diff --git a/cmd/atenet/internal/router/extproc/handler.go b/cmd/atenet/internal/router/extproc/handler.go index a5622252e8..948d4a6904 100644 --- a/cmd/atenet/internal/router/extproc/handler.go +++ b/cmd/atenet/internal/router/extproc/handler.go @@ -60,8 +60,8 @@ type Result struct { TemplateName string // Resume is the actor-resume outcome, as one of the ateattr.RouterResume* - // values. Empty means "none" — the direction never resumes an actor, or the - // request never got that far. + // values. Empty means "unattempted" — the direction never resumes an actor, + // or the request never got that far. Resume string // DynamicMetadata is attached to the ProcessingResponse alongside Response, @@ -72,10 +72,10 @@ type Result struct { } // resume returns the resume label for the route-duration metric, defaulting an -// unset outcome to "none". +// unset outcome to "unattempted". func (r Result) resume() string { if r.Resume == "" { - return ateattr.RouterResumeNone + return ateattr.RouterResumeUnattempted } return r.Resume } diff --git a/cmd/atenet/internal/router/extproc/metrics.go b/cmd/atenet/internal/router/extproc/metrics.go index 7cc2ddf37f..406a828af6 100644 --- a/cmd/atenet/internal/router/extproc/metrics.go +++ b/cmd/atenet/internal/router/extproc/metrics.go @@ -64,6 +64,8 @@ func (s *Server) recordRouteDuration(ctx context.Context, d time.Duration, tmplN if s.routeDuration == nil { return } + tmplNs = ateattr.NormalizeTemplateDimension(tmplNs) + tmplName = ateattr.NormalizeTemplateDimension(tmplName) s.routeDuration.Record(ctx, d.Seconds(), metric.WithAttributes( ateattr.TemplateAtespaceKey.String(tmplNs), ateattr.TemplateNameKey.String(tmplName), diff --git a/cmd/atenet/internal/router/extproc/metrics_test.go b/cmd/atenet/internal/router/extproc/metrics_test.go index ad35360139..138a7579a8 100644 --- a/cmd/atenet/internal/router/extproc/metrics_test.go +++ b/cmd/atenet/internal/router/extproc/metrics_test.go @@ -140,3 +140,59 @@ func TestRecordRouteDuration_Attributes(t *testing.T) { } } } + +func TestRecordRouteDuration_NormalizesEmptyTemplateDimensions(t *testing.T) { + reader := sdkmetric.NewManualReader() + mp := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader)) + h, err := mp.Meter("atenet-router").Float64Histogram(routeDurationMetricName) + if err != nil { + t.Fatalf("failed to create histogram: %v", err) + } + + s := NewServer(50051, h, nil) + s.recordRouteDuration(context.Background(), 5*time.Millisecond, "", "", classifyOutcome(errors.New("fail")), ateattr.RouterResumeUnattempted) + + var rm metricdata.ResourceMetrics + if err := reader.Collect(context.Background(), &rm); err != nil { + t.Fatalf("Collect failed: %v", err) + } + + dp := rm.ScopeMetrics[0].Metrics[0].Data.(metricdata.Histogram[float64]).DataPoints[0] + wantAttrs := map[string]string{ + "ate.template.atespace": "unknown", + "ate.template.name": "unknown", + "ate.router.outcome": "resume_error", + "ate.router.resume": "unattempted", + } + + for k, want := range wantAttrs { + val, exists := dp.Attributes.Value(attribute.Key(k)) + if !exists { + t.Errorf("missing metric attribute %q", k) + } else if val.AsString() != want { + t.Errorf("attribute %q = %q, want %q", k, val.AsString(), want) + } + } +} + +func TestResultResume_DefaultsToUnattempted(t *testing.T) { + tests := []struct { + name string + resume string + want string + }{ + {name: "empty defaults to unattempted", resume: "", want: ateattr.RouterResumeUnattempted}, + {name: "none preserved", resume: ateattr.RouterResumeNone, want: ateattr.RouterResumeNone}, + {name: "triggered preserved", resume: ateattr.RouterResumeTriggered, want: ateattr.RouterResumeTriggered}, + {name: "joined preserved", resume: ateattr.RouterResumeJoined, want: ateattr.RouterResumeJoined}, + {name: "unattempted preserved", resume: ateattr.RouterResumeUnattempted, want: ateattr.RouterResumeUnattempted}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + r := Result{Resume: tt.resume} + if got := r.resume(); got != tt.want { + t.Errorf("Result.resume() = %q, want %q", got, tt.want) + } + }) + } +} diff --git a/cmd/atenet/internal/router/ingress/resumer.go b/cmd/atenet/internal/router/ingress/resumer.go index b515e19e50..2e82e872ea 100644 --- a/cmd/atenet/internal/router/ingress/resumer.go +++ b/cmd/atenet/internal/router/ingress/resumer.go @@ -70,9 +70,10 @@ func (e *budgetExhaustedError) Unwrap() error { return e.lastErr } type ResumeOutcome string const ( - ResumeOutcomeNone ResumeOutcome = ateattr.RouterResumeNone - ResumeOutcomeTriggered ResumeOutcome = ateattr.RouterResumeTriggered - ResumeOutcomeJoined ResumeOutcome = ateattr.RouterResumeJoined + ResumeOutcomeNone ResumeOutcome = ateattr.RouterResumeNone + ResumeOutcomeTriggered ResumeOutcome = ateattr.RouterResumeTriggered + ResumeOutcomeJoined ResumeOutcome = ateattr.RouterResumeJoined + ResumeOutcomeUnattempted ResumeOutcome = ateattr.RouterResumeUnattempted ) type resumeCallResult struct { @@ -160,6 +161,18 @@ func (r *ActorResumer) retryable(err error) bool { } } +// isDefinitiveResumeError reports whether err represents a failure where no cold +// activation could be attempted (e.g. the actor does not exist, bad request, or +// permission denied), as opposed to in-flight capacity or transient failures. +func isDefinitiveResumeError(err error) bool { + switch status.Code(err) { + case codes.NotFound, codes.InvalidArgument, codes.PermissionDenied, codes.Unauthenticated: + return true + default: + return false + } +} + // ResumeActor ensures the requested actor is running. It deduplicates concurrent // requests within the process and, when parking is enabled, holds the request // while retrying transient failures until the budget elapses. @@ -244,27 +257,22 @@ func (r *ActorResumer) ResumeActor(ctx context.Context, actorRef resources.Actor select { case <-ctx.Done(): // The caller's request context was canceled before the singleflight resume completed. - // Return early with ResumeOutcomeNone ("none") - return nil, ResumeOutcomeNone, ctx.Err() + // Return early with ResumeOutcomeUnattempted ("unattempted") + return nil, ResumeOutcomeUnattempted, ctx.Err() case res := <-ch: callRes, _ := res.Val.(*resumeCallResult) if callRes == nil { if res.Err != nil { - return nil, ResumeOutcomeNone, res.Err + return nil, ResumeOutcomeUnattempted, res.Err } - return nil, ResumeOutcomeNone, status.Error(codes.Internal, "resume call returned nil result") - } - - // On error, return ResumeOutcomeNone ("none") so the failure is tagged - // under the 'outcome' label rather than misreported as an activation. - if callRes.err != nil { - return nil, ResumeOutcomeNone, callRes.err + return nil, ResumeOutcomeUnattempted, status.Error(codes.Internal, "resume call returned nil result") } // Disambiguate singleflight resume outcome: - // - ResumeOutcomeNone ("none"): resumed == false, actor was already active/running. - // - ResumeOutcomeTriggered ("triggered"): Cold activation leader (resumed == true, caller's reqID == leaderID). - // - ResumeOutcomeJoined ("joined"): Cold activation joiner (resumed == true, caller's reqID != leaderID). + // - ResumeOutcomeNone ("none"): resumed == false and err == nil, actor was already active/running (warm route). + // - ResumeOutcomeTriggered ("triggered"): Cold activation leader (resumed == true or in-flight activation errored, caller's reqID == leaderID). + // - ResumeOutcomeJoined ("joined"): Cold activation joiner (resumed == true or in-flight activation errored, caller's reqID != leaderID). + // - ResumeOutcomeUnattempted ("unattempted"): Definitive non-activation error (e.g. NotFound, InvalidArgument) where no activation was attempted. outcome := ResumeOutcomeNone if callRes.resumed { if callRes.leaderID == reqID { @@ -272,6 +280,18 @@ func (r *ActorResumer) ResumeActor(ctx context.Context, actorRef resources.Actor } else { outcome = ResumeOutcomeJoined } + } else if callRes.err != nil { + if isDefinitiveResumeError(callRes.err) { + outcome = ResumeOutcomeUnattempted + } else if callRes.leaderID == reqID { + outcome = ResumeOutcomeTriggered + } else { + outcome = ResumeOutcomeJoined + } + } + + if callRes.err != nil { + return nil, outcome, callRes.err } return callRes.actor, outcome, nil diff --git a/cmd/atenet/internal/router/ingress/resumer_test.go b/cmd/atenet/internal/router/ingress/resumer_test.go index 6483280467..b6a278c9d5 100644 --- a/cmd/atenet/internal/router/ingress/resumer_test.go +++ b/cmd/atenet/internal/router/ingress/resumer_test.go @@ -18,6 +18,7 @@ import ( "context" "errors" "sync" + "sync/atomic" "testing" "testing/synctest" "time" @@ -146,8 +147,87 @@ func TestActorResumer_ResumeActor(t *testing.T) { if got := status.Code(err); got != codes.NotFound { t.Errorf("expected gRPC code NotFound, got %v (err=%v)", got, err) } - if outcome != ResumeOutcomeNone { - t.Errorf("expected outcome %q on error, got %q", ResumeOutcomeNone, outcome) + if outcome != ResumeOutcomeUnattempted { + t.Errorf("expected outcome %q on definitive error for leader attempt, got %q", ResumeOutcomeUnattempted, outcome) + } + }) + + t.Run("ContextCanceled", func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + mock := &resumerMockClient{ + resumeFn: func(ctx context.Context, in *ateapipb.ResumeActorRequest, opts ...grpc.CallOption) (*ateapipb.ResumeActorResponse, error) { + return &ateapipb.ResumeActorResponse{Resumed: true}, nil + }, + } + + resumer := NewActorResumer(mock) + _, outcome, err := resumer.ResumeActor(ctx, testActorRef) + if !errors.Is(err, context.Canceled) { + t.Errorf("expected context.Canceled, got %v", err) + } + if outcome != ResumeOutcomeUnattempted { + t.Errorf("expected outcome %q on context cancellation, got %q", ResumeOutcomeUnattempted, outcome) + } + }) + + t.Run("SingleflightDeduplication_ErrorDisambiguation", func(t *testing.T) { + var resumeCalled int + var mu sync.Mutex + const concurrentRequests = 10 + var callersStarted atomic.Int32 + + mock := &resumerMockClient{ + resumeFn: func(ctx context.Context, in *ateapipb.ResumeActorRequest, opts ...grpc.CallOption) (*ateapipb.ResumeActorResponse, error) { + mu.Lock() + resumeCalled++ + mu.Unlock() + // Barrier: wait until all concurrent goroutines have launched ResumeActor + for callersStarted.Load() < concurrentRequests { + time.Sleep(time.Millisecond) + } + time.Sleep(10 * time.Millisecond) + return nil, status.Error(codes.ResourceExhausted, "no free workers available") + }, + } + + resumer := NewActorResumer(mock) + + var wg sync.WaitGroup + outcomes := make([]ResumeOutcome, concurrentRequests) + errs := make([]error, concurrentRequests) + + wg.Add(concurrentRequests) + for i := 0; i < concurrentRequests; i++ { + go func(idx int) { + defer wg.Done() + callersStarted.Add(1) + _, outcomes[idx], errs[idx] = resumer.ResumeActor(context.Background(), testActorRef) + }(i) + } + wg.Wait() + + var triggeredCount, joinedCount int + for i := 0; i < concurrentRequests; i++ { + if got := status.Code(errs[i]); got != codes.ResourceExhausted { + t.Fatalf("request %d expected ResourceExhausted, got %v", i, errs[i]) + } + switch outcomes[i] { + case ResumeOutcomeTriggered: + triggeredCount++ + case ResumeOutcomeJoined: + joinedCount++ + default: + t.Errorf("unexpected outcome for request %d: %q", i, outcomes[i]) + } + } + + if triggeredCount != 1 { + t.Errorf("expected exactly 1 request to have outcome 'triggered', got %d", triggeredCount) + } + if joinedCount != concurrentRequests-1 { + t.Errorf("expected %d requests to have outcome 'joined', got %d", concurrentRequests-1, joinedCount) } }) diff --git a/docs/metrics/registry/metrics.yaml b/docs/metrics/registry/metrics.yaml index 7548f2cb1f..61cb0f51a7 100644 --- a/docs/metrics/registry/metrics.yaml +++ b/docs/metrics/registry/metrics.yaml @@ -69,19 +69,24 @@ groups: type: attribute_group brief: > The identity of an ActorTemplate. The values are not limited to a list. - But an operator makes each template. No value comes from a request. Thus - the number of values stays small. + An operator makes each template (or "unknown" when the direction has no + template or the request failed before resolving one). No value comes from + a request. Thus the number of values stays small. attributes: - id: ate.template.atespace stability: development type: string - brief: The atespace of the ActorTemplate of the actor. - examples: [ate-demo-counter] + brief: > + The atespace of the ActorTemplate of the actor, or "unknown" when the + direction has no template (or the request failed before resolving one). + examples: [ate-demo-counter, unknown] - id: ate.template.name stability: development type: string - brief: The name of the ActorTemplate of the actor. - examples: [counter] + brief: > + The name of the ActorTemplate of the actor, or "unknown" when the + direction has no template (or the request failed before resolving one). + examples: [counter, unknown] - id: registry.ate.workerpool type: attribute_group @@ -422,6 +427,13 @@ groups: A resume was already in operation. This request waited for it. This value is separate because it is not a correct sample. One cold start with 50 requests is one slow activation and not 51. + - id: unattempted + stability: development + value: unattempted + brief: > + No cold activation was in flight. The request was invalid, the + actor did not exist, the direction does not resume, or the + request canceled before activation was attempted. - id: ate.router.outcome stability: development brief: The result of the route attempt. diff --git a/docs/metrics/substrate.yaml b/docs/metrics/substrate.yaml index fbf89d361f..8eb2061901 100644 --- a/docs/metrics/substrate.yaml +++ b/docs/metrics/substrate.yaml @@ -150,7 +150,8 @@ cardinality_rules: - id: bounded-or-catalog-scoped brief: > Each ate.* metric label has a list of permitted values, or it names an - object that an operator made. The names of the templates and the pools + object that an operator made (or "unknown" when a template cannot be + resolved or is not applicable). The names of the templates and the pools never come from a request. enforced: false could_be_enforced_by: > diff --git a/internal/ateattr/ateattr.go b/internal/ateattr/ateattr.go index 702bf1263b..3a1f79d642 100644 --- a/internal/ateattr/ateattr.go +++ b/internal/ateattr/ateattr.go @@ -56,6 +56,21 @@ const ( ActorVersionKey = attribute.Key("ate.actor.version") ) +// TemplateUnknown is the fallback for unresolvable or missing template dimensions. +// Note: "unknown" is syntactically a legal atespace/template name; like +// ate.sandbox.class="unknown", this trades potential collision with a real +// object named "unknown" for maintaining a bounded, non-empty metric dimension. +const TemplateUnknown = "unknown" + +// NormalizeTemplateDimension ensures a template dimension (atespace or name) is +// non-empty, falling back to TemplateUnknown if unset. +func NormalizeTemplateDimension(dim string) string { + if dim == "" { + return TemplateUnknown + } + return dim +} + // ReservedNamespace is substrate's. A producer that merges untrusted fields into a // record drops everything under it, so nothing a workload sets can read as // platform-issued attribution downstream. @@ -197,6 +212,9 @@ const ( RouterResumeTriggered = "triggered" // RouterResumeJoined indicates this request parked on an in-flight singleflight resume. RouterResumeJoined = "joined" + // RouterResumeUnattempted indicates no cold activation was in flight (e.g. + // definitive errors before activation, request cancellation, or non-resuming route). + RouterResumeUnattempted = "unattempted" ) // Values for ImageCacheOutcomeKey. A hit is a complete image record; a miss diff --git a/internal/ateattr/ateattr_test.go b/internal/ateattr/ateattr_test.go index 0fd7bdb312..b3b7881597 100644 --- a/internal/ateattr/ateattr_test.go +++ b/internal/ateattr/ateattr_test.go @@ -416,6 +416,12 @@ func TestMetricLabelValues(t *testing.T) { {SnapshotPhaseTotal, "total"}, {SandboxClassUnknown, "unknown"}, + {TemplateUnknown, "unknown"}, + + {RouterResumeNone, "none"}, + {RouterResumeTriggered, "triggered"}, + {RouterResumeJoined, "joined"}, + {RouterResumeUnattempted, "unattempted"}, } for _, tt := range tests { t.Run(tt.want, func(t *testing.T) { @@ -770,3 +776,22 @@ func TestActorRefLogAttrs(t *testing.T) { t.Errorf("ActorRefLogAttrs() = %v, want %v", got, want) } } + +func TestNormalizeTemplateDimension(t *testing.T) { + tests := []struct { + name string + in string + want string + }{ + {name: "empty falls back to unknown", in: "", want: TemplateUnknown}, + {name: "valid name preserved", in: "counter", want: "counter"}, + {name: "valid atespace preserved", in: "ate-demo", want: "ate-demo"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := NormalizeTemplateDimension(tt.in); got != tt.want { + t.Errorf("NormalizeTemplateDimension(%q) = %q, want %q", tt.in, got, tt.want) + } + }) + } +}