rzo1 commented on code in PR #9154:
URL: https://github.com/apache/storm/pull/9154#discussion_r4269207001
##########
storm-client/src/jvm/org/apache/storm/serialization/KryoTupleSerializer.java:
##########
@@ -64,4 +81,18 @@ public byte[] serialize(Tuple tuple) {
throw new RuntimeException(e);
}
}
+
+ /**
+ * Appends an entry after the values: a tag byte, the payload length as a
varint, then the payload. Readers skip entries with
+ * an unknown tag, and readers that stop after the values ignore all of
them.
+ */
+ private void writeTraceContext(Object traceContext) {
+ byte[] payload = tracer.encode(traceContext);
Review Comment:
Minor: with an all grouping to tasks on remote workers, the same context is
encoded once per destination, including the construction of the tracestate
string if one is present. Caching the encoded form per context object would
avoid this, although the gain is presumably small.
##########
external/storm-opentelemetry/src/main/java/org/apache/storm/opentelemetry/OpenTelemetryTupleTracer.java:
##########
@@ -0,0 +1,290 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership. The
ASF licenses this file to you under the Apache License, Version
+ * 2.0 (the "License"); you may not use this file except in compliance with
the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions
+ * and limitations under the License.
+ */
+
+package org.apache.storm.opentelemetry;
+
+import io.opentelemetry.api.GlobalOpenTelemetry;
+import io.opentelemetry.api.common.AttributeKey;
+import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.api.common.AttributesBuilder;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.SpanBuilder;
+import io.opentelemetry.api.trace.SpanContext;
+import io.opentelemetry.api.trace.StatusCode;
+import io.opentelemetry.api.trace.Tracer;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.context.Scope;
+import java.net.UnknownHostException;
+import java.time.Instant;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import org.apache.storm.Config;
+import org.apache.storm.task.WorkerTopologyContext;
+import org.apache.storm.tracing.TupleTracer;
+import org.apache.storm.tuple.Tuple;
+import org.apache.storm.utils.Time;
+import org.apache.storm.utils.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Traces tuple trees with OpenTelemetry. To use it, set {@link
Config#TOPOLOGY_TRACING_TRACER} to this class name. Spans go to
+ * the SDK registered as the global OpenTelemetry instance, usually by the
OpenTelemetry Java agent, version 2.23.0 or later. Until
+ * one is registered, tuples are not traced and one warning is logged.
+ *
+ * <p>Each spout emit starts a trace with a root span named "component emit",
started and ended at once. Each bolt
+ * {@code execute()} of a traced tuple runs in a child span named "component
execute", current while {@code execute()} runs. A bolt
+ * emit carries the context of its anchors: the anchor's context if they share
one span. Otherwise it carries the context of a span
+ * named "component emit", started and ended at once: a child of the first
anchor's span linked to the others if all anchors belong
+ * to one trace, else a new root span linked to each.
+ *
+ * <p>The spout records the ack, fail or timeout of each traced tuple tree as
a span whose duration, also set as the
+ * {@code storm.tuple.latency_ms} attribute, is the time from the emit to the
outcome. A bolt records each {@code fail()} as a span
+ * started and ended at once. Contexts with the sampled flag unset are
propagated too.
+ */
+public class OpenTelemetryTupleTracer implements TupleTracer {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(OpenTelemetryTupleTracer.class);
+ private static final String INSTRUMENTATION_SCOPE = "org.apache.storm";
+ private static final long SDK_LOOKUP_INTERVAL_MS = 1000;
+ private static final long UNTRACED_EMIT_LOG_INTERVAL_MS = 60_000;
+
+ private static final AttributeKey<String> TOPOLOGY_NAME_KEY =
AttributeKey.stringKey("storm.topology.name");
+ private static final AttributeKey<String> TOPOLOGY_ID_KEY =
AttributeKey.stringKey("storm.topology.id");
+ private static final AttributeKey<String> COMPONENT_ID_KEY =
AttributeKey.stringKey("storm.component.id");
+ private static final AttributeKey<Long> TASK_ID_KEY =
AttributeKey.longKey("storm.task.id");
+ private static final AttributeKey<String> SOURCE_COMPONENT_ID_KEY =
AttributeKey.stringKey("storm.source.component.id");
+ private static final AttributeKey<String> SOURCE_STREAM_ID_KEY =
AttributeKey.stringKey("storm.source.stream.id");
+ private static final AttributeKey<String> WORKER_HOST_KEY =
AttributeKey.stringKey("storm.worker.host");
+ private static final AttributeKey<Long> WORKER_PORT_KEY =
AttributeKey.longKey("storm.worker.port");
+ private static final AttributeKey<Long> TUPLE_LATENCY_KEY =
AttributeKey.longKey("storm.tuple.latency_ms");
+
+ private WorkerTopologyContext context;
+ private Attributes workerAttributes;
+ /** Indexed by task id; null for the tasks of other workers and for system
tasks with a negative id. */
+ private TaskSpans[] taskSpans;
+ private volatile Tracer tracer;
+ private volatile long nextSdkLookupMs;
+ private final AtomicBoolean warnedNoSdk = new AtomicBoolean();
+
+ @Override
+ public void prepare(Map<String, Object> topoConf, WorkerTopologyContext
context) {
+ this.context = context;
+ AttributesBuilder attributes = Attributes.builder()
+ .put(TOPOLOGY_NAME_KEY, (String)
topoConf.get(Config.TOPOLOGY_NAME))
+ .put(TOPOLOGY_ID_KEY, context.getStormId())
+ .put(WORKER_PORT_KEY, context.getThisWorkerPort().longValue());
+ try {
+ attributes.put(WORKER_HOST_KEY, Utils.hostname());
+ } catch (UnknownHostException e) {
+ LOG.warn("Execute spans get no {} attribute: the host name is
unknown", WORKER_HOST_KEY, e);
+ }
+ this.workerAttributes = attributes.build();
+ int maxTaskId =
context.getThisWorkerTasks().stream().mapToInt(Integer::intValue).max().orElse(-1);
+ TaskSpans[] byTask = new TaskSpans[maxTaskId + 1];
+ for (int taskId : context.getThisWorkerTasks()) {
+ if (taskId >= 0) {
+ byTask[taskId] = newTaskSpans(taskId);
+ }
+ }
+ this.taskSpans = byTask;
+ }
+
+ @Override
+ public Object spoutEmit(int taskId, String streamId) {
+ Tracer current = tracer();
+ return current == null ? null :
emitContext(current.spanBuilder(spans(taskId).emitName()).setNoParent());
+ }
+
+ @Override
+ public Object boltEmit(int taskId, String streamId, List<Object>
anchorContexts) {
+ if (anchorContexts.isEmpty()) {
+ logUntracedEmitUnderSpan(taskId, streamId);
+ return null;
+ }
+ Context first = (Context) anchorContexts.get(0);
+ if (anchorContexts.size() == 1) {
+ return first;
+ }
+ SpanContext firstSpan = Span.fromContext(first).getSpanContext();
+ Set<SpanContext> otherSpans = new LinkedHashSet<>();
Review Comment:
This set grows with the number of anchors, whereas at most 128 links are
retained (line 135). Stopping once 129 distinct span contexts have been
collected would bound the cost for emits with many anchors without changing the
result.
##########
storm-client/src/jvm/org/apache/storm/daemon/worker/WorkerState.java:
##########
@@ -662,6 +674,17 @@ public final WorkerUserContext getWorkerUserContext() {
}
}
+ private TupleTracer mkTupleTracer() {
+ String className = (String)
topologyConf.get(Config.TOPOLOGY_TRACING_TRACER);
+ if (className == null) {
+ return null;
+ }
+ TupleTracer tracer = ReflectionUtils.newInstance(className);
Review Comment:
Minor: if the class is not on the worker classpath, for instance because
`storm-opentelemetry` is missing from `lib-worker`, this call fails on every
worker, and the workers are restarted repeatedly. A sentence in `Tracing.md`
would make this failure mode easier to diagnose.
##########
storm-client/src/jvm/org/apache/storm/executor/bolt/BoltOutputCollectorImpl.java:
##########
@@ -117,6 +126,26 @@ private List<Integer> boltEmit(String streamId,
Collection<Tuple> anchors, List<
return outTasks;
}
+ /**
+ * Returns the trace contexts of the anchors, in anchor order, skipping
anchors without one.
+ */
+ private static List<Object> traceContexts(Collection<Tuple> anchors) {
Review Comment:
For windowed bolts, every emit is anchored to the whole window, so this
method allocates a list of the window's size on each emit, and `boltEmit`
subsequently builds a set of all span contexts (see the corresponding comment
in `OpenTelemetryTupleTracer`). Since the SDK retains at most 128 links by
default, the work is largely discarded.
##########
storm-client/src/jvm/org/apache/storm/executor/bolt/BoltExecutor.java:
##########
@@ -224,7 +225,18 @@ public void tupleActionFn(int taskId, TupleImpl tuple)
throws Exception {
if (isExecuteSampler) {
tuple.setExecuteSampleStartTime(now);
}
- boltObject.execute(tuple);
+ Object received = tupleTracer == null ? null :
tuple.getTraceContext();
+ TupleTracer.ExecuteScope scope =
+ received == null ? null : tupleTracer.startExecute(taskId,
tuple, received);
+ if (scope == null) {
+ boltObject.execute(tuple);
+ } else {
+ try (scope) {
+ // anchored emits and fail() read the context from the
tuple, also after execute() returns
+ tuple.setTraceContext(scope.context());
+ boltObject.execute(tuple);
+ }
+ }
Long ms = tuple.getExecuteSampleStartTime();
Review Comment:
Since `delta` is computed after the `try (scope)` block, the execute latency
now includes the start of the span, `span.end()` and the enqueueing into the
batch span processor. This affects the execute latency and capacity shown in
the UI, the task metrics and `BoltExecuteInfo`, so that a comparison with
tracing enabled and disabled overstates the cost of the bolt itself. The time
could be taken before the scope is closed; alternatively, the behaviour should
be documented.
##########
storm-client/src/jvm/org/apache/storm/tracing/TupleTracer.java:
##########
@@ -0,0 +1,103 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership. The
ASF licenses this file to you under the Apache License, Version
+ * 2.0 (the "License"); you may not use this file except in compliance with
the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions
+ * and limitations under the License.
+ */
+
+package org.apache.storm.tracing;
+
+import java.util.List;
+import java.util.Map;
+import org.apache.storm.Config;
+import org.apache.storm.task.WorkerTopologyContext;
+import org.apache.storm.tuple.Tuple;
+
+/**
+ * Attaches a trace context to tuples and records how their tuple trees are
processed. Each worker creates one instance from the
+ * class named in {@link Config#TOPOLOGY_TRACING_TRACER}, through its zero-arg
constructor, and shares it between its executors
+ * and the threads that serialize and deserialize tuples, so implementations
must be thread safe.
+ *
+ * <p>A context is any object the implementation chooses; null means not
traced. Storm keeps it on the tuple and, for the tuple
+ * trees a spout starts, until the tree completes. Tuples sent to another
worker carry the bytes {@link #encode} returns, and the
+ * receiving worker turns them back into a context with {@link #decode}.
+ *
+ * <p>An exception from {@link #decode} leaves the tuple without a context.
Exceptions from the other methods propagate to the
Review Comment:
According to this contract, exceptions from every method except `decode`
propagate to the caller. An exception in `encode` therefore fails
`KryoTupleSerializer.serialize()` and terminates the worker, and the same holds
for `spoutEmit` and `boltEmit` (which run within the user's `emit()`),
`startExecute`, and `spoutOutcome` (which runs before `spout.ack()`). The
OpenTelemetry API is designed not to throw, so the bundled tracer is unlikely
to be affected, but the contract applies to every implementation. Since the
tracer serves observability only, I would expect a failure to leave the
processing of tuples unaffected, as is already the case for `decode`. Catching
`RuntimeException` around the calls in storm-client, logging at a limited rate
and treating the result as null would achieve this without cost on the regular
path. Was the propagation a deliberate choice?
##########
storm-client/src/jvm/org/apache/storm/tracing/TupleTracer.java:
##########
@@ -0,0 +1,103 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership. The
ASF licenses this file to you under the Apache License, Version
+ * 2.0 (the "License"); you may not use this file except in compliance with
the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions
+ * and limitations under the License.
+ */
+
+package org.apache.storm.tracing;
+
+import java.util.List;
+import java.util.Map;
+import org.apache.storm.Config;
+import org.apache.storm.task.WorkerTopologyContext;
+import org.apache.storm.tuple.Tuple;
+
+/**
+ * Attaches a trace context to tuples and records how their tuple trees are
processed. Each worker creates one instance from the
+ * class named in {@link Config#TOPOLOGY_TRACING_TRACER}, through its zero-arg
constructor, and shares it between its executors
+ * and the threads that serialize and deserialize tuples, so implementations
must be thread safe.
+ *
+ * <p>A context is any object the implementation chooses; null means not
traced. Storm keeps it on the tuple and, for the tuple
+ * trees a spout starts, until the tree completes. Tuples sent to another
worker carry the bytes {@link #encode} returns, and the
+ * receiving worker turns them back into a context with {@link #decode}.
+ *
+ * <p>An exception from {@link #decode} leaves the tuple without a context.
Exceptions from the other methods propagate to the
+ * caller.
+ */
+public interface TupleTracer {
+
+ /**
+ * Called once, before any other method.
+ */
+ void prepare(Map<String, Object> topoConf, WorkerTopologyContext context);
Review Comment:
The interface defines `prepare` but no counterpart that is called on worker
shutdown, so an implementation has no opportunity to flush or release
resources. This is irrelevant with the agent, but not for an SDK registered by
the application, nor in local mode with several topologies in one JVM. I
suggest adding `default void cleanup() {}` before the interface becomes public
API.
##########
storm-client/src/jvm/org/apache/storm/executor/spout/SpoutExecutor.java:
##########
@@ -379,6 +385,12 @@ public void failSpoutMsg(SpoutExecutor executor, Task
taskData, Long timeDelta,
if (executor.getIsDebug()) {
LOG.info("SPOUT Failing {} : {} REASON: {}",
tupleInfo.getRootId(), tupleInfo, reason);
}
+ Object traceContext = tupleInfo.getTraceContext();
+ if (traceContext != null) {
+ TupleTracer.Outcome outcome = "TIMEOUT".equals(reason) ?
TupleTracer.Outcome.TIMEOUT : TupleTracer.Outcome.FAIL;
Review Comment:
Minor: the outcome is derived from the literal reason passed by the expiry
callback of the pending map. A constant shared by both places would make this
dependency explicit.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
