diff --git a/internal/env/supported_configurations.gen.go b/internal/env/supported_configurations.gen.go index 3601c992b85..d6b263a4377 100644 --- a/internal/env/supported_configurations.gen.go +++ b/internal/env/supported_configurations.gen.go @@ -67,6 +67,7 @@ var SupportedConfigurations = map[string]struct{}{ "DD_EXPERIMENTAL_FLAGGING_PROVIDER_ENABLED": {}, "DD_EXPERIMENTAL_PROPAGATE_PROCESS_TAGS_ENABLED": {}, "DD_EXTERNAL_ENV": {}, + "DD_FLAGGING_EVALUATION_COUNTS_ENABLED": {}, "DD_GIT_BRANCH": {}, "DD_GIT_COMMIT_AUTHOR_DATE": {}, "DD_GIT_COMMIT_AUTHOR_EMAIL": {}, diff --git a/internal/env/supported_configurations.json b/internal/env/supported_configurations.json index 15ca43133d2..f0049cbed4d 100644 --- a/internal/env/supported_configurations.json +++ b/internal/env/supported_configurations.json @@ -413,6 +413,13 @@ "default": null } ], + "DD_FLAGGING_EVALUATION_COUNTS_ENABLED": [ + { + "implementation": "A", + "type": "boolean", + "default": "true" + } + ], "DD_GIT_BRANCH": [ { "implementation": "A", diff --git a/openfeature/evaluator.go b/openfeature/evaluator.go index de581309b2d..58021cc3f82 100644 --- a/openfeature/evaluator.go +++ b/openfeature/evaluator.go @@ -34,9 +34,12 @@ type evaluationResult struct { Metadata map[string]any } -// evaluateFlag evaluates a feature flag with the given context. +// evaluateFlag evaluates a feature flag with the given context. The caller supplies the +// evaluation time (now) so a single timestamp is shared between the allocation time-window +// checks here and the EVP eval-time metadata stamped by DatadogProvider.evaluate — avoiding a +// second time.Now() on the evaluation path. // It returns the variant value, reason, and any error that occurred. -func evaluateFlag(flag *flag, defaultValue any, context map[string]any) evaluationResult { +func evaluateFlag(flag *flag, defaultValue any, context map[string]any, now time.Time) evaluationResult { if flag == nil { return evaluationResult{Value: defaultValue, Reason: of.DefaultReason} } @@ -48,8 +51,7 @@ func evaluateFlag(flag *flag, defaultValue any, context map[string]any) evaluati } } - // Evaluate allocations in order - first match wins - now := time.Now() + // Evaluate allocations in order - first match wins (using the caller-supplied eval time) for _, allocation := range flag.Allocations { split, matched, err := evaluateAllocation(allocation, context, now) if err != nil { @@ -80,7 +82,7 @@ func evaluateFlag(flag *flag, defaultValue any, context map[string]any) evaluati } // Build metadata for exposure tracking - metadata := make(map[string]any) + metadata := make(map[string]any, 3) metadata[metadataAllocationKey] = allocation.Key // Get doLog value (defaults to true if not specified) diff --git a/openfeature/evaluator_test.go b/openfeature/evaluator_test.go index d75b58540f9..0969c6130c8 100644 --- a/openfeature/evaluator_test.go +++ b/openfeature/evaluator_test.go @@ -703,7 +703,7 @@ func TestEvaluateFlag_VariantTypeMismatchReturnsParseError(t *testing.T) { }, } - result := evaluateFlag(flag, nil, map[string]any{"targetingKey": "user-123"}) + result := evaluateFlag(flag, nil, map[string]any{"targetingKey": "user-123"}, time.Now()) if result.Reason != of.ErrorReason { t.Errorf("expected ErrorReason, got %s", result.Reason) @@ -768,7 +768,7 @@ func TestEvaluateFlag_JSONFixtures(t *testing.T) { ctx["targetingKey"] = *tc.TargetingKey } - result := evaluateFlag(cfg.Flags[tc.Flag], tc.DefaultValue, ctx) + result := evaluateFlag(cfg.Flags[tc.Flag], tc.DefaultValue, ctx, time.Now()) if fmt.Sprintf("%v", result.Value) != fmt.Sprintf("%v", tc.Result.Value) { t.Errorf("value: got %v, want %v", result.Value, tc.Result.Value) diff --git a/openfeature/evp.go b/openfeature/evp.go new file mode 100644 index 00000000000..c13433a8849 --- /dev/null +++ b/openfeature/evp.go @@ -0,0 +1,81 @@ +// Unless explicitly stated otherwise all files in this repository are licensed +// under the Apache License Version 2.0. +// This product includes software developed at Datadog (https://www.datadoghq.com/). +// Copyright 2025 Datadog, Inc. + +package openfeature + +import ( + "bytes" + "context" + "fmt" + "io" + "net/http" + "net/url" + + jsoniter "github.com/json-iterator/go" + + "github.com/DataDog/dd-trace-go/v2/internal" + "github.com/DataDog/dd-trace-go/v2/internal/log" +) + +type evpClient struct { + httpClient *http.Client + agentURL *url.URL + jsonConfig jsoniter.API +} + +func newEVPClient() *evpClient { + agentURL := internal.AgentURLFromEnv() + var httpClient *http.Client + if agentURL.Scheme == "unix" { + httpClient = internal.UDSClient(agentURL.Path, defaultHTTPTimeout) + agentURL = internal.UnixDataSocketURL(agentURL.Path) + } else { + httpClient = internal.DefaultHTTPClient(defaultHTTPTimeout, false) + } + + return &evpClient{ + httpClient: httpClient, + agentURL: agentURL, + jsonConfig: jsoniter.Config{}.Froze(), + } +} + +func (c *evpClient) post(endpoint, eventName string, payload any) error { + if c == nil { + return fmt.Errorf("EVP client is not configured") + } + + var bytesBuffer bytes.Buffer + encoder := c.jsonConfig.NewEncoder(&bytesBuffer) + if err := encoder.Encode(payload); err != nil { + return fmt.Errorf("failed to encode %s payload: %w", eventName, err) + } + + u := *c.agentURL + u.Path = endpoint + requestURL := u.String() + + req, err := http.NewRequestWithContext(context.Background(), "POST", requestURL, &bytesBuffer) + if err != nil { + return fmt.Errorf("failed to create request: %w", err) + } + + req.Header.Set("Content-Type", "application/json") + req.Header.Set(evpSubdomainHeader, evpSubdomainValue) + + log.Debug("openfeature: sending %s events to %s", eventName, requestURL) + + resp, err := c.httpClient.Do(req) + if err != nil { + return fmt.Errorf("request failed: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + body, _ := io.ReadAll(io.LimitReader(resp.Body, 256)) + return fmt.Errorf("unexpected status code %d: %s", resp.StatusCode, string(body)) + } + return nil +} diff --git a/openfeature/exposure.go b/openfeature/exposure.go index cc1df0fef8c..ccd46b76bb8 100644 --- a/openfeature/exposure.go +++ b/openfeature/exposure.go @@ -6,22 +6,13 @@ package openfeature import ( - "bytes" "cmp" "container/list" - "context" - "fmt" - "io" "log/slog" - "net/http" - "net/url" "os" "sync" "time" - jsoniter "github.com/json-iterator/go" - - "github.com/DataDog/dd-trace-go/v2/internal" "github.com/DataDog/dd-trace-go/v2/internal/env" "github.com/DataDog/dd-trace-go/v2/internal/globalconfig" "github.com/DataDog/dd-trace-go/v2/internal/log" @@ -176,36 +167,27 @@ type exposureWriter struct { buffer []exposureEvent // Buffer for exposure events cache *exposureLRUCache // LRU deduplication cache flushInterval time.Duration - httpClient *http.Client - agentURL *url.URL + evp *evpClient context exposureContext ticker *time.Ticker stopChan chan struct{} stopped bool - jsonConfig jsoniter.API } // newExposureWriter creates a new exposure writer with the given configuration func newExposureWriter(config ProviderConfig) *exposureWriter { - agentURL := internal.AgentURLFromEnv() - var httpClient *http.Client - if agentURL.Scheme == "unix" { - httpClient = internal.UDSClient(agentURL.Path, defaultHTTPTimeout) - agentURL = internal.UnixDataSocketURL(agentURL.Path) - } else { - httpClient = internal.DefaultHTTPClient(defaultHTTPTimeout, false) - } + return newExposureWriterWithEVP(config, newEVPClient()) +} +func newExposureWriterWithEVP(config ProviderConfig, evp *evpClient) *exposureWriter { executable, _ := os.Executable() return &exposureWriter{ buffer: make([]exposureEvent, 0, 1<<8), // Initial capacity of 256 cache: newExposureLRUCache(defaultExposureCacheCapacity), flushInterval: cmp.Or(config.ExposureFlushInterval, defaultExposureFlushInterval), - httpClient: httpClient, - agentURL: agentURL, + evp: evp, stopChan: make(chan struct{}), - jsonConfig: jsoniter.Config{}.Froze(), context: exposureContext{ Service: cmp.Or(env.Get("DD_SERVICE"), globalconfig.ServiceName(), executable), Version: env.Get("DD_VERSION"), @@ -298,44 +280,7 @@ func (w *exposureWriter) flush() { // sendToAgent sends the exposure payload to the Datadog Agent via EVP proxy func (w *exposureWriter) sendToAgent(payload exposurePayload) error { - // Serialize payload - var bytesBuffer bytes.Buffer - encoder := w.jsonConfig.NewEncoder(&bytesBuffer) - if err := encoder.Encode(payload); err != nil { - return fmt.Errorf("failed to encode exposure payload: %w", err) - } - - // Build request URL - u := *w.agentURL - u.Path = exposureEndpoint - requestURL := u.String() - - // Create HTTP request - req, err := http.NewRequestWithContext(context.Background(), "POST", requestURL, &bytesBuffer) - if err != nil { - return fmt.Errorf("failed to create request: %w", err) - } - - // Set headers - req.Header.Set("Content-Type", "application/json") - req.Header.Set(evpSubdomainHeader, evpSubdomainValue) - - log.Debug("openfeature: sending exposure events to %s", requestURL) - - // Send request - resp, err := w.httpClient.Do(req) - if err != nil { - return fmt.Errorf("request failed: %w", err) - } - defer resp.Body.Close() - - // Check response status - if resp.StatusCode < 200 || resp.StatusCode >= 300 { - body, _ := io.ReadAll(io.LimitReader(resp.Body, 256)) - return fmt.Errorf("unexpected status code %d: %s", resp.StatusCode, string(body)) - } - - return nil + return w.evp.post(exposureEndpoint, "exposure", payload) } // stop stops the exposure writer and flushes any remaining events diff --git a/openfeature/exposure_hook.go b/openfeature/exposure_hook.go index 301f72a7437..31ad834e39a 100644 --- a/openfeature/exposure_hook.go +++ b/openfeature/exposure_hook.go @@ -19,6 +19,9 @@ const ( // Metadata keys for exposure tracking metadataAllocationKey = "dd.allocation.key" metadataDoLogKey = "dd.doLog" + // metadataEvalTimeKey carries the evaluation timestamp (UnixMilli, int64). It is stamped in + // DatadogProvider.evaluate at evaluation entry so EVP first/last bounds use eval-time. + metadataEvalTimeKey = "dd.eval.timestamp_ms" ) // exposureHook implements the OpenFeature Hook interface to track feature flag exposures. diff --git a/openfeature/flagevaluation.go b/openfeature/flagevaluation.go new file mode 100644 index 00000000000..6575ae9c96f --- /dev/null +++ b/openfeature/flagevaluation.go @@ -0,0 +1,889 @@ +// Unless explicitly stated otherwise all files in this repository are licensed +// under the Apache License Version 2.0. +// This product includes software developed at Datadog (https://www.datadoghq.com/). +// Copyright 2025 Datadog, Inc. + +package openfeature + +import ( + "cmp" + "fmt" + "log/slog" + "os" + "sort" + "strconv" + "sync" + "sync/atomic" + "time" + + "github.com/DataDog/dd-trace-go/v2/internal/env" + "github.com/DataDog/dd-trace-go/v2/internal/globalconfig" + "github.com/DataDog/dd-trace-go/v2/internal/log" + telemetrylog "github.com/DataDog/dd-trace-go/v2/internal/telemetry/log" + + of "github.com/open-feature/go-sdk/openfeature" +) + +const ( + // defaultFlagEvalFlushInterval is the flush interval for EVP flag evaluation events. + // Dedicated 10 s timer; separate from exposureWriter's 1 s interval. + defaultFlagEvalFlushInterval = 10 * time.Second + + // flagEvaluationEndpoint is the EVP proxy endpoint for flag evaluation events. + flagEvaluationEndpoint = "/evp_proxy/v2/api/v2/flagevaluation" + + // Context pruning limits — mirror worker.ts MAX_EVALUATION_CONTEXT_FIELDS / MAX_FIELD_LENGTH. + maxContextFields = 256 + maxFieldLength = 256 + + // Aggregation cap sizing inputs. + evalScaleTargetFlags = 2_500 + evalScaleFullBucketsPerFlag = 50 + evalScaleUsersPerFlag = 1_000 + evalScalePerFlagHeadroomMultiplier = 10 + evalScaleDegradedBucketsPerFlag = 10 + evalScaleFullBucketTarget = evalScaleTargetFlags * evalScaleFullBucketsPerFlag + evalScalePerFlagBucketTarget = evalScalePerFlagHeadroomMultiplier * evalScaleUsersPerFlag + evalScaleDegradedBucketTarget = evalScaleTargetFlags * evalScaleDegradedBucketsPerFlag + + // Aggregation caps. + // + // The aggregation path is full → degraded → drop(counted). degradedCap must hold the + // legitimate degraded cardinality at the target scale, or legitimate counts would be + // dropped. Degraded cardinality is based only on schema-visible dimensions retained by + // the degraded payload; OpenFeature reason is not an EVP field and is not part of bucket + // identity. + // + // Sizing arithmetic: + // - globalCap 131,072 (2^17) > evalScaleFullBucketTarget = 125,000 + // - perFlagCap 10,000 = evalScalePerFlagBucketTarget + // - degradedCap 32,768 (2^15) > evalScaleDegradedBucketTarget = 25,000 + // These are starting caps, not schema limits; overflow is counted and dropped. + defaultEvalGlobalCap = 131_072 // bounds full-tier buckets only; degraded is bounded separately + defaultEvalPerFlagCap = evalScalePerFlagBucketTarget + defaultEvalDegradedCap = 32_768 // bounds degraded buckets; overflow is counted and dropped + + // defaultEvalEventBufferSize bounds the async hand-off queue between the (hot-path) + // Finally hook and the background aggregation worker. On overflow the hook drops the + // event and increments a counter rather than blocking the evaluation. + defaultEvalEventBufferSize = 4096 +) + +// evaluationAggregationKey identifies one full-tier aggregation bucket. Every field is an +// EXACT, comparable string compared byte-for-byte by Go map equality, so distinct keys can +// NEVER alias into the same bucket: +// +// - The enumerable dimensions are schema-visible fields only. OpenFeature reason is not +// accepted by the flageval-worker schema and must never be hidden cardinality. +// - contextKey is the EXACT canonical type-tagged, length-delimited encoding of the pruned +// context (see canonicalContextKey). Because it is the full encoding — not a lossy 64-bit +// digest — two distinct contexts ALWAYS produce distinct contextKey strings: +// int 1 vs string "1" differ by type tag, and '='/'\n'-bearing values cannot fake a field +// boundary thanks to the length prefixes. There is no hash, so there is no hash collision, +// so a count can never be misattributed to the wrong context. Go's map hashes and compares +// the full struct key (including contextKey) natively. +// +// The contextKey string is stored once per full-tier bucket; its memory footprint is therefore +// bounded by the number of full-tier buckets (globalCap) and the pruned context size +// (256 fields × 256 chars), and measured in BenchmarkFlagEvaluationOTelPlusEVPParallel. +type evaluationAggregationKey struct { + flagKey string + variant string + allocationKey string + runtimeDefault bool + errorMessage string + targetingKey string + contextKey string // exact canonical encoding of the pruned context; comparable, not a digest +} + +// evaluationDegradedKey is the key for the degraded aggregation map in the +// full → degraded → drop path. It drops targeting key, context, and other full-only fields, +// keeping only schema-visible fields emitted by the degraded payload. When a NEW degraded +// bucket would exceed degradedCap, the count is dropped and counted. +type evaluationDegradedKey struct { + flagKey string + variant string + allocationKey string + runtimeDefault bool + errorMessage string +} + +// evaluationEntry holds per-window counts and time bounds for one aggregation bucket. +type evaluationEntry struct { + count int64 + firstEvaluation int64 // milliseconds — min across all recordings in this window + lastEvaluation int64 // milliseconds — max across all recordings in this window + runtimeDefault bool + // For full tier only: + targetingKey string + contextAttrs map[string]any + errorMessage string +} + +// observe records one more evaluation against an existing bucket: it bumps the count and +// widens the [firstEvaluation, lastEvaluation] window to include evaluationTimeMs. Every existing-bucket +// path across the aggregation tiers funnels through here so the count++/min/max logic lives once. +// evaluationTimeMs is captured before enqueue; concurrent evaluations can reach the worker out +// of timestamp order, so an existing bucket can legitimately observe an older timestamp. +func (e *evaluationEntry) observe(evaluationTimeMs int64) { + e.count++ + if evaluationTimeMs < e.firstEvaluation { + e.firstEvaluation = evaluationTimeMs + } + if evaluationTimeMs > e.lastEvaluation { + e.lastEvaluation = evaluationTimeMs + } +} + +// newEvaluationEntry returns a fresh bucket for evaluationTimeMs with count 1 and +// first==last==evaluationTimeMs. +// Callers set any tier-specific fields (runtimeDefault, targetingKey, contextAttrs, +// errorMessage) on the returned entry. +func newEvaluationEntry(evaluationTimeMs int64) *evaluationEntry { + return &evaluationEntry{ + count: 1, + firstEvaluation: evaluationTimeMs, + lastEvaluation: evaluationTimeMs, + } +} + +// flagEvaluationAggregator holds the two-tier aggregation maps (full → degraded → drop). +type flagEvaluationAggregator struct { + mu sync.Mutex + full map[evaluationAggregationKey]*evaluationEntry + degraded map[evaluationDegradedKey]*evaluationEntry + perFlagFull map[string]int // flagKey → count of full-fidelity entries for this flag + globalCount int + globalCap int + perFlagCap int + degradedCap int + // dropped counts evaluations whose count was lost because a NEW degraded bucket would have + // exceeded degradedCap. It is the observable signal that legitimate counts are being + // dropped and that degradedCap should be raised. It is distinct from flagEvaluationWriter.dropped + // (which counts async-queue backpressure drops). + droppedDegradedOverflow int64 +} + +// flagEvaluationEvent matches flagevaluation.json — required fields always present; +// optional fields use omitempty (absent = schema-valid for the degraded tier). +type flagEvaluationEvent struct { + Timestamp int64 `json:"timestamp"` + Flag flagEvalFlag `json:"flag"` + FirstEvaluation int64 `json:"first_evaluation"` + LastEvaluation int64 `json:"last_evaluation"` + EvaluationCount int64 `json:"evaluation_count"` + RuntimeDefault bool `json:"runtime_default_used,omitempty"` + TargetingKey string `json:"targeting_key,omitempty"` + Variant *flagEvalVariant `json:"variant,omitempty"` + Allocation *flagEvalAllocation `json:"allocation,omitempty"` + Error *flagEvalError `json:"error,omitempty"` + Context *flagEvalEventContext `json:"context,omitempty"` +} + +// flagEvalFlag holds the flag key. +type flagEvalFlag struct { + Key string `json:"key"` +} + +// flagEvalVariant holds the variant key. +type flagEvalVariant struct { + Key string `json:"key"` +} + +// flagEvalAllocation holds the allocation key. +type flagEvalAllocation struct { + Key string `json:"key"` +} + +// flagEvalError holds the error message. +type flagEvalError struct { + Message string `json:"message"` +} + +// flagEvalEventContext holds the per-event context object. +type flagEvalEventContext struct { + Evaluation map[string]any `json:"evaluation,omitempty"` + DD *flagEvalContextDD `json:"dd,omitempty"` +} + +// flagEvalContextDD holds the Datadog-specific context inside per-event context.dd. +type flagEvalContextDD struct { + Service string `json:"service,omitempty"` +} + +// flagEvaluationPayload is the SDK's EVP flagevaluation batch envelope. +// Keep JSON field names aligned with the worker contract; do not vendor the +// worker schema here, because dd-source owns that contract. +type flagEvaluationPayload struct { + Context flagEvalDDContext `json:"context"` + FlagEvaluations []flagEvaluationEvent `json:"flagEvaluations"` +} + +// flagEvalDDContext carries service/env/version for the batch-level context. +type flagEvalDDContext struct { + Service string `json:"service"` + Env string `json:"env,omitempty"` + Version string `json:"version,omitempty"` +} + +// flagEvaluationWriter manages aggregation and periodic flushing of EVP flag evaluation events. +type flagEvaluationWriter struct { + aggregator flagEvaluationAggregator + flushInterval time.Duration + evp *evpClient + ddContext flagEvalDDContext // service/env/version — same source as exposureContext + ticker *time.Ticker + stopChan chan struct{} + stopped atomic.Bool // single idempotency gate for stop(); also read lock-free by record() + + // Asynchronous hand-off: the Finally hook enqueues a bounded snapshot here; a single + // background worker (started in start()) drains it and performs aggregate/flush off the + // evaluation hot path. events is bounded; on overflow the hook drops + // the event and bumps dropped — best-effort telemetry that never blocks the request. + events chan evalEvent + dropped atomic.Int64 + workerDone chan struct{} + enqueueMu sync.RWMutex +} + +// evalEvent is the bounded snapshot the Finally hook hands to the worker. +type evalEvent struct { + d evalDetails + evaluationContext of.EvaluationContext + evaluationTimeMs int64 +} + +// evalDetails holds extracted flag evaluation fields for EVP aggregation. +// Used only by flagEvaluationHook; does NOT replace extraction in flageval_metrics.go. +type evalDetails struct { + flagKey string + variant string + allocationKey string + targetingKey string + errorMessage string + runtimeDefault bool + evalTimeMs int64 +} + +// newFlagEvaluationWriter creates a new flag evaluation writer. +func newFlagEvaluationWriter(config ProviderConfig) *flagEvaluationWriter { + return newFlagEvaluationWriterWithEVP(config, newEVPClient()) +} + +func newFlagEvaluationWriterWithEVP(config ProviderConfig, evp *evpClient) *flagEvaluationWriter { + executable, _ := os.Executable() + + flushInterval := cmp.Or(config.FlagEvaluationFlushInterval, defaultFlagEvalFlushInterval) + + return &flagEvaluationWriter{ + flushInterval: flushInterval, + evp: evp, + stopChan: make(chan struct{}), + workerDone: make(chan struct{}), + events: make(chan evalEvent, defaultEvalEventBufferSize), + ddContext: flagEvalDDContext{ + Service: cmp.Or(env.Get("DD_SERVICE"), globalconfig.ServiceName(), executable), + Version: env.Get("DD_VERSION"), + Env: env.Get("DD_ENV"), + }, + aggregator: flagEvaluationAggregator{ + full: make(map[evaluationAggregationKey]*evaluationEntry), + degraded: make(map[evaluationDegradedKey]*evaluationEntry), + perFlagFull: make(map[string]int), + globalCap: defaultEvalGlobalCap, + perFlagCap: defaultEvalPerFlagCap, + degradedCap: defaultEvalDegradedCap, + }, + } +} + +// start begins the periodic flushing — called from InitWithContext(), NOT from the constructor. +// Mirrors exposure.go's start() goroutine + panic recovery pattern. +func (w *flagEvaluationWriter) start() { + w.ticker = time.NewTicker(w.flushInterval) + go func() { + defer func() { + if r := recover(); r != nil { + log.Error("openfeature: flag evaluation writer recovered panic: %s", r) + var errAttr slog.Attr + if err, ok := r.(error); ok { + errAttr = slog.Any("panic", telemetrylog.NewSafeError(err)) + } else { + errAttr = slog.Any("panic", r) + } + telemetrylog.Error("openfeature: flag evaluation writer recovered panic", errAttr) + } + // Always signal completion so stop() unblocks, even on panic. + close(w.workerDone) + }() + + // Single owner of aggregate/flush. The hook enqueues only bounded snapshots. + for { + select { + case ev := <-w.events: + w.aggregate(ev) + case <-w.ticker.C: + w.flush() + case <-w.stopChan: + w.drainAndFlush() + return + } + } + }() +} + +// stop stops the flush ticker and marks the writer as stopped. +func (w *flagEvaluationWriter) stop() { + w.enqueueMu.Lock() + // Single idempotency gate: the atomic Swap is the guard for both "mark stopped" and the + // downstream close(stopChan). enqueueMu prevents a record() call that observed stopped=false + // from sending into events after the worker has drained and exited. + if w.stopped.Swap(true) { + w.enqueueMu.Unlock() + return + } + + // Signal the worker to drain the queue and do a final flush. + close(w.stopChan) + w.enqueueMu.Unlock() + if w.ticker != nil { + // Worker was started: wait for its final flush before returning, then stop the + // ticker. (ticker is set only in start(), so it gates "was the worker started".) + <-w.workerDone + w.ticker.Stop() + } + + log.Debug("openfeature: flag evaluation writer stopped") +} + +// flush drains the aggregator, assembles per-tier events, and sends them to the agent. +func (w *flagEvaluationWriter) flush() { + // Surface best-effort backpressure drops (queue full) as an observable signal. + if d := w.dropped.Swap(0); d > 0 { + log.Debug("openfeature: flag evaluation queue full — dropped %d evaluation(s) under backpressure (best-effort telemetry)", d) + } + + w.aggregator.mu.Lock() + + // Under lock: drain both maps. + full := w.aggregator.full + degraded := w.aggregator.degraded + + // Surface degraded-overflow drops so an undersized degradedCap is observable rather than + // a silent loss of legitimate counts. + degradedOverflow := w.aggregator.droppedDegradedOverflow + + if (len(full) + len(degraded)) == 0 { + w.aggregator.droppedDegradedOverflow = 0 + w.aggregator.mu.Unlock() + if degradedOverflow > 0 { + log.Warn("openfeature: degraded aggregation tier full — dropped %d evaluation(s); raise degradedCap (best-effort telemetry)", degradedOverflow) + } + return + } + + // Reset maps. + w.aggregator.full = make(map[evaluationAggregationKey]*evaluationEntry) + w.aggregator.degraded = make(map[evaluationDegradedKey]*evaluationEntry) + w.aggregator.perFlagFull = make(map[string]int) + w.aggregator.globalCount = 0 + w.aggregator.droppedDegradedOverflow = 0 + + w.aggregator.mu.Unlock() + + if degradedOverflow > 0 { + log.Warn("openfeature: degraded aggregation tier full — dropped %d evaluation(s); raise degradedCap (best-effort telemetry)", degradedOverflow) + } + + flushTimeMs := time.Now().UnixMilli() + var events []flagEvaluationEvent + + // Full tier: required fields + variant + allocation + targeting_key + context + error. + // runtime_default_used decorates this tier when the caller default was returned. + for key, e := range full { + ev := baseFlagEvaluationEvent(key.flagKey, e, flushTimeMs) + ev.RuntimeDefault = e.runtimeDefault + ev.TargetingKey = e.targetingKey + if key.variant != "" { + ev.Variant = &flagEvalVariant{Key: key.variant} + } + if key.allocationKey != "" { + ev.Allocation = &flagEvalAllocation{Key: key.allocationKey} + } + if e.errorMessage != "" { + ev.Error = &flagEvalError{Message: e.errorMessage} + } + if len(e.contextAttrs) > 0 { + ev.Context = &flagEvalEventContext{Evaluation: e.contextAttrs} + } + events = append(events, ev) + } + + // Degraded tier: required fields + variant + allocation + error; no targeting_key, no context. + // runtime_default_used decorates this tier when the caller default was returned. + for key, e := range degraded { + ev := baseFlagEvaluationEvent(key.flagKey, e, flushTimeMs) + ev.RuntimeDefault = e.runtimeDefault + if key.variant != "" { + ev.Variant = &flagEvalVariant{Key: key.variant} + } + if key.allocationKey != "" { + ev.Allocation = &flagEvalAllocation{Key: key.allocationKey} + } + if e.errorMessage != "" { + ev.Error = &flagEvalError{Message: e.errorMessage} + } + events = append(events, ev) + } + + if len(events) == 0 { + return + } + + payload := flagEvaluationPayload{ + Context: w.ddContext, + FlagEvaluations: events, + } + + if err := w.sendToAgent(payload); err != nil { + log.Error("openfeature: failed to send flag evaluation events: %v", err.Error()) + } else { + log.Debug("openfeature: successfully sent %d flag evaluation events", len(events)) + } +} + +// baseFlagEvaluationEvent builds a flagEvaluationEvent with ONLY the five required schema +// fields (timestamp, flag.key, first/last evaluation, evaluation_count). It is tier-agnostic +// and sets no optional field — RuntimeDefault and the rest are decoration applied by each tier +// loop in flush() after the call. +func baseFlagEvaluationEvent(flagKey string, e *evaluationEntry, flushTimeMs int64) flagEvaluationEvent { + return flagEvaluationEvent{ + Timestamp: flushTimeMs, + Flag: flagEvalFlag{Key: flagKey}, + FirstEvaluation: e.firstEvaluation, + LastEvaluation: e.lastEvaluation, + EvaluationCount: e.count, + } +} + +// record runs on the evaluation hot path (the Finally hook). It does only cheap scalar +// extraction plus an immutable context handoff, then a non-blocking enqueue — no aggregation +// or context flattening happens here; the background worker does that. If the queue is full +// the event is dropped and counted (best-effort), never blocking the evaluation. Called from +// the Finally hook after every evaluation. +func (w *flagEvaluationWriter) record(hookContext of.HookContext, details of.InterfaceEvaluationDetails) { + w.enqueueMu.RLock() + defer w.enqueueMu.RUnlock() + + // Post-stop no-op: after stop() the worker no longer drains w.events, so enqueuing would + // silently lose the event. Check the atomic gate lock-free (reading under the aggregator + // lock would add hot-path contention) and count the event as dropped so it stays observable. + if w.stopped.Load() { + w.dropped.Add(1) + return + } + if len(w.events) == cap(w.events) { + w.dropped.Add(1) + return + } + d := extractEvalDetails(hookContext, details) + evaluationTimeMs := d.evalTimeMs + if evaluationTimeMs == 0 { + evaluationTimeMs = time.Now().UnixMilli() + } + ev := evalEvent{ + d: d, + evaluationContext: hookContext.EvaluationContext(), + evaluationTimeMs: evaluationTimeMs, + } + select { + case w.events <- ev: + default: + w.dropped.Add(1) + } +} + +// aggregate updates the aggregator. It runs only on the writer's single worker goroutine. +func (w *flagEvaluationWriter) aggregate(ev evalEvent) { + contextAttrs := flattenAndPruneContext(ev.evaluationContext.Attributes()) + w.aggregator.add(ev.d, contextAttrs, ev.evaluationTimeMs) +} + +// flattenAndPruneContext produces the pruned context map for EVP aggregation in a single +// traversal of the flattened keyspace. It merges the +// two former steps — flattenContext (flatten.go) + pruneContext — into one pass with the SAME +// pruned output: +// +// 1. Flatten nested objects into a single-level dot-notation map (reusing flattenRecursive, so +// the flatten semantics stay identical to the exposure path which still calls +// flattenContext directly — that caller is unchanged). +// 2. Apply the deterministic prune: sort the flattened keys, then keep the first +// maxContextFields that are not oversized strings (>maxFieldLength). +// +// Allocation win: when the flattened context already fits the limits (the common case — fewer +// than maxContextFields fields and no oversized string), the flattened map is returned DIRECTLY, +// so the separate pruned-output map that the old flatten→prune pipeline always allocated is +// elided. The pruned map is allocated only when trimming actually changes the result. Output is +// byte-for-byte identical to the previous flattenContext+pruneContext pipeline: same surviving +// keys, same 256/256 limits, same deterministic ordering of the cut. +func flattenAndPruneContext(attrs map[string]any) map[string]any { + if len(attrs) == 0 { + return nil + } + if contextFitsWithoutFlattening(attrs) { + return attrs + } + + flat := make(map[string]any, len(attrs)) + flattenRecursive("", attrs, flat) + if len(flat) == 0 { + return nil + } + + // Determine whether any pruning is actually required: an over-cap field count or any + // oversized string value. If neither, the flattened map already IS the pruned result — + // return it directly and skip allocating a second map. + needsPrune := len(flat) > maxContextFields + if !needsPrune { + for _, v := range flat { + if s, ok := v.(string); ok && len(s) > maxFieldLength { + needsPrune = true + break + } + } + } + if !needsPrune { + return flat + } + + // Deterministic prune: sort keys, then keep the first maxContextFields non-oversized values. + // Sorting BEFORE the oversized-string skip and the field cap makes the kept subset stable + // across calls (Go map iteration is randomized), so logically-identical contexts always + // prune to the same subset and the same canonicalContextKey. + keys := make([]string, 0, len(flat)) + for k := range flat { + keys = append(keys, k) + } + sort.Strings(keys) + + out := make(map[string]any, min(len(flat), maxContextFields)) + count := 0 + for _, k := range keys { + if count >= maxContextFields { + break + } + v := flat[k] + if s, ok := v.(string); ok && len(s) > maxFieldLength { + // Skip oversized string values (matches worker.ts pruneFields behavior). + continue + } + out[k] = v + count++ + } + if len(out) == 0 { + return nil + } + return out +} + +func contextFitsWithoutFlattening(attrs map[string]any) bool { + if len(attrs) > maxContextFields { + return false + } + for _, v := range attrs { + switch x := v.(type) { + case string: + if len(x) > maxFieldLength { + return false + } + case int, int8, int16, int32, int64, + uint, uint8, uint16, uint32, uint64, + float32, float64, bool: + default: + return false + } + } + return true +} + +// drainAndFlush processes any buffered events and performs a final flush. Called by the +// worker when stopping so a final batch is not lost on shutdown. +func (w *flagEvaluationWriter) drainAndFlush() { + for { + select { + case ev := <-w.events: + w.aggregate(ev) + default: + w.flush() + return + } + } +} + +// sendToAgent sends the flag evaluation payload to the Datadog Agent via EVP proxy. +// Reuses evpSubdomainHeader / evpSubdomainValue constants from exposure.go. +func (w *flagEvaluationWriter) sendToAgent(payload flagEvaluationPayload) error { + return w.evp.post(flagEvaluationEndpoint, "flag evaluation", payload) +} + +// add records one evaluation observation into the appropriate aggregation tier. +// Must be called WITHOUT the aggregator lock held (it acquires the lock internally). +// Routes observations through full → degraded → drop(counted). +// +// Per-flag attempt counting: perFlagFull[flag] is incremented on every call for a flag +// (whether or not a full-tier bucket is actually created). This ensures that once +// globalCap is full, a flag that accumulates enough attempts (>= perFlagCap) still +// follows the degraded path — keeping the per-flag overflow path alive even after the +// global full-tier cap is exhausted. +func (a *flagEvaluationAggregator) add(d evalDetails, contextAttrs map[string]any, evaluationTimeMs int64) { + a.mu.Lock() + defer a.mu.Unlock() + + // Build the full key from schema-visible dimensions including the canonical context encoding. + // No hash, so distinct contexts get distinct buckets. + fullKey := evaluationAggregationKey{ + flagKey: d.flagKey, + variant: d.variant, + allocationKey: d.allocationKey, + runtimeDefault: d.runtimeDefault, + errorMessage: d.errorMessage, + targetingKey: d.targetingKey, + contextKey: canonicalContextKey(contextAttrs), + } + + // Fast path: this exact full-tier bucket already exists → increment its count. Because + // contextKey is the full canonical encoding (not a digest), this fast path is hit only by a + // genuinely identical pruned context — never by an aliasing collision. + if e, ok := a.full[fullKey]; ok { + e.observe(evaluationTimeMs) + return + } + + // Check per-flag cap. + if a.perFlagFull[d.flagKey] >= a.perFlagCap { + // perFlagCap exceeded — route to degraded tier. + a.addToDegraded(d, evaluationTimeMs) + return + } + + // Per-flag cap not yet reached. Increment the attempt count for this flag + // regardless of whether we can actually create a full-tier bucket. This ensures + // the degraded overflow path activates correctly even when globalCap is full. + a.perFlagFull[d.flagKey]++ + + // Check globalCap before creating a new full-tier bucket. + if a.globalCount >= a.globalCap { + // Global full-tier cap full — count must not be lost. Route into the degraded tier + // (which drops targeting_key + context), sized to hold the legitimate degraded + // cardinality at the target scale. The per-flag attempt counter was + // already incremented above; once it reaches perFlagCap this flag routes through + // addToDegraded directly as well. + a.addToDegraded(d, evaluationTimeMs) + return + } + + // New full-tier entry. + a.full[fullKey] = &evaluationEntry{ + count: 1, + firstEvaluation: evaluationTimeMs, + lastEvaluation: evaluationTimeMs, + runtimeDefault: d.runtimeDefault, + targetingKey: d.targetingKey, + contextAttrs: contextAttrs, + errorMessage: d.errorMessage, + } + a.globalCount++ +} + +// addToDegraded adds an entry to the degraded map (drops targeting_key + context). +// Called with the aggregator lock held. When a NEW degraded bucket would exceed degradedCap, +// the evaluation's count is DROPPED and counted (droppedDegradedOverflow). degradedCap is +// sized (defaultEvalDegradedCap) to hold the legitimate degraded cardinality at the target +// scale, so this drop only fires under cardinality far beyond that target (e.g. an unbounded +// dynamic/abusive flag key) and the dropped counter makes such overflow observable. +func (a *flagEvaluationAggregator) addToDegraded(d evalDetails, evaluationTimeMs int64) { + degKey := evaluationDegradedKey{ + flagKey: d.flagKey, + variant: d.variant, + allocationKey: d.allocationKey, + runtimeDefault: d.runtimeDefault, + errorMessage: d.errorMessage, + } + + if e, ok := a.degraded[degKey]; ok { + e.observe(evaluationTimeMs) + return + } + + // New degraded bucket — check degradedCap. + if len(a.degraded) >= a.degradedCap { + // degradedCap exceeded — terminal tier full. Drop the count but keep it observable so an + // undersized cap surfaces in the flush warning instead of silently losing legitimate data. + a.droppedDegradedOverflow++ + return + } + + e := newEvaluationEntry(evaluationTimeMs) + e.runtimeDefault = d.runtimeDefault + e.errorMessage = d.errorMessage + a.degraded[degKey] = e +} + +// context value type discriminators for the canonical key encoding. Each distinct Go type +// gets a distinct tag byte so that, e.g., int 1 and string "1" cannot render identically. +const ( + ctxTagString byte = 's' + ctxTagBool byte = 'b' + ctxTagInt byte = 'i' + ctxTagInt64 byte = 'l' + ctxTagInt32 byte = 'j' + ctxTagFloat64 byte = 'f' + ctxTagFloat32 byte = 'g' + ctxTagOther byte = 'o' +) + +// canonicalContextKey builds the EXACT, comparable string key for the pruned context map, +// used as the contextKey field of evaluationAggregationKey. +// +// The encoding is CANONICAL — each field is a length-delimited key followed by a type-tag byte +// and a length-delimited value — so distinct contexts cannot ALIAS by construction (int 1 vs +// string "1" differ by tag; '=' / '\n' cannot fake a field boundary). Unlike the prior FNV-1a +// digest, the full encoding is emitted AS THE KEY, so Go's map compares it byte-for-byte: there +// is no hash and therefore no hash collision, so distinct contexts ALWAYS land in distinct +// full-tier buckets. The returned string is stored once per full-tier bucket. +func canonicalContextKey(attrs map[string]any) string { + if len(attrs) == 0 { + return "" + } + // Encode over a deterministic key ordering. Go map iteration is randomized, and the + // concatenated encoding is order-sensitive, so ranging the map directly would produce a + // different key for identical contexts and fragment aggregation buckets. + keys := make([]string, 0, len(attrs)) + for k := range attrs { + keys = append(keys, k) + } + sort.Strings(keys) + // Build the encoding into a single buffer, then convert once to a string for the key. The + // per-field append uses the same canonical, allocation-light path as before. + var buf []byte + for _, k := range keys { + buf = appendLengthDelimited(buf, []byte(k)) // length-delimited key + buf = appendContextValue(buf, attrs[k]) // tag + length-delimited value + } + return string(buf) +} + +// appendLengthDelimited writes a fixed-width 8-byte big-endian length prefix followed by the +// raw bytes, so the boundary between fields is unambiguous regardless of the byte content. +func appendLengthDelimited(buf, b []byte) []byte { + var lenBuf [8]byte + n := uint64(len(b)) + for i := range 8 { + lenBuf[7-i] = byte(n) + n >>= 8 + } + buf = append(buf, lenBuf[:]...) + return append(buf, b...) +} + +// appendContextValue appends a CANONICAL, length-delimited rendering of v to buf: a type-tag +// byte (distinct per Go type) followed by a length-delimited rendered value. This avoids +// allocation for the common scalar types; rare/complex types fall back to a type-qualified +// fmt rendering. The encoding only needs to be deterministic within a run and collision-free +// across distinct values. +func appendContextValue(buf []byte, v any) []byte { + var scratch [32]byte + tmp := scratch[:0] + var tag byte + switch x := v.(type) { + case string: + tag = ctxTagString + tmp = append(tmp, x...) + case bool: + tag = ctxTagBool + tmp = strconv.AppendBool(tmp, x) + case int: + tag = ctxTagInt + tmp = strconv.AppendInt(tmp, int64(x), 10) + case int64: + tag = ctxTagInt64 + tmp = strconv.AppendInt(tmp, x, 10) + case int32: + tag = ctxTagInt32 + tmp = strconv.AppendInt(tmp, int64(x), 10) + case float64: + tag = ctxTagFloat64 + tmp = strconv.AppendFloat(tmp, x, 'g', -1, 64) + case float32: + tag = ctxTagFloat32 + tmp = strconv.AppendFloat(tmp, float64(x), 'g', -1, 32) + default: + tag = ctxTagOther + tmp = fmt.Appendf(tmp, "%T:%v", x, x) + } + buf = append(buf, tag) + return appendLengthDelimited(buf, tmp) +} + +// pruneContext applies 256-field / 256-char limits before buffering. +// Mirrors worker.ts MAX_EVALUATION_CONTEXT_FIELDS / MAX_FIELD_LENGTH exactly. +// Must be called AFTER flattenContext() (from flatten.go) to expand nested objects first. +// +// The kept subset is DETERMINISTIC: keys are sorted BEFORE the oversized-string skip and the +// 256-field cap are applied, so two logically-identical contexts always prune to the exact +// same subset (and therefore the same canonicalContextKey). Ranging the map directly (Go map +// iteration is randomized) would cut a different 256-field subset each call and fragment +// otherwise-identical contexts into separate aggregation buckets. +func pruneContext(raw map[string]any) map[string]any { + if len(raw) == 0 { + return nil + } + keys := make([]string, 0, len(raw)) + for k := range raw { + keys = append(keys, k) + } + sort.Strings(keys) + + out := make(map[string]any, min(len(raw), maxContextFields)) + count := 0 + for _, k := range keys { + if count >= maxContextFields { + break + } + v := raw[k] + if s, ok := v.(string); ok && len(s) > maxFieldLength { + // Skip oversized string values (matches worker.ts pruneFields behavior). + // Applied against the deterministic ordering so the kept subset is stable. + continue + } + out[k] = v + count++ + } + if len(out) == 0 { + return nil + } + return out +} + +// extractEvalDetails extracts EVP-relevant fields from hook context and evaluation details. +// This helper is used only by flagEvaluationHook — it does NOT replace the extraction in +// flageval_metrics.go (that file is left untouched to preserve the OTel path). +func extractEvalDetails(hookContext of.HookContext, details of.InterfaceEvaluationDetails) evalDetails { + allocationKey, _ := details.FlagMetadata[metadataAllocationKey].(string) + // Prefer OpenFeature's human-readable ErrorMessage; fall back to the ErrorCode string only + // when ErrorMessage is empty (some providers populate just the code). + errMsg := details.ErrorMessage + if errMsg == "" && details.ErrorCode != "" { + errMsg = string(details.ErrorCode) + } + evalTimeMs, _ := details.FlagMetadata[metadataEvalTimeKey].(int64) + return evalDetails{ + flagKey: hookContext.FlagKey(), + variant: details.Variant, + allocationKey: allocationKey, + targetingKey: hookContext.EvaluationContext().TargetingKey(), + errorMessage: errMsg, + runtimeDefault: isRuntimeDefault(details), + evalTimeMs: evalTimeMs, + } +} diff --git a/openfeature/flagevaluation_hook.go b/openfeature/flagevaluation_hook.go new file mode 100644 index 00000000000..f6b169d724c --- /dev/null +++ b/openfeature/flagevaluation_hook.go @@ -0,0 +1,53 @@ +// Unless explicitly stated otherwise all files in this repository are licensed +// under the Apache License Version 2.0. +// This product includes software developed at Datadog (https://www.datadoghq.com/). +// Copyright 2025 Datadog, Inc. + +package openfeature + +import ( + "context" + + of "github.com/open-feature/go-sdk/openfeature" +) + +// flagEvaluationHook implements the OpenFeature Hook interface to record EVP flagevaluation events. +// It uses the Finally hook stage (same as flagEvalHook) to cover success, error, and default paths. +// Finally fires on error/default paths, unlike After. +type flagEvaluationHook struct { + of.UnimplementedHook + writer *flagEvaluationWriter +} + +// newFlagEvaluationHook creates a new EVP flag evaluation hook. +func newFlagEvaluationHook(w *flagEvaluationWriter) *flagEvaluationHook { + return &flagEvaluationHook{writer: w} +} + +// Finally is called after every flag evaluation (success or error). +// Using Finally (not After) ensures error-path and provider-not-ready evaluations are counted. +// Mirrors flageval_metrics.go's Finally stage; the EVP hook is a separate registered hook. +func (h *flagEvaluationHook) Finally( + _ context.Context, + hookContext of.HookContext, + details of.InterfaceEvaluationDetails, + _ of.HookHints, +) { + // Do NOT gate buffering on the evaluation context. In real servers ctx is + // frequently the request context, which may already be cancelled by the time + // Finally runs — and record() is a non-blocking in-memory add with no network + // call. Gating on ctx.Done() would silently drop legitimate evaluation counts + // for cancelled-request evals. + if h.writer == nil { + return + } + h.writer.record(hookContext, details) +} + +// isRuntimeDefault returns true when this provider path returns the caller's supplied default. +// The primary signal is an absent variant key. Type mismatches are the exception: the provider +// may have produced a real variant, but the OpenFeature SDK returns the caller default after +// conversion fails. +func isRuntimeDefault(details of.InterfaceEvaluationDetails) bool { + return details.Variant == "" || details.ErrorCode == of.TypeMismatchCode +} diff --git a/openfeature/flagevaluation_hook_test.go b/openfeature/flagevaluation_hook_test.go new file mode 100644 index 00000000000..4fb6979f2fa --- /dev/null +++ b/openfeature/flagevaluation_hook_test.go @@ -0,0 +1,352 @@ +// Unless explicitly stated otherwise all files in this repository are licensed +// under the Apache License Version 2.0. +// This product includes software developed at Datadog (https://www.datadoghq.com/). +// Copyright 2025 Datadog, Inc. + +package openfeature + +import ( + "context" + "testing" + "time" + + of "github.com/open-feature/go-sdk/openfeature" +) + +// setupTestWriter creates a flagEvaluationWriter configured for unit testing. +// The writer uses a large flush interval (24 h) so no automatic flush fires during tests. +func setupTestWriter(t *testing.T) *flagEvaluationWriter { + t.Helper() + return &flagEvaluationWriter{ + flushInterval: 24 * time.Hour, // effectively disabled; tests control flush manually + stopChan: make(chan struct{}), + events: make(chan evalEvent, defaultEvalEventBufferSize), + aggregator: flagEvaluationAggregator{ + full: make(map[evaluationAggregationKey]*evaluationEntry), + degraded: make(map[evaluationDegradedKey]*evaluationEntry), + perFlagFull: make(map[string]int), + globalCap: 10, + perFlagCap: 3, + degradedCap: 3, + }, + } +} + +// makeHookContext creates an of.HookContext for testing. +func makeHookContext(flagKey string, targetingKey string, attrs map[string]any) of.HookContext { + evalCtx := of.NewEvaluationContext(targetingKey, attrs) + return of.NewHookContext( + flagKey, + of.Boolean, + false, + of.ClientMetadata{}, + of.Metadata{}, + evalCtx, + ) +} + +// makeEvalDetails constructs an InterfaceEvaluationDetails for hook testing. +func makeEvalDetails(variant string, reason of.Reason, errorCode of.ErrorCode, metadata ...of.FlagMetadata) of.InterfaceEvaluationDetails { + d := of.InterfaceEvaluationDetails{ + EvaluationDetails: of.EvaluationDetails{ + ResolutionDetail: of.ResolutionDetail{ + Variant: variant, + Reason: reason, + ErrorCode: errorCode, + }, + }, + } + if len(metadata) > 0 { + d.FlagMetadata = metadata[0] + } + return d +} + +type panicStringer struct{} + +func (panicStringer) String() string { + panic("String must not be called when the queue is already full") +} + +// TestIsRuntimeDefault verifies the runtime-default detection rule. +// Signal: absent variant key. Our evaluator sets a variant ONLY on a matched +// allocation (TARGETING_MATCH/SPLIT/STATIC); every DEFAULT/DISABLED/ERROR path +// leaves the variant empty. A present variant therefore means a real assignment, +// never a default — regardless of the reported reason. +func TestIsRuntimeDefault(t *testing.T) { + tests := []struct { + name string + details of.InterfaceEvaluationDetails + want bool + }{ + { + name: "empty variant is runtime default", + details: makeEvalDetails("", of.TargetingMatchReason, ""), + want: true, + }, + { + name: "empty variant with DEFAULT reason is runtime default", + details: makeEvalDetails("", of.DefaultReason, ""), + want: true, + }, + { + name: "empty variant with DISABLED reason is runtime default", + details: makeEvalDetails("", of.DisabledReason, ""), + want: true, + }, + { + name: "empty variant with ERROR reason is runtime default", + details: makeEvalDetails("", of.ErrorReason, of.FlagNotFoundCode), + want: true, + }, + { + name: "variant present with TARGETING_MATCH is NOT runtime default", + details: makeEvalDetails("on", of.TargetingMatchReason, ""), + want: false, + }, + { + name: "variant present with SPLIT is NOT runtime default", + details: makeEvalDetails("variant-a", of.SplitReason, ""), + want: false, + }, + { + name: "variant present with STATIC is NOT runtime default", + details: makeEvalDetails("on", of.StaticReason, ""), + want: false, + }, + { + // Divergent case: a present variant means a real assignment even if the + // reason is DISABLED. The old secondary reason-clause wrongly returned true. + name: "variant present with DISABLED reason is NOT runtime default", + details: makeEvalDetails("on", of.DisabledReason, ""), + want: false, + }, + { + // Divergent case: a present variant means a real assignment even if the + // reason is DEFAULT. The old secondary reason-clause wrongly returned true. + name: "variant present with DEFAULT reason is NOT runtime default", + details: makeEvalDetails("on", of.DefaultReason, ""), + want: false, + }, + { + name: "variant present with TYPE_MISMATCH is runtime default", + details: makeEvalDetails("on", of.ErrorReason, of.TypeMismatchCode), + want: true, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + got := isRuntimeDefault(tc.details) + if got != tc.want { + t.Errorf("isRuntimeDefault() = %v, want %v", got, tc.want) + } + }) + } +} + +// TestExtractEvalDetailsReadsEvalTime verifies provider-stamped evaluation time +// is read into evalDetails.evalTimeMs, and is 0 when absent. +func TestExtractEvalDetailsReadsEvalTime(t *testing.T) { + const evalTime int64 = 1_717_171_717_123 + tests := []struct { + name string + md of.FlagMetadata + want int64 + }{ + {"eval-time present", of.FlagMetadata{metadataEvalTimeKey: evalTime}, evalTime}, + {"eval-time present alongside allocation key", of.FlagMetadata{metadataAllocationKey: "alloc-1", metadataEvalTimeKey: evalTime}, evalTime}, + {"eval-time absent", of.FlagMetadata{metadataAllocationKey: "alloc-1"}, 0}, + {"nil metadata", nil, 0}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + hookCtx := makeHookContext("flag-x", "user-1", nil) + details := makeEvalDetails("on", of.TargetingMatchReason, "", tc.md) + if got := extractEvalDetails(hookCtx, details).evalTimeMs; got != tc.want { + t.Errorf("evalTimeMs = %d, want %d", got, tc.want) + } + }) + } +} + +// TestRecordUsesEvalTimeFromMetadata verifies record() stamps the aggregated entry's +// first/last evaluation from provider-supplied eval-time, and falls back to hook time when absent. +func TestRecordUsesEvalTimeFromMetadata(t *testing.T) { + t.Run("uses provider eval-time", func(t *testing.T) { + const evalTime int64 = 1_700_000_000_000 + w := setupTestWriter(t) + w.record(makeHookContext("flag-y", "user-2", nil), + makeEvalDetails("on", of.TargetingMatchReason, "", of.FlagMetadata{metadataEvalTimeKey: evalTime})) + w.aggregate(<-w.events) + + if len(w.aggregator.full) != 1 { + t.Fatalf("expected 1 full-tier entry, got %d", len(w.aggregator.full)) + } + for _, e := range w.aggregator.full { + if e.firstEvaluation != evalTime || e.lastEvaluation != evalTime { + t.Errorf("entry first/last = %d/%d, want %d", e.firstEvaluation, e.lastEvaluation, evalTime) + } + } + }) + + t.Run("falls back to hook time when absent", func(t *testing.T) { + before := time.Now().UnixMilli() + w := setupTestWriter(t) + w.record(makeHookContext("flag-z", "user-3", nil), + makeEvalDetails("on", of.TargetingMatchReason, "")) + w.aggregate(<-w.events) + after := time.Now().UnixMilli() + + for _, e := range w.aggregator.full { + if e.firstEvaluation < before || e.firstEvaluation > after { + t.Errorf("fallback first=%d not within hook window [%d,%d]", e.firstEvaluation, before, after) + } + } + }) +} + +// TestFlagEvaluationHookFinally verifies that the Finally hook records an entry for +// success, error-reason, and provider-not-ready paths. +func TestFlagEvaluationHookFinally(t *testing.T) { + runtimeDefaultTrue := true + + tests := []struct { + name string + flagKey string + targetingKey string + attrs map[string]any + variant string + reason of.Reason + errorCode of.ErrorCode + metadata []of.FlagMetadata + wantRuntimeDefault *bool // when set, assert the recorded full-tier entry's runtimeDefault + }{ + { + name: "success path records entry", + flagKey: "test-flag", + targetingKey: "user-123", + attrs: map[string]any{"country": "US"}, + variant: "on", + reason: of.TargetingMatchReason, + metadata: []of.FlagMetadata{{metadataAllocationKey: "default-alloc"}}, + }, + { + name: "error-reason path records entry (flag-not-found)", + flagKey: "missing-flag", + targetingKey: "user-123", + reason: of.ErrorReason, + errorCode: of.FlagNotFoundCode, + }, + { + name: "provider-not-ready path records entry", + flagKey: "any-flag", + reason: of.ErrorReason, + errorCode: of.ProviderNotReadyCode, + }, + { + name: "DEFAULT reason path records entry with runtime_default_used=true", + flagKey: "absent-flag", + targetingKey: "user-456", + reason: of.DefaultReason, + wantRuntimeDefault: &runtimeDefaultTrue, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + w := setupTestWriter(t) + hook := newFlagEvaluationHook(w) + + hookCtx := makeHookContext(tc.flagKey, tc.targetingKey, tc.attrs) + details := makeEvalDetails(tc.variant, tc.reason, tc.errorCode, tc.metadata...) + + hook.Finally(context.Background(), hookCtx, details, of.HookHints{}) + + // record() enqueues asynchronously; drain the one event into the aggregator so the + // assertions below observe it deterministically (no worker runs in this test). + w.aggregate(<-w.events) + + w.aggregator.mu.Lock() + defer w.aggregator.mu.Unlock() + + total := len(w.aggregator.full) + len(w.aggregator.degraded) + if total == 0 { + t.Error("expected Finally to record an entry, got none") + } + + if tc.wantRuntimeDefault != nil { + for _, e := range w.aggregator.full { + if e.runtimeDefault != *tc.wantRuntimeDefault { + t.Errorf("expected runtimeDefault=%v, got %v", *tc.wantRuntimeDefault, e.runtimeDefault) + } + } + } + }) + } +} + +// TestFlagEvaluationHookContextCancelled verifies that a cancelled context does NOT +// drop the evaluation: record() is a non-blocking enqueue that ignores the request +// context, so a cancelled request must still be counted. +func TestFlagEvaluationHookContextCancelled(t *testing.T) { + w := setupTestWriter(t) + hook := newFlagEvaluationHook(w) + + ctx, cancel := context.WithCancel(context.Background()) + cancel() // cancel BEFORE calling Finally + + hookCtx := makeHookContext("test-flag", "user-123", nil) + details := makeEvalDetails("on", of.TargetingMatchReason, "") + + hook.Finally(ctx, hookCtx, details, of.HookHints{}) + + // record() enqueues asynchronously; drain the one event so the assertion sees it. + w.aggregate(<-w.events) + + w.aggregator.mu.Lock() + defer w.aggregator.mu.Unlock() + + total := len(w.aggregator.full) + len(w.aggregator.degraded) + if total != 1 { + t.Errorf("expected the cancelled-context evaluation to still be counted, got %d entries", total) + } +} + +// TestFlagEvaluationBackpressureDrops verifies the explicit backpressure policy: when the +// async hand-off queue is full, record() drops the event and counts it (observable) rather +// than blocking the evaluation. No worker drains in this test, so the queue fills after +// defaultEvalEventBufferSize enqueues and every further record() is a counted drop. +func TestFlagEvaluationBackpressureDrops(t *testing.T) { + w := setupTestWriter(t) + hookCtx := makeHookContext("bp-flag", "user-1", nil) + details := makeEvalDetails("on", of.TargetingMatchReason, "") + + const overflow = 100 + for range defaultEvalEventBufferSize + overflow { + w.record(hookCtx, details) // must never block, even once the queue is full + } + + if got := w.dropped.Load(); got != overflow { + t.Errorf("expected exactly %d dropped evaluations once the queue filled, got %d", overflow, got) + } +} + +func TestFlagEvaluationBackpressureDropsBeforeContextSnapshot(t *testing.T) { + w := setupTestWriter(t) + for range cap(w.events) { + w.events <- evalEvent{} + } + + hookCtx := makeHookContext("bp-flag", "user-1", map[string]any{"expensive": panicStringer{}}) + details := makeEvalDetails("on", of.TargetingMatchReason, "") + + w.record(hookCtx, details) + + if got := w.dropped.Load(); got != 1 { + t.Fatalf("expected one dropped evaluation when queue is already full, got %d", got) + } + if got := len(w.events); got != cap(w.events) { + t.Fatalf("full queue length changed: got %d, want %d", got, cap(w.events)) + } +} diff --git a/openfeature/flagevaluation_provider_test.go b/openfeature/flagevaluation_provider_test.go new file mode 100644 index 00000000000..d4b2f594238 --- /dev/null +++ b/openfeature/flagevaluation_provider_test.go @@ -0,0 +1,101 @@ +// Unless explicitly stated otherwise all files in this repository are licensed +// under the Apache License Version 2.0. +// This product includes software developed at Datadog (https://www.datadoghq.com/). +// Copyright 2025 Datadog, Inc. + +package openfeature + +import ( + "testing" + + of "github.com/open-feature/go-sdk/openfeature" +) + +// TestFlagEvaluationKillswitch verifies that DD_FLAGGING_EVALUATION_COUNTS_ENABLED (default true) +// controls ONLY the EVP flagevaluation hook/writer, leaving the OTel flagEvalHook unaffected. +// +// When the killswitch is "false": the EVP hook (flagEvalEVPHook) is NOT registered in Hooks() +// and flagEvalWriter is nil. +// When the killswitch is unset or "true": the EVP hook IS registered. +// The OTel flagEvalHook is present in Hooks() in BOTH cases. +func TestFlagEvaluationKillswitch(t *testing.T) { + tests := []struct { + name string + envValue string + wantEVPEnabled bool + }{ + { + name: "killswitch disabled: EVP hook absent from Hooks(), OTel hook present", + envValue: "false", + wantEVPEnabled: false, + }, + { + // "1" exercises the default-true behavior (any truthy value enables the EVP path). + name: "killswitch enabled (unset = default true): EVP hook present in Hooks()", + envValue: "1", + wantEVPEnabled: true, + }, + { + name: "killswitch explicitly true: EVP hook present in Hooks()", + envValue: "true", + wantEVPEnabled: true, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Setenv(flagEvalCountsEnabledEnvVar, tc.envValue) + + p := newDatadogProvider(ProviderConfig{}) + + if tc.wantEVPEnabled { + if p.flagEvalWriter == nil { + t.Error("expected flagEvalWriter to be non-nil when killswitch is enabled") + } + if p.flagEvalEVPHook == nil { + t.Error("expected flagEvalEVPHook to be non-nil when killswitch is enabled") + } + } else { + if p.flagEvalWriter != nil { + t.Error("expected flagEvalWriter to be nil when killswitch is disabled") + } + if p.flagEvalEVPHook != nil { + t.Error("expected flagEvalEVPHook to be nil when killswitch is disabled") + } + } + + hooks := p.Hooks() + + otelPresent := false + evpPresent := false + for _, h := range hooks { + switch h.(type) { + case *flagEvalHook: + otelPresent = true + case *flagEvaluationHook: + evpPresent = true + } + } + + // The OTel hook must be present in EVERY case — the killswitch never affects it. + if !otelPresent { + t.Error("expected OTel flagEvalHook to be present in Hooks() regardless of the killswitch") + } + + if evpPresent != tc.wantEVPEnabled { + if tc.wantEVPEnabled { + t.Error("expected EVP flagEvaluationHook to be present in Hooks() when killswitch is enabled") + } else { + t.Errorf("expected EVP flagEvaluationHook to be absent from Hooks() when killswitch is disabled, but found one") + } + } + + if tc.wantEVPEnabled && p.exposureWriter.evp != p.flagEvalWriter.evp { + t.Error("expected exposures and flag evaluations to share one EVP client") + } + }) + } +} + +// Compile-time assertion: flagEvaluationHook implements the OpenFeature Hook interface. +var _ of.Hook = (*flagEvaluationHook)(nil) diff --git a/openfeature/flagevaluation_scale_test.go b/openfeature/flagevaluation_scale_test.go new file mode 100644 index 00000000000..0333b34ac6b --- /dev/null +++ b/openfeature/flagevaluation_scale_test.go @@ -0,0 +1,341 @@ +// Unless explicitly stated otherwise all files in this repository are licensed +// under the Apache License Version 2.0. +// This product includes software developed at Datadog (https://www.datadoghq.com/). +// Copyright 2025 Datadog, Inc. + +package openfeature + +import ( + "fmt" + "testing" + "time" +) + +// These tests drive flagEvaluationAggregator.add directly (no client, no hooks, no async worker) +// to assert the 2,500-flag scale target against production aggregation caps. + +// scaleFlagShape describes the realistic per-flag structure used to size the degraded tier. +// Degraded key includes schema-visible retained fields only. OpenFeature reason is not accepted +// by the worker schema and is not part of degraded cardinality. +type scaleFlagShape struct { + variants int // distinct variant keys this flag can return + allocations int // distinct allocation keys this flag can return +} + +// makeScaleFlags builds n flags with a realistic spread of variants/allocations. The spread is +// deterministic (driven by index modulo) so the degraded cardinality is exactly reproducible: +// - ~1/3 of flags: 2 variants, 1 allocation (simple on/off) +// - ~1/3 of flags: 3 variants, 1 allocation (multivariate) +// - ~1/3 of flags: 4 variants, 2 allocations (multivariate + multiple allocations) +func makeScaleFlags(n int) []scaleFlagShape { + flags := make([]scaleFlagShape, n) + for i := range n { + switch i % 3 { + case 0: + flags[i] = scaleFlagShape{variants: 2, allocations: 1} + case 1: + flags[i] = scaleFlagShape{variants: 3, allocations: 1} + default: + flags[i] = scaleFlagShape{variants: 4, allocations: 2} + } + } + return flags +} + +// legitimateDegradedCardinality returns the number of DISTINCT degraded buckets the given flag +// shapes would produce if every (variant × allocation) combination were observed. +// This is the count degradedCap must hold without dropping legitimate buckets. +func legitimateDegradedCardinality(flags []scaleFlagShape) int { + total := 0 + for _, f := range flags { + total += f.variants * f.allocations + } + return total +} + +// driveScale records evaluations into agg for the given flag shapes, distributing each flag's +// evaluations across its variant/allocation combinations and across numContexts distinct +// evaluation contexts (subjects). evalsPerCombo controls how many subjects hit each combination +// (so counts accumulate). It returns the total number of add() calls made. +// +// The context cardinality knob (numContexts) is what splits the FULL tier: the full key includes +// targetingKey + contextKey, so more distinct subjects => more full buckets => earlier full-tier +// saturation => earlier cascade into degraded. +func driveScale(agg *flagEvaluationAggregator, flags []scaleFlagShape, numContexts, evalsPerCombo int) int64 { + nowMs := time.Now().UnixMilli() + var calls int64 + ctxCounter := 0 + for fi, f := range flags { + flagKey := fmt.Sprintf("flag-%05d", fi) + for v := range f.variants { + variant := fmt.Sprintf("v%d", v) + for a := range f.allocations { + alloc := fmt.Sprintf("alloc-%d", a) + for range evalsPerCombo { + // Spread subjects across numContexts distinct targeting keys + contexts. + subj := ctxCounter % numContexts + ctxCounter++ + d := evalDetails{ + flagKey: flagKey, + variant: variant, + allocationKey: alloc, + targetingKey: fmt.Sprintf("user-%d", subj), + } + attrs := map[string]any{ + "country": fmt.Sprintf("c%d", subj%50), + "plan": fmt.Sprintf("p%d", subj%5), + } + agg.add(d, attrs, nowMs) + calls++ + } + } + } + } + return calls +} + +// tierCounts returns observable aggregator state after a run. Over-cap degraded counts land in +// droppedDegradedOverflow (observable, not silent). +// sumCounts includes the drop counter so the count-preservation invariant is Σ == add() calls. +type tierCounts struct { + full, degraded int + dropped int64 + globalCount int + sumCounts int64 +} + +func snapshot(agg *flagEvaluationAggregator) tierCounts { + agg.mu.Lock() + defer agg.mu.Unlock() + tc := tierCounts{ + full: len(agg.full), + degraded: len(agg.degraded), + dropped: agg.droppedDegradedOverflow, + globalCount: agg.globalCount, + } + for _, e := range agg.full { + tc.sumCounts += e.count + } + for _, e := range agg.degraded { + tc.sumCounts += e.count + } + tc.sumCounts += agg.droppedDegradedOverflow + return tc +} + +// TestScaleDegradedCardinality2500Flags verifies the production degradedCap has at least 2x +// headroom over the legitimate degraded cardinality of the 2,500-flag target shape. +func TestScaleDegradedCardinality2500Flags(t *testing.T) { + const n = 2500 + flags := makeScaleFlags(n) + deg := legitimateDegradedCardinality(flags) + + // Per-shape breakdown for the report. + var s0, s1, s2 int + for i := range n { + switch i % 3 { + case 0: + s0++ + case 1: + s1++ + default: + s2++ + } + } + t.Logf("2,500-flag realistic shape:") + t.Logf(" %d flags @ 2v×1a = %d degraded buckets", s0, s0*2*1) + t.Logf(" %d flags @ 3v×1a = %d degraded buckets", s1, s1*3*1) + t.Logf(" %d flags @ 4v×2a = %d degraded buckets", s2, s2*4*2) + t.Logf("LEGITIMATE degraded cardinality (Σ variants×allocations) = %d", deg) + t.Logf("production degradedCap = %d", defaultEvalDegradedCap) + t.Logf("production globalCap = %d", defaultEvalGlobalCap) + + recDegraded := roundUpTo(deg*2, 1000) + if defaultEvalDegradedCap < recDegraded { + t.Fatalf("degradedCap=%d does not provide 2x headroom over legitimate cardinality=%d; want >= %d", + defaultEvalDegradedCap, deg, recDegraded) + } + t.Logf("degradedCap headroom OK: cap=%d legitimate=%d requiredWith2xHeadroom=%d", + defaultEvalDegradedCap, deg, recDegraded) +} + +func roundUpTo(v, mult int) int { + if v%mult == 0 { + return v + } + return ((v / mult) + 1) * mult +} + +// TestScaleDropTriggerSweep verifies that the 2,500-flag target shape does not drop legitimate +// counts across representative context-cardinality points with production caps. +func TestScaleDropTriggerSweep(t *testing.T) { + const n = 2500 + flags := makeScaleFlags(n) + deg := legitimateDegradedCardinality(flags) + t.Logf("2,500 flags; legitimate degraded cardinality = %d; production caps "+ + "full=%d perFlag=%d degraded=%d (full -> degraded -> drop)", + deg, defaultEvalGlobalCap, defaultEvalPerFlagCap, defaultEvalDegradedCap) + + // Sweep distinct-context cardinality. Low = few subjects (full tier stays small); + // high = many subjects (full tier saturates, cascading to degraded then drop). + sweep := []struct { + name string + numContexts int + evalsPer int + }{ + {"few-contexts (10 subjects)", 10, 2}, + {"moderate-contexts (1k subjects)", 1000, 1}, + {"many-contexts (100k subjects)", 100_000, 1}, + {"extreme-contexts (1M subjects)", 1_000_000, 1}, + } + + for _, sp := range sweep { + t.Run(sp.name, func(t *testing.T) { + agg := newTestAggregator( + defaultEvalGlobalCap, + defaultEvalPerFlagCap, + defaultEvalDegradedCap, + ) + calls := driveScale(agg, flags, sp.numContexts, sp.evalsPer) + tc := snapshot(agg) + + t.Logf("contexts=%d evalsPerCombo=%d => add() calls=%d", sp.numContexts, sp.evalsPer, calls) + t.Logf(" full=%d (globalCount=%d, cap=%d) degraded=%d (cap=%d) droppedDegradedOverflow=%d", + tc.full, tc.globalCount, defaultEvalGlobalCap, + tc.degraded, defaultEvalDegradedCap, tc.dropped) + t.Logf(" Σ counts (full+degraded+dropped)=%d (must == add() calls=%d => %v)", + tc.sumCounts, calls, tc.sumCounts == calls) + + // Count preservation must hold: nothing silently lost. + if tc.sumCounts != calls { + t.Errorf("count preservation violated: Σ=%d != calls=%d", tc.sumCounts, calls) + } + + if tc.dropped != 0 { + t.Errorf("unexpected degraded overflow drops at %s: got %d", sp.name, tc.dropped) + } + }) + } +} + +// TestScaleDropRequiresDegradedSaturation isolates the precondition for a terminal-tier drop: +// the degraded tier must be full. With production degradedCap and realistic 2,500-flag structure, +// it forces all buckets through the degraded tier and asserts that no legitimate counts drop. +func TestScaleDropRequiresDegradedSaturation(t *testing.T) { + const n = 2500 + flags := makeScaleFlags(n) + deg := legitimateDegradedCardinality(flags) + + // Force the full tier to be useless (globalCap=0) so EVERY new bucket cascades immediately + // to the degraded path — the worst case for the degraded tier at 2,500 flags. + agg := newTestAggregator(0, defaultEvalPerFlagCap, defaultEvalDegradedCap) + calls := driveScale(agg, flags, 100_000, 1) + tc := snapshot(agg) + + t.Logf("WORST CASE for degraded tier (globalCap=0 => everything cascades to degraded):") + t.Logf(" legitimate degraded cardinality = %d, degradedCap = %d", deg, defaultEvalDegradedCap) + t.Logf(" result: full=%d degraded=%d droppedDegradedOverflow=%d Σ=%d/calls=%d", + tc.full, tc.degraded, tc.dropped, tc.sumCounts, calls) + + if deg > defaultEvalDegradedCap { + t.Errorf("legitimate degraded cardinality %d EXCEEDS degradedCap %d — aggregation would DROP "+ + "legitimate counts at 2,500 flags; raise degradedCap", deg, defaultEvalDegradedCap) + } else { + t.Logf(" => all %d legitimate degraded buckets FIT under degradedCap %d (headroom %d); "+ + "no legitimate count dropped at the scale target.", + deg, defaultEvalDegradedCap, defaultEvalDegradedCap-deg) + if tc.dropped != 0 { + t.Errorf("unexpected drops (%d) when legitimate cardinality fits under degradedCap", tc.dropped) + } + } + + if tc.sumCounts != calls { + t.Errorf("count preservation violated: Σ=%d != calls=%d", tc.sumCounts, calls) + } +} + +// TestScaleFullSaturationCascade saturates the full tier naturally with production caps and +// verifies that overflow cascades to degraded without dropping legitimate counts. +func TestScaleFullSaturationCascade(t *testing.T) { + const n = 2500 + flags := makeScaleFlags(n) + deg := legitimateDegradedCardinality(flags) + + agg := newTestAggregator( + defaultEvalGlobalCap, + defaultEvalPerFlagCap, + defaultEvalDegradedCap, + ) + // With 16 distinct subjects per combo the full tier sees ~173k distinct full keys, over + // globalCap, forcing overflow into degraded. + calls := driveScale(agg, flags, 1_000_000, 16) + tc := snapshot(agg) + + t.Logf("FULL-saturation cascade (production caps; 16 subjects/combo):") + t.Logf(" legitimate degraded cardinality=%d add() calls=%d", deg, calls) + t.Logf(" full=%d (globalCount=%d, cap=%d) degraded=%d (cap=%d) droppedDegradedOverflow=%d", + tc.full, tc.globalCount, defaultEvalGlobalCap, + tc.degraded, defaultEvalDegradedCap, tc.dropped) + t.Logf(" Σ counts=%d / calls=%d (preserved=%v)", tc.sumCounts, calls, tc.sumCounts == calls) + + if tc.full > defaultEvalGlobalCap { + t.Errorf("full tier %d exceeded globalCap %d", tc.full, defaultEvalGlobalCap) + } + if tc.globalCount != defaultEvalGlobalCap { + t.Errorf("full tier did not saturate: globalCount=%d, want %d", tc.globalCount, defaultEvalGlobalCap) + } + if tc.degraded == 0 { + t.Error("expected degraded tier to absorb full-tier overflow") + } + if tc.dropped != 0 { + t.Errorf("unexpected degraded overflow drops: got %d", tc.dropped) + } + if tc.sumCounts != calls { + t.Errorf("count preservation violated: Σ=%d != calls=%d", tc.sumCounts, calls) + } +} + +// TestScaleHotFlagPerFlagCap drives a single flag past perFlagCap and asserts that overflow +// reaches the degraded tier, then becomes counted drops once degradedCap is saturated. +func TestScaleHotFlagPerFlagCap(t *testing.T) { + agg := newTestAggregator( + defaultEvalGlobalCap, + defaultEvalPerFlagCap, + defaultEvalDegradedCap, + ) + nowMs := time.Now().UnixMilli() + + // One hot flag, many distinct (variant, subject) combos so it blows past perFlagCap and then + // keeps generating distinct degraded keys. To fill the resized degradedCap we need that many + // distinct schema-visible variant/allocation combinations for this one flag. + const distinctVariants = 50_000 + var calls int64 + for v := range distinctVariants { + d := evalDetails{ + flagKey: "hot-flag", + variant: fmt.Sprintf("v%d", v), + allocationKey: "alloc-0", + targetingKey: fmt.Sprintf("user-%d", v), + } + agg.add(d, map[string]any{"k": v}, nowMs) + calls++ + } + tc := snapshot(agg) + + t.Logf("HOT-flag perFlagCap path (single flag, %d distinct variants):", distinctVariants) + t.Logf(" full=%d (perFlagCap=%d) degraded=%d (cap=%d) droppedDegradedOverflow=%d", + tc.full, defaultEvalPerFlagCap, tc.degraded, defaultEvalDegradedCap, tc.dropped) + t.Logf(" Σ counts=%d / calls=%d (preserved=%v)", tc.sumCounts, calls, tc.sumCounts == calls) + if tc.full != defaultEvalPerFlagCap { + t.Errorf("full tier did not stop at perFlagCap: full=%d, want %d", tc.full, defaultEvalPerFlagCap) + } + if tc.degraded != defaultEvalDegradedCap { + t.Errorf("degraded tier did not stop at degradedCap: degraded=%d, want %d", tc.degraded, defaultEvalDegradedCap) + } + if tc.dropped == 0 { + t.Error("expected counted drops after single hot flag saturates degradedCap") + } + if tc.sumCounts != calls { + t.Errorf("count preservation violated: Σ=%d != calls=%d", tc.sumCounts, calls) + } +} diff --git a/openfeature/flagevaluation_test.go b/openfeature/flagevaluation_test.go new file mode 100644 index 00000000000..ac64f2319da --- /dev/null +++ b/openfeature/flagevaluation_test.go @@ -0,0 +1,1284 @@ +// Unless explicitly stated otherwise all files in this repository are licensed +// under the Apache License Version 2.0. +// This product includes software developed at Datadog (https://www.datadoghq.com/). +// Copyright 2025 Datadog, Inc. + +package openfeature + +import ( + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "net/url" + "reflect" + "strings" + "sync" + "testing" + "time" + + of "github.com/open-feature/go-sdk/openfeature" +) + +func mustMarshalJSONMap(t *testing.T, payload any) map[string]any { + t.Helper() + + b, err := json.Marshal(payload) + if err != nil { + t.Fatalf("failed to marshal payload: %v", err) + } + var m map[string]any + if err := json.Unmarshal(b, &m); err != nil { + t.Fatalf("failed to unmarshal payload for schema validation: %v", err) + } + return m +} + +func TestFlagEvaluationEndpointUsesTrackName(t *testing.T) { + const want = "/evp_proxy/v2/api/v2/flagevaluation" + if flagEvaluationEndpoint != want { + t.Fatalf("flagEvaluationEndpoint = %q, want %q", flagEvaluationEndpoint, want) + } +} + +func TestDefaultEvalCapSizing(t *testing.T) { + if defaultEvalGlobalCap <= evalScaleFullBucketTarget { + t.Fatalf("defaultEvalGlobalCap = %d, want > %d", defaultEvalGlobalCap, evalScaleFullBucketTarget) + } + + if defaultEvalPerFlagCap != evalScalePerFlagBucketTarget { + t.Fatalf("defaultEvalPerFlagCap = %d, want %d", defaultEvalPerFlagCap, evalScalePerFlagBucketTarget) + } + + if defaultEvalDegradedCap <= evalScaleDegradedBucketTarget { + t.Fatalf("defaultEvalDegradedCap = %d, want > %d", defaultEvalDegradedCap, evalScaleDegradedBucketTarget) + } +} + +// TestFlattenAndPruneContextEquivalence verifies the merged single-pass +// flattenAndPruneContext must produce a pruned result byte-for-byte identical to the prior +// two-step flattenContext + pruneContext pipeline across nested, oversized, and >256-field +// inputs (and the determinism + 256/256 limits are preserved). +func TestFlattenAndPruneContextEquivalence(t *testing.T) { + bigFields := func() map[string]any { + m := make(map[string]any, 400) + for i := range 400 { + m[fmt.Sprintf("key%04d", i)] = fmt.Sprintf("value%04d", i) + } + return m + } + + cases := []struct { + name string + input map[string]any + }{ + { + name: "nested objects flatten to dot notation", + input: map[string]any{"user": map[string]any{"id": "123", "email": "a@b.com"}, "country": "US"}, + }, + { + name: "deeply nested + arrays", + input: map[string]any{"a": map[string]any{"b": map[string]any{"c": 1}}, "tags": []string{"x", "y", "z"}}, + }, + { + name: "oversized string value is skipped", + input: map[string]any{"short": "ok", "long": strings.Repeat("x", maxFieldLength+10)}, + }, + { + name: "more than 256 fields truncated to 256", + input: bigFields(), + }, + { + name: "nested oversized string among many fields", + input: map[string]any{"u": map[string]any{"bio": strings.Repeat("y", maxFieldLength+1), "id": 7}}, + }, + { + name: "mixed scalar types retained", + input: map[string]any{"i": 42, "b": true, "f": 3.14, "s": "hi"}, + }, + { + name: "empty input", + input: map[string]any{}, + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + // Reference pipeline (the two former steps) vs the merged single-pass procedure. + want := pruneContext(flattenContext(tc.input)) + got := flattenAndPruneContext(tc.input) + + if !reflect.DeepEqual(got, want) { + t.Errorf("merged flatten+prune differs from flattenContext+pruneContext:\n got=%v\nwant=%v", got, want) + } + + // 256-field limit preserved. + if len(got) > maxContextFields { + t.Errorf("merged result has %d fields, exceeds maxContextFields %d", len(got), maxContextFields) + } + + // Determinism: repeated calls yield an identical canonical key. + first := canonicalContextKey(flattenAndPruneContext(tc.input)) + for range 25 { + if k := canonicalContextKey(flattenAndPruneContext(tc.input)); k != first { + t.Fatalf("merged flatten+prune nondeterministic: canonical keys differ across calls") + } + } + }) + } +} + +// setupTestAggregator creates a flagEvaluationAggregator with small caps for testing. +// Caps are deliberately small to trigger tier-cascade behavior in unit tests. +func setupTestAggregator(t *testing.T) *flagEvaluationAggregator { + t.Helper() + return &flagEvaluationAggregator{ + full: make(map[evaluationAggregationKey]*evaluationEntry), + degraded: make(map[evaluationDegradedKey]*evaluationEntry), + perFlagFull: make(map[string]int), + globalCap: 10, // small cap to trigger overflow in tests + perFlagCap: 3, + degradedCap: 3, + } +} + +// newTestAggregator builds a flagEvaluationAggregator with explicit, caller-supplied caps. +// Unlike setupTestAggregator (which fixes small caps), each cap is a parameter so a test can +// drive a specific tier-overflow scenario. The cap NUMBERS are load-bearing — callers pass +// the exact values their scenario requires. +func newTestAggregator(globalCap, perFlagCap, degradedCap int) *flagEvaluationAggregator { + return &flagEvaluationAggregator{ + full: make(map[evaluationAggregationKey]*evaluationEntry), + degraded: make(map[evaluationDegradedKey]*evaluationEntry), + perFlagFull: make(map[string]int), + globalCap: globalCap, + perFlagCap: perFlagCap, + degradedCap: degradedCap, + } +} + +// TestPruneContext verifies that pruneContext applies the 256-field / 256-char limits +// before evaluation context enters the aggregation buffer. +func TestPruneContext(t *testing.T) { + tests := []struct { + name string + input map[string]any + assert func(t *testing.T, out map[string]any) + }{ + { + name: "300 fields truncated to exactly 256", + input: func() map[string]any { + raw := make(map[string]any, 300) + for i := range 300 { + raw[fmt.Sprintf("key%d", i)] = fmt.Sprintf("value%d", i) + } + return raw + }(), + assert: func(t *testing.T, out map[string]any) { + if len(out) != 256 { + t.Errorf("expected exactly 256 fields after prune, got %d", len(out)) + } + }, + }, + { + name: "string value exceeding 256 chars is dropped", + input: map[string]any{ + "short": "ok", + "long": strings.Repeat("x", 300), // 300 chars > maxFieldLength(256) + }, + assert: func(t *testing.T, out map[string]any) { + if _, ok := out["long"]; ok { + t.Error("expected long string value to be dropped from pruned context") + } + if _, ok := out["short"]; !ok { + t.Error("expected short string value to be retained in pruned context") + } + }, + }, + { + name: "nil input returns nil", + input: nil, + assert: func(t *testing.T, out map[string]any) { + if out != nil { + t.Errorf("expected nil for nil input, got %v", out) + } + }, + }, + { + name: "empty input returns nil or empty", + input: map[string]any{}, + assert: func(t *testing.T, out map[string]any) { + if out != nil && len(out) != 0 { + t.Errorf("expected nil or empty for empty input, got %v", out) + } + }, + }, + { + name: "non-string values are retained regardless of notional length", + input: map[string]any{ + "intVal": 42, + "boolVal": true, + }, + assert: func(t *testing.T, out map[string]any) { + if out["intVal"] == nil { + t.Error("expected integer value to be retained") + } + if out["boolVal"] == nil { + t.Error("expected boolean value to be retained") + } + }, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + tc.assert(t, pruneContext(tc.input)) + }) + } +} + +// TestFlagEvaluationPayloadSchema verifies that full, degraded, and required-only events +// marshal to JSON that omits the expected optional fields per tier while always including +// the 5 required fields. +func TestFlagEvaluationPayloadSchema(t *testing.T) { + nowMs := time.Now().UnixMilli() + + requiredFields := []string{"timestamp", "flag", "first_evaluation", "last_evaluation", "evaluation_count"} + + tierTests := []struct { + name string + event flagEvaluationEvent + requiredFlgKey bool // full tier additionally asserts flag.key is present + optionalAbsent []string // optional fields that must NOT appear for this tier + }{ + { + name: "full tier has all required fields", + event: flagEvaluationEvent{ + Timestamp: nowMs, + Flag: flagEvalFlag{Key: "test-flag"}, + FirstEvaluation: nowMs, + LastEvaluation: nowMs, + EvaluationCount: 1, + Variant: &flagEvalVariant{Key: "on"}, + TargetingKey: "user-123", + Context: &flagEvalEventContext{ + Evaluation: map[string]any{"country": "US"}, + }, + }, + requiredFlgKey: true, + }, + { + name: "degraded tier omits targeting_key and context.evaluation", + // Degraded tier: no targeting_key, no context.evaluation; variant + allocation present. + event: flagEvaluationEvent{ + Timestamp: nowMs, + Flag: flagEvalFlag{Key: "test-flag"}, + FirstEvaluation: nowMs, + LastEvaluation: nowMs, + EvaluationCount: 5, + Variant: &flagEvalVariant{Key: "on"}, + Allocation: &flagEvalAllocation{Key: "default"}, + // TargetingKey / Context intentionally absent. + }, + optionalAbsent: []string{"targeting_key", "context"}, + }, + { + name: "required-only event omits all optional fields", + // A bare event carrying only flag key + counts; no variant, allocation, targeting, + // context. (This shape is not emitted by a dedicated tier, but the + // schema must still accept a required-fields-only event.) + event: flagEvaluationEvent{ + Timestamp: nowMs, + Flag: flagEvalFlag{Key: "test-flag"}, + FirstEvaluation: nowMs, + LastEvaluation: nowMs, + EvaluationCount: 1000, + // All optional fields intentionally absent. + }, + optionalAbsent: []string{"targeting_key", "variant", "allocation", "targeting_rule", "error", "context", "runtime_default_used"}, + }, + } + + for _, tc := range tierTests { + t.Run(tc.name, func(t *testing.T) { + b, err := json.Marshal(tc.event) + if err != nil { + t.Fatalf("failed to marshal event: %v", err) + } + var m map[string]any + if err := json.Unmarshal(b, &m); err != nil { + t.Fatalf("failed to unmarshal: %v", err) + } + + for _, req := range requiredFields { + if _, ok := m[req]; !ok { + t.Errorf("required field %q missing from marshaled JSON", req) + } + } + if tc.requiredFlgKey { + if flagObj, ok := m["flag"].(map[string]any); !ok { + t.Error("flag is not an object") + } else if _, ok := flagObj["key"]; !ok { + t.Error("flag.key missing") + } + } + for _, opt := range tc.optionalAbsent { + if _, ok := m[opt]; ok { + t.Errorf("optional field %q should be absent", opt) + } + } + }) + } + + t.Run("first_evaluation and last_evaluation meet minimum constraint", func(t *testing.T) { + // Schema minimum: 1759276800000 (2025-08-01 Unix ms) + // Using time.Now().UnixMilli() always satisfies this. + const schemaMin int64 = 1759276800000 + if nowMs < schemaMin { + t.Errorf("time.Now().UnixMilli() = %d is below schema minimum %d; use current timestamps only", nowMs, schemaMin) + } + }) + + t.Run("batch payload wraps events in flagEvaluations array", func(t *testing.T) { + payload := flagEvaluationPayload{ + Context: flagEvalDDContext{ + Service: "test-service", + Env: "test", + Version: "1.0.0", + }, + FlagEvaluations: []flagEvaluationEvent{ + { + Timestamp: nowMs, + Flag: flagEvalFlag{Key: "test-flag"}, + FirstEvaluation: nowMs, + LastEvaluation: nowMs, + EvaluationCount: 1, + }, + }, + } + + b, err := json.Marshal(payload) + if err != nil { + t.Fatalf("failed to marshal batch payload: %v", err) + } + + var m map[string]any + if err := json.Unmarshal(b, &m); err != nil { + t.Fatalf("failed to unmarshal: %v", err) + } + + if _, ok := m["context"]; !ok { + t.Error("batch payload: context missing") + } + if _, ok := m["flagEvaluations"]; !ok { + t.Error("batch payload: flagEvaluations array missing") + } + if arr, ok := m["flagEvaluations"].([]any); !ok || len(arr) != 1 { + t.Errorf("batch payload: expected 1 flagEvaluations entry, got %v", m["flagEvaluations"]) + } + }) + + t.Run("batch payload uses only stable EVP flagevaluation fields", func(t *testing.T) { + payload := flagEvaluationPayload{ + Context: flagEvalDDContext{ + Service: "test-service", + Env: "test", + Version: "1.0.0", + }, + FlagEvaluations: []flagEvaluationEvent{ + { + Timestamp: nowMs, + Flag: flagEvalFlag{Key: "full-flag"}, + FirstEvaluation: nowMs, + LastEvaluation: nowMs, + EvaluationCount: 2, + RuntimeDefault: true, + TargetingKey: "user-123", + Variant: &flagEvalVariant{Key: "on"}, + Allocation: &flagEvalAllocation{Key: "alloc-a"}, + Error: &flagEvalError{Message: string(of.TypeMismatchCode)}, + Context: &flagEvalEventContext{ + Evaluation: map[string]any{"country": "US", "plan": "pro"}, + }, + }, + { + Timestamp: nowMs, + Flag: flagEvalFlag{Key: "degraded-flag"}, + FirstEvaluation: nowMs, + LastEvaluation: nowMs, + EvaluationCount: 5, + Variant: &flagEvalVariant{Key: "off"}, + Allocation: &flagEvalAllocation{Key: "alloc-b"}, + Error: &flagEvalError{Message: string(of.FlagNotFoundCode)}, + }, + { + Timestamp: nowMs, + Flag: flagEvalFlag{Key: "required-only-flag"}, + FirstEvaluation: nowMs, + LastEvaluation: nowMs, + EvaluationCount: 1, + }, + }, + } + + m := mustMarshalJSONMap(t, payload) + for k := range m { + switch k { + case "context", "flagEvaluations": + default: + t.Fatalf("unexpected batch payload field %q", k) + } + } + + arr, ok := m["flagEvaluations"].([]any) + if !ok { + t.Fatalf("flagEvaluations is not an array: %T", m["flagEvaluations"]) + } + if len(arr) != 3 { + t.Fatalf("expected 3 flagEvaluations entries, got %d", len(arr)) + } + + allowedEventFields := map[string]struct{}{ + "timestamp": {}, + "flag": {}, + "first_evaluation": {}, + "last_evaluation": {}, + "evaluation_count": {}, + "runtime_default_used": {}, + "targeting_key": {}, + "variant": {}, + "allocation": {}, + "targeting_rule": {}, + "error": {}, + "context": {}, + } + for i, raw := range arr { + event, ok := raw.(map[string]any) + if !ok { + t.Fatalf("flagEvaluations[%d] is not an object: %T", i, raw) + } + for k := range event { + if _, ok := allowedEventFields[k]; !ok { + t.Fatalf("flagEvaluations[%d] emitted unexpected field %q", i, k) + } + } + for _, required := range requiredFields { + if _, ok := event[required]; !ok { + t.Fatalf("flagEvaluations[%d] missing required field %q", i, required) + } + } + flag, ok := event["flag"].(map[string]any) + if !ok { + t.Fatalf("flagEvaluations[%d].flag is not an object: %T", i, event["flag"]) + } + if _, ok := flag["key"]; !ok { + t.Fatalf("flagEvaluations[%d].flag.key missing", i) + } + } + + full := arr[0].(map[string]any) + if _, ok := full["reason"]; ok { + t.Fatal("full EVP event must not emit OpenFeature reason") + } + if _, ok := full["targeting_key"]; !ok { + t.Fatal("full EVP event should retain targeting_key") + } + if _, ok := full["context"]; !ok { + t.Fatal("full EVP event should retain context") + } + + degraded := arr[1].(map[string]any) + if _, ok := degraded["reason"]; ok { + t.Fatal("degraded EVP event must not emit OpenFeature reason") + } + if _, ok := degraded["targeting_key"]; ok { + t.Fatal("degraded EVP event should omit targeting_key") + } + if _, ok := degraded["context"]; ok { + t.Fatal("degraded EVP event should omit context") + } + if _, ok := degraded["variant"]; !ok { + t.Fatal("degraded EVP event should retain schema-visible variant") + } + if _, ok := degraded["allocation"]; !ok { + t.Fatal("degraded EVP event should retain schema-visible allocation") + } + + requiredOnly := arr[2].(map[string]any) + for _, optional := range []string{"reason", "targeting_key", "variant", "allocation", "targeting_rule", "error", "context", "runtime_default_used"} { + if _, ok := requiredOnly[optional]; ok { + t.Fatalf("required-only EVP event should omit %q", optional) + } + } + }) +} + +// TestAggregatorDistinctAllocationBuckets verifies that allocation is part of the aggregation key. +func TestAggregatorDistinctAllocationBuckets(t *testing.T) { + agg := setupTestAggregator(t) + nowMs := time.Now().UnixMilli() + + // Two evaluations that differ only in allocationKey — they must be in separate full buckets. + d1 := evalDetails{ + flagKey: "my-flag", + variant: "on", + allocationKey: "alloc-a", + } + d2 := evalDetails{ + flagKey: "my-flag", + variant: "on", + allocationKey: "alloc-b", + } + + agg.add(d1, nil, nowMs) + agg.add(d2, nil, nowMs) + + agg.mu.Lock() + defer agg.mu.Unlock() + + if len(agg.full) != 2 { + t.Errorf("expected 2 separate full-tier buckets for distinct allocationKeys, got %d", len(agg.full)) + } + + // A second add with d1 must increment the existing bucket, not create a third + agg.mu.Unlock() + agg.add(d1, nil, nowMs) + agg.mu.Lock() + + if len(agg.full) != 2 { + t.Errorf("re-adding d1 must increment existing bucket, not create new one; got %d buckets", len(agg.full)) + } +} + +func TestOpenFeatureReasonIsNotEVPCardinality(t *testing.T) { + w := newFlagEvaluationWriter(ProviderConfig{}) + hookCtx := of.NewHookContext( + "reasonless-flag", + of.Boolean, + false, + of.NewClientMetadata(""), + of.Metadata{Name: "test-provider"}, + of.NewEvaluationContext("user-1", map[string]any{"country": "US"}), + ) + metadata := of.FlagMetadata{metadataAllocationKey: "alloc-a"} + detailsA := makeEvalDetails("on", of.TargetingMatchReason, "", metadata) + detailsB := makeEvalDetails("on", of.SplitReason, "", metadata) + + w.record(hookCtx, detailsA) + w.record(hookCtx, detailsB) + for len(w.events) > 0 { + w.aggregate(<-w.events) + } + + w.aggregator.mu.Lock() + defer w.aggregator.mu.Unlock() + if len(w.aggregator.full) != 1 { + t.Fatalf("reason-only differences must not split EVP buckets; got %d full buckets", len(w.aggregator.full)) + } + for _, e := range w.aggregator.full { + if e.count != 2 { + t.Fatalf("reason-only differences should aggregate into count=2, got %d", e.count) + } + } +} + +// TestAggregatorConcurrentMinMax verifies that 1000 goroutines recording the same key +// produce count==1000 and firstEvaluation<=lastEvaluation. +// Must be run with -race to satisfy the race-free requirement. +func TestAggregatorConcurrentMinMax(t *testing.T) { + // Caps large enough not to overflow during this test. + agg := newTestAggregator(100_000, 100_000, 100_000) + + d := evalDetails{ + flagKey: "concurrent-flag", + variant: "on", + } + + const goroutines = 1000 + var wg sync.WaitGroup + wg.Add(goroutines) + + for range goroutines { + go func() { + defer wg.Done() + nowMs := time.Now().UnixMilli() + agg.add(d, nil, nowMs) + }() + } + wg.Wait() + + agg.mu.Lock() + defer agg.mu.Unlock() + + if len(agg.full) != 1 { + t.Fatalf("expected exactly 1 full-tier bucket, got %d", len(agg.full)) + } + + for _, entry := range agg.full { + if entry.count != goroutines { + t.Errorf("expected count=%d, got %d", goroutines, entry.count) + } + if entry.firstEvaluation > entry.lastEvaluation { + t.Errorf("firstEvaluation=%d > lastEvaluation=%d — min/max invariant violated", + entry.firstEvaluation, entry.lastEvaluation) + } + } +} + +func TestEvaluationEntryObserveOutOfOrderTimestamps(t *testing.T) { + entry := newEvaluationEntry(200) + + entry.observe(150) + entry.observe(250) + + if entry.count != 3 { + t.Fatalf("count = %d, want 3", entry.count) + } + if entry.firstEvaluation != 150 { + t.Fatalf("firstEvaluation = %d, want 150", entry.firstEvaluation) + } + if entry.lastEvaluation != 250 { + t.Fatalf("lastEvaluation = %d, want 250", entry.lastEvaluation) + } +} + +// TestSaturationCountPreservation is the regression guard against a SILENT drop at saturation. +// The invariant is: Σ(full+degraded counts) + droppedDegradedOverflow == add() calls. +// No evaluation may vanish without being COUNTED — silent loss is the defect this guards against. +func TestSaturationCountPreservation(t *testing.T) { + // Use small caps so we can saturate them quickly. + // globalCap=5 means only 5 full-tier buckets ever created. + // perFlagCap=2 means after 2 distinct full-tier buckets per flag, it overflows to degraded. + // degradedCap=3 means only 3 degraded buckets; further overflow is dropped(counted). + agg := newTestAggregator(5, 2, 3) + nowMs := time.Now().UnixMilli() + + // Drive 100 distinct evaluations. Each add() must contribute exactly 1 count unit to either + // the full tier, the degraded tier, or the droppedDegradedOverflow counter. After all calls, + // Σ(full+degraded) + dropped must equal 100 — nothing silently lost. + const totalCalls = 100 + for i := range totalCalls { + flagIdx := i % 20 + allocIdx := i % 5 + d := evalDetails{ + flagKey: fmt.Sprintf("sat-flag-%d", flagIdx), + variant: "on", + allocationKey: fmt.Sprintf("alloc-%d", allocIdx), + targetingKey: fmt.Sprintf("user-%d", i%10), + } + agg.add(d, nil, nowMs) + } + + // Sum counts across both tiers plus the observable drop counter. + agg.mu.Lock() + defer agg.mu.Unlock() + + var totalCounted int64 + for _, e := range agg.full { + totalCounted += e.count + } + for _, e := range agg.degraded { + totalCounted += e.count + } + totalCounted += agg.droppedDegradedOverflow + + if totalCounted != totalCalls { + t.Errorf( + "count preservation violated: Σ(full+degraded)+dropped=%d, expected=%d (add() calls); "+ + "silent drops detected (full buckets=%d, degraded buckets=%d, droppedDegradedOverflow=%d)", + totalCounted, totalCalls, + len(agg.full), len(agg.degraded), agg.droppedDegradedOverflow, + ) + } +} + +// TestAggregatorCapOverflow verifies that: +// - Exceeding perFlagCap routes new entries to the degraded map. +// - Exceeding degradedCap drops new entries and counts the drop. +// - Global cap bounds total bucket growth. +func TestAggregatorCapOverflow(t *testing.T) { + t.Run("perFlagCap overflow routes to degraded", func(t *testing.T) { + agg := setupTestAggregator(t) // perFlagCap=3 + nowMs := time.Now().UnixMilli() + + // Fill perFlagCap (3) full-tier buckets for "flag-a" + for i := range 3 { + d := evalDetails{ + flagKey: "flag-a", + variant: "on", + allocationKey: fmt.Sprintf("alloc-%d", i), + targetingKey: fmt.Sprintf("user-%d", i), + } + agg.add(d, map[string]any{"key": fmt.Sprintf("v%d", i)}, nowMs) + } + + agg.mu.Lock() + if agg.perFlagFull["flag-a"] != 3 { + t.Errorf("expected perFlagFull[flag-a]=3, got %d", agg.perFlagFull["flag-a"]) + } + agg.mu.Unlock() + + // The 4th distinct entry for "flag-a" must overflow to degraded + d4 := evalDetails{ + flagKey: "flag-a", + variant: "on", + allocationKey: "alloc-overflow", + targetingKey: "user-overflow", + } + agg.add(d4, map[string]any{"extra": "data"}, nowMs) + + agg.mu.Lock() + defer agg.mu.Unlock() + + if len(agg.degraded) == 0 { + t.Error("expected at least one degraded bucket after perFlagCap overflow") + } + }) + + t.Run("degradedCap overflow is dropped and counted", func(t *testing.T) { + agg := setupTestAggregator(t) // degradedCap=3 + nowMs := time.Now().UnixMilli() + + // Pre-fill the degraded map to capacity by forcing overflow from full tier. + // Use different variants to get 3 distinct degraded buckets. + for i := range 4 { // 4 fills full to cap=3 then overflows once + for j := range 3 { // perFlagCap=3; 4 distinct allocs per flag => overflow on 4th + d := evalDetails{ + flagKey: fmt.Sprintf("flag-%d", i), + variant: fmt.Sprintf("v%d", j), + allocationKey: fmt.Sprintf("alloc-%d", j), + targetingKey: fmt.Sprintf("user-%d", j), + } + agg.add(d, nil, nowMs) + } + } + + // Continue adding until degradedCap is also exhausted. At that point, new degraded + // buckets must be dropped and COUNTED (droppedDegradedOverflow). + for i := range 10 { + d := evalDetails{ + flagKey: fmt.Sprintf("overflow-flag-%d", i), + variant: "on", + } + // Force each into degraded by also filling its full tier + for j := range 4 { + d2 := evalDetails{ + flagKey: d.flagKey, + variant: d.variant, + allocationKey: fmt.Sprintf("a%d", j), + } + agg.add(d2, nil, nowMs) + } + } + + agg.mu.Lock() + defer agg.mu.Unlock() + + if len(agg.degraded) > agg.degradedCap { + t.Errorf("degraded tier %d exceeds degradedCap %d — terminal tier not bounded", len(agg.degraded), agg.degradedCap) + } + if agg.droppedDegradedOverflow == 0 { + t.Error("expected droppedDegradedOverflow > 0 after degradedCap exhaustion (drops must be counted, not silent)") + } + }) + + t.Run("globalCap bounds full-tier bucket growth only", func(t *testing.T) { + agg := setupTestAggregator(t) // globalCap=10, perFlagCap=3, degradedCap=3 + nowMs := time.Now().UnixMilli() + + // Add 50 distinct evaluations (each a unique flag key). + // globalCap=10 caps the full tier; overflow cascades to degraded (then drops if degraded + // is also full). The full tier must stay at or below globalCap, and the total count + // across both tiers plus the drop counter must equal the number of add() calls. + const calls = 50 + for i := range calls { + d := evalDetails{ + flagKey: fmt.Sprintf("flag-%d", i), + variant: "on", + } + agg.add(d, nil, nowMs) + } + + agg.mu.Lock() + defer agg.mu.Unlock() + + // Full tier must be bounded by globalCap. + if agg.globalCount > agg.globalCap { + t.Errorf("full-tier globalCount %d exceeds globalCap %d", agg.globalCount, agg.globalCap) + } + if len(agg.full) > agg.globalCap { + t.Errorf("full-tier buckets %d exceeds globalCap %d", len(agg.full), agg.globalCap) + } + + // Every add() call must have produced a count unit somewhere observable (no silent drops). + var totalCounted int64 + for _, e := range agg.full { + totalCounted += e.count + } + for _, e := range agg.degraded { + totalCounted += e.count + } + totalCounted += agg.droppedDegradedOverflow + if totalCounted != calls { + t.Errorf("count preservation violated: Σ(full+degraded)+dropped=%d, expected=%d", totalCounted, calls) + } + }) +} + +// TestPruneContextDeterministic verifies deterministic context pruning. +// A >256-field context must prune to an IDENTICAL kept subset (and therefore an identical +// canonical key) on every call, and two independently-built maps with the same logical entries +// must produce an equal key. On the pre-fix code (map-range cap BEFORE sort) the kept subset is +// random, so the key varies across iterations and identical logical contexts fragment into +// separate buckets. +func TestPruneContextDeterministic(t *testing.T) { + const fields = 400 // > maxContextFields (256) + build := func() map[string]any { + m := make(map[string]any, fields) + for i := range fields { + m[fmt.Sprintf("key%04d", i)] = fmt.Sprintf("value%04d", i) + } + return m + } + + first := canonicalContextKey(pruneContext(build())) + for range 50 { + got := canonicalContextKey(pruneContext(build())) + if got != first { + t.Fatalf("pruneContext+canonicalContextKey nondeterministic over >256 fields: keys differ across iterations") + } + } + + // Two independently-built maps with the SAME 400 logical entries must produce an equal key. + if a, b := canonicalContextKey(pruneContext(build())), canonicalContextKey(pruneContext(build())); a != b { + t.Errorf("identical logical contexts produced different canonical keys") + } +} + +// TestPruneContextOversizedStringDeterministic verifies an oversized-string +// skip among >256 fields: the oversized-string skip must be applied against a deterministic +// key ordering, so the kept subset (and hash) is stable across iterations. +func TestPruneContextOversizedStringDeterministic(t *testing.T) { + const fields = 400 + longVal := strings.Repeat("x", maxFieldLength+44) // > maxFieldLength → skipped + build := func() map[string]any { + m := make(map[string]any, fields) + for i := range fields { + m[fmt.Sprintf("key%04d", i)] = fmt.Sprintf("value%04d", i) + } + m["zzz-oversized"] = longVal // sorts last; deterministically skipped + return m + } + + first := canonicalContextKey(pruneContext(build())) + for range 50 { + got := canonicalContextKey(pruneContext(build())) + if got != first { + t.Fatalf("oversized-string prune nondeterministic: canonical keys differ across iterations") + } + } + + // The oversized value must never appear in the pruned subset. + pruned := pruneContext(build()) + if _, ok := pruned["zzz-oversized"]; ok { + t.Error("oversized string value should be skipped from pruned context") + } +} + +// TestCanonicalContextKeyEncoding verifies that the comparable canonical-context key replaces +// the lossy FNV-1a discriminator). Distinct contexts must produce DISTINCT keys — int 1 vs +// string "1" must differ, and '='/'\n'-bearing values/keys must not fake a multi-field context. +// Because the full canonical encoding IS the map key (no hash), distinct contexts ALWAYS land +// in separate full-tier buckets via add() with ZERO misattribution. +func TestCanonicalContextKeyEncoding(t *testing.T) { + // Distinct contexts must produce distinct keys — no aliasing across type or delimiter tricks. + inequalityTests := []struct { + name string + mapA, mapB map[string]any + }{ + { + // Type-tagged encoding must distinguish int 1 from string "1". + name: "int 1 != string 1", + mapA: map[string]any{"x": 1}, + mapB: map[string]any{"x": "1"}, + }, + { + // {"a=b":"c"} vs {"a":"b=c"} render identically under key+"="+value with no + // length delimiter; canonical encoding must keep them distinct. + name: "'=' in key/value cannot alias a multi-field context", + mapA: map[string]any{"a=b": "c"}, + mapB: map[string]any{"a": "b=c"}, + }, + { + // Under key+"="+value+"\n", a newline in a value would collide with a two-field map. + name: "'\\n' in value cannot alias a multi-field context", + mapA: map[string]any{"a": "x\ny", "b": "z"}, + mapB: map[string]any{"a": "x", "y=z": ""}, + }, + } + + for _, tc := range inequalityTests { + t.Run(tc.name, func(t *testing.T) { + if canonicalContextKey(tc.mapA) == canonicalContextKey(tc.mapB) { + t.Errorf("canonical key must distinguish %v from %v", tc.mapA, tc.mapB) + } + }) + } + + t.Run("supported numeric scalar types produce distinct keys", func(t *testing.T) { + values := []struct { + name string + value any + }{ + {name: "int", value: int(1)}, + {name: "int8", value: int8(1)}, + {name: "int16", value: int16(1)}, + {name: "int32", value: int32(1)}, + {name: "int64", value: int64(1)}, + {name: "uint", value: uint(1)}, + {name: "uint8", value: uint8(1)}, + {name: "uint16", value: uint16(1)}, + {name: "uint32", value: uint32(1)}, + {name: "uint64", value: uint64(1)}, + {name: "float32", value: float32(1)}, + {name: "float64", value: float64(1)}, + } + + keys := make(map[string]string, len(values)) + for _, item := range values { + key := canonicalContextKey(map[string]any{"x": item.value}) + if other, ok := keys[key]; ok { + t.Fatalf("%s and %s produced the same canonical context key", item.name, other) + } + keys[key] = item.name + } + }) + + // Logically-identical contexts must produce the SAME key (so they aggregate into one bucket). + t.Run("identical contexts produce identical keys", func(t *testing.T) { + a := canonicalContextKey(map[string]any{"x": 1, "y": "two"}) + b := canonicalContextKey(map[string]any{"y": "two", "x": 1}) + if a != b { + t.Errorf("logically-identical contexts produced different canonical keys") + } + }) + + // Each distinct-context case must land in its OWN full-tier bucket via add() — the count is + // never misattributed to the other context (the defect the lossy FNV discriminator risked). + for _, tc := range inequalityTests { + t.Run("distinct buckets via add(): "+tc.name, func(t *testing.T) { + agg := setupTestAggregator(t) + nowMs := time.Now().UnixMilli() + d := evalDetails{flagKey: "f", variant: "on"} + agg.add(d, tc.mapA, nowMs) + agg.add(d, tc.mapB, nowMs) + + agg.mu.Lock() + defer agg.mu.Unlock() + if len(agg.full) != 2 { + t.Errorf("expected 2 full-tier buckets for distinct contexts %v vs %v, got %d", tc.mapA, tc.mapB, len(agg.full)) + } + // Zero misattribution: every bucket holds exactly the one count it received. + for k, e := range agg.full { + if e.count != 1 { + t.Errorf("bucket %+v has count %d; distinct contexts must not merge (misattribution)", k, e.count) + } + } + }) + } + + // A multi-field context with a key/value containing '\n' and '=' must still aggregate + // identically with itself (re-adding increments the SAME bucket, not a third). + t.Run("re-adding identical multi-field context increments same bucket", func(t *testing.T) { + agg := setupTestAggregator(t) + nowMs := time.Now().UnixMilli() + d := evalDetails{flagKey: "f", variant: "on"} + ctx := map[string]any{"a": "x\ny", "b": 7, "c=d": true} + agg.add(d, ctx, nowMs) + agg.add(d, map[string]any{"b": 7, "a": "x\ny", "c=d": true}, nowMs) // same logical context, different insertion order + + agg.mu.Lock() + defer agg.mu.Unlock() + if len(agg.full) != 1 { + t.Errorf("expected 1 full-tier bucket for identical context, got %d", len(agg.full)) + } + for _, e := range agg.full { + if e.count != 2 { + t.Errorf("expected count 2 for re-added identical context, got %d", e.count) + } + } + }) +} + +// TestDegradedCapBounded verifies that unbounded dynamic/abusive flag keys stay bounded. +// An unbounded number of distinct flag keys must NOT grow the degraded map without bound: +// len(degraded) <= degradedCap, and the over-cap counts must be DROPPED-AND-COUNTED +// (droppedDegradedOverflow), never silently lost. +// Σ(full+degraded counts) + droppedDegradedOverflow must equal the add() call count. +func TestDegradedCapBounded(t *testing.T) { + const cap = 3 + // globalCap=0 forces every distinct full key straight past the full tier into degraded. + agg := newTestAggregator(0, 100_000, cap) + nowMs := time.Now().UnixMilli() + + const calls = 100 + for i := range calls { + d := evalDetails{ + flagKey: fmt.Sprintf("dynamic-flag-%d", i), // every key distinct + variant: "on", + } + agg.add(d, nil, nowMs) + } + + agg.mu.Lock() + defer agg.mu.Unlock() + + if len(agg.degraded) > cap { + t.Errorf("degraded cardinality %d exceeds degradedCap (%d) — terminal tier not bounded", len(agg.degraded), cap) + } + + var total int64 + for _, e := range agg.full { + total += e.count + } + for _, e := range agg.degraded { + total += e.count + } + total += agg.droppedDegradedOverflow + if total != calls { + t.Errorf("count preservation violated under degradedCap: Σ(full+degraded)+dropped=%d, expected=%d", total, calls) + } + + // The over-cap counts must be observable in the drop counter. + if agg.droppedDegradedOverflow == 0 { + t.Errorf("expected droppedDegradedOverflow > 0 when distinct keys exceed degradedCap") + } +} + +// TestRecordAfterStopIsNoop verifies record() cannot enqueue after shutdown. After +// stop(), record() must NOT enqueue into the never-drained events channel; the event must be +// counted as dropped instead. On the pre-fix code (no stopped check in record()) the event is +// enqueued and silently lost. +func TestRecordAfterStopIsNoop(t *testing.T) { + w := newFlagEvaluationWriter(ProviderConfig{}) + + // Do NOT start the worker. stop() must still be safe (ticker==nil path) and mark stopped. + w.stop() + + before := w.dropped.Load() + + // Build a minimal hook context + details to drive record(). + hookCtx := of.NewHookContext( + "test-flag", + of.Boolean, + false, + of.NewClientMetadata(""), + of.Metadata{Name: "test-provider"}, + of.NewEvaluationContext("user-1", nil), + ) + details := of.InterfaceEvaluationDetails{ + Value: true, + EvaluationDetails: of.EvaluationDetails{ + FlagKey: "test-flag", + FlagType: of.Boolean, + ResolutionDetail: of.ResolutionDetail{ + Variant: "on", + Reason: of.TargetingMatchReason, + }, + }, + } + + w.record(hookCtx, details) + + if got := len(w.events); got != 0 { + t.Errorf("record() after stop() enqueued %d event(s); expected 0 (no-op into never-drained channel)", got) + } + if got := w.dropped.Load(); got != before+1 { + t.Errorf("record() after stop() must count the event as dropped: dropped=%d, expected=%d", got, before+1) + } +} + +func TestStopDrainsAndFlushesQueuedFlagEvaluations(t *testing.T) { + payloads := make(chan flagEvaluationPayload, 1) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + defer r.Body.Close() + + if r.URL.Path != flagEvaluationEndpoint { + t.Errorf("unexpected EVP path: got %q, want %q", r.URL.Path, flagEvaluationEndpoint) + } + + var payload flagEvaluationPayload + if err := json.NewDecoder(r.Body).Decode(&payload); err != nil { + t.Errorf("failed to decode flagevaluation payload: %v", err) + w.WriteHeader(http.StatusBadRequest) + return + } + payloads <- payload + w.WriteHeader(http.StatusAccepted) + })) + defer server.Close() + + agentURL, err := url.Parse(server.URL) + if err != nil { + t.Fatalf("failed to parse test server URL: %v", err) + } + + evp := newEVPClient() + evp.agentURL = agentURL + evp.httpClient = server.Client() + + w := newFlagEvaluationWriterWithEVP(ProviderConfig{FlagEvaluationFlushInterval: time.Hour}, evp) + w.start() + + hookCtx := of.NewHookContext( + "test-flag", + of.Boolean, + false, + of.NewClientMetadata(""), + of.Metadata{Name: "test-provider"}, + of.NewEvaluationContext("user-1", map[string]any{"country": "US"}), + ) + details := of.InterfaceEvaluationDetails{ + Value: true, + EvaluationDetails: of.EvaluationDetails{ + FlagKey: "test-flag", + FlagType: of.Boolean, + ResolutionDetail: of.ResolutionDetail{ + Variant: "on", + Reason: of.TargetingMatchReason, + }, + }, + } + w.record(hookCtx, details) + + w.stop() + + select { + case payload := <-payloads: + if len(payload.FlagEvaluations) != 1 { + t.Fatalf("expected 1 flushed flagevaluation event, got %d", len(payload.FlagEvaluations)) + } + event := payload.FlagEvaluations[0] + if event.Flag.Key != "test-flag" { + t.Errorf("unexpected flag key: got %q, want test-flag", event.Flag.Key) + } + if event.EvaluationCount != 1 { + t.Errorf("unexpected evaluation count: got %d, want 1", event.EvaluationCount) + } + case <-time.After(2 * time.Second): + t.Fatal("stop() returned without flushing queued flagevaluation event") + } +} + +func TestRecordAggregatesPrunedContextSnapshot(t *testing.T) { + w := newFlagEvaluationWriter(ProviderConfig{}) + attrs := make(map[string]any, maxContextFields+50) + for i := range maxContextFields + 50 { + attrs[fmt.Sprintf("field-%03d", i)] = fmt.Sprintf("value-%03d", i) + } + attrs["zzz-oversized"] = strings.Repeat("x", maxFieldLength+1) + hookCtx := of.NewHookContext( + "test-flag", + of.Boolean, + false, + of.NewClientMetadata(""), + of.Metadata{Name: "test-provider"}, + of.NewEvaluationContext("user-1", attrs), + ) + details := of.InterfaceEvaluationDetails{ + Value: true, + EvaluationDetails: of.EvaluationDetails{ + FlagKey: "test-flag", + FlagType: of.Boolean, + ResolutionDetail: of.ResolutionDetail{ + Variant: "on", + Reason: of.TargetingMatchReason, + }, + }, + } + + w.record(hookCtx, details) + + if len(w.events) != 1 { + t.Fatalf("expected one queued event, got %d", len(w.events)) + } + w.aggregate(<-w.events) + + if len(w.aggregator.full) != 1 { + t.Fatalf("expected one aggregated entry, got %d", len(w.aggregator.full)) + } + for _, entry := range w.aggregator.full { + if got := len(entry.contextAttrs); got != maxContextFields { + t.Fatalf("aggregated context should be pruned to %d fields, got %d", maxContextFields, got) + } + if _, ok := entry.contextAttrs["zzz-oversized"]; ok { + t.Fatal("aggregated context should not contain oversized string values") + } + } +} + +// TestExtractEvalDetailsPrefersErrorMessage verifies ErrorMessage is preferred when present; +// ErrorCode is the fallback only when ErrorMessage is empty. +func TestExtractEvalDetailsPrefersErrorMessage(t *testing.T) { + mkHookCtx := func() of.HookContext { + return of.NewHookContext( + "test-flag", + of.Boolean, + false, + of.NewClientMetadata(""), + of.Metadata{Name: "test-provider"}, + of.NewEvaluationContext("user-1", nil), + ) + } + + tests := []struct { + name string + details of.InterfaceEvaluationDetails + wantErrorMessage string + }{ + { + name: "ErrorMessage present is preferred over ErrorCode", + details: of.InterfaceEvaluationDetails{ + EvaluationDetails: of.EvaluationDetails{ + ResolutionDetail: of.ResolutionDetail{ + Reason: of.ErrorReason, + ErrorCode: of.GeneralCode, + ErrorMessage: "boom", + }, + }, + }, + wantErrorMessage: "boom", + }, + { + name: "empty ErrorMessage falls back to ErrorCode", + details: of.InterfaceEvaluationDetails{ + EvaluationDetails: of.EvaluationDetails{ + ResolutionDetail: of.ResolutionDetail{ + Reason: of.ErrorReason, + ErrorCode: of.TypeMismatchCode, + }, + }, + }, + wantErrorMessage: string(of.TypeMismatchCode), + }, + { + name: "both empty yields empty errorMessage", + details: of.InterfaceEvaluationDetails{ + EvaluationDetails: of.EvaluationDetails{ + ResolutionDetail: of.ResolutionDetail{ + Reason: of.TargetingMatchReason, + }, + }, + }, + wantErrorMessage: "", + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + if got := extractEvalDetails(mkHookCtx(), tc.details).errorMessage; got != tc.wantErrorMessage { + t.Errorf("errorMessage = %q, want %q", got, tc.wantErrorMessage) + } + }) + } +} diff --git a/openfeature/provider.go b/openfeature/provider.go index 78bda66b7bc..d385a9d1a07 100644 --- a/openfeature/provider.go +++ b/openfeature/provider.go @@ -34,6 +34,11 @@ var ( const ( // ffeProductEnvVar is the environment variable to enable the experimental flagging provider ffeProductEnvVar = "DD_EXPERIMENTAL_FLAGGING_PROVIDER_ENABLED" + // flagEvalCountsEnabledEnvVar is the operator killswitch for the EVP flagevaluation emission path. + // Default: true (EVP path is ON by default). Set to "false" to disable only the EVP path + // while leaving the OTel feature_flag.evaluations path unaffected. + // Mirrors the internal.BoolEnv convention used by ffeProductEnvVar. + flagEvalCountsEnabledEnvVar = "DD_FLAGGING_EVALUATION_COUNTS_ENABLED" // Default timeout for provider initialization defaultInitTimeout = 30 * time.Second // Default timeout for provider shutdown @@ -45,6 +50,10 @@ type ProviderConfig struct { // ExposureFlushInterval is the interval at which exposure events are flushed to the agent // Default: 1 second ExposureFlushInterval time.Duration + + // FlagEvaluationFlushInterval is the interval for flushing EVP flag evaluation events. + // Default: 10 seconds. Leave zero to use the default. + FlagEvaluationFlushInterval time.Duration } // DatadogProvider is an OpenFeature provider that evaluates feature flags @@ -62,6 +71,12 @@ type DatadogProvider struct { // Flag evaluation metrics hook (OTel counter via Finally hook) flagEvalHook *flagEvalHook + + // Flag evaluation EVP writer + hook (new Path B — EVP flagevaluation track). + // Both fields are nil when DD_FLAGGING_EVALUATION_COUNTS_ENABLED=false (killswitch). + // Named distinctly from flagEvalHook (OTel) to avoid collisions. + flagEvalWriter *flagEvaluationWriter + flagEvalEVPHook *flagEvaluationHook } // NewDatadogProvider creates a new Datadog OpenFeature provider with default configuration. @@ -83,8 +98,10 @@ func NewDatadogProvider(config ProviderConfig) (openfeature.FeatureProvider, err } func newDatadogProvider(config ProviderConfig) *DatadogProvider { + evp := newEVPClient() + // Create exposure writer - writer := newExposureWriter(config) + writer := newExposureWriterWithEVP(config, evp) // Create exposure hook hook := newExposureHook(writer) @@ -103,6 +120,17 @@ func newDatadogProvider(config ProviderConfig) *DatadogProvider { exposureHook: hook, flagEvalHook: newFlagEvalHook(metrics), } + + // Conditionally construct the EVP flagevaluation writer + hook. + // Gated by DD_FLAGGING_EVALUATION_COUNTS_ENABLED (default true). + // When false, both fields are left nil and the EVP path is disabled. + // The OTel hook (flagEvalHook above) is registered unconditionally. + if internal.BoolEnv(flagEvalCountsEnabledEnvVar, true) { + evalWriter := newFlagEvaluationWriterWithEVP(config, evp) + p.flagEvalWriter = evalWriter + p.flagEvalEVPHook = newFlagEvaluationHook(evalWriter) + } + p.configChange.L = &p.mu return p } @@ -183,8 +211,12 @@ func (p *DatadogProvider) InitWithContext(ctx context.Context, _ openfeature.Eva } } - // Start periodic flushing + // Start periodic flushing for exposure writer. p.exposureWriter.start() + // Start periodic flushing for EVP flag evaluation writer (nil when killswitch disabled). + if p.flagEvalWriter != nil { + p.flagEvalWriter.start() + } return nil } @@ -215,6 +247,10 @@ func (p *DatadogProvider) ShutdownWithContext(ctx context.Context) error { p.exposureWriter.flush() p.exposureWriter.stop() } + // Stop the EVP flag evaluation writer (nil when killswitch disabled). + if p.flagEvalWriter != nil { + p.flagEvalWriter.stop() + } // Shut down flag evaluation metrics if p.flagEvalHook != nil && p.flagEvalHook.metrics != nil { _ = p.flagEvalHook.metrics.shutdown(ctx) @@ -409,15 +445,21 @@ func (p *DatadogProvider) ObjectEvaluation( } // Hooks returns the hooks for this provider. -// This includes the exposure tracking hook and the flag evaluation metrics hook. +// This includes the exposure tracking hook, the OTel flag evaluation metrics hook (always), +// and the EVP flagevaluation hook (only when DD_FLAGGING_EVALUATION_COUNTS_ENABLED is true). func (p *DatadogProvider) Hooks() []openfeature.Hook { var hooks []openfeature.Hook if p.exposureHook != nil { hooks = append(hooks, p.exposureHook) } + // OTel hook is always registered — untouched by the EVP killswitch. if p.flagEvalHook != nil { hooks = append(hooks, p.flagEvalHook) } + // EVP hook is nil when the killswitch disabled it. + if p.flagEvalEVPHook != nil { + hooks = append(hooks, p.flagEvalEVPHook) + } return hooks } @@ -428,9 +470,15 @@ func (p *DatadogProvider) evaluate( defaultValue any, flatCtx openfeature.FlattenedContext, ) (res evaluationResult) { + // Capture the evaluation time once, at evaluation entry. It is used for allocation + // time-window checks and EVP first/last evaluation bounds. + evalNow := time.Now() log.Debug("openfeature: evaluating flag %q", flagKey) defer func() { - log.Debug("openfeature: evaluated flag %q: value=%v, reason=%s, error=%v", flagKey, res.Value, res.Reason, res.Error) + if res.Metadata == nil { + res.Metadata = make(map[string]any, 1) + } + res.Metadata[metadataEvalTimeKey] = evalNow.UnixMilli() }() // Check if context was cancelled before starting evaluation @@ -465,8 +513,8 @@ func (p *DatadogProvider) evaluate( } } - // Evaluate the flag (pass context for potential future use in evaluateFlag) - return evaluateFlag(flag, defaultValue, flatCtx) + // Evaluate the flag, sharing the eval-time captured at entry. + return evaluateFlag(flag, defaultValue, flatCtx, evalNow) } // toResolutionError converts a Go error to an OpenFeature ResolutionError. diff --git a/openfeature/provider_bench_test.go b/openfeature/provider_bench_test.go index 1375583e9bd..552b6478e20 100644 --- a/openfeature/provider_bench_test.go +++ b/openfeature/provider_bench_test.go @@ -7,7 +7,10 @@ package openfeature import ( "context" + "sort" + "strconv" "testing" + "time" "github.com/open-feature/go-sdk/openfeature" "github.com/stretchr/testify/require" @@ -280,3 +283,255 @@ func BenchmarkConcurrentEvaluations(b *testing.B) { } }) } + +// makeBenchmarkConfig creates a test config with the specified number of flags. +// Extends createTestConfig() for benchmark load profiles. +// Flag keys are unique monotonic integers ("bench-flag-N") so the claimed cardinality +// is fully exercised without wrapping. +func makeBenchmarkConfig(numFlags int) *universalFlagsConfiguration { + config := createTestConfig() + for i := len(config.Flags); i < numFlags; i++ { + flagKey := "bench-flag-" + strconv.Itoa(i) + config.Flags[flagKey] = &flag{ + Key: flagKey, + Enabled: true, + VariationType: valueTypeBoolean, + Variations: map[string]*variant{ + "on": {Key: "on", Value: true}, + }, + Allocations: []*allocation{ + { + Key: "alloc-bench", + Rules: []*rule{}, + Splits: []*split{ + {Shards: []*shard{}, VariationKey: "on"}, + }, + }, + }, + } + } + return config +} + +// boolBenchmarkFlagKeys returns the boolean flag keys in cfg in deterministic order, so a +// benchmark can rotate the evaluated flag and exercise flag-cardinality (not just users). +func boolBenchmarkFlagKeys(cfg *universalFlagsConfiguration) []string { + keys := make([]string, 0, len(cfg.Flags)) + for k, f := range cfg.Flags { + if f.VariationType == valueTypeBoolean { + keys = append(keys, k) + } + } + sort.Strings(keys) + return keys +} + +// makeBenchmarkAttrs builds an evaluation-context attribute map with numFields-1 distinct +// fields (the targeting key is supplied separately to NewEvaluationContext, so the flattened +// context has numFields entries total). Field names are "fieldN" so all are distinct. +func makeBenchmarkAttrs(numFields int) map[string]any { + attrs := make(map[string]any, numFields) + for i := 1; i < numFields; i++ { + attrs["field"+strconv.Itoa(i)] = "value" + } + return attrs +} + +// flagEvalBenchProfile is one load profile for the three-column overhead comparison. +type flagEvalBenchProfile struct { + name string + numFlags int + numUsers int + numFields int +} + +// flagEvalBenchProfiles are shared by the Noop / OTel-only / OTel+EVP benchmarks so the +// three columns are directly comparable. +var flagEvalBenchProfiles = []flagEvalBenchProfile{ + {"typical/100flags_50users_10fields", 100, 50, 10}, + {"stress/10flags_1000users_250fields", 10, 1000, 250}, + // scale profile targets the team's >=2,500-flag goal. Flag count is the dimension under + // test, so it dominates; users/fields are kept modest (500/20) so the flag-cardinality + // signal is not swamped by per-evaluation context cost and the suite stays tractable. + {"scale/2500flags_500users_20fields", 2500, 500, 20}, +} + +// runFlagEvalBenchmark drives evaluations through a real OpenFeature client so the provider's +// registered hooks actually run - calling provider.BooleanEvaluation directly would bypass +// Hooks() and every column would measure the same bare evaluator. Flag and targeting keys +// rotate to exercise flag- and user-cardinality. +// +// configureHooks nils whichever provider hooks the tier under test should exclude (it runs +// before registration; Init does not recreate hooks, so the choice sticks). The 24h flush +// interval keeps the EVP writer from flushing during the run, so no HTTP round-trip enters +// the hot path - isolating hook + aggregator cost. +func runFlagEvalBenchmark(b *testing.B, configureHooks func(p *DatadogProvider)) { + b.Helper() + for _, p := range flagEvalBenchProfiles { + b.Run(p.name, func(b *testing.B) { + provider := newDatadogProvider(ProviderConfig{FlagEvaluationFlushInterval: 24 * time.Hour}) + configureHooks(provider) + + config := makeBenchmarkConfig(p.numFlags) + provider.updateConfiguration(config) + + initCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + if err := openfeature.SetProviderWithContextAndWait(initCtx, provider); err != nil { + cancel() + b.Fatalf("set provider: %v", err) + } + cancel() + client := openfeature.NewClient("flageval-bench") + + flagKeys := boolBenchmarkFlagKeys(config) + attrs := makeBenchmarkAttrs(p.numFields) + ctx := context.Background() + + b.ReportAllocs() + b.ResetTimer() + + i := 0 + for b.Loop() { + evalCtx := openfeature.NewEvaluationContext("bench-user-"+strconv.Itoa(i%p.numUsers), attrs) + _ = client.Boolean(ctx, flagKeys[i%len(flagKeys)], false, evalCtx) + i++ + } + }) + } +} + +// BenchmarkFlagEvaluationNoop measures the client+provider evaluation cost with no hooks - +// the baseline column for the three-column overhead comparison. +// +// Run command: +// +// GOFLAGS=-mod=readonly go test ./openfeature -run='^$' -bench='^BenchmarkFlagEvaluation' \ +// -benchmem -count=3 -cpu=8 +func BenchmarkFlagEvaluationNoop(b *testing.B) { + runFlagEvalBenchmark(b, func(p *DatadogProvider) { + p.flagEvalHook = nil + p.flagEvalEVPHook = nil + p.flagEvalWriter = nil + p.exposureHook = nil + }) +} + +// BenchmarkFlagEvaluationOTelOnly measures the cost with only the existing OTel +// feature_flag.evaluations hook (Path A - the preserved baseline). The EVP hook and writer +// are disabled so this column isolates the OTel hook's cost. +func BenchmarkFlagEvaluationOTelOnly(b *testing.B) { + runFlagEvalBenchmark(b, func(p *DatadogProvider) { + p.flagEvalEVPHook = nil + p.flagEvalWriter = nil + p.exposureHook = nil + }) +} + +// BenchmarkFlagEvaluationOTelPlusEVP measures the marginal cost of adding the new EVP +// flagevaluation hook alongside the existing OTel hook (Path A + Path B). Only the exposure +// hook is disabled; the OTel hook, EVP hook, and aggregator all run. +func BenchmarkFlagEvaluationOTelPlusEVP(b *testing.B) { + runFlagEvalBenchmark(b, func(p *DatadogProvider) { + p.exposureHook = nil + }) +} + +// BenchmarkFlagEvaluationEVPRecord isolates the EVP hook's synchronous hot-path cost: scalar +// extraction + shallow context copy + non-blocking enqueue, all on the evaluation goroutine. +// A discard drainer keeps the queue empty so the asynchronous aggregation (flatten/prune/hash/ +// add, done by the real worker off the hot path) is NOT attributed here - this is the latency a +// flag evaluation actually pays when the EVP path is enabled. +func BenchmarkFlagEvaluationEVPRecord(b *testing.B) { + for _, p := range flagEvalBenchProfiles { + b.Run(p.name, func(b *testing.B) { + w := newFlagEvaluationWriter(ProviderConfig{FlagEvaluationFlushInterval: 24 * time.Hour}) + + // Discard drainer: keep the queue empty without aggregating, so only the + // synchronous enqueue cost is measured here. + done := make(chan struct{}) + go func() { + for { + select { + case <-w.events: + case <-done: + return + } + } + }() + defer close(done) + + hookCtx := makeHookContext("bool-flag", "bench-user", makeBenchmarkAttrs(p.numFields)) + details := makeEvalDetails("on", openfeature.TargetingMatchReason, "") + + b.ReportAllocs() + b.ResetTimer() + for b.Loop() { + w.record(hookCtx, details) + } + }) + } +} + +// scaleBenchProfile is the >=2,500-flag profile (the team's scale target) selected from +// flagEvalBenchProfiles by name. It panics if the profile is removed so the scale benches +// fail loudly rather than silently degrade to a smaller config. +func scaleBenchProfile() flagEvalBenchProfile { + const name = "scale/2500flags_500users_20fields" + for _, p := range flagEvalBenchProfiles { + if p.name == name { + return p + } + } + panic("scale profile " + name + " missing from flagEvalBenchProfiles") +} + +// BenchmarkFlagEvaluationOTelPlusEVPParallel measures concurrent server-side evaluation +// through a real OpenFeature client with BOTH the OTel hook and the new EVP flagevaluation +// hook enabled, at the >=2,500-flag scale profile. Because every +// concurrent evaluation funnels through the EVP writer's single aggregator mutex +// (flagEvaluationAggregator.add), this is the bench that surfaces lock contention under +// realistic multi-goroutine server load. +// +// Each goroutine rotates its own flag key and targeting key across the full cardinality so the +// aggregator sees a realistic spread of buckets, not a single hot key. The 24h flush interval +// keeps the EVP writer from flushing mid-run so no HTTP round-trip enters the hot path. +// +// Run command: +// +// GOFLAGS=-mod=readonly go test ./openfeature -run='^$' \ +// -bench='^BenchmarkFlagEvaluationOTelPlusEVPParallel$' -benchmem -count=2 -cpu=8 +func BenchmarkFlagEvaluationOTelPlusEVPParallel(b *testing.B) { + p := scaleBenchProfile() + + provider := newDatadogProvider(ProviderConfig{FlagEvaluationFlushInterval: 24 * time.Hour}) + // Disable only the exposure hook; OTel hook + EVP hook + aggregator all run, so this + // measures the combined Path A + Path B cost under contention. + provider.exposureHook = nil + + config := makeBenchmarkConfig(p.numFlags) + provider.updateConfiguration(config) + + initCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + if err := openfeature.SetProviderWithContextAndWait(initCtx, provider); err != nil { + cancel() + b.Fatalf("set provider: %v", err) + } + cancel() + client := openfeature.NewClient("flageval-bench-parallel") + + flagKeys := boolBenchmarkFlagKeys(config) + attrs := makeBenchmarkAttrs(p.numFields) + ctx := context.Background() + + b.ReportAllocs() + b.ResetTimer() + + b.RunParallel(func(pb *testing.PB) { + i := 0 + for pb.Next() { + evalCtx := openfeature.NewEvaluationContext("bench-user-"+strconv.Itoa(i%p.numUsers), attrs) + _ = client.Boolean(ctx, flagKeys[i%len(flagKeys)], false, evalCtx) + i++ + } + }) +} diff --git a/openfeature/provider_test.go b/openfeature/provider_test.go index 76ee7e217c6..51e6d6489c7 100644 --- a/openfeature/provider_test.go +++ b/openfeature/provider_test.go @@ -255,8 +255,47 @@ func TestNewDatadogProvider(t *testing.T) { } hooks := provider.Hooks() - if len(hooks) != 2 { - t.Errorf("expected 2 hooks (exposure + flag eval metrics), got %d", len(hooks)) + // 3 hooks: exposure hook + OTel flag eval metrics hook + EVP flagevaluation hook. + // The EVP hook is enabled by default (DD_FLAGGING_EVALUATION_COUNTS_ENABLED=true). + if len(hooks) != 3 { + t.Errorf("expected 3 hooks (exposure + flag eval metrics + EVP flagevaluation), got %d", len(hooks)) + } +} + +// TestEvaluateStampsEvalTimeMetadata verifies the provider stamps evaluation-entry +// time into FlagMetadata on every path so EVP first/last bounds use eval-time. +func TestEvaluateStampsEvalTimeMetadata(t *testing.T) { + provider := newDatadogProvider(ProviderConfig{}) + provider.updateConfiguration(createTestConfig()) + ctx := context.Background() + + cases := []struct { + name string + flagKey string + flatCtx openfeature.FlattenedContext + }{ + {"matched allocation", "bool-flag", openfeature.FlattenedContext{"targetingKey": "user-123", "country": "US"}}, + {"default (no match)", "bool-flag", openfeature.FlattenedContext{"targetingKey": "user-123", "country": "CA"}}, + {"disabled flag", "disabled-flag", openfeature.FlattenedContext{"targetingKey": "user-123"}}, + {"flag not found", "nonexistent-flag", openfeature.FlattenedContext{"targetingKey": "user-123"}}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + before := time.Now().UnixMilli() + result := provider.BooleanEvaluation(ctx, tc.flagKey, false, tc.flatCtx) + after := time.Now().UnixMilli() + + got, ok := result.FlagMetadata[metadataEvalTimeKey].(int64) + if !ok { + t.Fatalf("FlagMetadata[%q] missing or wrong type on %q path; metadata=%v", + metadataEvalTimeKey, tc.name, result.FlagMetadata) + } + if got < before || got > after { + t.Fatalf("FlagMetadata[%q] = %d, want within [%d,%d] on %q path", + metadataEvalTimeKey, got, before, after, tc.name) + } + }) } }