COCOS-44 - Add agent address to run responses (#45)

* Add agent address to run responses

The manager service's Run method now retrieves the agent address upon successful computation execution, providing more informative responses across gRPC, HTTP, and logging endpoints. This change improves service transparency by returning the agent's address to be used by client services, making the manager service's external communication more comprehensive. Updated the `RunResponse` structure in the corresponding protocol buffers definition and response handling in gRPC and HTTP APIs, ensuring that agent address information is serialized appropriately. It also necessitates a slight adjustment in the QEMU configuration to manage port forwarding rules more dynamically, simplifying the process as only agent-relevant ports are incremented with each new computation. This extra detail in responses aids in debugging and offers better integration capabilities for clients.

Signed-off-by: SammyOina <sammyoina@gmail.com>

* Allocate dynamic ports for VM guests and expand error handling

Refactored service initialization to accept host IP and incorporated dynamic port allocation for VM guests, replacing the prior static increment method. Introduced a new error type 'ErrFailedToAllocatePort' to capture instances where the system is unable to find a free port. Integrated a third-party error package for improved error wrapping and context. These changes prevent port conflicts between VM guests and enhance error diagnostics for service operations.

Resolves issue with static port allocation leading to conflicts.

Signed-off-by: SammyOina <sammyoina@gmail.com>

* Add HOST_IP to service configuration

Extend service configuration to include the host machine's IP address, allowing instances to be aware of their deployment environment. This update passes the new HostIP field to the service constructor, ensuring the service can now operate with host-specific logic.

Signed-off-by: SammyOina <sammyoina@gmail.com>

* Populate AgentAddress in gRPC Responses

Enhanced the gRPC encode/decode functions to properly populate the 'AgentAddress' field in 'RunResponse' objects. This ensures that consumers of the gRPC interface receive complete response data, which previously omitted the important 'AgentAddress' information. The change impacts both server-side response encoding and client-side response decoding, aligning the implementation with the expected interface contract.

Signed-off-by: SammyOina <sammyoina@gmail.com>

* fix ci

Signed-off-by: SammyOina <sammyoina@gmail.com>

---------

Signed-off-by: SammyOina <sammyoina@gmail.com>

cmd/manager/main.go

@@ -47,6 +47,7 @@ type config struct {
 	JaegerURL             string `env:"COCOS_JAEGER_URL"         envDefault:"http://localhost:14268/api/traces"`
 	InstanceID            string `env:"MANAGER_INSTANCE_ID"      envDefault:""`
 	NotificationServerURL string `env:"COCOS_NOTIFICATION_SERVER_URL" envDefault:"http://localhost:9000"`
+	HostIP                string `env:"MANAGER_HOST_IP"          envDefault:"localhost"`
 }
 
 func main() {
@@ -91,7 +92,7 @@ func main() {
 	}
 	logger.Info(fmt.Sprintf("%s %s", exe, strings.Join(args, " ")))
 
-	svc := newService(logger, tracer, qemuCfg, events.New(svcName, cfg.NotificationServerURL))
+	svc := newService(logger, tracer, qemuCfg, events.New(svcName, cfg.NotificationServerURL), cfg)
 
 	httpServerConfig := server.Config{Port: defSvcHTTPPort}
 	if err := env.Parse(&httpServerConfig, env.Options{Prefix: envPrefixHTTP}); err != nil {
@@ -132,8 +133,8 @@ func main() {
 	}
 }
 
-func newService(logger mglog.Logger, tracer trace.Tracer, qemuCfg qemu.Config, eventSvc events.Service) manager.Service {
-	svc := manager.New(qemuCfg, logger, eventSvc)
+func newService(logger mglog.Logger, tracer trace.Tracer, qemuCfg qemu.Config, eventSvc events.Service, cfg config) manager.Service {
+	svc := manager.New(qemuCfg, logger, eventSvc, cfg.HostIP)
 
 	svc = api.LoggingMiddleware(svc, logger)
 	counter, latency := internal.MakeMetrics(svcName, "api")

manager/README.md

@@ -20,6 +20,7 @@ The service is configured using the environment variables from the following tab
 | COCOS_JAEGER_URL              | Jaeger server URL                                        | http://localhost:14268/api/traces |
 | MANAGER_INSTANCE_ID           | Manager service instance ID                              |                                   |
 | COCOS_NOTIFICATION_SERVER_URL | Server to receive notification events from agent.        | http:/localhost:9000              |
+| MANAGER_HOST_IP               | Mnagaer host IP address                                  | localhost                         |
 
 ## Deployment
 

manager/api/grpc/client.go

@@ -53,11 +53,11 @@ func encodeRunRequest(_ context.Context, request interface{}) (interface{}, erro
 // decodeRunResponse is a transport/grpc.DecodeResponseFunc that
 // converts a gRPC RunResponse to a user-domain response.
 func decodeRunResponse(_ context.Context, grpcResponse interface{}) (interface{}, error) {
-	_, ok := grpcResponse.(*manager.RunResponse)
+	res, ok := grpcResponse.(*manager.RunResponse)
 	if !ok {
 		return nil, fmt.Errorf("invalid response type: %T", grpcResponse)
 	}
-	return runRes{}, nil
+	return runRes{AgentAddress: res.AgentAddress}, nil
 }
 
 func (client grpcClient) Run(ctx context.Context, req *manager.RunRequest, _ ...grpc.CallOption) (*manager.RunResponse, error) {
@@ -76,10 +76,11 @@ func (client grpcClient) Run(ctx context.Context, req *manager.RunRequest, _ ...
 		Timeout:     dur,
 	}
 
-	_, err = client.run(ctx, runReq)
+	res, err := client.run(ctx, runReq)
 	if err != nil {
 		return nil, err
 	}
 
-	return &manager.RunResponse{}, nil
+	runRes := res.(runRes)
+	return &manager.RunResponse{AgentAddress: runRes.AgentAddress}, nil
 }

manager/api/grpc/endpoint.go

@@ -28,10 +28,11 @@ func runEndpoint(svc manager.Service) endpoint.Endpoint {
 			agentConf.Timeout = 60 * time.Second
 		}
 
-		if err := svc.Run(ctx, req.Computation); err != nil {
+		agAddr, err := svc.Run(ctx, req.Computation)
+		if err != nil {
 			return runRes{}, err
 		}
 
-		return runRes{}, nil
+		return runRes{AgentAddress: agAddr}, nil
 	}
 }

manager/api/grpc/responses.go

@@ -2,4 +2,6 @@
 // SPDX-License-Identifier: Apache-2.0
 package grpc
 
-type runRes struct{}
+type runRes struct {
+	AgentAddress string `json:"agent_address"`
+}

manager/api/grpc/server.go

@@ -41,7 +41,8 @@ func decodeRunRequest(_ context.Context, grpcReq interface{}) (interface{}, erro
 }
 
 func encodeRunResponse(_ context.Context, response interface{}) (interface{}, error) {
-	return &manager.RunResponse{}, nil
+	res := response.(runRes)
+	return &manager.RunResponse{AgentAddress: res.AgentAddress}, nil
 }
 
 func (s *grpcServer) Run(ctx context.Context, req *manager.RunRequest) (*manager.RunResponse, error) {

manager/api/http/endpoint.go

@@ -41,13 +41,11 @@ func runEndpoint(svc manager.Service) endpoint.Endpoint {
 		}
 
 		// Call the Run method on the service
-		if err := svc.Run(ctx, &mc); err != nil {
+		agAddr, err := svc.Run(ctx, &mc)
+		if err != nil {
 			return nil, err
 		}
 
-		// Create the response
-		res := runRes{}
-
-		return res, nil
+		return runRes{AgentAddress: agAddr}, nil
 	}
 }

manager/api/http/responses.go

@@ -10,7 +10,9 @@ import (
 
 var _ magistrala.Response = (*runRes)(nil)
 
-type runRes struct{}
+type runRes struct {
+	AgentAddress string `json:"agent_address"`
+}
 
 func (res runRes) Code() int {
 	return http.StatusOK

manager/api/logging.go

@@ -27,7 +27,7 @@ func LoggingMiddleware(svc manager.Service, logger mglog.Logger) manager.Service
 	return &loggingMiddleware{logger, svc}
 }
 
-func (lm *loggingMiddleware) Run(ctx context.Context, mc *manager.Computation) (err error) {
+func (lm *loggingMiddleware) Run(ctx context.Context, mc *manager.Computation) (agentAddr string, err error) {
 	defer func(begin time.Time) {
 		message := fmt.Sprintf("Method Run for computation took %s to complete", time.Since(begin))
 		if err != nil {

manager/api/metrics.go

@@ -32,7 +32,7 @@ func MetricsMiddleware(svc manager.Service, counter metrics.Counter, latency met
 	}
 }
 
-func (ms *metricsMiddleware) Run(ctx context.Context, mc *manager.Computation) error {
+func (ms *metricsMiddleware) Run(ctx context.Context, mc *manager.Computation) (string, error) {
 	defer func(begin time.Time) {
 		ms.counter.With("method", "Run").Add(1)
 		ms.latency.With("method", "Run").Observe(time.Since(begin).Seconds())

manager/manager.pb.go

@@ -295,6 +295,8 @@ type RunResponse struct {
 	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache
 	unknownFields protoimpl.UnknownFields
+
+	AgentAddress string `protobuf:"bytes,1,opt,name=agent_address,json=agentAddress,proto3" json:"agent_address,omitempty"`
 }
 
 func (x *RunResponse) Reset() {
@@ -329,6 +331,13 @@ func (*RunResponse) Descriptor() ([]byte, []int) {
 	return file_manager_manager_proto_rawDescGZIP(), []int{4}
 }
 
+func (x *RunResponse) GetAgentAddress() string {
+	if x != nil {
+		return x.AgentAddress
+	}
+	return ""
+}
+
 var File_manager_manager_proto protoreflect.FileDescriptor
 
 var file_manager_manager_proto_rawDesc = []byte{
@@ -364,14 +373,16 @@ var file_manager_manager_proto_rawDesc = []byte{
 	0x09, 0x52, 0x02, 0x69, 0x64, 0x22, 0x37, 0x0a, 0x09, 0x41, 0x6c, 0x67, 0x6f, 0x72, 0x69, 0x74,
 	0x68, 0x6d, 0x12, 0x1a, 0x0a, 0x08, 0x70, 0x72, 0x6f, 0x76, 0x69, 0x64, 0x65, 0x72, 0x18, 0x01,
 	0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x70, 0x72, 0x6f, 0x76, 0x69, 0x64, 0x65, 0x72, 0x12, 0x0e,
-	0x0a, 0x02, 0x69, 0x64, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x02, 0x69, 0x64, 0x22, 0x0d,
-	0x0a, 0x0b, 0x52, 0x75, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x32, 0x44, 0x0a,
-	0x0e, 0x4d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x53, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x12,
-	0x32, 0x0a, 0x03, 0x52, 0x75, 0x6e, 0x12, 0x13, 0x2e, 0x6d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72,
-	0x2e, 0x52, 0x75, 0x6e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x14, 0x2e, 0x6d, 0x61,
-	0x6e, 0x61, 0x67, 0x65, 0x72, 0x2e, 0x52, 0x75, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73,
-	0x65, 0x22, 0x00, 0x42, 0x0b, 0x5a, 0x09, 0x2e, 0x2f, 0x6d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72,
-	0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+	0x0a, 0x02, 0x69, 0x64, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x02, 0x69, 0x64, 0x22, 0x32,
+	0x0a, 0x0b, 0x52, 0x75, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x23, 0x0a,
+	0x0d, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x5f, 0x61, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73, 0x18, 0x01,
+	0x20, 0x01, 0x28, 0x09, 0x52, 0x0c, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x41, 0x64, 0x64, 0x72, 0x65,
+	0x73, 0x73, 0x32, 0x44, 0x0a, 0x0e, 0x4d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x53, 0x65, 0x72,
+	0x76, 0x69, 0x63, 0x65, 0x12, 0x32, 0x0a, 0x03, 0x52, 0x75, 0x6e, 0x12, 0x13, 0x2e, 0x6d, 0x61,
+	0x6e, 0x61, 0x67, 0x65, 0x72, 0x2e, 0x52, 0x75, 0x6e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74,
+	0x1a, 0x14, 0x2e, 0x6d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x2e, 0x52, 0x75, 0x6e, 0x52, 0x65,
+	0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x42, 0x0b, 0x5a, 0x09, 0x2e, 0x2f, 0x6d, 0x61,
+	0x6e, 0x61, 0x67, 0x65, 0x72, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
 }
 
 var (

manager/manager.proto

@@ -37,4 +37,6 @@ message Algorithm {
   string id = 2;
 }
 
-message RunResponse { }
+message RunResponse {
+  string agent_address = 1;
+}

manager/qemu/config.go

@@ -28,13 +28,9 @@ type OVMFVarsConfig struct {
 }
 
 type NetDevConfig struct {
-	ID        string `env:"NETDEV_ID" envDefault:"vmnic"`
-	HostFwd1  int    `env:"HOST_FWD_1" envDefault:"2222"`
-	GuestFwd1 int    `env:"GUEST_FWD_1" envDefault:"22"`
-	HostFwd2  int    `env:"HOST_FWD_2" envDefault:"9301"`
-	GuestFwd2 int    `env:"GUEST_FWD_2" envDefault:"9031"`
-	HostFwd3  int    `env:"HOST_FWD_3" envDefault:"7020"`
-	GuestFwd3 int    `env:"GUEST_FWD_3" envDefault:"7002"`
+	ID            string `env:"NETDEV_ID"       envDefault:"vmnic"`
+	HostFwdAgent  int    `env:"HOST_FWD_AGENT"  envDefault:"7020"`
+	GuestFwdAgent int    `env:"GUEST_FWD_AGENT" envDefault:"7002"`
 }
 
 type VirtioNetPciConfig struct {
@@ -149,11 +145,9 @@ func constructQemuArgs(config Config, computation string) []string {
 
 	// network
 	args = append(args, "-netdev",
-		fmt.Sprintf("user,id=%s,hostfwd=tcp::%d-:%d,hostfwd=tcp::%d-:%d,hostfwd=tcp::%d-:%d",
+		fmt.Sprintf("user,id=%s,hostfwd=tcp::%d-:%d",
 			config.NetDevConfig.ID,
-			config.NetDevConfig.HostFwd1, config.NetDevConfig.GuestFwd1,
-			config.NetDevConfig.HostFwd2, config.NetDevConfig.GuestFwd2,
-			config.NetDevConfig.HostFwd3, config.NetDevConfig.GuestFwd3))
+			config.NetDevConfig.HostFwdAgent, config.NetDevConfig.GuestFwdAgent))
 
 	args = append(args, "-device",
 		fmt.Sprintf("virtio-net-pci,disable-legacy=%s,iommu_platform=%v,netdev=%s,romfile=%s",

manager/service.go

@@ -5,9 +5,12 @@ package manager
 import (
 	"context"
 	"encoding/json"
-	"errors"
+	"fmt"
+	"net"
+	"strconv"
 
 	mglog "github.com/absmach/magistrala/logger"
+	"github.com/absmach/magistrala/pkg/errors"
 	"github.com/ultravioletrs/cocos/agent"
 	"github.com/ultravioletrs/cocos/internal/events"
 	"github.com/ultravioletrs/cocos/manager/qemu"
@@ -24,31 +27,36 @@ var (
 
 	// ErrNotFound indicates a non-existent entity request.
 	ErrNotFound = errors.New("entity not found")
+
+	// ErrFailedToAllocatePort indicates no free port was found on host.
+	ErrFailedToAllocatePort = errors.New("failed to allocate free port on host")
 )
 
 // Service specifies an API that must be fulfilled by the domain service
 // implementation, and all of its decorators (e.g. logging & metrics).
 type Service interface {
-	Run(ctx context.Context, c *Computation) error
+	Run(ctx context.Context, c *Computation) (string, error)
 }
 
 type managerService struct {
 	qemuCfg  qemu.Config
 	logger   mglog.Logger
 	eventSvc events.Service
+	hostIP   string
 }
 
 var _ Service = (*managerService)(nil)
 
 // New instantiates the manager service implementation.
-func New(qemuCfg qemu.Config, logger mglog.Logger, eventSvc events.Service) Service {
+func New(qemuCfg qemu.Config, logger mglog.Logger, eventSvc events.Service, hostIP string) Service {
 	return &managerService{
 		qemuCfg:  qemuCfg,
 		eventSvc: eventSvc,
+		hostIP:   hostIP,
 	}
 }
 
-func (ms *managerService) Run(ctx context.Context, c *Computation) error {
+func (ms *managerService) Run(ctx context.Context, c *Computation) (string, error) {
 	ms.publishEvent("vm-provision", c.Id, "starting", json.RawMessage{})
 	ac := agent.Computation{
 		ID:              c.Id,
@@ -63,20 +71,38 @@ func (ms *managerService) Run(ctx context.Context, c *Computation) error {
 		ac.Datasets = append(ac.Datasets, agent.Dataset{ID: data.Id, Provider: data.Provider})
 	}
 
-	ms.publishEvent("vm-provision", c.Id, "in-progress", json.RawMessage{})
-	if _, err := qemu.CreateVM(ctx, ms.qemuCfg, ac); err != nil {
+	agentPort, err := getFreePort()
+	if err != nil {
 		ms.publishEvent("vm-provision", c.Id, "failed", json.RawMessage{})
-		return err
+		return "", errors.Wrap(ErrFailedToAllocatePort, err)
+	}
+	ms.qemuCfg.HostFwdAgent = agentPort
+
+	ms.publishEvent("vm-provision", c.Id, "in-progress", json.RawMessage{})
+	if _, err = qemu.CreateVM(ctx, ms.qemuCfg, ac); err != nil {
+		ms.publishEvent("vm-provision", c.Id, "failed", json.RawMessage{})
+		return "", err
 	}
-	// Different VM guests can't forward ports to the same ports on the same host.
-	defer func() {
-		ms.qemuCfg.HostFwd1++
-		ms.qemuCfg.NetDevConfig.HostFwd2++
-		ms.qemuCfg.NetDevConfig.HostFwd3++
-	}()
 
 	ms.publishEvent("vm-provision", c.Id, "complete", json.RawMessage{})
-	return nil
+	return fmt.Sprintf("%s:%d", ms.hostIP, ms.qemuCfg.HostFwdAgent), nil
+}
+
+func getFreePort() (int, error) {
+	listener, err := net.Listen("tcp", "")
+	if err != nil {
+		return 0, err
+	}
+	defer listener.Close()
+	_, portStr, err := net.SplitHostPort(listener.Addr().String())
+	if err != nil {
+		return 0, err
+	}
+	port, err := strconv.Atoi(portStr)
+	if err != nil {
+		return 0, err
+	}
+	return port, nil
 }
 
 func (ms *managerService) publishEvent(event, cmpID, status string, details json.RawMessage) {

manager/tracing/tracing.go

@@ -21,7 +21,7 @@ func New(svc manager.Service, tracer trace.Tracer) manager.Service {
 	return &tracingMiddleware{tracer, svc}
 }
 
-func (tm *tracingMiddleware) Run(ctx context.Context, mc *manager.Computation) error {
+func (tm *tracingMiddleware) Run(ctx context.Context, mc *manager.Computation) (string, error) {
 	ctx, span := tm.tracer.Start(ctx, "run")
 	defer span.End()