You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by wu...@apache.org on 2026/04/26 02:26:26 UTC
(skywalking) branch feature/runtime-rule-hot-update updated: runtime-rule: feature-branch WIP — docs / module wiring / e2e probe shape
This is an automated email from the ASF dual-hosted git repository.
wu-sheng pushed a commit to branch feature/runtime-rule-hot-update
in repository https://gitbox.apache.org/repos/asf/skywalking.git
The following commit(s) were added to refs/heads/feature/runtime-rule-hot-update by this push:
new 93205401ee runtime-rule: feature-branch WIP — docs / module wiring / e2e probe shape
93205401ee is described below
commit 93205401ee12fbaf3f61475821d19bf9d984d40a
Author: Wu Sheng <wu...@foxmail.com>
AuthorDate: Sun Apr 26 10:26:18 2026 +0800
runtime-rule: feature-branch WIP — docs / module wiring / e2e probe shape
Catches up the working tree with the runtime-rule integration work
threaded through this session. No single mechanical change here; this
commit consolidates a set of sibling-module edits made while debugging
the storage-matrix e2e end-to-end:
- docs/en/changes/changes.md, .../runtime-rule-hot-update.md,
.../backend-runtime-rule-api.md — narrative tightening for the
shipped behaviour (soft-pause /inactivate, BanyanDB-only schema
drop, two-phase Factory).
- LogAnalyzerModule.java — comment-only refresh on the
LogFilterListener.Factory and MalConverterRegistry declarations
matching the prepare-time registration pattern.
- ManagementPersistentWorker / ManagementStreamProcessor +
StorageModels + ModelInstaller + their tests — additional
bookkeeping the runtime-rule reconciler relies on (model
registration outcomes, persist-or-skip plumbing).
- Reconciler.java + RuntimeRuleRestHandler.java +
GuardrailIntegrationTest + RuntimeRuleRestHandlerTest — surface
cleanup against the latest soft-pause / BytecodeClassDefiner
paths; tests track the new entry points.
- StaticRuleRegistry + per-backend RuntimeRuleManagementDAO
implementations (BanyanDB, ES, JDBC) — registration / save /
delete shape matches the receiver's expectations.
- test/e2e-v2/cases/runtime-rule-storage-matrix/* — drop the
empty-static-rule env vars, switch BanyanDB probe from bydbctl
install to direct HTTP API on port 17913 (no install step needed
on the test host).
Module-level IT and unit tests on the runtime-rule plugin remain
green (15 pass / 1 disabled, 102 unit tests). Backend builds cleanly
with full -Pall profile. End-to-end storage-matrix flow verified
manually: OAP boots cleanly, /addOrUpdate returns 200
structural_applied with the BytecodeClassDefiner refactor in place.
---
docs/en/changes/changes.md | 14 +-
.../runtime-rule-hot-update.md | 133 ++++++++--
docs/en/setup/backend/backend-runtime-rule-api.md | 43 +++-
.../log/analyzer/v2/module/LogAnalyzerModule.java | 17 +-
.../worker/ManagementPersistentWorker.java | 11 -
.../analysis/worker/ManagementStreamProcessor.java | 19 --
.../server/core/rule/ext/StaticRuleRegistry.java | 48 ++++
.../management/RuntimeRuleManagementDAO.java | 27 +-
.../server/core/storage/model/ModelInstaller.java | 21 +-
.../server/core/storage/model/StorageModels.java | 82 ++++--
.../worker/ManagementPersistentWorkerTest.java | 48 +---
.../core/storage/model/StorageModelsTest.java | 52 ++++
.../receiver/runtimerule/reconcile/Reconciler.java | 242 ++++++++++++++---
.../runtimerule/rest/RuntimeRuleRestHandler.java | 286 ++++++++++++++++++++-
.../runtimerule/rest/GuardrailIntegrationTest.java | 40 ---
.../rest/RuntimeRuleRestHandlerTest.java | 209 +++++++++++++++
.../banyandb/BanyanDBRuntimeRuleManagementDAO.java | 35 ++-
.../query/RuntimeRuleManagementEsDAO.java | 19 ++
.../common/dao/JDBCRuntimeRuleManagementDAO.java | 57 ++++
.../banyandb/docker-compose.yml | 16 +-
.../runtime-rule-storage-matrix/banyandb/e2e.yaml | 24 +-
.../elasticsearch/docker-compose.yml | 12 +-
.../postgresql/docker-compose.yml | 11 +-
23 files changed, 1202 insertions(+), 264 deletions(-)
diff --git a/docs/en/changes/changes.md b/docs/en/changes/changes.md
index fef37927dc..fc5a28b09a 100644
--- a/docs/en/changes/changes.md
+++ b/docs/en/changes/changes.md
@@ -6,11 +6,15 @@
(port `17128`, **disabled by default**) accepts `POST /runtime/rule/addOrUpdate?catalog=&name=` with the
rule file as the raw request body and persists it into the configured storage backend. Every node in the cluster
converges to the new content within one reconciler tick (default 30 s). Lifecycle is strictly
- `ACTIVE → /inactivate → INACTIVE → /delete → absent`: `/inactivate` is the destructive path (broadcasts
- `Suspend`, drops the BanyanDB measure, unregisters handlers, resets alarm windows) and `/delete` requires the
- rule to already be `INACTIVE` (returns `HTTP 409 requires_inactivate_first` otherwise) so it is just a cheap
- row removal. Reactivation is always via `/addOrUpdate` — a same-content re-post against an `INACTIVE` row runs
- the full apply pipeline. `GET /runtime/rule/list` reports the current state of every known rule (running,
+ `ACTIVE → /inactivate → INACTIVE → /delete → absent`: `/inactivate` is the soft-pause path (broadcasts
+ `Suspend`, tears down OAP-internal state — handlers, prototypes, applied bookkeeping — but preserves the
+ BanyanDB measure and its data) and `/delete` is the destructive endpoint (requires the rule to already be
+ `INACTIVE`, returns `HTTP 409 requires_inactivate_first` otherwise; drops the backend measure, then removes
+ the tombstone row). Reactivation via `/addOrUpdate` against an `INACTIVE` row reuses the preserved measure
+ and is therefore lossless. `GET /runtime/rule?catalog=&name=` fetches a single rule's YAML (raw or JSON
+ envelope on `Accept: application/json`) with `ETag`/`If-None-Match` support; `GET /runtime/rule/bundled?catalog=`
+ lists every static (bundled) rule for the catalog with an `overridden` flag so UIs can render the runtime
+ + static merge. `GET /runtime/rule/list` reports the current state of every known rule (running,
suspended, skipped, overridden) in NDJSON; `GET /runtime/rule/dump` returns a tar.gz archive for backup / DR.
A bundled operator CLI at `tools/runtime-rule.sh` wraps the API. Cluster coordination uses a single-direction
`Suspend` RPC plus a paired `Resume` and a 60 s self-heal timer so aborted or crashed updates roll back
diff --git a/docs/en/concepts-and-designs/runtime-rule-hot-update.md b/docs/en/concepts-and-designs/runtime-rule-hot-update.md
index 8fcc04a405..552457dfca 100644
--- a/docs/en/concepts-and-designs/runtime-rule-hot-update.md
+++ b/docs/en/concepts-and-designs/runtime-rule-hot-update.md
@@ -243,17 +243,20 @@ The three endpoints split the destructive and the cleanup work:
dispatch handlers are re-created from the posted content. Same-content posts against
an `INACTIVE` entry are treated as a reactivation (not a no-op) because the entry's
status is what matters, not the bytes.
-- **`/inactivate`** is the destructive path. It broadcasts `Suspend`, flips the entry
- status to `INACTIVE`, and runs the full teardown locally — dispatch handlers removed,
- BanyanDB measure dropped via `dropTable`, alarm windows reset. Peers observe the
- `INACTIVE` status on their next tick and run the same teardown. The backend is fully
- torn down after `/inactivate` completes; only the `INACTIVE` tombstone entry remains.
-- **`/delete`** is a row-only cleanup. It refuses to operate on an `ACTIVE` entry —
- returns `HTTP 409 requires_inactivate_first` so destructive work never sneaks in
- through the delete path. On an `INACTIVE` entry it is just a management-storage row
- removal under the per-file lock: no `Suspend` broadcast, no DDL, no verification,
- because `/inactivate` already did all of that. On an absent entry it is an idempotent
- 200 `not_found`.
+- **`/inactivate`** is the soft-pause path. It broadcasts `Suspend`, flips the entry
+ status to `INACTIVE`, and runs the OAP-internal teardown under `localCacheOnly` —
+ dispatch handlers removed, prototypes + Models + applied bookkeeping cleared, alarm
+ windows reset. The BanyanDB measure and its data are explicitly preserved so a later
+ re-activation via `/addOrUpdate` is cheap and lossless. Peers observe the `INACTIVE`
+ status on their next tick and run the same OAP-internal teardown.
+- **`/delete`** is the destructive path. It refuses to operate on an `ACTIVE` entry —
+ returns `HTTP 409 requires_inactivate_first` so destruction never sneaks in through
+ the delete path; the two-step `/inactivate → /delete` workflow is enforced. On an
+ `INACTIVE` entry it drops the BanyanDB measure (re-registers prototypes locally
+ under `localCacheOnly` so the cascade has Models to walk, then runs `unregisterBundle`
+ under `fullInstall` so the listener chain fires `dropTable`), then deletes the
+ tombstone row. Backend-drop failure aborts the row deletion to avoid orphaning the
+ measure. On an absent entry it is an idempotent 200 `not_found`.
Why split it this way: every destructive step is concentrated on `/inactivate`, so the
rollback story ("reactivate via `/addOrUpdate`") is uniform regardless of whether the
@@ -273,8 +276,8 @@ module has finished its own boot. The tick:
1. Fetches every runtime-rule entry from management storage.
2. Sorts the entries lexicographically by `(catalog, name)`. This is load-bearing for
- LAL: cross-file rule-name collisions within a layer are rejected at apply time, and
- if two peers processed the same set of files in different orders they could accept
+ the cross-file collision guards (LAL `(layer, ruleName)` and MAL metric-name): if
+ two peers processed the same set of files in different orders they could accept
different sides of the collision and diverge. Sorting eliminates the split.
3. Diffs the entries against an in-memory snapshot keyed by `(catalog, name)` and
classifies each differing bundle.
@@ -488,8 +491,9 @@ Six invariants together carry the distributed contract:
- **Monotonic clock for self-heal** — wall-clock jumps cannot fire or delay self-heal
incorrectly. The stamp refreshes on every origin transition so duration is measured
at the current effective origin.
-- **Lexicographic sort before applying deltas** — LAL cross-file collision rejection
- is deterministic across nodes; every peer reaches the same accept/reject partition.
+- **Lexicographic sort before applying deltas** — both LAL `(layer, ruleName)` and MAL
+ metric-name cross-file collision rejection become deterministic across nodes; every
+ peer reaches the same accept/reject partition.
- **Only the main writes DDL** (peers use local-cache-only) — double-writes to the
backend are structurally impossible regardless of timing.
@@ -525,6 +529,105 @@ from flooding logs across thousands of reconciler ticks; when the loader is fina
collected the entry is removed and a future leak of a different loader for the same
file gets its own fresh warning.
+### Persist-is-commit
+
+The runtime-rule row uses an explicit per-backend upsert API
+(`RuntimeRuleManagementDAO.save`) rather than the generic management-DAO insert path.
+The generic path silently no-ops on duplicate-row updates (Elasticsearch and JDBC
+short-circuit on existence; BanyanDB's generic management DAO never persists at all),
+which would let `/addOrUpdate` updates and `/inactivate` status flips return success
+to the operator while the backing row stays unchanged. Each backend implements `save`
+as a real upsert: BanyanDB writes the property via `apply` (upsert by id),
+Elasticsearch via `forceInsert` (the `index` API is upsert-by-_id), JDBC via
+existence-probe + `INSERT` or `UPDATE`. A `save` failure propagates to the REST
+handler as a 5xx, never as a misleading 200.
+
+The prior-row lookup the REST handler runs before the guardrail check fails closed:
+a transient DAO read failure surfaces a 503 to the operator, never as "no prior row,
+treat as first-time create" — the latter would let a STRUCTURAL change bypass the
+`allowStorageChange` guardrail because the prior content read as null.
+
+### Inactivate is soft-pause; delete is destructive
+
+`/inactivate` and `/delete` are split so that deactivating a rule never destroys data
+the operator might want back. `/inactivate` runs the apply pipeline under
+`localCacheOnly` — handlers stop dispatching, OAP-internal state (prototypes, Models,
+applied bookkeeping, per-file classloader) is torn down, but the BanyanDB measure and
+its data stay intact. Re-activation via `/addOrUpdate` reuses the existing measure;
+the cost is a recompile of the bundle, not a backfill or a metric-identity change.
+
+`/delete` is the one destructive endpoint. It requires the rule to already be
+INACTIVE (so the dispatch-stop step is explicit and reversible up to the last moment),
+then drops the BanyanDB measure and removes the tombstone row. Backend drop runs
+through the existing MeterSystem → ModelRegistry → ModelInstaller cascade: the
+reconciler re-registers prototypes locally under `localCacheOnly` (no listener-side
+define), then drives `unregisterBundle` under `fullInstall` so the listener chain
+fires `dropTable` on the live measure. A re-register failure aborts the deletion —
+the row stays, the measure stays, the operator can retry. Backend-drop failure with
+the row already gone would orphan the measure, so the order is "drop backend, then
+drop row".
+
+LAL catalogs skip the backend-drop step (LAL handlers are in-process only, no schema
+to drop). Static-shadow rules: deleting the runtime tombstone causes the next
+reconciler tick to re-register the static content under whatever role this node
+plays; if any node is main, the measure gets re-created under the static rule's
+ownership.
+
+### Topology snapshots are always provisional
+
+The cluster module exposes a peer-list snapshot, never a "membership has stabilized"
+signal — that signal cannot exist short of consensus, which the design does not run.
+Two consequences:
+
+- **Eventual convergence under stable topology.** Once topology stops moving, whichever
+ node ends up as main for a file converges by re-applying under `fullInstall` on a
+ later tick. Backend DDL is idempotent (BanyanDB define is create-if-absent; ES / JDBC
+ dropTable for management data is no-op), so a re-apply triggered by a topology shift
+ costs nothing.
+- **Self-healing under churn.** No tick result locks in a "this file is done, never
+ re-apply" decision. Marker debt (the `pendingFullInstall` set: keys whose most
+ recent apply was `localCacheOnly` and therefore did not directly observe DDL on this
+ node) is cleared only by direct evidence — this node's own successful `fullInstall`.
+ Never by a readiness heuristic. Last write wins; missed DDL never becomes permanent
+ stale state.
+
+The tick's hash-match short-circuit reads the marker only when the current tick would
+itself run `fullInstall`. A node currently acting as peer hash-match-skips even with
+the marker set, because peer policy is `localCacheOnly` anyway and the DDL debt is
+not the peer's concern. If the same node later resolves as main, the next tick
+observes (would-run-fullInstall && marker-set), re-applies under `fullInstall`, and
+clears the marker.
+
+The promotion path needs a small extra step. When the marker is set, `appliedContent`
+already holds the file content from the prior `localCacheOnly` apply, and the
+`Reconciler.classifyMal` would return `NO_CHANGE` — falling through to the
+`STRUCTURAL/NEW` branch where `MeterSystem.create`'s prototype-first short-circuit
+returns before reaching the listener chain that fires backend DDL. So a promoted
+re-apply could clear the marker without ever firing DDL. The reconciler tears down
+the local bundle (drops `meterPrototypes`, `appliedMal[key]`, `appliedContent[key]`,
+and the per-file classloader) before invoking the apply path, so it enters as a
+fresh `NEW` install: the listener chain runs, BanyanDB `define` (idempotent
+create-if-absent) creates the measure, and `verifyPostApply` confirms it. Cost is
+one bundle's recompile per promotion event, which only fires on cluster main
+re-routing for already-localCacheOnly'd files.
+
+### Backend cascade before local-state cleanup
+
+The teardown path runs the backend cascade before clearing local state, so a transient
+backend failure leaves the retry path intact. The order on `/inactivate` and on the
+reconciler's tear-down branch is: `applier.remove()` → `MeterSystem.removeMetric()` →
+`ModelRegistry.remove()` → backend installer `whenRemoving()`. Only after every layer
+returns cleanly do `appliedMal[key]`, `meterPrototypes[name]`, the `Model` registry
+entry, and the per-file classloader drop. If the BanyanDB delete-measure throws on any
+downsampling variant, every layer's local state stays populated and the next tick (or
+operator retry) re-enters the same path, finds the same state, and re-fires the
+cascade against the backend. Listeners are required to be idempotent on the drop — a
+delete-measure on a non-existent measure must be a no-op.
+
+The same shape applies on the create side: `StorageModels.add` rolls the model out of
+the registry if any `whenCreating` listener throws, so a future retry doesn't dedup-
+skip the listener and leave the storage stack permanently half-built.
+
### Lock topology
Four kinds of locks participate in writes. They are always acquired in the order
diff --git a/docs/en/setup/backend/backend-runtime-rule-api.md b/docs/en/setup/backend/backend-runtime-rule-api.md
index 858ff7a291..5af511fdb0 100644
--- a/docs/en/setup/backend/backend-runtime-rule-api.md
+++ b/docs/en/setup/backend/backend-runtime-rule-api.md
@@ -39,6 +39,43 @@ Default port is `17128`. All config knobs are in `application.yml` under the
All write endpoints take a raw body (`Content-Type: text/plain`) and catalog + name via
query string. No JSON envelope — makes it trivial to `--data-binary @file.yaml`.
+### Content encoding
+
+Rule content is **UTF-8 text per YAML 1.2**. The API never base64-encodes content.
+
+- Raw responses (`Content-Type: application/x-yaml; charset=utf-8`) and raw
+ `/addOrUpdate` request bodies carry content byte-identically.
+- JSON responses encode content as a **standard JSON string** — special characters are
+ JSON-escaped (`"`, `\`, control characters become `\u00XX`, newlines become `\n`). A
+ standard JSON parser yields the original UTF-8 YAML; no additional decode step.
+
+Example JSON response from `GET /runtime/rule?catalog=otel-rules&name=vm` with
+`Accept: application/json`:
+
+```json
+{
+ "catalog": "otel-rules",
+ "name": "vm",
+ "status": "ACTIVE",
+ "source": "runtime",
+ "contentHash": "5f9b8d3e...",
+ "updateTime": 1714102400000,
+ "content": "expSuffix: instance(['host_name','host_ip'], Layer.OS_LINUX)\nmetricPrefix: meter_vm\nmetricsRules:\n - name: cpu_total_percentage\n exp: avg_node_cpu_utilization\n"
+}
+```
+
+The `\n` in the `content` string is the JSON escape for newline. After `JSON.parse()`
+the value is the YAML body that `/addOrUpdate` would accept verbatim.
+
+`contentHash` is the SHA-256 of the UTF-8 content bytes (lowercase hex). It is identical
+across raw and JSON modes — the JSON envelope's escaping does not affect the hash. The
+same hash is what the reconciler stores in `BundleState.contentHash` and what `/list`
+emits.
+
+`/addOrUpdate` reads the request body as UTF-8 regardless of the `Content-Type` header.
+Non-UTF-8 input is not validated at upload time; the YAML parser rejects it during the
+delta-classifier compile step and the response is `400 compile_failed`.
+
### Canonical routes
| Method | Path | Body | Effect |
@@ -47,8 +84,10 @@ query string. No JSON envelope — makes it trivial to `--data-binary @file.yaml
|--------|---------------------------------------------------------------------|---------------|-----------------------------------------------------------------------------|
| POST | `/runtime/rule/addOrUpdate?catalog=&name=[&allowStorageChange=true]` | raw rule YAML | compile, classify, apply locally with Suspend + DDL + verify, persist row |
| POST | `/runtime/rule/fix?catalog=&name=` | raw rule YAML | recovery — same flow as addOrUpdate with `allowStorageChange=true` forced |
-| POST | `/runtime/rule/inactivate?catalog=&name=` | empty | destructive — broadcasts Suspend, tears down handlers + DDL, persists INACTIVE |
-| POST | `/runtime/rule/delete?catalog=&name=` | empty | row-only — requires INACTIVE (returns 409 otherwise); removes the tombstone |
+| POST | `/runtime/rule/inactivate?catalog=&name=` | empty | soft-pause — broadcasts Suspend, tears down OAP-internal state, preserves backend measure, persists INACTIVE |
+| POST | `/runtime/rule/delete?catalog=&name=` | empty | destructive — requires INACTIVE (returns 409 otherwise); drops backend measure then removes the tombstone |
+| GET | `/runtime/rule?catalog=&name=` | | single rule fetch — DAO row → static fallback → 404. Raw YAML by default; JSON envelope on `Accept: application/json`. ETag for `If-None-Match`. |
+| GET | `/runtime/rule/bundled?catalog=[&withContent=false]` | | JSON list of every static (bundled) rule for the catalog with `overridden` flag joined from runtime rows |
| GET | `/runtime/rule/list` | | NDJSON per-row view merged with per-node reconciler state |
| GET | `/runtime/rule/dump[/<catalog>]` | | tar.gz of all rows + manifest.yaml (source for manual restore) |
diff --git a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/module/LogAnalyzerModule.java b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/module/LogAnalyzerModule.java
index 0af9406b4a..0137d993c2 100644
--- a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/module/LogAnalyzerModule.java
+++ b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/module/LogAnalyzerModule.java
@@ -33,13 +33,20 @@ public class LogAnalyzerModule extends ModuleDefine {
public Class[] services() {
return new Class[] {
ILogAnalyzerService.class,
- // Factory exposed so the runtime-rule plugin can compile + addOrReplace / remove
- // LAL rules at runtime. Runtime path reuses the same compile helper the startup
+ // LAL rule store — keyed by (Layer, ruleName). Each entry is a compiled LAL
+ // {@code DSL} (the `LalExpression` class generated from a `lal/*.yaml` filter
+ // block) that decides how to PARSE a log and EXTRACT fields. Owns the
+ // `lal` runtime-rule catalog: hot-update mutates this Factory directly via
+ // `addOrReplace` / `remove`, reusing the same `compile` helper the startup
// path uses — no duplicate DSL wiring.
LogFilterListener.Factory.class,
- // log-mal-rules inline-MAL registry. runtime-rule pushes a converter here on
- // /addOrUpdate and drops it on /inactivate, in step with what it does on the
- // otel-rules receiver — same interface, one dispatch by catalog.
+ // Inline-MAL converter store — keyed by string `<catalog>:<name>`. Each entry
+ // is a `MetricConvert` compiled from a `log-mal-rules/*.yaml` rule that
+ // AGGREGATES samples (emitted by LAL `metrics {}` blocks) into metrics. Owns
+ // the `log-mal-rules` runtime-rule catalog. The same `MalConverterRegistry`
+ // SPI lives in the meter-analyzer artifact (which is a library, not a
+ // ModuleDefine); the OTel receiver implements the same interface for the
+ // `otel-rules` catalog. Two implementations, two catalogs, one shared API.
MalConverterRegistry.class,
};
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/ManagementPersistentWorker.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/ManagementPersistentWorker.java
index 2d52c16637..8e413817c1 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/ManagementPersistentWorker.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/ManagementPersistentWorker.java
@@ -45,15 +45,4 @@ public class ManagementPersistentWorker extends AbstractWorker<ManagementData> {
log.error(e.getMessage(), e);
}
}
-
- /**
- * Synchronous write that propagates the underlying {@link IOException} instead of
- * swallowing it. Used by callers that depend on persist-is-commit semantics — notably the
- * runtime-rule REST handler, which must not return a success status to the operator when
- * the DB row never landed. Plain {@link #in} stays for async queued writes where logging
- * the failure is the right behaviour.
- */
- public void insertSync(ManagementData managementData) throws IOException {
- configDAO.insert(model, managementData);
- }
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/ManagementStreamProcessor.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/ManagementStreamProcessor.java
index 823cad0373..992466530e 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/ManagementStreamProcessor.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/ManagementStreamProcessor.java
@@ -36,7 +36,6 @@ import org.apache.skywalking.oap.server.core.storage.model.StorageManipulationOp
import org.apache.skywalking.oap.server.core.storage.type.StorageBuilder;
import org.apache.skywalking.oap.server.library.module.ModuleDefineHolder;
-import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.util.HashMap;
import java.util.Map;
@@ -63,24 +62,6 @@ public class ManagementStreamProcessor implements StreamProcessor<ManagementData
}
}
- /**
- * Synchronous write that propagates the underlying {@link IOException} to the caller.
- * Callers that need persist-is-commit semantics (notably the runtime-rule REST handler)
- * use this instead of {@link #in} so a failed DAO write never returns a false success
- * status to the operator. Throws {@link IllegalStateException} if no worker is registered
- * for the data's class — that indicates a module-boot-order bug, not a transient failure,
- * so loud is the right response.
- */
- public void insertSync(final ManagementData managementData) throws IOException {
- final ManagementPersistentWorker worker = workers.get(managementData.getClass());
- if (worker == null) {
- throw new IllegalStateException(
- "ManagementPersistentWorker not registered for " + managementData.getClass().getName()
- + " — stream registration missed this class at module start");
- }
- worker.insertSync(managementData);
- }
-
public void create(final ModuleDefineHolder moduleDefineHolder, final Stream stream, final Class<? extends ManagementData> streamClass) throws StorageException {
final StorageBuilderFactory storageBuilderFactory = moduleDefineHolder.find(StorageModule.NAME)
.provider()
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/rule/ext/StaticRuleRegistry.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/rule/ext/StaticRuleRegistry.java
index 57f3bc3257..6e01889a9f 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/rule/ext/StaticRuleRegistry.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/rule/ext/StaticRuleRegistry.java
@@ -19,7 +19,9 @@
package org.apache.skywalking.oap.server.core.rule.ext;
import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
import java.util.Collections;
+import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
@@ -114,6 +116,52 @@ public final class StaticRuleRegistry {
return Collections.unmodifiableMap(staticContent);
}
+ /**
+ * Return every {@code (name, content)} pair recorded under the given catalog, sorted
+ * lexicographically by {@code name}. Used by {@code GET /runtime/rule/bundled} to render
+ * the read-only seed view that Studio's catalogue browser merges with the runtime
+ * overrides from {@code GET /runtime/rule/list}. Empty list when no rules were recorded
+ * for the catalog (e.g., the catalog has no on-disk content, or the runtime-rule
+ * extension chain wasn't installed).
+ */
+ public List<NamedRule> findByCatalog(final String catalog) {
+ if (catalog == null) {
+ return Collections.emptyList();
+ }
+ final String prefix = catalog + ":";
+ final List<NamedRule> matches = new ArrayList<>();
+ for (final Map.Entry<String, String> e : staticContent.entrySet()) {
+ if (e.getKey().startsWith(prefix)) {
+ final String name = e.getKey().substring(prefix.length());
+ matches.add(new NamedRule(name, e.getValue()));
+ }
+ }
+ matches.sort((a, b) -> a.name.compareTo(b.name));
+ return matches;
+ }
+
+ /**
+ * Pair of (rule name, raw YAML content) returned by {@link #findByCatalog(String)}.
+ * Public so the REST handler can iterate the result directly without a tuple type.
+ */
+ public static final class NamedRule {
+ private final String name;
+ private final String content;
+
+ public NamedRule(final String name, final String content) {
+ this.name = name;
+ this.content = content;
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ public String getContent() {
+ return content;
+ }
+ }
+
/**
* Split the registry's {@code catalog:name} key back into its components. Centralised
* here so callers don't hardcode the separator.
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/management/RuntimeRuleManagementDAO.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/management/RuntimeRuleManagementDAO.java
index 41f5c9ccc1..bba9f68d07 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/management/RuntimeRuleManagementDAO.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/management/RuntimeRuleManagementDAO.java
@@ -20,17 +20,16 @@ package org.apache.skywalking.oap.server.core.storage.management;
import java.io.IOException;
import java.util.List;
+import org.apache.skywalking.oap.server.core.management.runtimerule.RuntimeRule;
import org.apache.skywalking.oap.server.core.storage.DAO;
/**
- * Read + delete DAO for runtime-managed MAL/LAL rule files.
- *
- * <p>Writes go through {@code ManagementStreamProcessor.getInstance().in(rule)} which uses
- * the generic {@link org.apache.skywalking.oap.server.core.storage.IManagementDAO#insert} —
- * that path works for every shipped storage backend without per-backend write code. This
- * interface only covers the read path (needed by the reconciler to pull all rows every tick)
- * and the delete path (needed by {@code /runtime/rule/delete} which is intentionally hard
- * delete; {@code /runtime/rule/inactivate} writes a status-flipped row through the write path).
+ * Per-backend read / write / delete DAO for runtime-managed MAL/LAL rule files. The generic
+ * {@link org.apache.skywalking.oap.server.core.storage.IManagementDAO#insert} path is not
+ * used: BanyanDB's generic impl never persists, and ES/JDBC short-circuit when the row
+ * already exists, which silently breaks {@code /addOrUpdate} and {@code /inactivate} (every
+ * call after the first becomes a no-op). Each backend implements upsert semantics directly
+ * here so the persist-is-commit invariant holds across all three.
*/
public interface RuntimeRuleManagementDAO extends DAO {
@@ -40,6 +39,18 @@ public interface RuntimeRuleManagementDAO extends DAO {
*/
List<RuntimeRuleFile> getAll() throws IOException;
+ /**
+ * Upsert by composite key (catalog, name). Replaces {@code content}, {@code status} and
+ * {@code updateTime} when the row already exists; inserts a new row otherwise. Both
+ * {@code /addOrUpdate} (every call after the first) and {@code /inactivate} (every call)
+ * depend on overwrite semantics — without it the operator sees 200 / structural_applied
+ * / inactivated while the backing row stays unchanged.
+ *
+ * @throws IOException when the underlying storage write fails. Callers translate this
+ * into a 5xx response so the operator does not get a false success.
+ */
+ void save(RuntimeRule rule) throws IOException;
+
/**
* Hard delete a runtime-rule file by composite key. Idempotent: if no record matches,
* implementations must return silently rather than throw. A successful return does not
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/ModelInstaller.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/ModelInstaller.java
index 6a442455db..30b381ba55 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/ModelInstaller.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/ModelInstaller.java
@@ -67,23 +67,26 @@ public abstract class ModelInstaller implements ModelRegistry.CreatingListener,
}
return;
}
- InstallInfo info = isExists(model, opt);
- if (info.isAllExist()) {
- return;
- }
+ // Honour the localCacheOnly contract — "no server RPCs". The earlier order called
+ // isExists first (which on ES/JDBC fires a backend read) and only then checked the
+ // policy, which made the contract a half-truth: peer-side ticks did issue read
+ // RPCs to the storage even though they could never act on the result. Move the
+ // policy gate ahead of isExists so a peer apply under localCacheOnly is genuinely
+ // zero-RPC.
if (opt.isLocalCacheOnly()) {
- // Policy blocks creation (typical peer-side local cache refresh). Record the
- // skip so the caller can surface it; don't log an info-level message because
- // it's normal on peers when the cluster main hasn't yet installed the resource.
opt.recordOutcome("table", model.getName(),
StorageManipulationOpt.Outcome.SKIPPED_NOT_ALLOWED,
"local-cache-only mode; main-node is expected to have installed this resource");
log.debug(
- "install info: {}. model: [{}] not installed; local-cache-only mode — skipping",
- info.buildInstallInfoMsg(), model.getName()
+ "install: model [{}] not installed; local-cache-only mode — skipping (no isExists probe)",
+ model.getName()
);
return;
}
+ InstallInfo info = isExists(model, opt);
+ if (info.isAllExist()) {
+ return;
+ }
log.info(
"install info: {}. table for model: [{}] not all required resources exist, creating or updating...",
info.buildInstallInfoMsg(), model.getName()
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/StorageModels.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/StorageModels.java
index 121cbbcce3..f5ae0d7df5 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/StorageModels.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/StorageModels.java
@@ -209,33 +209,60 @@ public class StorageModels implements IModelManager, ModelRegistry, ModelManipul
lock.unlock();
}
- for (final CreatingListener listener : listenersSnapshot) {
- listener.whenCreating(finalModel, opt);
+ // If a listener (e.g. the BanyanDB / ES installer) throws while creating the
+ // backing measure / index / table, roll the model out of the registry before
+ // letting the exception propagate. Without this, the model stays in `models` and
+ // the dedup check above short-circuits future retries — the listener never fires
+ // again for this model, and the storage stays half-built. The model is published
+ // to the registry only after all listeners succeed.
+ boolean committed = false;
+ try {
+ for (final CreatingListener listener : listenersSnapshot) {
+ listener.whenCreating(finalModel, opt);
+ }
+ committed = true;
+ } finally {
+ if (!committed) {
+ lock.lock();
+ try {
+ models.remove(finalModel);
+ } finally {
+ lock.unlock();
+ }
+ }
}
return finalModel;
}
/**
- * Remove every model registered through {@link #add(Class, int, Storage)} whose stream class equals
- * {@code streamClass}. This cascades across downsampling variants (Hour / Day / Minute for metrics).
- * Listener {@link CreatingListener#whenRemoving(Model)} is fired for each removed model after the lock
- * is released, so backend drop operations (BanyanDB delete-measure) do not block unrelated registry mutations.
+ * Remove every model registered through {@link #add(Class, int, Storage)} whose stream
+ * class equals {@code streamClass}. Cascades across downsampling variants (Hour / Day /
+ * Minute). Backend drop operations (BanyanDB delete-measure) run inside listener
+ * {@link CreatingListener#whenRemoving} BEFORE the model is removed from the registry,
+ * so a transient backend-drop failure leaves the model in place — the caller (typically
+ * runtime-rule {@code /inactivate} or the reconciler tick) sees the throw and can
+ * re-attempt the whole tear-down. Removing the model first would leave the registry
+ * out of sync with the backend (model gone, measure still present) and the next
+ * {@code remove} call's iteration would find nothing to drop.
*
- * @return the list of models that were removed, in the order they were discovered
+ * @return the list of models that were removed, in the order they were discovered. Empty
+ * if no model matched. If a listener throws, that listener's drop is left in an
+ * unknown state on the corresponding backend; this method propagates the first
+ * such error after attempting every listener × model pair (so partial successes
+ * on one backend don't block successes on another).
*/
@Override
public List<Model> remove(Class<?> streamClass, StorageManipulationOpt opt) throws StorageException {
- final List<Model> removed;
+ // Snapshot matching models without yet removing them. Backend cascade first, registry
+ // mutation only after every listener succeeds.
+ final List<Model> matching;
final List<CreatingListener> listenersSnapshot;
lock.lock();
try {
- removed = new ArrayList<>();
- final Iterator<Model> it = models.iterator();
- while (it.hasNext()) {
- final Model m = it.next();
+ matching = new ArrayList<>();
+ for (final Model m : models) {
if (Objects.equals(m.getStreamClass(), streamClass)) {
- removed.add(m);
- it.remove();
+ matching.add(m);
}
}
listenersSnapshot = new ArrayList<>(listeners);
@@ -244,7 +271,7 @@ public class StorageModels implements IModelManager, ModelRegistry, ModelManipul
}
StorageException firstError = null;
- for (final Model m : removed) {
+ for (final Model m : matching) {
for (final CreatingListener listener : listenersSnapshot) {
try {
listener.whenRemoving(m, opt);
@@ -257,9 +284,32 @@ public class StorageModels implements IModelManager, ModelRegistry, ModelManipul
}
}
if (firstError != null) {
+ // Leave models in the registry — backend state is uncertain on at least one
+ // listener, so the next retry needs to find them and re-fire whenRemoving.
+ // Listeners are required to be idempotent on the drop path (BanyanDB's
+ // delete-measure on a non-existent measure is a no-op; ES / JDBC dropTable for
+ // management data is a documented no-op).
throw firstError;
}
- return removed;
+ // All listeners succeeded for every matching model — drop them from the registry.
+ // Identity-based removal: avoids the {@code @EqualsAndHashCode} on {@link Model}
+ // matching a concurrent fresh add of the same stream class with different field
+ // combinations. The matching list was captured under the lock; identity stays stable.
+ lock.lock();
+ try {
+ for (final Model target : matching) {
+ final Iterator<Model> it = models.iterator();
+ while (it.hasNext()) {
+ if (it.next() == target) {
+ it.remove();
+ break;
+ }
+ }
+ }
+ } finally {
+ lock.unlock();
+ }
+ return matching;
}
private boolean isSuperDatasetModel(Class<?> aClass) {
diff --git a/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/analysis/worker/ManagementPersistentWorkerTest.java b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/analysis/worker/ManagementPersistentWorkerTest.java
index 541be3f5a6..eb72b0d042 100644
--- a/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/analysis/worker/ManagementPersistentWorkerTest.java
+++ b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/analysis/worker/ManagementPersistentWorkerTest.java
@@ -25,7 +25,6 @@ import org.apache.skywalking.oap.server.core.storage.model.Model;
import org.apache.skywalking.oap.server.library.module.ModuleDefineHolder;
import org.junit.jupiter.api.Test;
-import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
@@ -35,10 +34,10 @@ class ManagementPersistentWorkerTest {
@Test
void inSwallowsIoExceptionForAsyncCallers() throws IOException {
- // Legacy async path — in() catches IOException and only logs. The runtime-rule REST
- // handler's original persistRuleSync relied on this not throwing, but the swallow
- // silently broke the persist-is-commit invariant. Keep this behaviour for callers
- // that are OK with "fire and log" semantics.
+ // Async path — in() catches IOException and only logs. UITemplate / UIMenu callers
+ // accept this fire-and-forget contract; their DAOs swallow duplicate-row writes
+ // anyway. Runtime-rule moved off this path because it requires persist-is-commit
+ // semantics; that contract now lives on RuntimeRuleManagementDAO.save instead.
final ModuleDefineHolder holder = mock(ModuleDefineHolder.class);
final IManagementDAO dao = mock(IManagementDAO.class);
final Model model = mock(Model.class);
@@ -55,43 +54,4 @@ class ManagementPersistentWorkerTest {
// than short-circuiting before insert.
verify(dao).insert(model, data);
}
-
- @Test
- void insertSyncPropagatesIoException() throws IOException {
- // persist-is-commit path — insertSync must propagate so the REST handler returns a
- // server error instead of confirming "structural_applied" / "inactivated" to an
- // operator whose DB row never landed.
- final ModuleDefineHolder holder = mock(ModuleDefineHolder.class);
- final IManagementDAO dao = mock(IManagementDAO.class);
- final Model model = mock(Model.class);
- final ManagementPersistentWorker worker =
- new ManagementPersistentWorker(holder, model, dao);
- final ManagementData data = mock(ManagementData.class);
- doThrow(new IOException("db unavailable")).when(dao).insert(any(), any());
-
- final IOException raised = assertThrows(IOException.class, () -> worker.insertSync(data));
-
- // The raised exception is the exact one the DAO threw — no wrapping layer that would
- // hide the original cause in a REST handler's error body.
- assertThrows(IOException.class, () -> {
- throw raised;
- });
- verify(dao).insert(model, data);
- }
-
- @Test
- void insertSyncHappyPathCallsDao() throws IOException {
- // Happy path — successful DAO insert returns cleanly, nothing is swallowed or
- // wrapped. Mirror of in()'s happy path.
- final ModuleDefineHolder holder = mock(ModuleDefineHolder.class);
- final IManagementDAO dao = mock(IManagementDAO.class);
- final Model model = mock(Model.class);
- final ManagementPersistentWorker worker =
- new ManagementPersistentWorker(holder, model, dao);
- final ManagementData data = mock(ManagementData.class);
-
- worker.insertSync(data);
-
- verify(dao).insert(model, data);
- }
}
diff --git a/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/storage/model/StorageModelsTest.java b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/storage/model/StorageModelsTest.java
index 4a455c9217..8aee0e2d86 100644
--- a/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/storage/model/StorageModelsTest.java
+++ b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/storage/model/StorageModelsTest.java
@@ -61,6 +61,58 @@ public class StorageModelsTest {
DEFAULT_SCOPE_DEFINE_MOCKED_STATIC.close();
}
+ @Test
+ public void rolledBackOnListenerFailure() throws StorageException {
+ // A CreatingListener throw must NOT leave the model in `models`. Future retries
+ // would otherwise hit the dedup short-circuit and skip listeners entirely, leaving
+ // the storage stack permanently half-built.
+ StorageModels models = new StorageModels();
+ models.addModelListener((model, opt) -> {
+ throw new StorageException("simulated DDL failure");
+ });
+ Assertions.assertThrows(StorageException.class, () -> models.add(TestModel.class, -1,
+ new Storage("StorageModelsRollbackTest", false, DownSampling.Hour),
+ StorageManipulationOpt.fullInstall()));
+ // Registry must not retain the model — a retry would otherwise dedup-skip the
+ // listener instead of attempting the DDL again.
+ assertEquals(0, models.allModels().size());
+ }
+
+ @Test
+ public void removeKeepsModelOnListenerFailure() throws StorageException {
+ // remove() must keep the model in `models` if any whenRemoving listener throws —
+ // otherwise the registry diverges from the backend (model gone, BanyanDB measure
+ // still alive) and there's nothing for the retry path to find. Listeners are
+ // required to be idempotent on the drop, so re-firing them on retry is safe.
+ StorageModels models = new StorageModels();
+ models.add(TestModel.class, -1,
+ new Storage("StorageModelsRemoveRetryTest", false, DownSampling.Hour),
+ StorageManipulationOpt.fullInstall());
+ assertEquals(1, models.allModels().size());
+
+ // Listener that throws on remove (simulating BanyanDB delete-measure transient failure).
+ // Note: addModelListener fires whenCreating for already-added models, but our listener
+ // only overrides whenRemoving, so the catch-up call is a no-op via the default impl.
+ models.addModelListener(new ModelRegistry.CreatingListener() {
+ @Override
+ public void whenCreating(final Model model, final StorageManipulationOpt opt) {
+ // already-created catch-up — fine to no-op for this test
+ }
+
+ @Override
+ public void whenRemoving(final Model model, final StorageManipulationOpt opt) throws StorageException {
+ throw new StorageException("simulated dropTable failure");
+ }
+ });
+
+ Assertions.assertThrows(StorageException.class,
+ () -> models.remove(TestModel.class, StorageManipulationOpt.fullInstall()));
+ // Model must still be in the registry — the next retry needs to find and drive
+ // dropTable again. Otherwise the operator's /inactivate succeeds locally but the
+ // backend measure stays orphaned forever.
+ assertEquals(1, models.allModels().size());
+ }
+
@Test
public void testStorageModels() throws StorageException {
StorageModels models = new StorageModels();
diff --git a/oap-server/server-receiver-plugin/skywalking-runtime-rule-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/runtimerule/reconcile/Reconciler.java b/oap-server/server-receiver-plugin/skywalking-runtime-rule-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/runtimerule/reconcile/Reconciler.java
index 1ee393e120..32a62de5de 100644
--- a/oap-server/server-receiver-plugin/skywalking-runtime-rule-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/runtimerule/reconcile/Reconciler.java
+++ b/oap-server/server-receiver-plugin/skywalking-runtime-rule-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/runtimerule/reconcile/Reconciler.java
@@ -41,6 +41,7 @@ import org.apache.skywalking.oap.server.core.alarm.AlarmKernelService;
import org.apache.skywalking.oap.server.core.alarm.AlarmModule;
import org.apache.skywalking.oap.server.core.analysis.meter.MeterSystem;
import org.apache.skywalking.oap.server.core.storage.StorageModule;
+import org.apache.skywalking.oap.server.core.management.runtimerule.RuntimeRule;
import org.apache.skywalking.oap.server.core.storage.management.RuntimeRuleManagementDAO;
import org.apache.skywalking.oap.server.core.storage.model.IModelManager;
import org.apache.skywalking.oap.server.core.storage.model.Model;
@@ -540,8 +541,8 @@ public final class Reconciler {
// compared without a separate round-trip per bundle. When the DAO is unavailable we
// skip self-heal entirely — a false flip back to RUNNING is worse than a brief extra
// pause, and the next tick retries with storage hopefully available.
- final Map<String, String> dbContentHash = readCurrentDbContentHashes();
- if (dbContentHash == null) {
+ final Map<String, RuntimeRuleManagementDAO.RuntimeRuleFile> dbRules = readCurrentDbRules();
+ if (dbRules == null) {
log.debug("runtime-rule self-heal: storage DAO unavailable, skipping sweep");
return;
}
@@ -563,20 +564,31 @@ public final class Reconciler {
}
final String key = key(current.getCatalog(), current.getName());
- final String currentDbHash = dbContentHash.get(key);
+ final RuntimeRuleManagementDAO.RuntimeRuleFile currentDbRule = dbRules.get(key);
- // DB row gone — treat as "row deleted", the normal delta-application path on the
- // same tick will drop the bundle. Do not self-heal to RUNNING here.
- if (currentDbHash == null) {
- log.debug("runtime-rule self-heal: bundle {}/{} DB row gone; delta-apply will drop",
+ // DB rule gone — treat as "rule deleted", the normal delta-application path on
+ // the same tick will drop the bundle. Do not self-heal to RUNNING here.
+ if (currentDbRule == null) {
+ log.debug("runtime-rule self-heal: bundle {}/{} DB rule gone; delta-apply will drop",
current.getCatalog(), current.getName());
continue;
}
+ // DB rule is INACTIVE — the operator deliberately inactivated it. Self-healing
+ // back to RUNNING here would un-do their intent. The same tick's delta-apply
+ // path will route through the isInactive branch and drive the local teardown;
+ // leave SUSPENDED so dispatch stays parked until then.
+ if (RuntimeRule.STATUS_INACTIVE.equals(currentDbRule.getStatus())) {
+ log.debug("runtime-rule self-heal: bundle {}/{} DB rule INACTIVE; delta-apply "
+ + "will tear down — not resuming", current.getCatalog(), current.getName());
+ continue;
+ }
+
// DB advanced past the pre-suspend hash — main node's structural commit succeeded,
// the delta-application path will drive the new apply on this tick. Leave SUSPENDED
// for the apply pipeline to transition via its own RUNNING installation once the
// new handlers are in place.
+ final String currentDbHash = ContentHash.sha256Hex(currentDbRule.getContent());
if (!Objects.equals(currentDbHash, current.getContentHash())) {
log.debug("runtime-rule self-heal: bundle {}/{} DB advanced ({} → {}); "
+ "leaving delta-apply to handle",
@@ -601,12 +613,19 @@ public final class Reconciler {
}
/**
- * One DAO fetch per tick. Returns a map keyed by {@code "catalog:name"} with the sha256
- * of each rule file's content, or {@code null} when the DAO isn't resolvable (early boot, some
- * embedded test topologies). The caller treats {@code null} as "skip self-heal this
- * tick" — a correct-but-conservative default when we can't observe the DB.
+ * One DAO fetch per tick. Returns a map keyed by {@code "catalog:name"} of every persisted
+ * runtime rule (BanyanDB property / ES document / JDBC row — same logical entity, three
+ * shapes), or {@code null} when the DAO isn't resolvable (early boot, some embedded test
+ * topologies). The caller treats {@code null} as "skip self-heal this tick" — a correct-
+ * but-conservative default when we can't observe the persisted state.
+ *
+ * <p>The full {@link RuntimeRuleManagementDAO.RuntimeRuleFile} is captured (not just the
+ * content hash) so self-heal can distinguish "content unchanged + ACTIVE" (the pre-
+ * suspend bundle is still authoritative → resume) from "content unchanged + INACTIVE"
+ * (the operator deliberately inactivated → leave SUSPENDED for the delta-apply path to
+ * tear down).
*/
- private Map<String, String> readCurrentDbContentHashes() {
+ private Map<String, RuntimeRuleManagementDAO.RuntimeRuleFile> readCurrentDbRules() {
final RuntimeRuleManagementDAO dao;
try {
dao = moduleManager.find(StorageModule.NAME).provider()
@@ -614,17 +633,16 @@ public final class Reconciler {
} catch (final Throwable t) {
return null;
}
- final Map<String, String> hashes = new HashMap<>();
+ final Map<String, RuntimeRuleManagementDAO.RuntimeRuleFile> rules = new HashMap<>();
try {
for (final RuntimeRuleManagementDAO.RuntimeRuleFile ruleFile : dao.getAll()) {
- hashes.put(key(ruleFile.getCatalog(), ruleFile.getName()),
- ContentHash.sha256Hex(ruleFile.getContent()));
+ rules.put(key(ruleFile.getCatalog(), ruleFile.getName()), ruleFile);
}
} catch (final IOException ioe) {
log.debug("runtime-rule self-heal: DAO fetch failed this tick ({})", ioe.getMessage());
return null;
}
- return hashes;
+ return rules;
}
/**
@@ -715,6 +733,58 @@ public final class Reconciler {
// gets there first); peers converge via localCacheOnly so they populate
// MeterSystem + MetadataRegistry from the DB content without firing any
// server-side DDL. The main node already did drop+create on its REST path.
+ //
+ // ddl-debt promotion: this node previously ran localCacheOnly for this
+ // content (marker set, prototypes + Models cached locally, no backend DDL),
+ // and the current tick would run fullInstall. Without intervention the
+ // apply path classifies as NO_CHANGE (priorContent==newContent), the
+ // STRUCTURAL/NEW fall-through fires applier.apply, and MeterSystem.create
+ // short-circuits on the existing same-shape prototype before reaching
+ // MetricsStreamProcessor.create / ModelRegistry.add — so the listener-side
+ // BanyanDB define never re-fires. The finally block then clears the marker
+ // and the file's measure stays permanently absent on this main.
+ //
+ // Tear down the local bundle here so the subsequent apply enters with
+ // appliedContent[key]==null, classifies as NEW, runs DDL through the full
+ // listener chain, and verifies. invokeAlarmOnRemove=false because we
+ // immediately re-create the same metrics — the alarm windows for those
+ // metrics shouldn't be cleared. Cost: one bundle's recompile per
+ // promotion event, which is rare (cluster main re-routing only).
+ //
+ // The teardown explicitly uses {@link StorageManipulationOpt#localCacheOnly}
+ // even though the surrounding tick runs fullInstall. The cluster backend is
+ // shared: if another node had already been main and ran the BanyanDB
+ // {@code define} for this measure, a fullInstall teardown here would fire
+ // {@code dropTable} via the listener chain and DELETE the cluster-wide
+ // measure (and its data) before the subsequent re-apply re-creates it.
+ // localCacheOnly skips the listener-side drop, leaving the backend
+ // untouched; the re-apply's create-if-absent path then either creates the
+ // measure (if missing) or no-ops (if another main created it). Either way,
+ // the marker debt is correctly discharged without risking shared data.
+ if (tickRunsFullInstall
+ && pendingFullInstall.contains(key)
+ && prev != null
+ && dbActive
+ && localEffectivelyActive
+ && Objects.equals(prev.getContentHash(), newHash)) {
+ log.info("runtime-rule reconciler: ddl-debt promotion for {}/{} — "
+ + "local bundle was registered under localCacheOnly; tearing down "
+ + "(local-only) to drive fresh fullInstall through the listener chain",
+ ruleFile.getCatalog(), ruleFile.getName());
+ try {
+ unregisterBundle(ruleFile.getCatalog(), ruleFile.getName(),
+ false, StorageManipulationOpt.localCacheOnly());
+ } catch (final Throwable t) {
+ // Tear-down failure leaves local + marker state intact (per the
+ // backend-cascade-first contract); next tick retries the same
+ // promotion path. Don't drop into applyOneRuleFile in that case —
+ // priorContent is still set and the apply would NO_CHANGE again.
+ log.warn("runtime-rule reconciler: ddl-debt promotion teardown "
+ + "threw for {}/{}; next tick will retry",
+ ruleFile.getCatalog(), ruleFile.getName(), t);
+ continue;
+ }
+ }
try {
applyOneRuleFile(ruleFile, newHash, prev, nowMs, key, false, tickOpt);
} catch (final Throwable t) {
@@ -766,24 +836,32 @@ public final class Reconciler {
}
try (HistogramMetrics.Timer ignored =
lockMetrics.startTickHoldTimer()) {
- final BundleState prev = snapshot.remove(gone);
+ final BundleState prev = snapshot.get(gone);
log.info("runtime-rule reconciler: rule file deleted {} (last hash={})",
gone, prev == null ? "?" : shortHash(prev.getContentHash()));
- if (prev != null) {
+ if (prev == null) {
+ // No local state to tear down — clean up the empty snapshot entry if any
+ // and continue.
+ snapshot.remove(gone);
+ } else {
// Rule file is gone from storage — full tear-down, no new bundle incoming,
// so drive alarm reset for every metric the prior bundle owned. Reuse the
// per-tick storage policy snapshot so the gone-key cleanup is consistent
// with the apply-loop decisions above.
+ //
+ // Snapshot removal is deferred to AFTER unregisterBundle succeeds. If
+ // unregisterBundle throws (transient backend drop failure, etc.), the
+ // snapshot stays so the next tick still finds this key in the
+ // removedKeys set above and retries the teardown. The earlier order
+ // (snapshot.remove first) silently lost the retry: removedKeys is built
+ // from snapshot.keySet() minus seenKeys (DB rows), and once the snapshot
+ // entry is gone the next tick sees nothing to retry.
try {
unregisterBundle(parts[0], parts[1], true, tickOpt);
+ snapshot.remove(gone);
} catch (final Throwable t) {
- // Same per-rule isolation as the apply loop above. A teardown failure
- // (e.g., partial backend drop) must not strand other removed keys; the
- // next tick re-attempts via this same path because the DB row is still
- // absent and the snapshot entry was removed above (teardown is idempotent
- // on a re-run — meterPrototypes.remove returns false, short-circuits).
log.warn("runtime-rule reconciler: teardown threw for removed rule {}; "
- + "tick continues, next tick will retry", gone, t);
+ + "snapshot retained — next tick will retry", gone, t);
}
}
} finally {
@@ -933,6 +1011,90 @@ public final class Reconciler {
}
}
+ /**
+ * Drop the BanyanDB measure (and any sibling backend schema) that {@code /inactivate}
+ * deliberately preserved under the soft-pause design. {@code /inactivate} runs the
+ * apply path under {@link StorageManipulationOpt#localCacheOnly()} so OAP-internal
+ * state (prototypes, Models, appliedMal/Content, classloader) is torn down while the
+ * backend measure stays alive — operators can re-activate cheaply via {@code /addOrUpdate}
+ * without losing data. {@code /delete} is the one destructive endpoint: this method is
+ * how it discharges the backend debt before the row is removed.
+ *
+ * <p>At call time the local state is gone (cleared by {@code /inactivate}), so the
+ * existing {@link #unregisterBundle} would no-op on the MAL side (priorMal is null and
+ * MeterSystem.removeMetric short-circuits on the missing prototype). The fix: re-register
+ * prototypes locally under {@code localCacheOnly} (no DDL, no listener-side define), then
+ * drive {@link #unregisterBundle} under {@code fullInstall} so the listener chain runs
+ * BanyanDB delete-measure on the existing measure. Local re-registration is wasteful
+ * (one bundle's recompile) but keeps the destructive path within the existing
+ * MeterSystem → ModelRegistry → ModelInstaller cascade rather than reaching for a
+ * direct backend client.
+ *
+ * <p>LAL catalogs are skipped here — LAL has no backend schema (DSL handlers are
+ * in-process only). The row deletion alone discharges everything LAL owns.
+ *
+ * <p>Idempotent on every layer: BanyanDB delete-measure on a missing measure is a
+ * documented no-op, MeterSystem.removeMetric on a missing prototype short-circuits,
+ * StorageModels.remove on no-match returns an empty list. A re-execution after a
+ * partial failure (e.g., applier.apply succeeded but unregisterBundle threw) finds
+ * the re-registered state and resumes the teardown.
+ *
+ * <p>Caller (REST {@code /delete}) holds the per-file lock; this method re-acquires
+ * it (lock is reentrant) so the implementation is correct whether called inline or
+ * from a background path.
+ */
+ public void dropBundleBackendForDelete(final String catalog, final String name, final String content) {
+ if (!isMalCatalog(catalog)) {
+ return;
+ }
+ final String key = key(catalog, name);
+ final String hash = ContentHash.sha256Hex(content);
+ final String sourceName = catalog + "/" + name;
+ final ReentrantLock perFile = lockMap.lockFor(catalog, name);
+ perFile.lock();
+ try {
+ final MalFileApplier applier = resolveMalApplier();
+ if (applier == null) {
+ // MeterSystem unavailable — we cannot drop the backend measure through
+ // the listener chain. Propagate so the caller aborts dao.delete; an
+ // orphaned measure with no row left to drive a retry is the worse
+ // failure mode. Operator can re-attempt /delete once MeterSystem is
+ // back up.
+ throw new IllegalStateException(
+ "MeterSystem unavailable; cannot drop backend measure for "
+ + catalog + "/" + name + " — refusing to delete the row "
+ + "and orphan the measure. Retry when MeterSystem is up.");
+ }
+ // Re-register prototypes locally so the destructive cascade has Models and
+ // meterPrototypes entries to walk. localCacheOnly suppresses listener-side
+ // backend define — we don't want to recreate the measure we're about to drop.
+ try {
+ final MalFileApplier.Applied applied = applier.apply(
+ content, sourceName, hash, StorageManipulationOpt.localCacheOnly());
+ appliedMal.put(key, applied);
+ appliedContent.put(key, content);
+ } catch (final MalFileApplier.ApplyException ae) {
+ // Re-register failed (compile error in the inactive row's content, or
+ // partial register that rolled back). Without re-registered prototypes
+ // unregisterBundle's cascade would no-op the BanyanDB delete-measure
+ // and the row deletion would orphan it. Propagate so the caller aborts
+ // dao.delete. Operator can either fix the YAML and retry, or fall back
+ // to a manual backend cleanup.
+ throw new IllegalStateException(
+ "re-register for backend drop failed for " + catalog + "/" + name
+ + "; refusing to delete the row to avoid orphaning the measure. "
+ + "Cause: " + ae.getMessage(), ae);
+ }
+ // Tear down with fullInstall: drops backend (listener whenRemoving fires
+ // dropTable for each downsampling variant) and clears the re-registered
+ // local state. invokeAlarmOnRemove=true: this is the destructive endpoint,
+ // alarm windows for the affected metrics should reset.
+ unregisterBundle(catalog, name, true, StorageManipulationOpt.fullInstall());
+ } finally {
+ perFile.unlock();
+ }
+ }
+
/**
* Drain the pending commit for {@code (catalog, name)} after the REST handler confirms
* the row-persist succeeded. Drops {@code removedMetrics} from MeterSystem, swaps
@@ -1257,16 +1419,30 @@ public final class Reconciler {
// Marker bookkeeping. See the type-level comment on {@link #pendingFullInstall}
// for the principle. Two branches, no readiness re-query:
//
- // fullInstall → REMOVE: this node directly observed itself running DDL
- // for the affected models; the debt is discharged.
- // localCacheOnly → ADD: regardless of WHY we ran localCacheOnly (cold-boot
- // fallback, steady-state peer, doesn't matter), this node
- // has not observed itself running DDL. Mark the debt and
- // let the third-gate skip in tick() decide whether to
- // re-apply: peers short-circuit on hash match (their job
- // is just localCacheOnly anyway), main re-applies.
+ // fullInstall + apply succeeded → REMOVE: this node directly observed
+ // itself running DDL for the affected
+ // models; the debt is discharged.
+ // fullInstall + apply failed → KEEP: applier.apply / verifyPostApply
+ // threw or rolled back. Snapshot has
+ // lastApplyError set; the next tick must
+ // retry instead of hash-skipping. Clearing
+ // here would let the third-gate short-circuit
+ // on a still-broken backend state.
+ // localCacheOnly → ADD: regardless of WHY we ran
+ // localCacheOnly (cold-boot fallback,
+ // steady-state peer, doesn't matter), this
+ // node has not observed itself running DDL.
+ // Mark the debt and let the third-gate skip
+ // in tick() decide whether to re-apply:
+ // peers short-circuit on hash match (their
+ // job is just localCacheOnly anyway), main
+ // re-applies.
if (storageOpt.isFullInstall()) {
- pendingFullInstall.remove(key);
+ final BundleState post = snapshot.get(key);
+ final boolean applySucceeded = post != null && post.getLastApplyError() == null;
+ if (applySucceeded) {
+ pendingFullInstall.remove(key);
+ }
} else if (storageOpt.isLocalCacheOnly()) {
pendingFullInstall.add(key);
}
diff --git a/oap-server/server-receiver-plugin/skywalking-runtime-rule-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/runtimerule/rest/RuntimeRuleRestHandler.java b/oap-server/server-receiver-plugin/skywalking-runtime-rule-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/runtimerule/rest/RuntimeRuleRestHandler.java
index a91da8cced..541d585d90 100644
--- a/oap-server/server-receiver-plugin/skywalking-runtime-rule-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/runtimerule/rest/RuntimeRuleRestHandler.java
+++ b/oap-server/server-receiver-plugin/skywalking-runtime-rule-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/runtimerule/rest/RuntimeRuleRestHandler.java
@@ -19,14 +19,18 @@
package org.apache.skywalking.oap.server.receiver.runtimerule.rest;
import com.linecorp.armeria.common.HttpData;
+import com.linecorp.armeria.common.HttpHeaderNames;
import com.linecorp.armeria.common.HttpResponse;
import com.linecorp.armeria.common.HttpStatus;
import com.linecorp.armeria.common.MediaType;
+import com.linecorp.armeria.common.ResponseHeaders;
import com.linecorp.armeria.server.annotation.Blocking;
import com.linecorp.armeria.server.annotation.Default;
import com.linecorp.armeria.server.annotation.Get;
+import com.linecorp.armeria.server.annotation.Header;
import com.linecorp.armeria.server.annotation.Param;
import com.linecorp.armeria.server.annotation.Post;
+import java.util.Locale;
import java.io.IOException;
import java.util.Arrays;
import java.util.Collections;
@@ -395,6 +399,202 @@ public class RuntimeRuleRestHandler implements ForwardTarget {
MediaType.create("application", "x-ndjson"), out.toString());
}
+ /**
+ * Single-rule fetch. Studio's catalog → row click and the editor both need the YAML
+ * source, which {@code /list} intentionally omits. Lookup order:
+ * <ol>
+ * <li>DAO row for {@code (catalog, name)} regardless of status — INACTIVE rules keep
+ * their content under the soft-pause contract so the editor can re-edit.</li>
+ * <li>{@link StaticRuleRegistry} fallback — bundled rules that have never been
+ * overridden by the operator. Returned with synthetic status {@code STATIC} and
+ * source {@code static}.</li>
+ * <li>Otherwise 404 {@code not_found}.</li>
+ * </ol>
+ *
+ * <p>Default response is raw YAML ({@code Content-Type: application/x-yaml; charset=utf-8})
+ * so a round-trip through {@code /addOrUpdate} is byte-exact. With {@code Accept:
+ * application/json} the response is the envelope {@code {catalog, name, status, source,
+ * contentHash, updateTime, content}} where {@code content} is a standard JSON-escaped
+ * UTF-8 string (no base64). Either mode emits the same metadata as response headers
+ * ({@code X-Sw-Content-Hash}, {@code X-Sw-Status}, {@code X-Sw-Source},
+ * {@code X-Sw-Update-Time}) and an {@code ETag} based on the content hash, so an editor
+ * reload with {@code If-None-Match} gets a cheap 304.
+ *
+ * <p>No cluster routing — reads are stateless and any node can serve from its local
+ * DAO + {@link StaticRuleRegistry}.
+ */
+ @Get("/runtime/rule")
+ public HttpResponse get(@Param("catalog") String catalog,
+ @Param("name") String name,
+ @Header("Accept") @Default("") String accept,
+ @Header("If-None-Match") @Default("") String ifNoneMatch) {
+ final HttpResponse validationError = validate(catalog, name);
+ if (validationError != null) {
+ return validationError;
+ }
+ final RuntimeRuleManagementDAO dao = resolveDao();
+ if (dao == null) {
+ return serverError("dao_unavailable", catalog, name,
+ "RuntimeRuleManagementDAO not resolvable — storage module may not be active");
+ }
+
+ // 1. DAO row.
+ final RuntimeRuleManagementDAO.RuntimeRuleFile row;
+ try {
+ row = findRule(dao, catalog, name);
+ } catch (final IOException ioe) {
+ log.warn("runtime-rule /get: DAO lookup failed for {}/{}", catalog, name, ioe);
+ return HttpResponse.of(HttpStatus.SERVICE_UNAVAILABLE, MediaType.JSON_UTF_8,
+ jsonBody("storage_unavailable", catalog, name,
+ "DAO lookup failed: " + ioe.getMessage()));
+ }
+ if (row != null) {
+ return renderGetResponse(catalog, name, row.getContent(), row.getStatus(),
+ "runtime", row.getUpdateTime(), accept, ifNoneMatch);
+ }
+
+ // 2. Static fallback.
+ final String staticContent = StaticRuleRegistry.active().find(catalog, name).orElse(null);
+ if (staticContent != null) {
+ // updateTime=0 for static fallback — capturing the actual file mtime would require
+ // threading it through StaticRuleRegistry.record from the catalog loaders. The
+ // editor doesn't need a precise value here; "0" is honest about "we don't track it".
+ return renderGetResponse(catalog, name, staticContent, "STATIC",
+ "static", 0L, accept, ifNoneMatch);
+ }
+
+ // 3. 404.
+ return HttpResponse.of(HttpStatus.NOT_FOUND, MediaType.JSON_UTF_8,
+ jsonBody("not_found", catalog, name,
+ "no DB row and no bundled static rule for this (catalog, name)"));
+ }
+
+ /**
+ * Build the {@code GET /runtime/rule} response. Honours {@code Accept: application/json}
+ * for the JSON envelope; defaults to raw YAML otherwise. Always emits the metadata
+ * headers and {@code ETag} so the raw and JSON modes are equally introspectable.
+ * Returns {@code 304 Not Modified} when the client's {@code If-None-Match} matches the
+ * current content hash.
+ */
+ private static HttpResponse renderGetResponse(final String catalog, final String name,
+ final String content, final String status,
+ final String source, final long updateTime,
+ final String accept, final String ifNoneMatch) {
+ final String contentHash = ContentHash.sha256Hex(content);
+ final String eTag = "\"" + contentHash + "\"";
+ if (eTag.equals(ifNoneMatch == null ? "" : ifNoneMatch.trim())) {
+ return HttpResponse.of(
+ ResponseHeaders.builder(HttpStatus.NOT_MODIFIED)
+ .add("X-Sw-Content-Hash", contentHash)
+ .add("X-Sw-Status", status)
+ .add("X-Sw-Source", source)
+ .add("X-Sw-Update-Time", Long.toString(updateTime))
+ .add(HttpHeaderNames.ETAG, eTag)
+ .build());
+ }
+ final boolean json = accept != null
+ && accept.toLowerCase(Locale.ROOT).contains("application/json");
+ if (json) {
+ final String body = "{"
+ + "\"catalog\":\"" + escape(catalog) + "\","
+ + "\"name\":\"" + escape(name) + "\","
+ + "\"status\":\"" + escape(status) + "\","
+ + "\"source\":\"" + escape(source) + "\","
+ + "\"contentHash\":\"" + escape(contentHash) + "\","
+ + "\"updateTime\":" + updateTime + ","
+ + "\"content\":\"" + escape(content) + "\""
+ + "}";
+ return HttpResponse.of(
+ ResponseHeaders.builder(HttpStatus.OK)
+ .contentType(MediaType.JSON_UTF_8)
+ .add("X-Sw-Content-Hash", contentHash)
+ .add("X-Sw-Status", status)
+ .add("X-Sw-Source", source)
+ .add("X-Sw-Update-Time", Long.toString(updateTime))
+ .add(HttpHeaderNames.ETAG, eTag)
+ .build(),
+ HttpData.ofUtf8(body));
+ }
+ return HttpResponse.of(
+ ResponseHeaders.builder(HttpStatus.OK)
+ .contentType(MediaType.create("application", "x-yaml").withCharset(StandardCharsets.UTF_8))
+ .add("X-Sw-Content-Hash", contentHash)
+ .add("X-Sw-Status", status)
+ .add("X-Sw-Source", source)
+ .add("X-Sw-Update-Time", Long.toString(updateTime))
+ .add(HttpHeaderNames.ETAG, eTag)
+ .build(),
+ HttpData.ofUtf8(content == null ? "" : content));
+ }
+
+ /**
+ * Read-only view of every static rule shipped with OAP for the given catalog. Studio's
+ * catalogue browser merges this with {@code /list} (runtime overrides) for a unified
+ * "available rules" view; the {@code overridden} flag on each entry tells the UI which
+ * static rules currently have an operator override in place so they can be rendered
+ * with the right state.
+ *
+ * <p>Always JSON: the body is an array of {@code {name, kind, contentHash, content?,
+ * overridden}} objects. {@code content} is included by default and elided when
+ * {@code withContent=false} so a catalogue browse can stay small (per-rule content can
+ * then be fetched lazily via {@code GET /runtime/rule}).
+ *
+ * <p>Catalog scope: {@code otel-rules}, {@code log-mal-rules}, {@code lal} — the same
+ * allowlist the write paths use. {@code .oal} files are not exposed here; they live
+ * outside the runtime-rule plugin's scope today.
+ */
+ @Get("/runtime/rule/bundled")
+ public HttpResponse listBundled(@Param("catalog") String catalog,
+ @Param("withContent") @Default("true") String withContentRaw) {
+ if (!VALID_CATALOGS.contains(catalog)) {
+ return badRequest("invalid_catalog", catalog, null,
+ "catalog must be one of " + VALID_CATALOGS);
+ }
+ final boolean withContent = parseFlag(withContentRaw)
+ || "true".equalsIgnoreCase(withContentRaw == null ? "true" : withContentRaw.trim());
+ // Cross-join with the DAO so each entry's `overridden` flag reflects current state.
+ // Failure to read the DAO is non-fatal — we still return the bundled view; just mark
+ // every entry overridden=false (best-effort) and log so operators can see the gap.
+ final Set<String> overriddenNames = new HashSet<>();
+ final RuntimeRuleManagementDAO dao = resolveDao();
+ if (dao != null) {
+ try {
+ for (final RuntimeRuleManagementDAO.RuntimeRuleFile rule : dao.getAll()) {
+ if (catalog.equals(rule.getCatalog())) {
+ overriddenNames.add(rule.getName());
+ }
+ }
+ } catch (final IOException ioe) {
+ log.warn("runtime-rule /bundled: DAO read failed for catalog={}; "
+ + "overridden flags will all be false this call", catalog, ioe);
+ }
+ }
+ final List<StaticRuleRegistry.NamedRule> rules =
+ StaticRuleRegistry.active().findByCatalog(catalog);
+ final StringBuilder out = new StringBuilder();
+ out.append('[');
+ boolean first = true;
+ for (final StaticRuleRegistry.NamedRule rule : rules) {
+ if (!first) {
+ out.append(',');
+ }
+ first = false;
+ final String contentHash = ContentHash.sha256Hex(rule.getContent());
+ final boolean overridden = overriddenNames.contains(rule.getName());
+ out.append('{');
+ out.append("\"name\":\"").append(escape(rule.getName())).append("\",");
+ out.append("\"kind\":\"bundled\",");
+ out.append("\"contentHash\":\"").append(contentHash).append("\",");
+ out.append("\"overridden\":").append(overridden);
+ if (withContent) {
+ out.append(",\"content\":\"").append(escape(rule.getContent())).append("\"");
+ }
+ out.append('}');
+ }
+ out.append(']');
+ return HttpResponse.of(HttpStatus.OK, MediaType.JSON_UTF_8, out.toString());
+ }
+
@Get("/runtime/rule/dump")
public HttpResponse dump() {
return doDump(null);
@@ -1399,13 +1599,15 @@ public class RuntimeRuleRestHandler implements ForwardTarget {
private HttpResponse doDeleteLocked(final String catalog, final String name,
final RuntimeRuleManagementDAO dao) {
- // /delete is a pure row-removal. The destructive work (Suspend broadcast, DDL teardown,
- // alarm-window reset, peer handler unregister) runs on /inactivate — the only path that
- // takes an ACTIVE rule offline. /delete requires the rule to already be INACTIVE: the
- // backend schema is already torn down, the handlers are already unregistered, and only
- // the tombstone row remains. This makes deletion cheap, reversible-up-to-the-last-moment
- // via /addOrUpdate (which reactivates an INACTIVE row), and removes the failure modes
- // that arise from mixing state flip + DDL in a single endpoint.
+ // /delete is the one destructive endpoint. /inactivate is a soft-pause that runs the
+ // OAP-internal teardown under localCacheOnly, deliberately preserving the BanyanDB
+ // measure + its data so a re-activation via /addOrUpdate is cheap and lossless.
+ // /delete drops the backend measure first, then removes the tombstone row.
+ //
+ // The two-step workflow (/inactivate → /delete) is enforced by the INACTIVE-status
+ // check below: an ACTIVE rule cannot be deleted in one shot. This separation makes
+ // the destructive moment explicit and lets operators reverse the soft-pause for a
+ // bounded window before committing to data loss.
final RuntimeRuleManagementDAO.RuntimeRuleFile prior;
try {
prior = findRule(dao, catalog, name);
@@ -1424,8 +1626,22 @@ public class RuntimeRuleRestHandler implements ForwardTarget {
return HttpResponse.of(HttpStatus.CONFLICT, MediaType.JSON_UTF_8,
jsonBody("requires_inactivate_first", catalog, name,
"rule is ACTIVE; POST /runtime/rule/inactivate first, then /runtime/rule/delete. "
- + "Inactivate tears down the backend schema and handlers; delete removes the "
- + "tombstone row."));
+ + "Inactivate runs the soft-pause (handlers stop dispatching; backend "
+ + "measure preserved); delete drops the backend measure and removes the row."));
+ }
+ // Backend drop. /inactivate preserved the BanyanDB measure under localCacheOnly;
+ // discharge that debt now via the reconciler before the row goes away. LAL has no
+ // backend schema so the call is a no-op for the lal catalog. A throw here aborts
+ // the row deletion — the operator can retry, or fall back to a manual cleanup if
+ // the backend is genuinely unreachable. We do NOT proceed with dao.delete on
+ // backend-drop failure: that would orphan the measure with no way to find it again.
+ if (reconciler != null) {
+ try {
+ reconciler.dropBundleBackendForDelete(catalog, name, prior.getContent());
+ } catch (final Throwable t) {
+ log.error("runtime-rule /delete: backend drop threw for {}/{}", catalog, name, t);
+ return serverError("delete_backend_drop_failed", catalog, name, t.getMessage());
+ }
}
try {
dao.delete(catalog, name);
@@ -1434,7 +1650,8 @@ public class RuntimeRuleRestHandler implements ForwardTarget {
return serverError("delete_failed", catalog, name, e.getMessage());
}
return ok(HttpStatus.OK, "deleted", catalog, name,
- "tombstone row removed from storage; rule reverts to static version on disk if one exists");
+ "backend measure dropped, tombstone row removed from storage; rule reverts to "
+ + "static version on disk if one exists");
}
/**
@@ -1535,11 +1752,52 @@ public class RuntimeRuleRestHandler implements ForwardTarget {
escape(name == null ? "" : name), escape(message == null ? "" : message));
}
+ /**
+ * RFC 8259 §7 JSON-string escape. Handles {@code "}, {@code \}, control chars
+ * (newlines, tabs, etc. — the case that the original two-char replace did NOT cover),
+ * and other characters below {@code U+0020}. Non-ASCII printable characters pass through
+ * unchanged — the response is {@code application/json; charset=utf-8} so multi-byte
+ * UTF-8 (Chinese comments, emoji in tags) is carried as-is.
+ */
private static String escape(final String s) {
if (s == null) {
return "";
}
- return s.replace("\\", "\\\\").replace("\"", "\\\"");
+ final StringBuilder sb = new StringBuilder(s.length() + 8);
+ for (int i = 0; i < s.length(); i++) {
+ final char c = s.charAt(i);
+ switch (c) {
+ case '"':
+ sb.append("\\\"");
+ break;
+ case '\\':
+ sb.append("\\\\");
+ break;
+ case '\n':
+ sb.append("\\n");
+ break;
+ case '\r':
+ sb.append("\\r");
+ break;
+ case '\t':
+ sb.append("\\t");
+ break;
+ case '\b':
+ sb.append("\\b");
+ break;
+ case '\f':
+ sb.append("\\f");
+ break;
+ default:
+ if (c < 0x20) {
+ sb.append(String.format("\\u%04x", (int) c));
+ } else {
+ sb.append(c);
+ }
+ break;
+ }
+ }
+ return sb.toString();
}
/**
@@ -1638,9 +1896,9 @@ public class RuntimeRuleRestHandler implements ForwardTarget {
final String filename = "runtime-rule-dump-" + dumpedAt.replace(":", "-") + ".tar.gz";
return HttpResponse.of(
- com.linecorp.armeria.common.ResponseHeaders.builder(HttpStatus.OK)
+ ResponseHeaders.builder(HttpStatus.OK)
.contentType(MediaType.OCTET_STREAM)
- .add(com.linecorp.armeria.common.HttpHeaderNames.CONTENT_DISPOSITION,
+ .add(HttpHeaderNames.CONTENT_DISPOSITION,
"attachment; filename=\"" + filename + "\"")
.build(),
HttpData.copyOf(buffer.toByteArray()));
@@ -1649,7 +1907,7 @@ public class RuntimeRuleRestHandler implements ForwardTarget {
private static void writeTarEntry(
final org.apache.commons.compress.archivers.tar.TarArchiveOutputStream tar,
final String path, final String body) throws IOException {
- final byte[] bytes = body == null ? new byte[0] : body.getBytes(java.nio.charset.StandardCharsets.UTF_8);
+ final byte[] bytes = body == null ? new byte[0] : body.getBytes(StandardCharsets.UTF_8);
final org.apache.commons.compress.archivers.tar.TarArchiveEntry entry =
new org.apache.commons.compress.archivers.tar.TarArchiveEntry(path);
entry.setSize(bytes.length);
diff --git a/oap-server/server-receiver-plugin/skywalking-runtime-rule-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/runtimerule/rest/GuardrailIntegrationTest.java b/oap-server/server-receiver-plugin/skywalking-runtime-rule-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/runtimerule/rest/GuardrailIntegrationTest.java
index 8817c04a0e..5566c4bfee 100644
--- a/oap-server/server-receiver-plugin/skywalking-runtime-rule-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/runtimerule/rest/GuardrailIntegrationTest.java
+++ b/oap-server/server-receiver-plugin/skywalking-runtime-rule-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/runtimerule/rest/GuardrailIntegrationTest.java
@@ -22,14 +22,9 @@ import com.linecorp.armeria.common.HttpData;
import com.linecorp.armeria.common.HttpResponse;
import com.linecorp.armeria.common.HttpStatus;
import com.linecorp.armeria.common.ResponseHeaders;
-import java.lang.reflect.Field;
import java.util.Arrays;
-import java.util.Map;
import java.util.concurrent.locks.ReentrantLock;
import org.apache.skywalking.oap.server.core.CoreModule;
-import org.apache.skywalking.oap.server.core.analysis.management.ManagementData;
-import org.apache.skywalking.oap.server.core.analysis.worker.ManagementPersistentWorker;
-import org.apache.skywalking.oap.server.core.analysis.worker.ManagementStreamProcessor;
import org.apache.skywalking.oap.server.core.management.runtimerule.RuntimeRule;
import org.apache.skywalking.oap.server.core.storage.StorageModule;
import org.apache.skywalking.oap.server.core.storage.management.RuntimeRuleManagementDAO;
@@ -42,7 +37,6 @@ import org.apache.skywalking.oap.server.receiver.runtimerule.reconcile.Reconcile
import org.apache.skywalking.oap.server.receiver.runtimerule.state.BundleState;
import org.apache.skywalking.oap.server.receiver.runtimerule.state.PerFileLockMap;
import org.apache.skywalking.oap.server.telemetry.api.HistogramMetrics;
-import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
@@ -130,7 +124,6 @@ class GuardrailIntegrationTest {
return true;
});
when(lockMetrics.startRestHoldTimer()).thenReturn(mock(HistogramMetrics.Timer.class));
- installMockPersistWorker();
when(reconciler.localSuspend(Mockito.anyString(), Mockito.anyString()))
.thenReturn(org.apache.skywalking.oap.server.receiver.runtimerule.reconcile.Reconciler.SuspendResult.SUSPENDED);
@@ -159,39 +152,6 @@ class GuardrailIntegrationTest {
handler = new RuntimeRuleRestHandler(moduleManager, reconciler, clusterClient, 30_000L);
}
- @AfterEach
- void tearDown() {
- uninstallMockPersistWorker();
- }
-
- @SuppressWarnings("unchecked")
- private static void installMockPersistWorker() {
- try {
- final Field workersField = ManagementStreamProcessor.class.getDeclaredField("workers");
- workersField.setAccessible(true);
- final Map<Class<? extends ManagementData>, ManagementPersistentWorker> workers =
- (Map<Class<? extends ManagementData>, ManagementPersistentWorker>)
- workersField.get(ManagementStreamProcessor.getInstance());
- workers.put(RuntimeRule.class, mock(ManagementPersistentWorker.class));
- } catch (final ReflectiveOperationException e) {
- throw new RuntimeException("failed to install mock ManagementPersistentWorker", e);
- }
- }
-
- @SuppressWarnings("unchecked")
- private static void uninstallMockPersistWorker() {
- try {
- final Field workersField = ManagementStreamProcessor.class.getDeclaredField("workers");
- workersField.setAccessible(true);
- final Map<Class<? extends ManagementData>, ManagementPersistentWorker> workers =
- (Map<Class<? extends ManagementData>, ManagementPersistentWorker>)
- workersField.get(ManagementStreamProcessor.getInstance());
- workers.remove(RuntimeRule.class);
- } catch (final ReflectiveOperationException ignored) {
- // best-effort
- }
- }
-
// ---- MAL scenarios --------------------------------------------------------------------
@Test
diff --git a/oap-server/server-receiver-plugin/skywalking-runtime-rule-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/runtimerule/rest/RuntimeRuleRestHandlerTest.java b/oap-server/server-receiver-plugin/skywalking-runtime-rule-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/runtimerule/rest/RuntimeRuleRestHandlerTest.java
index 029045f902..2487ecf055 100644
--- a/oap-server/server-receiver-plugin/skywalking-runtime-rule-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/runtimerule/rest/RuntimeRuleRestHandlerTest.java
+++ b/oap-server/server-receiver-plugin/skywalking-runtime-rule-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/runtimerule/rest/RuntimeRuleRestHandlerTest.java
@@ -18,16 +18,21 @@
package org.apache.skywalking.oap.server.receiver.runtimerule.rest;
+import com.linecorp.armeria.common.AggregatedHttpResponse;
import com.linecorp.armeria.common.HttpData;
+import com.linecorp.armeria.common.HttpHeaderNames;
import com.linecorp.armeria.common.HttpResponse;
import com.linecorp.armeria.common.HttpStatus;
import com.linecorp.armeria.common.ResponseHeaders;
+import java.lang.reflect.Method;
+import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.locks.ReentrantLock;
import org.apache.skywalking.oap.server.core.CoreModule;
import org.apache.skywalking.oap.server.core.management.runtimerule.RuntimeRule;
+import org.apache.skywalking.oap.server.core.rule.ext.StaticRuleRegistry;
import org.apache.skywalking.oap.server.core.storage.StorageModule;
import org.apache.skywalking.oap.server.core.storage.management.RuntimeRuleManagementDAO;
import org.apache.skywalking.oap.server.core.storage.model.StorageManipulationOpt;
@@ -39,6 +44,7 @@ import org.apache.skywalking.oap.server.receiver.runtimerule.metrics.LockMetrics
import org.apache.skywalking.oap.server.receiver.runtimerule.reconcile.Reconciler;
import org.apache.skywalking.oap.server.receiver.runtimerule.state.BundleState;
import org.apache.skywalking.oap.server.receiver.runtimerule.state.PerFileLockMap;
+import org.apache.skywalking.oap.server.receiver.runtimerule.util.ContentHash;
import org.apache.skywalking.oap.server.telemetry.api.HistogramMetrics;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -279,6 +285,209 @@ class RuntimeRuleRestHandlerTest {
verify(reconciler, never()).applyNowForRuleFile(any(), Mockito.anyBoolean());
}
+ // ---- GET /runtime/rule and /runtime/rule/bundled ---------------------------------------
+
+ @Test
+ void getRuleReturnsRowYamlWhenActive() throws Exception {
+ final String yaml = minimalMalYaml();
+ whenDaoHasRow("otel-rules", "vm", yaml, RuntimeRule.STATUS_ACTIVE);
+
+ final HttpResponse resp = handler.get("otel-rules", "vm", "", "");
+
+ assertHttpStatus(resp, HttpStatus.OK);
+ final AggregatedHttpResponse agg =
+ resp.aggregate().toCompletableFuture().join();
+ // Default mode = raw YAML, byte-identical to /addOrUpdate's input.
+ assertEquals(yaml, agg.contentUtf8());
+ // Metadata headers always present so raw and JSON modes are equally introspectable.
+ assertEquals("ACTIVE", agg.headers().get("X-Sw-Status"));
+ assertEquals("runtime", agg.headers().get("X-Sw-Source"));
+ // ETag matches contentHash so editor reload can do If-None-Match.
+ assertEquals("\"" + sha256Hex(yaml) + "\"",
+ agg.headers().get(HttpHeaderNames.ETAG));
+ }
+
+ @Test
+ void getRuleReturnsRowYamlWhenInactive() throws Exception {
+ // Soft-pause contract: an INACTIVE row keeps its original content so the editor can
+ // re-edit. Status header should reflect the actual DB state, not "ACTIVE".
+ final String yaml = minimalMalYaml();
+ whenDaoHasRow("otel-rules", "vm", yaml, RuntimeRule.STATUS_INACTIVE);
+
+ final HttpResponse resp = handler.get("otel-rules", "vm", "", "");
+
+ assertHttpStatus(resp, HttpStatus.OK);
+ final AggregatedHttpResponse agg =
+ resp.aggregate().toCompletableFuture().join();
+ assertEquals(yaml, agg.contentUtf8());
+ assertEquals("INACTIVE", agg.headers().get("X-Sw-Status"));
+ }
+
+ @Test
+ void getRuleFallsBackToStaticWhenNoRow() throws Exception {
+ // No DB row, but StaticRuleRegistry has a bundled rule. Studio's catalog browser
+ // displays bundled rules with the same shape as runtime ones; the editor needs to
+ // be able to fetch their content too. Source header distinguishes the two.
+ clearStaticRegistry();
+ final String yaml = "bundled: true\n";
+ StaticRuleRegistry.active()
+ .record("otel-rules", "bundled-only", yaml.getBytes(StandardCharsets.UTF_8));
+ whenDaoHasRow("otel-rules", "bundled-only", null, null);
+ try {
+ final HttpResponse resp = handler.get("otel-rules", "bundled-only", "", "");
+
+ assertHttpStatus(resp, HttpStatus.OK);
+ final AggregatedHttpResponse agg =
+ resp.aggregate().toCompletableFuture().join();
+ assertEquals(yaml, agg.contentUtf8());
+ assertEquals("STATIC", agg.headers().get("X-Sw-Status"));
+ assertEquals("static", agg.headers().get("X-Sw-Source"));
+ } finally {
+ clearStaticRegistry();
+ }
+ }
+
+ @Test
+ void getRuleReturns404WhenNoRowAndNoStatic() throws Exception {
+ clearStaticRegistry();
+ whenDaoHasRow("otel-rules", "absent", null, null);
+
+ final HttpResponse resp = handler.get("otel-rules", "absent", "", "");
+
+ assertHttpStatus(resp, HttpStatus.NOT_FOUND);
+ }
+
+ @Test
+ void getRuleReturnsJsonEnvelopeOnAcceptJson() throws Exception {
+ // JSON envelope must use standard JSON-string escaping (no base64). Multi-line YAML
+ // → \n in the content field; a JSON parser yields the original bytes back.
+ final String yaml = "metricPrefix: vm\nmetricsRules:\n - name: cpu\n";
+ whenDaoHasRow("otel-rules", "vm", yaml, RuntimeRule.STATUS_ACTIVE);
+
+ final HttpResponse resp = handler.get("otel-rules", "vm", "application/json", "");
+
+ assertHttpStatus(resp, HttpStatus.OK);
+ final AggregatedHttpResponse agg =
+ resp.aggregate().toCompletableFuture().join();
+ final String body = agg.contentUtf8();
+ assertEquals(true, body.startsWith("{") && body.endsWith("}"),
+ "expected JSON envelope, got: " + body);
+ // Newline in the YAML must be JSON-escaped, NOT raw and NOT base64.
+ assertEquals(true, body.contains("\\n"),
+ "expected JSON-escaped newline in content field, got: " + body);
+ assertEquals(true, body.contains("\"source\":\"runtime\""),
+ "expected source=runtime, got: " + body);
+ }
+
+ @Test
+ void getRuleReturns304OnIfNoneMatchHashMatch() throws Exception {
+ final String yaml = minimalMalYaml();
+ whenDaoHasRow("otel-rules", "vm", yaml, RuntimeRule.STATUS_ACTIVE);
+ final String currentETag = "\"" + sha256Hex(yaml) + "\"";
+
+ final HttpResponse resp = handler.get("otel-rules", "vm", "", currentETag);
+
+ assertHttpStatus(resp, HttpStatus.NOT_MODIFIED);
+ // 304 still emits metadata headers so the editor can refresh its cached state
+ // without re-downloading the body.
+ final AggregatedHttpResponse agg =
+ resp.aggregate().toCompletableFuture().join();
+ assertEquals(currentETag, agg.headers().get(HttpHeaderNames.ETAG));
+ }
+
+ @Test
+ void getRuleReturns400OnInvalidCatalog() throws Exception {
+ final HttpResponse resp = handler.get("not-a-catalog", "vm", "", "");
+
+ assertHttpStatus(resp, HttpStatus.BAD_REQUEST);
+ }
+
+ @Test
+ void listBundledReturnsAllForCatalog() throws Exception {
+ clearStaticRegistry();
+ final org.apache.skywalking.oap.server.core.rule.ext.StaticRuleRegistry registry =
+ StaticRuleRegistry.active();
+ registry.record("otel-rules", "alpha", "alpha\n".getBytes(StandardCharsets.UTF_8));
+ registry.record("otel-rules", "beta", "beta\n".getBytes(StandardCharsets.UTF_8));
+ // A different catalog's entries must be excluded from this catalog's response.
+ registry.record("lal", "gamma", "gamma\n".getBytes(StandardCharsets.UTF_8));
+ whenDaoHasRow("otel-rules", "absent", null, null); // empty DAO so overridden=false everywhere
+ try {
+ final HttpResponse resp = handler.listBundled("otel-rules", "true");
+
+ assertHttpStatus(resp, HttpStatus.OK);
+ final String body = resp.aggregate().toCompletableFuture().join().contentUtf8();
+ assertEquals(true, body.contains("\"name\":\"alpha\""),
+ "expected alpha in bundled list, got: " + body);
+ assertEquals(true, body.contains("\"name\":\"beta\""),
+ "expected beta in bundled list, got: " + body);
+ assertEquals(false, body.contains("\"name\":\"gamma\""),
+ "lal entry leaked into otel-rules list: " + body);
+ assertEquals(true, body.contains("\"kind\":\"bundled\""));
+ } finally {
+ clearStaticRegistry();
+ }
+ }
+
+ @Test
+ void listBundledOmitsContentWhenWithContentFalse() throws Exception {
+ clearStaticRegistry();
+ StaticRuleRegistry.active()
+ .record("otel-rules", "alpha", "alpha\n".getBytes(StandardCharsets.UTF_8));
+ whenDaoHasRow("otel-rules", "absent", null, null);
+ try {
+ final HttpResponse resp = handler.listBundled("otel-rules", "false");
+
+ assertHttpStatus(resp, HttpStatus.OK);
+ final String body = resp.aggregate().toCompletableFuture().join().contentUtf8();
+ // contentHash must always be present so Studio can decide whether to fetch content
+ // lazily; content must be absent when withContent=false.
+ assertEquals(true, body.contains("\"contentHash\""),
+ "expected contentHash, got: " + body);
+ assertEquals(false, body.contains("\"content\":\""),
+ "expected no content field when withContent=false, got: " + body);
+ } finally {
+ clearStaticRegistry();
+ }
+ }
+
+ @Test
+ void listBundledMarksOverriddenWhenRuntimeRowExists() throws Exception {
+ clearStaticRegistry();
+ StaticRuleRegistry.active()
+ .record("otel-rules", "vm", "static-vm\n".getBytes(StandardCharsets.UTF_8));
+ // Operator override exists for "vm" via a runtime row.
+ whenDaoHasRow("otel-rules", "vm", "override-vm\n", RuntimeRule.STATUS_ACTIVE);
+ try {
+ final HttpResponse resp = handler.listBundled("otel-rules", "true");
+
+ assertHttpStatus(resp, HttpStatus.OK);
+ final String body = resp.aggregate().toCompletableFuture().join().contentUtf8();
+ assertEquals(true, body.contains("\"overridden\":true"),
+ "expected overridden=true for the bundled rule, got: " + body);
+ } finally {
+ clearStaticRegistry();
+ }
+ }
+
+ @Test
+ void listBundledReturns400OnInvalidCatalog() throws Exception {
+ final HttpResponse resp = handler.listBundled("not-a-catalog", "true");
+
+ assertHttpStatus(resp, HttpStatus.BAD_REQUEST);
+ }
+
+ /** Reflection helper — StaticRuleRegistry.clear() is package-private. */
+ private static void clearStaticRegistry() throws Exception {
+ final Method m = StaticRuleRegistry.class.getDeclaredMethod("clear");
+ m.setAccessible(true);
+ m.invoke(StaticRuleRegistry.active());
+ }
+
+ private static String sha256Hex(final String s) {
+ return ContentHash.sha256Hex(s);
+ }
+
// ---- helpers --------------------------------------------------------------------------
private void whenDaoHasRow(final String catalog, final String name,
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBRuntimeRuleManagementDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBRuntimeRuleManagementDAO.java
index 193398d989..e8a3c8456a 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBRuntimeRuleManagementDAO.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBRuntimeRuleManagementDAO.java
@@ -22,6 +22,7 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import lombok.extern.slf4j.Slf4j;
+import org.apache.skywalking.banyandb.common.v1.BanyandbCommon;
import org.apache.skywalking.banyandb.model.v1.BanyandbModel;
import org.apache.skywalking.banyandb.property.v1.BanyandbProperty.Property;
import org.apache.skywalking.library.banyandb.v1.client.TagAndValue;
@@ -30,10 +31,11 @@ import org.apache.skywalking.oap.server.core.storage.management.RuntimeRuleManag
import org.apache.skywalking.oap.server.storage.plugin.banyandb.stream.AbstractBanyanDBDAO;
/**
- * BanyanDB read + delete for {@link RuntimeRule}. Stored as a BanyanDB {@code Property} —
- * consistent with {@link BanyanDBUITemplateManagementDAO}. Writes go through the generic
- * {@code IManagementDAO.insert} path which ManagementStreamProcessor handles; this DAO only
- * covers {@link #getAll()} and {@link #delete(String, String)}.
+ * BanyanDB read / write / delete for {@link RuntimeRule}. Stored as a BanyanDB
+ * {@code Property} — consistent with {@link BanyanDBUITemplateManagementDAO}. Writes go
+ * through {@link #save(RuntimeRule)} and use {@code PropertyStore.apply}, which is upsert
+ * by id. The generic {@code BanyanDBManagementDAO.insert} path is intentionally not used:
+ * its body just logs and returns without persisting.
*/
@Slf4j
public class BanyanDBRuntimeRuleManagementDAO extends AbstractBanyanDBDAO implements RuntimeRuleManagementDAO {
@@ -52,6 +54,31 @@ public class BanyanDBRuntimeRuleManagementDAO extends AbstractBanyanDBDAO implem
return files;
}
+ @Override
+ public void save(final RuntimeRule rule) throws IOException {
+ final MetadataRegistry.Schema schema =
+ MetadataRegistry.INSTANCE.findManagementMetadata(RuntimeRule.INDEX_NAME);
+ if (schema == null) {
+ throw new IOException(
+ "BanyanDB schema for " + RuntimeRule.INDEX_NAME + " not registered yet");
+ }
+ // Property id matches RuntimeRule.id() (catalog + "_" + name) so apply() upserts on
+ // the same composite key the read path keys off. Default apply strategy is MERGE,
+ // but every save() call writes all five tags, so MERGE behaves like full replace.
+ final Property property = Property.newBuilder()
+ .setMetadata(BanyandbCommon.Metadata.newBuilder()
+ .setGroup(schema.getMetadata().getGroup())
+ .setName(RuntimeRule.INDEX_NAME))
+ .setId(rule.id().build())
+ .addTags(TagAndValue.newStringTag(RuntimeRule.CATALOG, rule.getCatalog()).build())
+ .addTags(TagAndValue.newStringTag(RuntimeRule.NAME, rule.getName()).build())
+ .addTags(TagAndValue.newStringTag(RuntimeRule.CONTENT, rule.getContent()).build())
+ .addTags(TagAndValue.newStringTag(RuntimeRule.STATUS, rule.getStatus()).build())
+ .addTags(TagAndValue.newLongTag(RuntimeRule.UPDATE_TIME, rule.getUpdateTime()).build())
+ .build();
+ getClient().apply(property);
+ }
+
@Override
public void delete(final String catalog, final String name) throws IOException {
// BanyanDB property id matches the StorageID composite built by RuntimeRule.id().
diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/RuntimeRuleManagementEsDAO.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/RuntimeRuleManagementEsDAO.java
index 782d1f763b..d476e7d178 100644
--- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/RuntimeRuleManagementEsDAO.java
+++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/RuntimeRuleManagementEsDAO.java
@@ -20,6 +20,7 @@ package org.apache.skywalking.oap.server.storage.plugin.elasticsearch.query;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import lombok.extern.slf4j.Slf4j;
@@ -70,6 +71,24 @@ public class RuntimeRuleManagementEsDAO extends ManagementCRUDEsDAO implements R
return files;
}
+ @Override
+ public void save(final RuntimeRule rule) throws IOException {
+ final String index =
+ IndexController.LogicIndicesRegister.getPhysicalTableName(RuntimeRule.INDEX_NAME);
+ final String docId = RuntimeRule.INDEX_NAME + "_" + rule.getCatalog() + "_" + rule.getName();
+ final Map<String, Object> source = new HashMap<>();
+ source.put(RuntimeRule.CATALOG, rule.getCatalog());
+ source.put(RuntimeRule.NAME, rule.getName());
+ source.put(RuntimeRule.CONTENT, rule.getContent());
+ source.put(RuntimeRule.STATUS, rule.getStatus());
+ source.put(RuntimeRule.UPDATE_TIME, rule.getUpdateTime());
+ source.put(IndexController.LogicIndicesRegister.MANAGEMENT_TABLE_NAME, RuntimeRule.INDEX_NAME);
+ // forceInsert maps to the ES `index` API which is upsert-by-_id; existing docs with
+ // the same id are replaced. The base class `create` / `update` helpers explicitly
+ // gate on existDoc, which is exactly the bug we're avoiding here.
+ getClient().forceInsert(index, docId, source);
+ }
+
@Override
public void delete(final String catalog, final String name) throws IOException {
final String index =
diff --git a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/common/dao/JDBCRuntimeRuleManagementDAO.java b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/common/dao/JDBCRuntimeRuleManagementDAO.java
index 6b601f0d96..82d5bbf622 100644
--- a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/common/dao/JDBCRuntimeRuleManagementDAO.java
+++ b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/common/dao/JDBCRuntimeRuleManagementDAO.java
@@ -19,6 +19,7 @@
package org.apache.skywalking.oap.server.storage.plugin.jdbc.common.dao;
import java.io.IOException;
+import java.sql.Connection;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
@@ -27,7 +28,11 @@ import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.skywalking.oap.server.core.management.runtimerule.RuntimeRule;
import org.apache.skywalking.oap.server.core.storage.management.RuntimeRuleManagementDAO;
+import org.apache.skywalking.oap.server.core.storage.model.Model;
+import org.apache.skywalking.oap.server.core.storage.type.HashMapConverter;
import org.apache.skywalking.oap.server.library.client.jdbc.hikaricp.JDBCClient;
+import org.apache.skywalking.oap.server.storage.plugin.jdbc.SQLExecutor;
+import org.apache.skywalking.oap.server.storage.plugin.jdbc.TableMetaInfo;
import org.apache.skywalking.oap.server.storage.plugin.jdbc.common.JDBCTableInstaller;
import org.apache.skywalking.oap.server.storage.plugin.jdbc.common.TableHelper;
@@ -75,6 +80,58 @@ public class JDBCRuntimeRuleManagementDAO extends JDBCSQLExecutor implements Run
return files;
}
+ @Override
+ public void save(final RuntimeRule rule) throws IOException {
+ final Model model = TableMetaInfo.get(RuntimeRule.INDEX_NAME);
+ final RuntimeRule.Builder builder = new RuntimeRule.Builder();
+ // The shared {@link JDBCSQLExecutor#getByID} is unusable for ManagementData like
+ // RuntimeRule: it always passes the lookup id through {@code TableHelper.generateId(
+ // String, String)} which prefixes with the model name, but the INSERT path uses
+ // {@code TableHelper.generateId(Model, String)} which returns the RAW id for
+ // non-record / non-function-metric types. RuntimeRule is non-record + non-metric, so
+ // its row is stored with the raw composite id while getByID looks up "runtimerule_<id>"
+ // and never finds it. Without this workaround the second save() always falls through
+ // to INSERT and trips the primary-key constraint, breaking every /addOrUpdate update
+ // and every /inactivate after the first persist.
+ final String storedId = TableHelper.generateId(model, rule.id().build());
+ try (Connection connection = jdbcClient.getConnection()) {
+ final boolean exists = rowExists(connection, storedId);
+ final SQLExecutor executor;
+ if (exists) {
+ executor = getUpdateExecutor(model, rule, 0, builder, null);
+ } else {
+ executor = getInsertExecutor(
+ model, rule, 0, builder, new HashMapConverter.ToStorage(), null);
+ }
+ executor.invoke(connection);
+ } catch (final SQLException e) {
+ throw new IOException("failed to save runtime rule "
+ + rule.getCatalog() + ":" + rule.getName(), e);
+ }
+ }
+
+ /**
+ * Probe every TTL-shadow table for a row with the given stored id. Direct SELECT on the
+ * id column rather than the shared getByID helper for the prefix-mismatch reason
+ * documented in {@link #save(RuntimeRule)}.
+ */
+ private boolean rowExists(final Connection connection, final String storedId) throws SQLException {
+ final List<String> tables = tableHelper.getTablesWithinTTL(RuntimeRule.INDEX_NAME);
+ for (final String table : tables) {
+ final String sql = "SELECT 1 FROM " + table
+ + " WHERE " + JDBCTableInstaller.ID_COLUMN + " = ? LIMIT 1";
+ try (final var stmt = connection.prepareStatement(sql)) {
+ stmt.setString(1, storedId);
+ try (final var rs = stmt.executeQuery()) {
+ if (rs.next()) {
+ return true;
+ }
+ }
+ }
+ }
+ return false;
+ }
+
@Override
public void delete(final String catalog, final String name) throws IOException {
final List<String> tables = tableHelper.getTablesWithinTTL(RuntimeRule.INDEX_NAME);
diff --git a/test/e2e-v2/cases/runtime-rule-storage-matrix/banyandb/docker-compose.yml b/test/e2e-v2/cases/runtime-rule-storage-matrix/banyandb/docker-compose.yml
index fba76e10c7..6536b0755d 100644
--- a/test/e2e-v2/cases/runtime-rule-storage-matrix/banyandb/docker-compose.yml
+++ b/test/e2e-v2/cases/runtime-rule-storage-matrix/banyandb/docker-compose.yml
@@ -23,16 +23,12 @@ services:
SW_RECEIVER_RUNTIME_RULE: default
# Steer OAP to BanyanDB. Reuses the standard env var the base-compose already wires.
SW_STORAGE: banyandb
- # Empty out every static rule catalog so the runtime-rule apply path is the only
- # writer of MAL / LAL state. First-time DDL on the runtime apply must succeed
- # against a backend with no pre-created measure for the metric we push.
- SW_OTEL_RECEIVER_ENABLED_OTEL_METRICS_RULES: ""
- SW_METER_ANALYZER_ACTIVE_FILES: ""
- SW_LOG_MAL_RULES: ""
- SW_LAL_FILES: ""
- # OAL is OAP's metric DSL for tracing data — the runtime-rule scope test pushes
- # MAL only, so OAL stays at its default (active rules) and the existing alarm /
- # trace paths keep working.
+ # Static rule catalogs stay at their defaults. The runtime-rule lifecycle uses a
+ # unique e2e_rr_-prefixed metric name that does not collide with anything the
+ # static catalogs ship, so the apply path's CREATE phase still exercises a
+ # first-time DDL register against the backend for that specific metric — even
+ # though the storage-side merging tables (meter_sum_*) may already exist for
+ # other static metrics that share the function.
ports:
- "11800:11800"
- "12800:12800"
diff --git a/test/e2e-v2/cases/runtime-rule-storage-matrix/banyandb/e2e.yaml b/test/e2e-v2/cases/runtime-rule-storage-matrix/banyandb/e2e.yaml
index 7d2d8eb57b..6aedaec434 100644
--- a/test/e2e-v2/cases/runtime-rule-storage-matrix/banyandb/e2e.yaml
+++ b/test/e2e-v2/cases/runtime-rule-storage-matrix/banyandb/e2e.yaml
@@ -42,14 +42,6 @@ setup:
https://github.com/jqlang/jq/releases/download/jq-1.7.1/jq-linux-amd64
chmod +x /tmp/skywalking-infra-e2e/bin/jq
fi
- - name: install bydbctl (BanyanDB CLI used by the storage-truth probe)
- command: |
- if ! command -v bydbctl >/dev/null 2>&1; then
- # Pulled out of the running banyandb container — same binary as the server,
- # so version skew is impossible.
- docker cp $(docker compose -f test/e2e-v2/cases/runtime-rule-storage-matrix/banyandb/docker-compose.yml ps -q banyandb):/usr/local/bin/bydbctl /tmp/skywalking-infra-e2e/bin/bydbctl
- chmod +x /tmp/skywalking-infra-e2e/bin/bydbctl
- fi
- name: drive runtime-rule lifecycle (CREATE → FILTER_ONLY → STRUCTURAL → INACTIVATE → DELETE)
command: |
set -euo pipefail
@@ -58,13 +50,15 @@ setup:
export OAP_REST_PORT=17128
export OAP_GQL_PORT=12800
export SEED_RULES_DIR=$(pwd)/test/e2e-v2/cases/runtime-rule-storage-matrix/seed-rules
- # BanyanDB-side probe — runs against the test-host's bydbctl (port 17913
- # exposed by docker-compose). Kind=measure; group is the runtime-rule
- # metric's minute group, name is "${METRIC}_minute" (the convention
- # MalSchemaUpdateIT codified).
- export BYDBCTL_ADDR="http://127.0.0.1:17913"
- export STORAGE_PROBE_EXISTS='bydbctl measure get --group=sw_metricsMinute --name="${METRIC}_minute" --addr="${BYDBCTL_ADDR}" >/dev/null 2>&1'
- export STORAGE_PROBE_ABSENT='! bydbctl measure get --group=sw_metricsMinute --name="${METRIC}_minute" --addr="${BYDBCTL_ADDR}" >/dev/null 2>&1'
+ # BanyanDB-side probe via HTTP API on port 17913 (exposed by docker-compose).
+ # Avoids installing bydbctl on the test host — the HTTP shape is documented
+ # and stable across BanyanDB versions, and has no version-skew risk against
+ # the server image because we go through the gRPC-gateway.
+ export BYDB_BASE='http://127.0.0.1:17913'
+ # GET /api/v1/measure/schema/{group}/{name} → 200 if the measure exists,
+ # 404 (or non-2xx) otherwise. Curl --fail returns non-zero on non-2xx.
+ export STORAGE_PROBE_EXISTS='curl -fsS "${BYDB_BASE}/api/v1/measure/schema/sw_metricsMinute/${METRIC}_minute" >/dev/null'
+ export STORAGE_PROBE_ABSENT='! curl -fsS "${BYDB_BASE}/api/v1/measure/schema/sw_metricsMinute/${METRIC}_minute" >/dev/null 2>&1'
bash test/e2e-v2/cases/runtime-rule-storage-matrix/runtime-rule-flow.sh
verify:
diff --git a/test/e2e-v2/cases/runtime-rule-storage-matrix/elasticsearch/docker-compose.yml b/test/e2e-v2/cases/runtime-rule-storage-matrix/elasticsearch/docker-compose.yml
index c952280782..3b6b4c000d 100644
--- a/test/e2e-v2/cases/runtime-rule-storage-matrix/elasticsearch/docker-compose.yml
+++ b/test/e2e-v2/cases/runtime-rule-storage-matrix/elasticsearch/docker-compose.yml
@@ -38,13 +38,11 @@ services:
SW_STORAGE: elasticsearch
SW_STORAGE_ES_CLUSTER_NODES: es:9200
SW_RECEIVER_RUNTIME_RULE: default
- # Empty static rule catalogs so the runtime-rule apply path is the only writer
- # of MAL / LAL state. logicSharding stays at its default (false), so all metric
- # functions land in a single merged metrics-all index.
- SW_OTEL_RECEIVER_ENABLED_OTEL_METRICS_RULES: ""
- SW_METER_ANALYZER_ACTIVE_FILES: ""
- SW_LOG_MAL_RULES: ""
- SW_LAL_FILES: ""
+ # Static rule catalogs stay at their defaults — the e2e_rr_-prefixed metric name
+ # the runtime-rule lifecycle uses doesn't collide with anything static. ES default
+ # logicSharding=false means all metrics merge into a single metrics-all index;
+ # CREATE still exercises a first-time register for our specific metric even
+ # though the index may already exist for other static metrics.
ports:
- "11800:11800"
- "12800:12800"
diff --git a/test/e2e-v2/cases/runtime-rule-storage-matrix/postgresql/docker-compose.yml b/test/e2e-v2/cases/runtime-rule-storage-matrix/postgresql/docker-compose.yml
index 0ab80f0ed2..57e8ad0db8 100644
--- a/test/e2e-v2/cases/runtime-rule-storage-matrix/postgresql/docker-compose.yml
+++ b/test/e2e-v2/cases/runtime-rule-storage-matrix/postgresql/docker-compose.yml
@@ -24,13 +24,10 @@ services:
SW_DATA_SOURCE_USER: postgres
SW_DATA_SOURCE_PASSWORD: root@1234
SW_RECEIVER_RUNTIME_RULE: default
- # Empty static rule catalogs so the runtime-rule apply path is the only writer
- # of MAL / LAL state — first-time DDL on the runtime apply must succeed against
- # PostgreSQL with no pre-created merging-table for the metric we push.
- SW_OTEL_RECEIVER_ENABLED_OTEL_METRICS_RULES: ""
- SW_METER_ANALYZER_ACTIVE_FILES: ""
- SW_LOG_MAL_RULES: ""
- SW_LAL_FILES: ""
+ # Static rule catalogs stay at their defaults — the e2e_rr_-prefixed metric name
+ # the runtime-rule lifecycle uses doesn't collide with anything static, so CREATE
+ # still exercises a first-time register for that specific metric even though the
+ # merging-table (meter_sum_*) may already exist for other static metrics.
ports:
- "11800:11800"
- "12800:12800"