diff --git a/agent/README.md b/agent/README.md index 60fa1fdf..078ae862 100644 --- a/agent/README.md +++ b/agent/README.md @@ -9,15 +9,10 @@ The service is configured using the environment variables from the following tab | Variable | Description | Default | | ----------------------------- | ------------------------------------------------------ | ------------------------------ | | AGENT_LOG_LEVEL | Log level for agent service (debug, info, warn, error) | info | -| AGENT_HTTP_HOST | Agent service HTTP host | "" | -| AGENT_HTTP_PORT | Agent service HTTP port | 9031 | -| AGENT_HTTP_SERVER_CERT | Path to HTTP server certificate in pem format | "" | -| AGENT_HTTP_SERVER_KEY | Path to HTTP server key in pem format | "" | | AGENT_GRPC_HOST | Agent service gRPC host | "" | | AGENT_GRPC_PORT | Agent service gRPC port | 7002 | | AGENT_GRPC_SERVER_CERT | Path to gRPC server certificate in pem format | "" | | AGENT_GRPC_SERVER_KEY | Path to gRPC server key in pem format | "" | -| COCOS_NOTIFICATION_SERVER_URL | Server to receive notification events from agent. | http:/localhost:9000 | ## Deployment diff --git a/agent/computations.go b/agent/computations.go index 41947390..8170cad8 100644 --- a/agent/computations.go +++ b/agent/computations.go @@ -14,12 +14,11 @@ var ( ) type AgentConfig struct { - LogLevel string `json:"log_level"` - InstanceID string `json:"instance_id"` - Host string `json:"host"` - Port string `json:"port"` - CertFile string `json:"cert_file"` - KeyFile string `json:"server_key"` + LogLevel string `json:"log_level"` + Host string `json:"host"` + Port string `json:"port"` + CertFile string `json:"cert_file"` + KeyFile string `json:"server_key"` } type Computation struct { diff --git a/api/manager.yml b/api/manager.yml deleted file mode 100644 index 9faeecd9..00000000 --- a/api/manager.yml +++ /dev/null @@ -1,138 +0,0 @@ -openapi: 3.0.1 -info: - title: CoCos-AI Manager - description: | - The manager service creates a VM with CoCos-AI agent: - - [The CoCos-AI repository](https://github.com/ultravioletrs/cocos-ai) - contact: - email: info@ultraviolet.com - license: - name: Apache 2.0 - url: https://github.com/ultravioletrs/cocos-ai/blob/main/LICENCE.md - version: "1" - -servers: - - url: http://localhost:9021 - - url: https://localhost:9021 - -paths: - /run: - post: - summary: Run computation on VM. - description: | - Send computation to agent. This creates agent virtual machine and sends - computation manifest. - requestBody: - $ref: "#/components/requestBodies/Run" - responses: - "200": - description: Computation manifest created successfully. - "400": - description: Request has malformed content. - '500': - $ref: "#/components/responses/ServiceError" - /health: - get: - summary: Retrieves service health check info. - tags: - - health - responses: - '200': - $ref: "#/components/responses/HealthRes" - '500': - $ref: "#/components/responses/ServiceError" - -components: - schemas: - HealthRes: - type: object - properties: - status: - type: string - description: Service status. - enum: - - pass - version: - type: string - description: Service version. - example: 0.0.1 - commit: - type: string - description: Service commit hash. - example: 7d6f4dc4f7f0c1fa3dc24eddfb18bb5073ff4f62 - description: - type: string - description: Service description. - example: service - build_time: - type: string - description: Service build time. - example: 1970-01-01_00:00:00 - requestBodies: - Run: - required: true - content: - application/json: - schema: - type: object - properties: - computation: - description: byte array of computation request as defined in ComputationReq - type: object - properties: - id: - type: string - name: - type: string - description: - type: string - datasets: - type: array - items: - type: object - properties: - provider: - type: string - id: - type: string - algorithms: - type: array - items: - type: object - properties: - provider: - type: string - id: - type: string - results_consumers: - type: array - items: - type: string - agent_config: - type: object - properties: - log_level: - type: string - instance_id: - type: string - notification_server_url: - type: string - host: - type: string - port: - type: string - cert_file: - type: string - key_file: - type: string - - responses: - ServiceError: - description: Unexpected server-side error occurred. - - HealthRes: - description: Service Health Check. - content: - application/json: - schema: - $ref: "#/components/schemas/HealthRes" diff --git a/cli/run.go b/cli/run.go index ebccaa4a..698e561d 100644 --- a/cli/run.go +++ b/cli/run.go @@ -31,12 +31,11 @@ func (cli *CLI) NewRunCmd() *cobra.Command { Name: cmp.Name, ResultConsumers: cmp.ResultConsumers, AgentConfig: &manager.AgentConfig{ - Port: cmp.AgentConfig.Port, - Host: cmp.AgentConfig.Host, - CertFile: cmp.AgentConfig.CertFile, - KeyFile: cmp.AgentConfig.KeyFile, - LogLevel: cmp.AgentConfig.LogLevel, - InstanceId: cmp.AgentConfig.InstanceID, + Port: cmp.AgentConfig.Port, + Host: cmp.AgentConfig.Host, + CertFile: cmp.AgentConfig.CertFile, + KeyFile: cmp.AgentConfig.KeyFile, + LogLevel: cmp.AgentConfig.LogLevel, }, } diff --git a/cmd/agent/main.go b/cmd/agent/main.go index c519a83d..c802aae7 100644 --- a/cmd/agent/main.go +++ b/cmd/agent/main.go @@ -10,7 +10,6 @@ import ( "log/slog" mglog "github.com/absmach/magistrala/logger" - "github.com/absmach/magistrala/pkg/uuid" "github.com/mdlayher/vsock" "github.com/ultravioletrs/cocos/agent" "github.com/ultravioletrs/cocos/agent/api" @@ -50,14 +49,6 @@ func main() { return } - if cfg.AgentConfig.InstanceID == "" { - cfg.AgentConfig.InstanceID, err = uuid.New().ID() - if err != nil { - log.Printf("Failed to generate instanceID: %s", err) - return - } - } - eventSvc, err := events.New(svcName, cfg.ID) if err != nil { log.Printf("failed to create events service %s", err.Error()) diff --git a/cmd/manager/main.go b/cmd/manager/main.go index 6e329653..00eaa3b1 100644 --- a/cmd/manager/main.go +++ b/cmd/manager/main.go @@ -19,12 +19,10 @@ import ( jaegerclient "github.com/ultravioletrs/cocos/internal/jaeger" "github.com/ultravioletrs/cocos/internal/server" grpcserver "github.com/ultravioletrs/cocos/internal/server/grpc" - httpserver "github.com/ultravioletrs/cocos/internal/server/http" "github.com/ultravioletrs/cocos/manager" "github.com/ultravioletrs/cocos/manager/agentevents" "github.com/ultravioletrs/cocos/manager/api" managergrpc "github.com/ultravioletrs/cocos/manager/api/grpc" - httpapi "github.com/ultravioletrs/cocos/manager/api/http" "github.com/ultravioletrs/cocos/manager/qemu" "github.com/ultravioletrs/cocos/manager/tracing" "go.opentelemetry.io/otel/trace" @@ -116,7 +114,6 @@ func main() { logger.Error(fmt.Sprintf("failed to load %s gRPC server configuration: %s", svcName, err)) return } - hs := httpserver.New(ctx, cancel, svcName, httpServerConfig, httpapi.MakeHandler(svc, cfg.InstanceID), logger) grpcServerConfig := server.Config{Port: defSvcGRPCPort} if err := env.Parse(&grpcServerConfig, env.Options{Prefix: envPrefixGRPC}); err != nil { @@ -129,18 +126,10 @@ func main() { } gs := grpcserver.New(ctx, cancel, svcName, grpcServerConfig, registerManagerServiceServer, logger) - g.Go(func() error { - return hs.Start() - }) - g.Go(func() error { return gs.Start() }) - g.Go(func() error { - return server.StopHandler(ctx, cancel, logger, svcName, hs, gs) - }) - if err := g.Wait(); err != nil { logger.Error(fmt.Sprintf("%s service terminated: %s", svcName, err)) } diff --git a/go.mod b/go.mod index a070948d..10236d9f 100644 --- a/go.mod +++ b/go.mod @@ -8,7 +8,6 @@ require ( github.com/cenkalti/backoff/v4 v4.2.1 github.com/digitalocean/go-libvirt v0.0.0-20221205150000-2939327a8519 github.com/go-kit/kit v0.13.0 - github.com/go-zoo/bone v1.3.0 github.com/gofrs/uuid v4.4.0+incompatible github.com/google/go-sev-guest v0.10.1 github.com/mdlayher/vsock v1.2.1 @@ -16,7 +15,6 @@ require ( github.com/spf13/cobra v1.8.0 github.com/spf13/pflag v1.0.5 go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.46.1 - go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.46.1 go.opentelemetry.io/otel v1.21.0 go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.21.0 go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.21.0 @@ -31,7 +29,6 @@ require ( cloud.google.com/go/compute v1.23.3 // indirect github.com/beorn7/perks v1.0.1 // indirect github.com/cespare/xxhash/v2 v2.2.0 // indirect - github.com/felixge/httpsnoop v1.0.4 // indirect github.com/go-kit/log v0.2.1 // indirect github.com/go-logfmt/logfmt v0.6.0 // indirect github.com/go-logr/logr v1.4.1 // indirect diff --git a/go.sum b/go.sum index d80ffdb2..feaa6721 100644 --- a/go.sum +++ b/go.sum @@ -26,8 +26,6 @@ github.com/digitalocean/go-libvirt v0.0.0-20221205150000-2939327a8519 h1:OpkN/n4 github.com/digitalocean/go-libvirt v0.0.0-20221205150000-2939327a8519/go.mod h1:WyJJyfmJ0gWJvjV+ZH4DOgtOYZc1KOvYyBXWCLKxsUU= github.com/envoyproxy/protoc-gen-validate v1.0.2 h1:QkIBuU5k+x7/QXPvPPnWXWlCdaBFApVqftFV6k087DA= github.com/envoyproxy/protoc-gen-validate v1.0.2/go.mod h1:GpiZQP3dDbg4JouG/NNS7QWXpgx6x8QiMKdmN72jogE= -github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg= -github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U= github.com/go-kit/kit v0.13.0 h1:OoneCcHKHQ03LfBpoQCUfCluwd2Vt3ohz+kvbJneZAU= github.com/go-kit/kit v0.13.0/go.mod h1:phqEHMMUbyrCFCTgH48JueqrM3md2HcAZ8N3XE4FKDg= github.com/go-kit/log v0.2.1 h1:MRVx0/zhvdseW+Gza6N9rVzU/IVzaeE1SFI4raAhmBU= @@ -39,8 +37,6 @@ github.com/go-logr/logr v1.4.1 h1:pKouT5E8xu9zeFC39JXRDukb6JFQPXM5p5I91188VAQ= github.com/go-logr/logr v1.4.1/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= -github.com/go-zoo/bone v1.3.0 h1:PY6sHq37FnQhj+4ZyqFIzJQHvrrGx0GEc3vTZZC/OsI= -github.com/go-zoo/bone v1.3.0/go.mod h1:HI3Lhb7G3UQcAwEhOJ2WyNcsFtQX1WYHa0Hl4OBbhW8= github.com/gofrs/uuid v4.4.0+incompatible h1:3qXRTX8/NbyulANqlc0lchS1gqAVxRgsuW1YrTJupqA= github.com/gofrs/uuid v4.4.0+incompatible/go.mod h1:b2aQJv3Z4Fp6yNu3cdSllBxTCLRxnplIgP/c0N/04lM= github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk= @@ -99,8 +95,6 @@ github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXl github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY= go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.46.1 h1:SpGay3w+nEwMpfVnbqOLH5gY52/foP8RE8UzTZ1pdSE= go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.46.1/go.mod h1:4UoMYEZOC0yN/sPGH76KPkkU7zgiEWYWL9vwmbnTJPE= -go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.46.1 h1:aFJWCqJMNjENlcleuuOkGAPH82y0yULBScfXcIEdS24= -go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.46.1/go.mod h1:sEGXWArGqc3tVa+ekntsN65DmVbVeW+7lTKTjZF3/Fo= go.opentelemetry.io/otel v1.21.0 h1:hzLeKBZEL7Okw2mGzZ0cc4k/A7Fta0uoPgaJCr8fsFc= go.opentelemetry.io/otel v1.21.0/go.mod h1:QZzNPQPm1zLX4gZK4cMi+71eaorMSGT3A4znnUvNNEo= go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.21.0 h1:cl5P5/GIfFh4t6xyruOgJP5QiA1pw4fYYdv6nc6CBWw= diff --git a/internal/server/http/doc.go b/internal/server/http/doc.go deleted file mode 100644 index f890b14a..00000000 --- a/internal/server/http/doc.go +++ /dev/null @@ -1,5 +0,0 @@ -// Copyright (c) Ultraviolet -// SPDX-License-Identifier: Apache-2.0 - -// Package http contains the HTTP server implementation. -package http diff --git a/internal/server/http/http.go b/internal/server/http/http.go deleted file mode 100644 index 4cd5b608..00000000 --- a/internal/server/http/http.go +++ /dev/null @@ -1,78 +0,0 @@ -// Copyright (c) Ultraviolet -// SPDX-License-Identifier: Apache-2.0 -package http - -import ( - "context" - "fmt" - "log/slog" - "net/http" - "time" - - "github.com/ultravioletrs/cocos/internal/server" -) - -const ( - stopWaitTime = 5 * time.Second - httpProtocol = "http" - httpsProtocol = "https" -) - -type Server struct { - server.BaseServer - server *http.Server -} - -var _ server.Server = (*Server)(nil) - -func New(ctx context.Context, cancel context.CancelFunc, name string, config server.Config, handler http.Handler, logger *slog.Logger) server.Server { - listenFullAddress := fmt.Sprintf("%s:%s", config.Host, config.Port) - httpServer := &http.Server{Addr: listenFullAddress, Handler: handler} - return &Server{ - BaseServer: server.BaseServer{ - Ctx: ctx, - Cancel: cancel, - Name: name, - Address: listenFullAddress, - Config: config, - Logger: logger, - }, - server: httpServer, - } -} - -func (s *Server) Start() error { - errCh := make(chan error) - s.Protocol = httpProtocol - switch { - case s.Config.CertFile != "" || s.Config.KeyFile != "": - s.Protocol = httpsProtocol - s.Logger.Info(fmt.Sprintf("%s service %s server listening at %s with TLS cert %s and key %s", s.Name, s.Protocol, s.Address, s.Config.CertFile, s.Config.KeyFile)) - go func() { - errCh <- s.server.ListenAndServeTLS(s.Config.CertFile, s.Config.KeyFile) - }() - default: - s.Logger.Info(fmt.Sprintf("%s service %s server listening at %s without TLS", s.Name, s.Protocol, s.Address)) - go func() { - errCh <- s.server.ListenAndServe() - }() - } - select { - case <-s.Ctx.Done(): - return s.Stop() - case err := <-errCh: - return err - } -} - -func (s *Server) Stop() error { - defer s.Cancel() - ctxShutdown, cancelShutdown := context.WithTimeout(context.Background(), stopWaitTime) - defer cancelShutdown() - if err := s.server.Shutdown(ctxShutdown); err != nil { - s.Logger.Error(fmt.Sprintf("%s service %s server error occurred during shutdown at %s: %s", s.Name, s.Protocol, s.Address, err)) - return fmt.Errorf("%s service %s server error occurred during shutdown at %s: %w", s.Name, s.Protocol, s.Address, err) - } - s.Logger.Info(fmt.Sprintf("%s %s service shutdown of http at %s", s.Name, s.Protocol, s.Address)) - return nil -} diff --git a/manager/README.md b/manager/README.md index c8b8bbf7..59751d62 100644 --- a/manager/README.md +++ b/manager/README.md @@ -1,6 +1,6 @@ # Manager -Manager service provides a barebones HTTP and gRPC API and Service interface implementation for the development of the manager service. +Manager service provides a barebones gRPC API and Service interface implementation for the development of the manager service. ## Configuration @@ -9,10 +9,6 @@ The service is configured using the environment variables from the following tab | Variable | Description | Default | | ----------------------------- | -------------------------------------------------------- | --------------------------------- | | MANAGER_LOG_LEVEL | Log level for manager service (debug, info, warn, error) | info | -| MANAGER_HTTP_HOST | Manager service HTTP host | | -| MANAGER_HTTP_PORT | Manager service HTTP port | 9021 | -| MANAGER_HTTP_SERVER_CERT | Path to server certificate in pem format | | -| MANAGER_HTTP_SERVER_KEY | Path to server key in pem format | | | MANAGER_GRPC_HOST | Manager service gRPC host | | | MANAGER_GRPC_PORT | Manager service gRPC port | 7001 | | MANAGER_GRPC_SERVER_CERT | Path to server certificate in pem format | | @@ -87,7 +83,7 @@ qemu-system-x86_64 \ -m 2048M,slots=5,maxmem=10240M \ -no-reboot \ -drive if=pflash,format=raw,unit=0,file=$OVMF_CODE,readonly=on \ - -netdev user,id=vmnic,hostfwd=tcp::2222-:22,hostfwd=tcp::9301-:9031,hostfwd=tcp::7020-:7002 \ + -netdev user,id=vmnic,hostfwd=tcp::7020-:7002 \ -device virtio-net-pci,disable-legacy=on,iommu_platform=true,netdev=vmnic,romfile= \ -device vhost-vsock-pci,id=vhost-vsock-pci0,guest-cid=3 -vnc :0 \ -kernel $KERNEL \ @@ -101,9 +97,6 @@ Once the VM is booted press enter and on the login use username `root`. #### Build and run Agent ```sh -# Start the 'agent' executable in the background using '&' at the end. -cocos-agent & - # List running processes and use 'grep' to filter for processes containing 'agent' in their names. ps aux | grep cocos-agent # This command helps verify that the 'agent' process is running. @@ -114,12 +107,6 @@ ps aux | grep cocos-agent We can also check if `Agent` is reachable from the host machine: ```sh -# Use netcat (nc) to test the connection to localhost on port 9301. -nc -zv localhost 9301 -# Output: -# nc: connect to localhost (::1) port 9301 (tcp) failed: Connection refused -# Connection to localhost (127.0.0.1) 9301 port [tcp/*] succeeded! - # Use netcat (nc) to test the connection to localhost on port 7020. nc -zv localhost 7020 # Output: @@ -186,37 +173,7 @@ MANAGER_QEMU_SEV_CBITPOS=51 \ To create an instance of VM and run a computation, run ```sh -curl -sSi -X POST \ - http://localhost:9021/run \ - -H "Content-Type: application/json" \ - -d '{ - "computation": { - "id":"c0d15c5e-e37d-4426-b3b7-b432c966fb09", - "name":"Sample_Computation", - "description":"A_sample_computation", - "datasets":[ - { - "provider":"Provider1", - "id":"Dataset1" - }, - { - "provider":"Provider2", - "id":"Dataset2" - } - ], - "algorithms":[ - { - "provider":"AlgorithmProvider1", - "id":"Algorithm1" - } - ], - "result_consumers":[ - "Consumer1" - ], - "timeout":"10m" - } -}' - +go run cmd/cli/main.go manager run '{"id":"123","name":"Sample Computation","description":"A sample computation","status":"Processing","owner":"John Doe","start_time":"2023-11-03T12:03:21.705171284+03:00","end_time":"2023-11-03T13:03:21.705171532+03:00","datasets":[{"provider":"Provider1","id":"Dataset1"},{"provider":"Provider2","id":"Dataset2"}],"algorithms":[{"provider":"AlgorithmProvider1","id":"Algorithm1"}],"result_consumers":["Consumer1","Consumer2"], "agent_config": {"port":"7002"}}' ``` You should be able to create multiple instances by reruning the command. diff --git a/manager/api/http/doc.go b/manager/api/http/doc.go deleted file mode 100644 index b3af73d4..00000000 --- a/manager/api/http/doc.go +++ /dev/null @@ -1,5 +0,0 @@ -// Copyright (c) Ultraviolet -// SPDX-License-Identifier: Apache-2.0 - -// Package http contains implementation of kit service HTTP API. -package http diff --git a/manager/api/http/endpoint.go b/manager/api/http/endpoint.go deleted file mode 100644 index e3297385..00000000 --- a/manager/api/http/endpoint.go +++ /dev/null @@ -1,49 +0,0 @@ -// Copyright (c) Ultraviolet -// SPDX-License-Identifier: Apache-2.0 -package http - -import ( - "context" - - "github.com/go-kit/kit/endpoint" - "github.com/ultravioletrs/cocos/manager" -) - -func runEndpoint(svc manager.Service) endpoint.Endpoint { - return func(ctx context.Context, request interface{}) (interface{}, error) { - req := request.(runReq) - - if err := req.validate(); err != nil { - return nil, err - } - - mc := manager.Computation{ - Id: req.Computation.ID, - Name: req.Computation.Name, - Description: req.Computation.Description, - ResultConsumers: req.Computation.ResultConsumers, - AgentConfig: &manager.AgentConfig{ - Port: req.Computation.AgentConfig.Port, - Host: req.Computation.AgentConfig.Host, - LogLevel: req.Computation.AgentConfig.LogLevel, - InstanceId: req.Computation.AgentConfig.InstanceID, - CertFile: req.Computation.AgentConfig.CertFile, - KeyFile: req.Computation.AgentConfig.KeyFile, - }, - } - for _, algo := range req.Computation.Algorithms { - mc.Algorithms = append(mc.Algorithms, &manager.Algorithm{Id: algo.ID, Provider: algo.Provider}) - } - for _, data := range req.Computation.Datasets { - mc.Datasets = append(mc.Datasets, &manager.Dataset{Id: data.ID, Provider: data.Provider}) - } - - // Call the Run method on the service - agAddr, err := svc.Run(ctx, &mc) - if err != nil { - return nil, err - } - - return runRes{AgentAddress: agAddr}, nil - } -} diff --git a/manager/api/http/requests.go b/manager/api/http/requests.go deleted file mode 100644 index 43204414..00000000 --- a/manager/api/http/requests.go +++ /dev/null @@ -1,21 +0,0 @@ -// Copyright (c) Ultraviolet -// SPDX-License-Identifier: Apache-2.0 -package http - -import ( - "github.com/ultravioletrs/cocos/agent" -) - -var _ apiReq = (*runReq)(nil) - -type apiReq interface { - validate() error -} - -type runReq struct { - Computation agent.Computation `json:"computation"` -} - -func (req *runReq) validate() error { - return nil -} diff --git a/manager/api/http/responses.go b/manager/api/http/responses.go deleted file mode 100644 index 1a8f5548..00000000 --- a/manager/api/http/responses.go +++ /dev/null @@ -1,27 +0,0 @@ -// Copyright (c) Ultraviolet -// SPDX-License-Identifier: Apache-2.0 -package http - -import ( - "net/http" - - "github.com/absmach/magistrala" -) - -var _ magistrala.Response = (*runRes)(nil) - -type runRes struct { - AgentAddress string `json:"agent_address"` -} - -func (res runRes) Code() int { - return http.StatusOK -} - -func (res runRes) Headers() map[string]string { - return map[string]string{} -} - -func (res runRes) Empty() bool { - return false -} diff --git a/manager/api/http/transport.go b/manager/api/http/transport.go deleted file mode 100644 index 987dae64..00000000 --- a/manager/api/http/transport.go +++ /dev/null @@ -1,108 +0,0 @@ -// Copyright (c) Ultraviolet -// SPDX-License-Identifier: Apache-2.0 -package http - -import ( - "context" - "encoding/json" - "errors" - "io" - "net/http" - "strings" - - "github.com/absmach/magistrala" - kithttp "github.com/go-kit/kit/transport/http" - "github.com/go-zoo/bone" - "github.com/prometheus/client_golang/prometheus/promhttp" - "github.com/ultravioletrs/cocos/manager" - "go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp" -) - -const contentType = "application/json" - -var ( - errUnsupportedContentType = errors.New("unsupported content type") - errInvalidQueryParams = errors.New("invalid query params") -) - -// MakeHandler returns a HTTP handler for API endpoints. -func MakeHandler(svc manager.Service, instanceID string) http.Handler { - opts := []kithttp.ServerOption{ - kithttp.ServerErrorEncoder(encodeError), - } - - r := bone.New() - - r.Post("/run", otelhttp.NewHandler(kithttp.NewServer( - runEndpoint(svc), - decodeRun, - encodeResponse, - opts..., - ), "run")) - - r.GetFunc("/health", magistrala.Health("manager", instanceID)) - r.Handle("/metrics", promhttp.Handler()) - - return r -} - -func decodeRun(_ context.Context, r *http.Request) (interface{}, error) { - if !strings.Contains(r.Header.Get("Content-Type"), contentType) { - return nil, errUnsupportedContentType - } - - var req runReq - if err := json.NewDecoder(r.Body).Decode(&req); err != nil { - return nil, err - } - - return req, nil -} - -func encodeResponse(_ context.Context, w http.ResponseWriter, response interface{}) error { - w.Header().Set("Content-Type", contentType) - - if ar, ok := response.(magistrala.Response); ok { - for k, v := range ar.Headers() { - w.Header().Set(k, v) - } - - w.WriteHeader(ar.Code()) - - if ar.Empty() { - return nil - } - } - - return json.NewEncoder(w).Encode(response) -} - -func encodeError(_ context.Context, err error, w http.ResponseWriter) { - w.Header().Set("Content-Type", contentType) - - switch err { - case manager.ErrMalformedEntity: - w.WriteHeader(http.StatusBadRequest) - case manager.ErrNotFound: - w.WriteHeader(http.StatusNotFound) - case manager.ErrUnauthorizedAccess: - w.WriteHeader(http.StatusForbidden) - case errUnsupportedContentType: - w.WriteHeader(http.StatusUnsupportedMediaType) - case errInvalidQueryParams: - w.WriteHeader(http.StatusBadRequest) - case io.ErrUnexpectedEOF: - w.WriteHeader(http.StatusBadRequest) - case io.EOF: - w.WriteHeader(http.StatusBadRequest) - default: - switch err.(type) { - case *json.SyntaxError: - w.WriteHeader(http.StatusBadRequest) - case *json.UnmarshalTypeError: - w.WriteHeader(http.StatusBadRequest) - default: - w.WriteHeader(http.StatusInternalServerError) - } - } -} diff --git a/manager/manager.pb.go b/manager/manager.pb.go index e4d6c68b..05fee9b4 100644 --- a/manager/manager.pb.go +++ b/manager/manager.pb.go @@ -327,12 +327,11 @@ type AgentConfig struct { sizeCache protoimpl.SizeCache unknownFields protoimpl.UnknownFields - Port string `protobuf:"bytes,1,opt,name=port,proto3" json:"port,omitempty"` - Host string `protobuf:"bytes,2,opt,name=host,proto3" json:"host,omitempty"` - CertFile string `protobuf:"bytes,3,opt,name=cert_file,json=certFile,proto3" json:"cert_file,omitempty"` - KeyFile string `protobuf:"bytes,4,opt,name=key_file,json=keyFile,proto3" json:"key_file,omitempty"` - LogLevel string `protobuf:"bytes,5,opt,name=log_level,json=logLevel,proto3" json:"log_level,omitempty"` - InstanceId string `protobuf:"bytes,6,opt,name=instance_id,json=instanceId,proto3" json:"instance_id,omitempty"` + Port string `protobuf:"bytes,1,opt,name=port,proto3" json:"port,omitempty"` + Host string `protobuf:"bytes,2,opt,name=host,proto3" json:"host,omitempty"` + CertFile string `protobuf:"bytes,3,opt,name=cert_file,json=certFile,proto3" json:"cert_file,omitempty"` + KeyFile string `protobuf:"bytes,4,opt,name=key_file,json=keyFile,proto3" json:"key_file,omitempty"` + LogLevel string `protobuf:"bytes,5,opt,name=log_level,json=logLevel,proto3" json:"log_level,omitempty"` } func (x *AgentConfig) Reset() { @@ -402,13 +401,6 @@ func (x *AgentConfig) GetLogLevel() string { return "" } -func (x *AgentConfig) GetInstanceId() string { - if x != nil { - return x.InstanceId - } - return "" -} - var File_manager_manager_proto protoreflect.FileDescriptor var file_manager_manager_proto_rawDesc = []byte{ @@ -446,7 +438,7 @@ var file_manager_manager_proto_rawDesc = []byte{ 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, 0x22, 0xab, 0x01, 0x0a, 0x0b, 0x41, 0x67, 0x65, 0x6e, 0x74, + 0x64, 0x64, 0x72, 0x65, 0x73, 0x73, 0x22, 0x8a, 0x01, 0x0a, 0x0b, 0x41, 0x67, 0x65, 0x6e, 0x74, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x12, 0x12, 0x0a, 0x04, 0x70, 0x6f, 0x72, 0x74, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x70, 0x6f, 0x72, 0x74, 0x12, 0x12, 0x0a, 0x04, 0x68, 0x6f, 0x73, 0x74, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x68, 0x6f, 0x73, 0x74, 0x12, 0x1b, @@ -455,14 +447,12 @@ var file_manager_manager_proto_rawDesc = []byte{ 0x65, 0x79, 0x5f, 0x66, 0x69, 0x6c, 0x65, 0x18, 0x04, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x6b, 0x65, 0x79, 0x46, 0x69, 0x6c, 0x65, 0x12, 0x1b, 0x0a, 0x09, 0x6c, 0x6f, 0x67, 0x5f, 0x6c, 0x65, 0x76, 0x65, 0x6c, 0x18, 0x05, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x6c, 0x6f, 0x67, 0x4c, 0x65, - 0x76, 0x65, 0x6c, 0x12, 0x1f, 0x0a, 0x0b, 0x69, 0x6e, 0x73, 0x74, 0x61, 0x6e, 0x63, 0x65, 0x5f, - 0x69, 0x64, 0x18, 0x06, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0a, 0x69, 0x6e, 0x73, 0x74, 0x61, 0x6e, - 0x63, 0x65, 0x49, 0x64, 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, + 0x76, 0x65, 0x6c, 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 d1025b85..921dbef4 100644 --- a/manager/manager.proto +++ b/manager/manager.proto @@ -45,5 +45,4 @@ message AgentConfig { string cert_file = 3; string key_file = 4; string log_level = 5; - string instance_id = 6; } diff --git a/manager/service.go b/manager/service.go index 87daae4e..4e070c26 100644 --- a/manager/service.go +++ b/manager/service.go @@ -70,12 +70,11 @@ func (ms *managerService) Run(ctx context.Context, c *Computation) (string, erro Description: c.Description, ResultConsumers: c.ResultConsumers, AgentConfig: agent.AgentConfig{ - Port: c.AgentConfig.Port, - Host: c.AgentConfig.Host, - KeyFile: c.AgentConfig.KeyFile, - CertFile: c.AgentConfig.CertFile, - LogLevel: c.AgentConfig.LogLevel, - InstanceID: c.AgentConfig.InstanceId, + Port: c.AgentConfig.Port, + Host: c.AgentConfig.Host, + KeyFile: c.AgentConfig.KeyFile, + CertFile: c.AgentConfig.CertFile, + LogLevel: c.AgentConfig.LogLevel, }, } for _, algo := range c.Algorithms { diff --git a/test/manual/README.md b/test/manual/README.md index 49ec3f14..de626fd0 100644 --- a/test/manual/README.md +++ b/test/manual/README.md @@ -26,9 +26,10 @@ Open console on the host, and run ```sh export AGENT_GRPC_URL=localhost:7002 +export MANAGER_GRPC_URL=localhost:7001 # Run CLI to provide manifest -go run cmd/cli/main.go run --computation '{"id":"123","name":"Sample Computation","description":"A sample computation","status":"Processing","owner":"John Doe","start_time":"2023-11-03T12:03:21.705171284+03:00","end_time":"2023-11-03T13:03:21.705171532+03:00","datasets":[{"provider":"Provider1","id":"Dataset1"},{"provider":"Provider2","id":"Dataset2"}],"algorithms":[{"provider":"AlgorithmProvider1","id":"Algorithm1"}],"result_consumers":["Consumer1","Consumer2"],"ttl":3600,"metadata":{"key1":"value1","key2":42}, "timeout": "2m"}' +go run cmd/cli/main.go run '{"id":"123","name":"Sample Computation","description":"A sample computation","status":"Processing","owner":"John Doe","start_time":"2023-11-03T12:03:21.705171284+03:00","end_time":"2023-11-03T13:03:21.705171532+03:00","datasets":[{"provider":"Provider1","id":"Dataset1"},{"provider":"Provider2","id":"Dataset2"}],"algorithms":[{"provider":"AlgorithmProvider1","id":"Algorithm1"}],"result_consumers":["Consumer1","Consumer2"],"ttl":3600,"metadata":{"key1":"value1","key2":42}, "timeout": "2m"}' # Run the CLI program with algorithm input go run cmd/cli/main.go algo test/manual/algo/lin_reg.py Algorithm1 AlgorithmProvider1