diff --git a/cmd/atenet/internal/router/extproc/handler.go b/cmd/atenet/internal/router/extproc/handler.go index 01c055d87..7105493d1 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 "unknown" — the direction never resumes an actor, or + // the request never got far enough to learn whether an activation ran. 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 "unknown". func (r Result) resume() string { if r.Resume == "" { - return ateattr.RouterResumeNone + return ateattr.RouterResumeUnknown } return r.Resume } diff --git a/cmd/atenet/internal/router/extproc/metrics.go b/cmd/atenet/internal/router/extproc/metrics.go index 7cc2ddf37..406a828af 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 ad3536013..f4a3217a3 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.RouterResumeUnknown) + + 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": "unknown", + } + + 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_DefaultsToUnknown(t *testing.T) { + tests := []struct { + name string + resume string + want string + }{ + {name: "empty defaults to unknown", resume: "", want: ateattr.RouterResumeUnknown}, + {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: "unknown preserved", resume: ateattr.RouterResumeUnknown, want: ateattr.RouterResumeUnknown}, + } + 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/flight.go b/cmd/atenet/internal/router/ingress/flight.go index d72876f46..45956f0ec 100644 --- a/cmd/atenet/internal/router/ingress/flight.go +++ b/cmd/atenet/internal/router/ingress/flight.go @@ -53,16 +53,20 @@ func (f *resumeActorFlight) signalRetrying() { // callerResult classifies f's completed outcome for one caller. It must only // be called after f.done is closed. +// +// Only a resume that completed without an error says whether an activation +// ran, so the three activation labels are reserved for that case. A failed +// resume reports "unknown": the gRPC code alone does not carry the fact. A +// canceled leader's flight outlives its request and keeps restoring the actor, +// and a DeadlineExceeded can land in the middle of a restore. func (f *resumeActorFlight) callerResult(reqID uint64) (*ateapipb.Actor, ResumeOutcome, error) { res := f.result if res == nil { - return nil, ResumeOutcomeNone, status.Error(codes.Internal, "resume call returned nil result") + return nil, ResumeOutcomeUnknown, 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 res.err != nil { - return nil, ResumeOutcomeNone, res.err + return nil, ResumeOutcomeUnknown, res.err } // Disambiguate the shared-flight resume outcome: diff --git a/cmd/atenet/internal/router/ingress/resumer.go b/cmd/atenet/internal/router/ingress/resumer.go index 52def8e48..d80a90774 100644 --- a/cmd/atenet/internal/router/ingress/resumer.go +++ b/cmd/atenet/internal/router/ingress/resumer.go @@ -73,6 +73,7 @@ const ( ResumeOutcomeNone ResumeOutcome = ateattr.RouterResumeNone ResumeOutcomeTriggered ResumeOutcome = ateattr.RouterResumeTriggered ResumeOutcomeJoined ResumeOutcome = ateattr.RouterResumeJoined + ResumeOutcomeUnknown ResumeOutcome = ateattr.RouterResumeUnknown ) type resumeCallResult struct { @@ -271,8 +272,9 @@ func (r *ActorResumer) awaitFlight(ctx context.Context, f *resumeActorFlight, ac select { case <-ctx.Done(): // The caller's request context was canceled before the shared resume - // completed. Return early with ResumeOutcomeNone ("none"). - return nil, ResumeOutcomeNone, ctx.Err() + // completed. The flight continues and may still activate the actor, so + // the outcome is "unknown", not "none". + return nil, ResumeOutcomeUnknown, ctx.Err() case <-f.done: // Fast path: the flight finished without ever retrying, or before this // caller saw retrying. The lot is never touched. @@ -292,7 +294,7 @@ func (r *ActorResumer) awaitFlight(ctx context.Context, f *resumeActorFlight, ac release, ok := r.enterLot(ctx) if !ok { - return nil, ResumeOutcomeNone, parkingFullErr(actorRef.String()) + return nil, ResumeOutcomeUnknown, parkingFullErr(actorRef.String()) } var finalErr error defer func() { release(parkOutcomeFor(finalErr)) }() @@ -300,7 +302,7 @@ func (r *ActorResumer) awaitFlight(ctx context.Context, f *resumeActorFlight, ac select { case <-ctx.Done(): finalErr = ctx.Err() - return nil, ResumeOutcomeNone, finalErr + return nil, ResumeOutcomeUnknown, finalErr case <-f.done: actor, outcome, err := f.callerResult(reqID) finalErr = err diff --git a/cmd/atenet/internal/router/ingress/resumer_test.go b/cmd/atenet/internal/router/ingress/resumer_test.go index c9efa082f..3e497e2c6 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" @@ -149,11 +150,91 @@ 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 != ResumeOutcomeUnknown { + t.Errorf("expected outcome %q on a failed resume, got %q", ResumeOutcomeUnknown, outcome) } }) + t.Run("CallerContextCanceled", func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + // The caller selects on its own context and on the flight's completion. + // Hold the flight open for the whole call, so only the cancellation can + // be ready. A flight that can finish first makes both cases ready, and + // the select picks one of them at random. + gate := make(chan struct{}) + defer close(gate) + + mock := &resumerMockClient{ + resumeFn: func(ctx context.Context, in *ateapipb.ResumeActorRequest, opts ...grpc.CallOption) (*ateapipb.ResumeActorResponse, error) { + <-gate + 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 != ResumeOutcomeUnknown { + t.Errorf("expected outcome %q on a canceled caller, got %q", ResumeOutcomeUnknown, outcome) + } + }) + + // A failed resume tells no caller whether an activation ran — the leader no + // more than the joiners — so every caller on the flight reports "unknown". + t.Run("SingleflightDeduplication_FailedFlight", func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + const concurrentRequests = 10 + var resumeCalled atomic.Int32 + gate := make(chan struct{}) + + mock := &resumerMockClient{ + resumeFn: func(ctx context.Context, in *ateapipb.ResumeActorRequest, opts ...grpc.CallOption) (*ateapipb.ResumeActorResponse, error) { + resumeCalled.Add(1) + <-gate + 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() + _, outcomes[idx], errs[idx] = resumer.ResumeActor(context.Background(), testActorRef) + }(i) + } + // synctest.Wait returns once every caller is parked on the flight, + // so the release below cannot beat one of them to it. The flight + // leaves the registry when it completes, so a caller that arrived + // after that would start its own RPC and fail the count below. + synctest.Wait() + close(gate) + wg.Wait() + + 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]) + } + if outcomes[i] != ResumeOutcomeUnknown { + t.Errorf("request %d: expected outcome %q on a failed flight, got %q", i, ResumeOutcomeUnknown, outcomes[i]) + } + } + + if calls := resumeCalled.Load(); calls != 1 { + t.Errorf("ResumeActor calls = %d, want 1 for %d concurrent callers", calls, concurrentRequests) + } + }) + }) + t.Run("SingleflightDeduplication_Disambiguation", func(t *testing.T) { var resumeCalled int var mu sync.Mutex @@ -713,8 +794,8 @@ func TestActorResumer_LotAdmission(t *testing.T) { if !errors.As(err, &reqErr) || reqErr.StatusCode != int(envoy_type.StatusCode_ServiceUnavailable) { t.Fatalf("expected a 503 router-at-capacity denial, got %v", err) } - if outcome != ResumeOutcomeNone { - t.Errorf("shed caller outcome = %q, want %q", outcome, ResumeOutcomeNone) + if outcome != ResumeOutcomeUnknown { + t.Errorf("shed caller outcome = %q, want %q", outcome, ResumeOutcomeUnknown) } // The caller was turned away at the transition: exactly one attempt // had run. @@ -778,8 +859,8 @@ func TestActorResumer_LotAdmission(t *testing.T) { if !errors.As(err, &reqErr) || reqErr.StatusCode != int(envoy_type.StatusCode_ServiceUnavailable) { t.Fatalf("joiner: expected a 503 router-at-capacity denial, got %v", err) } - if outcome != ResumeOutcomeNone { - t.Errorf("joiner outcome = %q, want %q", outcome, ResumeOutcomeNone) + if outcome != ResumeOutcomeUnknown { + t.Errorf("joiner outcome = %q, want %q", outcome, ResumeOutcomeUnknown) } close(proceed) diff --git a/docs/metrics/registry/metrics.yaml b/docs/metrics/registry/metrics.yaml index 47fd18ab3..0174ebebc 100644 --- a/docs/metrics/registry/metrics.yaml +++ b/docs/metrics/registry/metrics.yaml @@ -160,18 +160,28 @@ groups: 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. + the number of values stays small. One metric, + atenet.router.route.duration, also reports the literal "unknown" when it + has no template to name. 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. On + atenet.router.route.duration only, the value is "unknown" when the + direction has no template, or the request failed before the router + resolved 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. On + atenet.router.route.duration only, the value is "unknown" when the + direction has no template, or the request failed before the router + resolved one. + examples: [counter, unknown] - id: registry.ate.workerpool type: attribute_group @@ -504,16 +514,26 @@ groups: stability: development value: triggered brief: > - This request got the singleflight lock and did the resume. Its - time is the activation time. Its rate is the number of resumes - each second. + This request got the singleflight lock and completed the resume. + Its time is the activation time. Its rate is the number of + resumes each second. - id: joined stability: development value: joined brief: > - 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. + A resume was already in operation. This request waited for it, + and the resume activated the actor. 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: unknown + stability: development + value: unknown + brief: > + The resume did not complete, thus the router cannot tell whether + an activation ran. The resume failed, or the request stopped + before the resume gave an answer, or the direction does not + resume an actor. Do not read the time in this series as an + activation time. - id: ate.router.outcome stability: development brief: The result of the route attempt. @@ -1191,7 +1211,9 @@ groups: This is the measurement at the user boundary. Divide the data by ate.router.resume before you read it. The triggered series is the activation time. The none series is the usual warm route. A sum across the - two series puts milliseconds and tens of seconds in one distribution. + two series puts milliseconds and tens of seconds in one distribution. The + unknown series holds the resumes that did not complete, thus it is neither + an activation time nor a warm route. annotations: substrate: emitted_by: [atenet-router] diff --git a/docs/metrics/substrate.yaml b/docs/metrics/substrate.yaml index 8cc8d80bf..7b85faed9 100644 --- a/docs/metrics/substrate.yaml +++ b/docs/metrics/substrate.yaml @@ -151,7 +151,11 @@ cardinality_rules: 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 - never come from a request. + never come from a request. One exception: on + atenet.router.route.duration, ate.template.atespace and ate.template.name + hold the literal "unknown" when the router has no template to name. The + router sends that constant in place of the empty string. It does not send + a value that the request gave. enforced: false could_be_enforced_by: > A Weaver Rego policy can make each ate.* attribute of type string give a diff --git a/docs/observability.md b/docs/observability.md index 2ddc37be4..0a07201da 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -267,7 +267,7 @@ For `ate.workerpool.desired_workers` and `ate.workerpool.ready_workers`: For `atenet.router.route.duration`: * `ate.router.outcome` categorizes the route attempt result: `ok`, `cancelled`, `timeout`, `no_capacity`, `failed_precondition`, `lock_conflict`, `not_found`, `unavailable`, `rate_limited`, or `resume_error`. -* `ate.router.resume` indicates the singleflight execution state of actor resumption: `none` (actor already running), `triggered` (initiated cold activation), or `joined` (parked on in-flight activation). +* `ate.router.resume` indicates the singleflight execution state of actor resumption: `none` (the resume found the actor already running), `triggered` (this request completed a cold activation), `joined` (this request waited on another request's resume, which activated the actor), or `unknown` (the resume did not complete, so whether an activation ran is unknown). `ate.template.atespace` and `ate.template.name` hold `unknown` when the router has no template to name. For `ate.scheduler.eligible_workers`: * `ate.scheduling.constraint` categorizes the scheduling request constraint type: `none` (unconstrained), `selector` (actor or template label selectors specified), or `required_nodes` (pinned to specific node VMs). diff --git a/internal/ateattr/ateattr.go b/internal/ateattr/ateattr.go index 2e7a59126..3ecf8d231 100644 --- a/internal/ateattr/ateattr.go +++ b/internal/ateattr/ateattr.go @@ -56,6 +56,23 @@ const ( ActorVersionKey = attribute.Key("ate.actor.version") ) +// TemplateUnknown is the fallback for a template dimension the emitter cannot +// resolve. "unknown" is a legal atespace or template name, so a real object +// with that name collides with the fallback. Like ate.sandbox.class="unknown", +// the registry accepts that collision to keep the dimension bounded and +// non-empty. +const TemplateUnknown = "unknown" + +// NormalizeTemplateDimension returns dim, or TemplateUnknown when dim is empty. +// Only the router calls it today. The other metrics that carry ate.template.* +// emit the raw value. +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. @@ -240,12 +257,19 @@ const ( // Values for RouterResumeKey. const ( - // RouterResumeNone indicates the actor was already running (steady-state route). + // RouterResumeNone indicates the resume completed and found the actor already + // running (steady-state route). RouterResumeNone = "none" - // RouterResumeTriggered indicates this request won the singleflight lock and initiated cold activation. + // RouterResumeTriggered indicates this request won the singleflight lock and + // completed a cold activation. RouterResumeTriggered = "triggered" - // RouterResumeJoined indicates this request parked on an in-flight singleflight resume. + // RouterResumeJoined indicates this request waited on another request's + // singleflight resume, which completed a cold activation. RouterResumeJoined = "joined" + // RouterResumeUnknown indicates the resume did not complete, so the router + // cannot tell whether an activation ran. The resume failed, or the request + // stopped first, or the direction never resumes an actor. + RouterResumeUnknown = "unknown" ) // 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 274b4436f..3b882bdb2 100644 --- a/internal/ateattr/ateattr_test.go +++ b/internal/ateattr/ateattr_test.go @@ -418,6 +418,12 @@ func TestMetricLabelValues(t *testing.T) { {SnapshotPhaseTotal, "total"}, {SandboxClassUnknown, "unknown"}, + {TemplateUnknown, "unknown"}, + + {RouterResumeNone, "none"}, + {RouterResumeTriggered, "triggered"}, + {RouterResumeJoined, "joined"}, + {RouterResumeUnknown, "unknown"}, } for _, tt := range tests { t.Run(tt.want, func(t *testing.T) { @@ -891,3 +897,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: "template name preserved", in: "counter", want: "counter"}, + {name: "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) + } + }) + } +}