NOISSUE - Introduce computation runner, log forwarder, ingress, and egress proxy services.

.mockery.yml

@@ -146,3 +146,10 @@ packages:
           dir: '{{.InterfaceDir}}/mocks'
           structname: '{{.InterfaceName}}'
           filename: "{{.InterfaceName | lower}}.go"
+  github.com/ultravioletrs/cocos/pkg/clients/grpc/runner:
+    interfaces:
+      Client:
+        config:
+          dir: '{{.InterfaceDir}}/mocks'
+          structname: '{{.InterfaceName}}'
+          filename: "{{.InterfaceName | lower}}.go"

Makefile

@@ -1,5 +1,5 @@
 BUILD_DIR = build
-SERVICES = manager agent cli attestation-service
+SERVICES = manager agent cli attestation-service log-forwarder computation-runner egress-proxy ingress-proxy
 ATTESTATION_POLICY = attestation_policy
 CGO_ENABLED ?= 0
 GOARCH ?= amd64
@@ -41,6 +41,8 @@ protoc:
 	protoc -I. --go_out=. --go_opt=paths=source_relative --go-grpc_out=. --go-grpc_opt=paths=source_relative agent/events/events.proto
 	protoc -I. --go_out=. --go_opt=paths=source_relative --go-grpc_out=. --go-grpc_opt=paths=source_relative agent/cvms/cvms.proto
 	protoc -I. --go_out=. --go_opt=paths=source_relative --go-grpc_out=. --go-grpc_opt=paths=source_relative internal/proto/attestation/v1/attestation.proto
+	protoc -I. --go_out=. --go_opt=paths=source_relative --go-grpc_out=. --go-grpc_opt=paths=source_relative agent/log/log.proto
+	protoc -I. --go_out=. --go_opt=paths=source_relative --go-grpc_out=. --go-grpc_opt=paths=source_relative agent/runner/runner.proto
 
 mocks:
 	mockery --config ./.mockery.yml

agent/computations.go

