-
Notifications
You must be signed in to change notification settings - Fork 3.1k
/
Collector.java
286 lines (250 loc) · 9.48 KB
/
Collector.java
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
/*
* Copyright The OpenZipkin Authors
* SPDX-License-Identifier: Apache-2.0
*/
package zipkin2.collector;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.Executor;
import java.util.function.Supplier;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import zipkin2.Callback;
import zipkin2.Span;
import zipkin2.SpanBytesDecoderDetector;
import zipkin2.codec.BytesDecoder;
import zipkin2.codec.SpanBytesDecoder;
import zipkin2.storage.StorageComponent;
import static zipkin2.Call.propagateIfFatal;
/**
* This component takes action on spans received from a transport. This includes deserializing,
* sampling and scheduling for storage.
*
* <p>Callbacks passed do not propagate to the storage layer. They only return success or failures
* before storage is attempted. This ensures that calling threads are disconnected from storage
* threads.
*/
public class Collector { // not final for mock
static final Callback<Void> NOOP_CALLBACK = new Callback<Void>() {
@Override public void onSuccess(Void value) {
}
@Override public void onError(Throwable t) {
}
};
/** Needed to scope this to the correct logging category */
public static Builder newBuilder(Class<?> loggingClass) {
if (loggingClass == null) throw new NullPointerException("loggingClass == null");
return new Builder(LoggerFactory.getLogger(loggingClass.getName()));
}
public static final class Builder {
final Logger logger;
StorageComponent storage;
CollectorSampler sampler;
CollectorMetrics metrics;
Builder(Logger logger) {
this.logger = logger;
}
/** Sets {@link {@link CollectorComponent.Builder#storage(StorageComponent)}} */
public Builder storage(StorageComponent storage) {
if (storage == null) throw new NullPointerException("storage == null");
this.storage = storage;
return this;
}
/** Sets {@link {@link CollectorComponent.Builder#metrics(CollectorMetrics)}} */
public Builder metrics(CollectorMetrics metrics) {
if (metrics == null) throw new NullPointerException("metrics == null");
this.metrics = metrics;
return this;
}
/** Sets {@link {@link CollectorComponent.Builder#sampler(CollectorSampler)}} */
public Builder sampler(CollectorSampler sampler) {
if (sampler == null) throw new NullPointerException("sampler == null");
this.sampler = sampler;
return this;
}
public Collector build() {
return new Collector(this);
}
}
final Logger logger;
final CollectorMetrics metrics;
final CollectorSampler sampler;
final StorageComponent storage;
Collector(Builder builder) {
if (builder.logger == null) throw new NullPointerException("logger == null");
this.logger = builder.logger;
this.metrics = builder.metrics == null ? CollectorMetrics.NOOP_METRICS : builder.metrics;
if (builder.storage == null) throw new NullPointerException("storage == null");
this.storage = builder.storage;
this.sampler = builder.sampler == null ? CollectorSampler.ALWAYS_SAMPLE : builder.sampler;
}
public void accept(List<Span> spans, Callback<Void> callback) {
accept(spans, callback, Runnable::run);
}
/**
* Calls to get the storage component could be blocking. This ensures requests that block
* callers (such as http or gRPC) do not add additional load during such events.
*
* @param executor the executor used to enqueue the storage request.
*/
public void accept(List<Span> spans, Callback<Void> callback, Executor executor) {
if (spans.isEmpty()) {
callback.onSuccess(null);
return;
}
metrics.incrementSpans(spans.size());
List<Span> sampledSpans = sample(spans);
if (sampledSpans.isEmpty()) {
callback.onSuccess(null);
return;
}
// In order to ensure callers are not blocked, we swap callbacks when we get to the storage
// phase of this process. Here, we create a callback whose sole purpose is classifying later
// errors on this bundle of spans in the same log category. This allows people to only turn on
// debug logging in one place.
try {
executor.execute(new StoreSpans(sampledSpans));
callback.onSuccess(null);
} catch (Throwable unexpected) { // ensure if a future is supplied we always set value or error
callback.onError(unexpected);
throw unexpected;
}
}
/** Like {@link #acceptSpans(byte[], BytesDecoder, Callback)}, except using a byte buffer. */
public void acceptSpans(ByteBuffer encoded, SpanBytesDecoder decoder, Callback<Void> callback,
Executor executor) {
List<Span> spans;
try {
spans = decoder.decodeList(encoded);
} catch (RuntimeException | Error e) {
handleDecodeError(e, callback);
return;
}
accept(spans, callback, executor);
}
/**
* Before calling this, call {@link CollectorMetrics#incrementMessages()}, and {@link
* CollectorMetrics#incrementBytes(int)}. Do not call any other metrics callbacks as those are
* handled internal to this method.
*
* @param serialized not empty message
*/
public void acceptSpans(byte[] serialized, Callback<Void> callback) {
BytesDecoder<Span> decoder;
try {
decoder = SpanBytesDecoderDetector.decoderForListMessage(serialized);
} catch (RuntimeException | Error e) {
handleDecodeError(e, callback);
return;
}
acceptSpans(serialized, decoder, callback);
}
/**
* Before calling this, call {@link CollectorMetrics#incrementMessages()}, and {@link
* CollectorMetrics#incrementBytes(int)}. Do not call any other metrics callbacks as those are
* handled internal to this method.
*
* @param serializedSpans not empty message
*/
public void acceptSpans(
byte[] serializedSpans, BytesDecoder<Span> decoder, Callback<Void> callback) {
List<Span> spans;
try {
spans = decodeList(decoder, serializedSpans);
} catch (RuntimeException | Error e) {
handleDecodeError(e, callback);
return;
}
accept(spans, callback);
}
List<Span> decodeList(BytesDecoder<Span> decoder, byte[] serialized) {
List<Span> out = new ArrayList<>();
decoder.decodeList(serialized, out);
return out;
}
void store(List<Span> sampledSpans, Callback<Void> callback) {
storage.spanConsumer().accept(sampledSpans).enqueue(callback);
}
String idString(Span span) {
return span.traceId() + "/" + span.id();
}
List<Span> sample(List<Span> input) {
List<Span> sampled = new ArrayList<>(input.size());
for (int i = 0, length = input.size(); i < length; i++) {
Span s = input.get(i);
if (sampler.isSampled(s.traceId(), Boolean.TRUE.equals(s.debug()))) {
sampled.add(s);
}
}
int dropped = input.size() - sampled.size();
if (dropped > 0) metrics.incrementSpansDropped(dropped);
return sampled;
}
class StoreSpans implements Callback<Void>, Runnable {
final List<Span> spans;
StoreSpans(List<Span> spans) {
this.spans = spans;
}
@Override public void run() {
try {
store(spans, this);
} catch (RuntimeException | Error e) {
// While unexpected, invoking the storage command could raise an error synchronously. When
// that's the case, we wouldn't have invoked callback.onSuccess, so we need to handle the
// error here.
onError(e);
}
}
@Override public void onSuccess(Void value) {
}
@Override public void onError(Throwable t) {
handleStorageError(spans, t, NOOP_CALLBACK);
}
@Override public String toString() {
return appendSpanIds(spans, new StringBuilder("StoreSpans(")) + ")";
}
}
void handleDecodeError(Throwable e, Callback<Void> callback) {
metrics.incrementMessagesDropped();
handleError(e, "Cannot decode spans"::toString, callback);
}
/**
* When storing spans, an exception can be raised before or after the fact. This adds context of
* span ids to give logs more relevance.
*/
void handleStorageError(List<Span> spans, Throwable e, Callback<Void> callback) {
metrics.incrementSpansDropped(spans.size());
// The exception could be related to a span being huge. Instead of filling logs,
// print trace id, span id pairs
handleError(e, () -> appendSpanIds(spans, new StringBuilder("Cannot store spans ")), callback);
}
void handleError(Throwable e, Supplier<String> defaultLogMessage, Callback<Void> callback) {
propagateIfFatal(e);
callback.onError(e);
if (!logger.isDebugEnabled()) return;
String error = e.getMessage() != null ? e.getMessage() : "";
// We have specific code that customizes log messages. Use this when the case.
if (error.startsWith("Malformed") || error.startsWith("Truncated")) {
logger.debug(error, e);
} else { // otherwise, beautify the message
String message =
"%s due to %s(%s)".formatted(defaultLogMessage.get(), e.getClass().getSimpleName(), error);
logger.debug(message, e);
}
}
// TODO: this logic needs to be redone as service names are more important than span IDs. Also,
// span IDs repeat between client and server!
String appendSpanIds(List<Span> spans, StringBuilder message) {
message.append("[");
int i = 0;
Iterator<Span> iterator = spans.iterator();
while (iterator.hasNext() && i++ < 3) {
message.append(idString(iterator.next()));
if (iterator.hasNext()) message.append(", ");
}
if (iterator.hasNext()) message.append("...");
return message.append("]").toString();
}
}