diff --git a/agent/README.md b/agent/README.md index a4a0cb15..5c999d4a 100644 --- a/agent/README.md +++ b/agent/README.md @@ -15,7 +15,7 @@ The agent supports two modes: ### All Nodes - WireGuard (`wg` and `wg-quick` commands) -- Podman +- Podman 4.8 or newer (required for command execution cleanup) - BuildKit + buildctl - Railpack @@ -226,7 +226,7 @@ WantedBy=multi-user.target ``` `KillMode=process` ensures only the agent process is killed on restart, not container processes. -The rootful Podman API socket at `/run/podman/podman.sock` is required for container metrics collection. +The rootful Podman API socket at `/run/podman/podman.sock` is required for container metrics collection and command execution. Command timeouts force-remove the foreground exec session; deliberately backgrounded descendants are not guaranteed to be removed. ```bash sudo systemctl daemon-reload diff --git a/agent/internal/agent/handlers.go b/agent/internal/agent/handlers.go index 0b8aeb29..1ffbf8a8 100644 --- a/agent/internal/agent/handlers.go +++ b/agent/internal/agent/handlers.go @@ -10,6 +10,7 @@ import ( "os/exec" "path/filepath" "time" + "unicode/utf8" "techulus/cloud-agent/internal/build" "techulus/cloud-agent/internal/container" @@ -19,6 +20,23 @@ import ( "techulus/cloud-agent/internal/registryauth" ) +func (a *Agent) ProcessCommand(item agenthttp.WorkQueueItem) (container.CommandResult, error) { + var payload struct { + CommandRunID string `json:"commandRunId"` + ServiceID string `json:"serviceId"` + DeploymentID string `json:"deploymentId"` + ContainerID string `json:"containerId"` + Command string `json:"command"` + } + if err := json.Unmarshal([]byte(item.Payload), &payload); err != nil { + return container.CommandResult{}, fmt.Errorf("failed to parse command payload: %w", err) + } + if payload.CommandRunID != item.ID || payload.ServiceID == "" || payload.DeploymentID == "" || payload.ContainerID == "" || payload.Command == "" || utf8.RuneCountInString(payload.Command) > 4096 { + return container.CommandResult{}, fmt.Errorf("invalid command payload") + } + return container.ExecCommand(payload.ContainerID, payload.ServiceID, payload.DeploymentID, payload.Command) +} + func (a *Agent) ProcessRestart(item agenthttp.WorkQueueItem) error { var payload struct { DeploymentID string `json:"deploymentId"` diff --git a/agent/internal/agent/handlers_command_test.go b/agent/internal/agent/handlers_command_test.go new file mode 100644 index 00000000..f7e3efcc --- /dev/null +++ b/agent/internal/agent/handlers_command_test.go @@ -0,0 +1,26 @@ +package agent + +import ( + "testing" + + agenthttp "techulus/cloud-agent/internal/http" +) + +func TestProcessCommandRequiresOwnershipIdentifiers(t *testing.T) { + tests := []struct { + name string + payload string + }{ + {"service ID", `{"commandRunId":"run","deploymentId":"deployment","containerId":"container","command":"true"}`}, + {"deployment ID", `{"commandRunId":"run","serviceId":"service","containerId":"container","command":"true"}`}, + {"container ID", `{"commandRunId":"run","serviceId":"service","deploymentId":"deployment","command":"true"}`}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + _, err := (&Agent{}).ProcessCommand(agenthttp.WorkQueueItem{ID: "run", Payload: tt.payload}) + if err == nil || err.Error() != "invalid command payload" { + t.Fatalf("expected invalid payload for missing %s, got %v", tt.name, err) + } + }) + } +} diff --git a/agent/internal/agent/workqueue.go b/agent/internal/agent/workqueue.go index 5791a21f..5f12116a 100644 --- a/agent/internal/agent/workqueue.go +++ b/agent/internal/agent/workqueue.go @@ -7,6 +7,7 @@ import ( "os" "time" + "techulus/cloud-agent/internal/container" agenthttp "techulus/cloud-agent/internal/http" ) @@ -93,7 +94,19 @@ func (a *Agent) processLeasedWorkItem(item agenthttp.WorkQueueItem) { status := "completed" errorMsg := "" restartAfterReport := false - if err := a.ProcessWorkItem(item); err != nil { + var commandResult *container.CommandResult + var processErr error + if item.Type == "command" { + result, err := a.ProcessCommand(item) + processErr = err + if err == nil { + commandResult = &result + } + } else { + processErr = a.ProcessWorkItem(item) + } + if processErr != nil { + err := processErr if errors.Is(err, errAgentUpgradeRestartNeeded) { restartAfterReport = true } else { @@ -109,12 +122,28 @@ func (a *Agent) processLeasedWorkItem(item agenthttp.WorkQueueItem) { if !restartAfterReport && a.activeWorkItem != nil && a.activeWorkItem.ID == item.ID && a.activeWorkItem.Attempt == item.Attempt { a.activeWorkItem = nil } - a.pendingWorkResults = append(a.pendingWorkResults, agenthttp.CompletedWorkItem{ + completed := agenthttp.CompletedWorkItem{ ID: item.ID, Attempt: item.Attempt, Status: status, Error: errorMsg, - }) + } + if commandResult != nil { + completed.Result = agenthttp.CommandWorkItemResult{ + Type: "command", + Output: commandResult.Output, + ExitCode: &commandResult.ExitCode, + OutputTruncated: commandResult.Truncated, + TimedOut: commandResult.TimedOut, + } + if commandResult.TimedOut { + completed.Status = "failed" + completed.Error = "command timed out after 60 seconds" + } else if commandResult.ExitCode != 0 { + completed.Status = "failed" + } + } + a.pendingWorkResults = append(a.pendingWorkResults, completed) a.workMutex.Unlock() a.RequestStatusReport("work item " + status) diff --git a/agent/internal/container/libpod_exec.go b/agent/internal/container/libpod_exec.go new file mode 100644 index 00000000..5e119162 --- /dev/null +++ b/agent/internal/container/libpod_exec.go @@ -0,0 +1,366 @@ +package container + +import ( + "bufio" + "bytes" + "context" + "encoding/binary" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/url" + "strings" + "time" + "unicode/utf8" +) + +const ( + libpodExecBasePath = "/v4.8.0/libpod" + libpodBodyLimit = 1024 * 1024 + libpodErrorLimit = 4 * 1024 + libpodRequestTime = 10 * time.Second +) + +type execContainerInspect struct { + ID string `json:"Id"` + State struct { + Running bool `json:"Running"` + } `json:"State"` + Config struct { + Labels map[string]string `json:"Labels"` + } `json:"Config"` +} + +type execCreateResponse struct { + ID string `json:"Id"` +} + +type execInspectResponse struct { + Running bool `json:"Running"` + ExitCode int `json:"ExitCode"` +} + +type attachedExecStream struct { + io.Reader + closers []io.Closer +} + +func (s *attachedExecStream) Close() error { + var closeErrors []error + for _, closer := range s.closers { + if err := closer.Close(); err != nil { + closeErrors = append(closeErrors, err) + } + } + return errors.Join(closeErrors...) +} + +func ExecCommand(containerID, serviceID, deploymentID, command string) (CommandResult, error) { + if err := execPreflight(); err != nil { + return CommandResult{}, err + } + if err := verifyExecContainer(containerID, serviceID, deploymentID); err != nil { + return CommandResult{}, err + } + + execID, err := createExec(containerID, command) + if err != nil { + return CommandResult{}, err + } + + startCtx, cancelStart := context.WithTimeout(context.Background(), libpodRequestTime) + stream, err := startAttachedExec(startCtx, execID) + cancelStart() + if err != nil { + cleanupErr := removeExecSession(execID, true) + if cleanupErr != nil { + return CommandResult{}, fmt.Errorf("failed to start exec session: %w; cleanup failed: %v", err, cleanupErr) + } + return CommandResult{}, fmt.Errorf("failed to start exec session: %w", err) + } + ctx, cancel := context.WithTimeout(context.Background(), commandTimeout) + defer cancel() + + var output limitedBuffer + streamDone := make(chan error, 1) + go func() { + streamDone <- copyMultiplexedOutput(stream, &output) + }() + + select { + case streamErr := <-streamDone: + _ = stream.Close() + if streamErr != nil { + if cleanupErr := removeExecSession(execID, true); cleanupErr != nil { + return commandResult(&output), fmt.Errorf("failed to read exec output: %w; cleanup failed: %v", streamErr, cleanupErr) + } + return commandResult(&output), fmt.Errorf("failed to read exec output: %w", streamErr) + } + inspect, err := inspectExecSession(execID) + if err != nil { + _ = removeExecSession(execID, true) + return commandResult(&output), err + } + if inspect.Running { + _ = removeExecSession(execID, true) + return commandResult(&output), fmt.Errorf("exec session remained running after attached stream ended") + } + if err := removeExecSession(execID, false); err != nil { + return commandResult(&output), err + } + result := commandResult(&output) + result.ExitCode = inspect.ExitCode + return result, nil + case <-ctx.Done(): + _ = stream.Close() + if err := removeExecSession(execID, true); err != nil { + return commandResult(&output), fmt.Errorf("command deadline exceeded but exec session cleanup failed: %w", err) + } + select { + case <-streamDone: + case <-time.After(time.Second): + } + result := commandResult(&output) + result.ExitCode = 124 + result.TimedOut = true + return result, nil + } +} + +func commandResult(output *limitedBuffer) CommandResult { + text, truncated := output.snapshot() + text = strings.ToValidUTF8(text, "�") + if len(text) > CommandOutputLimit { + text = text[:CommandOutputLimit] + for !utf8.ValidString(text) { + text = text[:len(text)-1] + } + truncated = true + } + return CommandResult{Output: text, Truncated: truncated} +} + +func execPreflight() error { + ctx, cancel := context.WithTimeout(context.Background(), libpodRequestTime) + defer cancel() + resp, err := libpodRequest(ctx, http.MethodGet, libpodExecBasePath+"/version", nil) + if err != nil { + return fmt.Errorf("podman 4.8 exec API preflight failed: %w", err) + } + defer resp.Body.Close() + if err := requireSuccess(resp, "Podman 4.8 exec API preflight"); err != nil { + return err + } + _, err = io.Copy(io.Discard, io.LimitReader(resp.Body, libpodBodyLimit+1)) + return err +} + +func verifyExecContainer(containerID, serviceID, deploymentID string) error { + var inspect execContainerInspect + if err := libpodJSON(http.MethodGet, libpodExecBasePath+"/containers/"+url.PathEscape(containerID)+"/json", nil, &inspect); err != nil { + return fmt.Errorf("failed to inspect command container: %w", err) + } + if inspect.ID != containerID { + return fmt.Errorf("inspected container ID does not match command target") + } + if !inspect.State.Running { + return fmt.Errorf("container is not running") + } + if inspect.Config.Labels["techulus.service.id"] != serviceID || inspect.Config.Labels["techulus.deployment.id"] != deploymentID { + return fmt.Errorf("container ownership does not match command target") + } + return nil +} + +func createExec(containerID, command string) (string, error) { + body := struct { + AttachStdout bool `json:"AttachStdout"` + AttachStderr bool `json:"AttachStderr"` + TTY bool `json:"Tty"` + Cmd []string `json:"Cmd"` + }{true, true, false, []string{"/bin/sh", "-c", command}} + var created execCreateResponse + if err := libpodJSON(http.MethodPost, libpodExecBasePath+"/containers/"+url.PathEscape(containerID)+"/exec", body, &created); err != nil { + return "", fmt.Errorf("failed to create exec session: %w", err) + } + if created.ID == "" { + return "", fmt.Errorf("failed to create exec session: Podman returned an empty ID") + } + return created.ID, nil +} + +func inspectExecSession(execID string) (execInspectResponse, error) { + var inspect execInspectResponse + err := libpodJSON(http.MethodGet, libpodExecBasePath+"/exec/"+url.PathEscape(execID)+"/json", nil, &inspect) + if err != nil { + return inspect, fmt.Errorf("failed to inspect exec session: %w", err) + } + return inspect, nil +} + +func removeExecSession(execID string, force bool) error { + body := struct { + Force bool `json:"Force"` + }{force} + if err := libpodJSON(http.MethodPost, libpodExecBasePath+"/exec/"+url.PathEscape(execID)+"/remove", body, nil); err != nil { + return fmt.Errorf("failed to remove exec session: %w", err) + } + return nil +} + +func libpodJSON(method, path string, requestBody, responseBody any) error { + ctx, cancel := context.WithTimeout(context.Background(), libpodRequestTime) + defer cancel() + var body io.Reader + if requestBody != nil { + encoded, err := json.Marshal(requestBody) + if err != nil { + return err + } + body = bytes.NewReader(encoded) + } + resp, err := libpodRequest(ctx, method, path, body) + if err != nil { + return err + } + defer resp.Body.Close() + if err := requireSuccess(resp, "Podman request"); err != nil { + return err + } + if responseBody == nil { + data, err := io.ReadAll(io.LimitReader(resp.Body, libpodBodyLimit+1)) + if err != nil { + return err + } + if len(data) > libpodBodyLimit { + return fmt.Errorf("podman response exceeds limit") + } + return nil + } + limited := io.LimitReader(resp.Body, libpodBodyLimit+1) + data, err := io.ReadAll(limited) + if err != nil { + return err + } + if len(data) > libpodBodyLimit { + return fmt.Errorf("podman response exceeds limit") + } + if err := json.Unmarshal(data, responseBody); err != nil { + return fmt.Errorf("invalid Podman response: %w", err) + } + return nil +} + +func libpodRequest(ctx context.Context, method, path string, body io.Reader) (*http.Response, error) { + req, err := http.NewRequestWithContext(ctx, method, podmanBaseURL+path, body) + if err != nil { + return nil, err + } + if body != nil { + req.Header.Set("Content-Type", "application/json") + } + return podmanHTTPClient.Do(req) +} + +func requireSuccess(resp *http.Response, operation string) error { + if resp.StatusCode >= http.StatusOK && resp.StatusCode < http.StatusMultipleChoices { + return nil + } + message, _ := io.ReadAll(io.LimitReader(resp.Body, libpodErrorLimit)) + return fmt.Errorf("%s failed: Podman returned %s: %s", operation, resp.Status, strings.TrimSpace(string(message))) +} + +func startAttachedExec(ctx context.Context, execID string) (io.ReadCloser, error) { + conn, err := podmanDialContext(ctx, "unix", "podman") + if err != nil { + return nil, err + } + success := false + defer func() { + if !success { + _ = conn.Close() + } + }() + if deadline, ok := ctx.Deadline(); ok { + _ = conn.SetDeadline(deadline) + } + body := strings.NewReader(`{"Detach":false,"Tty":false}`) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, podmanBaseURL+libpodExecBasePath+"/exec/"+url.PathEscape(execID)+"/start", body) + if err != nil { + return nil, err + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Connection", "Upgrade") + req.Header.Set("Upgrade", "tcp") + if err := req.Write(conn); err != nil { + return nil, err + } + buffered := bufio.NewReader(conn) + resp, err := http.ReadResponse(buffered, req) + if err != nil { + return nil, err + } + if resp.StatusCode != http.StatusSwitchingProtocols && resp.StatusCode != http.StatusOK { + defer resp.Body.Close() + return nil, requireSuccess(resp, "start attached exec session") + } + if err := conn.SetDeadline(time.Time{}); err != nil { + _ = resp.Body.Close() + return nil, err + } + success = true + if resp.StatusCode == http.StatusSwitchingProtocols { + return &attachedExecStream{Reader: buffered, closers: []io.Closer{conn}}, nil + } + return &attachedExecStream{ + Reader: resp.Body, + closers: []io.Closer{resp.Body, conn}, + }, nil +} + +func copyMultiplexedOutput(reader io.Reader, output io.Writer) error { + var header [8]byte + buffer := make([]byte, 32*1024) + for { + _, err := io.ReadFull(reader, header[:]) + if err == io.EOF { + return nil + } + if err != nil { + return fmt.Errorf("invalid multiplexed frame header: %w", err) + } + if header[1] != 0 || header[2] != 0 || header[3] != 0 { + return fmt.Errorf("invalid multiplexed frame header") + } + remaining := uint64(binary.BigEndian.Uint32(header[4:])) + if header[0] == 3 { + messageLength := min(remaining, uint64(libpodErrorLimit)) + message := make([]byte, messageLength) + if _, err := io.ReadFull(reader, message); err != nil { + return fmt.Errorf("invalid multiplexed error frame: %w", err) + } + if _, err := io.CopyN(io.Discard, reader, int64(remaining-messageLength)); err != nil { + return fmt.Errorf("invalid multiplexed error frame: %w", err) + } + return fmt.Errorf("podman exec stream failed: %s", strings.TrimSpace(string(message))) + } + if header[0] != 0 && header[0] != 1 && header[0] != 2 { + return fmt.Errorf("invalid multiplexed stream %d", header[0]) + } + for remaining > 0 { + chunk := uint64(len(buffer)) + if remaining < chunk { + chunk = remaining + } + if _, err := io.ReadFull(reader, buffer[:int(chunk)]); err != nil { + return fmt.Errorf("invalid multiplexed frame payload: %w", err) + } + if _, err := output.Write(buffer[:int(chunk)]); err != nil { + return err + } + remaining -= chunk + } + } +} diff --git a/agent/internal/container/libpod_exec_test.go b/agent/internal/container/libpod_exec_test.go new file mode 100644 index 00000000..ad00d0c2 --- /dev/null +++ b/agent/internal/container/libpod_exec_test.go @@ -0,0 +1,301 @@ +package container + +import ( + "bytes" + "context" + "encoding/binary" + "encoding/json" + "net" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" +) + +func TestCopyMultiplexedOutputRejectsMalformedFrames(t *testing.T) { + var output limitedBuffer + if err := copyMultiplexedOutput(strings.NewReader("short"), &output); err == nil { + t.Fatal("expected truncated frame header to fail") + } + + var frame bytes.Buffer + var header [8]byte + header[0] = 9 + binary.BigEndian.PutUint32(header[4:], 1) + frame.Write(header[:]) + frame.WriteByte('x') + if err := copyMultiplexedOutput(&frame, &output); err == nil { + t.Fatal("expected unknown stream to fail") + } +} + +func TestCopyMultiplexedOutputBoundsPodmanErrors(t *testing.T) { + var frame bytes.Buffer + var header [8]byte + header[0] = 3 + binary.BigEndian.PutUint32(header[4:], 10_000) + frame.Write(header[:]) + frame.WriteString(strings.Repeat("x", 10_000)) + + err := copyMultiplexedOutput(&frame, &limitedBuffer{}) + if err == nil || len(err.Error()) > libpodErrorLimit+100 { + t.Fatalf("unexpected bounded stream error: length=%d err=%v", len(err.Error()), err) + } +} + +func TestLibpodJSONRejectsOversizedResponses(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, _ = w.Write([]byte(strings.Repeat("x", libpodBodyLimit+1))) + })) + defer server.Close() + oldClient, oldBase := podmanHTTPClient, podmanBaseURL + podmanHTTPClient, podmanBaseURL = server.Client(), server.URL + t.Cleanup(func() { podmanHTTPClient, podmanBaseURL = oldClient, oldBase }) + + if err := libpodJSON(http.MethodGet, "/oversized", nil, &struct{}{}); err == nil || !strings.Contains(err.Error(), "exceeds limit") { + t.Fatalf("expected oversized response rejection, got %v", err) + } +} + +func TestExecCommandPreflightFailsBeforeCreate(t *testing.T) { + api := installFakeExecAPI(t) + api.preflightStatus = http.StatusNotFound + if _, err := ExecCommand(api.containerID, api.serviceID, api.deploymentID, "true"); err == nil { + t.Fatal("expected old API rejection") + } + if api.createCount != 0 { + t.Fatalf("created %d exec sessions after failed preflight", api.createCount) + } +} + +func TestExecCommandRejectsInvalidOwnershipBeforeCreate(t *testing.T) { + tests := []struct { + name string + mutate func(*fakeExecAPI) + }{ + {"service mismatch", func(api *fakeExecAPI) { api.inspectServiceID = "other" }}, + {"deployment mismatch", func(api *fakeExecAPI) { api.inspectDeploymentID = "other" }}, + {"unmanaged", func(api *fakeExecAPI) { api.inspectServiceID, api.inspectDeploymentID = "", "" }}, + {"stopped", func(api *fakeExecAPI) { api.running = false }}, + {"ID mismatch", func(api *fakeExecAPI) { api.inspectContainerID = "other" }}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + api := installFakeExecAPI(t) + tt.mutate(api) + if _, err := ExecCommand(api.containerID, api.serviceID, api.deploymentID, "true"); err == nil { + t.Fatal("expected rejection") + } + if api.createCount != 0 { + t.Fatal("exec session was created") + } + }) + } +} + +func TestExecCommandLifecycleAndOutput(t *testing.T) { + tests := []struct { + name string + output string + exitCode int + truncated bool + }{ + {"multiplexed output", "stdoutstderr", 0, false}, + {"nonzero exit", "bad", 7, false}, + {"exact limit", strings.Repeat("x", CommandOutputLimit), 0, false}, + {"truncated", strings.Repeat("x", CommandOutputLimit+1), 0, true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + api := installFakeExecAPI(t) + api.output, api.exitCode = tt.output, tt.exitCode + result, err := ExecCommand(api.containerID, api.serviceID, api.deploymentID, "printf test") + if err != nil { + t.Fatal(err) + } + if result.ExitCode != tt.exitCode || result.Truncated != tt.truncated || result.TimedOut { + t.Fatalf("unexpected result: %+v", result) + } + if result.Output != tt.output[:min(len(tt.output), CommandOutputLimit)] { + t.Fatalf("unexpected output length/content: %d", len(result.Output)) + } + if api.removeCount != 1 || api.lastRemoveForce { + t.Fatalf("normal removal = count %d force %v", api.removeCount, api.lastRemoveForce) + } + if api.createdCommand != "printf test" { + t.Fatalf("created command = %q", api.createdCommand) + } + if !api.startRequestValid { + t.Fatal("attached exec start options were invalid") + } + }) + } +} + +func TestExecCommandSupportsAttachedOKResponse(t *testing.T) { + api := installFakeExecAPI(t) + api.startStatus = http.StatusOK + api.output = "ok response" + + result, err := ExecCommand(api.containerID, api.serviceID, api.deploymentID, "true") + if err != nil { + t.Fatal(err) + } + if result.Output != api.output || result.ExitCode != 0 { + t.Fatalf("unexpected result: %+v", result) + } +} + +func TestExecCommandTimeoutForceRemoves(t *testing.T) { + api := installFakeExecAPI(t) + api.hang = true + oldTimeout := commandTimeout + commandTimeout = 20 * time.Millisecond + t.Cleanup(func() { commandTimeout = oldTimeout }) + result, err := ExecCommand(api.containerID, api.serviceID, api.deploymentID, "sleep") + if err != nil { + t.Fatal(err) + } + if !result.TimedOut || result.ExitCode != 124 || api.removeCount != 1 || !api.lastRemoveForce { + t.Fatalf("unexpected timeout result=%+v remove=%d force=%v", result, api.removeCount, api.lastRemoveForce) + } +} + +func TestExecCommandForceRemoveFailureIsNotTimeout(t *testing.T) { + api := installFakeExecAPI(t) + api.hang = true + api.removeStatus = http.StatusInternalServerError + oldTimeout := commandTimeout + commandTimeout = 20 * time.Millisecond + t.Cleanup(func() { commandTimeout = oldTimeout }) + result, err := ExecCommand(api.containerID, api.serviceID, api.deploymentID, "sleep") + if err == nil || result.TimedOut { + t.Fatalf("result=%+v err=%v", result, err) + } +} + +type fakeExecAPI struct { + server *httptest.Server + containerID string + serviceID string + deploymentID string + inspectContainerID string + inspectServiceID string + inspectDeploymentID string + running bool + preflightStatus int + startStatus int + removeStatus int + output string + exitCode int + hang bool + createCount int + removeCount int + lastRemoveForce bool + createdCommand string + startRequestValid bool + stopOnce sync.Once + stop chan struct{} +} + +func installFakeExecAPI(t *testing.T) *fakeExecAPI { + t.Helper() + api := &fakeExecAPI{ + containerID: "container/id", serviceID: "service", deploymentID: "deployment", + inspectContainerID: "container/id", inspectServiceID: "service", inspectDeploymentID: "deployment", + running: true, preflightStatus: http.StatusOK, startStatus: http.StatusSwitchingProtocols, removeStatus: http.StatusOK, output: "ok", stop: make(chan struct{}), + } + api.server = httptest.NewServer(http.HandlerFunc(api.serveHTTP)) + address := api.server.Listener.Addr().String() + oldClient, oldDial, oldBase := podmanHTTPClient, podmanDialContext, podmanBaseURL + podmanDialContext = func(ctx context.Context, _, _ string) (net.Conn, error) { + return (&net.Dialer{}).DialContext(ctx, "tcp", address) + } + podmanHTTPClient = &http.Client{Transport: &http.Transport{DisableCompression: true, DialContext: podmanDialContext}} + podmanBaseURL = "http://podman" + t.Cleanup(func() { + api.stopOnce.Do(func() { close(api.stop) }) + api.server.Close() + podmanHTTPClient, podmanDialContext, podmanBaseURL = oldClient, oldDial, oldBase + }) + return api +} + +func (api *fakeExecAPI) serveHTTP(w http.ResponseWriter, r *http.Request) { + switch { + case r.URL.Path == libpodExecBasePath+"/version": + w.WriteHeader(api.preflightStatus) + case strings.Contains(r.URL.Path, "/containers/") && strings.HasSuffix(r.URL.Path, "/json"): + _ = json.NewEncoder(w).Encode(map[string]any{"Id": api.inspectContainerID, "State": map[string]any{"Running": api.running}, "Config": map[string]any{"Labels": map[string]string{"techulus.service.id": api.inspectServiceID, "techulus.deployment.id": api.inspectDeploymentID}}}) + case strings.Contains(r.URL.Path, "/containers/") && strings.HasSuffix(r.URL.Path, "/exec"): + api.createCount++ + var request struct { + Cmd []string `json:"Cmd"` + } + _ = json.NewDecoder(r.Body).Decode(&request) + if len(request.Cmd) == 3 { + api.createdCommand = request.Cmd[2] + } + _ = json.NewEncoder(w).Encode(map[string]string{"Id": "exec/id"}) + case strings.HasSuffix(r.URL.Path, "/start"): + api.serveStart(w, r) + case strings.HasSuffix(r.URL.Path, "/json"): + _ = json.NewEncoder(w).Encode(map[string]any{"Running": false, "ExitCode": api.exitCode}) + case strings.HasSuffix(r.URL.Path, "/remove"): + api.removeCount++ + var request struct { + Force bool `json:"Force"` + } + _ = json.NewDecoder(r.Body).Decode(&request) + api.lastRemoveForce = request.Force + w.WriteHeader(api.removeStatus) + api.stopOnce.Do(func() { close(api.stop) }) + default: + http.NotFound(w, r) + } +} + +func (api *fakeExecAPI) serveStart(w http.ResponseWriter, r *http.Request) { + var request struct { + Detach bool `json:"Detach"` + TTY bool `json:"Tty"` + } + if err := json.NewDecoder(r.Body).Decode(&request); err != nil { + http.Error(w, "invalid start request", http.StatusBadRequest) + return + } + api.startRequestValid = !request.Detach && !request.TTY + if !api.startRequestValid { + http.Error(w, "invalid start options", http.StatusBadRequest) + return + } + hijacker, ok := w.(http.Hijacker) + if !ok { + panic("hijacking unsupported") + } + conn, buffer, err := hijacker.Hijack() + if err != nil { + panic(err) + } + defer conn.Close() + if api.startStatus == http.StatusSwitchingProtocols { + _, _ = buffer.WriteString("HTTP/1.1 101 Switching Protocols\r\nConnection: Upgrade\r\nUpgrade: tcp\r\n\r\n") + } else { + _, _ = buffer.WriteString("HTTP/1.1 200 OK\r\nConnection: close\r\n\r\n") + } + _ = buffer.Flush() + if api.hang { + <-api.stop + return + } + middle := len(api.output) / 2 + for stream, payload := range []string{api.output[:middle], api.output[middle:]} { + var header [8]byte + header[0] = byte(stream + 1) + binary.BigEndian.PutUint32(header[4:], uint32(len(payload))) + _, _ = conn.Write(header[:]) + _, _ = conn.Write([]byte(payload)) + } +} diff --git a/agent/internal/container/runtime.go b/agent/internal/container/runtime.go index ee4d08fb..774997a2 100644 --- a/agent/internal/container/runtime.go +++ b/agent/internal/container/runtime.go @@ -1,6 +1,7 @@ package container import ( + "bytes" "context" "encoding/json" "errors" @@ -9,12 +10,54 @@ import ( "os" "os/exec" "strings" + "sync" "time" "techulus/cloud-agent/internal/retry" "techulus/cloud-agent/internal/wireguard" ) +const CommandOutputLimit = 64 * 1024 + +var commandTimeout = 60 * time.Second + +type CommandResult struct { + Output string + ExitCode int + Truncated bool + TimedOut bool +} + +type limitedBuffer struct { + mutex sync.Mutex + buffer bytes.Buffer + truncated bool +} + +func (b *limitedBuffer) snapshot() (string, bool) { + b.mutex.Lock() + defer b.mutex.Unlock() + return b.buffer.String(), b.truncated +} + +func (b *limitedBuffer) Write(p []byte) (int, error) { + b.mutex.Lock() + defer b.mutex.Unlock() + + n := len(p) + remaining := CommandOutputLimit - b.buffer.Len() + if remaining > 0 { + writeLength := min(remaining, len(p)) + _, _ = b.buffer.Write(p[:writeLength]) + if writeLength < len(p) { + b.truncated = true + } + } else if len(p) > 0 { + b.truncated = true + } + return n, nil +} + func ContainerExists(containerID string) (bool, error) { cmd := exec.Command("podman", "inspect", "--format", "json", containerID) output, err := cmd.CombinedOutput() diff --git a/agent/internal/container/stats.go b/agent/internal/container/stats.go index b5e1f795..a72dfd85 100644 --- a/agent/internal/container/stats.go +++ b/agent/internal/container/stats.go @@ -49,19 +49,19 @@ type podmanStatsReport struct { Stats []podmanStatsSample `json:"Stats"` } -const ( - podmanSocketPath = "/run/podman/podman.sock" - podmanStatsEndpoint = "http://podman/v4.0.0/libpod/containers/stats" -) +const podmanSocketPath = "/run/podman/podman.sock" var ( - podmanStatsClient = &http.Client{Transport: &http.Transport{ + podmanBaseURL = "http://podman" + podmanDialContext = func(ctx context.Context, _, _ string) (net.Conn, error) { + return (&net.Dialer{}).DialContext(ctx, "unix", podmanSocketPath) + } + podmanHTTPClient = &http.Client{Transport: &http.Transport{ DisableCompression: true, - DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) { - return (&net.Dialer{}).DialContext(ctx, "unix", podmanSocketPath) + DialContext: func(ctx context.Context, network, address string) (net.Conn, error) { + return podmanDialContext(ctx, network, address) }, }} - podmanStatsURL = podmanStatsEndpoint ) var previousResourceSamples = struct { @@ -97,7 +97,7 @@ func CollectResourceStats() ([]ResourceStats, error) { ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() - samples, err := fetchPodmanStats(ctx, podmanStatsClient, podmanStatsURL, containerIDs) + samples, err := fetchPodmanStats(ctx, podmanHTTPClient, podmanBaseURL+"/v4.0.0/libpod/containers/stats", containerIDs) if err != nil { return nil, err } diff --git a/agent/internal/container/stats_test.go b/agent/internal/container/stats_test.go index bd7b65d6..a477d021 100644 --- a/agent/internal/container/stats_test.go +++ b/agent/internal/container/stats_test.go @@ -286,13 +286,13 @@ fi w.WriteHeader(api.status) _, _ = w.Write(api.body) })) - previousClient := podmanStatsClient - previousURL := podmanStatsURL - podmanStatsClient = api.server.Client() - podmanStatsURL = api.server.URL + previousClient := podmanHTTPClient + previousBaseURL := podmanBaseURL + podmanHTTPClient = api.server.Client() + podmanBaseURL = api.server.URL t.Cleanup(func() { - podmanStatsClient = previousClient - podmanStatsURL = previousURL + podmanHTTPClient = previousClient + podmanBaseURL = previousBaseURL api.server.Close() }) return api diff --git a/agent/internal/http/client.go b/agent/internal/http/client.go index 5cf21ede..ecf17577 100644 --- a/agent/internal/http/client.go +++ b/agent/internal/http/client.go @@ -358,12 +358,27 @@ type StatusReport struct { } type CompletedWorkItem struct { - ID string `json:"id"` - Attempt int `json:"attempt"` - Status string `json:"status"` - Error string `json:"error,omitempty"` + ID string `json:"id"` + Attempt int `json:"attempt"` + Status string `json:"status"` + Error string `json:"error,omitempty"` + Result WorkItemResult `json:"result,omitempty"` +} + +type WorkItemResult interface { + isWorkItemResult() } +type CommandWorkItemResult struct { + Type string `json:"type"` + Output string `json:"output,omitempty"` + ExitCode *int `json:"exitCode,omitempty"` + OutputTruncated bool `json:"outputTruncated,omitempty"` + TimedOut bool `json:"timedOut,omitempty"` +} + +func (CommandWorkItemResult) isWorkItemResult() {} + type ActiveWorkItem struct { ID string `json:"id"` Attempt int `json:"attempt"` diff --git a/agent/internal/http/client_test.go b/agent/internal/http/client_test.go index d402012b..7a2afdaa 100644 --- a/agent/internal/http/client_test.go +++ b/agent/internal/http/client_test.go @@ -41,6 +41,28 @@ func TestSignedJSONRequests(t *testing.T) { } } +func TestCompletedWorkItemCommandResultJSON(t *testing.T) { + exitCode := 0 + encoded, err := json.Marshal(CompletedWorkItem{ + ID: "command-1", + Attempt: 1, + Status: "completed", + Result: CommandWorkItemResult{ + Type: "command", + Output: "hello\n", + ExitCode: &exitCode, + }, + }) + if err != nil { + t.Fatal(err) + } + + const expected = `{"id":"command-1","attempt":1,"status":"completed","result":{"type":"command","output":"hello\n","exitCode":0}}` + if string(encoded) != expected { + t.Fatalf("unexpected completion JSON: %s", encoded) + } +} + func TestUpdateBuildStatusImageURI(t *testing.T) { keyPair, err := crypto.GenerateKeyPair() if err != nil { diff --git a/web/app/(dashboard)/dashboard/projects/[slug]/[env]/services/[serviceId]/commands/page.tsx b/web/app/(dashboard)/dashboard/projects/[slug]/[env]/services/[serviceId]/commands/page.tsx new file mode 100644 index 00000000..e8e7aeeb --- /dev/null +++ b/web/app/(dashboard)/dashboard/projects/[slug]/[env]/services/[serviceId]/commands/page.tsx @@ -0,0 +1,362 @@ +"use client"; + +import { CheckCircle2, Clock, Loader2, Terminal, XCircle } from "lucide-react"; +import { useMemo, useState } from "react"; +import useSWRInfinite from "swr/infinite"; +import { useService } from "@/components/service/service-layout-client"; +import { Button } from "@/components/ui/button"; +import { Card, CardContent, CardHeader, CardTitle } from "@/components/ui/card"; +import { + Empty, + EmptyDescription, + EmptyMedia, + EmptyTitle, +} from "@/components/ui/empty"; +import { + Item, + ItemContent, + ItemDescription, + ItemGroup, + ItemTitle, +} from "@/components/ui/item"; +import { + NativeSelect, + NativeSelectOption, +} from "@/components/ui/native-select"; +import { StatusBadge } from "@/components/ui/status-badge"; +import { Textarea } from "@/components/ui/textarea"; +import { + formatDateTime, + formatElapsedDurationBetween, + formatRelativeTime, +} from "@/lib/date"; +import { isObservedReady } from "@/lib/deployment-status"; +import { fetcher } from "@/lib/fetcher"; +import { cn } from "@/lib/utils"; + +type CommandStatus = + | "pending" + | "running" + | "succeeded" + | "failed" + | "timed_out"; + +type CommandRun = { + id: string; + command: string; + status: CommandStatus; + output: string | null; + exitCode: number | null; + outputTruncated: boolean; + errorMessage: string | null; + actor: { name: string }; + serverName: string; + containerId: string; + createdAt: string; + startedAt: string | null; + completedAt: string | null; +}; + +type CommandHistory = { + commands: CommandRun[]; + nextCursor: string | null; +}; + +const STATUS_CONFIG: Record< + CommandStatus, + { label: string; icon: typeof Clock; className: string } +> = { + pending: { label: "Queued", icon: Clock, className: "text-slate-500" }, + running: { label: "Running", icon: Loader2, className: "text-blue-500" }, + succeeded: { + label: "Succeeded", + icon: CheckCircle2, + className: "text-green-500", + }, + failed: { label: "Failed", icon: XCircle, className: "text-red-500" }, + timed_out: { + label: "Timed out", + icon: Clock, + className: "text-orange-500", + }, +}; + +function CommandStatusBadge({ + status, + className, + size, +}: { + status: CommandStatus; + className?: string; + size?: "default" | "sm"; +}) { + const config = STATUS_CONFIG[status]; + return ( + + ); +} + +export default function CommandsPage() { + const { service } = useService(); + const targets = service.deployments.filter( + (deployment) => + deployment.containerId && + deployment.runtimeDesiredState === "running" && + isObservedReady(deployment.observedPhase) && + deployment.server?.status === "online", + ); + const [deploymentId, setDeploymentId] = useState(targets[0]?.id ?? ""); + const selectedDeploymentId = targets.some( + (target) => target.id === deploymentId, + ) + ? deploymentId + : (targets[0]?.id ?? ""); + const [command, setCommand] = useState(""); + const [submitting, setSubmitting] = useState(false); + const [submitError, setSubmitError] = useState(null); + + const { + data, + error: historyError, + isLoading, + isValidating, + mutate, + size, + setSize, + } = useSWRInfinite( + (pageIndex, previousPage) => { + if (previousPage && !previousPage.nextCursor) return null; + const cursor = pageIndex === 0 ? null : previousPage?.nextCursor; + return `/api/services/${service.id}/commands${cursor ? `?cursor=${encodeURIComponent(cursor)}` : ""}`; + }, + fetcher, + { + refreshInterval: (pages) => + pages?.some((page) => + page.commands.some( + (item) => item.status === "pending" || item.status === "running", + ), + ) + ? 2000 + : 0, + revalidateOnFocus: true, + }, + ); + + const history = useMemo(() => { + const byId = new Map(); + for (const page of data ?? []) { + for (const item of page.commands) byId.set(item.id, item); + } + return [...byId.values()]; + }, [data]); + const hasMore = data?.[data.length - 1]?.nextCursor != null; + const isLoadingMore = isValidating && Boolean(data?.[size - 1] === undefined); + + async function runCommand() { + setSubmitting(true); + setSubmitError(null); + try { + const response = await fetch(`/api/services/${service.id}/commands`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + deploymentId: selectedDeploymentId, + command, + }), + }); + if (!response.ok) { + const body = (await response.json()) as { + error?: string; + message?: string; + }; + throw new Error( + body.error ?? body.message ?? "Command could not be queued", + ); + } + setCommand(""); + await mutate(); + } catch (error) { + setSubmitError( + error instanceof Error ? error.message : "Command could not be queued", + ); + } finally { + setSubmitting(false); + } + } + + return ( +
+ + + Run command + + + {targets.length === 0 ? ( +

+ No ready, running container is available on an online server. +

+ ) : ( + setDeploymentId(event.target.value)} + > + {targets.map((target) => ( + + {target.server?.name} · {target.containerId?.slice(0, 12)} + + ))} + + )} +