diff --git a/README.md b/README.md index 696185e..7c60429 100644 --- a/README.md +++ b/README.md @@ -16,7 +16,8 @@ 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`). @@ -24,23 +25,75 @@ The suite: - **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 ``` diff --git a/chaos_test.go b/chaos_test.go index 15da002..7dc9d83 100644 --- a/chaos_test.go +++ b/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 "-" == "data-" (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)) }) diff --git a/durability_test.go b/durability_test.go new file mode 100644 index 0000000..2a129dd --- /dev/null +++ b/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)) +}) diff --git a/helpers_test.go b/helpers_test.go index 11d4850..7ebe843 100644 --- a/helpers_test.go +++ b/helpers_test.go @@ -3,6 +3,7 @@ package e2e import ( "context" "fmt" + "sort" "sync/atomic" "time" @@ -10,6 +11,7 @@ import ( . "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) { diff --git a/internal/nodeadmin/nodeadmin.go b/internal/nodeadmin/nodeadmin.go new file mode 100644 index 0000000..1fcd7e4 --- /dev/null +++ b/internal/nodeadmin/nodeadmin.go @@ -0,0 +1,131 @@ +// Package nodeadmin is a thin client for an oteldb node's admin API (`/api/v1/storage`), which +// reports the embedded storage engine's per-shard state and the cluster's part-mirroring activity. +// +// The admin listener (:8090) is deliberately not published on the operator's client Service, so the +// suite port-forwards to an individual pod. That is exactly what these specs want: the questions it +// answers — did *this* node flush parts, did *this* node mirror a peer's — are per-node, and a +// Service-level view would hide them behind the read fan-out. +package nodeadmin + +import ( + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "strings" + "time" +) + +// Port is the oteldb admin API bind inside the container. +const Port = 8090 + +// Storage is the subset of GET /api/v1/storage the suite asserts on. +type Storage struct { + StorageEnabled bool `json:"storage_enabled"` + Engine struct { + Tenants []Tenant `json:"tenants"` + Cluster Cluster `json:"cluster"` + } `json:"engine"` +} + +// Tenant is one shard's state on this node ("default/_s0" is tenant "default", shard 0). +type Tenant struct { + Tenant string `json:"tenant"` + Signals []Signal `json:"signals"` +} + +// Signal is one signal's engine state within a shard. +type Signal struct { + Signal string `json:"signal"` + Series int64 `json:"series"` + HeadItems int64 `json:"head_items"` + // Parts counts the flushed, durable parts this node holds locally. It is the observable that + // distinguishes "the node has the data" from "a peer answers for it". + Parts int64 `json:"parts"` +} + +// Cluster is this node's view of the storage cluster. +type Cluster struct { + Self string `json:"self"` + // Members is the ring membership as this node sees it. + Members []Member `json:"members"` + Owned []string `json:"owned"` + // PartSync is reported only when the node runs with cluster.private_backend set: the stats + // live on the shared-nothing part mirror, so a nil here means the node believes its backend is + // a store its peers can read, and cluster/partsync never runs. + PartSync *PartSync `json:"part_sync"` +} + +// Member is one ring member. +type Member struct { + ID string `json:"id"` + Addr string `json:"addr"` + Zone string `json:"zone"` +} + +// PartSync is the cumulative shared-nothing part-mirroring activity of this node. +type PartSync struct { + Passes int64 `json:"passes"` + Mirrored int64 `json:"mirrored"` + Copied int64 `json:"copied"` + CopiedBytes int64 `json:"copied_bytes"` + Pruned int64 `json:"pruned"` + Errors int64 `json:"errors"` +} + +// Client reads one node's admin API over a port-forwarded authority ("127.0.0.1:"). +type Client struct { + HTTP *http.Client + Base string +} + +// NewClient returns a Client for a port-forwarded admin authority. +func NewClient(base string) *Client { + return &Client{HTTP: &http.Client{Timeout: 15 * time.Second}, Base: base} +} + +// Storage fetches GET /api/v1/storage. +func (c *Client) Storage(ctx context.Context) (Storage, error) { + var out Storage + u := fmt.Sprintf("http://%s/api/v1/storage", c.Base) + req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil) + if err != nil { + return out, err + } + resp, err := c.HTTP.Do(req) + if err != nil { + return out, err + } + defer resp.Body.Close() + body, _ := io.ReadAll(resp.Body) + if resp.StatusCode != http.StatusOK { + return out, fmt.Errorf("GET %s -> %d: %s", u, resp.StatusCode, strings.TrimSpace(string(body))) + } + if err := json.Unmarshal(body, &out); err != nil { + return out, fmt.Errorf("decode %s: %w (body: %s)", u, err, string(body)) + } + return out, nil +} + +// PartsFor totals the flushed parts this node holds for a signal across every shard. +func (s Storage) PartsFor(signal string) int64 { + var n int64 + for _, t := range s.Engine.Tenants { + for _, sig := range t.Signals { + if sig.Signal == signal { + n += sig.Parts + } + } + } + return n +} + +// Shards returns the shard identifiers this node holds an engine for. +func (s Storage) Shards() []string { + out := make([]string, 0, len(s.Engine.Tenants)) + for _, t := range s.Engine.Tenants { + out = append(out, t.Tenant) + } + return out +} diff --git a/manifests/cluster.yaml b/manifests/cluster.yaml index 76697a1..d0918b5 100644 --- a/manifests/cluster.yaml +++ b/manifests/cluster.yaml @@ -1,6 +1,7 @@ # The OtelDBCluster exercised by the e2e suite: three symmetric nodes on the per-node `file` -# backend with replication factor 2, so a single node loss (chaos specs) is survivable and a -# wiped node re-replicates its shards from a peer. Image is pinned to the locally built, +# backend with replication factor 2 and cluster.privateBackend set, so a single node loss (chaos +# specs) is survivable and a wiped node re-replicates its shards from a peer. Image is pinned to +# the locally built, # kind-loaded oteldb (see internal/build); imagePullPolicy IfNotPresent keeps it from reaching # for the registry. Sizes/resources are trimmed to run comfortably inside kind. apiVersion: db.oteldb.io/v1alpha1 @@ -26,6 +27,11 @@ spec: replicationFactor: 2 shardsPerTenant: 4 peerPort: 7946 + # Shared-nothing: every node's `file` backend is its own PVC, not a store its peers can read, + # so flushed parts must be mirrored node-to-node by cluster/partsync. Without this the ring + # replicates only the unflushed head and each flushed part exists in exactly ONE copy at + # replicationFactor 2 (oteldb/oteldb#1262). The `private_backend` spec asserts it took effect. + privateBackend: true signals: metrics: true