diff --git a/.github/workflows/checkproto.yaml b/.github/workflows/checkproto.yaml index e507fd3e..555a6ddb 100644 --- a/.github/workflows/checkproto.yaml +++ b/.github/workflows/checkproto.yaml @@ -34,7 +34,7 @@ jobs: - name: Set up protoc run: | - PROTOC_VERSION=35.0 + PROTOC_VERSION=35.1 PROTOC_GEN_VERSION=v1.36.11 PROTOC_GRPC_VERSION=v1.6.0 diff --git a/agent/agent.pb.go b/agent/agent.pb.go index 2076b089..8ef7e6a2 100644 --- a/agent/agent.pb.go +++ b/agent/agent.pb.go @@ -4,7 +4,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: // protoc-gen-go v1.36.11 -// protoc v7.35.0 +// protoc v7.35.1 // source: agent/agent.proto package agent diff --git a/agent/agent_grpc.pb.go b/agent/agent_grpc.pb.go index adcac7b1..b9c8e43a 100644 --- a/agent/agent_grpc.pb.go +++ b/agent/agent_grpc.pb.go @@ -4,7 +4,7 @@ // Code generated by protoc-gen-go-grpc. DO NOT EDIT. // versions: // - protoc-gen-go-grpc v1.6.0 -// - protoc v7.35.0 +// - protoc v7.35.1 // source: agent/agent.proto package agent diff --git a/agent/algorithm/algorithm.go b/agent/algorithm/algorithm.go index 109e6713..07354bc5 100644 --- a/agent/algorithm/algorithm.go +++ b/agent/algorithm/algorithm.go @@ -18,11 +18,14 @@ const ( AlgoTypeKey = "algo_type" AlgoArgsKey = "algo_args" - ResultsDir = "results" - DatasetsDir = "datasets" - AlgoWorkingDir = "/cocos" + ResultsDir = "results" + DatasetsDir = "datasets" ) +// AlgoWorkingDir is the base directory used by algorithm runners (e.g. docker) +// to create datasets/results mounts. It is a variable so tests can override it. +var AlgoWorkingDir = "/cocos" + func AlgorithmTypeToContext(ctx context.Context, algoType string) context.Context { return metadata.AppendToOutgoingContext(ctx, AlgoTypeKey, algoType) } diff --git a/agent/algorithm/docker/docker.go b/agent/algorithm/docker/docker.go index 835df042..64fa2fa1 100644 --- a/agent/algorithm/docker/docker.go +++ b/agent/algorithm/docker/docker.go @@ -90,6 +90,16 @@ func (d *docker) Run() error { return fmt.Errorf("could not find image ID") } + datasetsDir := path.Join(algorithm.AlgoWorkingDir, algorithm.DatasetsDir) + resultsDir := path.Join(algorithm.AlgoWorkingDir, algorithm.ResultsDir) + + if err := os.MkdirAll(datasetsDir, 0o755); err != nil { + return fmt.Errorf("could not create datasets directory %s: %v", datasetsDir, err) + } + if err := os.MkdirAll(resultsDir, 0o755); err != nil { + return fmt.Errorf("could not create results directory %s: %v", resultsDir, err) + } + // Create and start the container. respContainer, err := cli.ContainerCreate(ctx, &container.Config{ Image: dockerImageName, diff --git a/agent/cvms/cvms.pb.go b/agent/cvms/cvms.pb.go index f139756c..b1342efc 100644 --- a/agent/cvms/cvms.pb.go +++ b/agent/cvms/cvms.pb.go @@ -4,7 +4,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: // protoc-gen-go v1.36.11 -// protoc v7.35.0 +// protoc v7.35.1 // source: agent/cvms/cvms.proto package cvms diff --git a/agent/cvms/cvms_grpc.pb.go b/agent/cvms/cvms_grpc.pb.go index 61559b68..6b032b7f 100644 --- a/agent/cvms/cvms_grpc.pb.go +++ b/agent/cvms/cvms_grpc.pb.go @@ -4,7 +4,7 @@ // Code generated by protoc-gen-go-grpc. DO NOT EDIT. // versions: // - protoc-gen-go-grpc v1.6.0 -// - protoc v7.35.0 +// - protoc v7.35.1 // source: agent/cvms/cvms.proto package cvms diff --git a/agent/events/events.pb.go b/agent/events/events.pb.go index df9730ef..b267606d 100644 --- a/agent/events/events.pb.go +++ b/agent/events/events.pb.go @@ -4,7 +4,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: // protoc-gen-go v1.36.11 -// protoc v7.35.0 +// protoc v7.35.1 // source: agent/events/events.proto package events diff --git a/agent/log/log.pb.go b/agent/log/log.pb.go index 188c5e28..a7288909 100644 --- a/agent/log/log.pb.go +++ b/agent/log/log.pb.go @@ -4,7 +4,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: // protoc-gen-go v1.36.11 -// protoc v7.35.0 +// protoc v7.35.1 // source: agent/log/log.proto package log diff --git a/agent/log/log_grpc.pb.go b/agent/log/log_grpc.pb.go index 084cd0e1..545d45a4 100644 --- a/agent/log/log_grpc.pb.go +++ b/agent/log/log_grpc.pb.go @@ -4,7 +4,7 @@ // Code generated by protoc-gen-go-grpc. DO NOT EDIT. // versions: // - protoc-gen-go-grpc v1.6.0 -// - protoc v7.35.0 +// - protoc v7.35.1 // source: agent/log/log.proto package log diff --git a/agent/runner/runner.pb.go b/agent/runner/runner.pb.go index f02295ee..8a587284 100644 --- a/agent/runner/runner.pb.go +++ b/agent/runner/runner.pb.go @@ -4,7 +4,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: // protoc-gen-go v1.36.11 -// protoc v7.35.0 +// protoc v7.35.1 // source: agent/runner/runner.proto package runner @@ -26,15 +26,15 @@ const ( ) type RunRequest struct { - state protoimpl.MessageState `protogen:"open.v1"` - ComputationId string `protobuf:"bytes,1,opt,name=computation_id,json=computationId,proto3" json:"computation_id,omitempty"` - AlgoType string `protobuf:"bytes,2,opt,name=algo_type,json=algoType,proto3" json:"algo_type,omitempty"` // "binary", "python", "wasm", "docker" - Algorithm []byte `protobuf:"bytes,3,opt,name=algorithm,proto3" json:"algorithm,omitempty"` // The algorithm binary/script content - Requirements []byte `protobuf:"bytes,4,opt,name=requirements,proto3" json:"requirements,omitempty"` // Python requirements.txt content - Args []string `protobuf:"bytes,5,rep,name=args,proto3" json:"args,omitempty"` - Datasets []*Dataset `protobuf:"bytes,6,rep,name=datasets,proto3" json:"datasets,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + state protoimpl.MessageState `protogen:"open.v1"` + ComputationId string `protobuf:"bytes,1,opt,name=computation_id,json=computationId,proto3" json:"computation_id,omitempty"` + AlgoType string `protobuf:"bytes,2,opt,name=algo_type,json=algoType,proto3" json:"algo_type,omitempty"` // "binary", "python", "wasm", "docker" + AlgorithmPath string `protobuf:"bytes,3,opt,name=algorithm_path,json=algorithmPath,proto3" json:"algorithm_path,omitempty"` // Path to the staged algorithm artifact + RequirementsPath string `protobuf:"bytes,4,opt,name=requirements_path,json=requirementsPath,proto3" json:"requirements_path,omitempty"` // Optional path to staged Python requirements.txt + Args []string `protobuf:"bytes,5,rep,name=args,proto3" json:"args,omitempty"` + Datasets []*Dataset `protobuf:"bytes,6,rep,name=datasets,proto3" json:"datasets,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *RunRequest) Reset() { @@ -81,18 +81,18 @@ func (x *RunRequest) GetAlgoType() string { return "" } -func (x *RunRequest) GetAlgorithm() []byte { +func (x *RunRequest) GetAlgorithmPath() string { if x != nil { - return x.Algorithm + return x.AlgorithmPath } - return nil + return "" } -func (x *RunRequest) GetRequirements() []byte { +func (x *RunRequest) GetRequirementsPath() string { if x != nil { - return x.Requirements + return x.RequirementsPath } - return nil + return "" } func (x *RunRequest) GetArgs() []string { @@ -261,13 +261,13 @@ var File_agent_runner_runner_proto protoreflect.FileDescriptor const file_agent_runner_runner_proto_rawDesc = "" + "\n" + - "\x19agent/runner/runner.proto\x12\x06runner\x1a\x1bgoogle/protobuf/empty.proto\"\xd3\x01\n" + + "\x19agent/runner/runner.proto\x12\x06runner\x1a\x1bgoogle/protobuf/empty.proto\"\xe5\x01\n" + "\n" + "RunRequest\x12%\n" + "\x0ecomputation_id\x18\x01 \x01(\tR\rcomputationId\x12\x1b\n" + - "\talgo_type\x18\x02 \x01(\tR\balgoType\x12\x1c\n" + - "\talgorithm\x18\x03 \x01(\fR\talgorithm\x12\"\n" + - "\frequirements\x18\x04 \x01(\fR\frequirements\x12\x12\n" + + "\talgo_type\x18\x02 \x01(\tR\balgoType\x12%\n" + + "\x0ealgorithm_path\x18\x03 \x01(\tR\ralgorithmPath\x12+\n" + + "\x11requirements_path\x18\x04 \x01(\tR\x10requirementsPath\x12\x12\n" + "\x04args\x18\x05 \x03(\tR\x04args\x12+\n" + "\bdatasets\x18\x06 \x03(\v2\x0f.runner.DatasetR\bdatasets\"9\n" + "\aDataset\x12\x1a\n" + diff --git a/agent/runner/runner.proto b/agent/runner/runner.proto index 6f39a20f..cbb8f3f9 100644 --- a/agent/runner/runner.proto +++ b/agent/runner/runner.proto @@ -17,8 +17,8 @@ service ComputationRunner { message RunRequest { string computation_id = 1; string algo_type = 2; // "binary", "python", "wasm", "docker" - bytes algorithm = 3; // The algorithm binary/script content - bytes requirements = 4; // Python requirements.txt content + string algorithm_path = 3; // Path to the staged algorithm artifact + string requirements_path = 4; // Optional path to staged Python requirements.txt repeated string args = 5; repeated Dataset datasets = 6; } diff --git a/agent/runner/runner_grpc.pb.go b/agent/runner/runner_grpc.pb.go index 8f619bcc..de7631db 100644 --- a/agent/runner/runner_grpc.pb.go +++ b/agent/runner/runner_grpc.pb.go @@ -4,7 +4,7 @@ // Code generated by protoc-gen-go-grpc. DO NOT EDIT. // versions: // - protoc-gen-go-grpc v1.6.0 -// - protoc v7.35.0 +// - protoc v7.35.1 // source: agent/runner/runner.proto package runner diff --git a/agent/runner/service/service.go b/agent/runner/service/service.go index 9d5b874d..4929c245 100644 --- a/agent/runner/service/service.go +++ b/agent/runner/service/service.go @@ -6,8 +6,6 @@ import ( "context" "fmt" "log/slog" - "os" - "path/filepath" "sync" "github.com/ultravioletrs/cocos/agent/algorithm" @@ -20,10 +18,6 @@ import ( "google.golang.org/protobuf/types/known/emptypb" ) -const ( - algoFilePermission = 0o700 -) - var _ pb.ComputationRunnerServer = (*RunnerService)(nil) type RunnerService struct { @@ -58,65 +52,24 @@ func (s *RunnerService) Run(ctx context.Context, req *pb.RunRequest) (*pb.RunRes s.mu.Unlock() }() - currentDir, err := os.Getwd() - if err != nil { - return nil, fmt.Errorf("error getting current directory: %v", err) + if req.AlgorithmPath == "" { + return nil, fmt.Errorf("algorithm path is required") } - // Write Algo File - algoPath := filepath.Join(currentDir, "algo") - f, err := os.Create(algoPath) - if err != nil { - return nil, fmt.Errorf("error creating algorithm file: %v", err) - } - if _, err := f.Write(req.Algorithm); err != nil { - return nil, fmt.Errorf("error writing algorithm to file: %v", err) - } - if err := os.Chmod(algoPath, algoFilePermission); err != nil { - return nil, fmt.Errorf("error changing file permissions: %v", err) - } - if err := f.Close(); err != nil { - return nil, fmt.Errorf("error closing file: %v", err) - } - defer func() { - if err := os.Remove(algoPath); err != nil { - s.logger.Warn("error removing algorithm file", "error", err) - } - }() - var algo algorithm.Algorithm switch req.AlgoType { case string(algorithm.AlgoTypeBin): - algo = binary.NewAlgorithm(s.logger, s.eventSvc, algoPath, req.Args, req.ComputationId) + algo = binary.NewAlgorithm(s.logger, s.eventSvc, req.AlgorithmPath, req.Args, req.ComputationId) case string(algorithm.AlgoTypePython): - var requirementsFile string - if len(req.Requirements) > 0 { - fr, err := os.CreateTemp("", "requirements.txt") - if err != nil { - return nil, fmt.Errorf("error creating requirments file: %v", err) - } - defer func() { - if err := os.Remove(fr.Name()); err != nil { - s.logger.Warn("error removing requirements file", "error", err) - } - }() - if _, err := fr.Write(req.Requirements); err != nil { - return nil, fmt.Errorf("error writing requirements to file: %v", err) - } - if err := fr.Close(); err != nil { - return nil, fmt.Errorf("error closing file: %v", err) - } - requirementsFile = fr.Name() - } // Assuming default python runtime if not specified in request (proto doesn't have runtime field yet) // We can add it or assume. runtime := python.PyRuntime - algo = python.NewAlgorithm(s.logger, s.eventSvc, runtime, requirementsFile, algoPath, req.Args, req.ComputationId) + algo = python.NewAlgorithm(s.logger, s.eventSvc, runtime, req.RequirementsPath, req.AlgorithmPath, req.Args, req.ComputationId) case string(algorithm.AlgoTypeWasm): - algo = wasm.NewAlgorithm(s.logger, s.eventSvc, req.Args, algoPath, req.ComputationId) + algo = wasm.NewAlgorithm(s.logger, s.eventSvc, req.Args, req.AlgorithmPath, req.ComputationId) case string(algorithm.AlgoTypeDocker): - algo = docker.NewAlgorithm(s.logger, s.eventSvc, algoPath, req.ComputationId) + algo = docker.NewAlgorithm(s.logger, s.eventSvc, req.AlgorithmPath, req.ComputationId) default: return nil, fmt.Errorf("unsupported algorithm type: %s", req.AlgoType) } diff --git a/agent/runner/service/service_test.go b/agent/runner/service/service_test.go index 78d1e1a0..ba5ec2cc 100644 --- a/agent/runner/service/service_test.go +++ b/agent/runner/service/service_test.go @@ -8,11 +8,13 @@ import ( "fmt" "log/slog" "os" + "path/filepath" "testing" "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "github.com/ultravioletrs/cocos/agent/algorithm" pb "github.com/ultravioletrs/cocos/agent/runner" ) @@ -30,6 +32,15 @@ func (m *MockEventService) SendEvent(cmpID, event, status string, details json.R }) } +func writeRunnerTestFile(t *testing.T, dir, name string, data []byte, mode os.FileMode) string { + t.Helper() + + path := filepath.Join(dir, name) + require.NoError(t, os.WriteFile(path, data, mode)) + + return path +} + // TestNewRunnerService tests the creation of a new runner service. func TestNewRunnerService(t *testing.T) { logger := slog.New(slog.NewTextHandler(os.Stdout, nil)) @@ -48,6 +59,7 @@ func TestRunWithBinaryAlgorithm(t *testing.T) { tmpDir := t.TempDir() require.NoError(t, os.Chdir(tmpDir)) defer func() { require.NoError(t, os.Chdir(origDir)) }() + algoPath := writeRunnerTestFile(t, tmpDir, "algo", []byte("#!/bin/bash\necho 'test'"), 0o700) logger := slog.New(slog.NewTextHandler(os.Stdout, nil)) eventSvc := &MockEventService{} @@ -56,7 +68,7 @@ func TestRunWithBinaryAlgorithm(t *testing.T) { req := &pb.RunRequest{ ComputationId: "test-1", AlgoType: "bin", - Algorithm: []byte("#!/bin/bash\necho 'test'"), + AlgorithmPath: algoPath, Args: []string{"arg1", "arg2"}, } @@ -72,13 +84,16 @@ func TestRunWithPythonAlgorithm(t *testing.T) { logger := slog.New(slog.NewTextHandler(os.Stdout, nil)) eventSvc := &MockEventService{} rs := New(logger, eventSvc) + tmpDir := t.TempDir() + algoPath := writeRunnerTestFile(t, tmpDir, "algo.py", []byte("print('hello')"), 0o600) + requirementsPath := writeRunnerTestFile(t, tmpDir, "requirements.txt", []byte("numpy==2.2.0"), 0o600) req := &pb.RunRequest{ - ComputationId: "test-python", - AlgoType: "python", - Algorithm: []byte("print('hello')"), - Args: []string{}, - Requirements: []byte("numpy==2.2.0"), + ComputationId: "test-python", + AlgoType: "python", + AlgorithmPath: algoPath, + Args: []string{}, + RequirementsPath: requirementsPath, } resp, err := rs.Run(context.Background(), req) @@ -86,9 +101,6 @@ func TestRunWithPythonAlgorithm(t *testing.T) { require.NotNil(t, resp) assert.Empty(t, resp.Error) assert.Equal(t, "test-python", resp.ComputationId) - t.Cleanup(func() { - _ = os.Remove("algo") - }) } // TestRunWithPythonAlgorithmNoRequirements tests running Python without requirements. @@ -96,11 +108,13 @@ func TestRunWithPythonAlgorithmNoRequirements(t *testing.T) { logger := slog.New(slog.NewTextHandler(os.Stdout, nil)) eventSvc := &MockEventService{} rs := New(logger, eventSvc) + tmpDir := t.TempDir() + algoPath := writeRunnerTestFile(t, tmpDir, "algo.py", []byte("print('hello')"), 0o600) req := &pb.RunRequest{ ComputationId: "test-python-noreq", AlgoType: "python", - Algorithm: []byte("print('hello')"), + AlgorithmPath: algoPath, Args: []string{}, } @@ -109,9 +123,6 @@ func TestRunWithPythonAlgorithmNoRequirements(t *testing.T) { require.NotNil(t, resp) assert.Empty(t, resp.Error) assert.Equal(t, "test-python-noreq", resp.ComputationId) - t.Cleanup(func() { - _ = os.Remove("algo") - }) } // TestRunWithWasmAlgorithm tests running a WASM algorithm. @@ -119,11 +130,13 @@ func TestRunWithWasmAlgorithm(t *testing.T) { logger := slog.New(slog.NewTextHandler(os.Stdout, nil)) eventSvc := &MockEventService{} rs := New(logger, eventSvc) + tmpDir := t.TempDir() + algoPath := writeRunnerTestFile(t, tmpDir, "algo.wasm", []byte{0x00, 0x61, 0x73, 0x6d}, 0o600) req := &pb.RunRequest{ ComputationId: "test-wasm", AlgoType: "wasm", - Algorithm: []byte{0x00, 0x61, 0x73, 0x6d}, + AlgorithmPath: algoPath, Args: []string{}, } @@ -135,9 +148,6 @@ func TestRunWithWasmAlgorithm(t *testing.T) { t.Skip("wasmedge not found, skipping test") } assert.Equal(t, "test-wasm", resp.ComputationId) - t.Cleanup(func() { - _ = os.Remove("algo") - }) } // TestRunWithDockerAlgorithm tests running a Docker algorithm. @@ -145,11 +155,17 @@ func TestRunWithDockerAlgorithm(t *testing.T) { logger := slog.New(slog.NewTextHandler(os.Stdout, nil)) eventSvc := &MockEventService{} rs := New(logger, eventSvc) + tmpDir := t.TempDir() + algoPath := writeRunnerTestFile(t, tmpDir, "Dockerfile", []byte("FROM ubuntu:latest\nRUN echo 'test'"), 0o600) + + origWorkingDir := algorithm.AlgoWorkingDir + algorithm.AlgoWorkingDir = tmpDir + t.Cleanup(func() { algorithm.AlgoWorkingDir = origWorkingDir }) req := &pb.RunRequest{ ComputationId: "test-docker", AlgoType: "docker", - Algorithm: []byte("FROM ubuntu:latest\nRUN echo 'test'"), + AlgorithmPath: algoPath, Args: []string{}, } @@ -161,9 +177,6 @@ func TestRunWithDockerAlgorithm(t *testing.T) { t.Skip("Docker issue, skipping test") } assert.Equal(t, "test-docker", resp.ComputationId) - t.Cleanup(func() { - _ = os.Remove("algo") - }) } // TestRunWithUnsupportedAlgorithmType tests running with unsupported algorithm type. @@ -175,7 +188,7 @@ func TestRunWithUnsupportedAlgorithmType(t *testing.T) { req := &pb.RunRequest{ ComputationId: "test-unsupported", AlgoType: "unsupported", - Algorithm: []byte("test"), + AlgorithmPath: "/tmp/test", Args: []string{}, } @@ -189,12 +202,14 @@ func TestRunAlreadyRunning(t *testing.T) { logger := slog.New(slog.NewTextHandler(os.Stdout, nil)) eventSvc := &MockEventService{} rs := New(logger, eventSvc) + tmpDir := t.TempDir() + algoPath := writeRunnerTestFile(t, tmpDir, "algo", []byte("#!/bin/bash\nsleep 30"), 0o700) // Use a long-running bash script req := &pb.RunRequest{ ComputationId: "test-running", AlgoType: "bin", - Algorithm: []byte("#!/bin/bash\nsleep 30"), + AlgorithmPath: algoPath, Args: []string{}, } @@ -211,9 +226,6 @@ func TestRunAlreadyRunning(t *testing.T) { require.NoError(t, err) require.NotNil(t, resp) assert.Equal(t, "computation already running", resp.Error) - t.Cleanup(func() { - _ = os.Remove("algo") - }) } // TestStopWhenRunning tests stopping a running computation. @@ -221,11 +233,13 @@ func TestStopWhenRunning(t *testing.T) { logger := slog.New(slog.NewTextHandler(os.Stdout, nil)) eventSvc := &MockEventService{} rs := New(logger, eventSvc) + tmpDir := t.TempDir() + algoPath := writeRunnerTestFile(t, tmpDir, "algo", []byte("#!/bin/bash\nsleep 10"), 0o700) req := &pb.RunRequest{ ComputationId: "test-stop", AlgoType: "bin", - Algorithm: []byte("#!/bin/bash\nsleep 10"), + AlgorithmPath: algoPath, Args: []string{}, } @@ -243,9 +257,6 @@ func TestStopWhenRunning(t *testing.T) { stopResp, err := rs.Stop(context.Background(), stopReq) require.NoError(t, err) require.NotNil(t, stopResp) - t.Cleanup(func() { - _ = os.Remove("algo") - }) } // TestRunErrors tests error paths in Run. @@ -255,59 +266,27 @@ func TestRunErrors(t *testing.T) { rs := New(logger, eventSvc) t.Run("create algo file failure", func(t *testing.T) { - // Create a directory named "algo" to make os.Create("algo") fail - err := os.Mkdir("algo", 0o755) - require.NoError(t, err) - defer os.RemoveAll("algo") - + var err error req := &pb.RunRequest{ ComputationId: "test-err", AlgoType: "bin", - Algorithm: []byte("test"), + AlgorithmPath: "", } _, err = rs.Run(context.Background(), req) assert.Error(t, err) - assert.Contains(t, err.Error(), "error creating algorithm file") - }) - - t.Run("getwd failure", func(t *testing.T) { - origDir, _ := os.Getwd() - tmpDir := t.TempDir() - err := os.Chdir(tmpDir) - require.NoError(t, err) - - // Remove the current working directory to trigger Getwd failure - err = os.RemoveAll(tmpDir) - require.NoError(t, err) - - req := &pb.RunRequest{ - ComputationId: "test-err-getwd", - AlgoType: "bin", - Algorithm: []byte("test"), - } - _, err = rs.Run(context.Background(), req) - assert.Error(t, err) - assert.Contains(t, err.Error(), "error getting current directory") - - // Restore working directory - _ = os.Chdir(origDir) + assert.Contains(t, err.Error(), "algorithm path is required") }) t.Run("requirements file creation failure", func(t *testing.T) { - // This one is harder because it uses os.CreateTemp("", "requirements.txt") - // We can't easily make this fail without reaching into the system's temp dir. - // Skipping for now as it's a very unlikely edge case. + // Requirements are now staged by the agent, so the runner no longer creates temp files. }) t.Run("chmod failure", func(t *testing.T) { - // We can't easily mock os.Chmod, but we can try to make the file unmodifiable - // On Linux, we can set the immutable attribute, but that requires root. - // Alternatively, we can try to use a directory with permissions that prevent chmod? - // No, chmod usually works if you own the file. + // Permission management is now the agent's responsibility during staging. }) t.Run("write algorithm failure", func(t *testing.T) { - // This is also hard without mocking os.File.Write or reaching internal limits. + // Write failures are now handled by the agent before invoking the runner. }) } @@ -316,11 +295,13 @@ func TestConcurrentRun(t *testing.T) { logger := slog.New(slog.NewTextHandler(os.Stdout, nil)) eventSvc := &MockEventService{} rs := New(logger, eventSvc) + tmpDir := t.TempDir() + algoPath := writeRunnerTestFile(t, tmpDir, "algo", []byte("#!/bin/bash\nsleep 15"), 0o700) req := &pb.RunRequest{ ComputationId: "test-concurrent", AlgoType: "bin", - Algorithm: []byte("#!/bin/bash\nsleep 15"), + AlgorithmPath: algoPath, Args: []string{}, } @@ -336,9 +317,6 @@ func TestConcurrentRun(t *testing.T) { resp2, err := rs.Run(context.Background(), req) require.NoError(t, err) assert.Equal(t, "computation already running", resp2.Error) - t.Cleanup(func() { - _ = os.Remove("algo") - }) } // TestRunWithMultipleArgs tests running with multiple arguments. @@ -346,11 +324,13 @@ func TestRunWithMultipleArgs(t *testing.T) { logger := slog.New(slog.NewTextHandler(os.Stdout, nil)) eventSvc := &MockEventService{} rs := New(logger, eventSvc) + tmpDir := t.TempDir() + algoPath := writeRunnerTestFile(t, tmpDir, "algo", []byte("#!/bin/bash\necho $@"), 0o700) req := &pb.RunRequest{ ComputationId: "test-multi-args", AlgoType: "bin", - Algorithm: []byte("#!/bin/bash\necho $@"), + AlgorithmPath: algoPath, Args: []string{"arg1", "arg2", "arg3", "arg4"}, } @@ -359,9 +339,6 @@ func TestRunWithMultipleArgs(t *testing.T) { require.NotNil(t, resp) assert.Empty(t, resp.Error) assert.Equal(t, "test-multi-args", resp.ComputationId) - t.Cleanup(func() { - _ = os.Remove("algo") - }) } func TestStopFailure(t *testing.T) { diff --git a/agent/service.go b/agent/service.go index 184ff6ce..a13e3de5 100644 --- a/agent/service.go +++ b/agent/service.go @@ -82,7 +82,9 @@ const ( ) const ( - algoFilePermission = 0o700 + algoFilePermission = 0o700 + algoFileName = "algo" + requirementsFileName = "requirements.txt" ) var ( @@ -113,6 +115,30 @@ func ensureDir(path string, mode os.FileMode) error { return nil } +func writeFile(path string, data []byte, mode os.FileMode) error { + if err := os.WriteFile(path, data, mode); err != nil { + return fmt.Errorf("writing file %q: %w", path, err) + } + + return nil +} + +func stageRequirementsFile(workDir string, requirements []byte) (string, error) { + requirementsPath := filepath.Join(workDir, requirementsFileName) + if len(requirements) == 0 { + if err := os.Remove(requirementsPath); err != nil && !os.IsNotExist(err) { + return "", fmt.Errorf("removing stale requirements file: %w", err) + } + return "", nil + } + + if err := writeFile(requirementsPath, requirements, 0o600); err != nil { + return "", err + } + + return requirementsPath, nil +} + var ( // ErrMalformedEntity indicates malformed entity specification (e.g. // invalid username or password). @@ -364,10 +390,14 @@ func (as *agentService) StopComputation(ctx context.Context) error { return fmt.Errorf("error removing results directory: %v", err) } - if err := os.Remove("algo"); err != nil && !os.IsNotExist(err) { + if err := os.Remove(algoFileName); err != nil && !os.IsNotExist(err) { as.logger.Warn("error removing algorithm file", "error", err) } + if err := os.Remove(requirementsFileName); err != nil && !os.IsNotExist(err) { + as.logger.Warn("error removing requirements file", "error", err) + } + as.sm.Reset(Idle) as.computation = Computation{} @@ -448,9 +478,9 @@ func (as *agentService) downloadAlgorithmIfRemote(state statemachine.State) { } // Write algorithm to file - currentDir, err := os.Getwd() - if err != nil { - as.runError = fmt.Errorf("error getting current directory: %w", err) + currentDir, getwdErr := os.Getwd() + if getwdErr != nil { + as.runError = fmt.Errorf("error getting current directory: %w", getwdErr) as.logger.Error(as.runError.Error()) as.sm.SendEvent(RunFailed) return @@ -504,7 +534,14 @@ func (as *agentService) downloadAlgorithmIfRemote(state statemachine.State) { } as.algoReceived = true - as.algoRequirements = res.Requirements // Store requirements for installation + as.algoRequirements = res.Requirements + + if _, err := stageRequirementsFile(currentDir, as.algoRequirements); err != nil { + as.runError = fmt.Errorf("error staging requirements file: %w", err) + as.logger.Error(as.runError.Error()) + as.sm.SendEvent(RunFailed) + return + } // The initramfs may have already provisioned /cocos/datasets. if err := ensureDir(algorithm.DatasetsDir, 0o755); err != nil { @@ -1053,21 +1090,9 @@ func (as *agentService) Algo(ctx context.Context, algo Algorithm) error { return fmt.Errorf("error getting current directory: %v", err) } - f, err := os.Create(filepath.Join(currentDir, "algo")) - if err != nil { - return fmt.Errorf("error creating algorithm file: %v", err) - } - - if _, err := f.Write(algoData); err != nil { - return fmt.Errorf("error writing algorithm to file: %v", err) - } - - if err := os.Chmod(f.Name(), algoFilePermission); err != nil { - return fmt.Errorf("error changing file permissions: %v", err) - } - - if err := f.Close(); err != nil { - return fmt.Errorf("error closing file: %v", err) + algoPath := filepath.Join(currentDir, algoFileName) + if err := writeFile(algoPath, algoData, algoFilePermission); err != nil { + return fmt.Errorf("error staging algorithm file: %w", err) } algoType := algorithm.AlgorithmTypeFromContext(ctx) @@ -1079,7 +1104,13 @@ func (as *agentService) Algo(ctx context.Context, algo Algorithm) error { as.algoType = algoType as.algoArgs = args - as.algoRequirements = algo.Requirements + if len(as.algoRequirements) == 0 { + as.algoRequirements = algo.Requirements + } + + if _, err := stageRequirementsFile(currentDir, as.algoRequirements); err != nil { + return fmt.Errorf("error staging requirements file: %w", err) + } as.algoReceived = true if err := ensureDir(algorithm.DatasetsDir, 0o755); err != nil { @@ -1276,9 +1307,9 @@ func (as *agentService) runComputation(state statemachine.State) { } }() - // Read algo file currentDir, _ := os.Getwd() - algoFile := filepath.Join(currentDir, "algo") + algoFile := filepath.Join(currentDir, algoFileName) + requirementsFile := filepath.Join(currentDir, requirementsFileName) defer func() { if err := os.RemoveAll(algorithm.ResultsDir); err != nil { @@ -1290,6 +1321,9 @@ func (as *agentService) runComputation(state statemachine.State) { if err := os.Remove(algoFile); err != nil && !os.IsNotExist(err) { as.logger.Warn(fmt.Sprintf("error removing algorithm file: %s", err.Error())) } + if err := os.Remove(requirementsFile); err != nil && !os.IsNotExist(err) { + as.logger.Warn(fmt.Sprintf("error removing requirements file: %s", err.Error())) + } }() if err := ensureDir(algorithm.ResultsDir, 0o755); err != nil { @@ -1301,10 +1335,21 @@ func (as *agentService) runComputation(state statemachine.State) { return } - algoBytes, err := os.ReadFile(algoFile) - if err != nil { + if _, err := os.Stat(algoFile); err != nil { as.mu.Lock() - as.runError = fmt.Errorf("failed to read algo file: %w", err) + as.runError = fmt.Errorf("failed to stat algo file: %w", err) + as.mu.Unlock() + as.logger.Warn(as.runError.Error()) + as.publishEvent(Failed.String())(state) + return + } + + requirementsPath := "" + if _, err := os.Stat(requirementsFile); err == nil { + requirementsPath = requirementsFile + } else if !os.IsNotExist(err) { + as.mu.Lock() + as.runError = fmt.Errorf("failed to stat requirements file: %w", err) as.mu.Unlock() as.logger.Warn(as.runError.Error()) as.publishEvent(Failed.String())(state) @@ -1315,11 +1360,11 @@ func (as *agentService) runComputation(state statemachine.State) { // Call Runner resp, err := as.runnerClient.Run(context.Background(), &runnerpb.RunRequest{ - ComputationId: as.computation.ID, - AlgoType: as.algoType, - Algorithm: algoBytes, - Requirements: as.algoRequirements, - Args: as.algoArgs, + ComputationId: as.computation.ID, + AlgoType: as.algoType, + AlgorithmPath: algoFile, + RequirementsPath: requirementsPath, + Args: as.algoArgs, // Datasets implicit on shared FS }) if err != nil { diff --git a/agent/service_test.go b/agent/service_test.go index 6c01876f..374a8b14 100644 --- a/agent/service_test.go +++ b/agent/service_test.go @@ -1014,7 +1014,7 @@ func TestRunComputation(t *testing.T) { svc.runComputation(Running) assert.Error(t, svc.runError) - assert.Contains(t, svc.runError.Error(), "failed to read algo file") + assert.Contains(t, svc.runError.Error(), "failed to stat algo file") sm.AssertExpectations(t) }) @@ -1026,7 +1026,9 @@ func TestRunComputation(t *testing.T) { require.NoError(t, os.WriteFile("algo", []byte("#!/bin/sh\necho ok\n"), 0o755)) runnerCli := new(runnermocks.Client) - runnerCli.On("Run", mock.Anything, mock.Anything).Return((*runnerpb.RunResponse)(nil), fmt.Errorf("runner unavailable")) + runnerCli.On("Run", mock.Anything, mock.MatchedBy(func(req *runnerpb.RunRequest) bool { + return req != nil && filepath.Base(req.AlgorithmPath) == algoFileName && req.RequirementsPath == "" && len(req.Args) == 0 + })).Return((*runnerpb.RunResponse)(nil), fmt.Errorf("runner unavailable")) eventsSvc := new(mocks.Service) eventsSvc.EXPECT().SendEvent(mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return().Maybe() diff --git a/cmd/agent/main.go b/cmd/agent/main.go index 7ff48504..59bd21df 100644 --- a/cmd/agent/main.go +++ b/cmd/agent/main.go @@ -15,6 +15,7 @@ import ( "os" "os/signal" "syscall" + "time" "github.com/absmach/certs/sdk" mglog "github.com/absmach/magistrala/logger" @@ -274,7 +275,15 @@ func main() { }) g.Go(func() error { - return mc.Process(ctx, cancel) + for { + if err := mc.Process(ctx, cancel); err != nil { + if ctx.Err() != nil { + return ctx.Err() + } + logger.Warn("CVMS client process connection loop exited, retrying in 2s...", "error", err) + time.Sleep(2 * time.Second) + } + } }) attest, certSerialNumber, err := attestationFromCert(ctx, cvmGrpcConfig.ClientCert, svc, ccPlatform) diff --git a/hal/disk/configs/cocos_defconfig b/hal/disk/configs/cocos_defconfig index 80e505ed..56bce129 100644 --- a/hal/disk/configs/cocos_defconfig +++ b/hal/disk/configs/cocos_defconfig @@ -38,9 +38,9 @@ BR2_ROOTFS_POST_IMAGE_SCRIPT="$(BR2_EXTERNAL_COCOS_PATH)/board/cocos/post-image. BR2_ROOTFS_POST_SCRIPT_ARGS="$(BR2_DEFCONFIG)" # Linux headers same as kernel -BR2_PACKAGE_HOST_LINUX_HEADERS_CUSTOM_6_11=y +BR2_PACKAGE_HOST_LINUX_HEADERS_CUSTOM_7_0=y BR2_TOOLCHAIN_HEADERS_LATEST=y -BR2_TOOLCHAIN_HEADERS_AT_LEAST="6.11-rc7" +BR2_TOOLCHAIN_HEADERS_AT_LEAST="7.0" # Kernel BR2_LINUX_KERNEL=y diff --git a/hal/linux/board/cocos/linux.config b/hal/linux/board/cocos/linux.config index 37d075df..f19fdaf4 100644 --- a/hal/linux/board/cocos/linux.config +++ b/hal/linux/board/cocos/linux.config @@ -1,8 +1,18 @@ CONFIG_SYSVIPC=y CONFIG_CGROUPS=y +CONFIG_CGROUP_DEVICE=y +CONFIG_CGROUP_PIDS=y +CONFIG_MEMCG=y +CONFIG_BLK_CGROUP=y CONFIG_MODULES=y CONFIG_MODULE_UNLOAD=y CONFIG_SMP=y +CONFIG_NAMESPACES=y +CONFIG_UTS_NS=y +CONFIG_IPC_NS=y +CONFIG_USER_NS=y +CONFIG_PID_NS=y +CONFIG_NET_NS=y CONFIG_HYPERVISOR_GUEST=y CONFIG_PARAVIRT=y CONFIG_NET=y @@ -14,12 +24,17 @@ CONFIG_PCI=y CONFIG_DEVTMPFS=y CONFIG_DEVTMPFS_MOUNT=y CONFIG_VIRTIO_BLK=y +CONFIG_BLK_DEV_LOOP=y +CONFIG_BLK_DEV_LOOP_MIN_COUNT=8 CONFIG_BLK_DEV_SD=y CONFIG_SCSI_VIRTIO=y CONFIG_ATA=y CONFIG_ATA_PIIX=y CONFIG_NETDEVICES=y CONFIG_VIRTIO_NET=y +CONFIG_VETH=y +CONFIG_BRIDGE=y +CONFIG_BRIDGE_NETFILTER=y CONFIG_NE2K_PCI=y CONFIG_8139CP=y # CONFIG_WLAN is not set @@ -34,15 +49,43 @@ CONFIG_VIRTIO_INPUT=y CONFIG_VIRTIO_MMIO=y CONFIG_VIRTIO_MMIO_CMDLINE_DEVICES=y CONFIG_EXT4_FS=y +CONFIG_OVERLAY_FS=y CONFIG_AUTOFS4_FS=y CONFIG_TMPFS=y CONFIG_TMPFS_POSIX_ACL=y +CONFIG_FHANDLE=y +CONFIG_INOTIFY_USER=y +CONFIG_SIGNALFD=y +CONFIG_TIMERFD=y +CONFIG_EPOLL=y +CONFIG_POSIX_MQUEUE=y +CONFIG_POSIX_MQUEUE_SYSCTL=y CONFIG_UNWINDER_FRAME_POINTER=y CONFIG_VSOCKETS=y CONFIG_VIRTIO_VSOCKETS=y +CONFIG_NETFILTER=y +CONFIG_NETFILTER_ADVANCED=y +CONFIG_NETFILTER_XTABLES_LEGACY=y +CONFIG_NF_CONNTRACK=y +CONFIG_NF_CONNTRACK_MARK=y +CONFIG_NF_NAT=y +CONFIG_NF_NAT_MASQUERADE=y +CONFIG_NETFILTER_XTABLES=y +CONFIG_NETFILTER_XT_NAT=y +CONFIG_NETFILTER_XT_TARGET_MASQUERADE=y +CONFIG_NF_DEFRAG_IPV4=y +CONFIG_IP_NF_IPTABLES_LEGACY=y +CONFIG_IP_NF_IPTABLES=y +CONFIG_IP_NF_FILTER=y +CONFIG_IP_NF_NAT=y +CONFIG_IP_NF_TARGET_MASQUERADE=y +CONFIG_NETFILTER_XT_MATCH_ADDRTYPE=y +CONFIG_NETFILTER_XT_MATCH_CONNTRACK=y CONFIG_NF_TABLES=y CONFIG_BPF_SYSCALL=y CONFIG_CGROUP_BPF=y +CONFIG_SECCOMP=y +CONFIG_SECCOMP_FILTER=y ### # AMD SEV-SNP diff --git a/hal/linux/configs/cocos_defconfig b/hal/linux/configs/cocos_defconfig index 7bf91c50..d03b6070 100644 --- a/hal/linux/configs/cocos_defconfig +++ b/hal/linux/configs/cocos_defconfig @@ -25,9 +25,9 @@ BR2_ROOTFS_POST_IMAGE_SCRIPT="$(BR2_EXTERNAL_COCOS_PATH)/board/cocos/post-image. BR2_ROOTFS_POST_SCRIPT_ARGS="$(BR2_DEFCONFIG)" # Linux headers same as kernel -BR2_PACKAGE_HOST_LINUX_HEADERS_CUSTOM_6_11=y +BR2_PACKAGE_HOST_LINUX_HEADERS_CUSTOM_7_0=y BR2_TOOLCHAIN_HEADERS_LATEST=y -BR2_TOOLCHAIN_HEADERS_AT_LEAST="6.11-rc7" +BR2_TOOLCHAIN_HEADERS_AT_LEAST="7.0" # Kernel BR2_LINUX_KERNEL=y diff --git a/hal/linux/package/agent/agent.mk b/hal/linux/package/agent/agent.mk index 66a9eb87..aaab99e2 100644 --- a/hal/linux/package/agent/agent.mk +++ b/hal/linux/package/agent/agent.mk @@ -22,6 +22,12 @@ define AGENT_INSTALL_INIT_SYSTEMD $(INSTALL) -D -m 0640 $(@D)/init/systemd/cocos-agent.service $(TARGET_DIR)/usr/lib/systemd/system/cocos-agent.service $(INSTALL) -D -m 0750 $(@D)/init/systemd/agent_setup.sh $(TARGET_DIR)/cocos_init/agent_setup.sh $(INSTALL) -D -m 0750 $(@D)/init/systemd/agent_start_script.sh $(TARGET_DIR)/cocos_init/agent_start_script.sh + mkdir -p $(TARGET_DIR)/usr/lib/systemd/system/multi-user.target.wants + ln -sf ../docker.service $(TARGET_DIR)/usr/lib/systemd/system/multi-user.target.wants/docker.service + ln -sf ../log-forwarder.service $(TARGET_DIR)/usr/lib/systemd/system/multi-user.target.wants/log-forwarder.service + ln -sf ../computation-runner.service $(TARGET_DIR)/usr/lib/systemd/system/multi-user.target.wants/computation-runner.service + ln -sf ../egress-proxy.service $(TARGET_DIR)/usr/lib/systemd/system/multi-user.target.wants/egress-proxy.service + ln -sf ../cocos-agent.service $(TARGET_DIR)/usr/lib/systemd/system/multi-user.target.wants/cocos-agent.service endef $(eval $(generic-package)) diff --git a/init/systemd/agent_setup.sh b/init/systemd/agent_setup.sh index f8404a7c..db3aa8c3 100644 --- a/init/systemd/agent_setup.sh +++ b/init/systemd/agent_setup.sh @@ -1,6 +1,24 @@ #!/bin/sh WORK_DIR="/cocos" +DOCKER_OVERRIDE_DIR="/etc/systemd/system/docker.service.d" +DOCKER_OVERRIDE_FILE="$DOCKER_OVERRIDE_DIR/override.conf" + +# Docker runs on a writable bind mount while the rootfs itself stays ephemeral, +# so force RAM-disk mode to avoid pivot_root failures inside containers. +mkdir -p "$DOCKER_OVERRIDE_DIR" +if ! grep -qs '^Environment=DOCKER_RAMDISK=true$' "$DOCKER_OVERRIDE_FILE"; then + tee "$DOCKER_OVERRIDE_FILE" > /dev/null <