Surface LAL drop reason in DSL live-debugging

When a LAL rule stops a log at a parse step, live-debug could only show
continueOn=false (that it stopped, never why). Add a human-readable reason:

- ExecutionContext.dropReason(String)/dropReason(): set at the LAL drop sites
  (json/yaml parse failure, text regexp non-match, non-log-body input) in
  FilterSpec/TextParserSpec.
- The shared debug Sample gains an optional reason field (non-final, so the
  5-arg constructor and MAL/OAL recorders are unchanged); the LAL recorder
  stamps it from ctx.dropReason() only when the step stopped (continueOn=false),
  so the reason lands on the parser step that halted the pipeline.
- Reason flows through the dsl-debugging REST session response and the cluster
  forward proto (ClusterSample.reason) for single-node and cross-node sessions.
- Document per-DSL TYPE_* usage inline on Sample.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: wu-sheng

docs/en/changes/changes.md

@@ -265,6 +265,7 @@
 * 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-call deadline never adds to the operator's apply latency. On the peer side the notify-triggered reconcile is coalesced: a burst of notifies (a multi-rule file, or several applies) collapses to a single queued full reconcile rather than one redundant `dao.getAll()` scan per notify. The notify is best-effort and idempotent (the peer runs its normal per-file-locked reconcile; a lost notify is harmless — the peer still self-converges on its next tick), so it tightens the cluster-convergence window without adding a hard dependency on the main being reachable.
 * 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 evicting) while the 30s reconcile only covers runtime-rule rows, so nothing re-derived it. The peer / local-cache-only install path now (re)derives and overwrites the local schema entry from the declared model with zero server RPC — honoring the `inspectBackend=false` contract so the cache can never lag the worker, including across a reshape — and a model removal now evicts its cache entry so a dropped or reshaped model leaves no stale translation behind; the persist DAOs keep an RPC-free re-derivation as a read-side backstop, and the no-init defer poll loop retries a transient backend probe error instead of escaping and crash-looping the pod.
 * 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 content type `JSON`; the normalization is scoped to that rule's context — other rules analyzing the same log still see the original text body.
