From fc85042adb5f17676ec2a9299d7f3138a0d92adb Mon Sep 17 00:00:00 2001 From: labkey-jeckels Date: Fri, 28 Aug 2026 15:17:06 -0700 Subject: [PATCH 1/3] Correctly associate async queries with the HTTP request for Datadog --- .../labkey/api/data/AsyncQueryRequest.java | 33 +++++++++++++------ .../api/data/MaterializedQueryHelper.java | 33 ++++++++++++++++++- .../org/labkey/api/data/TableSelector.java | 17 ++++++++-- 3 files changed, 70 insertions(+), 13 deletions(-) diff --git a/api/src/org/labkey/api/data/AsyncQueryRequest.java b/api/src/org/labkey/api/data/AsyncQueryRequest.java index 1130a6d0e1d..3838e98711f 100644 --- a/api/src/org/labkey/api/data/AsyncQueryRequest.java +++ b/api/src/org/labkey/api/data/AsyncQueryRequest.java @@ -17,9 +17,12 @@ package org.labkey.api.data; import datadog.trace.api.CorrelationIdentifier; +import datadog.trace.api.DDTags; import io.opentracing.Scope; import io.opentracing.Span; import io.opentracing.Tracer; +import io.opentracing.log.Fields; +import io.opentracing.tag.Tags; import io.opentracing.util.GlobalTracer; import org.apache.logging.log4j.Logger; import org.apache.logging.log4j.ThreadContext; @@ -39,6 +42,7 @@ import java.io.IOException; import java.sql.SQLException; import java.sql.Statement; +import java.util.Map; import java.util.concurrent.Callable; public class AsyncQueryRequest @@ -52,16 +56,20 @@ private static class CancelledException extends RuntimeException @Nullable private final StackTraceElement[] _creationStackTrace; private final HttpServletResponse _rootResponse; + /** Names the APM span; without it every async query aggregates under the bare operation name */ + private final @Nullable String _resourceName; boolean _cancelled; @Nullable Statement _statement; + @Nullable Span _span; T _result; Throwable _exception; - public AsyncQueryRequest(HttpServletResponse response) + public AsyncQueryRequest(HttpServletResponse response, @Nullable String resourceName) { _creationStackTrace = MiniProfiler.getTroubleshootingStackTrace(); + _resourceName = resourceName; _rootResponse = getRootResponse(response); } @@ -103,29 +111,33 @@ synchronized public T waitForResult(final Callable callable) throws SQLExcept final Object state = qs.cloneEnvironment(); final RequestInfo current = MemTracker.getInstance().current(); - String traceId = CorrelationIdentifier.getTraceId(); - // Create the span, so we can connect the async query with its owning thread final Tracer tracer = GlobalTracer.get(); final Span span = tracer.buildSpan("AsyncRequest").start(); - String spanId = CorrelationIdentifier.getSpanId(); + if (_resourceName != null) + span.setTag(DDTags.RESOURCE_NAME, _resourceName); + _span = span; Runnable runnable = () -> { if (current != null) MemTracker.get().startProfiler("async query"); assert ThreadContext.isEmpty(); // Prevent/detect leaks - // Connect log messages with the active trace and span - ThreadContext.put(CorrelationIdentifier.getTraceIdKey(), traceId); - ThreadContext.put(CorrelationIdentifier.getSpanIdKey(), spanId); qs.copyEnvironment(state); - try + // Activate on this thread, not the caller's, or the JDBC integration starts a new trace instead of parenting here + try (Scope ignored = tracer.activateSpan(span)) { + // Connect log messages with the active trace and span + ThreadContext.put(CorrelationIdentifier.getTraceIdKey(), CorrelationIdentifier.getTraceId()); + ThreadContext.put(CorrelationIdentifier.getSpanIdKey(), CorrelationIdentifier.getSpanId()); + setResult(callable.call()); } catch (Throwable t) { + Tags.ERROR.set(span, true); + span.log(Map.of(Fields.ERROR_OBJECT, t)); setException(t); } finally @@ -150,8 +162,7 @@ synchronized public T waitForResult(final Callable callable) throws SQLExcept Thread thread = new Thread(runnable, threadName); // We want the async thread to use the same database connection, in case we have a transaction open, and // so that when the original thread finishes processing the results it ends up closing the right connection - try (DbScope.ConnectionSharingCloseable ignored = DbScope.shareConnections(Thread.currentThread(), thread); - Scope ignore = tracer.activateSpan(span)) + try (DbScope.ConnectionSharingCloseable ignored = DbScope.shareConnections(Thread.currentThread(), thread)) { thread.start(); @@ -237,6 +248,8 @@ synchronized public void setException(Throwable exception) synchronized private void cancel() { _cancelled = true; + if (_span != null) + _span.setTag("labkey.async_query.cancelled", true); if (_statement != null) { try diff --git a/api/src/org/labkey/api/data/MaterializedQueryHelper.java b/api/src/org/labkey/api/data/MaterializedQueryHelper.java index 226abff3031..a30716d449f 100644 --- a/api/src/org/labkey/api/data/MaterializedQueryHelper.java +++ b/api/src/org/labkey/api/data/MaterializedQueryHelper.java @@ -15,9 +15,18 @@ */ package org.labkey.api.data; +import datadog.trace.api.CorrelationIdentifier; +import datadog.trace.api.DDTags; +import io.opentracing.Scope; +import io.opentracing.Span; +import io.opentracing.Tracer; +import io.opentracing.log.Fields; +import io.opentracing.tag.Tags; +import io.opentracing.util.GlobalTracer; import org.apache.commons.lang3.StringUtils; import org.apache.commons.lang3.Strings; import org.apache.logging.log4j.Logger; +import org.apache.logging.log4j.ThreadContext; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; import org.junit.After; @@ -422,17 +431,39 @@ public void materializeAsync() { if (_backgroundTaskRunning.compareAndSet(false, true)) { + String triggeringTraceId = CorrelationIdentifier.getTraceId(); + String triggeringSpanId = CorrelationIdentifier.getSpanId(); + _materializationRunner.execute(() -> { - try + // Materialization outlives the request that triggers it and then serves every later request, so it gets its own trace; the trigger is recorded as tags rather than as a parent + Tracer tracer = GlobalTracer.get(); + Span span = tracer.buildSpan("MaterializeAsync").ignoreActiveSpan().start(); + span.setTag(DDTags.RESOURCE_NAME, StringUtils.defaultIfEmpty(_prefix, getClass().getSimpleName())); + if (!"0".equals(triggeringTraceId)) + { + span.setTag("labkey.triggering_trace_id", triggeringTraceId); + span.setTag("labkey.triggering_span_id", triggeringSpanId); + } + + try (Scope ignored = tracer.activateSpan(span)) { + // Connect log messages with the active trace and span + ThreadContext.put(CorrelationIdentifier.getTraceIdKey(), CorrelationIdentifier.getTraceId()); + ThreadContext.put(CorrelationIdentifier.getSpanIdKey(), CorrelationIdentifier.getSpanId()); + getFromSql("_bg_"); } catch (Exception e) { + Tags.ERROR.set(span, true); + span.log(Map.of(Fields.ERROR_OBJECT, e)); LOG.warn("Background materialization failed.", e); } finally { + span.finish(); + ThreadContext.remove(CorrelationIdentifier.getTraceIdKey()); + ThreadContext.remove(CorrelationIdentifier.getSpanIdKey()); _backgroundTaskRunning.set(false); } }); diff --git a/api/src/org/labkey/api/data/TableSelector.java b/api/src/org/labkey/api/data/TableSelector.java index e3284dec54d..cf873ba6d27 100644 --- a/api/src/org/labkey/api/data/TableSelector.java +++ b/api/src/org/labkey/api/data/TableSelector.java @@ -379,10 +379,23 @@ public Results getResults(boolean cache, boolean scrollable) return new ResultsImpl(rs, tableSqlFactory.getSelectedColumns()); } + /** Names the APM span for an async query. Public schema/query name when there is one, otherwise the DB table. */ + private String getAsyncResourceName(String operation) + { + String schema = _table.getPublicSchemaName(); + String name = _table.getPublicName(); + if (null == schema || null == name) + { + schema = null != _table.getSchema() ? _table.getSchema().getName() : null; + name = _table.getName(); + } + return operation + " " + (null != schema ? schema + "." : "") + name; + } + public Results getResultsAsync(final boolean cache, final boolean scrollable, HttpServletResponse response) throws SQLException { setLogger(ConnectionWrapper.getConnectionLogger()); - AsyncQueryRequest asyncRequest = new AsyncQueryRequest<>(response); + AsyncQueryRequest asyncRequest = new AsyncQueryRequest<>(response, getAsyncResourceName("getResults")); setAsyncRequest(asyncRequest); try @@ -554,7 +567,7 @@ public Map> getAggregates(final List aggregates) public Map> getAggregatesAsync(final List aggregates, HttpServletResponse response) { setLogger(ConnectionWrapper.getConnectionLogger()); - AsyncQueryRequest>> asyncRequest = new AsyncQueryRequest<>(response); + AsyncQueryRequest>> asyncRequest = new AsyncQueryRequest<>(response, getAsyncResourceName("getAggregates")); setAsyncRequest(asyncRequest); try From 5850067f1e39d4efb3715203e530bd086477c23b Mon Sep 17 00:00:00 2001 From: labkey-jeckels Date: Fri, 28 Aug 2026 16:13:33 -0700 Subject: [PATCH 2/3] Code review --- .../labkey/api/data/AsyncQueryRequest.java | 28 +++++++++++++++++-- .../api/data/MaterializedQueryHelper.java | 9 ++++-- .../org/labkey/api/data/TableSelector.java | 23 +++++++++++---- 3 files changed, 49 insertions(+), 11 deletions(-) diff --git a/api/src/org/labkey/api/data/AsyncQueryRequest.java b/api/src/org/labkey/api/data/AsyncQueryRequest.java index 3838e98711f..4a3927a1fcc 100644 --- a/api/src/org/labkey/api/data/AsyncQueryRequest.java +++ b/api/src/org/labkey/api/data/AsyncQueryRequest.java @@ -26,6 +26,7 @@ import io.opentracing.util.GlobalTracer; import org.apache.logging.log4j.Logger; import org.apache.logging.log4j.ThreadContext; +import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; import org.labkey.api.miniprofiler.MiniProfiler; import org.labkey.api.miniprofiler.RequestInfo; @@ -58,6 +59,8 @@ private static class CancelledException extends RuntimeException private final HttpServletResponse _rootResponse; /** Names the APM span; without it every async query aggregates under the bare operation name */ private final @Nullable String _resourceName; + /** Extra tags for APM search and grouping, beyond the resource name */ + private final @NotNull Map _spanTags; boolean _cancelled; @Nullable Statement _statement; @@ -66,10 +69,11 @@ private static class CancelledException extends RuntimeException T _result; Throwable _exception; - public AsyncQueryRequest(HttpServletResponse response, @Nullable String resourceName) + public AsyncQueryRequest(HttpServletResponse response, @Nullable String resourceName, @NotNull Map spanTags) { _creationStackTrace = MiniProfiler.getTroubleshootingStackTrace(); _resourceName = resourceName; + _spanTags = spanTags; _rootResponse = getRootResponse(response); } @@ -116,6 +120,7 @@ synchronized public T waitForResult(final Callable callable) throws SQLExcept final Span span = tracer.buildSpan("AsyncRequest").start(); if (_resourceName != null) span.setTag(DDTags.RESOURCE_NAME, _resourceName); + _spanTags.forEach(span::setTag); _span = span; Runnable runnable = () -> { @@ -136,8 +141,12 @@ synchronized public T waitForResult(final Callable callable) throws SQLExcept } catch (Throwable t) { - Tags.ERROR.set(span, true); - span.log(Map.of(Fields.ERROR_OBJECT, t)); + // Cancellation arrives here as a CancelledException or the driver's "statement cancelled" error; tagging that as an APM error makes every client disconnect look like a failure + if (!isCancelled()) + { + Tags.ERROR.set(span, true); + span.log(Map.of(Fields.ERROR_OBJECT, t)); + } setException(t); } finally @@ -162,9 +171,11 @@ synchronized public T waitForResult(final Callable callable) throws SQLExcept Thread thread = new Thread(runnable, threadName); // We want the async thread to use the same database connection, in case we have a transaction open, and // so that when the original thread finishes processing the results it ends up closing the right connection + boolean started = false; try (DbScope.ConnectionSharingCloseable ignored = DbScope.shareConnections(Thread.currentThread(), thread)) { thread.start(); + started = true; // Stash the original disconnect exception if the client has dropped off IOException clientDisconnectException = null; @@ -220,6 +231,12 @@ synchronized public T waitForResult(final Callable callable) throws SQLExcept } } } + finally + { + // The runnable finishes the span; without a thread to run it the span stays open and its parent trace never completes + if (!started) + span.finish(); + } } synchronized public void setResult(T result) @@ -245,6 +262,11 @@ synchronized public void setException(Throwable exception) notify(); } + synchronized private boolean isCancelled() + { + return _cancelled; + } + synchronized private void cancel() { _cancelled = true; diff --git a/api/src/org/labkey/api/data/MaterializedQueryHelper.java b/api/src/org/labkey/api/data/MaterializedQueryHelper.java index a30716d449f..1d8083d86cf 100644 --- a/api/src/org/labkey/api/data/MaterializedQueryHelper.java +++ b/api/src/org/labkey/api/data/MaterializedQueryHelper.java @@ -453,11 +453,14 @@ public void materializeAsync() getFromSql("_bg_"); } - catch (Exception e) + catch (Throwable t) { Tags.ERROR.set(span, true); - span.log(Map.of(Fields.ERROR_OBJECT, e)); - LOG.warn("Background materialization failed.", e); + span.log(Map.of(Fields.ERROR_OBJECT, t)); + LOG.warn("Background materialization failed.", t); + // Broad enough to tag an Error on the span, but only Exceptions are swallowed + if (t instanceof Error e) + throw e; } finally { diff --git a/api/src/org/labkey/api/data/TableSelector.java b/api/src/org/labkey/api/data/TableSelector.java index cf873ba6d27..60cced9da13 100644 --- a/api/src/org/labkey/api/data/TableSelector.java +++ b/api/src/org/labkey/api/data/TableSelector.java @@ -36,6 +36,7 @@ import java.util.ArrayList; import java.util.Collection; import java.util.Collections; +import java.util.HashMap; import java.util.Iterator; import java.util.LinkedHashMap; import java.util.LinkedHashSet; @@ -379,8 +380,8 @@ public Results getResults(boolean cache, boolean scrollable) return new ResultsImpl(rs, tableSqlFactory.getSelectedColumns()); } - /** Names the APM span for an async query. Public schema/query name when there is one, otherwise the DB table. */ - private String getAsyncResourceName(String operation) + /** @return "schema.query", using the public (Query) names when the table has them, otherwise the DB schema and table */ + private String getAsyncQueryName() { String schema = _table.getPublicSchemaName(); String name = _table.getPublicName(); @@ -389,13 +390,24 @@ private String getAsyncResourceName(String operation) schema = null != _table.getSchema() ? _table.getSchema().getName() : null; name = _table.getName(); } - return operation + " " + (null != schema ? schema + "." : "") + name; + return (null != schema ? schema + "." : "") + name; + } + + /** Extra APM span tags. resource.name concatenates the operation, so these are what let one query be grouped across getResults and getAggregates. */ + private Map getAsyncSpanTags(String queryName) + { + Map tags = new HashMap<>(); + tags.put("labkey.query", queryName); + if (null != _table.getSchema()) + tags.put("labkey.db_schema", _table.getSchema().getName()); + return tags; } public Results getResultsAsync(final boolean cache, final boolean scrollable, HttpServletResponse response) throws SQLException { setLogger(ConnectionWrapper.getConnectionLogger()); - AsyncQueryRequest asyncRequest = new AsyncQueryRequest<>(response, getAsyncResourceName("getResults")); + String queryName = getAsyncQueryName(); + AsyncQueryRequest asyncRequest = new AsyncQueryRequest<>(response, "getResults " + queryName, getAsyncSpanTags(queryName)); setAsyncRequest(asyncRequest); try @@ -567,7 +579,8 @@ public Map> getAggregates(final List aggregates) public Map> getAggregatesAsync(final List aggregates, HttpServletResponse response) { setLogger(ConnectionWrapper.getConnectionLogger()); - AsyncQueryRequest>> asyncRequest = new AsyncQueryRequest<>(response, getAsyncResourceName("getAggregates")); + String queryName = getAsyncQueryName(); + AsyncQueryRequest>> asyncRequest = new AsyncQueryRequest<>(response, "getAggregates " + queryName, getAsyncSpanTags(queryName)); setAsyncRequest(asyncRequest); try From 0eae3d3bbdfb5b12b67d107fcd4552f7a96457e5 Mon Sep 17 00:00:00 2001 From: labkey-jeckels Date: Fri, 28 Aug 2026 17:22:36 -0700 Subject: [PATCH 3/3] Better materialization tracing --- .../api/data/MaterializedQueryHelper.java | 82 ++++++++++++++----- .../experiment/api/ExpMaterialTableImpl.java | 18 ++-- 2 files changed, 72 insertions(+), 28 deletions(-) diff --git a/api/src/org/labkey/api/data/MaterializedQueryHelper.java b/api/src/org/labkey/api/data/MaterializedQueryHelper.java index 1d8083d86cf..69ad2371b4f 100644 --- a/api/src/org/labkey/api/data/MaterializedQueryHelper.java +++ b/api/src/org/labkey/api/data/MaterializedQueryHelper.java @@ -162,31 +162,33 @@ boolean load(SQLFragment selectQuery, boolean isSelectInto) DbSchema temp = DbSchema.getTemp(); TempTableTracker.track(_tableName, this); - SQLFragment selectInto; - if (isSelectInto) - { - String sql = selectQuery.getSQL().replace("${NAME}", _tableName); - List params = selectQuery.getParams(); - selectInto = new SQLFragment(sql,params); - } - else - { - // UNLOGGED skips WAL when populating and indexing the table; only supported in PostgreSQL. - selectInto = new SQLFragment("SELECT * INTO ") - .append(_mqh._unlogged && _mqh._scope.getSqlDialect().isPostgreSQL() ? "UNLOGGED " : "") - .appendIdentifier(temp.getName()).append(".").appendIdentifier(_tableName).append("\nFROM (\n"); - selectInto.append(selectQuery); - selectInto.append("\n) _sql_"); - } - new SqlExecutor(_mqh._scope).execute(selectInto); + traced("full", _mqh.getMaterializationName(), () -> { + SQLFragment selectInto; + if (isSelectInto) + { + String sql = selectQuery.getSQL().replace("${NAME}", _tableName); + List params = selectQuery.getParams(); + selectInto = new SQLFragment(sql,params); + } + else + { + // UNLOGGED skips WAL when populating and indexing the table; only supported in PostgreSQL. + selectInto = new SQLFragment("SELECT * INTO ") + .append(_mqh._unlogged && _mqh._scope.getSqlDialect().isPostgreSQL() ? "UNLOGGED " : "") + .appendIdentifier(temp.getName()).append(".").appendIdentifier(_tableName).append("\nFROM (\n"); + selectInto.append(selectQuery); + selectInto.append("\n) _sql_"); + } + new SqlExecutor(_mqh._scope).execute(selectInto); - try (var ignored = SpringActionController.ignoreSqlUpdates()) - { - for (String index : _mqh._indexes) + try (var ignored = SpringActionController.ignoreSqlUpdates()) { - new SqlExecutor(_mqh._scope).execute(StringUtils.replace(index, "${NAME}", _tableName)); + for (String index : _mqh._indexes) + { + new SqlExecutor(_mqh._scope).execute(StringUtils.replace(index, "${NAME}", _tableName)); + } } - } + }); _loadingState.set(LoadingState.LOADED); return true; @@ -439,6 +441,7 @@ public void materializeAsync() Tracer tracer = GlobalTracer.get(); Span span = tracer.buildSpan("MaterializeAsync").ignoreActiveSpan().start(); span.setTag(DDTags.RESOURCE_NAME, StringUtils.defaultIfEmpty(_prefix, getClass().getSimpleName())); + span.setTag("labkey.materialized_view", getMaterializationName()); if (!"0".equals(triggeringTraceId)) { span.setTag("labkey.triggering_trace_id", triggeringTraceId); @@ -586,6 +589,41 @@ protected void incrementalUpdateBeforeSelect(Materialized m) { } + /** + * Runs DB work inside its own Datadog APM span so each kind of materialization is a separate resource. Nests under + * MaterializeAsync on the background thread and under the HTTP request when a stale view is rebuilt inline. + */ + protected static void traced(String resource, String viewName, Runnable work) + { + Tracer tracer = GlobalTracer.get(); + Span span = tracer.buildSpan("labkey.materialize").start(); + span.setTag(DDTags.RESOURCE_NAME, resource); + // Never a service-entry span, so Datadog computes no hits/duration/error metrics for it without this + span.setTag(DDTags.MEASURED, true); + span.setTag("labkey.materialized_view", viewName); + + try (Scope ignored = tracer.activateSpan(span)) + { + work.run(); + } + catch (Throwable t) + { + Tags.ERROR.set(span, true); + span.log(Map.of(Fields.ERROR_OBJECT, t)); + throw t; + } + finally + { + span.finish(); + } + } + + /** Identifies the view in APM. Carried as a tag, never a resource name, to keep per-view cardinality out of trace metrics. */ + protected String getMaterializationName() + { + return StringUtils.defaultIfEmpty(_prefix, getClass().getSimpleName()); + } + /** * A Materialized represents a particular instance of materialized view (stored in a temp table). * We want to avoid two threads materializing the same view. This is why we synchronize first creating the diff --git a/experiment/src/org/labkey/experiment/api/ExpMaterialTableImpl.java b/experiment/src/org/labkey/experiment/api/ExpMaterialTableImpl.java index 38f71e66222..31468e5eb82 100644 --- a/experiment/src/org/labkey/experiment/api/ExpMaterialTableImpl.java +++ b/experiment/src/org/labkey/experiment/api/ExpMaterialTableImpl.java @@ -1493,10 +1493,10 @@ protected void incrementalUpdateBeforeSelect(Materialized m) if (Materialized.LoadingState.ERROR == materialized._loadingState.get()) throw materialized._loadException; - runIncremental(materialized.incrementalDeleteCheck, this::executeIncrementalDelete); - runIncremental(materialized.incrementalUpdateCheck, this::executeIncrementalUpdate); - runIncremental(materialized.incrementalRollupCheck, this::executeIncrementalRollup); - runIncremental(materialized.incrementalInsertCheck, this::executeIncrementalInsert); + runIncremental("delete", materialized.incrementalDeleteCheck, this::executeIncrementalDelete); + runIncremental("update", materialized.incrementalUpdateCheck, this::executeIncrementalUpdate); + runIncremental("rollup", materialized.incrementalRollupCheck, this::executeIncrementalRollup); + runIncremental("insert", materialized.incrementalInsertCheck, this::executeIncrementalInsert); } catch (RuntimeException|InterruptedException ex) { @@ -1524,15 +1524,21 @@ protected void incrementalUpdateBeforeSelect(Materialized m) * reflects the change, instead of briefly serving a stale materialized view. Must be called while holding the * materialized's loading lock so only one updater runs at a time. */ - private static void runIncremental(MaterializedQueryHelper.SupplierInvalidator check, Runnable work) + private void runIncremental(String kind, MaterializedQueryHelper.SupplierInvalidator check, Runnable work) { if (check.peekValid()) return; String token = check.current(); - work.run(); + traced("incremental." + kind, getMaterializationName(), work); check.markValidAs(token); } + @Override + protected String getMaterializationName() + { + return StringUtils.defaultIfEmpty(Lsid.parse(_lsid).getObjectId(), _lsid); + } + void upsertWithRetry(SQLFragment sql) { // not actually read-only, but we don't want to start an explicit transaction