This is an automated email from the ASF dual-hosted git repository. wu-sheng pushed a commit to branch feat/otlp-log-json-body in repository https://gitbox.apache.org/repos/asf/skywalking.git
commit d5b562e567dfeb1bcdb0a50a050473b8a3e6606f Author: Wu Sheng <[email protected]> AuthorDate: Thu Jul 2 21:48:26 2026 +0800 Address review: rule-scoped JSON normalization, rate-limited parse-failure WARNs, need-based unload probes - json{} text-fallback swaps this rule's context input to a JSON-bodied copy instead of mutating the LogData.Builder shared by every rule on the same log. - Parse failures that abort are WARN'd at most once per minute per parser with the suppressed count reported on the next emission; abortOnFailure=false failures are DEBUG-only and continue with a metadata-backed parsed map so parsed.* reads stay null-safe. Typed-proto inputs (Envoy ALS routing guard) stay quiet as before. - ClassLoaderGc arms the unload probe on demand once a pending entry's settle window elapses, recording the collected probe's mint time — sound evidence with single-GC detection, no drain-time overshoot. Co-Authored-By: Claude Fable 5 <[email protected]> --- docs/en/changes/changes.md | 4 +- docs/en/concepts-and-designs/lal.md | 4 +- oap-server/analyzer/log-analyzer/CLAUDE.md | 4 +- .../analyzer/v2/compiler/rt/LalRuntimeHelper.java | 6 +- .../oap/log/analyzer/v2/dsl/ExecutionContext.java | 10 ++ .../analyzer/v2/dsl/spec/filter/FilterSpec.java | 109 ++++++++++++++---- .../dsl/spec/parser/ParseFailureWarnLimiter.java | 52 +++++++++ .../v2/dsl/spec/parser/TextParserSpec.java | 13 ++- .../feature-cases/execution-basic.data.yaml | 5 +- .../test-lal/feature-cases/execution-basic.yaml | 3 + .../oap/server/core/classloader/ClassLoaderGc.java | 123 ++++++++++++++------- .../core/classloader/DSLClassLoaderManager.java | 2 +- 12 files changed, 260 insertions(+), 75 deletions(-) diff --git a/docs/en/changes/changes.md b/docs/en/changes/changes.md index b4e271e7fc..8f06148561 100644 --- a/docs/en/changes/changes.md +++ b/docs/en/changes/changes.md @@ -264,8 +264,8 @@ * 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. A parse failure keeps the existing `abortOnFailure` behavior; on a successful parse from a text body [...] -* Fix the LAL parser `abortOnFailure` option being silently ignored in the v2 (ANTLR4) compiler. The DSL flag was parsed into the rule model but never emitted into the generated code, so every `json {}` / `yaml {}` / `text { regexp }` block always used the built-in default and `abortOnFailure false` had no effect. The compiler now bakes each rule's flag into the generated parser call (its default is `true`, as documented), and a parse failure or regexp non-match is now logged at WARN ins [...] +* 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 the LAL parser `abortOnFailure` option being silently ignored in the v2 (ANTLR4) compiler. The DSL flag was parsed into the rule model but never emitted into the generated code, so every `json {}` / `yaml {}` / `text { regexp }` block always used the built-in default and `abortOnFailure false` had no effect. The compiler now bakes each rule's flag into the generated parser call (default `true`, as documented). A parse failure or regexp non-match that aborts the log is reported at W [...] * 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 793fdd7ac7..d233c10b3c 100644 --- a/docs/en/concepts-and-designs/lal.md +++ b/docs/en/concepts-and-designs/lal.md @@ -221,7 +221,9 @@ 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: it is stored with content type `JSON` and the original text content is no longer available (`log.body` reads the JSON content). This makes a JSON-shaped log delivered over any transport persist and render as JSON once a `json {}` rule matches it. +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/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/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 5e259ec339..6381299278 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,6 +20,7 @@ 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; @@ -29,6 +30,7 @@ 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; @@ -67,6 +69,16 @@ public class FilterSpec extends AbstractSpec { private final TypeReference<Map<String, Object>> parsedType; + /** One WARN per minute per parser; suppressed failures are counted and reported + * with the next emitted WARN. */ + private static final long PARSE_FAILURE_WARN_INTERVAL_MS = 60_000L; + + private final ParseFailureWarnLimiter jsonWarnLimiter = + new ParseFailureWarnLimiter(PARSE_FAILURE_WARN_INTERVAL_MS); + + private final ParseFailureWarnLimiter yamlWarnLimiter = + new ParseFailureWarnLimiter(PARSE_FAILURE_WARN_INTERVAL_MS); + public FilterSpec(final ModuleManager moduleManager, final LogAnalyzerModuleConfig moduleConfig) throws ModuleStartException { super(moduleManager, moduleConfig); @@ -122,15 +134,21 @@ public class FilterSpec extends AbstractSpec { * {@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 parse failure is logged at WARN 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 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, the body is rewritten to a - * JSON body carrying the same content, so the log persists as {@code ContentType.JSON} - * (persistence derives the stored content type from the body case in {@code - * LogBuilder.toLog()}); the original text case is dropped and no longer readable. + * <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 @@ -141,8 +159,13 @@ public class FilterSpec extends AbstractSpec { 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 LogDataBody body = logData.getBody(); String content = body.getJson().getJson(); final boolean fromText = content.isEmpty(); @@ -153,43 +176,81 @@ public class FilterSpec extends AbstractSpec { addMetadataFields(parsed, ctx.metadata()); ctx.parsed(parsed); if (fromText) { - logData.setBody(LogDataBody.newBuilder() + ctx.input(logData.build().toBuilder().setBody(LogDataBody.newBuilder() .setJson(JSONLog.newBuilder().setJson(content).build()) - .build()); + .build())); } } catch (final Exception e) { - LOGGER.warn("LAL json parser failed to parse the log body (service={}, abortOnFailure={}): {}", - ctx.metadata().getService(), abortOnFailure, e.getMessage()); - if (abortOnFailure) { - ctx.abort(); - } + 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 and {@code abortOnFailure} are handled the same way as - * {@link #json(ExecutionContext, boolean)}: a parse failure is logged at WARN - * before honoring the flag. + * 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, 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) { - LOGGER.warn("LAL yaml parser failed to parse the log body (service={}, abortOnFailure={}): {}", - ctx.metadata().getService(), abortOnFailure, e.getMessage()); - if (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(System.currentTimeMillis()); + 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.getMessage(), suppressed); } + } else if (LOGGER.isDebugEnabled()) { + LOGGER.debug("LAL {} parser failed to parse the log body (service={}, abortOnFailure=false): {}", + parser, ctx.metadata().getService(), e.getMessage()); } } 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..8a2076bc47 --- /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,52 @@ +/* + * 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; + +/** + * 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 call site; the number of failures suppressed in between is + * reported with the next emitted WARN so no failure goes uncounted. + */ +public final class ParseFailureWarnLimiter { + private final long intervalMs; + private long lastEmitMs; + private long suppressed; + + public ParseFailureWarnLimiter(final long intervalMs) { + this.intervalMs = intervalMs; + } + + /** + * @param nowMs current wall-clock time + * @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 synchronized long acquire(final long nowMs) { + if (nowMs - lastEmitMs >= intervalMs) { + final long count = suppressed; + suppressed = 0; + lastEmitMs = nowMs; + return count; + } + suppressed++; + 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 a404b70590..177dc36e5f 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 @@ -28,6 +28,8 @@ import org.apache.skywalking.oap.server.library.module.ModuleManager; @Slf4j public class TextParserSpec extends AbstractParserSpec { + private final ParseFailureWarnLimiter warnLimiter = new ParseFailureWarnLimiter(60_000L); + public TextParserSpec(final ModuleManager moduleManager, final LogAnalyzerModuleConfig moduleConfig) { super(moduleManager, moduleConfig); @@ -47,10 +49,17 @@ public class TextParserSpec extends AbstractParserSpec { if (matched) { ctx.parsed(matcher); } else { - log.warn("LAL text parser regexp did not match the log body (service={}, abortOnFailure={})", - ctx.metadata().getService(), abortOnFailure); if (abortOnFailure) { + final long suppressed = warnLimiter.acquire(System.currentTimeMillis()); + 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/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 579ed5fc2c..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 @@ -220,13 +220,16 @@ json-text-fallback-invalid: expect: abort: true -# abortOnFailure false: a json parse failure must NOT abort — the log flows through. +# 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: 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 ea4e180e0c..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 @@ -355,6 +355,9 @@ rules: json { abortOnFailure false } + extractor { + service parsed.service as String + } 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 11aea56a41..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; - /** Observed (probe-drain) time a class-unloading GC cycle was last confirmed complete; - * {@code 0} until the first is observed. Drain time, not the probe's mint time, so a - * single post-settle cycle a loader survives advances evidence past its settle window. */ + 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 observed time a class-unloading GC cycle was confirmed complete (via probe - * collection); {@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; } @@ -186,46 +184,72 @@ final class ClassLoaderGc { } /** - * Record that a class-unloading GC cycle was confirmed complete as of {@code observedAtMs}. - * Normally driven by {@link #drainUnloadProbe()} with the drain time; package-private so - * tests can exercise {@link #leakSuspects} without depending on real GC timing. + * Record that a class-unloading GC cycle completed after {@code probeMintedAtMs}. + * 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 observedAtMs) { + void recordUnloadEvidence(final long probeMintedAtMs) { synchronized (probeLock) { - if (observedAtMs > unloadEvidenceUpToMs) { - unloadEvidenceUpToMs = observedAtMs; + if (probeMintedAtMs > unloadEvidenceUpToMs) { + unloadEvidenceUpToMs = probeMintedAtMs; } } } - /** 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); + 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(System.currentTimeMillis()); - liveProbe = null; + if (ref == liveProbe) { + liveProbe = null; + } } } + if (latestMintMs >= 0) { + recordUnloadEvidence(latestMintMs); + } } private static byte[] readProbePayloadBytecode() { @@ -262,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()); }