@@ -13,7 +13,6 @@ import (
 var _ fmt.Stringer = (*Datasets)(nil)
 
 type AgentConfig struct {
-	Port         string `json:"port,omitempty"`
 	CertFile     string `json:"cert_file,omitempty"`
 	KeyFile      string `json:"server_key,omitempty"`
 	ServerCAFile string `json:"server_ca_file,omitempty"`

agent/computations_test.go

@@ -105,16 +105,15 @@ func TestDecompressToContext(t *testing.T) {
 }
 
 func TestAgentConfigJSON(t *testing.T) {
-	config := AgentConfig{
-		Port:         "8080",
+	cfg := AgentConfig{
 		CertFile:     "cert.pem",
 		KeyFile:      "key.pem",
-		ServerCAFile: "server_ca.pem",
-		ClientCAFile: "client_ca.pem",
+		ServerCAFile: "server-ca.pem",
+		ClientCAFile: "client-ca.pem",
 		AttestedTls:  true,
 	}
 
-	data, err := json.Marshal(config)
+	data, err := json.Marshal(cfg)
 	if err != nil {
 		t.Fatalf("Failed to marshal AgentConfig: %v", err)
 	}
@@ -125,7 +124,7 @@ func TestAgentConfigJSON(t *testing.T) {
 		t.Fatalf("Failed to unmarshal AgentConfig: %v", err)
 	}
 
-	if !reflect.DeepEqual(config, unmarshaledConfig) {
-		t.Errorf("Unmarshaled config does not match original. Got %+v, want %+v", unmarshaledConfig, config)
+	if !reflect.DeepEqual(cfg, unmarshaledConfig) {
+		t.Errorf("Unmarshaled config does not match original. Got %+v, want %+v", unmarshaledConfig, cfg)
 	}
 }

agent/cvms/api/grpc/client.go

@@ -5,6 +5,7 @@ package grpc
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"log/slog"
 	"sync"
 	"time"
@@ -17,6 +18,7 @@ import (
 	"github.com/ultravioletrs/cocos/pkg/attestation"
 	"github.com/ultravioletrs/cocos/pkg/attestation/vtpm"
 	"github.com/ultravioletrs/cocos/pkg/clients/grpc"
+	"github.com/ultravioletrs/cocos/pkg/ingress"
 	"golang.org/x/sync/errgroup"
 	"google.golang.org/protobuf/proto"
 )
@@ -44,13 +46,14 @@ type CVMSClient struct {
 	logger        *slog.Logger
 	runReqManager *runRequestManager
 	sp            server.AgentServer
+	ingressProxy  ingress.ProxyServer
 	storage       storage.Storage
 	reconnectFn   func(context.Context) (grpc.Client, cvms.Service_ProcessClient, error)
 	grpcClient    grpc.Client
 }
 
 // NewClient returns new gRPC client instance.
-func NewClient(stream cvms.Service_ProcessClient, svc agent.Service, messageQueue chan *cvms.ClientStreamMessage, logger *slog.Logger, sp server.AgentServer, storageDir string, reconnectFn func(context.Context) (grpc.Client, cvms.Service_ProcessClient, error), grpcClient grpc.Client) (*CVMSClient, error) {
+func NewClient(stream cvms.Service_ProcessClient, svc agent.Service, messageQueue chan *cvms.ClientStreamMessage, logger *slog.Logger, sp server.AgentServer, ingressProxy ingress.ProxyServer, storageDir string, reconnectFn func(context.Context) (grpc.Client, cvms.Service_ProcessClient, error), grpcClient grpc.Client) (*CVMSClient, error) {
 	store, err := storage.NewFileStorage(storageDir)
 	if err != nil {
 		return nil, err
@@ -63,6 +66,7 @@ func NewClient(stream cvms.Service_ProcessClient, svc agent.Service, messageQueu
 		logger:        logger,
 		runReqManager: newRunRequestManager(),
 		sp:            sp,
+		ingressProxy:  ingressProxy,
 		storage:       store,
 		reconnectFn:   reconnectFn,
 		grpcClient:    grpcClient,
@@ -205,14 +209,17 @@ func (client *CVMSClient) handleAgentStateReq(mes *cvms.ServerStreamMessage_Agen
 }
 
 func (client *CVMSClient) handleRunReqChunks(ctx context.Context, msg *cvms.ServerStreamMessage_RunReqChunks) error {
+	client.logger.Debug("Received RunReq chunk", "id", msg.RunReqChunks.Id, "size", len(msg.RunReqChunks.Data), "isLast", msg.RunReqChunks.IsLast)
 	buffer, complete := client.runReqManager.addChunk(msg.RunReqChunks.Id, msg.RunReqChunks.Data, msg.RunReqChunks.IsLast)
 
 	if complete {
+		client.logger.Info("Received complete computation run request", "id", msg.RunReqChunks.Id, "totalSize", len(buffer))
 		var runReq cvms.ComputationRunReq
 		if err := proto.Unmarshal(buffer, &runReq); err != nil {
 			return errors.Wrap(err, errCorruptedManifest)
 		}
 
+		client.logger.Info("Starting computation execution", "computationId", runReq.Id, "name", runReq.Name)
 		go client.executeRun(ctx, &runReq)
 	}
 
@@ -246,6 +253,15 @@ func (client *CVMSClient) executeRun(ctx context.Context, runReq *cvms.Computati
 		})
 	}
 
+	// Check if the agent is in the correct state to initialize a new computation.
+	// If the agent is already processing this computation (e.g., after a reconnection),
+	// skip initialization to avoid state errors.
+	currentState := client.svc.State()
+	if currentState != "ReceivingManifest" {
+		client.logger.Info("Agent already processing computation, skipping initialization", "state", currentState, "computationId", runReq.Id)
+		return
+	}
+
 	if err := client.svc.InitComputation(ctx, ac); err != nil {
 		client.logger.Warn(err.Error())
 		return
@@ -267,7 +283,6 @@ func (client *CVMSClient) executeRun(ctx context.Context, runReq *cvms.Computati
 	}
 
 	if err := client.sp.Start(agent.AgentConfig{
-		Port:         runReq.AgentConfig.Port,
 		CertFile:     runReq.AgentConfig.CertFile,
 		KeyFile:      runReq.AgentConfig.KeyFile,
 		ServerCAFile: runReq.AgentConfig.ServerCaFile,
@@ -278,6 +293,22 @@ func (client *CVMSClient) executeRun(ctx context.Context, runReq *cvms.Computati
 		runRes.RunRes.Error = err.Error()
 	}
 
+	// Start ingress proxy if available
+	if client.ingressProxy != nil {
+		if err := client.ingressProxy.Start(
+			ingress.AgentConfigToProxyConfig(agent.AgentConfig{
+				CertFile:     runReq.AgentConfig.CertFile,
+				KeyFile:      runReq.AgentConfig.KeyFile,
+				ServerCAFile: runReq.AgentConfig.ServerCaFile,
+				ClientCAFile: runReq.AgentConfig.ClientCaFile,
+				AttestedTls:  runReq.AgentConfig.AttestedTls,
+			}),
+			ingress.ComputationToProxyContext(ac),
+		); err != nil {
+			client.logger.Warn(fmt.Sprintf("failed to start ingress proxy: %s", err.Error()))
+		}
+	}
+
 	defer func() {
 		if ccPlatform == attestation.Azure || ccPlatform == attestation.SNPvTPM {
 			cmpJson, err := json.Marshal(ac)
@@ -309,6 +340,12 @@ func (client *CVMSClient) handleStopComputation(ctx context.Context, mes *cvms.S
 	if err := client.sp.Stop(); err != nil {
 		msg.StopComputationRes.Message = err.Error()
 	}
+	// Stop ingress proxy if available
+	if client.ingressProxy != nil {
+		if err := client.ingressProxy.Stop(); err != nil {
+			client.logger.Warn(fmt.Sprintf("failed to stop ingress proxy: %s", err.Error()))
+		}
+	}
 	client.mu.Unlock()
 
 	client.sendMessage(&cvms.ClientStreamMessage{Message: msg})