Class: Datadog::OpenFeature::FlagEvaluation::Aggregator

Inherits:
Object
  • Object
show all
Defined in:
lib/datadog/open_feature/flag_evaluation/aggregator.rb

Overview

Two-tier aggregation for EVP flagevaluation events.

Two-tier design:

  • full-tier key: (flag_key, variant, allocation_key, runtime_default, error_message, targeting_key, canonical_context_key, observe_full_evaluation_data)
  • degraded-tier key: (flag_key, variant, allocation_key, runtime_default, error_message)
  • Drop-and-count when degraded tier is full
  • canonical_context_key: sorted type-tagged length-delimited encoding (no hash digest)
  • Caps: global_cap=131_072 / per_flag_cap=10_000 / degraded_cap=32_768

Context bounding: the writer applies bounded_context_snapshot on the evaluation thread before enqueue, so the queue only ever holds an already-bounded, flattened snapshot. record receives that snapshot and keys on it directly.

Constant Summary collapse

MAX_CONTEXT_FIELDS =

Cross-SDK context caps. The per-dimension caps match Go and Java.

256
MAX_VALUE_LENGTH = 
256
MAX_KEY_LENGTH = 
256
MAX_LIST_ELEMENTS = 
256
MAX_STRUCTURE_PROPERTIES = 
256
MAX_SNAPSHOT_DEPTH = 
4
MAX_VISITED_NODES =

Total-node budget bounding leaf-free shared subtrees, which do not increase the retained field count. It still permits every retained field to sit at the maximum snapshot depth.

MAX_CONTEXT_FIELDS * (MAX_SNAPSHOT_DEPTH + 1)
REASON_MAX_CONTEXT_FIELDS =

Truncation reason labels, surfaced on the flagevaluation.context.truncated telemetry counter so operators can tell which cap was hit.

"max_context_fields"
REASON_MAX_VALUE_LENGTH = 
"max_value_length"
REASON_MAX_KEY_LENGTH = 
"max_key_length"
REASON_MAX_LIST_ELEMENTS = 
"max_list_elements"
REASON_MAX_STRUCTURE_PROPERTIES = 
"max_structure_properties"
REASON_MAX_SNAPSHOT_DEPTH = 
"max_snapshot_depth"
REASON_MAX_VISITED_NODES = 
"max_visited_nodes"
REASON_CYCLE = 
"cycle"
CTX_TAG_STRING =

Type tags so values of different Ruby types never collide in the canonical key.

"s"
CTX_TAG_BOOL = 
"b"
CTX_TAG_INTEGER = 
"i"
CTX_TAG_FLOAT = 
"f"
CTX_TAG_OTHER = 
"o"
EVAL_SCALE_TARGET_FLAGS = 
2_500
EVAL_SCALE_FULL_BUCKETS_PER_FLAG = 
50
EVAL_SCALE_USERS_PER_FLAG = 
1_000
EVAL_SCALE_PER_FLAG_HEADROOM_MULTIPLIER = 
10
EVAL_SCALE_DEGRADED_BUCKETS_PER_FLAG = 
10
EVAL_SCALE_FULL_BUCKET_TARGET = 
EVAL_SCALE_TARGET_FLAGS * EVAL_SCALE_FULL_BUCKETS_PER_FLAG
EVAL_SCALE_PER_FLAG_BUCKET_TARGET = 
EVAL_SCALE_PER_FLAG_HEADROOM_MULTIPLIER * EVAL_SCALE_USERS_PER_FLAG
EVAL_SCALE_DEGRADED_BUCKET_TARGET = 
EVAL_SCALE_TARGET_FLAGS * EVAL_SCALE_DEGRADED_BUCKETS_PER_FLAG
DEFAULT_GLOBAL_CAP = 
131_072
DEFAULT_PER_FLAG_CAP = 
EVAL_SCALE_PER_FLAG_BUCKET_TARGET
DEFAULT_DEGRADED_CAP = 
32_768

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(global_cap: DEFAULT_GLOBAL_CAP, per_flag_cap: DEFAULT_PER_FLAG_CAP, degraded_cap: DEFAULT_DEGRADED_CAP) ⇒ Aggregator

Returns a new instance of Aggregator.



68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
 
# File 'lib/datadog/open_feature/flag_evaluation/aggregator.rb', line 68

def initialize(
  global_cap: DEFAULT_GLOBAL_CAP,
  per_flag_cap: DEFAULT_PER_FLAG_CAP,
  degraded_cap: DEFAULT_DEGRADED_CAP
)
  @global_cap = global_cap
  @per_flag_cap = per_flag_cap
  @degraded_cap = degraded_cap

  @mutex = Mutex.new
  # full-tier: Array key -> Hash entry
  @full = {}
  # degraded-tier: Array key -> Hash entry
  @degraded = {}
  # per-flag full-bucket count for per_flag_cap enforcement
  @per_flag_full = Hash.new(0)
  @global_count = 0
  @dropped_degraded_overflow = 0
end

Instance Attribute Details

#dropped_degraded_overflow ⇒ Object (readonly)

Returns the value of attribute dropped_degraded_overflow.



66
67
68
 
# File 'lib/datadog/open_feature/flag_evaluation/aggregator.rb', line 66

def dropped_degraded_overflow
  @dropped_degraded_overflow
end

Class Method Details

.bounded_context_snapshot(attrs, excluded_key: nil) ⇒ Object

Bound and flatten the caller's evaluation context on the evaluation thread, before enqueue, so the async queue only ever holds an already-bounded snapshot. Returns the flattened (dot-notation) context and the truncation reasons hit, which the writer surfaces on the flagevaluation.context.truncated counter.

