Line data Source code
1 : // Copyright The OpenTelemetry Authors
2 : // SPDX-License-Identifier: Apache-2.0
3 :
4 : import 'dart:async';
5 : import 'dart:collection';
6 :
7 : import 'package:dartastic_opentelemetry_api/dartastic_opentelemetry_api.dart';
8 : import 'package:synchronized/synchronized.dart';
9 :
10 : import '../../environment/otel_env.dart';
11 : import '../span.dart';
12 : import '../span_processor.dart';
13 : import 'span_exporter.dart';
14 :
15 : /// Configuration for the [BatchSpanProcessor].
16 : ///
17 : /// This class configures how the batch span processor behaves, including
18 : /// queue size limits, export scheduling, and batch size parameters.
19 : class BatchSpanProcessorConfig {
20 : /// Stand-in for "no limit": `OTEL_BSP_EXPORT_TIMEOUT=0` means no timeout
21 : /// per spec, represented as the max milliseconds in a 32-bit integer
22 : /// (~24.8 days). Web-safe (as microseconds it is far below JS's 2^53
23 : /// safe-integer limit, so it behaves identically on VM, dart2js, and wasm)
24 : /// and within Dart timer limits for `Future.timeout()`.
25 : static const Duration noLimit = Duration(milliseconds: 0x7FFFFFFF);
26 :
27 : /// The maximum queue size for spans. After this is reached,
28 : /// spans will be dropped.
29 : final int maxQueueSize;
30 :
31 : /// The delay between two consecutive exports.
32 : final Duration scheduleDelay;
33 :
34 : /// The maximum batch size of spans that can be exported at once.
35 : final int maxExportBatchSize;
36 :
37 : /// The amount of time to wait for an export to complete before timing out.
38 : final Duration exportTimeout;
39 :
40 : /// Creates a new configuration for a [BatchSpanProcessor].
41 : ///
42 : /// [maxQueueSize] The maximum number of spans that can be queued for export. Default is 2048.
43 : /// If this limit is reached, additional spans will be dropped.
44 : /// [scheduleDelay] The time interval between two consecutive exports. Default is 5 seconds.
45 : /// This controls how frequently batches are sent to the exporter.
46 : /// [maxExportBatchSize] The maximum number of spans to export in a single batch. Default is 512.
47 : /// This helps control resource usage during export operations.
48 : /// [exportTimeout] The maximum time to wait for an export operation to complete. Default is 30 seconds.
49 : /// After this time, export operations will be considered failed.
50 172 : const BatchSpanProcessorConfig({
51 : this.maxQueueSize = 2048,
52 : this.scheduleDelay = const Duration(milliseconds: 5000),
53 : this.maxExportBatchSize = 512,
54 : this.exportTimeout = const Duration(seconds: 30),
55 : });
56 :
57 : /// Creates a configuration by reading `OTEL_BSP_*` environment variables
58 : /// via [OTelEnv]. Falls back to standard OTel defaults if variables are
59 : /// missing or invalid.
60 : ///
61 : /// Invalid or out-of-range values emit an [OTelLog.warn] diagnostic and
62 : /// fall back to the spec default.
63 136 : factory BatchSpanProcessorConfig.fromBspEnvironmentValues(
64 : BspEnvironmentValues env) {
65 : var queueSize = env.maxQueueSize ?? 2048;
66 : var batchSize = env.maxExportBatchSize ?? 512;
67 :
68 : // --- scheduleDelay ---
69 : // Spec type: Duration. Zero is valid ("export as fast as possible").
70 : // Negative values MUST warn and fall back to default.
71 : Duration scheduleDelay;
72 : if (env.scheduleDelay != null) {
73 : final delay = env.scheduleDelay!;
74 2 : if (delay.inMilliseconds >= 0) {
75 : scheduleDelay = delay;
76 : } else {
77 1 : if (OTelLog.isWarn()) {
78 2 : OTelLog.warn('BatchSpanProcessorConfig: Negative '
79 1 : 'OTEL_BSP_SCHEDULE_DELAY (${delay.inMilliseconds} ms) is '
80 : 'invalid per spec, using default 5000 ms.');
81 : }
82 : scheduleDelay = const Duration(milliseconds: 5000);
83 : }
84 : } else {
85 : scheduleDelay = const Duration(milliseconds: 5000);
86 : }
87 :
88 : // --- exportTimeout ---
89 : // Spec type: Timeout. Zero means "no limit" — substitute a very large
90 : // duration. Negative values MUST warn and fall back to default.
91 : Duration exportTimeout;
92 : if (env.exportTimeout != null) {
93 : final timeout = env.exportTimeout!;
94 2 : if (timeout.inMilliseconds == 0) {
95 : exportTimeout = noLimit;
96 2 : } else if (timeout.inMilliseconds > 0) {
97 : exportTimeout = timeout;
98 : } else {
99 1 : if (OTelLog.isWarn()) {
100 2 : OTelLog.warn('BatchSpanProcessorConfig: Negative '
101 1 : 'OTEL_BSP_EXPORT_TIMEOUT (${timeout.inMilliseconds} ms) is '
102 : 'invalid per spec, using default 30000 ms.');
103 : }
104 : exportTimeout = const Duration(milliseconds: 30000);
105 : }
106 : } else {
107 : exportTimeout = const Duration(milliseconds: 30000);
108 : }
109 :
110 : // --- Validation Logic ---
111 136 : if (queueSize <= 0) {
112 1 : if (OTelLog.isWarn()) {
113 2 : OTelLog.warn('BatchSpanProcessorConfig: Non-positive '
114 : 'OTEL_BSP_MAX_QUEUE_SIZE ($queueSize) is invalid per spec, '
115 : 'using default 2048.');
116 : }
117 : queueSize = 2048;
118 : }
119 136 : if (batchSize <= 0) {
120 1 : if (OTelLog.isWarn()) {
121 2 : OTelLog.warn('BatchSpanProcessorConfig: Non-positive '
122 : 'OTEL_BSP_MAX_EXPORT_BATCH_SIZE ($batchSize) is invalid per '
123 : 'spec, using default 512.');
124 : }
125 : batchSize = 512;
126 : }
127 : // Spec rule: maxExportBatchSize must be less than or equal to maxQueueSize
128 136 : if (batchSize > queueSize) {
129 : batchSize = queueSize;
130 : }
131 :
132 136 : return BatchSpanProcessorConfig(
133 : maxQueueSize: queueSize,
134 : maxExportBatchSize: batchSize,
135 : scheduleDelay: scheduleDelay,
136 : exportTimeout: exportTimeout,
137 : );
138 : }
139 :
140 : /// Creates a configuration by reading `OTEL_BSP_*` environment variables
141 : /// via [OTelEnv]. Falls back to standard OTel defaults if variables are
142 : /// missing or invalid.
143 : ///
144 : /// | Environment Variable | Type | Default | Notes |
145 : /// |-----------------------------------|----------|----------|----------------------------------------------------------|
146 : /// | `OTEL_BSP_SCHEDULE_DELAY` | Duration | `5000` | Delay between exports (ms). 0 is valid (export ASAP). |
147 : /// | `OTEL_BSP_EXPORT_TIMEOUT` | Timeout | `30000` | Export timeout (ms). 0 means no limit. |
148 : /// | `OTEL_BSP_MAX_QUEUE_SIZE` | Integer | `2048` | Maximum span queue size. |
149 : /// | `OTEL_BSP_MAX_EXPORT_BATCH_SIZE` | Integer | `512` | Maximum batch size. Must be ≤ `MAX_QUEUE_SIZE`. |
150 : ///
151 : /// Invalid or out-of-range values emit an [OTelLog.warn] diagnostic and
152 : /// fall back to the spec default.
153 1 : factory BatchSpanProcessorConfig.fromEnvironment() {
154 1 : return BatchSpanProcessorConfig.fromBspEnvironmentValues(
155 1 : OTelEnv.getBspConfig());
156 : }
157 : }
158 :
159 : /// A [SpanProcessor] that batches spans before export.
160 : ///
161 : /// This processor collects finished spans in a queue and exports them in batches
162 : /// at regular intervals, improving efficiency compared to exporting each span
163 : /// individually. Spans are added to a queue when they end, and periodically sent
164 : /// to the configured exporter in batches according to the configured schedule.
165 : ///
166 : /// The batch behavior can be tuned using [BatchSpanProcessorConfig] to control
167 : /// batch size, queue limits, and export timing.
168 : class BatchSpanProcessor implements SpanProcessor {
169 : /// The exporter used to send spans to the backend
170 : final SpanExporter exporter;
171 :
172 : /// Configuration for the batch processor behavior
173 : final BatchSpanProcessorConfig _config;
174 :
175 : /// Queue of spans waiting to be exported
176 : final Queue<Span> _spanQueue = Queue<Span>();
177 :
178 : /// Whether the processor has been shut down
179 : bool _isShutdown = false;
180 :
181 : /// Timer for scheduling periodic exports
182 : Timer? _timer;
183 :
184 : /// Lock for synchronizing queue access
185 : final _lock = Lock();
186 :
187 : /// Creates a new BatchSpanProcessor with the specified exporter and configuration.
188 : ///
189 : /// The BatchSpanProcessor collects finished spans in a queue and exports them in batches
190 : /// at regular intervals. This improves efficiency compared to exporting each span individually.
191 : ///
192 : /// A timer is started when this processor is created based on the [config]'s scheduleDelay.
193 : /// The timer triggers periodic batch exports of completed spans to the configured exporter.
194 : ///
195 : /// When the maximum queue size is reached, new spans will be dropped and not exported.
196 : ///
197 : /// This processor does not modify spans on start or when their names are updated,
198 : /// it only processes spans when they end.
199 : ///
200 : /// If an error occurs during export, it will be logged but not propagated.
201 : ///
202 : /// [exporter] The SpanExporter to use for exporting batches of spans
203 : /// [config] Optional configuration for the batch processor
204 137 : BatchSpanProcessor(this.exporter, [BatchSpanProcessorConfig? config])
205 : : _config = config ?? const BatchSpanProcessorConfig() {
206 554 : _timer = Timer.periodic(_config.scheduleDelay, (_) async {
207 : try {
208 6 : await _exportBatch();
209 : // Exporter errors are caught inside _exportSingleBatch; this is a
210 : // last-resort guard so an unexpected failure can never become an
211 : // unhandled async error inside a periodic timer.
212 : // coverage:ignore-start
213 : } catch (e) {
214 : if (OTelLog.isError()) OTelLog.error('Error in batch export timer: $e');
215 : }
216 : // coverage:ignore-end
217 : });
218 : }
219 :
220 39 : @override
221 : Future<void> onEnd(Span span) async {
222 39 : if (_isShutdown) {
223 : return;
224 : }
225 :
226 : // Per the Trace SDK spec (Sampling), span exporters MUST receive
227 : // spans with the Sampled flag set and SHOULD NOT receive the ones
228 : // that do not (e.g. RECORD_ONLY spans record but are not exported).
229 117 : if (!span.spanContext.traceFlags.isSampled) {
230 2 : if (OTelLog.isDebug()) {
231 2 : OTelLog.debug(
232 6 : 'BatchSpanProcessor: Skipping enqueue - span ${span.spanContext.spanId} is not sampled',
233 : );
234 : }
235 : return;
236 : }
237 :
238 114 : return _lock.synchronized(() {
239 190 : if (_spanQueue.length >= _config.maxQueueSize) {
240 1 : if (OTelLog.isDebug()) {
241 1 : OTelLog.debug('BatchSpanProcessor queue full - dropping span');
242 : }
243 : return;
244 : }
245 76 : _spanQueue.add(span);
246 : });
247 : }
248 :
249 43 : @override
250 : Future<void> onStart(Span span, Context? parentContext) async {
251 : // Nothing to do on start
252 : }
253 :
254 1 : @override
255 : Future<void> onNameUpdate(Span span, String newName) async {
256 : // Nothing to do on name update
257 : }
258 :
259 : /// Periodic export path: called by the timer. Exports up to one batch
260 : /// of [BatchSpanProcessorConfig.maxExportBatchSize] spans and returns.
261 : /// No-op once [_isShutdown] is set.
262 6 : Future<void> _exportBatch() async {
263 6 : if (_isShutdown) {
264 : return;
265 : }
266 6 : await _exportSingleBatch();
267 : }
268 :
269 : /// Pulls up to [BatchSpanProcessorConfig.maxExportBatchSize] spans
270 : /// off the queue and hands them to the exporter. Bypasses the
271 : /// [_isShutdown] check so it remains usable from inside [shutdown]
272 : /// (which sets the flag last). Returns true if any spans were
273 : /// exported, false if the queue was empty or the exporter threw.
274 39 : Future<bool> _exportSingleBatch() async {
275 39 : final spansToExport = <Span>[];
276 :
277 117 : await _lock.synchronized(() {
278 195 : final batchSize = _spanQueue.length > _config.maxExportBatchSize
279 4 : ? _config.maxExportBatchSize
280 78 : : _spanQueue.length;
281 76 : for (var i = 0; i < batchSize; i++) {
282 74 : if (_spanQueue.isEmpty) break;
283 111 : spansToExport.add(_spanQueue.removeFirst());
284 : }
285 : });
286 :
287 39 : if (spansToExport.isEmpty) {
288 : return false;
289 : }
290 :
291 : try {
292 185 : await exporter.export(spansToExport).timeout(_config.exportTimeout);
293 : return true;
294 : } catch (e) {
295 34 : if (OTelLog.isError()) {
296 68 : OTelLog.error('Error exporting batch of spans: $e');
297 : }
298 : // Bail out — don't loop forever on a broken exporter.
299 : return false;
300 : }
301 : }
302 :
303 : /// Drains the queue completely, in batches. Used by [forceFlush] and
304 : /// [shutdown]. Stops on the first export failure (so a broken
305 : /// exporter can't wedge shutdown forever).
306 136 : Future<void> _drainQueue() async {
307 272 : while (_spanQueue.isNotEmpty) {
308 36 : final exported = await _exportSingleBatch();
309 : if (!exported) break;
310 : }
311 : }
312 :
313 130 : @override
314 : Future<void> forceFlush() async {
315 130 : if (_isShutdown) {
316 : return;
317 : }
318 130 : await _drainQueue();
319 : }
320 :
321 136 : @override
322 : Future<void> shutdown() async {
323 136 : if (_isShutdown) {
324 : return;
325 : }
326 :
327 : // Stop the periodic timer first so it can't race with our drain.
328 272 : _timer?.cancel();
329 :
330 : // Drain queued spans BEFORE setting `_isShutdown` — otherwise the
331 : // `_isShutdown` early-return inside `_exportBatch` would skip them.
332 : // `_drainQueue` calls `_exportSingleBatch` directly so it's
333 : // unaffected by the flag in any case, but the ordering keeps the
334 : // behavior obvious.
335 136 : await _drainQueue();
336 :
337 136 : _isShutdown = true;
338 :
339 : // Shutdown the exporter
340 272 : await exporter.shutdown();
341 : }
342 : }
|
