Repository navigation
Expand file tree
/
Copy pathExecutionOtelPlugin.java
More file actions
624 lines (547 loc) · 28.5 KB
/
Copy pathExecutionOtelPlugin.java
File metadata and controls
624 lines (547 loc) · 28.5 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
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
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
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
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
package software.amazon.lambda.durable.otel;
import static software.amazon.lambda.durable.otel.SpanAttributes.*;
import io.opentelemetry.api.common.AttributeKey;
import io.opentelemetry.api.trace.Span;
import io.opentelemetry.api.trace.SpanBuilder;
import io.opentelemetry.api.trace.SpanContext;
import io.opentelemetry.api.trace.SpanKind;
import io.opentelemetry.api.trace.StatusCode;
import io.opentelemetry.api.trace.TraceFlags;
import io.opentelemetry.api.trace.TraceState;
import io.opentelemetry.api.trace.Tracer;
import io.opentelemetry.context.Context;
import io.opentelemetry.context.Scope;
import io.opentelemetry.sdk.trace.SdkTracerProvider;
import io.opentelemetry.sdk.trace.SdkTracerProviderBuilder;
import java.time.Instant;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
import software.amazon.lambda.durable.plugin.DurableExecutionPlugin;
import software.amazon.lambda.durable.plugin.InvocationEndInfo;
import software.amazon.lambda.durable.plugin.InvocationInfo;
import software.amazon.lambda.durable.plugin.OperationEndInfo;
import software.amazon.lambda.durable.plugin.OperationInfo;
import software.amazon.lambda.durable.plugin.UserFunctionEndInfo;
import software.amazon.lambda.durable.plugin.UserFunctionStartInfo;
/**
* Workflow-rooted OpenTelemetry plugin for the AWS Lambda Durable Execution SDK.
*
* <p>This is the Java port of the {@code ExecutionOtelPlugin} in the Python and JavaScript SDKs. It renders the full
* durable-execution hierarchy:
*
* <ul>
* <li><b>Workflow span</b> — one <em>logical</em> root span per durable execution. Its span ID is derived
* deterministically from the execution ARN, so every invocation of the same execution produces the same ID. It is
* ended (and therefore exported) exactly once, on the terminal invocation.
* <li><b>Invocation span</b> — one per Lambda invocation, a child of the ambient Lambda span when available and a
* root otherwise. Created and ended every invocation.
* <li><b>Operation span</b> — parented to its parent operation span (or the Workflow span) and carrying a
* <em>link</em> to the current Invocation span for correlation. Deterministic ID keyed by operation ID, so a
* suspended-then-resumed operation stitches into a single logical span across invocations.
* <li><b>Attempt span</b> — one per user-function execution (step attempt, child-context run), child of the operation
* span, linked to the current Invocation span.
* </ul>
*
* <p>Contrast with {@link InvocationOtelPlugin} (the invocation-rooted variant, equivalent to the reference
* {@code InvocationOtelPlugin}): there operations hang off the per-invocation span and link to the Workflow span. Here
* operations hang off the independent Workflow root span, while each operation/attempt links to the invocation that ran
* it. Both plugins share {@link DeterministicIdGenerator}, {@link ContextExtractor}, {@link SpanAttributes}, and
* {@link MdcSpanEnricher}.
*
* <p>The Workflow trace ID is derived from the execution start time and ARN, and is independent of the ambient
* Lambda/X-Ray trace. Invocation spans inherit the active ambient context, or extracted upstream context as a fallback.
* When using {@link #ExecutionOtelPlugin()}, the plugin resolves the global provider at invocation start. If the
* OpenTelemetry Java agent is not initialized yet, telemetry is disabled for that entire invocation and provider
* resolution is retried on the next invocation.
*
* <p>Status mapping (parity with the Python/JS references):
*
* <ul>
* <li>Invocation span: {@code SUCCEEDED}/{@code PENDING} → {@link StatusCode#OK}; {@code FAILED} →
* {@link StatusCode#ERROR}; {@code RETRYING} → {@link StatusCode#UNSET}. {@code RETRYING} is left {@code UNSET}
* because the plugin interface does not expose whether the invocation/Workflow was STOPPED or TIMED_OUT versus
* retried for a transient error, so an ERROR status cannot be asserted reliably.
* <li>Workflow span (terminal only): {@code SUCCEEDED} → {@link StatusCode#OK}; {@code FAILED} →
* {@link StatusCode#ERROR}. Non-terminal statuses ({@code PENDING}/{@code RETRYING}) never end the Workflow span,
* so it is not exported this invocation (effectively {@link StatusCode#UNSET}).
* </ul>
*
* <p>Thread-safe: uses {@link ConcurrentHashMap} for span/scope storage since the SDK runs user code on multiple
* threads.
*/
public class ExecutionOtelPlugin implements DurableExecutionPlugin {
private static final Logger logger = LoggerFactory.getLogger(ExecutionOtelPlugin.class);
private volatile SdkTracerProvider sdkTracerProvider;
private volatile Tracer tracer;
private final DeterministicIdGenerator idGenerator;
private final ContextExtractor contextExtractor;
private final boolean enableMdc;
private final String workflowSpanName;
private final String instrumentationName;
// Per-invocation state
private volatile boolean tracingEnabled;
private volatile Span workflowSpan;
private volatile Span invocationSpan;
private volatile String durableExecutionArn;
private volatile String workflowTraceId;
// Thread-safe storage for operation spans (keyed by operationId) — open spans that need ending
private final ConcurrentHashMap<String, Span> operationSpans = new ConcurrentHashMap<>();
// Thread-safe storage for attempt spans/scopes (keyed by operationId + "-" + attempt)
private final ConcurrentHashMap<String, Span> attemptSpans = new ConcurrentHashMap<>();
private final ConcurrentHashMap<String, Scope> attemptScopes = new ConcurrentHashMap<>();
// Store operation span contexts for parent resolution (keyed by operationId)
private final ConcurrentHashMap<String, SpanContext> operationContexts = new ConcurrentHashMap<>();
/**
* Creates a Workflow-rooted OTel plugin with default settings: X-Ray context extraction, MDC enabled, root span
* named {@code "Workflow"}.
*
* <p>Uses the provided tracer provider builder. For ADOT Java agent usage, prefer {@link #ExecutionOtelPlugin()}
* with the plugin jar configured through {@code OTEL_JAVAAGENT_EXTENSIONS}.
*
* @param tracerProviderBuilder the tracer provider builder (its ID generator will be wrapped)
*/
public ExecutionOtelPlugin(SdkTracerProviderBuilder tracerProviderBuilder) {
this(tracerProviderBuilder, OtelPluginConfig.defaults());
}
/**
* Creates a Workflow-rooted OTel plugin with default settings: X-Ray context extraction and MDC enabled.
*
* <p>Resolves {@code GlobalOpenTelemetry} at invocation start. If the ADOT Java agent has not initialized it yet,
* telemetry is disabled for that invocation and resolution is retried on the next invocation.
*/
public ExecutionOtelPlugin() {
this(OtelPluginConfig.defaults());
}
/**
* Creates a Workflow-rooted OTel plugin from the given tracer provider builder and configuration.
*
* <p>Customers configure exporters and span processors on the builder; all other tunables (context extractor, MDC
* toggle, Workflow span name, instrumentation scope name) come from {@link OtelPluginConfig}. Use
* {@link OtelPluginConfig#builder()} for readable, named configuration:
*
* <pre>{@code
* var plugin = new ExecutionOtelPlugin(
* SdkTracerProvider.builder().addSpanProcessor(SimpleSpanProcessor.create(exporter)),
* OtelPluginConfig.builder().enableMdc(false).workflowSpanName("Workflow").build());
* }</pre>
*
* @param tracerProviderBuilder the tracer provider builder (its ID generator will be wrapped)
* @param config the plugin configuration
*/
public ExecutionOtelPlugin(SdkTracerProviderBuilder tracerProviderBuilder, OtelPluginConfig config) {
this.idGenerator = DeterministicIdGenerator.installOn(tracerProviderBuilder);
this.sdkTracerProvider = tracerProviderBuilder.build();
this.tracer = sdkTracerProvider.get(config.instrumentationName());
this.contextExtractor = config.contextExtractor();
this.enableMdc = config.enableMdc();
this.workflowSpanName = config.workflowSpanName();
this.instrumentationName = config.instrumentationName();
}
/**
* Creates a Workflow-rooted OTel plugin from configuration alone (no caller-supplied tracer provider builder).
*
* <p>The config-only constructor uses the ADOT/global provider. Supply a {@code SdkTracerProviderBuilder} via the
* two-arg constructor for an application-owned provider.
*
* @param config the plugin configuration
*/
public ExecutionOtelPlugin(OtelPluginConfig config) {
this.contextExtractor = config.contextExtractor();
this.enableMdc = config.enableMdc();
this.workflowSpanName = config.workflowSpanName();
this.instrumentationName = config.instrumentationName();
this.idGenerator = OtelPluginSupport.createDefaultIdGenerator();
}
// ─── Invocation hooks ────────────────────────────────────────────────
@Override
public void onInvocationStart(InvocationInfo info) {
tracingEnabled = false;
if (!bindTracer()) {
return;
}
this.durableExecutionArn = info.durableExecutionArn();
// Prefer the active Java-agent span, then fall back to explicitly extracted upstream context.
var invocationParent = extractCurrentSpanContext();
if (invocationParent == null) {
invocationParent = contextExtractor.extract();
}
// Workflow root span — deterministic span ID from the ARN, no parent. Recreated every invocation with the
// same ID so it is exported once as a single logical span (on the terminal invocation only). Its start time
// is the execution start time from the backend.
var workflowSpanBuilder = tracer.spanBuilder(workflowSpanName)
.setSpanKind(SpanKind.INTERNAL)
.setNoParent()
.setAttribute(DURABLE_EXECUTION_ARN, info.durableExecutionArn())
.setStartTimestamp(info.executionStartTime());
workflowTraceId =
idGenerator.generateTraceIdForExecution(info.durableExecutionArn(), info.executionStartTime());
var workflowSpanId = idGenerator.generateWorkflowSpanId(info.durableExecutionArn());
workflowSpan = idGenerator.startSpan(workflowSpanBuilder, workflowTraceId, workflowSpanId);
Context parentContext;
if (invocationParent != null && invocationParent.parentSpanId() != null) {
var parentSpanContext = SpanContext.createFromRemoteParent(
invocationParent.traceId(),
invocationParent.parentSpanId(),
TraceFlags.getSampled(),
TraceState.getDefault());
parentContext = Context.root().with(Span.wrap(parentSpanContext));
} else {
parentContext = Context.root();
}
// Invocation span — child of the ambient Lambda span when available, otherwise a root.
var spanBuilder = tracer.spanBuilder("Invocation")
.setSpanKind(SpanKind.INTERNAL)
.setParent(parentContext)
.setAttribute(DURABLE_EXECUTION_ARN, info.durableExecutionArn())
.setAttribute(DURABLE_FIRST_INVOCATION, info.isFirstInvocation());
if (info.requestId() != null) {
spanBuilder.setAttribute(AttributeKey.stringKey("faas.invocation_id"), info.requestId());
}
invocationSpan = spanBuilder.startSpan();
// Inject MDC on the handler thread so handler-level logs (between steps) have trace context.
if (enableMdc) {
MDC.put(
MdcSpanEnricher.MDC_TRACE_ID,
invocationSpan.getSpanContext().getTraceId());
}
tracingEnabled = true;
}
@Override
public void onInvocationEnd(InvocationEndInfo info) {
if (!tracingEnabled) {
return;
}
tracingEnabled = false;
// Clear invocation-level MDC
if (enableMdc) {
MdcSpanEnricher.clear();
}
// Reset per-invocation operation state WITHOUT ending open operation spans. Matching the JS/Python
// ExecutionOtelPlugin, an operation span is only ended in onOperationEnd. An operation still open when the
// invocation suspends is left un-exported here and is re-materialized once (with its deterministic span ID,
// plus a link to the invocation that completes it) when onOperationEnd fires in a later invocation.
operationSpans.clear();
operationContexts.clear();
// Defensively close any lingering attempt scopes so OTel context is not leaked on worker threads (normally
// every onUserFunctionStart is paired with onUserFunctionEnd within the invocation). The attempt spans
// themselves are left un-ended rather than force-ended, consistent with not ending open spans here.
for (var scope : attemptScopes.values()) {
scope.close();
}
attemptScopes.clear();
attemptSpans.clear();
// End the invocation span every invocation.
if (invocationSpan != null) {
invocationSpan.setAttribute(
DURABLE_INVOCATION_STATUS, info.invocationStatus().name());
applyInvocationStatus(invocationSpan, info);
invocationSpan.end();
invocationSpan = null;
}
// End the Workflow span only on a terminal status, so it is exported exactly once per execution.
if (workflowSpan != null) {
if (isTerminal(info)) {
workflowSpan.setAttribute(
DURABLE_EXECUTION_STATUS, info.invocationStatus().name());
switch (info.invocationStatus()) {
case FAILED -> {
var message = info.executionError() != null
? info.executionError().getMessage()
: null;
workflowSpan.setStatus(StatusCode.ERROR, message);
if (info.executionError() != null) {
workflowSpan.recordException(info.executionError());
}
}
default -> workflowSpan.setStatus(StatusCode.OK); // SUCCEEDED
}
workflowSpan.end();
}
// Non-terminal (PENDING/RETRYING): leave the Workflow span un-ended (not exported this invocation).
workflowSpan = null;
}
// Flush spans before Lambda freezes
if (sdkTracerProvider != null) {
var flushResult = sdkTracerProvider.forceFlush().join(5, TimeUnit.SECONDS);
if (!flushResult.isSuccess()) {
logger.warn("OTel span flush failed or timed out — some spans may be lost");
}
}
}
// ─── Operation hooks ─────────────────────────────────────────────────
@Override
public void onOperationStart(OperationInfo info) {
if (!tracingEnabled) return;
if (info.id() == null) return;
var parentContext = resolveParentContext(info.parentId());
var spanBuilder = tracer.spanBuilder(spanName(info.type(), info.subType(), info.name()))
.setParent(parentContext)
.setAttribute(DURABLE_EXECUTION_ARN, durableExecutionArn)
.setAttribute(DURABLE_OPERATION_ID, info.id())
.setAttribute(DURABLE_OPERATION_TYPE, info.type());
addInvocationLink(spanBuilder);
if (info.startTimestamp() != null) {
spanBuilder.setStartTimestamp(info.startTimestamp());
}
if (info.name() != null) {
spanBuilder.setAttribute(DURABLE_OPERATION_NAME, info.name());
}
if (info.subType() != null) {
spanBuilder.setAttribute(DURABLE_OPERATION_SUBTYPE, info.subType());
}
var operationSpanId = idGenerator.generateSpanIdForOperation(durableExecutionArn, info.id());
var span = idGenerator.startSpan(spanBuilder, null, operationSpanId);
// Store the open span — will be ended in onOperationEnd or onInvocationEnd
operationSpans.put(info.id(), span);
operationContexts.put(info.id(), span.getSpanContext());
}
@Override
public void onOperationEnd(OperationEndInfo info) {
if (!tracingEnabled) return;
if (info.id() == null) return;
var span = operationSpans.remove(info.id());
if (span != null) {
// Operation was started in this invocation — end normally
if (info.status() != null) {
span.setAttribute(DURABLE_OPERATION_STATUS, info.status());
}
// Total attempts for retriable operations (STEP, WAIT_FOR_CONDITION) — emitted only at end.
if (info.attempt() != null) {
span.setAttribute(DURABLE_ATTEMPT_NUMBER, info.attempt().longValue());
}
if (info.error() != null) {
span.setStatus(StatusCode.ERROR, info.error().getMessage());
span.recordException(info.error());
} else if ("SUCCEEDED".equals(info.status()) || info.status() == null) {
// Only stamp OK on genuine success. onOperationEnd fires for every terminal status, and
// extractErrorFromOperation returns null for CANCELLED (always) and for FAILED/TIMED_OUT/STOPPED
// with no attached error object — those carry a non-null, non-SUCCEEDED status and must stay UNSET.
// A null status is a successful statusless virtual (FLAT CONTEXT) operation, which is OK.
span.setStatus(StatusCode.OK);
}
endSpan(span, info.endTimestamp());
} else {
// Operation completed between invocations: its onOperationStart ran in a prior invocation, whose
// in-memory span was dropped un-exported at that invocation's end. Emit the operation's single span
// now, using its deterministic span ID (stable across the execution), plus a link to the invocation
// that completed it.
operationContexts.remove(info.id());
var parentContext = resolveParentContext(info.parentId());
var spanBuilder = tracer.spanBuilder(spanName(info.type(), info.subType(), info.name()))
.setParent(parentContext)
.setAttribute(DURABLE_EXECUTION_ARN, durableExecutionArn)
.setAttribute(DURABLE_OPERATION_ID, info.id())
.setAttribute(DURABLE_OPERATION_TYPE, info.type());
addInvocationLink(spanBuilder);
if (info.startTimestamp() != null) {
spanBuilder.setStartTimestamp(info.startTimestamp());
}
if (info.name() != null) {
spanBuilder.setAttribute(DURABLE_OPERATION_NAME, info.name());
}
if (info.subType() != null) {
spanBuilder.setAttribute(DURABLE_OPERATION_SUBTYPE, info.subType());
}
var operationSpanId = idGenerator.generateSpanIdForOperation(durableExecutionArn, info.id());
var continuationSpan = idGenerator.startSpan(spanBuilder, null, operationSpanId);
if (info.status() != null) {
continuationSpan.setAttribute(DURABLE_OPERATION_STATUS, info.status());
}
// Total attempts for retriable operations (STEP, WAIT_FOR_CONDITION) — emitted only at end.
if (info.attempt() != null) {
continuationSpan.setAttribute(
DURABLE_ATTEMPT_NUMBER, info.attempt().longValue());
}
if (info.error() != null) {
continuationSpan.setStatus(StatusCode.ERROR, info.error().getMessage());
continuationSpan.recordException(info.error());
} else if ("SUCCEEDED".equals(info.status()) || info.status() == null) {
// See onOperationEnd (this-invocation branch): only genuine success (or a successful statusless
// virtual operation) is OK; error-less non-success statuses stay UNSET.
continuationSpan.setStatus(StatusCode.OK);
}
endSpan(continuationSpan, info.endTimestamp());
}
}
// ─── User function hooks ─────────────────────────────────────────────
@Override
public void onUserFunctionStart(UserFunctionStartInfo info) {
if (!tracingEnabled) return;
// Skip attempt spans for CONTEXT operations — they are a scoping construct, not a retriable unit of work. Still
// make the operation span current so auto-instrumented calls become children.
if ("CONTEXT".equals(info.type())) {
var operationSpan = operationSpans.get(info.id());
if (operationSpan != null) {
var scope = operationSpan.makeCurrent();
var key = attemptKey(info.id(), info.attempt());
attemptScopes.put(key, scope);
}
if (enableMdc) {
MdcSpanEnricher.inject();
}
return;
}
var key = attemptKey(info.id(), info.attempt());
// Parent the attempt span to its operation span.
var parentContext = resolveParentContext(info.id());
var spanBuilder = tracer.spanBuilder(attemptSpanName(info.type(), info.subType(), info.name(), info.attempt()))
.setParent(parentContext)
.setStartTimestamp(info.startTimestamp() != null ? info.startTimestamp() : Instant.now());
addInvocationLink(spanBuilder);
spanBuilder.setAttribute(DURABLE_EXECUTION_ARN, durableExecutionArn);
spanBuilder.setAttribute(DURABLE_OPERATION_ID, info.id());
if (info.type() != null) {
spanBuilder.setAttribute(DURABLE_OPERATION_TYPE, info.type());
}
if (info.name() != null) {
spanBuilder.setAttribute(DURABLE_OPERATION_NAME, info.name());
}
if (info.subType() != null) {
spanBuilder.setAttribute(DURABLE_OPERATION_SUBTYPE, info.subType());
}
if (info.attempt() != null) {
spanBuilder.setAttribute(DURABLE_ATTEMPT_NUMBER, info.attempt().longValue());
}
var span = spanBuilder.startSpan();
attemptSpans.put(key, span);
// Make span current on this thread so auto-instrumented calls become children
var scope = span.makeCurrent();
attemptScopes.put(key, scope);
if (enableMdc) {
MdcSpanEnricher.inject();
}
}
@Override
public void onUserFunctionEnd(UserFunctionEndInfo info) {
if (!tracingEnabled) return;
var key = attemptKey(info.id(), info.attempt());
// Close scope first (must happen on same thread as makeCurrent)
var scope = attemptScopes.remove(key);
if (scope != null) {
scope.close();
}
// CONTEXT operations don't have attempt spans — scope cleanup is all we need
if ("CONTEXT".equals(info.type())) {
return;
}
var span = attemptSpans.remove(key);
if (span == null) return;
span.setAttribute(DURABLE_ATTEMPT_OUTCOME, info.outcome().name());
switch (info.outcome()) {
case SUCCEEDED -> span.setStatus(StatusCode.OK);
case FAILED -> {
if (info.error() != null) {
span.setStatus(StatusCode.ERROR, info.error().getMessage());
span.recordException(info.error());
}
}
case INCOMPLETE -> {
// An incomplete user function is expected when the durable execution suspends.
}
}
endSpan(span, info.endTimestamp());
}
// ─── Helpers ─────────────────────────────────────────────────────────
private boolean bindTracer() {
if (tracer != null) {
return true;
}
synchronized (this) {
if (tracer != null) {
return true;
}
var setup = OtelPluginSupport.tryResolveGlobalProvider(instrumentationName, "ExecutionOtelPlugin");
if (setup == null) {
return false;
}
sdkTracerProvider = setup.sdkTracerProvider();
tracer = setup.tracer();
return true;
}
}
private void applyInvocationStatus(Span span, InvocationEndInfo info) {
// Invocation span status mapping:
// SUCCEEDED, PENDING -> OK
// FAILED -> ERROR (records the execution error when present)
// RETRYING -> UNSET (left unset)
// RETRYING is left UNSET on purpose: the plugin interface does not expose whether the invocation/workflow
// was STOPPED or TIMED_OUT versus retried for a transient error, so an ERROR status cannot be asserted
// reliably for a retrying invocation.
switch (info.invocationStatus()) {
case SUCCEEDED, PENDING -> span.setStatus(StatusCode.OK);
case FAILED -> {
var message =
info.executionError() != null ? info.executionError().getMessage() : null;
span.setStatus(StatusCode.ERROR, message);
if (info.executionError() != null) {
span.recordException(info.executionError());
}
}
case RETRYING -> {
// UNSET — see note above.
}
}
}
private static boolean isTerminal(InvocationEndInfo info) {
return switch (info.invocationStatus()) {
case SUCCEEDED, FAILED -> true;
case PENDING, RETRYING -> false;
};
}
/** Adds a link to the current invocation span, if one exists, for correlation. */
private void addInvocationLink(SpanBuilder spanBuilder) {
var currentInvocationSpan = invocationSpan;
if (currentInvocationSpan != null) {
spanBuilder.addLink(currentInvocationSpan.getSpanContext());
}
}
private Context resolveParentContext(String parentId) {
if (parentId != null) {
var parentSpanContext = operationContexts.get(parentId);
if (parentSpanContext != null) {
return Context.current().with(Span.wrap(parentSpanContext));
}
// Parent operation from a prior invocation — create a non-recording placeholder with its deterministic ID.
var deterministicParentSpanId = idGenerator.generateSpanIdForOperation(durableExecutionArn, parentId);
var placeholderContext = SpanContext.create(
workflowTraceId, deterministicParentSpanId, TraceFlags.getSampled(), TraceState.getDefault());
return Context.current().with(Span.wrap(placeholderContext));
}
// No parent operation — hang off the Workflow root span.
if (workflowSpan != null) {
return Context.current().with(workflowSpan);
}
return Context.current();
}
private static void endSpan(Span span, Instant endTimestamp) {
if (endTimestamp != null) {
span.end(endTimestamp);
} else {
span.end();
}
}
private static String spanName(String type, String subType, String name) {
if (name != null) {
return name;
}
return subType != null ? subType.toLowerCase() : type.toLowerCase();
}
private static String attemptSpanName(String type, String subType, String name, Integer attempt) {
var base = spanName(type, subType, name);
if (attempt != null) {
return base + " attempt " + attempt;
}
return base;
}
private static String attemptKey(String operationId, Integer attempt) {
return operationId + "-" + (attempt != null ? attempt : "ctx");
}
private static ExtractedContext extractCurrentSpanContext() {
return OtelPluginSupport.extractCurrentSpanContext();
}
}