Skip to content

Record processed logs before export complete and reject on shutdown #8698

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We鈥檒l occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
anuraaga wants to merge 3 commits into
base: main
Choose a base branch
from
Open

Record processed logs before export complete and reject on shutdown #8698

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
ad55311
Record processed logs before export complete and reject on shutdown
anuraaga Aug 6, 2026
4a54c28
grammar
anuraaga Aug 6, 2026
85a7b3f
cleanup
anuraaga Aug 7, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 12 additions & 13 deletions sdk/logs/src/main/java/io/opentelemetry/sdk/logs/export/BatchLogRecordProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,6 @@ public final class BatchLogRecordProcessor implements LogRecordProcessor {
BatchLogRecordProcessor.class.getSimpleName() + "_WorkerThread";

private final Worker worker;
private final AtomicBoolean isShutdown = new AtomicBoolean(false);

/**
* Returns a new Builder for {@link BatchLogRecordProcessor}.
Expand Down Expand Up @@ -94,9 +93,6 @@ public void onEmit(Context context, ReadWriteLogRecord logRecord) {

@Override
public CompletableResultCode shutdown() {
if (isShutdown.getAndSet(true)) {
return CompletableResultCode.ofSuccess();
}
return worker.shutdown();
}

Expand Down Expand Up @@ -159,6 +155,7 @@ private static final class Worker implements Runnable {
private final AtomicInteger logsNeeded = new AtomicInteger(Integer.MAX_VALUE);
private final BlockingQueue<Boolean> signal;
private final AtomicReference<CompletableResultCode> flushRequested = new AtomicReference<>();
private final AtomicBoolean isShutdown = new AtomicBoolean(false);

@jack-berg jack-berg Aug 6, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The BatchLogRecordProcessor also has an isShutdown field. Since we need it down in the Worker for instrumentation, can we get rid of it in the parent and just delegate BatchLogRecordProcessor#shutdown to Worker#shutdown()?

private volatile boolean continueWork = true;
private final ArrayList<LogRecordData> batch;
private final long maxQueueSize;
Expand Down Expand Up @@ -186,9 +183,13 @@ private Worker(
}

private void addLog(ReadWriteLogRecord logData) {
if (isShutdown.get()) {
logProcessorInstrumentation.dropLogsAlreadyShutdown(1);
return;
}
logProcessorInstrumentation.buildQueueMetricsOnce(maxQueueSize, queue::size);
if (!queue.offer(logData)) {
logProcessorInstrumentation.dropLogs(1);
logProcessorInstrumentation.dropLogsQueueFull(1);
} else {
if (queue.size() >= logsNeeded.get()) {
signal.offer(true);
Expand Down Expand Up @@ -251,6 +252,9 @@ private void updateNextExportTime() {
}

private CompletableResultCode shutdown() {
if (isShutdown.getAndSet(true)) {
return CompletableResultCode.ofSuccess();
}
CompletableResultCode result = new CompletableResultCode();

CompletableResultCode flushResult = forceFlush();
Expand Down Expand Up @@ -289,25 +293,20 @@ private void exportCurrentBatch() {
return;
}

String error = null;
try {
// We always increment for every export invocation, so we increment before the export call
// to make sure thrown errors don't affect it.
logProcessorInstrumentation.finishLogs(batch.size());
CompletableResultCode result =
logRecordExporter.export(Collections.unmodifiableList(batch));
result.join(exporterTimeoutNanos, TimeUnit.NANOSECONDS);
if (!result.isSuccess()) {
logger.log(Level.FINE, "Exporter failed");
if (result.getFailureThrowable() != null) {
error = result.getFailureThrowable().getClass().getName();
} else {
error = "export_failed";
}
}
} catch (Throwable t) {
ThrowableUtil.propagateIfFatal(t);
logger.log(Level.WARNING, "Exporter threw an Exception", t);
error = t.getClass().getName();
} finally {
logProcessorInstrumentation.finishLogs(batch.size(), error);
batch.clear();
}
}
Expand Down
14 changes: 8 additions & 6 deletions ...c/main/java/io/opentelemetry/sdk/logs/export/LegacyLogRecordProcessorInstrumentation.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,16 +43,18 @@ final class LegacyLogRecordProcessorInstrumentation implements LogRecordProcesso
}

@Override
public void dropLogs(int count) {
public void dropLogsQueueFull(int count) {
processedLogs().add(count, droppedAttrs);
}

@Override
public void finishLogs(int count, @Nullable String error) {
// Legacy metrics only record when no error.
if (error == null) {
processedLogs().add(count, standardAttrs);
}
public void dropLogsAlreadyShutdown(int count) {
// Legacy did not record this metric.
}

@Override
public void finishLogs(int count) {
processedLogs().add(count, standardAttrs);
}

@Override
Expand Down
10 changes: 6 additions & 4 deletions ...ogs/src/main/java/io/opentelemetry/sdk/logs/export/LogRecordProcessorInstrumentation.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,6 @@
import io.opentelemetry.sdk.common.InternalTelemetryVersion;
import io.opentelemetry.sdk.common.internal.ComponentId;
import java.util.function.Supplier;
import javax.annotation.Nullable;

/** Metrics exported by span processors. */
interface LogRecordProcessorInstrumentation {
Expand All @@ -27,10 +26,13 @@ static LogRecordProcessorInstrumentation get(
}

/** Records metrics for logs dropped because a queue is full. */
void dropLogs(int count);
void dropLogsQueueFull(int count);

/** Record metrics for logs processed, possibly with an error. */
void finishLogs(int count, @Nullable String error);
/** Record metrics for logs dropped since processor is shutdown. */
void dropLogsAlreadyShutdown(int count);

/** Record metrics for logs processed successfully. */
void finishLogs(int count);

/** Registers metrics for processor queue capacity and size. */
void buildQueueMetricsOnce(long capacity, LongCallable getSize);
Expand Down
31 changes: 19 additions & 12 deletions .../main/java/io/opentelemetry/sdk/logs/export/SemConvLogRecordProcessorInstrumentation.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,8 @@ final class SemConvLogRecordProcessorInstrumentation implements LogRecordProcess

private final Supplier<MeterProvider> meterProvider;
private final Attributes standardAttrs;
private final Attributes droppedAttrs;
private final Attributes queueFullAttrs;
private final Attributes shutdownAttrs;

@Nullable private Meter meter;
@Nullable private volatile LongCounter processedLogs;
Expand All @@ -42,31 +43,37 @@ final class SemConvLogRecordProcessorInstrumentation implements LogRecordProcess
componentId.getTypeName(),
SemConvAttributes.OTEL_COMPONENT_NAME,
componentId.getComponentName());
droppedAttrs =
queueFullAttrs =
Attributes.of(
SemConvAttributes.OTEL_COMPONENT_TYPE,
componentId.getTypeName(),
SemConvAttributes.OTEL_COMPONENT_NAME,
componentId.getComponentName(),
SemConvAttributes.ERROR_TYPE,
"queue_full");
shutdownAttrs =
Attributes.of(
SemConvAttributes.OTEL_COMPONENT_TYPE,
componentId.getTypeName(),
SemConvAttributes.OTEL_COMPONENT_NAME,
componentId.getComponentName(),
SemConvAttributes.ERROR_TYPE,
"already_shutdown");
}

