[Feature] Store BanyanDB's own logs in BanyanDB (self-stored logs)

### Search before asking

- [X] I had searched in the [issues](https://github.com/apache/skywalking/issues?q=is%3Aissue) and found no similar feature requirement.

### Description

## 1. Feature introduction

BanyanDB self-stores its **metrics**, but not its **logs**.

| | Metrics | Logs |
|---|---|---|
| Destinations | **2** — Prometheus + self-storage | **1** — console only |
| Abstraction | `meter.Provider` (pluggable) | a raw `io.Writer` |
| Fan-out | ✅ `factory` → every provider | ❌ |
| Lifecycle service | ✅ `metricService` | ❌ |
| Stored in BanyanDB | ✅ `_monitoring` group | ❌ |
| Survives pod restart | ✅ | only if something else collected it |

The entire log output story is one line, fixed at `logger.Init()` time:

```go
// pkg/logger/setting.go
if development { w = zerolog.ConsoleWriter{Out: os.Stderr, …} } else { w = os.Stderr }
ctx := zerolog.New(w).Level(lvl).With().Timestamp()
```

**Proposal:** give logs the same two-destination model metrics already have — console **and** self-storage — reusing the native-metrics machinery wherever it already solves the problem.

**Non-goals:** replacing console output; ingesting non-BanyanDB logs; changing the module tree, level filtering, or existing flags.

## 2. Deployment architecture

```mermaid
flowchart TB
C["clients (OAP, bydbctl)"] --> LB["gRPC load balancer :17912"]

subgraph LT["Liaison tier — stateless, NO disk"]
L0["liaison-0
:17912 client · :18912 peer
+ FODC agent"]
L1["liaison-1
+ FODC agent"]
end

subgraph DT["Data tier — OWNS THE DISK"]
DH["data-hot-0 :17912
+ FODC agent
+ lifecycle sidecar
+ backup sidecar
+ restore init container
PVC: measure/stream/trace/property"]
DW["data-warm-0"]
DC["data-cold-0"]
end

FP["fodc-proxy · 1 per cluster
:17913 /metrics · /cluster/topology"]

LB --> L0 & L1
L0 & L1 -- "write / query :17912" --> DH
DH -- "hot→warm→cold" --> DW --> DC
L0 & L1 -. "gRPC register" .-> FP
DH & DW & DC -. "gRPC register" .-> FP
FP --> P["Prometheus → Grafana"]
```

**Pairing rules**

| Component | Cardinality | Paired with | Form |
|---|---|---|---|
| `liaison` | N (≥2) | — | behind a gRPC LB, peers via `:18912` |
| `data` | M (≥2), hot/warm/cold | — | discovered by liaisons |
| FODC **agent** | **1 : 1 per node** | one liaison *or* one data node | sidecar |
| FODC **proxy** | **1 per cluster** | all agents | standalone pod |
| `lifecycle` | 1 per data node | co-located data node | sidecar, `127.0.0.1:17912` |
| `backup` | 1 per data node | co-located data node | sidecar, shares PVC |
| `restore` | 1 per data node | **nothing** | **init container** |
| `migration` | 1 per cluster | **all PVCs, zero live nodes** | standalone pod, data tier at `replicas=0` |

> `liaison` and `data` are cluster-scoped peers. Everything else is a **per-node companion** glued to one node by `127.0.0.1` or a shared volume.

## 3. Role → function → log destination

Self-storage needs a process owning the **storage engine** — the component holding queryable groups on disk. Only `data` and `standalone` do. Everything else must reach one, and the hop count differs.

| Role | Function | Storage engine | Log destination |
|---|---|---|---|
| `standalone` | all-in-one | ✅ | **in-process** — `queue.Local()`, own shard |
| `data` | stores & serves shards | ✅ | **in-process** — `queue.Local()`, own shard |
| `liaison` | routes writes/queries | ❌ | **two tiers** — liaison wqueue → part-sync → data node ([ref](https://github.com/apache/skywalking/issues/14076#issuecomment-5633309173)) |
| `lifecycle` | hot→warm→cold migration | ❌ | **one hop** — `pub` → co-located `127.0.0.1:17912` |
| `backup` | snapshots → S3/GCS/Azure | ❌ | **one hop** — `pub` → `--grpc-addr`; needs a schema-bootstrap decision |
| `restore` | remote backup → local dirs | ❌ | **console only** — init container, runs before its data node starts |
| `migration` | re-grid measure/stream data | ❌ | **console only** — runs with the data tier at `replicas=0` |

Two corrections to a common mental model:

- **A liaison is not diskless.** It has a write queue at `--measure-data-path` / `--stream-data-path` ([`banyand/measure/wqueue.go`](https://github.com/apache/skywalking-banyandb/blob/main/banyand/measure/wqueue.go)) where it buffers parts before syncing them out. It lacks the *storage engine*, so it can hold a log batch in flight but can never be where logs are read back from.
- **`lifecycle` and `backup` never go through a liaison.** `pub.NewWithoutMetadata(nil)` defaults to `ROLE_DATA`, so they publish straight to their co-located data node — a strictly shorter path than the liaison's.

> `restore` and `migration` are **not gaps to fill later**. A tool that runs while the database is down cannot log into that database — and for `migration`, a self-storing sink would violate its own precondition that nothing else writes to the target paths.

## 4. Proposed approach

### 4.1 The seam — one event, two writers

`zerolog` always encodes an event to JSON internally, and `ConsoleWriter` is itself an `io.Writer` that re-formats that JSON. So `MultiLevelWriter` hands the sink the **fully-encoded line** — module, level, timestamp, message, all structured fields — one source, one format, one set of parameters.

```mermaid
flowchart LR
A["call site
logger.GetLogger(measure).Info()
.Str(group, g).Msg(flushed part)"] --> B["zerolog encodes ONE JSON event"]
B --> CW["console writer
stderr · ALWAYS ON"]
B --> SW["switchableWriter
atomic.Pointer"]
SW --> S2["logSink — ONE ring buffer
alive from Init()"]
SW --> S3["io.Discard — after GracefulStop"]
```

> ❌ `zerolog.Hook` rejected — hooks see level + message but **not** the accumulated structured fields.

#### The writer contract

Three constraints come from zerolog's own implementation ([v1.34.0](https://github.com/rs/zerolog/blob/master/writer.go)), not from BanyanDB:

| zerolog internal | Constraint on the sink |
|---|---|
| `Event.write()` calls `putEvent(e)`, returning `e.buf` to a `sync.Pool` | **`p` is reused.** Buffering it without `append([]byte(nil), p...)` yields a slice the next log line overwrites. |
| `multiLevelWriter.Write` maps `_n != len(p)` → `io.ErrShortWrite` | **Always return `(len(p), nil)`.** A dropped line must still report a full write; drops surface via the counter, never the return value. |
| `MultiLevelWriter` type-switches on `zerolog.LevelWriter` before wrapping in `LevelWriterAdapter` | **Implement `WriteLevel(l zerolog.Level, p []byte)`** and zerolog hands over the level **as an enum** — so `--logging-native-level` filtering is an int compare with no JSON parsing on the hot path, and the `level` entity tag needs no extraction. |

```go
type switchableWriter struct{ target atomic.Pointer[zerolog.LevelWriter] }

func (s *switchableWriter) WriteLevel(l zerolog.Level, p []byte) (int, error) {
if w := s.target.Load(); w != nil {
(*w).WriteLevel(l, p) // errors deliberately ignored
}
return len(p), nil // always
}
```

`atomic.Pointer` rather than a mutex: `WriteLevel` runs on every log line from every goroutine, while the pointer is swapped **once** per process (at `GracefulStop`). A lock-free load is the right trade at that ratio.

### 4.2 Lifecycle — deferred activation

Same deferral `pendingMeasures` / `native.InitSchema` already use for metric schemas — but applied to **the consumer, not the buffer**.

```
Init() install switchableWriter → logSink (buffer live, NO consumer, no I/O)
PreRun() register drop counters (no metadata)
Serve() 1. create group + stream schema (idempotent)
2. START the consumer goroutine (§4.3)
GracefulStop closer.CloseNotify() → consumer drains once and exits (bounded)
swap → io.Discard · close publisher
```

> **One buffer, two phases.** The buffer is allocated at `Init()` and never replaced; "activation" only starts draining it. There is no second buffer and no hand-off, so a goroutine that loaded the writer just before activation cannot strand its line in an abandoned buffer. Producers see one unchanging target for the whole process lifetime.

### 4.3 Write workflow

The consumer is a **dedicated goroutine**, following [`accesslog.startConsumer`](https://github.com/apache/skywalking-banyandb/blob/main/pkg/accesslog/file.go) — *not* a `timestamp.Scheduler` job as `FlushMetrics` uses. A scheduler job is a periodic callback: it fires on the tick and can do nothing between ticks, so it supports a time trigger and nothing else. The size trigger needs something that observes **every push**, which only a goroutine selecting on the buffer can do.

```go
for {
select {
case <-s.closer.CloseNotify():
s.flush(batch) // final drain, bounded
return
case <-flushTicker.C: // TIME trigger (--logging-native-flush-interval, 5s)
if len(batch) > 0 { s.flush(batch); batch = batch[:0] }
case entry := <-s.buffered():
batch = append(batch, entry)
if len(batch) >= s.flushSize { // SIZE trigger (--logging-native-flush-size, 1024)
s.flush(batch); batch = batch[:0]
}
}
}
```

```mermaid
flowchart LR
R["buffer
bounded · drop on full"] --> T{"consumer goroutine
flush trigger"}
T -- "interval 5s" --> B["build InternalWriteRequest"]
T -- "size ≥ 1024" --> B
T -- "CloseNotify" --> B
B --> N{"nodeSelector"}
N -- "nil (data/standalone)" --> LOC["queue.Local() → own shard"]
N -- "set (liaison/lifecycle)" --> LO["Locate()"]
LO -- ok --> PUB["pub → TopicStreamWrite → data node"]
LO -- "fail" --> DROP["count no_node, drop
(do NOT publish empty nodeID)"]
```

### 4.4 Schema

Derived field by field from the native-metrics schema in [`pkg/meter/native/provider.go`](https://github.com/apache/skywalking-banyandb/blob/main/pkg/meter/native/provider.go). Each row names the proto field, so the diff against the existing implementation is explicit.

Running example — this line, emitted on node `data-hot-0`:

```json
{"level":"warn","module":"MEASURE","group":"sw_metric","time":"2026-09-11T10:23:45.123Z","message":"flush took longer than expected"}
```

**Group — `common.v1.Group`**

| Schema field | Metrics (`_monitoring`) | Logs (`_monitoring_log`) | Example value |
|---|---|---|---|
| `metadata.name` | `_monitoring` | `_monitoring_log` | `"_monitoring_log"` |
| `catalog` | `CATALOG_MEASURE` | `CATALOG_STREAM` | `Catalog_CATALOG_STREAM` (= 1) |
| `resource_opts.shard_num` | `1` | `1` (configurable) | `1` |
| `resource_opts.segment_interval` | `{UNIT_DAY, 1}` | `{UNIT_DAY, 1}` | `&IntervalRule{Unit: UNIT_DAY, Num: 1}` |
| `resource_opts.ttl` | `{UNIT_DAY, 1}` | `{UNIT_DAY, 7}` (configurable) | `&IntervalRule{Unit: UNIT_DAY, Num: 7}` |

**Resource — metrics use `database.v1.Measure`, logs use `database.v1.Stream`**

> **Not a rename.** `Measure` and `Stream` are two distinct proto messages that coexist; nothing is renamed or modified. Logs simply instantiate a different existing type. "Resource" is BanyanDB's own umbrella term for what a group holds — [`docs/concept/data-model.md`](https://github.com/apache/skywalking-banyandb/blob/main/docs/concept/data-model.md): *"A group's `catalog` fixes which one kind of resource it holds (`MEASURE`, `STREAM`, `TRACE`, or `PROPERTY`)."* There is no `Resource` type in code.

| Schema field | Metrics (`Measure`) | Logs (`Stream`) | Example value |
|---|---|---|---|
| `metadata.name` | one Measure **per metric name** | one Stream, `log` | metrics: `"total_written"` · logs: `"log"` |
| `tag_families[].name` | single `default` | `searchable` + `data` | `"searchable"`, `"data"` |
| `tag_families[].tags[]` | node/scope/metric labels | 7 searchable + 1 binary | `{Name:"level", Type:TAG_TYPE_STRING}`, `{Name:"body", Type:TAG_TYPE_DATA_BINARY}` |
| `fields[]` (`FieldSpec`) | `value` FLOAT/GORILLA/ZSTD | **field does not exist** | — (no such field on `Stream`) |
| `entity.tag_names` | all 4 node tags + every label | `[node_id, module, level]` | `[]string{"node_id","module","level"}` |

**Write payload**

| Schema field | Metrics | Logs | Example value (for the line above) |
|---|---|---|---|
| request / value type | `measure.v1.InternalWriteRequest` / `DataPointValue` | `stream.v1.InternalWriteRequest` / `ElementValue` | — |
| `…element_id` | **no counterpart** | `--` | `"data-hot-0-1757585021-42"` |
| `…timestamp` | `time.Now().Truncate(time.Second)` at flush | the event's own time | `2026-09-11T10:23:45.123Z` |
| `…tag_families[0]` (`searchable`) | `{Tags: labelValues}` | 7 tag values | `["data","data-hot-0","MEASURE","warn","10.1.2.3:17912","10.1.2.3:17913","flush took longer than expected"]` |
| `…tag_families[1]` (`data`) | — | the raw line | `[]byte("{\"level\":\"warn\",\"module\":\"MEASURE\",…}")` |
| `…fields` | one `FieldValue_Float` | **absent** | — |
| `entity_values` | label values | must match `entity.tag_names` order | `["data-hot-0","MEASURE","warn"]` |
| topic | `data.TopicMeasureWrite` | `data.TopicStreamWrite` | `data.TopicStreamWrite` |

```
tag_families:
searchable: node_type, node_id, module, level, grpc_address, http_address, message
data: body TAG_TYPE_DATA_BINARY ← the complete original JSON line
```

- `grpc_address` / `http_address` — **searchable, not in the entity**. Not cardinality (both follow from `node_id`) but **address churn**: a restarted pod keeps its `node_id` and gets a new IP, which would open a new series and fragment that node's history.
- `body` holds the whole original line — no field lost to schema drift, byte-identical to the console.
- `element_id` = `--`. `timestamp` = the **event's** time, not flush time, so lines buffered before activation land correctly on the time axis.

### 4.5 Two things that differ from metrics

| | Metrics | Logs |
|---|---|---|
| **Semantics** | sampled state — a missed flush costs nothing, next flush carries the current value | **events** — a dropped line is gone. Needs bounded ring + explicit drop-oldest + drop counters |
| **Recursion** | `gauge.Set()` doesn't log | **the write path logs.** `pub`, `sub`, `cluster-node-registry-*` all call `logger.GetLogger(...)` → a naive sink feeds itself |

Recursion guard, three layers: **module denylist** for write-path modules (primary) → sink never logs through `pkg/logger`, only rate-limited stderr → bounded ring caps amplification (backstop).

> ⚠️ The denylist is load-bearing for the liaison specifically: its [two-tier path](https://github.com/apache/skywalking/issues/14076#issuecomment-5633309173) traverses the write queue and part-sync, **both of which log**.

**Invariant across every failure mode: console output is never degraded by the sink.** Self-storage is best-effort; the console is not.

### 4.6 Implementation phases

See [this comment](https://github.com/apache/skywalking/issues/14076#issuecomment-5633893088) — delivery order, and why each phase adds exactly one new failure domain.

## 5. Parameters and configuration

Derived from the two existing surfaces — **prefix from console logging, vocabulary from metrics**:

```
pkg/logger/setting.go RegisterFlags observability/services/service.go FlagSet
--logging-env prod --observability-listener-addr :2121
--logging-level info --observability-modes [prometheus]
--logging-modules nil --observability-metrics-interval 15s
--logging-levels nil --observability-native-flush-interval 5s
└──── "logging-" prefix ──┐ ┌──── "modes" + "-native-" infix ────┘
▼ ▼
--logging-native-*
```

| Flag | Default | Meaning | From (where the idea comes from) |
|---|---|---|---|
| `--logging-modes` | `console` | `console`, `native`, or both | **metric** — the `modes` idea, from `--observability-modes` (`[prometheus]`) |
| `--logging-native-level` | `warn` | minimum level reaching storage | **console log** — `--logging-level`, but as an **independent** threshold. Metrics have no level concept |
| `--logging-native-flush-interval` | `5s` | time trigger | **metric** — `--observability-native-flush-interval`, same default |
| `--logging-native-flush-size` | `1024` | size trigger, in entries | **access log** — `accesslog.DefaultBatchSize` (100), a constant today |
| `--logging-native-buffer-size` | `8192` | ring capacity, in entries | **access log** — the `validRequests` channel capacity (100 sampled / 1000 not), a constant today |
| `--logging-native-group-ttl` | `7d` | `_monitoring_log` TTL | **metric** — `provider.go` hardcodes `ResourceOpts.Ttl = {UNIT_DAY, 1}` |
| `--logging-native-shard-num` | `1` | `_monitoring_log` shards | **metric** — `provider.go` hardcodes `ResourceOpts.ShardNum = 1`. Load-bearing for logs, given the single-node funnel |

**Deliberately not copied from metrics:** `--observability-listener-addr` (logs are push-only — no pull endpoint to scrape) and `--observability-metrics-interval` (that is the Prometheus *collection* tick; logs have no collection phase).

**Unchanged:** the four existing `--logging-env / -level / -modules / -levels` flags. They apply upstream of both writers; altering them is an explicit non-goal. Each new flag gets the standard `BYDB_*` env binding, as `logger.RegisterFlags` already provides. `restore` / `migration` accept `--logging-modes` but reject `native` at `Validate()`.

## 6. Reading logs back

No new query surface — `_monitoring_log` is an ordinary stream group:

```sh
bydbctl stream query -f - <