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 '../span.dart';
11 : import '../span_processor.dart';
12 : import 'span_exporter.dart';
13 :
14 : /// Configuration for the [BatchSpanProcessor].
15 : ///
16 : /// This class configures how the batch span processor behaves, including
17 : /// queue size limits, export scheduling, and batch size parameters.
18 : class BatchSpanProcessorConfig {
19 : /// The maximum queue size for spans. After this is reached,
20 : /// spans will be dropped.
21 : final int maxQueueSize;
22 :
23 : /// The delay between two consecutive exports.
24 : final Duration scheduleDelay;
25 :
26 : /// The maximum batch size of spans that can be exported at once.
27 : final int maxExportBatchSize;
28 :
29 : /// The amount of time to wait for an export to complete before timing out.
30 : final Duration exportTimeout;
31 :
32 : /// Creates a new configuration for a [BatchSpanProcessor].
33 : ///
34 : /// [maxQueueSize] The maximum number of spans that can be queued for export. Default is 2048.
35 : /// If this limit is reached, additional spans will be dropped.
36 : /// [scheduleDelay] The time interval between two consecutive exports. Default is 5 seconds.
37 : /// This controls how frequently batches are sent to the exporter.
38 : /// [maxExportBatchSize] The maximum number of spans to export in a single batch. Default is 512.
39 : /// This helps control resource usage during export operations.
40 : /// [exportTimeout] The maximum time to wait for an export operation to complete. Default is 30 seconds.
41 : /// After this time, export operations will be considered failed.
42 141 : const BatchSpanProcessorConfig({
43 : this.maxQueueSize = 2048,
44 : this.scheduleDelay = const Duration(milliseconds: 5000),
45 : this.maxExportBatchSize = 512,
46 : this.exportTimeout = const Duration(seconds: 30),
47 : });
48 : }
49 :
50 : /// A [SpanProcessor] that batches spans before export.
51 : ///
52 : /// This processor collects finished spans in a queue and exports them in batches
53 : /// at regular intervals, improving efficiency compared to exporting each span
54 : /// individually. Spans are added to a queue when they end, and periodically sent
55 : /// to the configured exporter in batches according to the configured schedule.
56 : ///
57 : /// The batch behavior can be tuned using [BatchSpanProcessorConfig] to control
58 : /// batch size, queue limits, and export timing.
59 : class BatchSpanProcessor implements SpanProcessor {
60 : /// The exporter used to send spans to the backend
61 : final SpanExporter exporter;
62 :
63 : /// Configuration for the batch processor behavior
64 : final BatchSpanProcessorConfig _config;
65 :
66 : /// Queue of spans waiting to be exported
67 : final Queue<Span> _spanQueue = Queue<Span>();
68 :
69 : /// Whether the processor has been shut down
70 : bool _isShutdown = false;
71 :
72 : /// Timer for scheduling periodic exports
73 : Timer? _timer;
74 :
75 : /// Lock for synchronizing queue access
76 : final _lock = Lock();
77 :
78 : /// Creates a new BatchSpanProcessor with the specified exporter and configuration.
79 : ///
80 : /// The BatchSpanProcessor collects finished spans in a queue and exports them in batches
81 : /// at regular intervals. This improves efficiency compared to exporting each span individually.
82 : ///
83 : /// A timer is started when this processor is created based on the [config]'s scheduleDelay.
84 : /// The timer triggers periodic batch exports of completed spans to the configured exporter.
85 : ///
86 : /// When the maximum queue size is reached, new spans will be dropped and not exported.
87 : ///
88 : /// This processor does not modify spans on start or when their names are updated,
89 : /// it only processes spans when they end.
90 : ///
91 : /// If an error occurs during export, it will be logged but not propagated.
92 : ///
93 : /// [exporter] The SpanExporter to use for exporting batches of spans
94 : /// [config] Optional configuration for the batch processor
95 121 : BatchSpanProcessor(this.exporter, [BatchSpanProcessorConfig? config])
96 : : _config = config ?? const BatchSpanProcessorConfig() {
97 491 : _timer = Timer.periodic(_config.scheduleDelay, (_) async {
98 : try {
99 7 : await _exportBatch();
100 : } catch (e) {
101 0 : if (OTelLog.isError()) OTelLog.error('Error in batch export timer: $e');
102 : }
103 : });
104 : }
105 :
106 32 : @override
107 : Future<void> onEnd(Span span) async {
108 32 : if (_isShutdown) {
109 : return;
110 : }
111 :
112 96 : return _lock.synchronized(() {
113 160 : if (_spanQueue.length >= _config.maxQueueSize) {
114 1 : if (OTelLog.isDebug()) {
115 1 : OTelLog.debug('BatchSpanProcessor queue full - dropping span');
116 : }
117 : return;
118 : }
119 64 : _spanQueue.add(span);
120 : });
121 : }
122 :
123 36 : @override
124 : Future<void> onStart(Span span, Context? parentContext) async {
125 : // Nothing to do on start
126 : }
127 :
128 1 : @override
129 : Future<void> onNameUpdate(Span span, String newName) async {
130 : // Nothing to do on name update
131 : }
132 :
133 : /// Periodic export path: called by the timer. Exports up to one batch
134 : /// of [BatchSpanProcessorConfig.maxExportBatchSize] spans and returns.
135 : /// No-op once [_isShutdown] is set.
136 7 : Future<void> _exportBatch() async {
137 7 : if (_isShutdown) {
138 : return;
139 : }
140 7 : await _exportSingleBatch();
141 : }
142 :
143 : /// Pulls up to [BatchSpanProcessorConfig.maxExportBatchSize] spans
144 : /// off the queue and hands them to the exporter. Bypasses the
145 : /// [_isShutdown] check so it remains usable from inside [shutdown]
146 : /// (which sets the flag last). Returns true if any spans were
147 : /// exported, false if the queue was empty or the exporter threw.
148 36 : Future<bool> _exportSingleBatch() async {
149 36 : final spansToExport = <Span>[];
150 :
151 108 : await _lock.synchronized(() {
152 180 : final batchSize = _spanQueue.length > _config.maxExportBatchSize
153 4 : ? _config.maxExportBatchSize
154 72 : : _spanQueue.length;
155 68 : for (var i = 0; i < batchSize; i++) {
156 64 : if (_spanQueue.isEmpty) break;
157 96 : spansToExport.add(_spanQueue.removeFirst());
158 : }
159 : });
160 :
161 36 : if (spansToExport.isEmpty) {
162 : return false;
163 : }
164 :
165 : try {
166 64 : await exporter.export(spansToExport);
167 : return true;
168 : } catch (e) {
169 31 : if (OTelLog.isError()) {
170 62 : OTelLog.error('Error exporting batch of spans: $e');
171 : }
172 : // Bail out — don't loop forever on a broken exporter.
173 : return false;
174 : }
175 : }
176 :
177 : /// Drains the queue completely, in batches. Used by [forceFlush] and
178 : /// [shutdown]. Stops on the first export failure (so a broken
179 : /// exporter can't wedge shutdown forever).
180 120 : Future<void> _drainQueue() async {
181 240 : while (_spanQueue.isNotEmpty) {
182 31 : final exported = await _exportSingleBatch();
183 : if (!exported) break;
184 : }
185 : }
186 :
187 115 : @override
188 : Future<void> forceFlush() async {
189 115 : if (_isShutdown) {
190 : return;
191 : }
192 115 : await _drainQueue();
193 : }
194 :
195 120 : @override
196 : Future<void> shutdown() async {
197 120 : if (_isShutdown) {
198 : return;
199 : }
200 :
201 : // Stop the periodic timer first so it can't race with our drain.
202 240 : _timer?.cancel();
203 :
204 : // Drain queued spans BEFORE setting `_isShutdown` — otherwise the
205 : // `_isShutdown` early-return inside `_exportBatch` would skip them.
206 : // `_drainQueue` calls `_exportSingleBatch` directly so it's
207 : // unaffected by the flag in any case, but the ordering keeps the
208 : // behavior obvious.
209 120 : await _drainQueue();
210 :
211 120 : _isShutdown = true;
212 :
213 : // Shutdown the exporter
214 240 : await exporter.shutdown();
215 : }
216 : }
|