@Override
public void dropLogs(int count) {
processedLogs().add(count, droppedAttrs);
public void dropLogsQueueFull(int count) {
processedLogs().add(count, queueFullAttrs);
}

@Override
public void finishLogs(int count, @Nullable String error) {
if (error == null) {
processedLogs().add(count, standardAttrs);
return;
}
public void dropLogsAlreadyShutdown(int count) {
processedLogs().add(count, shutdownAttrs);
}

Attributes attributes =
standardAttrs.toBuilder().put(SemConvAttributes.ERROR_TYPE, error).build();
processedLogs().add(count, attributes);
@Override
public void finishLogs(int count) {
processedLogs().add(count, standardAttrs);
}

@Override
Expand Down
15 changes: 8 additions & 7 deletions sdk/logs/src/main/java/io/opentelemetry/sdk/logs/export/SimpleLogRecordProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -85,28 +85,29 @@ public static SimpleLogRecordProcessorBuilder builder(LogRecordExporter exporter

@Override
public void onEmit(Context context, ReadWriteLogRecord logRecord) {
if (isShutdown.get()) {
logProcessorInstrumentation.dropLogsAlreadyShutdown(1);
return;
}

try {
List<LogRecordData> logs = Collections.singletonList(logRecord.toLogRecordData());
CompletableResultCode result;

synchronized (exporterLock) {
// We always increment for every export invocation, so we increment before the export call
// to make sure thrown errors don't affect it.
logProcessorInstrumentation.finishLogs(1);

@anuraaga anuraaga Aug 6, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I put it before the export call to match what languages with synchronous export like Go and Python must be doing

@jack-berg jack-berg Aug 6, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Calling it before does also mitigate against the risk of export throwing synchronously.

Probably worth a comment since the placement seems unintuitive at first glance.

result = logRecordExporter.export(logs);
}

pendingExports.add(result);
result.whenComplete(
() -> {
pendingExports.remove(result);
String error = null;
if (!result.isSuccess()) {
logger.log(Level.FINE, "Exporter failed");
if (result.getFailureThrowable() != null) {
error = result.getFailureThrowable().getClass().getName();
} else {
error = "export_failed";
}
}
logProcessorInstrumentation.finishLogs(1, error);
});
} catch (RuntimeException e) {
logger.log(Level.WARNING, "Exporter threw an Exception", e);
Expand Down
Loading
Loading