Skip to content

Commit 30f391f

Browse files
Quinn-With-Two-NsQuinn-With-Two-Ns
authored
Add Nexus Worker interceptor (#2278)
Add Nexus Worker interceptor
1 parent 9ac1af3 commit 30f391f

28 files changed

+1053
-29
lines changed

‎build.gradle

+1-1
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,7 @@ ext {
3131
// Platforms
3232
grpcVersion = '1.54.1' // [1.38.0,) Needed for io.grpc.protobuf.services.HealthStatusManager
3333
jacksonVersion = '2.14.2' // [2.9.0,)
34-
nexusVersion = '0.2.1-alpha'
34+
nexusVersion = '0.3.0-alpha' // [0.1.0,)
3535
// we don't upgrade to 1.10.x because it requires kotlin 1.6. Users may use 1.10.x in their environments though.
3636
micrometerVersion = project.hasProperty("edgeDepsTest") ? '1.13.6' : '1.9.9' // [1.0.0,)
3737

‎releases/v1.27.0

+44
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,44 @@
1+
# **💥 BREAKING CHANGES**
2+
3+
## Update With Start **(Pre-release)**
4+
5+
### WorkflowClient
6+
7+
- `WorkflowClient.updateWithStart` has been renamed to `WorkflowClient.startUpdateWithStart`.
8+
- Instead of taking the workflow method, workflow arguments and a `UpdateWithStartWorkflowOperation` , `WorkflowClient.startUpdateWithStart` now takes the update method, update arguments and a `WithStartWorkflowOperation` . `WithStartWorkflowOperation` holds to the workflow method and workflow arguments to be executed together with the update request.
9+
10+
### WorkflowStub
11+
12+
- `WorkflowStub.updateWithStart` has been renamed to `WorkflowStub.startUpdateWithStart`
13+
- `WorkflowStub.startUpdateWithStart` now just takes the `UpdateOptions`, update arguments and workflow arguments
14+
15+
## Update **(Public Preview)**
16+
17+
- The SDK now preforms more rigorous type validation when registering a Workflow with an `@UpdateValidatorMethod` to make sure the type parameters match the linked `@UpdateMethod`
18+
- The SDK will no longer sometimes throw `WorkflowUpdateException` when calling `WorkflowStub.startUpdate` if the update is rejected. `WorkflowUpdateException` is now consistently throw when getting the result of the update
19+
- `UpdateOptionsBuilder` no longer generates a update ID when built. Now a unique UUID is generated when the options are used. This is similar to how `WorkflowOptions` and workflow ID work.
20+
21+
## Nexus **(Public Preview)**
22+
23+
- Workflow started by a Nexus operation now require the Workflow ID to be specified
24+
- The SDK now preforms more rigorous type validation when registering a Nexus Service to make sure it implements the service properly
25+
- All header maps for Nexus operations are now properly case-insensitive.
26+
27+
# **Highlights**
28+
29+
## Virtual Threads **(Public Preview)**
30+
31+
The Java SDK now has experimental support for virtual threads when using a JVM with a version of 21 or higher. Virtual threads can be used inside workflows by enabling `WorkerFactoryOptions.setUsingVirtualWorkflowThreads` . Users can also use virtual threads for task processing in a worker by enabling `WorkerOptions.setUsingVirtualThreads` .
32+
33+
## Nexus **(Public Preview)**
34+
35+
`WorkerInterceptor` now has support for intercepting Nexus workers.
36+
37+
## Update **(Public Preview)**
38+
39+
`WorkflowClient` now has a set of static methods called `startUpdate` that can be used to start an update, but not immediately wait on the result. This is a type safe analog to `WorkflowStub.startUpdate`.
40+
41+
## Workflow Metadata **(Public Preview)**
42+
43+
- The Java SDK now exposes a fixed summary option for local and normal activities.
44+
- The Java SDK now support `__temporal_workflow_metadata` query, this query allows users to get details about a workflow like its’ current description and what signal, update, and query handlers are registered

‎temporal-opentracing/src/main/java/io/temporal/opentracing/OpenTracingWorkerInterceptor.java

+9-4
Original file line numberDiff line numberDiff line change
@@ -20,11 +20,9 @@
2020

2121
package io.temporal.opentracing;
2222

23+
import io.nexusrpc.handler.OperationContext;
2324
import io.temporal.common.interceptors.*;
24-
import io.temporal.opentracing.internal.ContextAccessor;
25-
import io.temporal.opentracing.internal.OpenTracingActivityInboundCallsInterceptor;
26-
import io.temporal.opentracing.internal.OpenTracingWorkflowInboundCallsInterceptor;
27-
import io.temporal.opentracing.internal.SpanFactory;
25+
import io.temporal.opentracing.internal.*;
2826

2927
public class OpenTracingWorkerInterceptor implements WorkerInterceptor {
3028
private final OpenTracingOptions options;
@@ -52,4 +50,11 @@ public ActivityInboundCallsInterceptor interceptActivity(ActivityInboundCallsInt
5250
return new OpenTracingActivityInboundCallsInterceptor(
5351
next, options, spanFactory, contextAccessor);
5452
}
53+
54+
@Override
55+
public NexusOperationInboundCallsInterceptor interceptNexusOperation(
56+
OperationContext context, NexusOperationInboundCallsInterceptor next) {
57+
return new OpenTracingNexusOperationInboundCallsInterceptor(
58+
next, options, spanFactory, contextAccessor);
59+
}
5560
}

‎temporal-opentracing/src/main/java/io/temporal/opentracing/SpanOperationType.java

+4-1
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,10 @@ public enum SpanOperationType {
3434
UPDATE_WORKFLOW("UpdateWorkflow"),
3535
HANDLE_QUERY("HandleQuery"),
3636
HANDLE_SIGNAL("HandleSignal"),
37-
HANDLE_UPDATE("HandleUpdate");
37+
HANDLE_UPDATE("HandleUpdate"),
38+
START_NEXUS_OPERATION("StartNexusOperation"),
39+
RUN_START_NEXUS_OPERATION("RunStartNexusOperationHandler"),
40+
RUN_CANCEL_NEXUS_OPERATION("RunCancelNexusOperationHandler");
3841

3942
private final String defaultPrefix;
4043

‎temporal-opentracing/src/main/java/io/temporal/opentracing/internal/ActionTypeAndNameSpanBuilderProvider.java

+6
Original file line numberDiff line numberDiff line change
@@ -100,6 +100,12 @@ protected Map<String, String> getSpanTags(SpanCreationContext context) {
100100
return ImmutableMap.of(
101101
StandardTagNames.WORKFLOW_ID, context.getWorkflowId(),
102102
StandardTagNames.RUN_ID, context.getRunId());
103+
case START_NEXUS_OPERATION:
104+
return ImmutableMap.of(
105+
StandardTagNames.WORKFLOW_ID, context.getWorkflowId(),
106+
StandardTagNames.RUN_ID, context.getRunId());
107+
case RUN_START_NEXUS_OPERATION:
108+
case RUN_CANCEL_NEXUS_OPERATION:
103109
case HANDLE_QUERY:
104110
return ImmutableMap.of();
105111
}

‎temporal-opentracing/src/main/java/io/temporal/opentracing/internal/ContextAccessor.java

+18
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,8 @@
2424
import io.opentracing.Span;
2525
import io.opentracing.SpanContext;
2626
import io.opentracing.Tracer;
27+
import io.opentracing.propagation.Format;
28+
import io.opentracing.propagation.TextMapAdapter;
2729
import io.temporal.api.common.v1.Payload;
2830
import io.temporal.common.converter.DefaultDataConverter;
2931
import io.temporal.common.converter.StdConverterBackwardsCompatAdapter;
@@ -72,4 +74,20 @@ public SpanContext readSpanContextFromHeader(Header header, Tracer tracer) {
7274
payload, HashMap.class, HASH_MAP_STRING_STRING_TYPE);
7375
return codec.decode(serializedSpanContext, tracer);
7476
}
77+
78+
public Span writeSpanContextToHeader(
79+
Supplier<Span> spanSupplier, Map<String, String> header, Tracer tracer) {
80+
Span span = spanSupplier.get();
81+
writeSpanContextToHeader(span.context(), header, tracer);
82+
return span;
83+
}
84+
85+
public void writeSpanContextToHeader(
86+
SpanContext spanContext, Map<String, String> header, Tracer tracer) {
87+
tracer.inject(spanContext, Format.Builtin.HTTP_HEADERS, new TextMapAdapter(header));
88+
}
89+
90+
public SpanContext readSpanContextFromHeader(Map<String, String> header, Tracer tracer) {
91+
return tracer.extract(Format.Builtin.HTTP_HEADERS, new TextMapAdapter(header));
92+
}
7593
}

‎temporal-opentracing/src/main/java/io/temporal/opentracing/internal/OpenTracingNexusOperationInboundCallsInterceptor.java

+97
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,97 @@
1+
/*
2+
* Copyright (C) 2022 Temporal Technologies, Inc. All Rights Reserved.
3+
*
4+
* Copyright (C) 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
5+
*
6+
* Modifications copyright (C) 2017 Uber Technologies, Inc.
7+
*
8+
* Licensed under the Apache License, Version 2.0 (the "License");
9+
* you may not use this material except in compliance with the License.
10+
* You may obtain a copy of the License at
11+
*
12+
* http://www.apache.org/licenses/LICENSE-2.0
13+
*
14+
* Unless required by applicable law or agreed to in writing, software
15+
* distributed under the License is distributed on an "AS IS" BASIS,
16+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
17+
* See the License for the specific language governing permissions and
18+
* limitations under the License.
19+
*/
20+
21+
package io.temporal.opentracing.internal;
22+
23+
import io.nexusrpc.OperationUnsuccessfulException;
24+
import io.opentracing.Scope;
25+
import io.opentracing.Span;
26+
import io.opentracing.SpanContext;
27+
import io.opentracing.Tracer;
28+
import io.temporal.common.interceptors.NexusOperationInboundCallsInterceptor;
29+
import io.temporal.common.interceptors.NexusOperationInboundCallsInterceptorBase;
30+
import io.temporal.opentracing.OpenTracingOptions;
31+
32+
public class OpenTracingNexusOperationInboundCallsInterceptor
33+
extends NexusOperationInboundCallsInterceptorBase {
34+
private final OpenTracingOptions options;
35+
private final SpanFactory spanFactory;
36+
private final Tracer tracer;
37+
private final ContextAccessor contextAccessor;
38+
39+
public OpenTracingNexusOperationInboundCallsInterceptor(
40+
NexusOperationInboundCallsInterceptor next,
41+
OpenTracingOptions options,
42+
SpanFactory spanFactory,
43+
ContextAccessor contextAccessor) {
44+
super(next);
45+
this.options = options;
46+
this.spanFactory = spanFactory;
47+
this.tracer = options.getTracer();
48+
this.contextAccessor = contextAccessor;
49+
}
50+
51+
@Override
52+
public StartOperationOutput startOperation(StartOperationInput input)
53+
throws OperationUnsuccessfulException {
54+
SpanContext rootSpanContext =
55+
contextAccessor.readSpanContextFromHeader(input.getOperationContext().getHeaders(), tracer);
56+
57+
Span operationStartSpan =
58+
spanFactory
59+
.createStartNexusOperationSpan(
60+
tracer,
61+
input.getOperationContext().getService(),
62+
input.getOperationContext().getOperation(),
63+
rootSpanContext)
64+
.start();
65+
try (Scope scope = tracer.scopeManager().activate(operationStartSpan)) {
66+
return super.startOperation(input);
67+
} catch (Throwable t) {
68+
spanFactory.logFail(operationStartSpan, t);
69+
throw t;
70+
} finally {
71+
operationStartSpan.finish();
72+
}
73+
}
74+
75+
@Override
76+
public CancelOperationOutput cancelOperation(CancelOperationInput input) {
77+
SpanContext rootSpanContext =
78+
contextAccessor.readSpanContextFromHeader(input.getOperationContext().getHeaders(), tracer);
79+
80+
Span operationCancelSpan =
81+
spanFactory
82+
.createCancelNexusOperationSpan(
83+
tracer,
84+
input.getOperationContext().getService(),
85+
input.getOperationContext().getOperation(),
86+
rootSpanContext)
87+
.start();
88+
try (Scope scope = tracer.scopeManager().activate(operationCancelSpan)) {
89+
return super.cancelOperation(input);
90+
} catch (Throwable t) {
91+
spanFactory.logFail(operationCancelSpan, t);
92+
throw t;
93+
} finally {
94+
operationCancelSpan.finish();
95+
}
96+
}
97+
}

‎temporal-opentracing/src/main/java/io/temporal/opentracing/internal/OpenTracingWorkflowOutboundCallsInterceptor.java

+30
Original file line numberDiff line numberDiff line change
@@ -100,6 +100,25 @@ public <R> ChildWorkflowOutput<R> executeChildWorkflow(ChildWorkflowInput<R> inp
100100
}
101101
}
102102

