dpol1 commented on code in PR #9154: URL: https://github.com/apache/storm/pull/9154#discussion_r4272419037
########## 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: With 129 distinct anchor spans in trace A and a later one in B, a stopped scan makes the emit a child of the first anchor instead of a new root. The link limit is also configurable (`otel.span.link.count.limit`) and a fixed cap would drop links the SDK keeps, so I'd keep the full scan and document its cost in Tracing.md. ########## 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: Right, the values are also serialized once per remote task, and the context adds 28 bytes plus any tracestate. I'd leave it until a measurement shows a gain. ########## 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 bundled tracer borrows a `Tracer` from the global SDK, the agent's or the application's, and has nothing to close. Adding a default method later stays binary compatible, so could we add it when a tracer that owns its SDK needs it? -- 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]
