test(durability): assert shared-nothing part replication and disk-loss recovery

README.md

@@ -16,31 +16,84 @@ The suite:
 2. **Deploys the operator** (CRDs + controller-manager) via the operator's own Makefile. There are
    no webhooks, so cert-manager is not required.
 3. **Brings up a clustered `OtelDBCluster`** — 3 symmetric nodes on the per-node `file` backend
-   with `replicationFactor: 2`, backed by a bring-your-own dev etcd (`manifests/etcd-dev.yaml`).
+   with `replicationFactor: 2` and `cluster.privateBackend` set, backed by a bring-your-own dev
+   etcd (`manifests/etcd-dev.yaml`).
 4. **Exercises the normal path**: ingests traces, logs and metrics over OTLP/gRPC and reads each
    back through its native query API (Tempo `/api/traces`, Loki `/loki/api/v1/query_range`,
    Prometheus `/api/v1/query`).
 5. **Exercises chaos-injection paths**:
    - **Pod kill** — force-delete a node (its PVC is retained); assert previously-ingested data
      stays queryable (it is restored from the node's own WAL/parts on the PVC), the cluster
      returns to Ready, and fresh telemetry ingests/queries.
-   - **Disk loss** — delete a node *and its PVC*; assert the cluster self-heals (the wiped node
-     rejoins with a fresh PVC and the cluster returns to Ready) and stays available (new telemetry
-     ingests/queries across all signals).
-
-   > **Durability note.** With the per-node `file` backend, a node's data lives on its own PVC
-   > (WAL + parts). RF replication and read fan-out provide write-durability and availability while
-   > a peer is transiently down, but reads are primary-authoritative: a node returning under the
-   > same identity with an *empty* disk is not auto-restored from replicas within the query window.
-   > That disaster-recovery case is what the shared S3 backend addresses — which is intentionally
-   > **out of scope here** (file backend only). So the disk-loss spec asserts self-healing +
-   > availability, not recovery of the wiped node's historical data.
+   - **Disk loss** — delete a node *and its PVC*; assert the wiped node mirrors its shards back
+     from a peer, holds the flushed parts again, and that the pre-disruption data stays queryable
+     and new telemetry still ingests across all signals.
+
+6. **Asserts the cluster really is in shared-nothing mode** (`durability_test.go`) — see below.
+
+## Durability on the `file` backend
+
+Each node's `file` backend is its own PVC, which its peers cannot read. Two different mechanisms
+cover the two halves of a node's data, and only one of them is on by default:
+
+| what | replicated by | when |
+|---|---|---|
+| unflushed head | the ring's primary-authoritative write path (`replicationFactor`) | always |
+| flushed parts  | `cluster/partsync`, mirroring backend objects over the parts HTTP endpoints | only with `storage.cluster.private_backend` |
+
+`private_backend` is *not* inferable from the backend type — an S3 bucket shared by every node and
+a per-node local disk are both legal `backend:` values — so it has to be declared, and the
+operator's own default topology (a StatefulSet with a per-pod `ReadWriteOnce` PVC) is exactly the
+shared-nothing case that needs it. `manifests/cluster.yaml` sets `spec.cluster.privateBackend: true`
+for that reason, and `durability_test.go` asserts the node is actually running that way rather than
+trusting the manifest, because the misconfigured shape is invisible from the outside: writes still
+reach every replica, and every query is still answered.
+
+**Measured, with `privateBackend` set.** A flushed part lands on `replicationFactor` nodes. Wiping a
+node's PVC and letting it return under the same identity restores it from its peers in roughly one
+maintenance tick (the tick cadence *is* `spec.engine.flushInterval`, 10s here): the node finds it
+owns a shard it has no engine for, `cluster/partsync` mirrors that shard's backend objects from a
+peer, and the engine loads them —
+
+```
+bootstrap: gained shard with no local engine   shard=default/_s0
+bootstrap: shard signal loaded                 shard=default/_s0 signal=log
+part_sync: mirrored=1 copied=9 copied_bytes=557
+```
+
+That is what the disk-loss chaos spec now asserts. It is a change from what this README used to
+claim ("a node returning with an empty disk is not auto-restored from replicas") — that note was
+written while the flag was silently off, and it is wrong once the flag is on.
+
+**Measured, without it** (oteldb/oteldb#1262 — the state this suite was previously running in): each
+flushed part exists in exactly **one** copy despite `replicationFactor: 2`, and a node that loses
+its disk never recovers. It logs
+
+```
+bootstrap: gained shard with no local engine   shard=default/_s3
+```
+
+once per maintenance tick, forever: the mirror pass is skipped as a no-op for a store the node
+believes its peers can already read, so no bucket index ever appears on its disk, so no engine is
+ever created, so it keeps rediscovering the same missing shard. The node stays permanently empty.
 
 ## Requirements
 
 `docker`, `kind`, `kubectl`, and a Go toolchain — all on `PATH`. The sibling `../oteldb` and
 `../operator` source trees must be present (override with env vars below).
 
+The operator must understand `spec.cluster.privateBackend` (oteldb/operator#16). It is not on
+`oteldb/operator@main` yet, so against an older operator the run stops at
+
+```
+error when creating "manifests/cluster.yaml": OtelDBCluster in version "v1alpha1" cannot be
+handled as a OtelDBCluster: strict decoding error: unknown field "spec.cluster.privateBackend"
+```
+
+That is the good failure. `durability_test.go` is the backstop for the bad one — an operator that
+accepts the field and drops it on the floor would otherwise leave the suite quietly testing a
+shared-store cluster that isn't one.
+
 ## Running
 
 ```bash
@@ -89,8 +142,10 @@ internal/
   build/            fast image path: host go build -> COPY image -> kind load
   kubeutil/         kubectl helpers: apply/wait/list/delete + Service port-forward
   otlp/             OTLP/gRPC ingest (traces/logs/metrics) + Tempo/Loki/Prometheus query clients
+  nodeadmin/        per-node admin API client (/api/v1/storage): parts, shards, part-sync stats
 suite_test.go       Ginkgo bootstrap: kind + build/load + deploy operator + bring up cluster
-helpers_test.go     shared port-forward + ingest/verify helpers
+helpers_test.go     shared port-forward + ingest/verify + flushed-part helpers
 normal_test.go      happy-path ingest/query specs
+durability_test.go  shared-nothing mode + flushed-part replication specs
 chaos_test.go       pod-kill and disk-loss recovery specs
 ```

chaos_test.go

@@ -1,6 +1,7 @@
 package e2e
 
 import (
+	"fmt"
 	"time"
 
 	. "github.com/onsi/ginkgo/v2"
@@ -56,25 +57,32 @@ var _ = Describe("Chaos", func() {
 	})
 
 	// Disk loss (a node's local PersistentVolume is wiped) vs. pod loss (above) exercises a
-	// different guarantee. With the per-node `file` backend, a node's data lives on its own PVC
-	// (WAL + parts); RF replication and read fan-out provide write-durability and availability
-	// while a peer is transiently down, but reads are primary-authoritative — a node that returns
-	// under the same identity with an empty disk is NOT auto-restored from replicas within the
-	// query window (that recovery is the shared S3 backend's job, and out of scope here). So the
-	// guarantee this spec asserts is self-healing + continued availability: the wiped node
-	// rejoins with a fresh PVC, the cluster returns to Ready, and it keeps ingesting and serving.
-	It("self-heals and stays available after a node loses its local disk (PVC)", func(ctx SpecContext) {
+	// different guarantee: the returning node has its identity but none of its data, so nothing on
+	// its own disk can restore it. In shared-nothing mode it is restored from its peers instead —
+	// the node discovers it owns a shard it has no engine for, mirrors that shard's backend objects
+	// from a peer over the parts endpoints (cluster/partsync), and loads them. The pass runs on the
+	// engine's maintenance tick, whose cadence is spec.engine.flushInterval (10s here), so recovery
+	// is a tick or two, not a query-window wait.
+	//
+	// This works only because spec.cluster.privateBackend is set (see durability_test.go). Without
+	// it the node loops "gained shard with no local engine" at the maintenance cadence forever: the
+	// sync is skipped as a no-op for a supposedly-shared store, so no bucket index ever appears
+	// locally, so no engine is ever created, so the node stays permanently empty.
+	It("restores a node's shards from its peers after it loses its local disk (PVC)", func(ctx SpecContext) {
 		runID := "chaos-disk-" + randID()
 
 		By("establishing a baseline before the disruption")
 		ingest, clients := endpoints()
 		ingestAndVerify(ctx, ingest, clients, runID)
 
-		By("deleting one oteldb pod together with its PersistentVolumeClaim (disk loss)")
-		pods, err := kubeutil.PodNames(conf.Namespace, selector())
-		Expect(err).NotTo(HaveOccurred())
-		Expect(pods).To(HaveLen(3))
-		victim := pods[len(pods)-1]
+		By("waiting for a flushed, already-mirrored part to exist")
+		// Only flushed parts can be mirrored — the head is the ring's business, not partsync's — so
+		// there must be something on the victim's disk worth recovering, and a peer copy to recover
+		// it from. Waiting for both makes the wipe a test of restoration rather than of luck.
+		signal, holders := waitReplicatedPart(ctx)
+
+		By(fmt.Sprintf("deleting a pod holding flushed %s parts, together with its PersistentVolumeClaim", signal))
+		victim := holders[0]
 		// The per-pod PVC is named "<dataVolume>-<pod>" == "data-<pod>" (StatefulSet convention).
 		// Mark the PVC for deletion first (non-blocking; it is pinned by the running pod's
 		// pvc-protection finalizer), then force-delete the pod so the finalizer clears and the
@@ -84,8 +92,37 @@ var _ = Describe("Chaos", func() {
 
 		waitClusterHealthy()
 
+		By("verifying the wiped node mirrors its shards back from a peer")
+		Eventually(func(ctx SpecContext) (int64, error) {
+			st, err := nodeStorage(ctx, victim)
+			if err != nil {
+				return 0, err
+			}
+			if st.Engine.Cluster.PartSync == nil {
+				return 0, fmt.Errorf("%s reports no part-sync stats (private_backend not in effect)", victim)
+			}
+			return st.Engine.Cluster.PartSync.Mirrored, nil
+		}).WithContext(ctx).WithTimeout(4*time.Minute).WithPolling(5*time.Second).
+			Should(BeNumerically(">", 0), "%s never mirrored a peer's parts after losing its disk", victim)
+
+		By(fmt.Sprintf("verifying the wiped node holds flushed %s parts again", signal))
+		Eventually(func(ctx SpecContext) (int64, error) {
+			st, err := nodeStorage(ctx, victim)
+			if err != nil {
+				return 0, err
+			}
+			return st.PartsFor(signal), nil
+		}).WithContext(ctx).WithTimeout(4*time.Minute).WithPolling(5*time.Second).
+			Should(BeNumerically(">", 0), "%s did not get its %s parts back from its peers", victim, signal)
+
+		By("verifying the pre-disruption telemetry is still queryable")
+		_, clientsAfter := endpoints()
+		Eventually(func() (bool, error) {
+			return clientsAfter.LogsContain(ctx, `{service_name="oteldb-e2e-logs"}`, runID)
+		}).Should(BeTrue(), "data ingested before the disk loss became unqueryable")
+
 		By("verifying the recovered cluster accepts and serves new telemetry across all signals")
 		ingest2, clients2 := endpoints()
 		ingestAndVerify(ctx, ingest2, clients2, runID+"-after")
-	})
+	}, SpecTimeout(15*time.Minute))
 })

durability_test.go

@@ -0,0 +1,60 @@
+package e2e
+
+import (
+	"fmt"
+	"time"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+
+	"github.com/oteldb/e2e/internal/kubeutil"
+)
+
+// The cluster runs shared-nothing: each node's `file` backend is its own PVC, which its peers
+// cannot read. Replication of the *unflushed head* is the ring's job and happens either way, but
+// replication of *flushed parts* is cluster/partsync's, and partsync only runs when the node is
+// told its backend is private (`storage.cluster.private_backend`, oteldb/oteldb#1264).
+//
+// Nothing about the misconfigured shape looks broken from the outside: writes still reach every
+// replica through the primary-authoritative path and every query is answered. It only shows up as
+// each flushed part existing in exactly one copy at replicationFactor 2, and as a node that lost
+// its disk never getting the data back (see chaos_test.go). So the suite asserts the mode directly
+// rather than inferring it, and asserts the RF copy count it is supposed to buy.
+var _ = Describe("Shared-nothing durability", func() {
+	It("runs every node with cluster.private_backend enabled", func(ctx SpecContext) {
+		pods, err := kubeutil.PodNames(conf.Namespace, selector())
+		Expect(err).NotTo(HaveOccurred())
+		Expect(pods).To(HaveLen(3))
+
+		for _, pod := range pods {
+			st, err := nodeStorage(ctx, pod)
+			Expect(err).NotTo(HaveOccurred(), "admin API on %s", pod)
+			Expect(st.StorageEnabled).To(BeTrue(), "%s is not serving from the storage engine", pod)
+			// part_sync is reported only by a node running in shared-nothing mode: the stats live on
+			// the part mirror, which is not constructed otherwise. A nil here means the rendered
+			// config lacks storage.cluster.private_backend — i.e. the operator dropped or never
+			// rendered spec.cluster.privateBackend, and flushed parts are not being replicated.
+			Expect(st.Engine.Cluster.PartSync).NotTo(BeNil(),
+				"%s reports no part-sync stats: it believes its backend is shared, so cluster/partsync "+
+					"is not running and flushed parts are not replicated node-to-node "+
+					"(is spec.cluster.privateBackend supported by the deployed operator?)", pod)
+		}
+	})
+
+	It("replicates a flushed part to replicationFactor nodes", func(ctx SpecContext) {
+		runID := "durability-" + randID()
+		ingest, _ := endpoints()
+		Expect(ingest.EmitLog(ctx, "oteldb-e2e-logs", runID)).To(Succeed())
+		Expect(ingest.EmitMetric(ctx, "oteldb-e2e-metrics", runID, 3)).To(Succeed())
+		_, err := ingest.EmitTrace(ctx, "oteldb-e2e-traces", runID)
+		Expect(err).NotTo(HaveOccurred())
+
+		By("waiting for a flushed part to be mirrored onto a second node")
+		signal, holders := waitReplicatedPart(ctx)
+		_, _ = fmt.Fprintf(GinkgoWriter, "%s parts are held by %v\n", signal, holders)
+
+		// Without the mirror the owner is the only holder, and that is invisible from every API the
+		// cluster exposes — the read path answers from whichever owner has the data.
+		Expect(len(holders)).To(BeNumerically(">=", 2))
+	}, SpecTimeout(8*time.Minute))
+})

helpers_test.go

@@ -3,13 +3,15 @@ package e2e
 import (
 	"context"
 	"fmt"
+	"sort"
 	"sync/atomic"
 	"time"
 
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 
 	"github.com/oteldb/e2e/internal/kubeutil"
+	"github.com/oteldb/e2e/internal/nodeadmin"
 	"github.com/oteldb/e2e/internal/otlp"
 )
 
@@ -29,6 +31,27 @@ const (
 	portTempo    = 3200
 )
 
+// nodeAdmin establishes a port-forward to one pod's admin API and returns a client. The admin
+// listener is not on the client Service (the operator does not publish it), so this forwards to the
+// pod directly — which is the point: the durability specs ask per-node questions.
+func nodeAdmin(pod string) *nodeadmin.Client {
+	ports, stop, err := kubeutil.PortForward(conf.Namespace, "pod/"+pod, nodeadmin.Port)
+	Expect(err).NotTo(HaveOccurred(), "Failed to port-forward to %s admin API", pod)
+	DeferCleanup(stop)
+	return nodeadmin.NewClient(ports[nodeadmin.Port])
+}
+
+// nodeStorage reads one pod's /api/v1/storage. It re-establishes the forward on every call so it
+// survives the pod restarts the chaos specs cause.
+func nodeStorage(ctx context.Context, pod string) (nodeadmin.Storage, error) {
+	ports, stop, err := kubeutil.PortForward(conf.Namespace, "pod/"+pod, nodeadmin.Port)
+	if err != nil {
+		return nodeadmin.Storage{}, err
+	}
+	defer stop()
+	return nodeadmin.NewClient(ports[nodeadmin.Port]).Storage(ctx)
+}
+
 // endpoints establishes a fresh port-forward to the OtelDBCluster's client Service and returns an
 // OTLP ingester plus read-side query clients. A cleanup that stops the forward is registered on
 // the current spec via DeferCleanup, so callers get a clean forward each time — important after
@@ -45,6 +68,67 @@ func endpoints() (*otlp.Ingester, *otlp.Clients) {
 	return ingest, clients
 }
 
+// signals are the three the suite ingests, named as the admin API reports them.
+var signals = []string{"logs", "metrics", "traces"}
+
+// partHolders reports, per signal, which oteldb pods hold at least one *flushed* part for it. Data
+// sitting only in a node's in-memory head does not count — that distinction is the whole subject of
+// the durability specs, since the ring replicates the head and cluster/partsync replicates parts.
+func partHolders(ctx context.Context) (map[string][]string, error) {
+	pods, err := kubeutil.PodNames(conf.Namespace, selector())
+	if err != nil {
+		return nil, err
+	}
+	out := make(map[string][]string, len(signals))
+	for _, pod := range pods {
+		st, err := nodeStorage(ctx, pod)
+		if err != nil {
+			return nil, fmt.Errorf("admin API on %s: %w", pod, err)
+		}
+		for _, sig := range signals {
+			if st.PartsFor(sig) > 0 {
+				out[sig] = append(out[sig], pod)
+			}
+		}
+	}
+	for sig := range out {
+		sort.Strings(out[sig])
+	}
+	return out, nil
+}
+
+// waitReplicatedPart blocks until some signal's flushed parts are held by at least two nodes, and
+// returns that signal with its holders.
+//
+// It deliberately asserts a *state* rather than a change. Part counts are not a monotone counter to
+// wait on: a merge folds several parts back into one, so a cluster that has been running for a
+// while sits at one part per signal per node and a freshly flushed part can leave the total exactly
+// where it was. Waiting for a count to grow passes on a cold cluster and hangs on a warm one, which
+// is precisely the kind of run-order dependence this suite must not have.
+//
+// It also does not care *which* signal gets there first. Only a shard's compaction owner flushes,
+// so which signal has durable parts at a given moment is a property of shard placement, not of what
+// was ingested — pinning the spec to one signal makes it a placement lottery.
+func waitReplicatedPart(ctx context.Context) (signal string, holders []string) {
+	Eventually(func() (bool, error) {
+		byPod, err := partHolders(ctx)
+		if err != nil {
+			return false, err
+		}
+		for _, sig := range signals {
+			if len(byPod[sig]) >= 2 {
+				signal, holders = sig, byPod[sig]
+				return true, nil
+			}
+		}
+		_, _ = fmt.Fprintf(GinkgoWriter, "flushed-part holders so far: %v\n", byPod)
+		return false, nil
+	}, 4*time.Minute, 5*time.Second).Should(BeTrue(),
+		"no signal's flushed parts reached two nodes: at replicationFactor 2 a peer must mirror "+
+			"each flushed part (is cluster.private_backend in effect?)")
+	return signal, holders
+}
+
 // ingestAndVerify pushes a trace, a log and a metric tagged with runID and asserts each is
 // queryable back through the Tempo, Loki and Prometheus APIs. Returns once all three are visible.
 func ingestAndVerify(ctx context.Context, ingest *otlp.Ingester, clients *otlp.Clients, runID string) {