103+
@Override
104+
public <R> ExecuteNexusOperationOutput<R> executeNexusOperation(
105+
ExecuteNexusOperationInput<R> input) {
106+
if (!WorkflowUnsafe.isReplaying()) {
107+
Span nexusOperationExecuteSpan =
108+
contextAccessor.writeSpanContextToHeader(
109+
() -> createStartNexusOperationSpanBuilder(input).start(),
110+
input.getHeaders(),
111+
tracer);
112+
try (Scope ignored = tracer.scopeManager().activate(nexusOperationExecuteSpan)) {
113+
return super.executeNexusOperation(input);
114+
} finally {
115+
nexusOperationExecuteSpan.finish();
116+
}
117+
} else {
118+
return super.executeNexusOperation(input);
119+
}
120+
}
121+
103122
@Override
104123
public SignalExternalOutput signalExternalWorkflow(SignalExternalInput input) {
105124
if (!WorkflowUnsafe.isReplaying()) {
@@ -176,6 +195,17 @@ private <R> Tracer.SpanBuilder createChildWorkflowStartSpanBuilder(ChildWorkflow
176195
parentWorkflowInfo.getRunId());
177196
}
178197

198+
private <R> Tracer.SpanBuilder createStartNexusOperationSpanBuilder(
199+
ExecuteNexusOperationInput<R> input) {
200+
WorkflowInfo parentWorkflowInfo = Workflow.getInfo();
201+
return spanFactory.createStartNexusOperationSpan(
202+
tracer,
203+
input.getService(),
204+
input.getOperation(),
205+
parentWorkflowInfo.getWorkflowId(),
206+
parentWorkflowInfo.getRunId());
207+
}
208+
179209
private Tracer.SpanBuilder createContinueAsNewWorkflowStartSpanBuilder(ContinueAsNewInput input) {
180210
WorkflowInfo continuedWorkflowInfo = Workflow.getInfo();
181211
return spanFactory.createContinueAsNewWorkflowStartSpan(

‎temporal-opentracing/src/main/java/io/temporal/opentracing/internal/SpanFactory.java

+32
Original file line numberDiff line numberDiff line change
@@ -59,6 +59,18 @@ public Tracer.SpanBuilder createWorkflowStartSpan(
5959
return createSpan(context, tracer, null, References.FOLLOWS_FROM);
6060
}
6161

62+
public Tracer.SpanBuilder createStartNexusOperationSpan(
63+
Tracer tracer, String serviceName, String operationName, String workflowId, String runId) {
64+
SpanCreationContext context =
65+
SpanCreationContext.newBuilder()
66+
.setSpanOperationType(SpanOperationType.START_NEXUS_OPERATION)
67+
.setActionName(serviceName + "/" + operationName)
68+
.setWorkflowId(workflowId)
69+
.setRunId(runId)
70+
.build();
71+
return createSpan(context, tracer, null, References.CHILD_OF);
72+
}
73+
6274
public Tracer.SpanBuilder createChildWorkflowStartSpan(
6375
Tracer tracer,
6476
String childWorkflowType,
@@ -173,6 +185,26 @@ public Tracer.SpanBuilder createActivityRunSpan(
173185
return createSpan(context, tracer, activityStartSpanContext, References.FOLLOWS_FROM);
174186
}
175187

188+
public Tracer.SpanBuilder createStartNexusOperationSpan(
189+
Tracer tracer, String serviceName, String operationName, SpanContext nexusStartSpanContext) {
190+
SpanCreationContext context =
191+
SpanCreationContext.newBuilder()
192+
.setSpanOperationType(SpanOperationType.RUN_START_NEXUS_OPERATION)
193+
.setActionName(serviceName + "/" + operationName)
194+
.build();
195+
return createSpan(context, tracer, nexusStartSpanContext, References.FOLLOWS_FROM);
196+
}
197+
198+
public Tracer.SpanBuilder createCancelNexusOperationSpan(
199+
Tracer tracer, String serviceName, String operationName, SpanContext nexusStartSpanContext) {
200+
SpanCreationContext context =
201+
SpanCreationContext.newBuilder()
202+
.setSpanOperationType(SpanOperationType.RUN_CANCEL_NEXUS_OPERATION)
203+
.setActionName(serviceName + "/" + operationName)
204+
.build();
205+
return createSpan(context, tracer, nexusStartSpanContext, References.FOLLOWS_FROM);
206+
}
207+
176208
public Tracer.SpanBuilder createWorkflowStartUpdateSpan(
177209
Tracer tracer, String updateName, String workflowId, String runId) {
178210
SpanCreationContext context =

0 commit comments

Comments
 (0)