diff --git a/cli/run.go b/cli/run.go deleted file mode 100644 index 49398219..00000000 --- a/cli/run.go +++ /dev/null @@ -1,58 +0,0 @@ -// Copyright (c) Ultraviolet -// SPDX-License-Identifier: Apache-2.0 -package cli - -import ( - "encoding/json" - "log" - - "github.com/spf13/cobra" - "github.com/ultravioletrs/cocos/agent" - "github.com/ultravioletrs/cocos/manager" -) - -func (cli *CLI) NewRunCmd() *cobra.Command { - return &cobra.Command{ - Use: "run", - Short: "Upload a computation manifest json", - Example: "run ''", - Args: cobra.ExactArgs(1), - Run: func(cmd *cobra.Command, args []string) { - computationStr := args[0] - - var cmp agent.Computation - if err := json.Unmarshal([]byte(computationStr), &cmp); err != nil { - log.Fatalf("Error unmarshling computation json: %v", err) - } - - req := manager.ComputationRunReq{ - Id: cmp.ID, - Description: cmp.Description, - 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, - }, - } - - for _, data := range cmp.Datasets { - req.Datasets = append(req.Datasets, &manager.Dataset{Id: data.ID, Provider: data.Provider}) - } - - for _, algo := range cmp.Algorithms { - req.Algorithms = append(req.Algorithms, &manager.Algorithm{Id: algo.ID, Provider: algo.Provider}) - } - - response, err := cli.managerSDK.Run(cmd.Context(), &req) - if err != nil { - log.Fatalf("Error running computation: %v", err) - } - - log.Printf("Successfully run computation, agent address: %v", response) - }, - } -} diff --git a/cli/sdk.go b/cli/sdk.go index eb424b62..9d9c3df7 100644 --- a/cli/sdk.go +++ b/cli/sdk.go @@ -4,17 +4,14 @@ package cli import ( "github.com/ultravioletrs/cocos/agent" - "github.com/ultravioletrs/cocos/manager" ) type CLI struct { - agentSDK agent.Service - managerSDK manager.Service + agentSDK agent.Service } -func New(agentSDK agent.Service, managerSDK manager.Service) *CLI { +func New(agentSDK agent.Service) *CLI { return &CLI{ - agentSDK: agentSDK, - managerSDK: managerSDK, + agentSDK: agentSDK, } } diff --git a/cmd/cli/main.go b/cmd/cli/main.go index 042ce235..52e9dec7 100644 --- a/cmd/cli/main.go +++ b/cmd/cli/main.go @@ -6,26 +6,21 @@ import ( "fmt" "log" "os" - "strings" mglog "github.com/absmach/magistrala/logger" "github.com/spf13/cobra" "github.com/spf13/pflag" "github.com/ultravioletrs/cocos/cli" "github.com/ultravioletrs/cocos/internal/env" - managersvc "github.com/ultravioletrs/cocos/manager" - "github.com/ultravioletrs/cocos/manager/qemu" "github.com/ultravioletrs/cocos/pkg/clients/grpc" "github.com/ultravioletrs/cocos/pkg/clients/grpc/agent" "github.com/ultravioletrs/cocos/pkg/sdk" ) const ( - svcName = "cli" - envPrefixAgentGRPC = "AGENT_GRPC_" - envPrefixManagerGRPC = "MANAGER_GRPC_" - completion = "completion" - envPrefixQemu = "MANAGER_QEMU_" + svcName = "cli" + envPrefixAgentGRPC = "AGENT_GRPC_" + completion = "completion" ) type config struct { @@ -56,22 +51,9 @@ func main() { } defer agentGRPCClient.Close() - qemuCfg := qemu.Config{} - if err := env.Parse(&qemuCfg, env.Options{Prefix: envPrefixQemu}); err != nil { - logger.Error(fmt.Sprintf("failed to load QEMU configuration: %s", err)) - return - } - exe, args, err := qemu.ExecutableAndArgs(qemuCfg) - if err != nil { - logger.Error(fmt.Sprintf("failed to parse QEMU configuration: %s", err)) - return - } - logger.Info(fmt.Sprintf("%s %s", exe, strings.Join(args, " "))) - agentSDK := sdk.NewAgentSDK(logger, agentClient) - managerSDK := managersvc.New(qemuCfg, logger, make(chan *managersvc.ClientStreamMessage)) - cliSVC := cli.New(agentSDK, managerSDK) + cliSVC := cli.New(agentSDK) rootCmd := &cobra.Command{ Use: "cocos-cli [command]", @@ -101,80 +83,17 @@ func main() { }, } - agentCmd := &cobra.Command{ - Use: "agent [command]", - Short: "CLI application for agent Service API", - Run: func(cmd *cobra.Command, args []string) { - fmt.Printf("CLI application for agent Service API\n\n") - fmt.Printf("Usage:\n %s [command]\n\n", cmd.CommandPath()) - fmt.Printf("Available Commands:\n") - - // Filter out "completion" command - availableCommands := make([]*cobra.Command, 0) - for _, subCmd := range cmd.Commands() { - if subCmd.Name() != completion { - availableCommands = append(availableCommands, subCmd) - } - } - - for _, subCmd := range availableCommands { - fmt.Printf(" %-15s%s\n", subCmd.Name(), subCmd.Short) - } - - fmt.Printf("\nFlags:\n") - cmd.Flags().VisitAll(func(flag *pflag.Flag) { - fmt.Printf(" -%s, --%s %s\n", flag.Shorthand, flag.Name, flag.Usage) - }) - fmt.Printf("\nUse \"%s [command] --help\" for more information about a command.\n", cmd.CommandPath()) - }, - } - - managerCmd := &cobra.Command{ - Use: "manager [command]", - Short: "CLI application for manager Service API", - Run: func(cmd *cobra.Command, args []string) { - fmt.Printf("CLI application for manager Service API\n\n") - fmt.Printf("Usage:\n %s [command]\n\n", cmd.CommandPath()) - fmt.Printf("Available Commands:\n") - - // Filter out "completion" command - availableCommands := make([]*cobra.Command, 0) - for _, subCmd := range cmd.Commands() { - if subCmd.Name() != completion { - availableCommands = append(availableCommands, subCmd) - } - } - - for _, subCmd := range availableCommands { - fmt.Printf(" %-15s%s\n", subCmd.Name(), subCmd.Short) - } - - fmt.Printf("\nFlags:\n") - cmd.Flags().VisitAll(func(flag *pflag.Flag) { - fmt.Printf(" -%s, --%s %s\n", flag.Shorthand, flag.Name, flag.Usage) - }) - fmt.Printf("\nUse \"%s [command] --help\" for more information about a command.\n", cmd.CommandPath()) - }, - } - - // Root Commands - rootCmd.AddCommand(agentCmd) - rootCmd.AddCommand(managerCmd) - // Agent Commands - agentCmd.AddCommand(cliSVC.NewAlgorithmsCmd()) - agentCmd.AddCommand(cliSVC.NewDatasetsCmd()) - agentCmd.AddCommand(cliSVC.NewResultsCmd()) + rootCmd.AddCommand(cliSVC.NewAlgorithmsCmd()) + rootCmd.AddCommand(cliSVC.NewDatasetsCmd()) + rootCmd.AddCommand(cliSVC.NewResultsCmd()) attestaionCmd := cliSVC.NewAttestationCmd() - agentCmd.AddCommand(attestaionCmd) + rootCmd.AddCommand(attestaionCmd) // Attestation commands attestaionCmd.AddCommand(cliSVC.NewGetAttestationCmd()) attestaionCmd.AddCommand(cliSVC.NewValidateAttestationValidationCmd()) - // Manager commands - managerCmd.AddCommand(cliSVC.NewRunCmd()) - if err := rootCmd.Execute(); err != nil { logger.Error(fmt.Sprintf("Command execution failed: %s", err)) return diff --git a/cmd/manager/main.go b/cmd/manager/main.go index 36e2a863..a8cf83f9 100644 --- a/cmd/manager/main.go +++ b/cmd/manager/main.go @@ -28,11 +28,9 @@ import ( ) const ( - svcName = "manager" - envPrefixGRPC = "MANAGER_GRPC_" - envPrefixQemu = "MANAGER_QEMU_" - defSvcGRPCPort = "7001" - defSvcHTTPPort = "9021" + svcName = "manager" + envPrefixGRPC = "MANAGER_GRPC_" + envPrefixQemu = "MANAGER_QEMU_" ) type config struct { diff --git a/manager/README.md b/manager/README.md index 959750fe..63ed2a3e 100644 --- a/manager/README.md +++ b/manager/README.md @@ -135,7 +135,7 @@ NB: we set environment variables that we will use in the shell process where we ## Deployment -To start the service outside of the container, execute the following shell script: +To start the service, execute the following shell script (note a server needs to be running see [here](../test/manager-server/README.md)): ```bash # download the latest version of the service @@ -150,6 +150,7 @@ make manager make install # set the environment variables and run the service +MANAGER_GRPC_URL=localhost:7001 MANAGER_LOG_LEVEL=debug \ MANAGER_QEMU_USE_SUDO=false \ MANAGER_QEMU_ENABLE_SEV=false \ @@ -159,6 +160,7 @@ MANAGER_QEMU_ENABLE_SEV=false \ To enable [AMD SEV](https://www.amd.com/en/developer/sev.html) support, start manager like this ```sh +MANAGER_GRPC_URL=localhost:7001 MANAGER_LOG_LEVEL=debug \ MANAGER_QEMU_USE_SUDO=true \ MANAGER_QEMU_ENABLE_SEV=true \ @@ -166,16 +168,6 @@ MANAGER_QEMU_SEV_CBITPOS=51 \ ./build/cocos-manager ``` -### Create QEMU virtual machine (VM) - -To create an instance of VM and run a computation, run - -```sh -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. - ### Verifying VM launch NB: To verify that the manager successfully launched the VM, you need to open two terminals on the same machine. In one terminal, you need to launch `go run main.go` (with the environment variables of choice) and in the other, you can run the verification commands. diff --git a/manager/api/grpc/server.go b/manager/api/grpc/server.go index 943a265f..82f2937a 100644 --- a/manager/api/grpc/server.go +++ b/manager/api/grpc/server.go @@ -3,49 +3,44 @@ package grpc import ( - "context" - "github.com/ultravioletrs/cocos/manager" - "golang.org/x/sync/errgroup" + "google.golang.org/grpc/peer" ) type grpcServer struct { manager.UnimplementedManagerServiceServer - incoming chan *manager.ClientStreamMessage - responses chan *manager.ComputationRunReq - ctx context.Context + incoming chan *manager.ClientStreamMessage + svc Service +} + +type Service interface { + Run(ipAddress string) manager.ComputationRunReq } // NewServer returns new AuthServiceServer instance. -func NewServer(ctx context.Context, incoming chan *manager.ClientStreamMessage, responses chan *manager.ComputationRunReq) manager.ManagerServiceServer { +func NewServer(incoming chan *manager.ClientStreamMessage, svc Service) manager.ManagerServiceServer { return &grpcServer{ - incoming: incoming, - responses: responses, + incoming: incoming, + svc: svc, } } func (s *grpcServer) Process(stream manager.ManagerService_ProcessServer) error { - eg, _ := errgroup.WithContext(s.ctx) - - eg.Go(func() error { - for { - req, err := stream.Recv() - if err != nil { - return err - } - - s.incoming <- req + for { + req, err := stream.Recv() + if err != nil { + return err } - }) - - eg.Go(func() error { - for resp := range s.responses { - if err := stream.Send(resp); err != nil { - return err + if _, ok := req.Message.(*manager.ClientStreamMessage_Whoami); ok { + client, ok := peer.FromContext(stream.Context()) + if ok { + req := s.svc.Run(client.Addr.String()) + if err := stream.Send(&req); err != nil { + return err + } } } - return nil - }) - return eg.Wait() + s.incoming <- req + } } diff --git a/manager/manager.pb.go b/manager/manager.pb.go index fc1b2f10..8e35b888 100644 --- a/manager/manager.pb.go +++ b/manager/manager.pb.go @@ -585,7 +585,6 @@ type AgentConfig struct { ClientCaFile string `protobuf:"bytes,5,opt,name=client_ca_file,json=clientCaFile,proto3" json:"client_ca_file,omitempty"` ServerCaFile string `protobuf:"bytes,6,opt,name=server_ca_file,json=serverCaFile,proto3" json:"server_ca_file,omitempty"` LogLevel string `protobuf:"bytes,7,opt,name=log_level,json=logLevel,proto3" json:"log_level,omitempty"` - InstanceId string `protobuf:"bytes,8,opt,name=instance_id,json=instanceId,proto3" json:"instance_id,omitempty"` } func (x *AgentConfig) Reset() { @@ -669,13 +668,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{ @@ -748,7 +740,7 @@ var file_manager_manager_proto_rawDesc = []byte{ 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, 0xf7, 0x01, 0x0a, 0x0b, 0x41, 0x67, 0x65, 0x6e, 0x74, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, + 0x22, 0xd6, 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, 0x0a, 0x09, 0x63, 0x65, 0x72, 0x74, @@ -761,16 +753,14 @@ var file_manager_manager_proto_rawDesc = []byte{ 0x5f, 0x63, 0x61, 0x5f, 0x66, 0x69, 0x6c, 0x65, 0x18, 0x06, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0c, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x43, 0x61, 0x46, 0x69, 0x6c, 0x65, 0x12, 0x1b, 0x0a, 0x09, 0x6c, 0x6f, 0x67, 0x5f, 0x6c, 0x65, 0x76, 0x65, 0x6c, 0x18, 0x07, 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, 0x08, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0a, - 0x69, 0x6e, 0x73, 0x74, 0x61, 0x6e, 0x63, 0x65, 0x49, 0x64, 0x32, 0x5b, 0x0a, 0x0e, 0x4d, 0x61, - 0x6e, 0x61, 0x67, 0x65, 0x72, 0x53, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x12, 0x49, 0x0a, 0x07, - 0x50, 0x72, 0x6f, 0x63, 0x65, 0x73, 0x73, 0x12, 0x1c, 0x2e, 0x6d, 0x61, 0x6e, 0x61, 0x67, 0x65, - 0x72, 0x2e, 0x43, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x4d, 0x65, - 0x73, 0x73, 0x61, 0x67, 0x65, 0x1a, 0x1a, 0x2e, 0x6d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x2e, - 0x43, 0x6f, 0x6d, 0x70, 0x75, 0x74, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x75, 0x6e, 0x52, 0x65, - 0x71, 0x22, 0x00, 0x28, 0x01, 0x30, 0x01, 0x42, 0x0b, 0x5a, 0x09, 0x2e, 0x2f, 0x6d, 0x61, 0x6e, - 0x61, 0x67, 0x65, 0x72, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, + 0x08, 0x6c, 0x6f, 0x67, 0x4c, 0x65, 0x76, 0x65, 0x6c, 0x32, 0x5b, 0x0a, 0x0e, 0x4d, 0x61, 0x6e, + 0x61, 0x67, 0x65, 0x72, 0x53, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x12, 0x49, 0x0a, 0x07, 0x50, + 0x72, 0x6f, 0x63, 0x65, 0x73, 0x73, 0x12, 0x1c, 0x2e, 0x6d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, + 0x2e, 0x43, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x4d, 0x65, 0x73, + 0x73, 0x61, 0x67, 0x65, 0x1a, 0x1a, 0x2e, 0x6d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x2e, 0x43, + 0x6f, 0x6d, 0x70, 0x75, 0x74, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x75, 0x6e, 0x52, 0x65, 0x71, + 0x22, 0x00, 0x28, 0x01, 0x30, 0x01, 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 4aa6de58..c07dca2b 100644 --- a/manager/manager.proto +++ b/manager/manager.proto @@ -72,5 +72,4 @@ message AgentConfig { string client_ca_file = 5; string server_ca_file = 6; string log_level = 7; - string instance_id = 8; } \ No newline at end of file diff --git a/test/manager-server/README.md b/test/manager-server/README.md new file mode 100644 index 00000000..c6d6b89e --- /dev/null +++ b/test/manager-server/README.md @@ -0,0 +1,19 @@ +# Manager test server +Manager service is a grpc client. It connects to a server and sends a whoAmIRequest. +The server then responds with a run computation request. Once manager service receives the computation request it will launch an agent service in a virtual machine and pass the computation manifest. Agent will then pass logs and events to manager which are forwarded to the server. `main.go` is a sample of how such a server would be implemented. This is a very simple example for testing purposes. + +## Configuration + +The service is configured using the environment variables from the following table. Note that any unset variables will be replaced with their default values. + +| Variable | Description | Default | +| ---------------- | ---------------------------------------- | ------- | +| HOST | Manager service gRPC host | | +| PORT | Manager service gRPC port | 7001 | +| SERVER_CERT | Path to server certificate in pem format | | +| SERVER_KEY | Path to server key in pem format | | + +## Running +```shell +go run main.go +``` diff --git a/test/manager-server/main.go b/test/manager-server/main.go new file mode 100644 index 00000000..9e118ef8 --- /dev/null +++ b/test/manager-server/main.go @@ -0,0 +1,97 @@ +// Copyright (c) Ultraviolet +// SPDX-License-Identifier: Apache-2.0 +package main + +import ( + "context" + "fmt" + "log" + "log/slog" + "os" + + mglog "github.com/absmach/magistrala/logger" + "github.com/ultravioletrs/cocos/internal/env" + "github.com/ultravioletrs/cocos/internal/server" + grpcserver "github.com/ultravioletrs/cocos/internal/server/grpc" + "github.com/ultravioletrs/cocos/manager" + managergrpc "github.com/ultravioletrs/cocos/manager/api/grpc" + "golang.org/x/sync/errgroup" + "google.golang.org/grpc" + "google.golang.org/grpc/reflection" +) + +const ( + svcName = "manager_test_server" + defaultPort = "7001" +) + +type svc struct { + logger *slog.Logger +} + +func (s *svc) Run(ipAdress string) manager.ComputationRunReq { + s.logger.Debug(fmt.Sprintf("received who am on ip address %s", ipAdress)) + return manager.ComputationRunReq{ + Id: "1", + Name: "sample computation", + Description: "sample descrption", + Datasets: []*manager.Dataset{{Id: "1", Provider: "provider1"}}, + Algorithms: []*manager.Algorithm{{Id: "1", Provider: "provider1"}}, + ResultConsumers: []string{"consumer1"}, + AgentConfig: &manager.AgentConfig{ + Port: "7002", + LogLevel: "debug", + }, + } +} + +func main() { + ctx, cancel := context.WithCancel(context.Background()) + g, ctx := errgroup.WithContext(ctx) + incomingChan := make(chan *manager.ClientStreamMessage) + + logger, err := mglog.New(os.Stdout, "debug") + if err != nil { + log.Fatalf(err.Error()) + } + + go func() { + for incoming := range incomingChan { + switch incoming.Message.(type) { + case *manager.ClientStreamMessage_Whoami: + fmt.Println("received whoamI") + case *manager.ClientStreamMessage_RunRes: + fmt.Println("received runRes") + case *manager.ClientStreamMessage_AgentEvent: + fmt.Println("received agent event") + case *manager.ClientStreamMessage_AgentLog: + fmt.Println("received agent log") + } + fmt.Println(incoming.Message) + } + }() + + registerAgentServiceServer := func(srv *grpc.Server) { + reflection.Register(srv) + manager.RegisterManagerServiceServer(srv, managergrpc.NewServer(incomingChan, &svc{logger: logger})) + } + grpcServerConfig := server.Config{Port: defaultPort} + if err := env.Parse(&grpcServerConfig, env.Options{}); err != nil { + logger.Error(fmt.Sprintf("failed to load %s gRPC client configuration : %s", svcName, err)) + return + } + + gs := grpcserver.New(ctx, cancel, svcName, grpcServerConfig, registerAgentServiceServer, logger) + + g.Go(func() error { + return gs.Start() + }) + + g.Go(func() error { + return server.StopHandler(ctx, cancel, logger, svcName, gs) + }) + + if err := g.Wait(); err != nil { + logger.Error(fmt.Sprintf("%s service terminated: %s", svcName, err)) + } +} diff --git a/test/manual/README.md b/test/manual/README.md index a37bb2b8..6dfdae4f 100644 --- a/test/manual/README.md +++ b/test/manual/README.md @@ -16,11 +16,7 @@ pip3 install pandas scikit-learn ### Agent-CLI interaction -In the VM, open a console and start `agent`: - -```sh -AGENT_LOG_LEVEL=info AGENT_GRPC_URL=10.0.2.15:7002 go run cmd/agent/main.go -``` +Agent is started automatically in the VM when launched but requires configuration and manifest to be passed by manager. Alternatively you can pass configuration using this [simplified script](./agent-config/main.go) Open console on the host, and run @@ -28,26 +24,23 @@ Open console on the host, and run export AGENT_GRPC_URL=localhost:7002 export MANAGER_GRPC_URL=localhost:7001 -# Run CLI to provide manifest -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"}' - # Retieve Attestation -go run cmd/cli/main.go agent attestation get '' +go run cmd/cli/main.go attestation get '' # Validate Attestation -go run cmd/cli/main.go agent attestation validate '' '' +go run cmd/cli/main.go attestation validate '' '' # Run the CLI program with algorithm input -go run cmd/cli/main.go agent algo test/manual/algo/lin_reg.py Algorithm1 AlgorithmProvider1 +go run cmd/cli/main.go algo test/manual/algo/lin_reg.py Algorithm1 AlgorithmProvider1 # 2023/09/21 10:43:53 Uploading algorithm binary: test/manual/algo/lin_reg.py # Run the CLI program with dataset input -go run cmd/cli/main.go agent data test/manual/data/iris.csv Dataset1 Provider1 -go run cmd/cli/main.go agent data test/manual/data/iris.csv Dataset2 Provider2 +go run cmd/cli/main.go data test/manual/data/iris.csv Dataset1 Provider1 +go run cmd/cli/main.go data test/manual/data/iris.csv Dataset2 Provider2 # 2023/09/21 10:45:25 Uploading dataset CSV: test/manual/data/iris.csv # Run the CLI program to fetch computation result -go run cmd/cli/main.go agent result Consumer1 +go run cmd/cli/main.go result Consumer1 # 2023/09/21 10:45:39 Retrieving computation result file # 2023/09/21 10:45:40 Computation result retrieved and saved successfully! ``` diff --git a/test/manual/agent-config/main.go b/test/manual/agent-config/main.go new file mode 100644 index 00000000..6289bf93 --- /dev/null +++ b/test/manual/agent-config/main.go @@ -0,0 +1,134 @@ +// Copyright (c) Ultraviolet +// SPDX-License-Identifier: Apache-2.0 + +// Simplified script to pass configs to agent without manager and read logs and events for manager. +// This tool is meant for testing purposes. +package main + +import ( + "encoding/json" + "fmt" + "log" + + "github.com/mdlayher/vsock" +) + +const VsockConfigPort uint32 = 9999 + +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"` +} + +type Computation struct { + ID string `json:"id,omitempty"` + Name string `json:"name,omitempty"` + Description string `json:"description,omitempty"` + Datasets Datasets `json:"datasets,omitempty"` + Algorithms Algorithms `json:"algorithms,omitempty"` + ResultConsumers []string `json:"result_consumers,omitempty"` + AgentConfig AgentConfig `json:"agent_config,omitempty"` +} + +func (d *Datasets) String() string { + dat, err := json.Marshal(d) + if err != nil { + return "" + } + return string(dat) +} + +func (a *Algorithms) String() string { + dat, err := json.Marshal(a) + if err != nil { + return "" + } + return string(dat) +} + +type Dataset struct { + Dataset []byte `json:"-"` + Provider string `json:"provider,omitempty"` + ID string `json:"id,omitempty"` +} + +type Datasets []Dataset + +type Algorithm struct { + Algorithm []byte `json:"-"` + Provider string `json:"provider,omitempty"` + ID string `json:"id,omitempty"` +} + +type Algorithms []Algorithm + +func main() { + l, err := vsock.Listen(9997, nil) + if err != nil { + log.Fatal(err) + } + l2, err := vsock.Listen(9998, nil) + if err != nil { + log.Fatal(err) + } + ac := Computation{ + ID: "123", + Datasets: Datasets{Dataset{ID: "1", Provider: "pr1"}}, + Algorithms: Algorithms{Algorithm{ID: "1", Provider: "pr1"}}, + ResultConsumers: []string{"1"}, + AgentConfig: AgentConfig{ + LogLevel: "debug", + Port: "7002", + }, + } + fmt.Println(SendAgentConfig(3, ac)) + + go func() { + for { + conn, err := l.Accept() + if err != nil { + continue + } + b := make([]byte, 1024) + n, err := conn.Read(b) + if err != nil { + continue + } + conn.Close() + fmt.Println(string(b[:n])) + } + }() + for { + conn, err := l2.Accept() + if err != nil { + continue + } + b := make([]byte, 1024) + n, err := conn.Read(b) + if err != nil { + continue + } + conn.Close() + fmt.Println(string(b[:n])) + } +} + +func SendAgentConfig(cid uint32, ac Computation) error { + conn, err := vsock.Dial(cid, VsockConfigPort, nil) + if err != nil { + return err + } + defer conn.Close() + payload, err := json.Marshal(ac) + if err != nil { + return err + } + if _, err := conn.Write(payload); err != nil { + return err + } + return nil +}