Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 10 additions & 5 deletions services/core/internal/sandbox/node/agent.go
Original file line number Diff line number Diff line change
Expand Up @@ -388,24 +388,29 @@ func (a *agent) executeWork(ctx context.Context, task work) {
operationCtx, cancel := context.WithDeadline(context.Background(), task.request.deadline)
a.active.Add(1)
started := time.Now()
result := execute(operationCtx, task.provider, task.request)
result, err := execute(operationCtx, task.provider, task.request)
finished := time.Now()
a.active.Add(-1)
cancel()
_ = task.connection.write(frame{Type: "response", Response: &result})
logOperation(ctx, task, started, finished, result.ErrorCode)
logOperation(ctx, task, started, finished, result.ErrorCode, err)
}

// Queue wait and execution share this node's monotonic clock. Read-only
// observations repeat on every reconciliation tick and log at debug level.
func logOperation(ctx context.Context, task work, started, finished time.Time, code string) {
// The local error stays in this host's journal; it may name host paths.
func logOperation(ctx context.Context, task work, started, finished time.Time, code string, err error) {
level := slog.LevelInfo
switch task.request.Operation {
case "info", "renew", "initial", "new_compute", "compute", "observe":
if code == "" {
level = slog.LevelDebug
}
}
log.Bg().Log(ctx, level, "sandbox node operation", "operation", task.request.Operation, "allocation_id", task.request.Reference.AllocationID,
"queue_ms", started.Sub(task.queued).Milliseconds(), "duration_ms", finished.Sub(started).Milliseconds(), "error_code", code)
attrs := []any{"operation", task.request.Operation, "allocation_id", task.request.Reference.AllocationID,
"queue_ms", started.Sub(task.queued).Milliseconds(), "duration_ms", finished.Sub(started).Milliseconds(), "error_code", code}
if err != nil {
attrs = append(attrs, "error", err)
}
log.Bg().Log(ctx, level, "sandbox node operation", attrs...)
}
4 changes: 2 additions & 2 deletions services/core/internal/sandbox/node/observations_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -152,7 +152,7 @@ func TestObservationWirePreservesUnavailableAndRejectsMismatchedIdentity(t *test
}
for _, test := range providers {
t.Run(test.name, func(t *testing.T) {
out := execute(t.Context(), test.provider, q)
out, _ := execute(t.Context(), test.provider, q)
if !errors.Is(responseError(out), test.want) || out.Sample != nil {
t.Fatal("observation error or missing sample was fabricated", out)
}
Expand All @@ -163,7 +163,7 @@ func TestObservationWirePreservesUnavailableAndRejectsMismatchedIdentity(t *test
invalid := target
mutate(&invalid)
q.Observation = &invalid
if out := execute(t.Context(), p, q); !errors.Is(responseError(out), sandbox.ErrInvalid) {
if out, _ := execute(t.Context(), p, q); !errors.Is(responseError(out), sandbox.ErrInvalid) {
t.Fatal("mismatched observation reached provider", out)
}
}
Expand Down
2 changes: 1 addition & 1 deletion services/core/internal/sandbox/node/operations_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@ import (
func TestUnsupportedWireIsExplicitAndDoesNotInvokeProvider(t *testing.T) {
p := &fakeProvider{}
q := request{ID: uuid.NewString(), ConnectionID: uuid.NewString(), Reference: reference(), TimeoutMillis: 1000, Operation: "initial"}
out := execute(t.Context(), p, q)
out, _ := execute(t.Context(), p, q)
if p.creates != 0 || p.reads != 0 || p.kills != 0 {
t.Fatal("unsupported operation reached provider")
}
Expand Down
10 changes: 6 additions & 4 deletions services/core/internal/sandbox/node/wire.go
Original file line number Diff line number Diff line change
Expand Up @@ -289,20 +289,22 @@ func (q request) validate() error {
}
return sandbox.ErrInvalid
}
func execute(ctx context.Context, p sandbox.SandboxProvider, q request) response {

// execute also returns the local error; only its code crosses the transport.
func execute(ctx context.Context, p sandbox.SandboxProvider, q request) (response, error) {
out := response{ID: q.ID, ConnectionID: q.ConnectionID}
err := q.validate()
if err != nil {
out.ErrorCode = errorCode(err)
return out
return out, err
}
if err := providercontract.Require(p, operationMethod(q.Operation)); err != nil {
out.ErrorCode = errorCode(err)
var unsupported *providercontract.UnsupportedError
if errors.As(err, &unsupported) {
out.Unsupported = unsupported
}
return out
return out, err
}
var info sandbox.Info
var command sandbox.CommandResult
Expand Down Expand Up @@ -374,7 +376,7 @@ func execute(ctx context.Context, p sandbox.SandboxProvider, q request) response
out.Command = nil
out.Compute = nil
}
return out
return out, err
}

func requiresReady(q request) bool {
Expand Down
5 changes: 4 additions & 1 deletion services/core/internal/sandbox/node/workspace_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -78,7 +78,10 @@ func TestActualWorkspaceRefusalKeepsSettledAbsenceOnWire(t *testing.T) {
bootstrap := sandbox.Bootstrap{Reference: ref, SessionID: ref.TenantID, DeviceID: ref.EnvironmentID, CoreURL: "https://core.example/api/v1", Credential: "fixture", Workspace: binding}
ctx, cancel := context.WithTimeout(t.Context(), time.Second)
defer cancel()
out := execute(ctx, provider, request{ID: ref.AllocationID, ConnectionID: ref.TenantID, Reference: ref, Operation: "create", TimeoutMillis: 1000, Bootstrap: &bootstrap})
out, local := execute(ctx, provider, request{ID: ref.AllocationID, ConnectionID: ref.TenantID, Reference: ref, Operation: "create", TimeoutMillis: 1000, Bootstrap: &bootstrap})
if !errors.Is(local, workspacefs.ErrUnsupported) {
t.Fatal("the node journal lost the local error", local)
}
raw, err := json.Marshal(frame{Version: ProtocolVersion, Type: "response", Response: &out})
if err != nil {
t.Fatal(err)
Expand Down
Loading