Skip to content

Commit 16fa0d1

Browse files
anuraagaanuraaga
committed
Implement SDK metrics for logs
1 parent 22d90b6 commit 16fa0d1

18 files changed

+826
-62
lines changed

‎docs/apidiffs/current_vs_latest/opentelemetry-sdk-logs.txt‎

Lines changed: 15 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,2 +1,16 @@
11
Comparing source compatibility of opentelemetry-sdk-logs-1.58.0-SNAPSHOT.jar against opentelemetry-sdk-logs-1.57.0.jar
2-
No changes.
2+
*** MODIFIED CLASS: PUBLIC FINAL io.opentelemetry.sdk.logs.export.BatchLogRecordProcessorBuilder (not serializable)
3+
=== CLASS FILE FORMAT VERSION: 52.0 <- 52.0
4+
+++ NEW METHOD: PUBLIC(+) io.opentelemetry.sdk.logs.export.BatchLogRecordProcessorBuilder setInternalTelemetryVersion(io.opentelemetry.sdk.common.InternalTelemetryVersion)
5+
+++ NEW METHOD: PUBLIC(+) io.opentelemetry.sdk.logs.export.BatchLogRecordProcessorBuilder setMeterProvider(java.util.function.Supplier<io.opentelemetry.api.metrics.MeterProvider>)
6+
*** MODIFIED CLASS: PUBLIC FINAL io.opentelemetry.sdk.logs.export.SimpleLogRecordProcessor (not serializable)
7+
=== CLASS FILE FORMAT VERSION: 52.0 <- 52.0
8+
+++ NEW METHOD: PUBLIC(+) STATIC(+) io.opentelemetry.sdk.logs.export.SimpleLogRecordProcessorBuilder builder(io.opentelemetry.sdk.logs.export.LogRecordExporter)
9+
+++ NEW CLASS: PUBLIC(+) FINAL(+) io.opentelemetry.sdk.logs.export.SimpleLogRecordProcessorBuilder (not serializable)
10+
+++ CLASS FILE FORMAT VERSION: 52.0 <- n.a.
11+
+++ NEW SUPERCLASS: java.lang.Object
12+
+++ NEW METHOD: PUBLIC(+) io.opentelemetry.sdk.logs.export.SimpleLogRecordProcessor build()
13+
+++ NEW METHOD: PUBLIC(+) io.opentelemetry.sdk.logs.export.SimpleLogRecordProcessorBuilder setMeterProvider(java.util.function.Supplier<io.opentelemetry.api.metrics.MeterProvider>)
14+
*** MODIFIED CLASS: PUBLIC FINAL io.opentelemetry.sdk.logs.SdkLoggerProviderBuilder (not serializable)
15+
=== CLASS FILE FORMAT VERSION: 52.0 <- 52.0
16+
+++ NEW METHOD: PUBLIC(+) io.opentelemetry.sdk.logs.SdkLoggerProviderBuilder setMeterProvider(java.util.function.Supplier<io.opentelemetry.api.metrics.MeterProvider>)

