From a5e6cae92c9dea5f6cbeff341ec842070183e07f Mon Sep 17 00:00:00 2001 From: Sammy Kerata Oina <44265300+SammyOina@users.noreply.github.com> Date: Thu, 11 Jan 2024 17:08:35 +0300 Subject: [PATCH] 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 * 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 * 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 * 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 * fix ci Signed-off-by: SammyOina --------- Signed-off-by: SammyOina --- cmd/manager/main.go | 7 +++-- manager/README.md | 1 + manager/api/grpc/client.go | 9 +++--- manager/api/grpc/endpoint.go | 5 ++-- manager/api/grpc/responses.go | 4 ++- manager/api/grpc/server.go | 3 +- manager/api/http/endpoint.go | 8 ++---- manager/api/http/responses.go | 4 ++- manager/api/logging.go | 2 +- manager/api/metrics.go | 2 +- manager/manager.pb.go | 27 ++++++++++++------ manager/manager.proto | 4 ++- manager/qemu/config.go | 16 ++++------- manager/service.go | 54 ++++++++++++++++++++++++++--------- manager/tracing/tracing.go | 2 +- 15 files changed, 94 insertions(+), 54 deletions(-) diff --git a/cmd/manager/main.go b/cmd/manager/main.go index f3f4c34e..a76b3541 100644 --- a/cmd/manager/main.go +++ b/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") diff --git a/manager/README.md b/manager/README.md index f64468bb..30ffdc53 100644 --- a/manager/README.md +++ b/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 diff --git a/manager/api/grpc/client.go b/manager/api/grpc/client.go index bfa86d39..8fed10b6 100644 --- a/manager/api/grpc/client.go +++ b/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 } diff --git a/manager/api/grpc/endpoint.go b/manager/api/grpc/endpoint.go index e7901496..122eafbb 100644 --- a/manager/api/grpc/endpoint.go +++ b/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 } } diff --git a/manager/api/grpc/responses.go b/manager/api/grpc/responses.go index a0304972..ee762e79 100644 --- a/manager/api/grpc/responses.go +++ b/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"` +} diff --git a/manager/api/grpc/server.go b/manager/api/grpc/server.go index e8da5344..28056995 100644 --- a/manager/api/grpc/server.go +++ b/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) { diff --git a/manager/api/http/endpoint.go b/manager/api/http/endpoint.go index 1ad0b697..3c8dee2e 100644 --- a/manager/api/http/endpoint.go +++ b/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 } } diff --git a/manager/api/http/responses.go b/manager/api/http/responses.go index 103337a5..fcf979d1 100644 --- a/manager/api/http/responses.go +++ b/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 diff --git a/manager/api/logging.go b/manager/api/logging.go index d9cb25da..ccab21e0 100644 --- a/manager/api/logging.go +++ b/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 { diff --git a/manager/api/metrics.go b/manager/api/metrics.go index c9f096d1..a1558224 100644 --- a/manager/api/metrics.go +++ b/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()) diff --git a/manager/manager.pb.go b/manager/manager.pb.go index 9aa8716f..92565b17 100644 --- a/manager/manager.pb.go +++ b/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 ( diff --git a/manager/manager.proto b/manager/manager.proto index f6daf0ba..102152ac 100644 --- a/manager/manager.proto +++ b/manager/manager.proto @@ -37,4 +37,6 @@ message Algorithm { string id = 2; } -message RunResponse { } +message RunResponse { + string agent_address = 1; +} diff --git a/manager/qemu/config.go b/manager/qemu/config.go index a833877d..910d34fb 100644 --- a/manager/qemu/config.go +++ b/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", diff --git a/manager/service.go b/manager/service.go index ceb8de04..e90ec7e6 100644 --- a/manager/service.go +++ b/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) { diff --git a/manager/tracing/tracing.go b/manager/tracing/tracing.go index aa9b65d7..c908fada 100644 --- a/manager/tracing/tracing.go +++ b/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()