This is an automated email from the ASF dual-hosted git repository.
wu-sheng pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/skywalking.git
The following commit(s) were added to refs/heads/master by this push:
new 5884dc8bae Support LAL json{} parsing JSON delivered in text log
bodies; fix abortOnFailure wiring and leak-probe timing (#13936)
5884dc8bae is described below
commit 5884dc8bae32d7529b37e0ee788a107ed1348bf5
Author: 吴晟 Wu Sheng <[email protected]>
AuthorDate: Fri Jul 3 08:32:21 2026 +0800
Support LAL json{} parsing JSON delivered in text log bodies; fix
abortOnFailure wiring and leak-probe timing (#13936)
The OTLP log receiver delivers every string body as a text body, even
JSON-shaped ones, while `json {}` read only the JSON body case — so `json {}`
rules on OTLP-fed layers silently aborted. `json {}` now reads the JSON body
first and, when it is empty, tries the text body as JSON. On a successful
text-fallback parse the matching rule's context input is swapped to a copy with
a JSON body, so that rule persists the log with content type `JSON`; the shared
input is never mutated, and ot [...]
The DSL flag was parsed into the rule model but never emitted into the
generated code, so `json {}` / `yaml {}` / `text { regexp }` always used the
built-in default and `abortOnFailure false` had no effect (for all three
parsers, including the regexp case). The compiler now bakes each rule's flag
into the generated parser call (default `true`, as documented, fixed from a
latent `false` model default). A parse failure or regexp non-match that aborts
the log is reported at WARN, rate-li [...]
---
docs/en/changes/changes.md | 1 +
docs/en/concepts-and-designs/lal.md | 6 +
docs/en/setup/backend/log-otlp.md | 2 +
oap-server/analyzer/log-analyzer/CLAUDE.md | 4 +-
.../analyzer/v2/compiler/LALClassGenerator.java | 8 +-
.../log/analyzer/v2/compiler/LALScriptParser.java | 11 +-
.../analyzer/v2/compiler/rt/LalRuntimeHelper.java | 6 +-
.../oap/log/analyzer/v2/dsl/ExecutionContext.java | 10 ++
.../analyzer/v2/dsl/spec/filter/FilterSpec.java | 131 ++++++++++++++++++---
.../v2/dsl/spec/parser/AbstractParserSpec.java | 25 +---
.../dsl/spec/parser/ParseFailureWarnLimiter.java | 65 ++++++++++
.../v2/dsl/spec/parser/TextParserSpec.java | 30 +++--
.../v2/compiler/LALClassGeneratorBasicTest.java | 3 +-
.../v2/compiler/LALScriptExecutionTest.java | 24 ++++
.../feature-cases/execution-basic.data.yaml | 39 ++++++
.../test-lal/feature-cases/execution-basic.yaml | 47 ++++++++
.../oap/server/core/classloader/ClassLoaderGc.java | 116 ++++++++++++------
.../core/classloader/DSLClassLoaderManager.java | 2 +-
18 files changed, 434 insertions(+), 96 deletions(-)
diff --git a/docs/en/changes/changes.md b/docs/en/changes/changes.md
index a766359d48..2bf842f8db 100644
--- a/docs/en/changes/changes.md
+++ b/docs/en/changes/changes.md
@@ -264,6 +264,7 @@
* Add a runtime-rule apply-status query. The cluster main now tracks each
structural apply through a phase machine (`SchemaApplyCoordinator`: pending →
DDL → fencing → rolling-out → applied, with `degraded` for a
committed-but-unconfirmed apply — the cluster schema fence did not confirm
within the timeout, in which case the lagging data-node ids are surfaced as
`fenceLaggards` and dispatch is resumed anyway, or the local commit-tail threw
— and `failed` carrying the specific reason). The [...]
* Push runtime-rule convergence to peers on commit. After a successful
structural apply — and on the `commit_deferred` path, where the DB row is
durable but this node's commit-tail threw — the main broadcasts a
`NotifyApplied` admin-internal RPC so peers reconcile against the
just-persisted DB row immediately, instead of waiting up to one refresh tick
(~30s) to notice it. The fan-out runs off the REST response thread
(fire-and-forget on a daemon executor) so an unreachable peer's per-cal [...]
* Fix BanyanDB peer nodes permanently flooding `<metric> is not registered`,
and a follow-on case where a peer kept translating writes with a stale schema
shape after a runtime-rule reshape, when a node held a live persist worker but
its local `MetadataRegistry` schema cache was missing or stale for that model —
a `withoutSchemaChange` peer apply or a runtime-rule bundled fall-over rebuilt
the dispatch worker but skipped the local-cache populate, and the registry was
insert-only (never e [...]
+* Support LAL `json {}` parsing JSON content delivered in a plain-text log
body. The parser reads the native protocol's JSON body first; when that is
empty, it tries the text body as JSON — e.g. the OTLP log receiver maps every
OTLP string body to a text body, even JSON-shaped ones, so previously-aborting
`json {}` rules on OTLP-fed layers now work without any receiver or protocol
change. On a successful parse from a text body, the matching rule persists the
log as a JSON body with conte [...]
* Fix a v2 MAL `CounterWindow` key collision: `rate()` / `increase()` /
`irate()` keyed each counter's sliding window on the rule's output metric name
(the same for every input metric of a rule) instead of the counter's own name,
so two or more counters that reduce to the same label set after `.sum(...)`
shared one window and computed rates against each other's values — fabricating
non-zero rates from unchanged counters (e.g. the BanyanDB liaison gRPC error
rate read a steady non-zero of [...]
* Fix the v2 MAL Elvis operator `?:` to honor Groovy-falsy semantics. It
compiled to `Optional.ofNullable(primary).orElse(fallback)`, applying the
fallback only when the primary is `null`, so an empty-string primary kept `""`
instead — e.g. a BanyanDB liaison `ServiceInstance` stored `node_type=""`
rather than `n/a`, because `.sum([...,'node_type'])` fills an absent group-by
label with `""`. The fallback now applies for falsy primaries such as null,
false, numeric zero, and empty strings [...]
* SWIP-15: rebuild BanyanDB self-observability around the cluster / container
/ group model (requires BanyanDB 0.11+). A BanyanDB cluster is modeled as one
`Service`, each container as a `ServiceInstance` (role/tier as attributes), and
each storage group as an `Endpoint`. The `otel-rules/banyandb/` rules are
category-separated by role (`node_*` / `liaison_*` / `data_*` / `lifecycle_*`)
and by data type (`measure_*` / `stream_*` / `trace_*` / `property_*`),
mirroring the upstream FODC-pro [...]
diff --git a/docs/en/concepts-and-designs/lal.md
b/docs/en/concepts-and-designs/lal.md
index 74acc3f7e3..d233c10b3c 100644
--- a/docs/en/concepts-and-designs/lal.md
+++ b/docs/en/concepts-and-designs/lal.md
@@ -219,6 +219,12 @@ filter {
}
```
+`json` reads the JSON body of the [native log
protocol](../api/log-data-protocol.md). When the JSON body is empty but a
plain-text body is present — for example, the [OTLP log
receiver](../setup/backend/log-otlp.md) delivers every string body as text,
even JSON-shaped ones — the parser tries the text body as JSON instead. A parse
failure of either form follows `abortOnFailure`.
+
+When `json` parses successfully from a text body, the log is normalized to a
JSON body for the matching rule: the rule persists it with content type `JSON`
and `log.body` reads the JSON content within that rule. This makes a
JSON-shaped log delivered over any transport persist and render as JSON once a
`json {}` rule matches it. The normalization is scoped to the matching rule —
other rules analyzing the same log still see the original text body.
+
+A parse failure that aborts the log is reported at WARN, rate-limited to one
report per minute per parser with the number of suppressed failures included in
the next report. With `abortOnFailure false`, a failure is expected control
flow: it is only logged at DEBUG, the log continues through the filter chain,
and `parsed.*` reads return the metadata fallback fields (or `null`).
+
#### `yaml`
```
diff --git a/docs/en/setup/backend/log-otlp.md
b/docs/en/setup/backend/log-otlp.md
index 77c56daa94..5390adac41 100644
--- a/docs/en/setup/backend/log-otlp.md
+++ b/docs/en/setup/backend/log-otlp.md
@@ -53,3 +53,5 @@ And several attributes are optional as add-on information for
the logs before an
- `service.instance`: the instance name that generates the logs. The default
value is empty.
Note, that these attributes should be set manually through OpenTelemetry SDK
or through [attribute#insert in OpenTelemetry
Collector](https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/main/processor/attributesprocessor/README.md).
+
+The OTLP log record body is delivered to
[LAL](../../concepts-and-designs/lal.md) as a plain-text body, even when the
string content is JSON-shaped. A LAL rule can still parse such content with the
`json {}` parser — when the JSON body is empty and a text body is present,
`json {}` tries the text as JSON. On a successful parse the log is normalized
to a JSON body, so it is stored and rendered as JSON. See
[`json`](../../concepts-and-designs/lal.md#json) for details.
diff --git a/oap-server/analyzer/log-analyzer/CLAUDE.md
b/oap-server/analyzer/log-analyzer/CLAUDE.md
index 74a93836b8..4c352e3a97 100644
--- a/oap-server/analyzer/log-analyzer/CLAUDE.md
+++ b/oap-server/analyzer/log-analyzer/CLAUDE.md
@@ -91,7 +91,7 @@ fields are now handled via the `outputType` mechanism with
`outputFieldStatement
All spec methods take `ExecutionContext ctx` as an explicit parameter — there
is no `BINDING` ThreadLocal or `bind()` method. The `execute()` method receives
`ctx` directly and passes it through:
- `execute(FilterSpec filterSpec, ExecutionContext ctx)` — entry point
-- `filterSpec.json(ctx)`, `filterSpec.text(ctx)`, `filterSpec.sink(ctx)` —
parser/sink calls
+- `filterSpec.json(ctx, true)`, `filterSpec.text(ctx)`, `filterSpec.sink(ctx)`
— parser/sink calls (json/yaml/text-regexp carry the rule's `abortOnFailure`
flag)
- `((OutputType) h.ctx().output()).setService(...)` — standard field setters
on the output builder
- `_e.prepareMetrics(h.ctx())`, `_e.submitMetrics(h.ctx(), _metrics)` —
metrics calls via MetricExtractor
- `_f.sampler().rateLimit(h.ctx(), ...)` — sink calls via `h.ctx()`
@@ -267,7 +267,7 @@ One class is generated (e.g., `default_L3_my_rule` when
`yamlSource=default.yaml
public class default_L3_my_rule implements LalExpression {
public void execute(FilterSpec filterSpec, ExecutionContext ctx) {
LalRuntimeHelper h = new LalRuntimeHelper(ctx);
- filterSpec.json(ctx);
+ filterSpec.json(ctx, true);
if (!ctx.shouldAbort()) {
_extractor(filterSpec.extractor(), h);
}
diff --git
a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/LALClassGenerator.java
b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/LALClassGenerator.java
index 082529d700..4c0cc8d7bc 100644
---
a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/LALClassGenerator.java
+++
b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/LALClassGenerator.java
@@ -757,16 +757,18 @@ public final class LALClassGenerator {
if (tp.getRegexpPattern() != null) {
sb.append(" filterSpec.textWithRegexp(ctx, \"")
.append(LALCodegenHelper.escapeJava(tp.getRegexpPattern()))
- .append("\");\n");
+ .append("\", ").append(tp.isAbortOnFailure()).append(");\n");
} else {
sb.append(" filterSpec.text(ctx);\n");
}
LALCodegenHelper.emitCaptureCall(sb, "Parser", genCtx.ruleName, 0,
"ctx", "");
} else if (stmt instanceof LALScriptModel.JsonParser) {
- sb.append(" filterSpec.json(ctx);\n");
+ sb.append(" filterSpec.json(ctx, ")
+ .append(((LALScriptModel.JsonParser)
stmt).isAbortOnFailure()).append(");\n");
LALCodegenHelper.emitCaptureCall(sb, "Parser", genCtx.ruleName, 0,
"ctx", "");
} else if (stmt instanceof LALScriptModel.YamlParser) {
- sb.append(" filterSpec.yaml(ctx);\n");
+ sb.append(" filterSpec.yaml(ctx, ")
+ .append(((LALScriptModel.YamlParser)
stmt).isAbortOnFailure()).append(");\n");
LALCodegenHelper.emitCaptureCall(sb, "Parser", genCtx.ruleName, 0,
"ctx", "");
} else if (stmt instanceof LALScriptModel.AbortStatement) {
sb.append(" filterSpec.abort(ctx);\n");
diff --git
a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/LALScriptParser.java
b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/LALScriptParser.java
index 38c8942337..3ff9e58f6e 100644
---
a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/LALScriptParser.java
+++
b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/LALScriptParser.java
@@ -151,7 +151,7 @@ public final class LALScriptParser {
private static FilterStatement visitParserBlock(final
LALParser.ParserBlockContext ctx) {
if (ctx.textBlock() != null) {
String pattern = null;
- boolean abortOnFail = false;
+ boolean abortOnFail = true;
if (ctx.textBlock().textContent() != null) {
for (final LALParser.RegexpStatementContext regCtx :
ctx.textBlock().textContent().regexpStatement()) {
@@ -176,13 +176,14 @@ public final class LALScriptParser {
/**
* Extract the {@code abortOnFailure} flag from an optional
- * {@code abortOnFailureStatement} node, defaulting to {@code false}
- * when the statement is absent. Shared between the JSON and YAML
- * parser blocks (text uses a list form, handled inline).
+ * {@code abortOnFailureStatement} node, defaulting to {@code true}
+ * when the statement is absent — the documented default for every parser.
+ * Shared between the JSON and YAML parser blocks (text uses a list form,
+ * handled inline).
*/
private static boolean extractAbortOnFail(
final LALParser.AbortOnFailureStatementContext ctx) {
- return ctx != null && parseBoolText(ctx.boolValue().getText());
+ return ctx == null || parseBoolText(ctx.boolValue().getText());
}
private static boolean parseBoolText(final String text) {
diff --git
a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/rt/LalRuntimeHelper.java
b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/rt/LalRuntimeHelper.java
index c66103cb7d..cfc09195cc 100644
---
a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/rt/LalRuntimeHelper.java
+++
b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/rt/LalRuntimeHelper.java
@@ -29,6 +29,7 @@ import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.List;
import java.util.Map;
+import java.util.regex.Matcher;
import org.apache.commons.lang3.StringUtils;
import org.apache.skywalking.apm.network.common.v3.KeyStringValuePair;
import org.apache.skywalking.apm.network.logging.v3.LogData;
@@ -170,7 +171,10 @@ public final class LalRuntimeHelper {
* </pre>
*/
public String group(final String name) {
- return ctx.parsed().getMatcher().group(name);
+ // Null when the regexp did not match and the rule continued via
abortOnFailure
+ // false — the continuation contract is "parsed reads yield null", not
NPE.
+ final Matcher matcher = ctx.parsed().getMatcher();
+ return matcher == null ? null : matcher.group(name);
}
// ==================== Data source: Log tags ====================
diff --git
a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/dsl/ExecutionContext.java
b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/dsl/ExecutionContext.java
index c47daf3f8e..8e98c5b6a0 100644
---
a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/dsl/ExecutionContext.java
+++
b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/dsl/ExecutionContext.java
@@ -127,6 +127,16 @@ public class ExecutionContext {
return getProperty(KEY_INPUT);
}
+ /**
+ * Replace the raw input for this rule's context. Used by parsers that
normalize the
+ * body (e.g. {@code json {}} rewriting a text body to a JSON body): the
original input
+ * object is shared by every rule analyzing the same log, so normalization
must swap in
+ * a rule-local copy instead of mutating the shared object.
+ */
+ public void input(final Object input) {
+ setProperty(KEY_INPUT, input);
+ }
+
public ExecutionContext parsed(final Matcher parsed) {
parsed().matcher = parsed;
return this;
diff --git
a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/dsl/spec/filter/FilterSpec.java
b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/dsl/spec/filter/FilterSpec.java
index cf265c7a54..e1b95c719f 100644
---
a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/dsl/spec/filter/FilterSpec.java
+++
b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/dsl/spec/filter/FilterSpec.java
@@ -20,13 +20,17 @@ package
org.apache.skywalking.oap.log.analyzer.v2.dsl.spec.filter;
import com.fasterxml.jackson.core.type.TypeReference;
import java.util.Arrays;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import org.apache.skywalking.apm.network.logging.v3.JSONLog;
import org.apache.skywalking.apm.network.logging.v3.LogData;
+import org.apache.skywalking.apm.network.logging.v3.LogDataBody;
import org.apache.skywalking.oap.log.analyzer.v2.dsl.ExecutionContext;
import org.apache.skywalking.oap.log.analyzer.v2.dsl.spec.AbstractSpec;
import
org.apache.skywalking.oap.log.analyzer.v2.dsl.spec.extractor.MetricExtractor;
import
org.apache.skywalking.oap.log.analyzer.v2.dsl.spec.parser.JsonParserSpec;
+import
org.apache.skywalking.oap.log.analyzer.v2.dsl.spec.parser.ParseFailureWarnLimiter;
import
org.apache.skywalking.oap.log.analyzer.v2.dsl.spec.parser.TextParserSpec;
import
org.apache.skywalking.oap.log.analyzer.v2.dsl.spec.parser.YamlParserSpec;
import org.apache.skywalking.oap.log.analyzer.v2.dsl.spec.sink.SamplerSpec;
@@ -65,6 +69,12 @@ public class FilterSpec extends AbstractSpec {
private final TypeReference<Map<String, Object>> parsedType;
+ private final ParseFailureWarnLimiter jsonWarnLimiter =
+ new
ParseFailureWarnLimiter(ParseFailureWarnLimiter.DEFAULT_INTERVAL_MS);
+
+ private final ParseFailureWarnLimiter yamlWarnLimiter =
+ new
ParseFailureWarnLimiter(ParseFailureWarnLimiter.DEFAULT_INTERVAL_MS);
+
public FilterSpec(final ModuleManager moduleManager,
final LogAnalyzerModuleConfig moduleConfig) throws
ModuleStartException {
super(moduleManager, moduleConfig);
@@ -99,61 +109,144 @@ public class FilterSpec extends AbstractSpec {
/**
* LAL {@code text { regexp '...' }} — applies a named-group regexp to the
* log body text. Matched groups are stored in {@code
ctx.parsed().getMatcher()}
- * and accessed via {@code parsed.groupName} in the LAL script.
+ * and accessed via {@code parsed.groupName} in the LAL script. {@code
abortOnFailure}
+ * is the rule's compile-time flag (default {@code true}): a non-matching
body aborts
+ * the filter chain only when it is set. The flag travels as a parameter
rather than
+ * as parser-spec state because one {@code FilterSpec} instance serves
every compiled
+ * rule concurrently.
*/
- public void textWithRegexp(final ExecutionContext ctx, final String
regexp) {
+ public void textWithRegexp(final ExecutionContext ctx, final String regexp,
+ final boolean abortOnFailure) {
if (ctx.shouldAbort()) {
return;
}
- textParser.regexp(ctx, regexp);
+ textParser.regexp(ctx, regexp, abortOnFailure);
}
/**
- * LAL {@code json {}} — parses {@code LogData.body.json.json} into a
- * {@code Map<String, Object>} and stores it in {@code ctx.parsed()}.
- * Metadata fields (service, serviceInstance, endpoint, layer, timestamp)
+ * LAL {@code json {}} — parses the JSON log body into a {@code
Map<String, Object>}
+ * and stores it in {@code ctx.parsed()}. Reads {@code
LogData.body.json.json}; when
+ * that is empty but a text body is present, falls back to parsing
+ * {@code LogData.body.text.text} as JSON — the OTLP log receiver delivers
every
+ * string body (even JSON-shaped ones) as a text body, and a rule that
declared
+ * {@code json {}} means the content is JSON regardless of which body case
carried it.
+ * A genuine parse failure is logged — rate-limited WARN when the failure
aborts the
+ * log, DEBUG when {@code abortOnFailure false} makes the miss expected
control flow —
+ * and then honors {@code abortOnFailure} (the rule's compile-time flag,
default
+ * {@code true}); the flag travels as a parameter rather than as
parser-spec state
+ * because one {@code FilterSpec} instance serves every compiled rule
concurrently.
+ * A typed-proto input (e.g. Envoy ALS {@code HTTPAccessLogEntry}) is not
a parse
+ * failure: shipped rules use {@code json {}} on such layers as a routing
guard, so
+ * the mismatch stays quiet and just honors the flag.
+ *
+ * <p>On a successful parse that fell back to the text body, this rule's
context input
+ * is swapped to a copy whose body is a JSON body carrying the same
content, so the log
+ * this rule persists gets {@code ContentType.JSON} (persistence derives
the stored
+ * content type from the body case in {@code LogBuilder.toLog()}). The
original input
+ * object is shared by every rule analyzing the same log and is never
mutated — other
+ * rules still see the original text body.
+ *
+ * <p>Metadata fields (service, serviceInstance, endpoint, layer,
timestamp)
* are also added to the map via {@code putIfAbsent}, so body values take
* priority while metadata fields serve as fallback — matching v1 Groovy
* {@code Binding.Parsed.getAt(key)} behavior.
*/
- public void json(final ExecutionContext ctx) {
+ public void json(final ExecutionContext ctx, final boolean abortOnFailure)
{
if (ctx.shouldAbort()) {
return;
}
+ final Object rawInput = ctx.input();
+ if (!(rawInput instanceof LogData.Builder)) {
+ abortOrContinueUnparsed(ctx, abortOnFailure);
+ return;
+ }
try {
- final LogData.Builder logData = (LogData.Builder) ctx.input();
- final Map<String, Object> parsed = jsonParser.create().readValue(
- logData.getBody().getJson().getJson(), parsedType
- );
+ final LogData.Builder logData = (LogData.Builder) rawInput;
+ final LogDataBody body = logData.getBody();
+ String content = body.getJson().getJson();
+ final boolean fromText = content.isEmpty();
+ if (fromText) {
+ content = body.getText().getText();
+ }
+ final Map<String, Object> parsed =
jsonParser.create().readValue(content, parsedType);
addMetadataFields(parsed, ctx.metadata());
ctx.parsed(parsed);
- } catch (final Exception e) {
- if (jsonParser.abortOnFailure()) {
- ctx.abort();
+ if (fromText) {
+
ctx.input(logData.build().toBuilder().setBody(LogDataBody.newBuilder()
+ .setJson(JSONLog.newBuilder().setJson(content).build())
+ .build()));
}
+ } catch (final Exception e) {
+ warnParseFailure(jsonWarnLimiter, "json", ctx, abortOnFailure, e);
+ abortOrContinueUnparsed(ctx, abortOnFailure);
}
}
/**
* LAL {@code yaml {}} — parses {@code LogData.body.yaml.yaml} into a
* {@code Map<String, Object>} and stores it in {@code ctx.parsed()}.
- * Metadata fields are added the same way as {@link
#json(ExecutionContext)}.
+ * Metadata fields, failure logging, typed-proto inputs, and {@code
abortOnFailure}
+ * are handled the same way as {@link #json(ExecutionContext, boolean)}.
*/
- public void yaml(final ExecutionContext ctx) {
+ public void yaml(final ExecutionContext ctx, final boolean abortOnFailure)
{
if (ctx.shouldAbort()) {
return;
}
+ final Object rawInput = ctx.input();
+ if (!(rawInput instanceof LogData.Builder)) {
+ abortOrContinueUnparsed(ctx, abortOnFailure);
+ return;
+ }
try {
- final LogData.Builder logData = (LogData.Builder) ctx.input();
+ final LogData.Builder logData = (LogData.Builder) rawInput;
final Map<String, Object> parsed = yamlParser.create().load(
logData.getBody().getYaml().getYaml()
);
addMetadataFields(parsed, ctx.metadata());
ctx.parsed(parsed);
} catch (final Exception e) {
- if (yamlParser.abortOnFailure()) {
- ctx.abort();
+ warnParseFailure(yamlWarnLimiter, "yaml", ctx, abortOnFailure, e);
+ abortOrContinueUnparsed(ctx, abortOnFailure);
+ }
+ }
+
+ /**
+ * Failed-parse epilogue: abort when the rule demands it; otherwise
install a
+ * metadata-only parsed map so downstream {@code parsed.*} reads stay
null-safe on
+ * the continuation path — without it the generated extractor would NPE on
the null
+ * map and drop the log despite {@code abortOnFailure false}.
+ */
+ private void abortOrContinueUnparsed(final ExecutionContext ctx, final
boolean abortOnFailure) {
+ if (abortOnFailure) {
+ ctx.abort();
+ return;
+ }
+ final Map<String, Object> parsed = new HashMap<>();
+ addMetadataFields(parsed, ctx.metadata());
+ ctx.parsed(parsed);
+ }
+
+ /**
+ * Parse-failure logging shared by {@code json {}} / {@code yaml {}}: WARN
when the
+ * failure aborts the log (rate-limited, with the suppressed count
reported on the
+ * next emission), DEBUG when {@code abortOnFailure false} makes the miss
expected
+ * control flow.
+ */
+ private static void warnParseFailure(final ParseFailureWarnLimiter limiter,
+ final String parser,
+ final ExecutionContext ctx,
+ final boolean abortOnFailure,
+ final Exception e) {
+ if (abortOnFailure) {
+ final long suppressed = limiter.acquire();
+ if (suppressed >= 0) {
+ LOGGER.warn("LAL {} parser failed to parse the log body
(service={}): {}"
+ + " ({} similar failures suppressed since the last
report)",
+ parser, ctx.metadata().getService(), e.toString(),
suppressed);
}
+ } else if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("LAL {} parser failed to parse the log body
(service={}, abortOnFailure=false)",
+ parser, ctx.metadata().getService(), e);
}
}
diff --git
a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/dsl/spec/parser/AbstractParserSpec.java
b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/dsl/spec/parser/AbstractParserSpec.java
index d27e4d42dc..64aed29305 100644
---
a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/dsl/spec/parser/AbstractParserSpec.java
+++
b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/dsl/spec/parser/AbstractParserSpec.java
@@ -18,32 +18,19 @@
package org.apache.skywalking.oap.log.analyzer.v2.dsl.spec.parser;
-import lombok.experimental.Accessors;
import org.apache.skywalking.oap.log.analyzer.v2.dsl.spec.AbstractSpec;
import
org.apache.skywalking.oap.log.analyzer.v2.provider.LogAnalyzerModuleConfig;
import org.apache.skywalking.oap.server.library.module.ModuleManager;
-@Accessors
+/**
+ * Base of the shared parser specs. Per-rule options such as {@code
abortOnFailure} are NOT
+ * state on the spec — one spec instance serves every compiled rule
concurrently, so the v2
+ * compiler bakes each rule's flag into the generated call site (see
+ * {@code LALClassGenerator#generateFilterStatement}) and it travels as a
method parameter.
+ */
public class AbstractParserSpec extends AbstractSpec {
- /**
- * Whether the filter chain should abort when parsing the logs failed.
- *
- * Failing to parse the logs means either parsing throws exceptions or the
logs not matching the
- * desired patterns.
- */
- private boolean abortOnFailure = true;
-
public AbstractParserSpec(final ModuleManager moduleManager,
final LogAnalyzerModuleConfig moduleConfig) {
super(moduleManager, moduleConfig);
}
-
- @SuppressWarnings("unused") // used in user LAL scripts
- public void abortOnFailure(final boolean abortOnFailure) {
- this.abortOnFailure = abortOnFailure;
- }
-
- public boolean abortOnFailure() {
- return this.abortOnFailure;
- }
}
diff --git
a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/dsl/spec/parser/ParseFailureWarnLimiter.java
b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/dsl/spec/parser/ParseFailureWarnLimiter.java
new file mode 100644
index 0000000000..e6afc27941
--- /dev/null
+++
b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/dsl/spec/parser/ParseFailureWarnLimiter.java
@@ -0,0 +1,65 @@
+/*
+ * 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.skywalking.oap.log.analyzer.v2.dsl.spec.parser;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * Rate limiter for per-log parse-failure WARNs. A high-volume layer can fail
parsing on
+ * every log, and an unbounded per-log WARN would flood the OAP log. At most
one WARN is
+ * emitted per interval per limiter instance (each parser holds its own); the
number of
+ * failures suppressed in between is reported with the next emitted WARN so no
failure
+ * goes uncounted. Intervals are measured on the monotonic clock, so wall-clock
+ * adjustments can neither over-suppress nor over-emit.
+ *
+ * <p>Lock-free: the failure path is ingest-hot, so window transitions race on
a CAS
+ * instead of a mutex. A failure racing the window boundary may be attributed
to the
+ * next report rather than the current one; the total is never lost.
+ */
+public final class ParseFailureWarnLimiter {
+ /** Shared default: one WARN per minute per parser. */
+ public static final long DEFAULT_INTERVAL_MS = 60_000L;
+
+ private final long intervalNanos;
+ private final AtomicLong lastEmitNanos;
+ private final AtomicLong suppressed = new AtomicLong();
+
+ public ParseFailureWarnLimiter(final long intervalMs) {
+ this.intervalNanos = TimeUnit.MILLISECONDS.toNanos(intervalMs);
+ // Back-date so the first failure always emits.
+ this.lastEmitNanos = new AtomicLong(System.nanoTime() -
this.intervalNanos);
+ }
+
+ /**
+ * @return the number of failures suppressed since the previously emitted
WARN when a
+ * WARN should be emitted now, or {@code -1} when this failure
should be
+ * suppressed
+ */
+ public long acquire() {
+ final long nowNanos = System.nanoTime();
+ final long lastNanos = lastEmitNanos.get();
+ if (nowNanos - lastNanos >= intervalNanos
+ && lastEmitNanos.compareAndSet(lastNanos, nowNanos)) {
+ return suppressed.getAndSet(0);
+ }
+ suppressed.incrementAndGet();
+ return -1;
+ }
+}
diff --git
a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/dsl/spec/parser/TextParserSpec.java
b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/dsl/spec/parser/TextParserSpec.java
index a0f07926c2..e8ee67037c 100644
---
a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/dsl/spec/parser/TextParserSpec.java
+++
b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/dsl/spec/parser/TextParserSpec.java
@@ -20,32 +20,48 @@ package
org.apache.skywalking.oap.log.analyzer.v2.dsl.spec.parser;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
+import lombok.extern.slf4j.Slf4j;
import org.apache.skywalking.apm.network.logging.v3.LogData;
import org.apache.skywalking.oap.log.analyzer.v2.dsl.ExecutionContext;
import
org.apache.skywalking.oap.log.analyzer.v2.provider.LogAnalyzerModuleConfig;
import org.apache.skywalking.oap.server.library.module.ModuleManager;
+@Slf4j
public class TextParserSpec extends AbstractParserSpec {
+ private final ParseFailureWarnLimiter warnLimiter =
+ new
ParseFailureWarnLimiter(ParseFailureWarnLimiter.DEFAULT_INTERVAL_MS);
+
public TextParserSpec(final ModuleManager moduleManager,
final LogAnalyzerModuleConfig moduleConfig) {
super(moduleManager, moduleConfig);
}
- public void regexp(final ExecutionContext ctx, final String regexp) {
- regexp(ctx, Pattern.compile(regexp));
+ public void regexp(final ExecutionContext ctx, final String regexp, final
boolean abortOnFailure) {
+ regexp(ctx, Pattern.compile(regexp), abortOnFailure);
}
- public void regexp(final ExecutionContext ctx, final Pattern pattern) {
+ public void regexp(final ExecutionContext ctx, final Pattern pattern,
final boolean abortOnFailure) {
if (ctx.shouldAbort()) {
return;
}
- final LogData.Builder log = (LogData.Builder) ctx.input();
- final Matcher matcher =
pattern.matcher(log.getBody().getText().getText());
+ final LogData.Builder logData = (LogData.Builder) ctx.input();
+ final Matcher matcher =
pattern.matcher(logData.getBody().getText().getText());
final boolean matched = matcher.find();
if (matched) {
ctx.parsed(matcher);
- } else if (abortOnFailure()) {
- ctx.abort();
+ } else {
+ if (abortOnFailure) {
+ final long suppressed = warnLimiter.acquire();
+ if (suppressed >= 0) {
+ log.warn("LAL text parser regexp did not match the log
body (service={})"
+ + " ({} similar failures suppressed since the last
report)",
+ ctx.metadata().getService(), suppressed);
+ }
+ ctx.abort();
+ } else if (log.isDebugEnabled()) {
+ log.debug("LAL text parser regexp did not match the log body
(service={}, abortOnFailure=false)",
+ ctx.metadata().getService());
+ }
}
}
diff --git
a/oap-server/analyzer/log-analyzer/src/test/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/LALClassGeneratorBasicTest.java
b/oap-server/analyzer/log-analyzer/src/test/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/LALClassGeneratorBasicTest.java
index 4e9c3310af..0330d7a1e1 100644
---
a/oap-server/analyzer/log-analyzer/src/test/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/LALClassGeneratorBasicTest.java
+++
b/oap-server/analyzer/log-analyzer/src/test/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/LALClassGeneratorBasicTest.java
@@ -80,7 +80,8 @@ class LALClassGeneratorBasicTest extends
LALClassGeneratorTestBase {
compileAndAssert(dsl);
final String source = generator.generateSource(dsl);
assertNotNull(source);
- assertTrue(source.contains("filterSpec.json(ctx)"));
+ // The codegen bakes the rule's abortOnFailure flag (default true)
into the call.
+ assertTrue(source.contains("filterSpec.json(ctx, true)"));
assertTrue(source.contains("filterSpec.sink(ctx)"));
}
diff --git
a/oap-server/analyzer/log-analyzer/src/test/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/LALScriptExecutionTest.java
b/oap-server/analyzer/log-analyzer/src/test/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/LALScriptExecutionTest.java
index 3a24bb294b..1cdca79c46 100644
---
a/oap-server/analyzer/log-analyzer/src/test/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/LALScriptExecutionTest.java
+++
b/oap-server/analyzer/log-analyzer/src/test/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/LALScriptExecutionTest.java
@@ -30,6 +30,7 @@ import java.util.ServiceLoader;
import com.google.protobuf.Message;
import org.apache.skywalking.apm.network.common.v3.KeyStringValuePair;
import org.apache.skywalking.apm.network.logging.v3.LogData;
+import org.apache.skywalking.apm.network.logging.v3.LogDataBody;
import org.apache.skywalking.oap.log.analyzer.v2.module.LogAnalyzerModule;
import
org.apache.skywalking.oap.log.analyzer.v2.provider.LogAnalyzerModuleConfig;
import org.apache.skywalking.oap.log.analyzer.v2.spi.LALSourceTypeProvider;
@@ -249,6 +250,10 @@ class LALScriptExecutionTest {
case "timestamp":
assertOutputField(ruleName, output, "timestamp", expected);
break;
+ case "contentType":
+ assertEquals(expected, bodyContentType(logBuilder),
+ ruleName + ": persisted body content type mismatch");
+ break;
default:
if (key.startsWith("tag.")) {
final String tagKey = key.substring(4);
@@ -275,6 +280,25 @@ class LALScriptExecutionTest {
}
}
+ /** The content type {@code LogBuilder.toLog()} would persist, derived
from the body
+ * oneof case of the (possibly LAL-rewritten) input. */
+ private static String bodyContentType(final LogData.Builder logBuilder) {
+ if (logBuilder == null) {
+ return "NONE";
+ }
+ final LogDataBody body = logBuilder.getBody();
+ if (body.hasJson()) {
+ return "JSON";
+ }
+ if (body.hasYaml()) {
+ return "YAML";
+ }
+ if (body.hasText()) {
+ return "TEXT";
+ }
+ return "NONE";
+ }
+
private static final Map<String, String[]> FIELD_GETTER_CANDIDATES;
static {
diff --git
a/oap-server/analyzer/log-analyzer/src/test/resources/scripts/lal/test-lal/feature-cases/execution-basic.data.yaml
b/oap-server/analyzer/log-analyzer/src/test/resources/scripts/lal/test-lal/feature-cases/execution-basic.data.yaml
index 7c67febff0..066dd39033 100644
---
a/oap-server/analyzer/log-analyzer/src/test/resources/scripts/lal/test-lal/feature-cases/execution-basic.data.yaml
+++
b/oap-server/analyzer/log-analyzer/src/test/resources/scripts/lal/test-lal/feature-cases/execution-basic.data.yaml
@@ -199,3 +199,42 @@ auto-layer-abort:
os.name: "Android"
expect:
abort: true
+
+# json {} falls back to the text body when the JSON body is empty — the OTLP
log
+# receiver wraps every string body as text, even JSON-shaped ones. On success
the
+# body is normalized to JSON, so the log persists with content type JSON.
+json-text-fallback:
+ body-type: text
+ body: '{"service":"otlp-svc","env":"prod"}'
+ expect:
+ save: true
+ abort: false
+ service: otlp-svc
+ tag.env: prod
+ contentType: JSON
+
+# Non-JSON text under the fallback still aborts (default abortOnFailure true).
+json-text-fallback-invalid:
+ body-type: text
+ body: 'plain non-json log line'
+ expect:
+ abort: true
+
+# abortOnFailure false: a json parse failure must NOT abort — the log flows
through,
+# and parsed.* reads stay null-safe with metadata values as fallback.
+json-abort-on-failure-false:
+ service: fallback-svc
+ body-type: text
+ body: 'plain non-json log line'
+ expect:
+ abort: false
+ save: true
+ service: fallback-svc
+
+# abortOnFailure false on a text regexp: a no-match must NOT abort.
+text-abort-on-failure-false:
+ body-type: text
+ body: 'a line that does not match the level pattern'
+ expect:
+ abort: false
+ save: true
diff --git
a/oap-server/analyzer/log-analyzer/src/test/resources/scripts/lal/test-lal/feature-cases/execution-basic.yaml
b/oap-server/analyzer/log-analyzer/src/test/resources/scripts/lal/test-lal/feature-cases/execution-basic.yaml
index 45a1398854..4e37e979ba 100644
---
a/oap-server/analyzer/log-analyzer/src/test/resources/scripts/lal/test-lal/feature-cases/execution-basic.yaml
+++
b/oap-server/analyzer/log-analyzer/src/test/resources/scripts/lal/test-lal/feature-cases/execution-basic.yaml
@@ -324,3 +324,50 @@ rules:
}
sink {}
}
+
+ - name: json-text-fallback
+ layer: GENERAL
+ dsl: |
+ filter {
+ json {}
+ extractor {
+ service parsed.service as String
+ tag env: parsed.env as String
+ }
+ sink {}
+ }
+
+ - name: json-text-fallback-invalid
+ layer: GENERAL
+ dsl: |
+ filter {
+ json {}
+ extractor {
+ service parsed.service as String
+ }
+ sink {}
+ }
+
+ - name: json-abort-on-failure-false
+ layer: GENERAL
+ dsl: |
+ filter {
+ json {
+ abortOnFailure false
+ }
+ extractor {
+ service parsed.service as String
+ }
+ sink {}
+ }
+
+ - name: text-abort-on-failure-false
+ layer: GENERAL
+ dsl: |
+ filter {
+ text {
+ regexp "^(?<level>INFO|WARN|ERROR) (?<msg>.*)$"
+ abortOnFailure false
+ }
+ sink {}
+ }
diff --git
a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/classloader/ClassLoaderGc.java
b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/classloader/ClassLoaderGc.java
index cd2286215d..915ae72420 100644
---
a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/classloader/ClassLoaderGc.java
+++
b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/classloader/ClassLoaderGc.java
@@ -49,13 +49,15 @@ import lombok.extern.slf4j.Slf4j;
* only reclaimed by a class-unloading-capable GC cycle (G1 concurrent mark /
full GC — young
* collections never unload classes), and an idle heap may not run one for
hours. So "retired
* N minutes ago and still uncollected" is NOT a leak signal by itself. To
tell a pinned
- * loader apart from plain GC inactivity, the graveyard arms an <em>unload
probe</em> whenever
- * entries are pending: a parent-less throwaway classloader that defines one
empty class
- * ({@link UnloadProbePayload}) and is immediately dereferenced. The probe has
the exact same
- * collection requirement as a retired rule loader, so its collection is proof
that a
- * class-unloading cycle completed after the probe was minted. {@link
#leakSuspects} flags an
- * entry only when such a cycle ran comfortably after the entry's retirement
and the entry
- * still survived it.
+ * loader apart from plain GC inactivity, the graveyard arms an <em>unload
probe</em>: a
+ * parent-less throwaway classloader that defines one empty class ({@link
UnloadProbePayload})
+ * and is immediately dereferenced. The probe has the exact same collection
requirement as a
+ * retired rule loader, so its collection is proof that a class-unloading
cycle completed
+ * after the probe was minted. A probe is armed only once a pending entry's
settle window has
+ * elapsed, so its mint time is directly comparable to {@code retiredAt +
settle}: when it is
+ * collected while the entry survives, the entry provably outlived a cycle
that ran after its
+ * full settle window — {@link #leakSuspects} flags exactly those entries.
Loaders collected
+ * before their settle window elapses never cause a probe to be minted at all.
*
* <p>This graveyard is internal to {@link DSLClassLoaderManager}. The manager
retires loaders
* here via {@code dropRuntime} (full teardown) and {@code retire}
(engine-decided "displaced
@@ -76,10 +78,10 @@ final class ClassLoaderGc {
private final Object probeLock = new Object();
/** Live probe generation; at most one at a time. Guarded by {@link
#probeLock}. The
* phantom ref must stay strongly held here or it would be GC'd before
enqueuing. */
- private PhantomReference<ClassLoader> liveProbe;
- private long liveProbeMintedAtMs;
- /** Latest probe mint-time for which a class-unloading GC cycle is
confirmed complete.
- * {@code 0} until the first probe collection is observed. */
+ private ProbeRef liveProbe;
+ /** Mint time of the newest collected probe — a sound lower bound on when a
+ * class-unloading GC cycle last completed. {@code 0} until the first
probe collection
+ * is observed. */
private volatile long unloadEvidenceUpToMs;
@Getter
@@ -99,16 +101,16 @@ final class ClassLoaderGc {
loader.getKind(), loader.getCatalog(), loader.getRule(),
loader.getContentHash(),
System.currentTimeMillis(), ref);
pending.put(ref, retired);
- armUnloadProbe();
}
/**
* Drain collected phantoms from the queue. Returns the entries the JVM
confirmed as
* unreachable since the last sweep. Entries that remain in {@link
#pending()} after this
* call have not been collected yet — {@link #leakSuspects} decides which
of them are
- * evidence-backed leaks. Called by the manager's internal sweeper thread.
+ * evidence-backed leaks. Called by the manager's internal sweeper thread;
{@code
+ * settleMs} is the manager's leak settle window, which drives on-demand
probe arming.
*/
- Collection<Retired> sweep() {
+ Collection<Retired> sweep(final long settleMs) {
drainUnloadProbe();
final List<Retired> drained = new ArrayList<>();
Reference<?> r;
@@ -126,11 +128,7 @@ final class ClassLoaderGc {
drained.add(done);
}
}
- // Re-arm so entries retired after the previous probe was minted get
their own
- // evidence generation on a later class-unloading cycle.
- if (!pending.isEmpty()) {
- armUnloadProbe();
- }
+ armUnloadProbeIfNeeded(settleMs);
return drained;
}
@@ -171,9 +169,9 @@ final class ClassLoaderGc {
return out;
}
- /** Latest probe mint-time confirmed survived-past by a class-unloading GC
cycle;
- * {@code 0} while no cycle has been observed. Exposed for the manager's
diagnostics
- * and for deterministic tests. */
+ /** Mint time of the newest collected probe — a class-unloading GC cycle
is confirmed
+ * to have completed after this time; {@code 0} while no probe collection
has been
+ * observed. Exposed for the manager's diagnostics and for deterministic
tests. */
long unloadEvidenceUpToMs() {
return unloadEvidenceUpToMs;
}
@@ -187,8 +185,9 @@ final class ClassLoaderGc {
/**
* Record that a class-unloading GC cycle completed after {@code
probeMintedAtMs}.
- * Normally driven by {@link #drainUnloadProbe()}; package-private so
tests can exercise
- * {@link #leakSuspects} without depending on real GC timing.
+ * Normally driven by {@link #drainUnloadProbe()} with a collected probe's
mint time;
+ * package-private so tests can exercise {@link #leakSuspects} without
depending on
+ * real GC timing.
*/
void recordUnloadEvidence(final long probeMintedAtMs) {
synchronized (probeLock) {
@@ -198,35 +197,59 @@ final class ClassLoaderGc {
}
}
- /** Mint a fresh probe if none is live. The probe loader leaves this
method with the
- * phantom reference as its only remaining reference, so the next
class-unloading GC
- * cycle is guaranteed to collect it. */
- private void armUnloadProbe() {
+ /**
+ * Mint a probe when a pending entry needs one: some unwarned entry's
settle window has
+ * elapsed and neither the recorded evidence nor the live probe's mint
time reaches that
+ * window's end. Arming on demand keeps the probe's mint time at-or-after
{@code
+ * retiredAt + settleMs}, so a collected probe proves a cycle ran after
the full settle
+ * window — a probe minted earlier could only prove a cycle the entry was
not yet
+ * required to have survived. A live probe minted too early for a newer
entry is
+ * replaced; the stale phantom is left to die untracked ({@link
#drainUnloadProbe}
+ * still credits its mint time if it happens to enqueue first).
+ */
+ private void armUnloadProbeIfNeeded(final long settleMs) {
if (PROBE_PAYLOAD_BYTECODE == null) {
return;
}
+ long neededMintMs = Long.MAX_VALUE;
+ for (final Retired r : pending.values()) {
+ if (!r.warnedAlready()) {
+ neededMintMs = Math.min(neededMintMs, r.retiredAtMs() +
settleMs);
+ }
+ }
+ if (neededMintMs == Long.MAX_VALUE || unloadEvidenceUpToMs >=
neededMintMs) {
+ return;
+ }
+ final long nowMs = System.currentTimeMillis();
+ if (nowMs < neededMintMs) {
+ return;
+ }
synchronized (probeLock) {
- if (liveProbe != null) {
+ if (liveProbe != null && liveProbe.mintedAtMs() >= neededMintMs) {
return;
}
final ProbeClassLoader probe = new ProbeClassLoader();
probe.definePayload();
- liveProbe = new PhantomReference<>(probe, probeQueue);
- liveProbeMintedAtMs = System.currentTimeMillis();
+ liveProbe = new ProbeRef(probe, probeQueue, nowMs);
}
}
private void drainUnloadProbe() {
- boolean collected = false;
- while (probeQueue.poll() != null) {
- collected = true;
- }
- if (collected) {
+ long latestMintMs = -1L;
+ Reference<?> ref;
+ while ((ref = probeQueue.poll()) != null) {
+ if (ref instanceof ProbeRef) {
+ latestMintMs = Math.max(latestMintMs, ((ProbeRef)
ref).mintedAtMs());
+ }
synchronized (probeLock) {
- recordUnloadEvidence(liveProbeMintedAtMs);
- liveProbe = null;
+ if (ref == liveProbe) {
+ liveProbe = null;
+ }
}
}
+ if (latestMintMs >= 0) {
+ recordUnloadEvidence(latestMintMs);
+ }
}
private static byte[] readProbePayloadBytecode() {
@@ -263,6 +286,23 @@ final class ClassLoaderGc {
}
}
+ /** Probe phantom carrying its mint time, so a drained probe proves "a
class-unloading
+ * cycle completed after {@code mintedAtMs}" without external bookkeeping
— even for a
+ * replaced probe that enqueues after its tracking was dropped. */
+ private static final class ProbeRef extends PhantomReference<ClassLoader> {
+ private final long mintedAtMs;
+
+ ProbeRef(final ClassLoader probe, final ReferenceQueue<ClassLoader>
queue,
+ final long mintedAtMs) {
+ super(probe, queue);
+ this.mintedAtMs = mintedAtMs;
+ }
+
+ long mintedAtMs() {
+ return mintedAtMs;
+ }
+ }
+
/** Informational record surfaced to the sweeper. Identity is immutable;
only the
* {@code warned} latch can flip, and only once per lifetime of this
{@code Retired}. */
static final class Retired {
diff --git
a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/classloader/DSLClassLoaderManager.java
b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/classloader/DSLClassLoaderManager.java
index eab1930329..a463888213 100644
---
a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/classloader/DSLClassLoaderManager.java
+++
b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/classloader/DSLClassLoaderManager.java
@@ -233,7 +233,7 @@ public final class DSLClassLoaderManager {
private void sweepInternal() {
try {
- final Collection<ClassLoaderGc.Retired> collected =
graveyard.sweep();
+ final Collection<ClassLoaderGc.Retired> collected =
graveyard.sweep(STALE_LOADER_WARN_THRESHOLD_MS);
if (!collected.isEmpty() && log.isDebugEnabled()) {
log.debug("dsl-classloader-gc: {} loader(s) confirmed
collected", collected.size());
}
