Skip to content

Commit 2156317

Browse files
AngerM-DDAngerM-DD
committed
port over changeset from our fork
new files undo defaultContextProp changes remove bad calls to method no longer in this pr remove health check from this PR remove default adding of otel Add Health check API to worker and service interface (#617)
1 parent dca853a commit 2156317

13 files changed

+838
-256
lines changed

‎build.gradle

+1
Original file line numberDiff line numberDiff line change
@@ -63,6 +63,7 @@ dependencies {
6363
compile group: 'io.micrometer', name: 'micrometer-core', version: '1.1.2'
6464
compile group: 'javax.annotation', name: 'javax.annotation-api', version: '1.3.2'
6565
compile group: 'com.auth0', name: 'java-jwt', version:'3.10.2'
66+
compile group: 'io.opentelemetry', name: 'opentelemetry-sdk', version: '1.1.0'
6667

6768
testCompile group: 'junit', name: 'junit', version: '4.12'
6869
testCompile group: 'com.googlecode.junit-toolbox', name: 'junit-toolbox', version: '2.4'

‎src/main/java/com/uber/cadence/context/ContextPropagator.java

+30
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,9 @@
2323
* Context Propagators are used to propagate information from workflow to activity, workflow to
2424
* child workflow, and workflow to child thread (using {@link com.uber.cadence.workflow.Async}).
2525
*
26+
* <p>It is important to note that all threads share one ContextPropagator instance, so your
27+
* implementation <b>must</b> be thread-safe and store any state in ThreadLocal variables.
28+
*
2629
* <p>A sample <code>ContextPropagator</code> that copies all {@link org.slf4j.MDC} entries starting
2730
* with a given prefix along the code path looks like this:
2831
*
@@ -136,4 +139,31 @@ public interface ContextPropagator {
136139

137140
/** Sets the current context */
138141
void setCurrentContext(Object context);
142+
143+
/**
144+
* This is a lifecycle method, called after the context has been propagated to the
145+
* workflow/activity thread but the workflow/activity has not yet started.
146+
*/
147+
default void setUp() {
148+
// No-op
149+
}
150+
151+
/**
152+
* This is a lifecycle method, called after the workflow/activity has completed. If the method
153+
* finished without exception, {@code successful} will be true. Otherwise, it will be false and
154+
* {@link #onError(Throwable)} will have already been called.
155+
*/
156+
default void finish() {
157+
// No-op
158+
}
159+
160+
/**
161+
* This is a lifecycle method, called when the workflow/activity finishes by throwing an unhandled
162+
* exception. {@link #finish()} is called after this method.
163+
*
164+
* @param t The unhandled exception that caused the workflow/activity to terminate
165+
*/
166+
default void onError(Throwable t) {
167+
// No-op
168+
}
139169
}

‎src/main/java/com/uber/cadence/context/OpenTelemetryContextPropagator.java

+144
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,144 @@
1+
/*
2+
* Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
3+
*
4+
* Modifications copyright (C) 2017 Uber Technologies, Inc.
5+
*
6+
* Licensed under the Apache License, Version 2.0 (the "License"). You may not
7+
* use this file except in compliance with the License. A copy of the License is
8+
* located at
9+
*
10+
* http://aws.amazon.com/apache2.0
11+
*
12+
* or in the "license" file accompanying this file. This file is distributed on
13+
* an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
14+
* express or implied. See the License for the specific language governing
15+
* permissions and limitations under the License.
16+
*/
17+
18+
package com.uber.cadence.context;
19+
20+
import io.opentelemetry.api.GlobalOpenTelemetry;
21+
import io.opentelemetry.api.baggage.Baggage;
22+
import io.opentelemetry.api.baggage.propagation.W3CBaggagePropagator;
23+
import io.opentelemetry.api.trace.Span;
24+
import io.opentelemetry.api.trace.SpanKind;
25+
import io.opentelemetry.api.trace.propagation.W3CTraceContextPropagator;
26+
import io.opentelemetry.context.Context;
27+
import io.opentelemetry.context.Scope;
28+
import io.opentelemetry.context.propagation.TextMapGetter;
29+
import io.opentelemetry.context.propagation.TextMapPropagator;
30+
import io.opentelemetry.context.propagation.TextMapSetter;
31+
import java.nio.charset.Charset;
32+
import java.util.HashMap;
33+
import java.util.Map;
34+
import javax.annotation.Nullable;
35+
import org.slf4j.MDC;
36+
37+
public class OpenTelemetryContextPropagator implements ContextPropagator {
38+
39+
private static final TextMapPropagator w3cTraceContextPropagator =
40+
W3CTraceContextPropagator.getInstance();
41+
private static final TextMapPropagator w3cBaggagePropagator = W3CBaggagePropagator.getInstance();
42+
private static ThreadLocal<Scope> currentContextOtelScope = new ThreadLocal<>();
43+
private static ThreadLocal<Span> currentOtelSpan = new ThreadLocal<>();
44+
private static ThreadLocal<Scope> currentOtelScope = new ThreadLocal<>();
45+
private static ThreadLocal<Iterable<String>> otelKeySet = new ThreadLocal<>();
46+
private static final TextMapSetter<Map<String, String>> setter = Map::put;
47+
private static final TextMapGetter<Map<String, String>> getter =
48+
new TextMapGetter<Map<String, String>>() {
49+
@Override
50+
public Iterable<String> keys(Map<String, String> carrier) {
51+
return otelKeySet.get();
52+
}
53+
54+
@Nullable
55+
@Override
56+
public String get(Map<String, String> carrier, String key) {
57+
return MDC.get(key);
58+
}
59+
};
60+
61+
@Override
62+
public String getName() {
63+
return "OpenTelemetry";
64+
}
65+
66+
@Override
67+
public Map<String, byte[]> serializeContext(Object context) {
68+
Map<String, byte[]> serializedContext = new HashMap<>();
69+
Map<String, String> contextMap = (Map<String, String>) context;
70+
if (contextMap != null) {
71+
for (Map.Entry<String, String> entry : contextMap.entrySet()) {
72+
serializedContext.put(entry.getKey(), entry.getValue().getBytes(Charset.defaultCharset()));
73+
}
74+
}
75+
return serializedContext;
76+
}
77+
78+
@Override
79+
public Object deserializeContext(Map<String, byte[]> context) {
80+
Map<String, String> contextMap = new HashMap<>();
81+
for (Map.Entry<String, byte[]> entry : context.entrySet()) {
82+
contextMap.put(entry.getKey(), new String(entry.getValue(), Charset.defaultCharset()));
83+
}
84+
return contextMap;
85+
}
86+
87+
@Override
88+
public Object getCurrentContext() {
89+
Map<String, String> carrier = new HashMap<>();
90+
w3cTraceContextPropagator.inject(Context.current(), carrier, setter);
91+
w3cBaggagePropagator.inject(Context.current(), carrier, setter);
92+
return carrier;
93+
}
94+
95+
@Override
96+
public void setCurrentContext(Object context) {
97+
Map<String, String> contextMap = (Map<String, String>) context;
98+
if (contextMap != null) {
99+
for (Map.Entry<String, String> entry : contextMap.entrySet()) {
100+
MDC.put(entry.getKey(), entry.getValue());
101+
}
102+
otelKeySet.set(contextMap.keySet());
103+
}
104+
}
105+
106+
@Override
107+
@SuppressWarnings("MustBeClosedChecker")
108+
public void setUp() {
109+
Context context =
110+
Baggage.fromContext(w3cBaggagePropagator.extract(Context.current(), null, getter))
111+
.toBuilder()
112+
.build()
113+
.storeInContext(w3cTraceContextPropagator.extract(Context.current(), null, getter));
114+
115+
currentContextOtelScope.set(context.makeCurrent());
116+
117+
Span span =
118+
GlobalOpenTelemetry.getTracer("cadence-client")
119+
.spanBuilder("cadence.workflow")
120+
.setParent(context)
121+
.setSpanKind(SpanKind.CLIENT)
122+
.startSpan();
123+
124+
Scope scope = span.makeCurrent();
125+
currentOtelSpan.set(span);
126+
currentOtelScope.set(scope);
127+
}
128+
129+
@Override
130+
public void finish() {
131+
Scope scope = currentOtelScope.get();
132+
if (scope != null) {
133+
scope.close();
134+
}
135+
Span span = currentOtelSpan.get();
136+
if (span != null) {
137+
span.end();
138+
}
139+
Scope contextScope = currentContextOtelScope.get();
140+
if (contextScope != null) {
141+
contextScope.close();
142+
}
143+
}
144+
}

‎src/main/java/com/uber/cadence/internal/context/ContextThreadLocal.java

+51-2
Original file line numberDiff line numberDiff line change
@@ -24,10 +24,14 @@
2424
import java.util.List;
2525
import java.util.Map;
2626
import java.util.function.Supplier;
27+
import org.slf4j.Logger;
28+
import org.slf4j.LoggerFactory;
2729

28-
/** This class holds the current set of context propagators */
30+
/** This class holds the current set of context propagators. */
2931
public class ContextThreadLocal {
3032

33+
private static final Logger log = LoggerFactory.getLogger(ContextThreadLocal.class);
34+
3135
private static WorkflowThreadLocal<List<ContextPropagator>> contextPropagators =
3236
WorkflowThreadLocal.withInitial(
3337
new Supplier<List<ContextPropagator>>() {
@@ -37,7 +41,7 @@ public List<ContextPropagator> get() {
3741
}
3842
});
3943

40-
/** Sets the list of context propagators for the thread */
44+
/** Sets the list of context propagators for the thread. */
4145
public static void setContextPropagators(List<ContextPropagator> propagators) {
4246
if (propagators == null || propagators.isEmpty()) {
4347
return;
@@ -57,6 +61,11 @@ public static Map<String, Object> getCurrentContextForPropagation() {
5761
return contextData;
5862
}
5963

64+
/**
65+
* Injects the context data into the thread for each configured context propagator.
66+
*
67+
* @param contextData The context data received from the server.
68+
*/
6069
public static void propagateContextToCurrentThread(Map<String, Object> contextData) {
6170
if (contextData == null || contextData.isEmpty()) {
6271
return;
@@ -67,4 +76,44 @@ public static void propagateContextToCurrentThread(Map<String, Object> contextDa
6776
}
6877
}
6978
}
79+
80+
/** Calls {@link ContextPropagator#setUp()} for each propagator. */
81+
public static void setUpContextPropagators() {
82+
for (ContextPropagator propagator : contextPropagators.get()) {
83+
try {
84+
propagator.setUp();
85+
} catch (Throwable t) {
86+
// Don't let an error in one propagator block the others
87+
log.error("Error calling setUp() on a contextpropagator", t);
88+
}
89+
}
90+
}
91+
92+
/**
93+
* Calls {@link ContextPropagator#onError(Throwable)} for each propagator.
94+
*
95+
* @param t The Throwable that caused the workflow/activity to finish.
96+
*/
97+
public static void onErrorContextPropagators(Throwable t) {
98+
for (ContextPropagator propagator : contextPropagators.get()) {
99+
try {
100+
propagator.onError(t);
101+
} catch (Throwable t1) {
102+
// Don't let an error in one propagator block the others
103+
log.error("Error calling onError() on a contextpropagator", t1);
104+
}
105+
}
106+
}
107+
108+
/** Calls {@link ContextPropagator#finish()} for each propagator. */
109+
public static void finishContextPropagators() {
110+
for (ContextPropagator propagator : contextPropagators.get()) {
111+
try {
112+
propagator.finish();
113+
} catch (Throwable t) {
114+
// Don't let an error in one propagator block the others
115+
log.error("Error calling finish() on a contextpropagator", t);
116+
}
117+
}
118+
}
70119
}

‎src/main/java/com/uber/cadence/internal/replay/WorkflowContext.java

+13-1
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323
import java.util.HashMap;
2424
import java.util.List;
2525
import java.util.Map;
26+
import java.util.stream.Collectors;
2627

2728
final class WorkflowContext {
2829

@@ -166,7 +167,18 @@ Map<String, Object> getPropagatedContexts() {
166167

167168
Map<String, Object> contextData = new HashMap<>();
168169
for (ContextPropagator propagator : contextPropagators) {
169-
contextData.put(propagator.getName(), propagator.deserializeContext(headerData));
170+
// Only send the context propagator the fields that belong to them
171+
// Change the map from MyPropagator:foo -> bar to foo -> bar
172+
Map<String, byte[]> filteredData =
173+
headerData
174+
.entrySet()
175+
.stream()
176+
.filter(e -> e.getKey().startsWith(propagator.getName()))
177+
.collect(
178+
Collectors.toMap(
179+
e -> e.getKey().substring(propagator.getName().length() + 1),
180+
Map.Entry::getValue));
181+
contextData.put(propagator.getName(), propagator.deserializeContext(filteredData));
170182
}
171183

172184
return contextData;

‎src/main/java/com/uber/cadence/internal/sync/SyncDecisionContext.java

+13-1
Original file line numberDiff line numberDiff line change
@@ -72,6 +72,7 @@
7272
import java.util.function.Consumer;
7373
import java.util.function.Function;
7474
import java.util.function.Supplier;
75+
import java.util.stream.Collectors;
7576
import org.slf4j.Logger;
7677
import org.slf4j.LoggerFactory;
7778

@@ -453,7 +454,18 @@ private Map<String, byte[]> extractContextsAndConvertToBytes(
453454
}
454455
Map<String, byte[]> result = new HashMap<>();
455456
for (ContextPropagator propagator : contextPropagators) {
456-
result.putAll(propagator.serializeContext(propagator.getCurrentContext()));
457+
// Get the serialized context from the propagator
458+
Map<String, byte[]> serializedContext =
459+
propagator.serializeContext(propagator.getCurrentContext());
460+
// Namespace each entry in case of overlaps, so foo -> bar becomes MyPropagator:foo -> bar
461+
Map<String, byte[]> namespacedSerializedContext =
462+
serializedContext
463+
.entrySet()
464+
.stream()
465+
.collect(
466+
Collectors.toMap(
467+
e -> propagator.getName() + ":" + e.getKey(), Map.Entry::getValue));
468+
result.putAll(namespacedSerializedContext);
457469
}
458470
return result;
459471
}

‎src/main/java/com/uber/cadence/internal/sync/WorkflowStubImpl.java

+13-1
Original file line numberDiff line numberDiff line change
@@ -52,6 +52,7 @@
5252
import java.util.concurrent.TimeUnit;
5353
import java.util.concurrent.TimeoutException;
5454
import java.util.concurrent.atomic.AtomicReference;
55+
import java.util.stream.Collectors;
5556

5657
class WorkflowStubImpl implements WorkflowStub {
5758

@@ -204,7 +205,18 @@ private Map<String, byte[]> extractContextsAndConvertToBytes(
204205
}
205206
Map<String, byte[]> result = new HashMap<>();
206207
for (ContextPropagator propagator : contextPropagators) {
207-
result.putAll(propagator.serializeContext(propagator.getCurrentContext()));
208+
// Get the serialized context from the propagator
209+
Map<String, byte[]> serializedContext =
210+
propagator.serializeContext(propagator.getCurrentContext());
211+
// Namespace each entry in case of overlaps, so foo -> bar becomes MyPropagator:foo -> bar
212+
Map<String, byte[]> namespacedSerializedContext =
213+
serializedContext
214+
.entrySet()
215+
.stream()
216+
.collect(
217+
Collectors.toMap(
218+
k -> propagator.getName() + ":" + k.getKey(), Map.Entry::getValue));
219+
result.putAll(namespacedSerializedContext);
208220
}
209221
return result;
210222
}

‎src/main/java/com/uber/cadence/internal/sync/WorkflowThreadImpl.java

+6
Original file line numberDiff line numberDiff line change
@@ -91,6 +91,7 @@ public void run() {
9191
// Repopulate the context(s)
9292
ContextThreadLocal.setContextPropagators(this.contextPropagators);
9393
ContextThreadLocal.propagateContextToCurrentThread(this.propagatedContexts);
94+
ContextThreadLocal.setUpContextPropagators();
9495

9596
try {
9697
// initialYield blocks thread until the first runUntilBlocked is called.
@@ -99,6 +100,7 @@ public void run() {
99100
cancellationScope.run();
100101
} catch (DestroyWorkflowThreadError e) {
101102
if (!threadContext.isDestroyRequested()) {
103+
ContextThreadLocal.onErrorContextPropagators(e);
102104
threadContext.setUnhandledException(e);
103105
}
104106
} catch (Error e) {
@@ -111,9 +113,11 @@ public void run() {
111113
log.error(
112114
String.format("Workflow thread \"%s\" run failed with Error:\n%s", name, stackTrace));
113115
}
116+
ContextThreadLocal.onErrorContextPropagators(e);
114117
threadContext.setUnhandledException(e);
115118
} catch (CancellationException e) {
116119
if (!isCancelRequested()) {
120+
ContextThreadLocal.onErrorContextPropagators(e);
117121
threadContext.setUnhandledException(e);
118122
}
119123
if (log.isDebugEnabled()) {
@@ -130,8 +134,10 @@ public void run() {
130134
"Workflow thread \"%s\" run failed with unhandled exception:\n%s",
131135
name, stackTrace));
132136
}
137+
ContextThreadLocal.onErrorContextPropagators(e);
133138
threadContext.setUnhandledException(e);
134139
} finally {
140+
ContextThreadLocal.finishContextPropagators();
135141
DeterministicRunnerImpl.setCurrentThreadInternal(null);
136142
threadContext.setStatus(Status.DONE);
137143
thread.setName(originalName);

0 commit comments

Comments
 (0)