+* Surface the drop reason in LAL live-debugging. When a LAL rule stops a log at a parse step (a `json {}` / `yaml {}` parse failure, a `text { regexp }` non-match, or a non-log-body input), the recorder now captures a human-readable `reason` (e.g. the parse exception) onto the DSL-debug `Sample`, exposed through the `dsl-debugging` REST session response and the cluster forward proto. Previously a live-debug watcher could only see `continueOn=false` — that a step stopped, never why — and had to read the OAP server log. `Sample.reason` is shared across all DSL debuggers but populated by LAL today.
 * 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 off three frozen error counters). The window is now keyed by the counter's own metric name.
 * 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/containers.
 * 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-proxy Grafana boards, and include queue batch/message granularity (apache/skywalking-banyandb#1169). Adds a `SERVICE_INSTANCE_RELATION` MAL scope and `serviceInstanceRelation(...)` builder powering a new intra-cluster pod-to-pod deployment topology (`banyandb-instance-relation.yaml`). The stale single-node `host_name` model is removed.

oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/dsl/ExecutionContext.java

@@ -57,6 +57,7 @@ public class ExecutionContext {
     public static final String KEY_DRY_RUN = "dry_run";
     public static final String KEY_AUTO_LAYER = "auto_layer";
     public static final String KEY_OUTPUT = "output";
+    public static final String KEY_DROP_REASON = "drop_reason";
 
     private final Map<String, Object> properties = new HashMap<>();
 
@@ -112,6 +113,7 @@ public ExecutionContext init(final LogMetadata metadata, final Object input) {
         setProperty(KEY_METRICS_CONTAINER, null);
         setProperty(KEY_DRY_RUN, false);
         setProperty(KEY_OUTPUT, null);
+        setProperty(KEY_DROP_REASON, null);
         return this;
     }
 
@@ -174,6 +176,23 @@ public boolean shouldAbort() {
         return (boolean) getProperty(KEY_ABORT);
     }
 
+    /**
+     * Record a human-readable reason the pipeline is dropping the log at the current step
+     * (e.g. a parse-failure message). Read by the DSL-debug recorder to populate
+     * {@code Sample.reason} so a live-debug watcher sees WHY a log was dropped, not just
+     * that it was. Set on the drop path only — alongside {@link #abort()}; a log that
+     * continues is not dropped and records no reason, so a later {@code abort {}} statement
+     * never inherits a stale reason from an earlier continued step.
+     */
+    public ExecutionContext dropReason(final String reason) {
+        setProperty(KEY_DROP_REASON, reason);
+        return this;
+    }
+
+    public String dropReason() {
+        return (String) getProperty(KEY_DROP_REASON);
+    }
+
     public ExecutionContext metricsContainer(final List<SampleFamily> container) {
         setProperty(KEY_METRICS_CONTAINER, container);
         return this;

oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/dsl/spec/filter/FilterSpec.java

@@ -157,7 +157,7 @@ public void json(final ExecutionContext ctx, final boolean abortOnFailure) {
         }
         final Object rawInput = ctx.input();
         if (!(rawInput instanceof LogData.Builder)) {
-            abortOrContinueUnparsed(ctx, abortOnFailure);
+            abortOrContinueUnparsed(ctx, abortOnFailure, notLogBodyReason("json", rawInput));
             return;
         }
         try {
@@ -178,7 +178,7 @@ public void json(final ExecutionContext ctx, final boolean abortOnFailure) {
             }
         } catch (final Exception e) {
             warnParseFailure(jsonWarnLimiter, "json", ctx, abortOnFailure, e);
-            abortOrContinueUnparsed(ctx, abortOnFailure);
+            abortOrContinueUnparsed(ctx, abortOnFailure, "json parse failed: " + e);
         }
     }
 
@@ -194,7 +194,7 @@ public void yaml(final ExecutionContext ctx, final boolean abortOnFailure) {
         }
         final Object rawInput = ctx.input();
         if (!(rawInput instanceof LogData.Builder)) {
-            abortOrContinueUnparsed(ctx, abortOnFailure);
+            abortOrContinueUnparsed(ctx, abortOnFailure, notLogBodyReason("yaml", rawInput));
             return;
         }
         try {
@@ -206,18 +206,23 @@ public void yaml(final ExecutionContext ctx, final boolean abortOnFailure) {
             ctx.parsed(parsed);
         } catch (final Exception e) {
             warnParseFailure(yamlWarnLimiter, "yaml", ctx, abortOnFailure, e);
-            abortOrContinueUnparsed(ctx, abortOnFailure);
+            abortOrContinueUnparsed(ctx, abortOnFailure, "yaml parse failed: " + e);
         }
     }
 
     /**
-     * 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}.
+     * Failed-parse epilogue: when the rule aborts, record the drop reason (surfaced in
+     * live-debug via {@code Sample.reason}) and abort; 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}. The reason is set ONLY on the aborting path:
+     * a continued log is not dropped, and leaving a stale reason on the context would let a
+     * later {@code abort {}} statement wrongly inherit this step's parse-failure text.
      */
-    private void abortOrContinueUnparsed(final ExecutionContext ctx, final boolean abortOnFailure) {
+    private void abortOrContinueUnparsed(final ExecutionContext ctx, final boolean abortOnFailure,
+                                         final String reason) {
         if (abortOnFailure) {
+            ctx.dropReason(reason);
             ctx.abort();
             return;
         }
@@ -226,6 +231,11 @@ private void abortOrContinueUnparsed(final ExecutionContext ctx, final boolean a
         ctx.parsed(parsed);
     }
 
+    private static String notLogBodyReason(final String parser, final Object rawInput) {
+        final String actual = rawInput == null ? "null" : rawInput.getClass().getSimpleName();
+        return parser + " parser: input is not a log body (expected LogData, got " + actual + ")";
+    }
+
     /**
      * 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

oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/dsl/spec/parser/TextParserSpec.java

@@ -44,24 +44,36 @@ public void regexp(final ExecutionContext ctx, final Pattern pattern, final bool
         if (ctx.shouldAbort()) {
             return;
         }
-        final LogData.Builder logData = (LogData.Builder) ctx.input();
+        final Object rawInput = ctx.input();
+        if (!(rawInput instanceof LogData.Builder)) {
+            // Typed-proto input (e.g. Envoy ALS) reaches a text{regexp} rule as a routing
+            // mismatch, not a text body — honor abortOnFailure without a ClassCastException.
+            if (abortOnFailure) {
+                final String actual = rawInput == null ? "null" : rawInput.getClass().getSimpleName();
+                ctx.dropReason("text parser: input is not a log body (expected LogData, got " + actual + ")");
+                ctx.abort();
+            }
+            return;
+        }
+        final LogData.Builder logData = (LogData.Builder) rawInput;
         final Matcher matcher = pattern.matcher(logData.getBody().getText().getText());
         final boolean matched = matcher.find();
         if (matched) {
             ctx.parsed(matcher);
-        } 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());
+        } else if (abortOnFailure) {
+            // Reason set only on the aborting path: a continued log is not dropped, and a
+            // stale reason would leak onto a later abort {} statement.
+            ctx.dropReason("text parser: regexp did not match the log body");
+            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());
         }
     }
 

oap-server/analyzer/log-analyzer/src/test/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/LALScriptExecutionTest.java

@@ -52,6 +52,8 @@
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.mock;
@@ -254,6 +256,17 @@ private void validateExpected(final String ruleName,
                     assertEquals(expected, bodyContentType(logBuilder),
                         ruleName + ": persisted body content type mismatch");
                     break;
+                case "dropReason": {
+                    final String reason = ctx.dropReason();
+                    assertNotNull(reason, ruleName + ": expected a drop reason on ctx");
+                    assertTrue(reason.contains(expected),
+                        ruleName + ": drop reason '" + reason + "' should contain '" + expected + "'");
+                    break;
+                }
+                case "noDropReason":
+                    assertNull(ctx.dropReason(),
+                        ruleName + ": expected no drop reason, got '" + ctx.dropReason() + "'");
+                    break;
                 default:
                     if (key.startsWith("tag.")) {
                         final String tagKey = key.substring(4);

oap-server/analyzer/log-analyzer/src/test/resources/scripts/lal/test-lal/feature-cases/execution-basic.data.yaml

@@ -213,15 +213,18 @@ json-text-fallback:
     tag.env: prod
     contentType: JSON
 
-# Non-JSON text under the fallback still aborts (default abortOnFailure true).
+# Non-JSON text under the fallback still aborts (default abortOnFailure true), and the
+# drop reason is recorded on ctx so live-debug can surface WHY it dropped.
 json-text-fallback-invalid:
   body-type: text
   body: 'plain non-json log line'
   expect:
     abort: true
+    dropReason: "json parse failed"
 
 # abortOnFailure false: a json parse failure must NOT abort — the log flows through,
-# and parsed.* reads stay null-safe with metadata values as fallback.
+# and parsed.* reads stay null-safe with metadata values as fallback. A continued log is
+# not dropped, so NO drop reason is recorded.
 json-abort-on-failure-false:
   service: fallback-svc
   body-type: text
@@ -230,11 +233,22 @@ json-abort-on-failure-false:
     abort: false
     save: true
     service: fallback-svc
+    noDropReason: true
 
-# abortOnFailure false on a text regexp: a no-match must NOT abort.
+# A parse failure that continued (abortOnFailure false) must NOT leave a stale reason for a
+# later explicit abort {} to inherit — the abort has no recorded cause.
+json-continue-then-abort:
+  body-type: text
+  body: 'plain non-json log line'
+  expect:
+    abort: true
+    noDropReason: true
+
+# abortOnFailure false on a text regexp: a no-match must NOT abort, and records no reason.
 text-abort-on-failure-false:
   body-type: text
   body: 'a line that does not match the level pattern'
   expect:
     abort: false
     save: true
+    noDropReason: true

oap-server/analyzer/log-analyzer/src/test/resources/scripts/lal/test-lal/feature-cases/execution-basic.yaml

@@ -361,6 +361,17 @@ rules:
         sink {}
       }
 
+  - name: json-continue-then-abort
+    layer: GENERAL
+    dsl: |
+      filter {
+        json {
+          abortOnFailure false
+        }
+        abort {}
+        sink {}
+      }
+
   - name: text-abort-on-failure-false
     layer: GENERAL
     dsl: |

oap-server/server-admin/dsl-debugging/src/main/java/org/apache/skywalking/oap/server/admin/dsl/debugging/cluster/DSLDebuggingClusterServiceImpl.java

@@ -195,6 +195,7 @@ private static ClusterSample toClusterSample(final Sample s) {
             .setContinueOn(s.isContinueOn())
             .setPayloadJson(s.getPayloadJson() == null ? "{}" : s.getPayloadJson())
             .setSourceLine(s.getSourceLine())
+            .setReason(s.getReason() == null ? "" : s.getReason())
             .build();
     }
 }