‎sdk/logs/src/main/java/io/opentelemetry/sdk/logs/LoggerSharedState.java‎

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -23,19 +23,22 @@ final class LoggerSharedState {
2323
private final LogRecordProcessor logRecordProcessor;
2424
private final Clock clock;
2525
private final ExceptionAttributeResolver exceptionAttributeResolver;
26+
private final SdkLoggerInstrumentation loggerInstrumentation;
2627
@Nullable private volatile CompletableResultCode shutdownResult = null;
2728

2829
LoggerSharedState(
2930
Resource resource,
3031
Supplier<LogLimits> logLimitsSupplier,
3132
LogRecordProcessor logRecordProcessor,
3233
Clock clock,
33-
ExceptionAttributeResolver exceptionAttributeResolver) {
34+
ExceptionAttributeResolver exceptionAttributeResolver,
35+
SdkLoggerInstrumentation loggerInstrumentation) {
3436
this.resource = resource;
3537
this.logLimitsSupplier = logLimitsSupplier;
3638
this.logRecordProcessor = logRecordProcessor;
3739
this.clock = clock;
3840
this.exceptionAttributeResolver = exceptionAttributeResolver;
41+
this.loggerInstrumentation = loggerInstrumentation;
3942
}
4043

4144
Resource getResource() {
@@ -58,6 +61,10 @@ ExceptionAttributeResolver getExceptionAttributeResolver() {
5861
return exceptionAttributeResolver;
5962
}
6063

64+
SdkLoggerInstrumentation getLoggerInstrumentation() {
65+
return loggerInstrumentation;
66+
}
67+
6168
boolean hasBeenShutdown() {
6269
return shutdownResult != null;
6370
}

‎sdk/logs/src/main/java/io/opentelemetry/sdk/logs/SdkLogRecordBuilder.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -132,6 +132,8 @@ public void emit() {
132132
this.observedTimestampEpochNanos == 0
133133
? this.loggerSharedState.getClock().now()
134134
: this.observedTimestampEpochNanos;
135+
136+
loggerSharedState.getLoggerInstrumentation().emitLog();
135137
loggerSharedState
136138
.getLogRecordProcessor()
137139
.onEmit(context, createLogRecord(context, observedTimestampEpochNanos));

‎sdk/logs/src/main/java/io/opentelemetry/sdk/logs/SdkLoggerInstrumentation.java‎

Lines changed: 61 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,61 @@
1+
/*
2+
* Copyright The OpenTelemetry Authors
3+
* SPDX-License-Identifier: Apache-2.0
4+
*/
5+
6+
package io.opentelemetry.sdk.logs;
7+
8+
import io.opentelemetry.api.metrics.LongCounter;
9+
import io.opentelemetry.api.metrics.Meter;
10+
import io.opentelemetry.api.metrics.MeterProvider;
11+
import java.util.function.Supplier;
12+
import javax.annotation.Nullable;
13+
14+
/**
15+
* SDK metrics exported for emitted logs as defined in the <a
16+
* href="https://opentelemetry.io/docs/specs/semconv/otel/sdk-metrics/#log-metrics">semantic
17+
* conventions</a>.
18+
*/
19+
final class SdkLoggerInstrumentation {
20+
private final Object lock = new Object();
21+
22+
private final Supplier<MeterProvider> meterProvider;
23+
24+
@Nullable private Meter meter;
25+
@Nullable private volatile LongCounter createdLogs;
26+
27+
SdkLoggerInstrumentation(Supplier<MeterProvider> meterProvider) {
28+
this.meterProvider = meterProvider;
29+
}
30+
31+
void emitLog() {
32+
createdLogs().add(1);
33+
}
34+
35+
private LongCounter createdLogs() {
36+
LongCounter createdLogs = this.createdLogs;
37+
if (createdLogs == null) {
38+
synchronized (lock) {
39+
createdLogs = this.createdLogs;
40+
if (createdLogs == null) {
41+
createdLogs =
42+
meter()
43+
.counterBuilder("otel.sdk.log.created")
44+
.setUnit("{log_record}")
45+
.setDescription("The number of logs submitted to enabled SDK Loggers.")
46+
.build();
47+
this.createdLogs = createdLogs;
48+
}
49+
}
50+
}
51+
return createdLogs;
52+
}
53+
54+
private Meter meter() {
55+
if (meter == null) {
56+
// Safe to call from multiple threads.
57+
meter = meterProvider.get().get("io.opentelemetry.sdk.logs");
58+
}
59+
return meter;
60+
}
61+
}

‎sdk/logs/src/main/java/io/opentelemetry/sdk/logs/SdkLoggerProvider.java‎

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@
99
import io.opentelemetry.api.logs.Logger;
1010
import io.opentelemetry.api.logs.LoggerBuilder;
1111
import io.opentelemetry.api.logs.LoggerProvider;
12+
import io.opentelemetry.api.metrics.MeterProvider;
1213
import io.opentelemetry.sdk.common.Clock;
1314
import io.opentelemetry.sdk.common.CompletableResultCode;
1415
import io.opentelemetry.sdk.common.InstrumentationScopeInfo;
@@ -58,11 +59,17 @@ public static SdkLoggerProviderBuilder builder() {
5859
List<LogRecordProcessor> processors,
5960
Clock clock,
6061
ScopeConfigurator<LoggerConfig> loggerConfigurator,
61-
ExceptionAttributeResolver exceptionAttributeResolver) {
62+
ExceptionAttributeResolver exceptionAttributeResolver,
63+
Supplier<MeterProvider> meterProvider) {
6264
LogRecordProcessor logRecordProcessor = LogRecordProcessor.composite(processors);
6365
this.sharedState =
6466
new LoggerSharedState(
65-
resource, logLimitsSupplier, logRecordProcessor, clock, exceptionAttributeResolver);
67+
resource,
68+
logLimitsSupplier,
69+
logRecordProcessor,
70+
clock,
71+
exceptionAttributeResolver,
72+
new SdkLoggerInstrumentation(meterProvider));
6673
this.loggerComponentRegistry =
6774
new ComponentRegistry<>(
6875
instrumentationScopeInfo ->

‎sdk/logs/src/main/java/io/opentelemetry/sdk/logs/SdkLoggerProviderBuilder.java‎

Lines changed: 15 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@
99

1010
import io.opentelemetry.api.logs.LogRecordBuilder;
1111
import io.opentelemetry.api.logs.Logger;
12+
import io.opentelemetry.api.metrics.MeterProvider;
1213
import io.opentelemetry.context.Context;
1314
import io.opentelemetry.sdk.common.Clock;
1415
import io.opentelemetry.sdk.common.InstrumentationScopeInfo;
@@ -40,6 +41,7 @@ public final class SdkLoggerProviderBuilder {
4041
LoggerConfig.configuratorBuilder();
4142
private ExceptionAttributeResolver exceptionAttributeResolver =
4243
ExceptionAttributeResolver.getDefault();
44+
private Supplier<MeterProvider> meterProvider = MeterProvider::noop;
4345

4446
SdkLoggerProviderBuilder() {}
4547

@@ -186,6 +188,17 @@ SdkLoggerProviderBuilder setExceptionAttributeResolver(
186188
return this;
187189
}
188190

191+
/**
192+
* Sets the {@link MeterProvider} to use to generate <a
193+
* href="https://opentelemetry.io/docs/specs/semconv/otel/sdk-metrics/#span-metrics">SDK Span
194+
* Metrics</a>.
195+
*/
196+
public SdkLoggerProviderBuilder setMeterProvider(Supplier<MeterProvider> meterProvider) {
197+
requireNonNull(meterProvider, "meterProvider");
198+
this.meterProvider = meterProvider;
199+
return this;
200+
}
201+
189202
/**
190203
* Create a {@link SdkLoggerProvider} instance.
191204
*
@@ -198,6 +211,7 @@ public SdkLoggerProvider build() {
198211
logRecordProcessors,
199212
clock,
200213
loggerConfiguratorBuilder.build(),
201-
exceptionAttributeResolver);
214+
exceptionAttributeResolver,
215+
meterProvider);
202216
}
203217
}

‎sdk/logs/src/main/java/io/opentelemetry/sdk/logs/export/BatchLogRecordProcessor.java‎

Lines changed: 31 additions & 53 deletions
Original file line numberDiff line numberDiff line change
@@ -5,13 +5,11 @@
55

66
package io.opentelemetry.sdk.logs.export;
77

8-
import io.opentelemetry.api.common.AttributeKey;
9-
import io.opentelemetry.api.common.Attributes;
10-
import io.opentelemetry.api.metrics.LongCounter;
11-
import io.opentelemetry.api.metrics.Meter;
128
import io.opentelemetry.api.metrics.MeterProvider;
139
import io.opentelemetry.context.Context;
1410
import io.opentelemetry.sdk.common.CompletableResultCode;
11+
import io.opentelemetry.sdk.common.InternalTelemetryVersion;
12+
import io.opentelemetry.sdk.internal.ComponentId;
1513
import io.opentelemetry.sdk.internal.DaemonThreadFactory;
1614
import io.opentelemetry.sdk.logs.LogRecordProcessor;
1715
import io.opentelemetry.sdk.logs.ReadWriteLogRecord;
@@ -26,6 +24,7 @@
2624
import java.util.concurrent.atomic.AtomicBoolean;
2725
import java.util.concurrent.atomic.AtomicInteger;
2826
import java.util.concurrent.atomic.AtomicReference;
27+
import java.util.function.Supplier;
2928
import java.util.logging.Level;
3029
import java.util.logging.Logger;
3130

@@ -42,14 +41,11 @@
4241
*/
4342
public final class BatchLogRecordProcessor implements LogRecordProcessor {
4443

44+
private static final ComponentId COMPONENT_ID =
45+
ComponentId.generateLazy("batching_log_processor");
46+
4547
private static final String WORKER_THREAD_NAME =
4648
BatchLogRecordProcessor.class.getSimpleName() + "_WorkerThread";
47-
private static final AttributeKey<String> LOG_RECORD_PROCESSOR_TYPE_LABEL =
48-
AttributeKey.stringKey("processorType");
49-
private static final AttributeKey<Boolean> LOG_RECORD_PROCESSOR_DROPPED_LABEL =
50-
AttributeKey.booleanKey("dropped");
51-
private static final String LOG_RECORD_PROCESSOR_TYPE_VALUE =
52-
BatchLogRecordProcessor.class.getSimpleName();
5349

5450
private final Worker worker;
5551
private final AtomicBoolean isShutdown = new AtomicBoolean(false);
@@ -67,7 +63,8 @@ public static BatchLogRecordProcessorBuilder builder(LogRecordExporter logRecord
6763

6864
BatchLogRecordProcessor(
6965
LogRecordExporter logRecordExporter,
70-
MeterProvider meterProvider,
66+
Supplier<MeterProvider> meterProvider,
67+
InternalTelemetryVersion telemetryVersion,
7168
long scheduleDelayNanos,
7269
int maxQueueSize,
7370
int maxExportBatchSize,
@@ -76,10 +73,12 @@ public static BatchLogRecordProcessorBuilder builder(LogRecordExporter logRecord
7673
new Worker(
7774
logRecordExporter,
7875
meterProvider,
76+
telemetryVersion,
7977
scheduleDelayNanos,
8078
maxExportBatchSize,
8179
exporterTimeoutNanos,
82-
new ArrayBlockingQueue<>(maxQueueSize)); // TODO: use JcTools.newFixedSizeQueue(..)
80+
new ArrayBlockingQueue<>(maxQueueSize),
81+
maxQueueSize); // TODO: use JcTools.newFixedSizeQueue(..)
8382
Thread workerThread = new DaemonThreadFactory(WORKER_THREAD_NAME).newThread(worker);
8483
workerThread.start();
8584
}
@@ -140,9 +139,7 @@ private static final class Worker implements Runnable {
140139

141140
private static final Logger logger = Logger.getLogger(Worker.class.getName());
142141

143-
private final LongCounter processedLogsCounter;
144-
private final Attributes droppedAttrs;
145-
private final Attributes exportedAttrs;
142+
private final LogRecordProcessorInstrumentation logProcessorInstrumentation;
146143

147144
private final LogRecordExporter logRecordExporter;
148145
private final long scheduleDelayNanos;
@@ -163,59 +160,34 @@ private static final class Worker implements Runnable {
163160
private final AtomicReference<CompletableResultCode> flushRequested = new AtomicReference<>();
164161
private volatile boolean continueWork = true;
165162
private final ArrayList<LogRecordData> batch;
163+
private final long maxQueueSize;
166164

167165
private Worker(
168166
LogRecordExporter logRecordExporter,
169-
MeterProvider meterProvider,
167+
Supplier<MeterProvider> meterProvider,
168+
InternalTelemetryVersion telemetryVersion,
170169
long scheduleDelayNanos,
171170
int maxExportBatchSize,
172171
long exporterTimeoutNanos,
173-
Queue<ReadWriteLogRecord> queue) {
172+
Queue<ReadWriteLogRecord> queue,
173+
long maxQueueSize) {
174174
this.logRecordExporter = logRecordExporter;
175175
this.scheduleDelayNanos = scheduleDelayNanos;
176176
this.maxExportBatchSize = maxExportBatchSize;
177177
this.exporterTimeoutNanos = exporterTimeoutNanos;
178178
this.queue = queue;
179179
this.signal = new ArrayBlockingQueue<>(1);
180-
Meter meter = meterProvider.meterBuilder("io.opentelemetry.sdk.logs").build();
181-
meter
182-
.gaugeBuilder("queueSize")
183-
.ofLongs()
184-
.setDescription("The number of items queued")
185-
.setUnit("1")
186-
.buildWithCallback(
187-
result ->
188-
result.record(
189-
queue.size(),
190-
Attributes.of(
191-
LOG_RECORD_PROCESSOR_TYPE_LABEL, LOG_RECORD_PROCESSOR_TYPE_VALUE)));
192-
processedLogsCounter =
193-
meter
194-
.counterBuilder("processedLogs")
195-
.setUnit("1")
196-
.setDescription(
197-
"The number of logs processed by the BatchLogRecordProcessor. "
198-
+ "[dropped=true if they were dropped due to high throughput]")
199-
.build();
200-
droppedAttrs =
201-
Attributes.of(
202-
LOG_RECORD_PROCESSOR_TYPE_LABEL,
203-
LOG_RECORD_PROCESSOR_TYPE_VALUE,
204-
LOG_RECORD_PROCESSOR_DROPPED_LABEL,
205-
true);
206-
exportedAttrs =
207-
Attributes.of(
208-
LOG_RECORD_PROCESSOR_TYPE_LABEL,
209-
LOG_RECORD_PROCESSOR_TYPE_VALUE,
210-
LOG_RECORD_PROCESSOR_DROPPED_LABEL,
211-
false);
180+
logProcessorInstrumentation =
181+
LogRecordProcessorInstrumentation.get(telemetryVersion, COMPONENT_ID, meterProvider);
182+
this.maxQueueSize = maxQueueSize;
212183

213184
this.batch = new ArrayList<>(this.maxExportBatchSize);
214185
}
215186

216187
private void addLog(ReadWriteLogRecord logData) {
188+
logProcessorInstrumentation.buildQueueMetricsOnce(maxQueueSize, queue::size);
217189
if (!queue.offer(logData)) {
218-
processedLogsCounter.add(1, droppedAttrs);
190+
logProcessorInstrumentation.dropLogs(1);
219191
} else {
220192
if (queue.size() >= logsNeeded.get()) {
221193
signal.offer(true);
@@ -316,18 +288,24 @@ private void exportCurrentBatch() {
316288
return;
317289
}
318290

291+
String error = null;
319292
try {
320293
CompletableResultCode result =
321294
logRecordExporter.export(Collections.unmodifiableList(batch));
322295
result.join(exporterTimeoutNanos, TimeUnit.NANOSECONDS);
323-
if (result.isSuccess()) {
324-
processedLogsCounter.add(batch.size(), exportedAttrs);
325-
} else {
296+
if (!result.isSuccess()) {
326297
logger.log(Level.FINE, "Exporter failed");
298+
if (result.getFailureThrowable() != null) {
299+
error = result.getFailureThrowable().getClass().getName();
300+
} else {
301+
error = "export_failed";
302+
}
327303
}
328304
} catch (RuntimeException e) {
329305
logger.log(Level.WARNING, "Exporter threw an Exception", e);
306+
error = e.getClass().getName();
330307
} finally {
308+
logProcessorInstrumentation.finishLogs(batch.size(), error);
331309
batch.clear();
332310
}
333311
}

0 commit comments

Comments
 (0)