Ruby Hash insertion order lets traversal stop at the caps without sorting or scanning the complete input. Ruby therefore truncates where Go omits, whose map iteration is randomized per call.



184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
 
# File 'lib/datadog/open_feature/flag_evaluation/aggregator.rb', line 184

def self.bounded_context_snapshot(attrs, excluded_key: nil)
  return [{}, []] unless attrs.is_a?(Hash) && !attrs.empty?

  flattened = {}
  reasons = Set.new
  seen = {attrs.object_id => true}
  walked = 0
  # Single-element array so the recursion shares one mutable counter.
  budget = [MAX_VISITED_NODES]
  attrs.each do |key, value|
    key = context_key_string(key)
    next unless key
    next if key == excluded_key

    if flattened.size >= MAX_CONTEXT_FIELDS
      reasons << REASON_MAX_CONTEXT_FIELDS
      reasons << REASON_MAX_STRUCTURE_PROPERTIES if walked >= MAX_STRUCTURE_PROPERTIES
      break
    end
    if walked >= MAX_STRUCTURE_PROPERTIES
      reasons << REASON_MAX_STRUCTURE_PROPERTIES
      break
    end
    if budget[0] <= 0
      reasons << REASON_MAX_VISITED_NODES
      break
    end
    walked += 1
    if key.length > MAX_KEY_LENGTH
      reasons << REASON_MAX_KEY_LENGTH
      next
    end

    key = key.encode(Encoding::UTF_8, invalid: :replace, undef: :replace)
    bounded_flatten(key, value, flattened, seen, 0, reasons, budget)
  end
  [flattened, reasons.to_a]
end

Instance Method Details

#canonical_context_key(attrs) ⇒ Object

Canonical context key: sorted type-tagged length-delimited encoding. Each field is: 8-byte big-endian key length + key bytes + type-tag byte + 8-byte big-endian value length + value bytes. No hash digest — the key IS the full encoding (collision-free, no FNV).



351
352
353
354
355
356
357
358
359
360
361
 
# File 'lib/datadog/open_feature/flag_evaluation/aggregator.rb', line 351

def canonical_context_key(attrs)
  return "" if attrs.nil? || attrs.empty?

  buffer = String.new("", encoding: Encoding::BINARY)
  attrs.keys.sort.each do |key|
    value = attrs[key]
    buffer << length_delimited(key.to_s)
    buffer << context_value_bytes(value)
  end
  buffer
end

#flush_and_reset ⇒ Object

Flush aggregation maps, reset state, return snapshot. The overflow count is included in the snapshot so the caller can emit it before it is reset (never reset-without-emit).



160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
 
# File 'lib/datadog/open_feature/flag_evaluation/aggregator.rb', line 160

def flush_and_reset
  @mutex.synchronize do
    full_snapshot = @full
    degraded_snapshot = @degraded
    dropped_snapshot = @dropped_degraded_overflow

    @full = {}
    @degraded = {}
    @per_flag_full = Hash.new(0)
    @global_count = 0
    @dropped_degraded_overflow = 0

    {full: full_snapshot, degraded: degraded_snapshot, dropped_degraded_overflow: dropped_snapshot}
  end
end

#record(flag_key:, variant:, allocation_key:, targeting_key:, eval_time_ms:, attrs:, error_message: nil, runtime_default: nil, observe_full_evaluation_data: false) ⇒ Object

Record one evaluation event. Thread-safe. Called from the background writer.



89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
 
# File 'lib/datadog/open_feature/flag_evaluation/aggregator.rb', line 89

def record(
  flag_key:, variant:, allocation_key:, targeting_key:, eval_time_ms:, attrs:, error_message: nil,
  runtime_default: nil, observe_full_evaluation_data: false
)
  runtime_default = variant.nil? if runtime_default.nil?
  runtime_default = !!runtime_default
  # Runtime defaults have no resolved variant or allocation in the EVP schema.
  if runtime_default
    variant = nil
    allocation_key = nil
  end
  observe_full_evaluation_data = !!observe_full_evaluation_data

  # Preserve targeting-key presence; other dimensions use empty strings.
  variant = variant.to_s
  allocation_key = allocation_key.to_s
  error_message = error_message.to_s
  targeting_key = targeting_key.to_s unless targeting_key.nil?

  context_key = observe_full_evaluation_data ? canonical_context_key(attrs) : nil
  # @type var full_key: full_key
  full_key = [
    flag_key, variant, allocation_key, runtime_default, error_message,
    targeting_key, context_key, observe_full_evaluation_data,
  ]
  evaluation_time_ms = eval_time_ms.to_i

  @mutex.synchronize do
    # --- Full tier ---
    if (entry = @full[full_key])
      observe(entry, evaluation_time_ms, observe_full_evaluation_data)
      return
    end

    per_flag_count = @per_flag_full[flag_key]
    if per_flag_count >= @per_flag_cap
      add_to_degraded(
        flag_key, variant, allocation_key, runtime_default, error_message, evaluation_time_ms,
        observe_full_evaluation_data
      )
      return
    end

    # Count the full-tier attempt before checking the global cap so per-flag overflow stays
    # active even when the global full-tier cap is already saturated.
    @per_flag_full[flag_key] = per_flag_count + 1

    if @global_count < @global_cap
      entry = new_entry(
        evaluation_time_ms,
        runtime_default: runtime_default,
        error_message: error_message,
        targeting_key: targeting_key,
        context_attrs: observe_full_evaluation_data ? attrs : nil,
        observe_full_evaluation_data: observe_full_evaluation_data
      )
      @full[full_key] = entry
      @global_count += 1
    else
      # Route to degraded tier
      add_to_degraded(
        flag_key, variant, allocation_key, runtime_default, error_message, evaluation_time_ms,
        observe_full_evaluation_data
      )
    end
  end
end