From 8b4d3d5712ca131a3967924282ec3c3d592e9118 Mon Sep 17 00:00:00 2001 From: Vladimir Iliakov Date: Thu, 10 Sep 2026 15:49:52 +0200 Subject: [PATCH 1/4] Add external HA database replication checker Inspect HDFS, Elasticsearch, Kafka and ClickHouse through namespace-scoped Kubernetes discovery and fixed read-only exec queries. Require complete evidence, reject observations whose Kubernetes membership changes, and support bounded waits with a stable healthy period and JSON output. Keep replication checks independent of backup configuration. Include dedicated Job RBAC and a SUSE BCI packaging example. Document that this is sampled replication evidence, not a node-removal safety certificate; live cluster validation remains outstanding. Tracking: https://github.com/StackVista/stackstate/issues/501 --- ARCHITECTURE.md | 2 + README.md | 14 ++ cmd/replication/replication.go | 154 +++++++++++++++ cmd/replication/replication_test.go | 79 ++++++++ cmd/root.go | 6 +- docs/replication.md | 157 ++++++++++++++++ examples/replication/.dockerignore | 2 + examples/replication/Dockerfile | 4 + examples/replication/job.yaml | 77 ++++++++ internal/app/replication.go | 17 ++ internal/clients/k8s/exec.go | 53 ++++++ internal/clients/k8s/exec_test.go | 20 ++ internal/orchestration/replication/checker.go | 86 +++++++++ .../orchestration/replication/checker_test.go | 158 ++++++++++++++++ .../orchestration/replication/clickhouse.go | 139 ++++++++++++++ .../orchestration/replication/discovery.go | 125 +++++++++++++ .../replication/elasticsearch.go | 78 ++++++++ .../replication/evaluators_test.go | 177 ++++++++++++++++++ internal/orchestration/replication/hdfs.go | 85 +++++++++ internal/orchestration/replication/json.go | 54 ++++++ internal/orchestration/replication/kafka.go | 137 ++++++++++++++ internal/orchestration/replication/report.go | 90 +++++++++ internal/orchestration/replication/wait.go | 46 +++++ .../orchestration/replication/wait_test.go | 59 ++++++ 24 files changed, 1817 insertions(+), 2 deletions(-) create mode 100644 cmd/replication/replication.go create mode 100644 cmd/replication/replication_test.go create mode 100644 docs/replication.md create mode 100644 examples/replication/.dockerignore create mode 100644 examples/replication/Dockerfile create mode 100644 examples/replication/job.yaml create mode 100644 internal/app/replication.go create mode 100644 internal/clients/k8s/exec.go create mode 100644 internal/clients/k8s/exec_test.go create mode 100644 internal/orchestration/replication/checker.go create mode 100644 internal/orchestration/replication/checker_test.go create mode 100644 internal/orchestration/replication/clickhouse.go create mode 100644 internal/orchestration/replication/discovery.go create mode 100644 internal/orchestration/replication/elasticsearch.go create mode 100644 internal/orchestration/replication/evaluators_test.go create mode 100644 internal/orchestration/replication/hdfs.go create mode 100644 internal/orchestration/replication/json.go create mode 100644 internal/orchestration/replication/kafka.go create mode 100644 internal/orchestration/replication/report.go create mode 100644 internal/orchestration/replication/wait.go create mode 100644 internal/orchestration/replication/wait_test.go diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 9883559..34cb71b 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -72,6 +72,7 @@ stackstate-backup-cli/ - `cmd/victoriametrics/`: VictoriaMetrics backup/restore commands (list, restore, check-and-finalize) - `cmd/settings/`: Settings backup/restore commands (list, restore, check-and-finalize) - `cmd/version/`: Version information +- `cmd/replication/`: Read-only HA replication observations and bounded waiting **Dependency Rules**: - ✅ Can import: `internal/app/*` (preferred), all other `internal/` packages @@ -126,6 +127,7 @@ appCtx.NewCHClient(backupAPIPort, dbPort) // ClickHouse client factory **Key Packages**: - `portforward/`: Manages Kubernetes port-forwarding lifecycle +- `replication/`: Discovers release workloads and evaluates fixed database queries, without backup configuration or restore operations - `scale/`: Deployment and StatefulSet scaling workflows with detailed logging - `restore/`: Restore job orchestration (confirmation, job lifecycle, finalization, resource management) - `restorelock/`: Prevents parallel restore operations using Kubernetes annotations diff --git a/README.md b/README.md index bac5f87..09a71b8 100644 --- a/README.md +++ b/README.md @@ -12,6 +12,7 @@ This CLI tool replaces the legacy Bash-based backup/restore scripts with a singl - Stackgraph backups and restores - VictoriaMetrics backups and restores - Settings backups and restores +- Read-only HA database replication checks ## Installation @@ -44,6 +45,19 @@ sts-backup [command] [subcommand] [flags] ## Commands +### replication check + +Inspect HDFS, Elasticsearch, Kafka and ClickHouse replication for a Helm release: + +```bash +sts-backup replication check --namespace observability --release suse-observability +sts-backup replication check --namespace observability --release suse-observability --wait --output json +``` + +This command has its own flags and does not require backup configuration. +See [Replication checks](docs/replication.md) for status and exit-code semantics, +authentication, a Kubernetes Job example, and the maintenance checks outside its scope. + ### version Display version information. diff --git a/cmd/replication/replication.go b/cmd/replication/replication.go new file mode 100644 index 0000000..6e81311 --- /dev/null +++ b/cmd/replication/replication.go @@ -0,0 +1,154 @@ +// Package replication exposes read-only database replication checks. +package replication + +import ( + "context" + "encoding/json" + "fmt" + "io" + "os" + "os/signal" + "strings" + "syscall" + "text/tabwriter" + "time" + + "github.com/spf13/cobra" + + "github.com/stackvista/stackstate-backup-cli/internal/app" + checker "github.com/stackvista/stackstate-backup-cli/internal/orchestration/replication" +) + +const ( + defaultTimeout = 10 * time.Minute + defaultRequestTimeout = 30 * time.Second + defaultInterval = 10 * time.Second + defaultStableFor = 30 * time.Second + tablePadding = 2 +) + +type flags struct { + options checker.Options + kubeconfig string + output string + wait bool + timeout time.Duration + interval time.Duration + stableFor time.Duration +} + +// Cmd creates the replication command independently of backup configuration. +func Cmd() *cobra.Command { + command := &cobra.Command{Use: "replication", Short: "Inspect HA database replication without changing cluster state"} + f := &flags{} + check := &cobra.Command{ + Use: "check", Short: "Check observed replication; return nonzero unless all selected checks pass", + Long: "Check chart-managed HDFS, Elasticsearch, Kafka and ClickHouse replication. " + + "This is a point-in-time observation, not permission to remove a node. " + + "Requires pods/exec access; only fixed read-only database queries are executed.", + Args: cobra.NoArgs, SilenceUsage: true, + RunE: func(command *cobra.Command, _ []string) error { return run(command, f) }, + } + check.Flags().StringVarP(&f.options.Namespace, "namespace", "n", "", "Kubernetes namespace (required)") + check.Flags().StringVar(&f.options.Release, "release", "", "Helm release name (required)") + check.Flags().StringVar(&f.kubeconfig, "kubeconfig", "", "Kubeconfig path; uses normal kubeconfig or in-cluster credentials") + check.Flags().StringSliceVar(&f.options.Components, "components", []string{"hdfs", "elasticsearch", "kafka", "clickhouse"}, "Components to check") + check.Flags().StringVarP(&f.output, "output", "o", "table", "Output format: table or json") + check.Flags().BoolVar(&f.wait, "wait", false, "Wait for sustained healthy replication") + check.Flags().DurationVar(&f.timeout, "timeout", defaultTimeout, "Overall deadline, including queries") + check.Flags().DurationVar(&f.options.RequestTimeout, "request-timeout", defaultRequestTimeout, "Deadline for each Kubernetes request or database query") + check.Flags().DurationVar(&f.interval, "interval", defaultInterval, "Interval between observations in wait mode") + check.Flags().DurationVar(&f.stableFor, "stable-for", defaultStableFor, "Required healthy observation period in wait mode") + check.Flags().StringVar(&f.options.KafkaClientProperties, "kafka-client-properties", "", "Kafka client properties file already mounted in broker pods") + check.Flags().StringVar(&f.options.KafkaBootstrapServer, "kafka-bootstrap-server", "localhost:9092", "Kafka bootstrap address reachable from the broker pod") + check.Flags().StringVar(&f.options.ElasticsearchScheme, "elasticsearch-scheme", "http", "Elasticsearch loopback protocol: http or https") + check.Flags().StringVar(&f.options.ElasticsearchCA, "elasticsearch-ca", "", "CA file already mounted in Elasticsearch pods") + check.Flags().StringVar(&f.options.ElasticsearchHost, "elasticsearch-server-name", "127.0.0.1", "Elasticsearch TLS server name, resolved to loopback inside the pod") + _ = check.MarkFlagRequired("namespace") + _ = check.MarkFlagRequired("release") + command.AddCommand(check) + return command +} + +func (f *flags) validate() error { + if f.output != "table" && f.output != "json" { + return fmt.Errorf("output must be table or json") + } + if f.timeout <= 0 || f.interval <= 0 || f.stableFor < 0 { + return fmt.Errorf("timeout and interval must be positive; stable-for cannot be negative") + } + if f.wait && f.stableFor >= f.timeout { + return fmt.Errorf("stable-for must be shorter than timeout") + } + return f.options.Validate() +} + +func run(command *cobra.Command, f *flags) error { + if err := f.validate(); err != nil { + return err + } + probe, err := app.NewReplicationChecker(f.kubeconfig, f.options) + if err != nil { + return err + } + ctx, stop := signal.NotifyContext(command.Context(), os.Interrupt, syscall.SIGTERM) + defer stop() + ctx, cancel := context.WithTimeout(ctx, f.timeout) + defer cancel() + report, checkErr := observe(ctx, probe.Check, f, command.ErrOrStderr()) + return finishReport(command.OutOrStdout(), f.output, report, checkErr) +} + +func finishReport(writer io.Writer, format string, report checker.Report, checkErr error) error { + if checkErr != nil { + report.Status = checker.Unknown + report.Error = checkErr.Error() + } + if err := writeReport(writer, format, report); err != nil { + return err + } + if checkErr != nil { + return checkErr + } + if report.Status != checker.Healthy { + return fmt.Errorf("replication is %s; see the report", report.Status) + } + return nil +} + +func observe(ctx context.Context, check func(context.Context) checker.Report, f *flags, progress io.Writer) (checker.Report, error) { + if !f.wait { + report := check(ctx) + if ctx.Err() != nil { + return report, fmt.Errorf("replication check ended: %w", ctx.Err()) + } + return report, nil + } + return checker.Wait(ctx, check, f.interval, f.stableFor, func(report checker.Report) { + for _, check := range report.Checks { + _, _ = fmt.Fprintf(progress, "%s %s: %s\n", check.Component, check.Status, strings.Join(check.Messages, "; ")) + } + }) +} + +func writeReport(writer io.Writer, format string, report checker.Report) error { + if format == "json" { + if err := json.NewEncoder(writer).Encode(report); err != nil { + return fmt.Errorf("write JSON report: %w", err) + } + return nil + } + table := tabwriter.NewWriter(writer, 0, 0, tablePadding, ' ', 0) + if _, err := fmt.Fprintln(table, "COMPONENT\tSTATUS\tDETAILS"); err != nil { + return fmt.Errorf("write report header: %w", err) + } + for _, check := range report.Checks { + if _, err := fmt.Fprintf(table, "%s\t%s\t%s\n", check.Component, check.Status, strings.Join(check.Messages, "; ")); err != nil { + return fmt.Errorf("write report row: %w", err) + } + } + if err := table.Flush(); err != nil { + return fmt.Errorf("flush report: %w", err) + } + return nil +} diff --git a/cmd/replication/replication_test.go b/cmd/replication/replication_test.go new file mode 100644 index 0000000..89787b0 --- /dev/null +++ b/cmd/replication/replication_test.go @@ -0,0 +1,79 @@ +package replication + +import ( + "bytes" + "context" + "encoding/json" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + checker "github.com/stackvista/stackstate-backup-cli/internal/orchestration/replication" +) + +func TestJSONOutputIsSeparateFromProgress(t *testing.T) { + var stdout, stderr bytes.Buffer + report, err := observe(context.Background(), func(context.Context) checker.Report { + return checker.Report{Status: checker.Healthy, Checks: []checker.Result{ + {Component: "kafka", Status: checker.Healthy, Messages: []string{"all assigned replicas in sync"}}, + }} + }, &flags{wait: true, interval: time.Millisecond}, &stderr) + require.NoError(t, err) + require.NoError(t, writeReport(&stdout, "json", report)) + var decoded checker.Report + require.NoError(t, json.Unmarshal(stdout.Bytes(), &decoded)) + assert.Equal(t, checker.Healthy, decoded.Status) + assert.Contains(t, stderr.String(), "kafka healthy") +} + +func TestCheckValidatesBeforeConnecting(t *testing.T) { + command := Cmd() + command.SetArgs([]string{"check", "--namespace=test", "--release=test", "--components=kafka", "--output=invalid"}) + command.SetOut(&bytes.Buffer{}) + command.SetErr(&bytes.Buffer{}) + err := command.Execute() + require.ErrorContains(t, err, "output must be table or json") +} + +func TestHelpDoesNotRequireBackupConfiguration(t *testing.T) { + command := Cmd() + command.SetArgs([]string{"check", "--help"}) + var output bytes.Buffer + command.SetOut(&output) + require.NoError(t, command.Execute()) + assert.Contains(t, output.String(), "--release") + assert.NotContains(t, output.String(), "--secret") + assert.NotContains(t, output.String(), "--configmap") +} + +func TestReportAndExitAgree(t *testing.T) { + tests := []struct { + name, status, expected string + checkErr error + }{ + {"healthy", checker.Healthy, checker.Healthy, nil}, + {"degraded", checker.Degraded, checker.Degraded, nil}, + {"unknown", checker.Unknown, checker.Unknown, nil}, + {"timeout after healthy sample", checker.Healthy, checker.Unknown, context.DeadlineExceeded}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + var output bytes.Buffer + err := finishReport(&output, "json", checker.Report{Status: test.status}, test.checkErr) + var report checker.Report + require.NoError(t, json.Unmarshal(output.Bytes(), &report)) + assert.Equal(t, test.expected, report.Status) + if test.expected == checker.Healthy { + require.NoError(t, err) + } else { + require.Error(t, err) + } + if test.checkErr != nil { + require.ErrorIs(t, err, test.checkErr) + assert.NotEmpty(t, report.Error) + } + }) + } +} diff --git a/cmd/root.go b/cmd/root.go index f2d7686..4ae0b28 100644 --- a/cmd/root.go +++ b/cmd/root.go @@ -6,6 +6,7 @@ import ( "github.com/spf13/cobra" "github.com/stackvista/stackstate-backup-cli/cmd/clickhouse" "github.com/stackvista/stackstate-backup-cli/cmd/elasticsearch" + "github.com/stackvista/stackstate-backup-cli/cmd/replication" "github.com/stackvista/stackstate-backup-cli/cmd/settings" "github.com/stackvista/stackstate-backup-cli/cmd/stackgraph" "github.com/stackvista/stackstate-backup-cli/cmd/version" @@ -56,12 +57,13 @@ func init() { // Add commands that don't need backup config flags rootCmd.AddCommand(version.Cmd()) + rootCmd.AddCommand(replication.Cmd()) } var rootCmd = &cobra.Command{ Use: "sts-backup", - Short: "Backup and restore tool for SUSE Observability platform", - Long: `A CLI tool for managing backups and restores for SUSE Observability platform running on Kubernetes.`, + Short: "Backup, restore and replication checks for SUSE Observability", + Long: `A CLI tool for managing backups, restores and database replication checks for SUSE Observability running on Kubernetes.`, } func Execute() { diff --git a/docs/replication.md b/docs/replication.md new file mode 100644 index 0000000..d8cd183 --- /dev/null +++ b/docs/replication.md @@ -0,0 +1,157 @@ +# Check database replication + +`sts-backup replication check` inspects the chart-managed HA databases in one +SUSE Observability Helm release. It works from a workstation or a Kubernetes +Job and does not require the backup ConfigMap, backup Secret, or enabled backups. + +```bash +sts-backup replication check --namespace observability --release suse-observability +``` + +The command returns exit code **0** only when every selected component reports +healthy replication. Any degraded, missing, inaccessible, unsupported or +incompletely described component returns exit code **1**. A missing component +is never silently skipped. Non-HA installations do not meet the checker's +minimum application replication requirement. + +Use JSON for automation: + +```bash +sts-backup replication check \ + --namespace observability --release suse-observability \ + --output json +``` + +The report identifies the namespace, release, observation time, selected +components, status (`healthy`, `degraded` or `unknown`) and diagnostic messages. +The status describes only the selected checks. + +## Wait for recovery + +```bash +sts-backup replication check \ + --namespace observability --release suse-observability \ + --wait --timeout 15m --interval 10s --stable-for 30s \ + --output json > replication.json +``` + +Wait mode requires consecutive healthy observations spanning `--stable-for`. +An unsuccessful observation resets that period. Progress goes to stderr; +stdout contains one final report. The overall deadline includes database +queries. A timeout or cancellation returns nonzero, even if an earlier +observation was healthy. `--request-timeout` bounds each API request or query. + +Select an explicit subset if a database is intentionally disabled: + +```bash +sts-backup replication check \ + --namespace observability --release suse-observability \ + --components hdfs,elasticsearch,kafka +``` + +This selection does not validate the omitted database. + +## What is checked + +The checker discovers StatefulSets by the Helm release's +`app.kubernetes.io/instance` label and identifies the database containers used +by the product chart. It verifies the desired pods exist, belong to those +StatefulSets, are Ready, and are not terminating or undergoing a rollout. +It queries database state and invalidates the observation if Kubernetes +resource versions change during those queries. + +| Component | Replication evidence | +|---|---| +| HDFS | Configured default block replication is at least two; all expected DataNodes are live; the NameNode is out of safe mode; no missing, corrupt, under-replicated or pending-replication blocks are reported by JMX. | +| Elasticsearch | Expected members are present; health is green; every returned index has at least one replica shard; no shards are unassigned, initializing or relocating. | +| Kafka | Every described partition has at least two distinct assigned replicas, complete ISR membership and an in-sync leader. Summaries and partition descriptions must agree. Both `__consumer_offsets` and `__transaction_state` must exist. | +| ClickHouse | Every discovered member is queried. Replicated table groups have their expected active replicas, live coordination sessions and no read-only members. Replication logs are caught up, and no replication queue tasks other than background `MERGE_PARTS` remain. Missing or duplicate table replicas and query exceptions fail the check. | + +The Kafka check does not create missing internal topics: initialize the +corresponding workloads and repeat the check. The ClickHouse check requires +replicated-table evidence and does not classify an empty result as healthy. +Unreplicated ClickHouse tables are outside its scope. + +## Access and supported layouts + +The command uses the normal kubeconfig selection or a Kubernetes service +account when running in a Pod. It requires listing Pods and StatefulSets and +creating `pods/exec` requests in the target namespace. + +**`pods/exec` permission allows arbitrary commands in pods.** The checker +itself executes only fixed read-only queries. Use a dedicated identity and +grant this permission only in the installation's namespace. It does not scale +workloads, change PDBs, annotate resources, repair databases or invoke restore +operations. No backup credentials are loaded. + +Queries use the database tools already present in the product containers: +`hdfs` and `curl` in the NameNode, `curl` in Elasticsearch, +`kafka-topics.sh` in Kafka, and `clickhouse-client` in ClickHouse. +Custom images, container names, external databases and alternative NameNode +topologies are not supported by these initial adapters. Failed discovery or +unsupported response formats produce `unknown`, not success. + +The initial HTTP adapters use the chart's NameNode and Elasticsearch ports. +For authenticated Kafka, supply a client properties file already mounted in +the broker and an appropriate bootstrap address: + +```bash +sts-backup replication check -n observability --release suse-observability \ + --components kafka \ + --kafka-bootstrap-server suse-observability-kafka:9092 \ + --kafka-client-properties /mounted/client.properties +``` + +The credentials must be able to describe all topics in the installation. +Elasticsearch uses the pod's `ELASTIC_PASSWORD` when present. For HTTPS, +use `--elasticsearch-scheme https`, `--elasticsearch-ca` with a CA path inside +the pod, and `--elasticsearch-server-name` matching the server certificate. +The server name is resolved to loopback inside that pod; certificate +verification is not disabled. + +ClickHouse uses the pod's `CLICKHOUSE_ADMIN_USER`, +`CLICKHOUSE_ADMIN_PASSWORD` and `CLICKHOUSE_TCP_PORT`. Credentials stay inside +the pod. Query stderr is suppressed because database tools can echo credentials; +when a query fails, inspect the component's configuration through your normal +administrative procedure. Query output is size-limited and excess output is +treated as unverified. + +## Kubernetes Job + +[examples/replication/job.yaml](../examples/replication/job.yaml) contains a +dedicated ServiceAccount, namespace-scoped RBAC and a Job using in-cluster +credentials. Adapt its namespace, release and image reference before use. +The Job has no retries: a failure requires investigation and an explicit rerun. + +No new container image is published by this change. To package the CLI, build +a static Linux binary from this repository and use the example SUSE BCI image: + +```bash +CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -o examples/replication/sts-backup . +docker build -t registry.example.com/observability/sts-backup:replication-checker \ + examples/replication +``` + +Publish the image to your registry through your normal image delivery process +and replace the Job's example image reference. Use the architecture required +by your nodes. The build copies the local binary; it does not download an +unverified executable. + +## Maintenance boundary + +This is a sampled replication report, **not a “safe to remove this node” +certificate or a maintenance lock**. It does not prevent another process from +starting maintenance, provide an atomic database snapshot, or prove continued +health between observations. + +This first version does not validate Longhorn volume health or placement, +spare capacity, HBase region assignment and WAL recovery, ZooKeeper quorum, +VictoriaMetrics redundancy, backup freshness, or the consequences of removing +a particular node. HDFS's default replication setting does not prove that +every file has the same replication policy. It is not a complete implementation +of the product's node-maintenance checklist. + +Continue to serialize maintenance, preserve storage redundancy, follow the +documented recovery procedure and check these additional requirements. +Run this checker after the affected node or replacement can schedule workloads. +Only proceed when the report and the remaining maintenance checks pass. diff --git a/examples/replication/.dockerignore b/examples/replication/.dockerignore new file mode 100644 index 0000000..b80331d --- /dev/null +++ b/examples/replication/.dockerignore @@ -0,0 +1,2 @@ +* +!sts-backup diff --git a/examples/replication/Dockerfile b/examples/replication/Dockerfile new file mode 100644 index 0000000..0a1d5a2 --- /dev/null +++ b/examples/replication/Dockerfile @@ -0,0 +1,4 @@ +FROM registry.suse.com/bci/bci-micro:15.7@sha256:9e01097b36048042e276dd40e7661941ac4ba909237904bae99540eb90c9a5c6 +COPY sts-backup /usr/local/bin/sts-backup +USER 65532:65532 +ENTRYPOINT ["/usr/local/bin/sts-backup"] diff --git a/examples/replication/job.yaml b/examples/replication/job.yaml new file mode 100644 index 0000000..203ab17 --- /dev/null +++ b/examples/replication/job.yaml @@ -0,0 +1,77 @@ +apiVersion: v1 +kind: ServiceAccount +metadata: + name: replication-checker + namespace: observability +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: Role +metadata: + name: replication-checker + namespace: observability +rules: + - apiGroups: [""] + resources: ["pods"] + verbs: ["get", "list"] + - apiGroups: ["apps"] + resources: ["statefulsets"] + verbs: ["get", "list"] + - apiGroups: [""] + resources: ["pods/exec"] + verbs: ["create"] +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: RoleBinding +metadata: + name: replication-checker + namespace: observability +subjects: + - kind: ServiceAccount + name: replication-checker + namespace: observability +roleRef: + apiGroup: rbac.authorization.k8s.io + kind: Role + name: replication-checker +--- +apiVersion: batch/v1 +kind: Job +metadata: + name: suse-observability-replication-check + namespace: observability +spec: + backoffLimit: 0 + activeDeadlineSeconds: 960 + ttlSecondsAfterFinished: 86400 + template: + spec: + serviceAccountName: replication-checker + restartPolicy: Never + securityContext: + runAsNonRoot: true + runAsUser: 65532 + runAsGroup: 65532 + seccompProfile: + type: RuntimeDefault + containers: + - name: checker + image: registry.example.com/observability/sts-backup:replication-checker + args: + - replication + - check + - --namespace=observability + - --release=suse-observability + - --wait + - --timeout=15m + - --stable-for=30s + resources: + requests: + cpu: 100m + memory: 128Mi + limits: + memory: 256Mi + securityContext: + allowPrivilegeEscalation: false + readOnlyRootFilesystem: true + capabilities: + drop: ["ALL"] diff --git a/internal/app/replication.go b/internal/app/replication.go new file mode 100644 index 0000000..0cf8937 --- /dev/null +++ b/internal/app/replication.go @@ -0,0 +1,17 @@ +package app + +import ( + "fmt" + + "github.com/stackvista/stackstate-backup-cli/internal/clients/k8s" + "github.com/stackvista/stackstate-backup-cli/internal/orchestration/replication" +) + +// NewReplicationChecker requires Kubernetes access, but no backup ConfigMap or Secret. +func NewReplicationChecker(kubeconfig string, options replication.Options) (*replication.Checker, error) { + client, err := k8s.NewClient(kubeconfig, false) + if err != nil { + return nil, fmt.Errorf("create Kubernetes client: %w", err) + } + return replication.New(client, options) +} diff --git a/internal/clients/k8s/exec.go b/internal/clients/k8s/exec.go new file mode 100644 index 0000000..25336a3 --- /dev/null +++ b/internal/clients/k8s/exec.go @@ -0,0 +1,53 @@ +package k8s + +import ( + "bytes" + "context" + "fmt" + "io" + "net/http" + + corev1 "k8s.io/api/core/v1" + "k8s.io/client-go/kubernetes/scheme" + "k8s.io/client-go/tools/remotecommand" +) + +const maxExecOutput = 8 << 20 + +type boundedBuffer struct { + buffer bytes.Buffer + err error +} + +func (b *boundedBuffer) Write(p []byte) (int, error) { + if len(p) > maxExecOutput-b.buffer.Len() { + b.err = fmt.Errorf("query output exceeds %d bytes", maxExecOutput) + return 0, b.err + } + return b.buffer.Write(p) +} + +// Exec runs a command without a TTY, bounded by the caller's context and output limit. +func (c *Client) Exec(ctx context.Context, namespace, pod, container string, command []string) ([]byte, error) { + request := c.clientset.CoreV1().RESTClient().Post(). + Namespace(namespace).Resource("pods").Name(pod).SubResource("exec"). + VersionedParams(&corev1.PodExecOptions{ + Container: container, + Command: command, + Stdout: true, + Stderr: true, + }, scheme.ParameterCodec) + executor, err := remotecommand.NewSPDYExecutor(c.restConfig, http.MethodPost, request.URL()) + if err != nil { + return nil, fmt.Errorf("create pod executor: %w", err) + } + var output boundedBuffer + // Database tools can echo authentication details in stderr. + if err := executor.StreamWithContext(ctx, remotecommand.StreamOptions{Stdout: &output, Stderr: io.Discard}); err != nil { + return nil, fmt.Errorf("query pod %s/%s: %w", namespace, pod, err) + } + if output.err != nil { + return nil, output.err + } + return output.buffer.Bytes(), nil +} diff --git a/internal/clients/k8s/exec_test.go b/internal/clients/k8s/exec_test.go new file mode 100644 index 0000000..8f1cad2 --- /dev/null +++ b/internal/clients/k8s/exec_test.go @@ -0,0 +1,20 @@ +package k8s + +import ( + "io" + "strings" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestExecOutputLimit(t *testing.T) { + var output boundedBuffer + n, err := output.Write([]byte("report")) + require.NoError(t, err) + assert.Equal(t, 6, n) + _, err = io.Copy(&output, io.LimitReader(strings.NewReader(strings.Repeat("x", maxExecOutput)), int64(maxExecOutput))) + require.ErrorContains(t, err, "query output exceeds") + assert.LessOrEqual(t, output.buffer.Len(), maxExecOutput) +} diff --git a/internal/orchestration/replication/checker.go b/internal/orchestration/replication/checker.go new file mode 100644 index 0000000..c5884b6 --- /dev/null +++ b/internal/orchestration/replication/checker.go @@ -0,0 +1,86 @@ +package replication + +import ( + "context" + "fmt" + "time" +) + +// Checker observes databases without invoking backup or restore operations. +type Checker struct { + kube Kubernetes + options Options +} + +// New creates a checker with an explicit release scope. +func New(kube Kubernetes, options Options) (*Checker, error) { + if err := options.Validate(); err != nil { + return nil, err + } + if options.KafkaBootstrapServer == "" { + options.KafkaBootstrapServer = "localhost:9092" + } + if options.ElasticsearchHost == "" { + options.ElasticsearchHost = "127.0.0.1" + } + return &Checker{kube: kube, options: options}, nil +} + +// Check queries every selected component and rechecks Kubernetes membership afterward. +func (c *Checker) Check(ctx context.Context) Report { + report := Report{ + CheckedAt: time.Now().UTC(), Namespace: c.options.Namespace, Release: c.options.Release, + Checks: make([]Result, 0, len(c.options.Components)), + } + before, err := c.discover(ctx) + for _, component := range c.options.Components { + if err != nil { + report.Checks = append(report.Checks, result(component, Unknown, err.Error())) + continue + } + report.Checks = append(report.Checks, c.checkComponent(ctx, before, component)) + } + if err == nil { + after, afterErr := c.discover(ctx) + if afterErr != nil || before.fingerprint() != after.fingerprint() { + message := "Kubernetes membership or status changed during the checks; repeat the observation" + if afterErr != nil { + message = afterErr.Error() + } + for n := range report.Checks { + report.Checks[n].Status = Unknown + report.Checks[n].Messages = append(report.Checks[n].Messages, message) + } + } + } + report.Status = reportStatus(report.Checks) + return report +} + +func (c *Checker) query(ctx context.Context, pod, container string, command []string) ([]byte, error) { + ctx, cancel := context.WithTimeout(ctx, c.options.RequestTimeout) + defer cancel() + return c.kube.Exec(ctx, c.options.Namespace, pod, container, command) +} + +func (c *Checker) checkComponent(ctx context.Context, inventory inventory, component string) Result { + switch component { + case "hdfs": + return c.checkHDFS(ctx, inventory) + case "elasticsearch": + return c.checkElasticsearch(ctx, inventory) + case "kafka": + return c.checkKafka(ctx, inventory) + case "clickhouse": + return c.checkClickHouse(ctx, inventory) + default: + return result(component, Unknown, "unsupported component") + } +} + +func expectedMembers(members []member) error { + if len(members) < minReplicas { + return fmt.Errorf("at least %d application replicas are required; discovered %d", minReplicas, len(members)) + } + return nil +} diff --git a/internal/orchestration/replication/checker_test.go b/internal/orchestration/replication/checker_test.go new file mode 100644 index 0000000..b866d72 --- /dev/null +++ b/internal/orchestration/replication/checker_test.go @@ -0,0 +1,158 @@ +package replication + +import ( + "context" + "fmt" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/kubernetes/fake" + "k8s.io/utils/ptr" +) + +type fakeKubernetes struct { + client kubernetes.Interface + exec func(context.Context, string, string, string, []string) ([]byte, error) + calls int +} + +func (f *fakeKubernetes) Clientset() kubernetes.Interface { return f.client } + +func (f *fakeKubernetes) Exec(ctx context.Context, namespace, pod, container string, command []string) ([]byte, error) { + f.calls++ + return f.exec(ctx, namespace, pod, container, command) +} + +func testOptions() Options { + return Options{Namespace: "test", Release: "observability", Components: []string{"kafka"}, RequestTimeout: time.Second, ElasticsearchScheme: "http"} +} + +func kafkaObjects() []runtime.Object { + labels := map[string]string{"app.kubernetes.io/instance": "observability"} + workload := &appsv1.StatefulSet{ + ObjectMeta: metav1.ObjectMeta{Name: "kafka", Namespace: "test", UID: "sts-kafka", Generation: 1, ResourceVersion: "1", Labels: labels}, + Spec: appsv1.StatefulSetSpec{ + Replicas: ptr.To(int32(2)), + Template: corev1.PodTemplateSpec{Spec: corev1.PodSpec{Containers: []corev1.Container{{Name: "kafka"}}}}, + }, + Status: appsv1.StatefulSetStatus{ReadyReplicas: 2, ObservedGeneration: 1, CurrentRevision: "one", UpdateRevision: "one"}, + } + objects := []runtime.Object{workload} + for n := 0; n < 2; n++ { + name := fmt.Sprintf("kafka-%d", n) + pod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, Namespace: "test", UID: types.UID(name), ResourceVersion: "1", Labels: labels, + OwnerReferences: []metav1.OwnerReference{{Kind: "StatefulSet", Name: "kafka", UID: workload.UID, Controller: ptr.To(true)}}, + }, + Spec: corev1.PodSpec{NodeName: fmt.Sprintf("node-%d", n), Containers: []corev1.Container{{Name: "kafka"}}}, + Status: corev1.PodStatus{Phase: corev1.PodRunning, Conditions: []corev1.PodCondition{ + {Type: corev1.PodReady, Status: corev1.ConditionTrue}, + }}, + } + objects = append(objects, pod) + } + return objects +} + +func TestCheckerQueriesReadyPodsAndDoesNotMutateKubernetes(t *testing.T) { + client := fake.NewSimpleClientset(kafkaObjects()...) + kube := &fakeKubernetes{client: client, exec: func(ctx context.Context, namespace, pod, container string, command []string) ([]byte, error) { + assert.Equal(t, "test", namespace) + assert.Equal(t, "kafka-0", pod) + assert.Equal(t, "kafka", container) + assert.Equal(t, []string{"kafka-topics.sh", "--bootstrap-server", "localhost:9092", "--describe"}, command) + _, deadline := ctx.Deadline() + assert.True(t, deadline) + return []byte(kafkaFixture()), nil + }} + probe, err := New(kube, testOptions()) + require.NoError(t, err) + report := probe.Check(context.Background()) + require.Equal(t, Healthy, report.Status, report) + assert.Equal(t, 1, kube.calls) + for _, action := range client.Actions() { + assert.Equal(t, "list", action.GetVerb()) + } +} + +func TestCheckerRejectsIncompleteTopology(t *testing.T) { + tests := []struct { + name string + mutate func([]runtime.Object) []runtime.Object + }{ + {"missing pod", func(objects []runtime.Object) []runtime.Object { return objects[:2] }}, + {"not ready", func(objects []runtime.Object) []runtime.Object { + objects[1].(*corev1.Pod).Status.Conditions[0].Status = corev1.ConditionFalse + return objects + }}, + {"terminating", func(objects []runtime.Object) []runtime.Object { + objects[1].(*corev1.Pod).DeletionTimestamp = ptr.To(metav1.Now()) + return objects + }}, + {"wrong owner", func(objects []runtime.Object) []runtime.Object { + objects[1].(*corev1.Pod).OwnerReferences[0].UID = "other" + return objects + }}, + {"wrong release", func(objects []runtime.Object) []runtime.Object { + objects[0].(*appsv1.StatefulSet).Labels = map[string]string{"app.kubernetes.io/instance": "other"} + return objects + }}, + {"rolling update", func(objects []runtime.Object) []runtime.Object { + objects[0].(*appsv1.StatefulSet).Status.UpdateRevision = "two" + return objects + }}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + kube := &fakeKubernetes{client: fake.NewSimpleClientset(test.mutate(kafkaObjects())...)} + probe, err := New(kube, testOptions()) + require.NoError(t, err) + assert.Equal(t, Unknown, probe.Check(context.Background()).Status) + assert.Zero(t, kube.calls) + }) + } +} + +func TestCheckerRejectsMembershipChangeDuringQueries(t *testing.T) { + client := fake.NewSimpleClientset(kafkaObjects()...) + kube := &fakeKubernetes{client: client, exec: func(ctx context.Context, namespace, pod, _ string, _ []string) ([]byte, error) { + current, err := client.CoreV1().Pods(namespace).Get(ctx, pod, metav1.GetOptions{}) + require.NoError(t, err) + current.ResourceVersion = "2" + _, err = client.CoreV1().Pods(namespace).Update(ctx, current, metav1.UpdateOptions{}) + require.NoError(t, err) + return []byte(kafkaFixture()), nil + }} + probe, err := New(kube, testOptions()) + require.NoError(t, err) + report := probe.Check(context.Background()) + assert.Equal(t, Unknown, report.Status) + assert.Contains(t, report.Checks[0].Messages, "Kubernetes membership or status changed during the checks; repeat the observation") +} + +func TestQueryFailureCannotPass(t *testing.T) { + kube := &fakeKubernetes{client: fake.NewSimpleClientset(kafkaObjects()...), exec: func(context.Context, string, string, string, []string) ([]byte, error) { + return nil, fmt.Errorf("query not authorized") + }} + probe, err := New(kube, testOptions()) + require.NoError(t, err) + assert.Equal(t, Unknown, probe.Check(context.Background()).Status) +} + +func TestInvalidScopeRejected(t *testing.T) { + for _, components := range [][]string{nil, {"kafka", "kafka"}, {"not-a-store"}} { + options := testOptions() + options.Components = components + _, err := New(nil, options) + require.Error(t, err) + } +} diff --git a/internal/orchestration/replication/clickhouse.go b/internal/orchestration/replication/clickhouse.go new file mode 100644 index 0000000..ff76adb --- /dev/null +++ b/internal/orchestration/replication/clickhouse.go @@ -0,0 +1,139 @@ +package replication + +import ( + "context" + "encoding/json" + "fmt" + "sort" +) + +const clickhouseSQL = `SELECT database, table, zookeeper_path, replica_name, +is_readonly, is_session_expired, total_replicas, active_replicas, +log_max_index, log_pointer, absolute_delay, last_queue_update_exception, zookeeper_exception, +ifNull(pending_data_tasks, 0) AS pending_data_tasks +FROM system.replicas +LEFT JOIN ( + SELECT database, table, countIf(type != 'MERGE_PARTS') AS pending_data_tasks + FROM system.replication_queue GROUP BY database, table +) USING (database, table) +WHERE database NOT IN ('system', 'INFORMATION_SCHEMA', 'information_schema') +FORMAT JSON` + +const clickhouseQuery = `export CLICKHOUSE_PASSWORD="${CLICKHOUSE_ADMIN_PASSWORD:?missing ClickHouse credentials}" +exec clickhouse-client --host 127.0.0.1 --port "${CLICKHOUSE_TCP_PORT:-9000}" \ + --user "${CLICKHOUSE_ADMIN_USER:?missing ClickHouse user}" --readonly 1 --query "$1"` + +type replicaObservation struct { + path string + name string + pod string + total int64 + problems []string +} + +func (c *Checker) checkClickHouse(ctx context.Context, inventory inventory) Result { + members, err := inventory.members("clickhouse") + if err != nil { + return result("clickhouse", Unknown, err.Error()) + } + var observations []replicaObservation + for _, member := range members { + if member.replicas < minReplicas { + return result("clickhouse", Degraded, fmt.Sprintf("%s belongs to a shard with fewer than two replicas", member.pod.Name)) + } + command := []string{"bash", "-ec", clickhouseQuery, "replication-check", clickhouseSQL} + data, err := c.query(ctx, member.pod.Name, "clickhouse", command) + if err != nil { + return result("clickhouse", Unknown, err.Error()) + } + rows, err := parseClickHouse(data, member.pod.Name, member.replicas) + if err != nil { + return result("clickhouse", Unknown, fmt.Sprintf("%s: %v", member.pod.Name, err)) + } + observations = append(observations, rows...) + } + return evaluateClickHouse(observations) +} + +func parseClickHouse(data []byte, pod string, expected int) ([]replicaObservation, error) { + var response struct { + Data []map[string]json.RawMessage `json:"data"` + } + if err := json.Unmarshal(data, &response); err != nil || len(response.Data) == 0 { + return nil, fmt.Errorf("no replicated-table evidence returned") + } + var observations []replicaObservation + for _, row := range response.Data { + observation, err := parseReplica(row, pod, expected) + if err != nil { + return nil, err + } + observations = append(observations, observation) + } + return observations, nil +} + +func parseReplica(row map[string]json.RawMessage, pod string, expected int) (replicaObservation, error) { + fields := make(map[string]string) + for _, key := range []string{"database", "table", "zookeeper_path", "replica_name", "last_queue_update_exception", "zookeeper_exception"} { + value, err := text(row, key) + if err != nil { + return replicaObservation{}, err + } + fields[key] = value + } + counts, err := numbers(row, "is_readonly", "is_session_expired", "total_replicas", "active_replicas", + "log_max_index", "log_pointer", "pending_data_tasks", "absolute_delay") + if err != nil { + return replicaObservation{}, err + } + if fields["zookeeper_path"] == "" || fields["replica_name"] == "" || fields["database"] == "" || fields["table"] == "" { + return replicaObservation{}, fmt.Errorf("missing replicated-table identity") + } + observation := replicaObservation{path: fields["zookeeper_path"], name: fields["replica_name"], pod: pod, total: counts["total_replicas"]} + if observation.total < minReplicas || observation.total != int64(expected) || counts["active_replicas"] != observation.total { + observation.problems = append(observation.problems, fmt.Sprintf("%d/%d replicas active; %d expected", counts["active_replicas"], observation.total, expected)) + } + if counts["is_readonly"] != 0 || counts["is_session_expired"] != 0 { + observation.problems = append(observation.problems, "replica is read-only or coordination session expired") + } + if counts["log_pointer"] <= counts["log_max_index"] || counts["pending_data_tasks"] != 0 { + observation.problems = append(observation.problems, fmt.Sprintf("replication backlog: %d data tasks; reported delay %ds", counts["pending_data_tasks"], counts["absolute_delay"])) + } + if fields["last_queue_update_exception"] != "" || fields["zookeeper_exception"] != "" { + observation.problems = append(observation.problems, "replication queue or coordination query reported an exception") + } + return observation, nil +} + +func evaluateClickHouse(observations []replicaObservation) Result { + groups := make(map[string][]replicaObservation) + var problems []string + for _, observation := range observations { + groups[observation.path] = append(groups[observation.path], observation) + for _, problem := range observation.problems { + problems = append(problems, fmt.Sprintf("%s %s: %s", observation.pod, observation.path, problem)) + } + } + if len(groups) == 0 { + return result("clickhouse", Unknown, "no replicated tables found") + } + for path, replicas := range groups { + names := make(map[string]bool) + pods := make(map[string]bool) + for _, replica := range replicas { + if names[replica.name] || pods[replica.pod] || replica.total != replicas[0].total { + return result("clickhouse", Unknown, "duplicate or inconsistent replica evidence for "+path) + } + names[replica.name], pods[replica.pod] = true, true + } + if int64(len(replicas)) != replicas[0].total { + problems = append(problems, fmt.Sprintf("%s: queried %d/%d table replicas", path, len(replicas), replicas[0].total)) + } + } + if len(problems) > 0 { + sort.Strings(problems) + return Result{Component: "clickhouse", Status: Degraded, Messages: problems} + } + return result("clickhouse", Healthy, fmt.Sprintf("%d replicated table groups checked on every member; no pending data replication tasks", len(groups))) +} diff --git a/internal/orchestration/replication/discovery.go b/internal/orchestration/replication/discovery.go new file mode 100644 index 0000000..981d49b --- /dev/null +++ b/internal/orchestration/replication/discovery.go @@ -0,0 +1,125 @@ +package replication + +import ( + "context" + "encoding/json" + "fmt" + "slices" + + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/labels" + "k8s.io/client-go/kubernetes" +) + +// Kubernetes is the read/query surface needed by the checker. +type Kubernetes interface { + Clientset() kubernetes.Interface + Exec(context.Context, string, string, string, []string) ([]byte, error) +} + +type inventory struct { + workloads []appsv1.StatefulSet + pods []corev1.Pod +} + +type member struct { + pod corev1.Pod + replicas int +} + +func (c *Checker) discover(ctx context.Context) (inventory, error) { + ctx, cancel := context.WithTimeout(ctx, c.options.RequestTimeout) + defer cancel() + options := metav1.ListOptions{LabelSelector: labels.Set{"app.kubernetes.io/instance": c.options.Release}.String()} + sets, err := c.kube.Clientset().AppsV1().StatefulSets(c.options.Namespace).List(ctx, options) + if err != nil { + return inventory{}, fmt.Errorf("list release StatefulSets: %w", err) + } + pods, err := c.kube.Clientset().CoreV1().Pods(c.options.Namespace).List(ctx, options) + if err != nil { + return inventory{}, fmt.Errorf("list release pods: %w", err) + } + return inventory{workloads: sets.Items, pods: pods.Items}, nil +} + +func hasContainer(containers []corev1.Container, name string) bool { + return slices.ContainsFunc(containers, func(container corev1.Container) bool { return container.Name == name }) +} + +func (i inventory) members(container string) ([]member, error) { + var members []member + found := false + for _, workload := range i.workloads { + if !hasContainer(workload.Spec.Template.Spec.Containers, container) { + continue + } + found = true + pods, err := i.workloadPods(workload) + if err != nil { + return nil, err + } + for _, pod := range pods { + members = append(members, member{pod: pod, replicas: int(*workload.Spec.Replicas)}) + } + } + if !found { + return nil, fmt.Errorf("no chart-managed StatefulSet with container %q; check release, selected components and chart layout", container) + } + slices.SortFunc(members, func(a, b member) int { + if a.pod.Name < b.pod.Name { + return -1 + } + if a.pod.Name > b.pod.Name { + return 1 + } + return 0 + }) + return members, nil +} + +func (i inventory) workloadPods(workload appsv1.StatefulSet) ([]corev1.Pod, error) { + if workload.Spec.Replicas == nil || *workload.Spec.Replicas < 1 { + return nil, fmt.Errorf("%s has no desired replicas", workload.Name) + } + expected := *workload.Spec.Replicas + if workload.DeletionTimestamp != nil || workload.Status.ObservedGeneration < workload.Generation || + workload.Status.ReadyReplicas != expected || workload.Status.CurrentRevision != workload.Status.UpdateRevision { + return nil, fmt.Errorf("%s is not converged: %d/%d replicas ready", workload.Name, workload.Status.ReadyReplicas, expected) + } + var pods []corev1.Pod + for _, pod := range i.pods { + owner := metav1.GetControllerOf(&pod) + if owner == nil || owner.UID != workload.UID || owner.Kind != "StatefulSet" { + continue + } + if pod.DeletionTimestamp != nil || pod.Spec.NodeName == "" || !podReady(pod) { + return nil, fmt.Errorf("%s is terminating, unscheduled or not Ready", pod.Name) + } + pods = append(pods, pod) + } + if len(pods) != int(expected) { + return nil, fmt.Errorf("%s has %d/%d current pods", workload.Name, len(pods), expected) + } + return pods, nil +} + +func podReady(pod corev1.Pod) bool { + return pod.Status.Phase == corev1.PodRunning && slices.ContainsFunc(pod.Status.Conditions, func(condition corev1.PodCondition) bool { + return condition.Type == corev1.PodReady && condition.Status == corev1.ConditionTrue + }) +} + +func (i inventory) fingerprint() string { + var entries []string + for _, workload := range i.workloads { + entries = append(entries, fmt.Sprintf("sts:%s:%s", workload.UID, workload.ResourceVersion)) + } + for _, pod := range i.pods { + entries = append(entries, fmt.Sprintf("pod:%s:%s", pod.UID, pod.ResourceVersion)) + } + slices.Sort(entries) + data, _ := json.Marshal(entries) + return string(data) +} diff --git a/internal/orchestration/replication/elasticsearch.go b/internal/orchestration/replication/elasticsearch.go new file mode 100644 index 0000000..bcbba56 --- /dev/null +++ b/internal/orchestration/replication/elasticsearch.go @@ -0,0 +1,78 @@ +package replication + +import ( + "context" + "encoding/json" + "fmt" + "sort" +) + +const elasticsearchQuery = `scheme="$1" +ca="$2" +host="$3" +set -- --fail --silent --show-error --max-time 20 +if [ -n "${ELASTIC_PASSWORD:-}" ]; then + set -- "$@" --user "elastic:${ELASTIC_PASSWORD}" +fi +if [ -n "$ca" ]; then + set -- "$@" --cacert "$ca" +fi +exec curl "$@" --resolve "${host}:9200:127.0.0.1" "${scheme}://${host}:9200/_cluster/health?level=indices"` + +func (c *Checker) checkElasticsearch(ctx context.Context, inventory inventory) Result { + members, err := inventory.members("elasticsearch") + if err != nil { + return result("elasticsearch", Unknown, err.Error()) + } + if err := expectedMembers(members); err != nil { + return result("elasticsearch", Degraded, err.Error()) + } + command := []string{"bash", "-ec", elasticsearchQuery, "replication-check", c.options.ElasticsearchScheme, c.options.ElasticsearchCA, c.options.ElasticsearchHost} + data, err := c.query(ctx, members[0].pod.Name, "elasticsearch", command) + if err != nil { + return result("elasticsearch", Unknown, err.Error()) + } + return evaluateElasticsearch(data, len(members)) +} + +func evaluateElasticsearch(data []byte, expected int) Result { + var fields map[string]json.RawMessage + if err := json.Unmarshal(data, &fields); err != nil { + return result("elasticsearch", Unknown, "invalid cluster health response") + } + counts, err := numbers(fields, "number_of_nodes", "unassigned_shards", "initializing_shards", "relocating_shards") + if err != nil { + return result("elasticsearch", Unknown, err.Error()) + } + status, err := text(fields, "status") + if err != nil { + return result("elasticsearch", Unknown, err.Error()) + } + var indices map[string]map[string]json.RawMessage + if err := json.Unmarshal(fields["indices"], &indices); err != nil || len(indices) == 0 { + return result("elasticsearch", Unknown, "no index replication evidence returned") + } + var problems []string + if status != "green" || counts["number_of_nodes"] != int64(expected) { + problems = append(problems, fmt.Sprintf("cluster status %s; %d/%d expected nodes", status, counts["number_of_nodes"], expected)) + } + for _, key := range []string{"unassigned_shards", "initializing_shards", "relocating_shards"} { + if counts[key] > 0 { + problems = append(problems, fmt.Sprintf("%s=%d", key, counts[key])) + } + } + for name, index := range indices { + replicas, err := number(index, "number_of_replicas") + if err != nil { + return result("elasticsearch", Unknown, fmt.Sprintf("index %s: %v", name, err)) + } + if replicas < minReplicas-1 { + problems = append(problems, fmt.Sprintf("index %s has no replica shard", name)) + } + } + if len(problems) > 0 { + sort.Strings(problems) + return Result{Component: "elasticsearch", Status: Degraded, Messages: problems} + } + return result("elasticsearch", Healthy, fmt.Sprintf("%d indices have replica shards; all shards allocated with no recovery or relocation", len(indices))) +} diff --git a/internal/orchestration/replication/evaluators_test.go b/internal/orchestration/replication/evaluators_test.go new file mode 100644 index 0000000..e52a39a --- /dev/null +++ b/internal/orchestration/replication/evaluators_test.go @@ -0,0 +1,177 @@ +package replication + +import ( + "encoding/json" + "fmt" + "strings" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func encode(t *testing.T, value any) []byte { + t.Helper() + data, err := json.Marshal(value) + require.NoError(t, err) + return data +} + +func hdfsFixture(t *testing.T, mutate func(map[string]any)) []byte { + t.Helper() + fields := map[string]any{ + "name": "Hadoop:service=NameNode,name=FSNamesystem", + "UnderReplicatedBlocks": 0, "MissingBlocks": 0, "CorruptBlocks": 0, "PendingReplicationBlocks": 0, "NumLiveDataNodes": 3, "Safemode": "", + } + mutate(fields) + return encode(t, map[string]any{"configuredReplication": 3, "jmx": map[string]any{"beans": []any{fields}}}) +} + +func TestHDFSEvidence(t *testing.T) { + tests := []struct { + name string + mutate func(map[string]any) + status string + }{ + {"recovered", func(map[string]any) {}, Healthy}, + {"replicating", func(m map[string]any) { m["UnderReplicatedBlocks"] = 12 }, Degraded}, + {"missing block", func(m map[string]any) { m["MissingBlocks"] = 1 }, Degraded}, + {"corrupt block", func(m map[string]any) { m["CorruptBlocks"] = 1 }, Degraded}, + {"pending replication", func(m map[string]any) { m["PendingReplicationBlocks"] = 2 }, Degraded}, + {"missing datanode", func(m map[string]any) { m["NumLiveDataNodes"] = 2 }, Degraded}, + {"safe mode", func(m map[string]any) { m["Safemode"] = "ON" }, Degraded}, + {"missing metric", func(m map[string]any) { delete(m, "UnderReplicatedBlocks") }, Unknown}, + {"null metric", func(m map[string]any) { m["CorruptBlocks"] = nil }, Unknown}, + {"negative metric", func(m map[string]any) { m["MissingBlocks"] = -1 }, Unknown}, + {"unrelated bean", func(m map[string]any) { m["name"] = "Hadoop:service=Other" }, Unknown}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + assert.Equal(t, test.status, evaluateHDFS(hdfsFixture(t, test.mutate), 3).Status) + }) + } + data := strings.Replace(string(hdfsFixture(t, func(map[string]any) {})), `"configuredReplication":3`, `"configuredReplication":1`, 1) + assert.Equal(t, Degraded, evaluateHDFS([]byte(data), 3).Status) + assert.Equal(t, Unknown, evaluateHDFS([]byte(`{"beans":[]}`), 3).Status) +} + +func esFixture() map[string]any { + return map[string]any{ + "status": "green", "number_of_nodes": 3, "unassigned_shards": 0, "initializing_shards": 0, "relocating_shards": 0, + "indices": map[string]any{"events": map[string]any{"number_of_replicas": 1}}, + } +} + +func TestElasticsearchEvidence(t *testing.T) { + tests := []struct { + name string + mutate func(map[string]any) + status string + }{ + {"recovered", func(map[string]any) {}, Healthy}, + {"yellow", func(m map[string]any) { m["status"] = "yellow" }, Degraded}, + {"node missing", func(m map[string]any) { m["number_of_nodes"] = 2 }, Degraded}, + {"initializing", func(m map[string]any) { m["initializing_shards"] = 1 }, Degraded}, + {"relocating", func(m map[string]any) { m["relocating_shards"] = 1 }, Degraded}, + {"unassigned", func(m map[string]any) { m["unassigned_shards"] = 1 }, Degraded}, + {"green without replicas", func(m map[string]any) { + m["indices"] = map[string]any{"events": map[string]any{"number_of_replicas": 0}} + }, Degraded}, + {"empty cluster", func(m map[string]any) { m["indices"] = map[string]any{} }, Unknown}, + {"partial response", func(m map[string]any) { delete(m, "initializing_shards") }, Unknown}, + {"missing index replication", func(m map[string]any) { m["indices"] = map[string]any{"events": map[string]any{}} }, Unknown}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + fields := esFixture() + test.mutate(fields) + assert.Equal(t, test.status, evaluateElasticsearch(encode(t, fields), 3).Status) + }) + } +} + +func kafkaFixture() string { + var lines []string + for _, topic := range []string{"events", "__consumer_offsets", "__transaction_state"} { + lines = append(lines, fmt.Sprintf("Topic: %s TopicId: x PartitionCount: 1 ReplicationFactor: 2 Configs: min.insync.replicas=1", topic)) + lines = append(lines, fmt.Sprintf(" Topic: %s Partition: 0 Leader: 0 Replicas: 0,1 Isr: 1,0", topic)) + } + return strings.Join(lines, "\n") +} + +func TestKafkaEvidence(t *testing.T) { + tests := []struct { + name, from, to, status string + }{ + {"recovered", "Isr: 1,0", "Isr: 1,0", Healthy}, + {"single replica transaction state", "Leader: 0 Replicas: 0,1 Isr: 1,0", "Leader: 0 Replicas: 0 Isr: 0", Degraded}, + {"missing ISR", "Isr: 1,0", "Isr: 0", Degraded}, + {"wrong ISR", "Isr: 1,0", "Isr: 2,0", Degraded}, + {"missing leader", "Leader: 0", "Leader: -1", Degraded}, + {"duplicate replica", "Replicas: 0,1", "Replicas: 0,0", Degraded}, + {"empty ISR", "Isr: 1,0", "Isr:", Degraded}, + {"missing internal topic", "__transaction_state", "another-topic", Unknown}, + {"incomplete partitions", "PartitionCount: 1", "PartitionCount: 2", Unknown}, + {"wrong partition ID", "Partition: 0", "Partition: 5", Unknown}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + output := strings.ReplaceAll(kafkaFixture(), test.from, test.to) + assert.Equal(t, test.status, evaluateKafka(output).Status) + }) + } + assert.Equal(t, Unknown, evaluateKafka("").Status) + assert.Equal(t, Unknown, evaluateKafka(kafkaFixture()+"\n"+kafkaFixture()).Status) +} + +func clickhouseFixture() map[string]any { + return map[string]any{ + "database": "otel", "table": "traces", "zookeeper_path": "/clickhouse/tables/shard0/traces", + "replica_name": "replica0", "is_readonly": 0, "is_session_expired": 0, + "total_replicas": "2", "active_replicas": "2", "log_max_index": "100", "log_pointer": "101", + "pending_data_tasks": "0", "absolute_delay": "0", "last_queue_update_exception": "", "zookeeper_exception": "", + } +} + +func TestClickHouseReplicaEvidence(t *testing.T) { + tests := []struct { + name string + mutate func(map[string]any) + status string + }{ + {"recovered", func(map[string]any) {}, Healthy}, + {"merge backlog alone", func(m map[string]any) { m["queue_size"] = 8; m["merges_in_queue"] = 8 }, Healthy}, + {"lagging log", func(m map[string]any) { m["log_pointer"] = "100" }, Degraded}, + {"data backlog", func(m map[string]any) { m["pending_data_tasks"] = "1" }, Degraded}, + {"read only", func(m map[string]any) { m["is_readonly"] = 1 }, Degraded}, + {"expired session", func(m map[string]any) { m["is_session_expired"] = 1 }, Degraded}, + {"inactive replica", func(m map[string]any) { m["active_replicas"] = "1" }, Degraded}, + {"wrong replication", func(m map[string]any) { m["total_replicas"] = "1" }, Degraded}, + {"coordination exception", func(m map[string]any) { m["zookeeper_exception"] = "connection failed" }, Degraded}, + {"missing backlog evidence", func(m map[string]any) { delete(m, "pending_data_tasks") }, Unknown}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + row := clickhouseFixture() + test.mutate(row) + observations, err := parseClickHouse(encode(t, map[string]any{"data": []any{row}}), "pod0", 2) + if test.status == Unknown { + require.Error(t, err) + return + } + require.NoError(t, err) + peer := observations[0] + peer.pod, peer.name, peer.problems = "pod1", "replica1", nil + assert.Equal(t, test.status, evaluateClickHouse(append(observations, peer)).Status) + }) + } +} + +func TestClickHouseChecksEveryTableReplica(t *testing.T) { + observations, err := parseClickHouse(encode(t, map[string]any{"data": []any{clickhouseFixture()}}), "pod0", 2) + require.NoError(t, err) + assert.Equal(t, Degraded, evaluateClickHouse(observations).Status) + assert.Equal(t, Unknown, evaluateClickHouse(append(observations, observations...)).Status) + _, err = parseClickHouse([]byte(`{"data":[]}`), "pod0", 2) + require.Error(t, err) +} diff --git a/internal/orchestration/replication/hdfs.go b/internal/orchestration/replication/hdfs.go new file mode 100644 index 0000000..6b6285e --- /dev/null +++ b/internal/orchestration/replication/hdfs.go @@ -0,0 +1,85 @@ +package replication + +import ( + "context" + "encoding/json" + "fmt" +) + +const hdfsQuery = `unset HADOOP_OPTS +printf '{"configuredReplication":' +hdfs getconf -confKey dfs.replication +printf ',"jmx":' +curl --fail --silent --show-error --max-time 20 'http://127.0.0.1:50070/jmx' +printf '}'` + +func (c *Checker) checkHDFS(ctx context.Context, inventory inventory) Result { + namenodes, err := inventory.members("namenode") + if err != nil { + return result("hdfs", Unknown, err.Error()) + } + if len(namenodes) != 1 { + return result("hdfs", Unknown, "expected the chart's single NameNode; external or HA NameNode layouts are not supported") + } + datanodes, err := inventory.members("datanode") + if err != nil { + return result("hdfs", Unknown, err.Error()) + } + if err := expectedMembers(datanodes); err != nil { + return result("hdfs", Degraded, err.Error()) + } + data, err := c.query(ctx, namenodes[0].pod.Name, "namenode", []string{"bash", "-ec", hdfsQuery}) + if err != nil { + return result("hdfs", Unknown, err.Error()) + } + return evaluateHDFS(data, len(datanodes)) +} + +func evaluateHDFS(data []byte, expected int) Result { + var response struct { + Replication *int `json:"configuredReplication"` + JMX struct { + Beans []map[string]json.RawMessage `json:"beans"` + } `json:"jmx"` + } + if err := json.Unmarshal(data, &response); err != nil || response.Replication == nil { + return result("hdfs", Unknown, "invalid HDFS replication/JMX response") + } + fields := make(map[string]json.RawMessage) + for _, bean := range response.JMX.Beans { + name, _ := text(bean, "name") + switch name { + case "Hadoop:service=NameNode,name=FSNamesystem", "Hadoop:service=NameNode,name=FSNamesystemState", "Hadoop:service=NameNode,name=NameNodeInfo": + for key, value := range bean { + fields[key] = value + } + } + } + counts, err := numbers(fields, "UnderReplicatedBlocks", "MissingBlocks", "CorruptBlocks", "PendingReplicationBlocks", "NumLiveDataNodes") + if err != nil { + return result("hdfs", Unknown, err.Error()) + } + safemode, err := text(fields, "Safemode") + if err != nil { + return result("hdfs", Unknown, err.Error()) + } + var problems []string + if *response.Replication < minReplicas { + problems = append(problems, fmt.Sprintf("configured block replication is %d; at least %d required", *response.Replication, minReplicas)) + } + if safemode != "" { + problems = append(problems, "NameNode is in safe mode") + } + if counts["NumLiveDataNodes"] != int64(expected) { + problems = append(problems, fmt.Sprintf("%d/%d expected DataNodes are live", counts["NumLiveDataNodes"], expected)) + } + for _, key := range []string{"UnderReplicatedBlocks", "MissingBlocks", "CorruptBlocks", "PendingReplicationBlocks"} { + if counts[key] != 0 { + problems = append(problems, fmt.Sprintf("%s=%d", key, counts[key])) + } + } + if len(problems) > 0 { + return Result{Component: "hdfs", Status: Degraded, Messages: problems} + } + return result("hdfs", Healthy, fmt.Sprintf("%d DataNodes live; no missing, corrupt, under-replicated or pending-replication blocks", expected)) +} diff --git a/internal/orchestration/replication/json.go b/internal/orchestration/replication/json.go new file mode 100644 index 0000000..ee6b998 --- /dev/null +++ b/internal/orchestration/replication/json.go @@ -0,0 +1,54 @@ +package replication + +import ( + "encoding/json" + "fmt" + "strconv" +) + +func number(fields map[string]json.RawMessage, key string) (int64, error) { + raw, found := fields[key] + if !found || string(raw) == "null" { + return 0, fmt.Errorf("missing numeric field %s", key) + } + var value int64 + if err := json.Unmarshal(raw, &value); err != nil { + var quoted string + if err := json.Unmarshal(raw, "ed); err != nil { + return 0, fmt.Errorf("invalid integer field %s", key) + } + var parseErr error + value, parseErr = strconv.ParseInt(quoted, 10, 64) + if parseErr != nil { + return 0, fmt.Errorf("invalid integer field %s", key) + } + } + if value < 0 { + return 0, fmt.Errorf("invalid nonnegative integer field %s", key) + } + return value, nil +} + +func text(fields map[string]json.RawMessage, key string) (string, error) { + raw, found := fields[key] + if !found || string(raw) == "null" { + return "", fmt.Errorf("missing string field %s", key) + } + var value string + if err := json.Unmarshal(raw, &value); err != nil { + return "", fmt.Errorf("invalid string field %s", key) + } + return value, nil +} + +func numbers(fields map[string]json.RawMessage, keys ...string) (map[string]int64, error) { + values := make(map[string]int64, len(keys)) + for _, key := range keys { + value, err := number(fields, key) + if err != nil { + return nil, err + } + values[key] = value + } + return values, nil +} diff --git a/internal/orchestration/replication/kafka.go b/internal/orchestration/replication/kafka.go new file mode 100644 index 0000000..6030d3c --- /dev/null +++ b/internal/orchestration/replication/kafka.go @@ -0,0 +1,137 @@ +package replication + +import ( + "context" + "fmt" + "regexp" + "strconv" + "strings" +) + +var ( + kafkaTopicPattern = regexp.MustCompile(`\bTopic:\s*(\S+)`) + kafkaPartitionPattern = regexp.MustCompile(`\bPartition:\s*(\d+)\b`) + kafkaCountPattern = regexp.MustCompile(`\bPartitionCount:\s*(\d+)\b`) + kafkaFieldsPattern = regexp.MustCompile(`\b(Leader|Replicas|Isr):\s*([-\d,]+)`) +) + +func (c *Checker) checkKafka(ctx context.Context, inventory inventory) Result { + members, err := inventory.members("kafka") + if err != nil { + return result("kafka", Unknown, err.Error()) + } + if err := expectedMembers(members); err != nil { + return result("kafka", Degraded, err.Error()) + } + command := []string{"kafka-topics.sh", "--bootstrap-server", c.options.KafkaBootstrapServer, "--describe"} + if c.options.KafkaClientProperties != "" { + command = append(command, "--command-config", c.options.KafkaClientProperties) + } + data, err := c.query(ctx, members[0].pod.Name, "kafka", command) + if err != nil { + return result("kafka", Unknown, err.Error()) + } + return evaluateKafka(string(data)) +} + +func brokerIDs(value string) (map[int]bool, error) { + if value == "" { + return nil, fmt.Errorf("missing broker IDs") + } + ids := make(map[int]bool) + for _, part := range strings.Split(value, ",") { + id, err := strconv.Atoi(part) + if err != nil || id < 0 || ids[id] { + return nil, fmt.Errorf("invalid or duplicate broker ID") + } + ids[id] = true + } + return ids, nil +} + +func partitionProblem(line string) string { + fields := make(map[string]string) + for _, match := range kafkaFieldsPattern.FindAllStringSubmatch(line, -1) { + fields[match[1]] = match[2] + } + replicas, err := brokerIDs(fields["Replicas"]) + if err != nil || len(replicas) < minReplicas { + return "fewer than two distinct assigned replicas or invalid assignment" + } + isr, err := brokerIDs(fields["Isr"]) + if err != nil || len(isr) != len(replicas) { + return "assigned replicas are not all in sync" + } + for id := range replicas { + if !isr[id] { + return "assigned replicas are not all in sync" + } + } + leader, err := strconv.Atoi(fields["Leader"]) + if err != nil || !isr[leader] { + return "no available in-sync leader" + } + return "" +} + +func evaluateKafka(output string) Result { + counts := make(map[string]int) + partitions := make(map[string]map[int]bool) + var problems []string + for _, line := range strings.Split(output, "\n") { + topicMatch := kafkaTopicPattern.FindStringSubmatch(line) + if topicMatch == nil { + continue + } + topic := topicMatch[1] + if count := kafkaCountPattern.FindStringSubmatch(line); count != nil { + value, err := strconv.Atoi(count[1]) + if err != nil || value < 1 || counts[topic] != 0 { + return result("kafka", Unknown, "invalid or duplicate topic summary") + } + counts[topic] = value + } + if partition := kafkaPartitionPattern.FindStringSubmatch(line); partition != nil { + id, err := strconv.Atoi(partition[1]) + if err != nil || partitions[topic][id] { + return result("kafka", Unknown, "invalid or duplicate partition description") + } + if partitions[topic] == nil { + partitions[topic] = make(map[int]bool) + } + partitions[topic][id] = true + if problem := partitionProblem(line); problem != "" { + problems = append(problems, fmt.Sprintf("%s partition %d: %s", topic, id, problem)) + } + } + } + if err := completeKafkaEvidence(counts, partitions); err != nil { + return result("kafka", Unknown, err.Error()) + } + if len(problems) > 0 { + return Result{Component: "kafka", Status: Degraded, Messages: problems} + } + return result("kafka", Healthy, fmt.Sprintf("all partitions of %d topics have at least two replicas, complete ISR and an in-sync leader", len(counts))) +} + +func completeKafkaEvidence(counts map[string]int, partitions map[string]map[int]bool) error { + for _, topic := range []string{"__consumer_offsets", "__transaction_state"} { + if counts[topic] == 0 { + return fmt.Errorf("required internal topic %s is absent; initialize its workload and repeat the check", topic) + } + } + if len(counts) != len(partitions) { + return fmt.Errorf("topic summaries and partition descriptions do not match") + } + for topic, count := range counts { + if len(partitions[topic]) != count { + return fmt.Errorf("incomplete partition descriptions for %s", topic) + } + for id := 0; id < count; id++ { + if !partitions[topic][id] { + return fmt.Errorf("missing partition %d for %s", id, topic) + } + } + } + return nil +} diff --git a/internal/orchestration/replication/report.go b/internal/orchestration/replication/report.go new file mode 100644 index 0000000..ffba7ee --- /dev/null +++ b/internal/orchestration/replication/report.go @@ -0,0 +1,90 @@ +// Package replication checks the observed replication state of chart-managed databases. +// It does not grant permission to disrupt a node or modify database state. +package replication + +import ( + "fmt" + "slices" + "time" +) + +const ( + Healthy = "healthy" + Degraded = "degraded" + Unknown = "unknown" + + minReplicas = 2 +) + +var supportedComponents = []string{"hdfs", "elasticsearch", "kafka", "clickhouse"} + +// Options identifies the installation and bounds each database query. +type Options struct { + Namespace string + Release string + Components []string + RequestTimeout time.Duration + KafkaClientProperties string + KafkaBootstrapServer string + ElasticsearchScheme string + ElasticsearchCA string + ElasticsearchHost string +} + +// Validate rejects ambiguous scope and unsupported probe settings. +func (o Options) Validate() error { + if o.Namespace == "" || o.Release == "" { + return fmt.Errorf("namespace and release are required") + } + if o.RequestTimeout <= 0 { + return fmt.Errorf("request-timeout must be positive") + } + if o.ElasticsearchScheme != "http" && o.ElasticsearchScheme != "https" { + return fmt.Errorf("elasticsearch-scheme must be http or https") + } + if len(o.Components) == 0 { + return fmt.Errorf("select at least one component") + } + seen := make(map[string]bool) + for _, component := range o.Components { + if !slices.Contains(supportedComponents, component) || seen[component] { + return fmt.Errorf("invalid or duplicate component %q; choose from %v", component, supportedComponents) + } + seen[component] = true + } + return nil +} + +// Result reports one component; missing evidence is unknown, never healthy. +type Result struct { + Component string `json:"component"` + Status string `json:"status"` + Messages []string `json:"messages"` +} + +// Report is a point-in-time observation, not a maintenance lock. +type Report struct { + CheckedAt time.Time `json:"checkedAt"` + Namespace string `json:"namespace"` + Release string `json:"release"` + Status string `json:"status"` + Error string `json:"error,omitempty"` + Checks []Result `json:"checks"` +} + +func result(component, status, message string) Result { + return Result{Component: component, Status: status, Messages: []string{message}} +} + +func reportStatus(checks []Result) string { + status := Healthy + for _, check := range checks { + if check.Status == Unknown { + return Unknown + } + if check.Status != Healthy { + status = Degraded + } + } + return status +} diff --git a/internal/orchestration/replication/wait.go b/internal/orchestration/replication/wait.go new file mode 100644 index 0000000..a830bf8 --- /dev/null +++ b/internal/orchestration/replication/wait.go @@ -0,0 +1,46 @@ +package replication + +import ( + "context" + "fmt" + "time" +) + +// Wait polls until replication remains healthy for stableFor or the caller's deadline expires. +// The observer receives every report and can write progress to stderr. +func Wait(ctx context.Context, check func(context.Context) Report, interval, stableFor time.Duration, observe func(Report)) (Report, error) { + if interval <= 0 || stableFor < 0 { + return Report{}, fmt.Errorf("interval must be positive and stable-for cannot be negative") + } + var healthySince time.Time + var report Report + for { + if err := ctx.Err(); err != nil { + return report, fmt.Errorf("replication wait ended: %w", err) + } + report = check(ctx) + if observe != nil { + observe(report) + } + if err := ctx.Err(); err != nil { + return report, fmt.Errorf("replication wait ended: %w", err) + } + if report.Status != Healthy { + healthySince = time.Time{} + } else { + if healthySince.IsZero() { + healthySince = time.Now() + } + if time.Since(healthySince) >= stableFor { + return report, nil + } + } + timer := time.NewTimer(interval) + select { + case <-ctx.Done(): + timer.Stop() + return report, fmt.Errorf("replication wait ended: %w", ctx.Err()) + case <-timer.C: + } + } +} diff --git a/internal/orchestration/replication/wait_test.go b/internal/orchestration/replication/wait_test.go new file mode 100644 index 0000000..87b01e9 --- /dev/null +++ b/internal/orchestration/replication/wait_test.go @@ -0,0 +1,59 @@ +package replication + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestWaitContinuesThroughUnknownAndDegraded(t *testing.T) { + statuses := []string{Unknown, Degraded, Healthy} + calls := 0 + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + report, err := Wait(ctx, func(context.Context) Report { + status := statuses[calls] + calls++ + return Report{Status: status} + }, time.Millisecond, 0, nil) + require.NoError(t, err) + assert.Equal(t, Healthy, report.Status) + assert.Equal(t, 3, calls) +} + +func TestWaitHonorsDeadlineDuringQuery(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Millisecond) + defer cancel() + _, err := Wait(ctx, func(ctx context.Context) Report { + <-ctx.Done() + return Report{Status: Healthy} + }, time.Millisecond, 0, nil) + require.ErrorIs(t, err, context.DeadlineExceeded) +} + +func TestWaitRequiresSustainedHealthyObservations(t *testing.T) { + calls := 0 + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + _, err := Wait(ctx, func(context.Context) Report { + calls++ + if calls == 2 { + return Report{Status: Degraded} + } + return Report{Status: Healthy} + }, time.Millisecond, 5*time.Millisecond, nil) + require.NoError(t, err) + assert.GreaterOrEqual(t, calls, 4) +} + +func TestWaitCancellation(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + _, err := Wait(ctx, func(context.Context) Report { + cancel() + return Report{Status: Unknown} + }, time.Hour, 0, nil) + require.ErrorIs(t, err, context.Canceled) +} From fad80b6fbed7bb0de58cdac9172ee309bdb1a6a7 Mon Sep 17 00:00:00 2001 From: Vladimir Iliakov Date: Thu, 10 Sep 2026 16:41:22 +0200 Subject: [PATCH 2/4] Discover replication targets by namespace and fix in-pod queries Use chart name and component labels instead of requiring a Helm release. Distinguish the HDFS NameNode from its SecondaryNameNode even though both use a namenode container. Clear the inherited broker JMX settings for Kafka CLI invocations while preserving authentication options, and alias the ClickHouse queue subquery. ClickHouse retains last_queue_update_exception after successful queue updates; report it as historical context instead of blocking recovered replicas. Current coordination errors and backlog still fail. Verified against StorageReplicatedMergeTree::queueUpdatingTask in ClickHouse v26.5.1.882-stable. Full Go tests and golangci-lint pass. Read-only validation in stackstate-nightly reports HDFS, Elasticsearch and ClickHouse healthy; Kafka correctly flags three perf-test topics with one assigned replica. Added namespace/component discovery, SecondaryNameNode, JMX isolation and recovered-exception regression coverage. --- ARCHITECTURE.md | 2 +- README.md | 6 +- cmd/replication/replication.go | 2 - cmd/replication/replication_test.go | 4 +- docs/replication.md | 46 +++++++----- examples/replication/job.yaml | 1 - internal/orchestration/replication/checker.go | 4 +- .../orchestration/replication/checker_test.go | 74 +++++++++++++++---- .../orchestration/replication/clickhouse.go | 31 +++++--- .../orchestration/replication/discovery.go | 15 ++-- .../replication/elasticsearch.go | 2 +- .../replication/evaluators_test.go | 11 ++- internal/orchestration/replication/hdfs.go | 4 +- internal/orchestration/replication/kafka.go | 8 +- .../orchestration/replication/kafka_test.go | 40 ++++++++++ internal/orchestration/replication/report.go | 6 +- 16 files changed, 188 insertions(+), 68 deletions(-) create mode 100644 internal/orchestration/replication/kafka_test.go diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 34cb71b..15a4a72 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -127,7 +127,7 @@ appCtx.NewCHClient(backupAPIPort, dbPort) // ClickHouse client factory **Key Packages**: - `portforward/`: Manages Kubernetes port-forwarding lifecycle -- `replication/`: Discovers release workloads and evaluates fixed database queries, without backup configuration or restore operations +- `replication/`: Discovers namespace workloads and evaluates fixed database queries, without backup configuration or restore operations - `scale/`: Deployment and StatefulSet scaling workflows with detailed logging - `restore/`: Restore job orchestration (confirmation, job lifecycle, finalization, resource management) - `restorelock/`: Prevents parallel restore operations using Kubernetes annotations diff --git a/README.md b/README.md index 09a71b8..ab926c5 100644 --- a/README.md +++ b/README.md @@ -47,11 +47,11 @@ sts-backup [command] [subcommand] [flags] ### replication check -Inspect HDFS, Elasticsearch, Kafka and ClickHouse replication for a Helm release: +Inspect HDFS, Elasticsearch, Kafka and ClickHouse replication in one namespace: ```bash -sts-backup replication check --namespace observability --release suse-observability -sts-backup replication check --namespace observability --release suse-observability --wait --output json +sts-backup replication check --namespace observability +sts-backup replication check --namespace observability --wait --output json ``` This command has its own flags and does not require backup configuration. diff --git a/cmd/replication/replication.go b/cmd/replication/replication.go index 6e81311..e222fdd 100644 --- a/cmd/replication/replication.go +++ b/cmd/replication/replication.go @@ -50,7 +50,6 @@ func Cmd() *cobra.Command { RunE: func(command *cobra.Command, _ []string) error { return run(command, f) }, } check.Flags().StringVarP(&f.options.Namespace, "namespace", "n", "", "Kubernetes namespace (required)") - check.Flags().StringVar(&f.options.Release, "release", "", "Helm release name (required)") check.Flags().StringVar(&f.kubeconfig, "kubeconfig", "", "Kubeconfig path; uses normal kubeconfig or in-cluster credentials") check.Flags().StringSliceVar(&f.options.Components, "components", []string{"hdfs", "elasticsearch", "kafka", "clickhouse"}, "Components to check") check.Flags().StringVarP(&f.output, "output", "o", "table", "Output format: table or json") @@ -65,7 +64,6 @@ func Cmd() *cobra.Command { check.Flags().StringVar(&f.options.ElasticsearchCA, "elasticsearch-ca", "", "CA file already mounted in Elasticsearch pods") check.Flags().StringVar(&f.options.ElasticsearchHost, "elasticsearch-server-name", "127.0.0.1", "Elasticsearch TLS server name, resolved to loopback inside the pod") _ = check.MarkFlagRequired("namespace") - _ = check.MarkFlagRequired("release") command.AddCommand(check) return command } diff --git a/cmd/replication/replication_test.go b/cmd/replication/replication_test.go index 89787b0..a7d4fd2 100644 --- a/cmd/replication/replication_test.go +++ b/cmd/replication/replication_test.go @@ -30,7 +30,7 @@ func TestJSONOutputIsSeparateFromProgress(t *testing.T) { func TestCheckValidatesBeforeConnecting(t *testing.T) { command := Cmd() - command.SetArgs([]string{"check", "--namespace=test", "--release=test", "--components=kafka", "--output=invalid"}) + command.SetArgs([]string{"check", "--namespace=test", "--components=kafka", "--output=invalid"}) command.SetOut(&bytes.Buffer{}) command.SetErr(&bytes.Buffer{}) err := command.Execute() @@ -43,7 +43,7 @@ func TestHelpDoesNotRequireBackupConfiguration(t *testing.T) { var output bytes.Buffer command.SetOut(&output) require.NoError(t, command.Execute()) - assert.Contains(t, output.String(), "--release") + assert.NotContains(t, output.String(), "--release") assert.NotContains(t, output.String(), "--secret") assert.NotContains(t, output.String(), "--configmap") } diff --git a/docs/replication.md b/docs/replication.md index d8cd183..96acee8 100644 --- a/docs/replication.md +++ b/docs/replication.md @@ -1,11 +1,12 @@ # Check database replication -`sts-backup replication check` inspects the chart-managed HA databases in one -SUSE Observability Helm release. It works from a workstation or a Kubernetes -Job and does not require the backup ConfigMap, backup Secret, or enabled backups. +`sts-backup replication check` inspects the chart-managed HA databases in a +namespace containing one SUSE Observability installation. It works from a +workstation or a Kubernetes Job and does not require the backup ConfigMap, +backup Secret, or enabled backups. ```bash -sts-backup replication check --namespace observability --release suse-observability +sts-backup replication check --namespace observability ``` The command returns exit code **0** only when every selected component reports @@ -18,11 +19,11 @@ Use JSON for automation: ```bash sts-backup replication check \ - --namespace observability --release suse-observability \ + --namespace observability \ --output json ``` -The report identifies the namespace, release, observation time, selected +The report identifies the namespace, observation time, selected components, status (`healthy`, `degraded` or `unknown`) and diagnostic messages. The status describes only the selected checks. @@ -30,7 +31,7 @@ The status describes only the selected checks. ```bash sts-backup replication check \ - --namespace observability --release suse-observability \ + --namespace observability \ --wait --timeout 15m --interval 10s --stable-for 30s \ --output json > replication.json ``` @@ -45,7 +46,7 @@ Select an explicit subset if a database is intentionally disabled: ```bash sts-backup replication check \ - --namespace observability --release suse-observability \ + --namespace observability \ --components hdfs,elasticsearch,kafka ``` @@ -53,10 +54,14 @@ This selection does not validate the omitted database. ## What is checked -The checker discovers StatefulSets by the Helm release's -`app.kubernetes.io/instance` label and identifies the database containers used -by the product chart. It verifies the desired pods exist, belong to those -StatefulSets, are Ready, and are not terminating or undergoing a rollout. +The checker discovers StatefulSets using `app.kubernetes.io/name` +(`hbase`, `elasticsearch`, `kafka`, `clickhouse`) and +`app.kubernetes.io/component` within the namespace. The HDFS components +`hdfs-nn` and `hdfs-dn` distinguish the NameNode and DataNodes from the +SecondaryNameNode. It assumes one SUSE Observability installation per +namespace; no Helm release name is required. It verifies the desired pods +exist, belong to those StatefulSets, are Ready, and are not terminating or +undergoing a rollout. It queries database state and invalidates the observation if Kubernetes resource versions change during those queries. @@ -65,13 +70,17 @@ resource versions change during those queries. | HDFS | Configured default block replication is at least two; all expected DataNodes are live; the NameNode is out of safe mode; no missing, corrupt, under-replicated or pending-replication blocks are reported by JMX. | | Elasticsearch | Expected members are present; health is green; every returned index has at least one replica shard; no shards are unassigned, initializing or relocating. | | Kafka | Every described partition has at least two distinct assigned replicas, complete ISR membership and an in-sync leader. Summaries and partition descriptions must agree. Both `__consumer_offsets` and `__transaction_state` must exist. | -| ClickHouse | Every discovered member is queried. Replicated table groups have their expected active replicas, live coordination sessions and no read-only members. Replication logs are caught up, and no replication queue tasks other than background `MERGE_PARTS` remain. Missing or duplicate table replicas and query exceptions fail the check. | +| ClickHouse | Every discovered member is queried. Replicated table groups have their expected active replicas, live coordination sessions and no read-only members. Replication logs are caught up, and no replication queue tasks other than background `MERGE_PARTS` remain. Missing or duplicate table replicas and current query exceptions fail the check. | The Kafka check does not create missing internal topics: initialize the corresponding workloads and repeat the check. The ClickHouse check requires replicated-table evidence and does not classify an empty result as healthy. Unreplicated ClickHouse tables are outside its scope. +ClickHouse can retain `last_queue_update_exception` after recovery. The checker +reports that history without failing otherwise healthy replication. Current +coordination-query errors, expired sessions and replication backlog still fail. + ## Access and supported layouts The command uses the normal kubeconfig selection or a Kubernetes service @@ -87,16 +96,17 @@ operations. No backup credentials are loaded. Queries use the database tools already present in the product containers: `hdfs` and `curl` in the NameNode, `curl` in Elasticsearch, `kafka-topics.sh` in Kafka, and `clickhouse-client` in ClickHouse. -Custom images, container names, external databases and alternative NameNode -topologies are not supported by these initial adapters. Failed discovery or -unsupported response formats produce `unknown`, not success. +Custom images, container names, external databases, alternative NameNode +topologies and overridden database name/component labels are not supported by +these initial adapters. Failed discovery or unsupported response formats +produce `unknown`, not success. The initial HTTP adapters use the chart's NameNode and Elasticsearch ports. For authenticated Kafka, supply a client properties file already mounted in the broker and an appropriate bootstrap address: ```bash -sts-backup replication check -n observability --release suse-observability \ +sts-backup replication check -n observability \ --components kafka \ --kafka-bootstrap-server suse-observability-kafka:9092 \ --kafka-client-properties /mounted/client.properties @@ -120,7 +130,7 @@ treated as unverified. [examples/replication/job.yaml](../examples/replication/job.yaml) contains a dedicated ServiceAccount, namespace-scoped RBAC and a Job using in-cluster -credentials. Adapt its namespace, release and image reference before use. +credentials. Adapt its namespace and image reference before use. The Job has no retries: a failure requires investigation and an explicit rerun. No new container image is published by this change. To package the CLI, build diff --git a/examples/replication/job.yaml b/examples/replication/job.yaml index 203ab17..34997df 100644 --- a/examples/replication/job.yaml +++ b/examples/replication/job.yaml @@ -60,7 +60,6 @@ spec: - replication - check - --namespace=observability - - --release=suse-observability - --wait - --timeout=15m - --stable-for=30s diff --git a/internal/orchestration/replication/checker.go b/internal/orchestration/replication/checker.go index c5884b6..38d7c9a 100644 --- a/internal/orchestration/replication/checker.go +++ b/internal/orchestration/replication/checker.go @@ -12,7 +12,7 @@ type Checker struct { options Options } -// New creates a checker with an explicit release scope. +// New creates a checker for the installation in one namespace. func New(kube Kubernetes, options Options) (*Checker, error) { if err := options.Validate(); err != nil { return nil, err @@ -29,7 +29,7 @@ func New(kube Kubernetes, options Options) (*Checker, error) { // Check queries every selected component and rechecks Kubernetes membership afterward. func (c *Checker) Check(ctx context.Context) Report { report := Report{ - CheckedAt: time.Now().UTC(), Namespace: c.options.Namespace, Release: c.options.Release, + CheckedAt: time.Now().UTC(), Namespace: c.options.Namespace, Checks: make([]Result, 0, len(c.options.Components)), } before, err := c.discover(ctx) diff --git a/internal/orchestration/replication/checker_test.go b/internal/orchestration/replication/checker_test.go index b866d72..43b7daa 100644 --- a/internal/orchestration/replication/checker_test.go +++ b/internal/orchestration/replication/checker_test.go @@ -32,28 +32,34 @@ func (f *fakeKubernetes) Exec(ctx context.Context, namespace, pod, container str } func testOptions() Options { - return Options{Namespace: "test", Release: "observability", Components: []string{"kafka"}, RequestTimeout: time.Second, ElasticsearchScheme: "http"} + return Options{Namespace: "test", Components: []string{"kafka"}, RequestTimeout: time.Second, ElasticsearchScheme: "http"} } func kafkaObjects() []runtime.Object { - labels := map[string]string{"app.kubernetes.io/instance": "observability"} + return databaseObjects("kafka", "kafka", "kafka", 2) +} + +func databaseObjects(application, component, container string, replicas int32) []runtime.Object { + labels := map[string]string{ + "app.kubernetes.io/name": application, "app.kubernetes.io/component": component, "app.kubernetes.io/instance": "arbitrary-release", + } workload := &appsv1.StatefulSet{ - ObjectMeta: metav1.ObjectMeta{Name: "kafka", Namespace: "test", UID: "sts-kafka", Generation: 1, ResourceVersion: "1", Labels: labels}, + ObjectMeta: metav1.ObjectMeta{Name: component, Namespace: "test", UID: types.UID("sts-" + component), Generation: 1, ResourceVersion: "1", Labels: labels}, Spec: appsv1.StatefulSetSpec{ - Replicas: ptr.To(int32(2)), - Template: corev1.PodTemplateSpec{Spec: corev1.PodSpec{Containers: []corev1.Container{{Name: "kafka"}}}}, + Replicas: ptr.To(replicas), + Template: corev1.PodTemplateSpec{Spec: corev1.PodSpec{Containers: []corev1.Container{{Name: container}}}}, }, - Status: appsv1.StatefulSetStatus{ReadyReplicas: 2, ObservedGeneration: 1, CurrentRevision: "one", UpdateRevision: "one"}, + Status: appsv1.StatefulSetStatus{ReadyReplicas: replicas, ObservedGeneration: 1, CurrentRevision: "one", UpdateRevision: "one"}, } objects := []runtime.Object{workload} - for n := 0; n < 2; n++ { - name := fmt.Sprintf("kafka-%d", n) + for n := int32(0); n < replicas; n++ { + name := fmt.Sprintf("%s-%d", component, n) pod := &corev1.Pod{ ObjectMeta: metav1.ObjectMeta{ Name: name, Namespace: "test", UID: types.UID(name), ResourceVersion: "1", Labels: labels, - OwnerReferences: []metav1.OwnerReference{{Kind: "StatefulSet", Name: "kafka", UID: workload.UID, Controller: ptr.To(true)}}, + OwnerReferences: []metav1.OwnerReference{{Kind: "StatefulSet", Name: component, UID: workload.UID, Controller: ptr.To(true)}}, }, - Spec: corev1.PodSpec{NodeName: fmt.Sprintf("node-%d", n), Containers: []corev1.Container{{Name: "kafka"}}}, + Spec: corev1.PodSpec{NodeName: fmt.Sprintf("node-%d", n), Containers: []corev1.Container{{Name: container}}}, Status: corev1.PodStatus{Phase: corev1.PodRunning, Conditions: []corev1.PodCondition{ {Type: corev1.PodReady, Status: corev1.ConditionTrue}, }}, @@ -69,7 +75,7 @@ func TestCheckerQueriesReadyPodsAndDoesNotMutateKubernetes(t *testing.T) { assert.Equal(t, "test", namespace) assert.Equal(t, "kafka-0", pod) assert.Equal(t, "kafka", container) - assert.Equal(t, []string{"kafka-topics.sh", "--bootstrap-server", "localhost:9092", "--describe"}, command) + assert.Equal(t, []string{"bash", "-ec", kafkaQuery, "replication-check", "--bootstrap-server", "localhost:9092", "--describe"}, command) _, deadline := ctx.Deadline() assert.True(t, deadline) return []byte(kafkaFixture()), nil @@ -102,8 +108,18 @@ func TestCheckerRejectsIncompleteTopology(t *testing.T) { objects[1].(*corev1.Pod).OwnerReferences[0].UID = "other" return objects }}, - {"wrong release", func(objects []runtime.Object) []runtime.Object { - objects[0].(*appsv1.StatefulSet).Labels = map[string]string{"app.kubernetes.io/instance": "other"} + {"wrong database label", func(objects []runtime.Object) []runtime.Object { + objects[0].(*appsv1.StatefulSet).Labels = map[string]string{"app.kubernetes.io/name": "other", "app.kubernetes.io/component": "kafka"} + return objects + }}, + {"wrong component label", func(objects []runtime.Object) []runtime.Object { + objects[0].(*appsv1.StatefulSet).Labels["app.kubernetes.io/component"] = "unrelated" + return objects + }}, + {"different namespace", func(objects []runtime.Object) []runtime.Object { + for _, object := range objects { + object.(metav1.Object).SetNamespace("other") + } return objects }}, {"rolling update", func(objects []runtime.Object) []runtime.Object { @@ -122,6 +138,38 @@ func TestCheckerRejectsIncompleteTopology(t *testing.T) { } } +func TestHDFSDiscoveryDistinguishesSecondaryNameNode(t *testing.T) { + objects := databaseObjects("hbase", "hdfs-nn", "namenode", 1) + objects = append(objects, databaseObjects("hbase", "hdfs-snn", "namenode", 1)...) + objects = append(objects, databaseObjects("hbase", "hdfs-dn", "datanode", 3)...) + kube := &fakeKubernetes{client: fake.NewSimpleClientset(objects...), exec: func(_ context.Context, namespace, pod, container string, _ []string) ([]byte, error) { + assert.Equal(t, "test", namespace) + assert.Equal(t, "hdfs-nn-0", pod) + assert.Equal(t, "namenode", container) + return hdfsFixture(t, func(map[string]any) {}), nil + }} + options := testOptions() + options.Components = []string{"hdfs"} + probe, err := New(kube, options) + require.NoError(t, err) + report := probe.Check(context.Background()) + assert.Equal(t, Healthy, report.Status, report) + assert.Equal(t, 1, kube.calls) +} + +func TestDiscoveryDoesNotRequireHelmReleaseLabel(t *testing.T) { + objects := kafkaObjects() + for _, object := range objects { + delete(object.(metav1.Object).GetLabels(), "app.kubernetes.io/instance") + } + kube := &fakeKubernetes{client: fake.NewSimpleClientset(objects...), exec: func(context.Context, string, string, string, []string) ([]byte, error) { + return []byte(kafkaFixture()), nil + }} + probe, err := New(kube, testOptions()) + require.NoError(t, err) + assert.Equal(t, Healthy, probe.Check(context.Background()).Status) +} + func TestCheckerRejectsMembershipChangeDuringQueries(t *testing.T) { client := fake.NewSimpleClientset(kafkaObjects()...) kube := &fakeKubernetes{client: client, exec: func(ctx context.Context, namespace, pod, _ string, _ []string) ([]byte, error) { diff --git a/internal/orchestration/replication/clickhouse.go b/internal/orchestration/replication/clickhouse.go index ff76adb..6783f5e 100644 --- a/internal/orchestration/replication/clickhouse.go +++ b/internal/orchestration/replication/clickhouse.go @@ -15,7 +15,7 @@ FROM system.replicas LEFT JOIN ( SELECT database, table, countIf(type != 'MERGE_PARTS') AS pending_data_tasks FROM system.replication_queue GROUP BY database, table -) USING (database, table) +) AS queues USING (database, table) WHERE database NOT IN ('system', 'INFORMATION_SCHEMA', 'information_schema') FORMAT JSON` @@ -24,15 +24,16 @@ exec clickhouse-client --host 127.0.0.1 --port "${CLICKHOUSE_TCP_PORT:-9000}" \ --user "${CLICKHOUSE_ADMIN_USER:?missing ClickHouse user}" --readonly 1 --query "$1"` type replicaObservation struct { - path string - name string - pod string - total int64 - problems []string + path string + name string + pod string + total int64 + historicalQueueErr bool + problems []string } func (c *Checker) checkClickHouse(ctx context.Context, inventory inventory) Result { - members, err := inventory.members("clickhouse") + members, err := inventory.members("clickhouse", "clickhouse", "clickhouse") if err != nil { return result("clickhouse", Unknown, err.Error()) } @@ -91,6 +92,8 @@ func parseReplica(row map[string]json.RawMessage, pod string, expected int) (rep return replicaObservation{}, fmt.Errorf("missing replicated-table identity") } observation := replicaObservation{path: fields["zookeeper_path"], name: fields["replica_name"], pod: pod, total: counts["total_replicas"]} + // ClickHouse retains this exception even after subsequent queue updates succeed. + observation.historicalQueueErr = fields["last_queue_update_exception"] != "" if observation.total < minReplicas || observation.total != int64(expected) || counts["active_replicas"] != observation.total { observation.problems = append(observation.problems, fmt.Sprintf("%d/%d replicas active; %d expected", counts["active_replicas"], observation.total, expected)) } @@ -100,8 +103,8 @@ func parseReplica(row map[string]json.RawMessage, pod string, expected int) (rep if counts["log_pointer"] <= counts["log_max_index"] || counts["pending_data_tasks"] != 0 { observation.problems = append(observation.problems, fmt.Sprintf("replication backlog: %d data tasks; reported delay %ds", counts["pending_data_tasks"], counts["absolute_delay"])) } - if fields["last_queue_update_exception"] != "" || fields["zookeeper_exception"] != "" { - observation.problems = append(observation.problems, "replication queue or coordination query reported an exception") + if fields["zookeeper_exception"] != "" { + observation.problems = append(observation.problems, "coordination query reported an exception") } return observation, nil } @@ -109,7 +112,11 @@ func parseReplica(row map[string]json.RawMessage, pod string, expected int) (rep func evaluateClickHouse(observations []replicaObservation) Result { groups := make(map[string][]replicaObservation) var problems []string + historicalErrors := 0 for _, observation := range observations { + if observation.historicalQueueErr { + historicalErrors++ + } groups[observation.path] = append(groups[observation.path], observation) for _, problem := range observation.problems { problems = append(problems, fmt.Sprintf("%s %s: %s", observation.pod, observation.path, problem)) @@ -135,5 +142,9 @@ func evaluateClickHouse(observations []replicaObservation) Result { sort.Strings(problems) return Result{Component: "clickhouse", Status: Degraded, Messages: problems} } - return result("clickhouse", Healthy, fmt.Sprintf("%d replicated table groups checked on every member; no pending data replication tasks", len(groups))) + report := result("clickhouse", Healthy, fmt.Sprintf("%d replicated table groups checked on every member; no pending data replication tasks", len(groups))) + if historicalErrors > 0 { + report.Messages = append(report.Messages, fmt.Sprintf("%d table replicas retain a previous queue-update exception; current replication checks pass", historicalErrors)) + } + return report } diff --git a/internal/orchestration/replication/discovery.go b/internal/orchestration/replication/discovery.go index 981d49b..15dffdc 100644 --- a/internal/orchestration/replication/discovery.go +++ b/internal/orchestration/replication/discovery.go @@ -9,7 +9,6 @@ import ( appsv1 "k8s.io/api/apps/v1" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "k8s.io/apimachinery/pkg/labels" "k8s.io/client-go/kubernetes" ) @@ -32,14 +31,14 @@ type member struct { func (c *Checker) discover(ctx context.Context) (inventory, error) { ctx, cancel := context.WithTimeout(ctx, c.options.RequestTimeout) defer cancel() - options := metav1.ListOptions{LabelSelector: labels.Set{"app.kubernetes.io/instance": c.options.Release}.String()} + options := metav1.ListOptions{LabelSelector: "app.kubernetes.io/name in (hbase,elasticsearch,kafka,clickhouse)"} sets, err := c.kube.Clientset().AppsV1().StatefulSets(c.options.Namespace).List(ctx, options) if err != nil { - return inventory{}, fmt.Errorf("list release StatefulSets: %w", err) + return inventory{}, fmt.Errorf("list database StatefulSets: %w", err) } pods, err := c.kube.Clientset().CoreV1().Pods(c.options.Namespace).List(ctx, options) if err != nil { - return inventory{}, fmt.Errorf("list release pods: %w", err) + return inventory{}, fmt.Errorf("list database pods: %w", err) } return inventory{workloads: sets.Items, pods: pods.Items}, nil } @@ -48,10 +47,14 @@ func hasContainer(containers []corev1.Container, name string) bool { return slices.ContainsFunc(containers, func(container corev1.Container) bool { return container.Name == name }) } -func (i inventory) members(container string) ([]member, error) { +func (i inventory) members(name, component, container string) ([]member, error) { var members []member found := false for _, workload := range i.workloads { + if workload.Labels["app.kubernetes.io/name"] != name || + (component != "" && workload.Labels["app.kubernetes.io/component"] != component) { + continue + } if !hasContainer(workload.Spec.Template.Spec.Containers, container) { continue } @@ -65,7 +68,7 @@ func (i inventory) members(container string) ([]member, error) { } } if !found { - return nil, fmt.Errorf("no chart-managed StatefulSet with container %q; check release, selected components and chart layout", container) + return nil, fmt.Errorf("no chart-managed StatefulSet for name=%q component=%q with container %q; check namespace, selected components and chart layout", name, component, container) } slices.SortFunc(members, func(a, b member) int { if a.pod.Name < b.pod.Name { diff --git a/internal/orchestration/replication/elasticsearch.go b/internal/orchestration/replication/elasticsearch.go index bcbba56..3cc1d50 100644 --- a/internal/orchestration/replication/elasticsearch.go +++ b/internal/orchestration/replication/elasticsearch.go @@ -20,7 +20,7 @@ fi exec curl "$@" --resolve "${host}:9200:127.0.0.1" "${scheme}://${host}:9200/_cluster/health?level=indices"` func (c *Checker) checkElasticsearch(ctx context.Context, inventory inventory) Result { - members, err := inventory.members("elasticsearch") + members, err := inventory.members("elasticsearch", "", "elasticsearch") if err != nil { return result("elasticsearch", Unknown, err.Error()) } diff --git a/internal/orchestration/replication/evaluators_test.go b/internal/orchestration/replication/evaluators_test.go index e52a39a..54fd6b0 100644 --- a/internal/orchestration/replication/evaluators_test.go +++ b/internal/orchestration/replication/evaluators_test.go @@ -148,6 +148,11 @@ func TestClickHouseReplicaEvidence(t *testing.T) { {"inactive replica", func(m map[string]any) { m["active_replicas"] = "1" }, Degraded}, {"wrong replication", func(m map[string]any) { m["total_replicas"] = "1" }, Degraded}, {"coordination exception", func(m map[string]any) { m["zookeeper_exception"] = "connection failed" }, Degraded}, + {"historical queue exception after recovery", func(m map[string]any) { m["last_queue_update_exception"] = "previous Keeper error" }, Healthy}, + {"queue exception with backlog", func(m map[string]any) { + m["last_queue_update_exception"] = "Keeper error" + m["log_pointer"] = "100" + }, Degraded}, {"missing backlog evidence", func(m map[string]any) { delete(m, "pending_data_tasks") }, Unknown}, } for _, test := range tests { @@ -162,7 +167,11 @@ func TestClickHouseReplicaEvidence(t *testing.T) { require.NoError(t, err) peer := observations[0] peer.pod, peer.name, peer.problems = "pod1", "replica1", nil - assert.Equal(t, test.status, evaluateClickHouse(append(observations, peer)).Status) + report := evaluateClickHouse(append(observations, peer)) + assert.Equal(t, test.status, report.Status) + if test.name == "historical queue exception after recovery" { + assert.Contains(t, strings.Join(report.Messages, " "), "previous queue-update exception") + } }) } } diff --git a/internal/orchestration/replication/hdfs.go b/internal/orchestration/replication/hdfs.go index 6b6285e..1472950 100644 --- a/internal/orchestration/replication/hdfs.go +++ b/internal/orchestration/replication/hdfs.go @@ -14,14 +14,14 @@ curl --fail --silent --show-error --max-time 20 'http://127.0.0.1:50070/jmx' printf '}'` func (c *Checker) checkHDFS(ctx context.Context, inventory inventory) Result { - namenodes, err := inventory.members("namenode") + namenodes, err := inventory.members("hbase", "hdfs-nn", "namenode") if err != nil { return result("hdfs", Unknown, err.Error()) } if len(namenodes) != 1 { return result("hdfs", Unknown, "expected the chart's single NameNode; external or HA NameNode layouts are not supported") } - datanodes, err := inventory.members("datanode") + datanodes, err := inventory.members("hbase", "hdfs-dn", "datanode") if err != nil { return result("hdfs", Unknown, err.Error()) } diff --git a/internal/orchestration/replication/kafka.go b/internal/orchestration/replication/kafka.go index 6030d3c..1880613 100644 --- a/internal/orchestration/replication/kafka.go +++ b/internal/orchestration/replication/kafka.go @@ -8,6 +8,10 @@ import ( "strings" ) +// The CLI must not bind the broker's inherited JMX port. +const kafkaQuery = `unset JMX_PORT KAFKA_JMX_OPTS +exec kafka-topics.sh "$@"` + var ( kafkaTopicPattern = regexp.MustCompile(`\bTopic:\s*(\S+)`) kafkaPartitionPattern = regexp.MustCompile(`\bPartition:\s*(\d+)\b`) @@ -16,14 +20,14 @@ var ( ) func (c *Checker) checkKafka(ctx context.Context, inventory inventory) Result { - members, err := inventory.members("kafka") + members, err := inventory.members("kafka", "kafka", "kafka") if err != nil { return result("kafka", Unknown, err.Error()) } if err := expectedMembers(members); err != nil { return result("kafka", Degraded, err.Error()) } - command := []string{"kafka-topics.sh", "--bootstrap-server", c.options.KafkaBootstrapServer, "--describe"} + command := []string{"bash", "-ec", kafkaQuery, "replication-check", "--bootstrap-server", c.options.KafkaBootstrapServer, "--describe"} if c.options.KafkaClientProperties != "" { command = append(command, "--command-config", c.options.KafkaClientProperties) } diff --git a/internal/orchestration/replication/kafka_test.go b/internal/orchestration/replication/kafka_test.go new file mode 100644 index 0000000..8777445 --- /dev/null +++ b/internal/orchestration/replication/kafka_test.go @@ -0,0 +1,40 @@ +package replication + +import ( + "context" + "os" + "os/exec" + "path/filepath" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestKafkaQueryDoesNotInheritBrokerJMXPort(t *testing.T) { + bash, err := exec.LookPath("bash") + if err != nil { + t.Skip("bash is required to exercise the in-pod query") + } + dir := t.TempDir() + script := `#!/bin/sh +test -z "${JMX_PORT:-}" || exit 1 +test -z "${KAFKA_JMX_OPTS:-}" || exit 1 +printf '%s\n' "$KAFKA_OPTS" "$@" +` + path := filepath.Join(dir, "kafka-topics.sh") + require.NoError(t, os.WriteFile(path, []byte(script), 0o600)) + require.NoError(t, os.Chmod(path, 0o700)) + t.Setenv("PATH", dir+string(os.PathListSeparator)+os.Getenv("PATH")) + t.Setenv("JMX_PORT", "5555") + t.Setenv("KAFKA_JMX_OPTS", "-Dcom.sun.management.jmxremote.port=5555") + t.Setenv("KAFKA_OPTS", "-Djava.security.auth.login.config=/mounted/jaas.conf") + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + command := exec.CommandContext(ctx, bash, "-ec", kafkaQuery, "replication-check", + "--bootstrap-server", "localhost:9092", "--describe", "--command-config", "/mounted/client properties") + output, err := command.CombinedOutput() + require.NoError(t, err, string(output)) + assert.Equal(t, "-Djava.security.auth.login.config=/mounted/jaas.conf\n--bootstrap-server\nlocalhost:9092\n--describe\n--command-config\n/mounted/client properties\n", string(output)) +} diff --git a/internal/orchestration/replication/report.go b/internal/orchestration/replication/report.go index ffba7ee..6d78d7c 100644 --- a/internal/orchestration/replication/report.go +++ b/internal/orchestration/replication/report.go @@ -21,7 +21,6 @@ var supportedComponents = []string{"hdfs", "elasticsearch", "kafka", "clickhouse // Options identifies the installation and bounds each database query. type Options struct { Namespace string - Release string Components []string RequestTimeout time.Duration KafkaClientProperties string @@ -33,8 +32,8 @@ type Options struct { // Validate rejects ambiguous scope and unsupported probe settings. func (o Options) Validate() error { - if o.Namespace == "" || o.Release == "" { - return fmt.Errorf("namespace and release are required") + if o.Namespace == "" { + return fmt.Errorf("namespace is required") } if o.RequestTimeout <= 0 { return fmt.Errorf("request-timeout must be positive") @@ -66,7 +65,6 @@ type Result struct { type Report struct { CheckedAt time.Time `json:"checkedAt"` Namespace string `json:"namespace"` - Release string `json:"release"` Status string `json:"status"` Error string `json:"error,omitempty"` Checks []Result `json:"checks"` From bd2d07151c0eb0b427e9818263bd5397b833112c Mon Sep 17 00:00:00 2001 From: Vladimir Iliakov Date: Fri, 11 Sep 2026 10:24:53 +0200 Subject: [PATCH 3/4] Include ZooKeeper in HA replication checks Discover the namespace's ZooKeeper ensemble using chart labels and query every member with read-only mntr. Require at least three voters, matching configured voting counts, one leader and all expected synchronized followers. Recheck the leader after sampling to detect elections without depending on Kubernetes readiness. Enable the check by default and document the supported chart layout and sampled-maintenance boundary. Cover incomplete membership, lag, elections, unavailable metrics and malformed responses. Full Go tests and golangci-lint pass; nightly ZooKeeper passed a live stable-period wait with three voters. --- README.md | 2 +- cmd/replication/replication.go | 4 +- cmd/replication/replication_test.go | 1 + docs/replication.md | 30 +++- internal/orchestration/replication/checker.go | 2 + .../orchestration/replication/discovery.go | 2 +- internal/orchestration/replication/report.go | 2 +- .../orchestration/replication/zookeeper.go | 169 ++++++++++++++++++ .../replication/zookeeper_test.go | 147 +++++++++++++++ 9 files changed, 349 insertions(+), 10 deletions(-) create mode 100644 internal/orchestration/replication/zookeeper.go create mode 100644 internal/orchestration/replication/zookeeper_test.go diff --git a/README.md b/README.md index ab926c5..fb39ac6 100644 --- a/README.md +++ b/README.md @@ -47,7 +47,7 @@ sts-backup [command] [subcommand] [flags] ### replication check -Inspect HDFS, Elasticsearch, Kafka and ClickHouse replication in one namespace: +Inspect HDFS, Elasticsearch, Kafka, ClickHouse and ZooKeeper replication in one namespace: ```bash sts-backup replication check --namespace observability diff --git a/cmd/replication/replication.go b/cmd/replication/replication.go index e222fdd..47f61ee 100644 --- a/cmd/replication/replication.go +++ b/cmd/replication/replication.go @@ -43,7 +43,7 @@ func Cmd() *cobra.Command { f := &flags{} check := &cobra.Command{ Use: "check", Short: "Check observed replication; return nonzero unless all selected checks pass", - Long: "Check chart-managed HDFS, Elasticsearch, Kafka and ClickHouse replication. " + + Long: "Check chart-managed HDFS, Elasticsearch, Kafka, ClickHouse and ZooKeeper replication. " + "This is a point-in-time observation, not permission to remove a node. " + "Requires pods/exec access; only fixed read-only database queries are executed.", Args: cobra.NoArgs, SilenceUsage: true, @@ -51,7 +51,7 @@ func Cmd() *cobra.Command { } check.Flags().StringVarP(&f.options.Namespace, "namespace", "n", "", "Kubernetes namespace (required)") check.Flags().StringVar(&f.kubeconfig, "kubeconfig", "", "Kubeconfig path; uses normal kubeconfig or in-cluster credentials") - check.Flags().StringSliceVar(&f.options.Components, "components", []string{"hdfs", "elasticsearch", "kafka", "clickhouse"}, "Components to check") + check.Flags().StringSliceVar(&f.options.Components, "components", []string{"hdfs", "elasticsearch", "kafka", "clickhouse", "zookeeper"}, "Components to check") check.Flags().StringVarP(&f.output, "output", "o", "table", "Output format: table or json") check.Flags().BoolVar(&f.wait, "wait", false, "Wait for sustained healthy replication") check.Flags().DurationVar(&f.timeout, "timeout", defaultTimeout, "Overall deadline, including queries") diff --git a/cmd/replication/replication_test.go b/cmd/replication/replication_test.go index a7d4fd2..fa2ade3 100644 --- a/cmd/replication/replication_test.go +++ b/cmd/replication/replication_test.go @@ -46,6 +46,7 @@ func TestHelpDoesNotRequireBackupConfiguration(t *testing.T) { assert.NotContains(t, output.String(), "--release") assert.NotContains(t, output.String(), "--secret") assert.NotContains(t, output.String(), "--configmap") + assert.Contains(t, output.String(), "hdfs,elasticsearch,kafka,clickhouse,zookeeper") } func TestReportAndExitAgree(t *testing.T) { diff --git a/docs/replication.md b/docs/replication.md index 96acee8..ae24ca6 100644 --- a/docs/replication.md +++ b/docs/replication.md @@ -55,7 +55,7 @@ This selection does not validate the omitted database. ## What is checked The checker discovers StatefulSets using `app.kubernetes.io/name` -(`hbase`, `elasticsearch`, `kafka`, `clickhouse`) and +(`hbase`, `elasticsearch`, `kafka`, `clickhouse`, `zookeeper`) and `app.kubernetes.io/component` within the namespace. The HDFS components `hdfs-nn` and `hdfs-dn` distinguish the NameNode and DataNodes from the SecondaryNameNode. It assumes one SUSE Observability installation per @@ -71,6 +71,7 @@ resource versions change during those queries. | Elasticsearch | Expected members are present; health is green; every returned index has at least one replica shard; no shards are unassigned, initializing or relocating. | | Kafka | Every described partition has at least two distinct assigned replicas, complete ISR membership and an in-sync leader. Summaries and partition descriptions must agree. Both `__consumer_offsets` and `__transaction_state` must exist. | | ClickHouse | Every discovered member is queried. Replicated table groups have their expected active replicas, live coordination sessions and no read-only members. Replication logs are caught up, and no replication queue tasks other than background `MERGE_PARTS` remain. Missing or duplicate table replicas and current query exceptions fail the check. | +| ZooKeeper | At least three voting members are available. Every member reports the expected voting membership; exactly one is leader and the rest are followers. The leader reports all expected followers synchronized and is checked again after sampling the ensemble. | The Kafka check does not create missing internal topics: initialize the corresponding workloads and repeat the check. The ClickHouse check requires @@ -81,6 +82,17 @@ ClickHouse can retain `last_queue_update_exception` after recovery. The checker reports that history without failing otherwise healthy replication. Current coordination-query errors, expired sessions and replication backlog still fail. +ZooKeeper is checked by default. To inspect it alone: + +```bash +sts-backup replication check -n observability --components zookeeper --wait +``` + +ZooKeeper's `ruok` response does not prove that the ensemble has recovered. +The checker reads `mntr` from every member instead, and requires full voting +membership to recover rather than accepting a surviving majority. Wait mode +applies the same healthy observation period as for the other databases. + ## Access and supported layouts The command uses the normal kubeconfig selection or a Kubernetes service @@ -95,13 +107,21 @@ operations. No backup credentials are loaded. Queries use the database tools already present in the product containers: `hdfs` and `curl` in the NameNode, `curl` in Elasticsearch, -`kafka-topics.sh` in Kafka, and `clickhouse-client` in ClickHouse. +`kafka-topics.sh` in Kafka, `clickhouse-client` in ClickHouse, and Bash TCP +access to ZooKeeper. Custom images, container names, external databases, alternative NameNode topologies and overridden database name/component labels are not supported by these initial adapters. Failed discovery or unsupported response formats produce `unknown`, not success. The initial HTTP adapters use the chart's NameNode and Elasticsearch ports. +The ZooKeeper adapter requires the chart's plaintext loopback client port +and `mntr` in its four-letter-command whitelist. It supports the chart's +single ensemble of voting participants; external ensembles, observer layouts, +weighted quorums and TLS-only client listeners are outside its scope. Missing +membership or synchronization metrics produce `unknown`. No configuration +changes, HTTP AdminServer or database writes are needed. + For authenticated Kafka, supply a client properties file already mounted in the broker and an appropriate bootstrap address: @@ -155,9 +175,9 @@ starting maintenance, provide an atomic database snapshot, or prove continued health between observations. This first version does not validate Longhorn volume health or placement, -spare capacity, HBase region assignment and WAL recovery, ZooKeeper quorum, -VictoriaMetrics redundancy, backup freshness, or the consequences of removing -a particular node. HDFS's default replication setting does not prove that +spare capacity, HBase region assignment and WAL recovery, quorum survival +after a specific node is removed, VictoriaMetrics redundancy, backup freshness, +or the consequences of removing a particular node. HDFS's default replication setting does not prove that every file has the same replication policy. It is not a complete implementation of the product's node-maintenance checklist. diff --git a/internal/orchestration/replication/checker.go b/internal/orchestration/replication/checker.go index 38d7c9a..7f8cdf4 100644 --- a/internal/orchestration/replication/checker.go +++ b/internal/orchestration/replication/checker.go @@ -73,6 +73,8 @@ func (c *Checker) checkComponent(ctx context.Context, inventory inventory, compo return c.checkKafka(ctx, inventory) case "clickhouse": return c.checkClickHouse(ctx, inventory) + case "zookeeper": + return c.checkZooKeeper(ctx, inventory) default: return result(component, Unknown, "unsupported component") } diff --git a/internal/orchestration/replication/discovery.go b/internal/orchestration/replication/discovery.go index 15dffdc..4b04b3c 100644 --- a/internal/orchestration/replication/discovery.go +++ b/internal/orchestration/replication/discovery.go @@ -31,7 +31,7 @@ type member struct { func (c *Checker) discover(ctx context.Context) (inventory, error) { ctx, cancel := context.WithTimeout(ctx, c.options.RequestTimeout) defer cancel() - options := metav1.ListOptions{LabelSelector: "app.kubernetes.io/name in (hbase,elasticsearch,kafka,clickhouse)"} + options := metav1.ListOptions{LabelSelector: "app.kubernetes.io/name in (hbase,elasticsearch,kafka,clickhouse,zookeeper)"} sets, err := c.kube.Clientset().AppsV1().StatefulSets(c.options.Namespace).List(ctx, options) if err != nil { return inventory{}, fmt.Errorf("list database StatefulSets: %w", err) diff --git a/internal/orchestration/replication/report.go b/internal/orchestration/replication/report.go index 6d78d7c..b3aad2d 100644 --- a/internal/orchestration/replication/report.go +++ b/internal/orchestration/replication/report.go @@ -16,7 +16,7 @@ const ( minReplicas = 2 ) -var supportedComponents = []string{"hdfs", "elasticsearch", "kafka", "clickhouse"} +var supportedComponents = []string{"hdfs", "elasticsearch", "kafka", "clickhouse", "zookeeper"} // Options identifies the installation and bounds each database query. type Options struct { diff --git a/internal/orchestration/replication/zookeeper.go b/internal/orchestration/replication/zookeeper.go new file mode 100644 index 0000000..f04478d --- /dev/null +++ b/internal/orchestration/replication/zookeeper.go @@ -0,0 +1,169 @@ +package replication + +import ( + "context" + "fmt" + "strconv" + "strings" +) + +const ( + minZooKeeperVoters = 3 + zooKeeperLeader = "leader" + zooKeeperFollower = "follower" +) + +const zookeeperQuery = `exec 3<>/dev/tcp/127.0.0.1/2181 +printf mntr >&3 +while true; do + if IFS= read -r -t 5 -u 3 line; then + printf '%s\n' "$line" + else + status=$? + if [ "$status" -eq 1 ] && [ -z "$line" ]; then + break + fi + exit 1 + fi +done` + +type zooKeeperObservation struct { + pod string + role string + peerState string + voters int + synced int +} + +func (c *Checker) checkZooKeeper(ctx context.Context, inventory inventory) Result { + members, err := inventory.members("zookeeper", "zookeeper", "zookeeper") + if err != nil { + return result("zookeeper", Unknown, err.Error()) + } + if len(members) < minZooKeeperVoters { + return result("zookeeper", Degraded, "at least three voting members are required for fault-tolerant ZooKeeper") + } + var observations []zooKeeperObservation + for _, member := range members { + if member.replicas != len(members) { + return result("zookeeper", Unknown, "expected one chart-managed ZooKeeper ensemble") + } + observation, err := c.queryZooKeeper(ctx, member.pod.Name) + if err != nil { + return result("zookeeper", Unknown, err.Error()) + } + observations = append(observations, observation) + } + report := evaluateZooKeeper(observations) + if report.Status != Healthy { + return report + } + // Recheck the leader after sampling followers; elections do not change Pod status. + for n, observation := range observations { + if observation.role != zooKeeperLeader { + continue + } + current, err := c.queryZooKeeper(ctx, observation.pod) + if err != nil { + return result("zookeeper", Unknown, err.Error()) + } + if current.role != zooKeeperLeader { + return result("zookeeper", Unknown, "ZooKeeper leadership changed during the checks; repeat the observation") + } + observations[n] = current + } + return evaluateZooKeeper(observations) +} + +func (c *Checker) queryZooKeeper(ctx context.Context, pod string) (zooKeeperObservation, error) { + data, err := c.query(ctx, pod, "zookeeper", []string{"bash", "-ec", zookeeperQuery}) + if err != nil { + return zooKeeperObservation{}, err + } + observation, err := parseZooKeeper(string(data), pod) + if err != nil { + return zooKeeperObservation{}, fmt.Errorf("%s: %w", pod, err) + } + return observation, nil +} + +func parseZooKeeper(data, pod string) (zooKeeperObservation, error) { + fields := make(map[string]string) + for _, line := range strings.Split(data, "\n") { + if strings.TrimSpace(line) == "" { + continue + } + key, value, ok := strings.Cut(line, "\t") + _, duplicate := fields[key] + if !ok || key == "" || duplicate { + return zooKeeperObservation{}, fmt.Errorf("invalid or unavailable ZooKeeper mntr response") + } + fields[key] = strings.TrimSpace(value) + } + if fields["zk_server_state"] == "" { + return zooKeeperObservation{}, fmt.Errorf("missing zk_server_state in mntr response") + } + observation := zooKeeperObservation{pod: pod, role: fields["zk_server_state"], peerState: fields["zk_peer_state"]} + voters, err := zooKeeperNumber(fields, "zk_quorum_size") + if err != nil { + return zooKeeperObservation{}, err + } + observation.voters = voters + if observation.role == zooKeeperLeader { + observation.synced, err = zooKeeperNumber(fields, "zk_synced_followers") + if err != nil { + return zooKeeperObservation{}, err + } + } + return observation, nil +} + +func zooKeeperNumber(fields map[string]string, key string) (int, error) { + value, err := strconv.Atoi(fields[key]) + if err != nil || value < 0 { + return 0, fmt.Errorf("missing or invalid %s in mntr response", key) + } + return value, nil +} + +func evaluateZooKeeper(observations []zooKeeperObservation) Result { + if len(observations) < minZooKeeperVoters { + return result("zookeeper", Degraded, "at least three voting members are required for fault-tolerant ZooKeeper") + } + leaders := 0 + seen := make(map[string]bool) + var problems []string + for _, observation := range observations { + if observation.pod == "" || seen[observation.pod] { + return result("zookeeper", Unknown, "missing or duplicate ZooKeeper member evidence") + } + seen[observation.pod] = true + if observation.voters != len(observations) { + return result("zookeeper", Unknown, fmt.Sprintf("%s reports %d configured voters; %d members discovered", observation.pod, observation.voters, len(observations))) + } + switch observation.role { + case zooKeeperLeader: + leaders++ + if observation.synced != len(observations)-1 { + problems = append(problems, fmt.Sprintf("%s reports %d/%d synchronized followers", observation.pod, observation.synced, len(observations)-1)) + } + case zooKeeperFollower: + default: + problems = append(problems, fmt.Sprintf("%s is %s; expected a voting leader or follower", observation.pod, observation.role)) + } + expectedPeerState := "following - broadcast" + if observation.role == zooKeeperLeader { + expectedPeerState = "leading - broadcast" + } + if observation.peerState != "" && observation.peerState != expectedPeerState { + problems = append(problems, fmt.Sprintf("%s is not in broadcast state: %s", observation.pod, observation.peerState)) + } + } + if leaders != 1 { + problems = append(problems, fmt.Sprintf("expected one ZooKeeper leader; observed %d", leaders)) + } + if len(problems) > 0 { + return Result{Component: "zookeeper", Status: Degraded, Messages: problems} + } + return result("zookeeper", Healthy, fmt.Sprintf("%d voting members available; one leader and %d synchronized followers", len(observations), len(observations)-1)) +} diff --git a/internal/orchestration/replication/zookeeper_test.go b/internal/orchestration/replication/zookeeper_test.go new file mode 100644 index 0000000..dfce640 --- /dev/null +++ b/internal/orchestration/replication/zookeeper_test.go @@ -0,0 +1,147 @@ +package replication + +import ( + "context" + "fmt" + "strings" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "k8s.io/client-go/kubernetes/fake" +) + +func zooKeeperFixture(role string) string { + data := "zk_version\t3.9.5\nzk_server_state\t" + role + "\nzk_quorum_size\t3\nzk_avg_latency\t0.123\n" + if role == zooKeeperLeader { + return data + "zk_peer_state\tleading - broadcast\nzk_synced_followers\t2\n" + } + return data + "zk_peer_state\tfollowing - broadcast\nzk_synced_observers\tnull\n" +} + +func TestZooKeeperParsing(t *testing.T) { + tests := []struct { + name, data string + valid bool + }{ + {zooKeeperLeader, zooKeeperFixture(zooKeeperLeader), true}, + {zooKeeperFollower, zooKeeperFixture(zooKeeperFollower), true}, + {"optional peer state absent", strings.ReplaceAll(zooKeeperFixture(zooKeeperFollower), "zk_peer_state\tfollowing - broadcast\n", ""), true}, + {"disabled command", "mntr is not executed because it is not in the whitelist.\n", false}, + {"not serving", "This ZooKeeper instance is not currently serving requests\n", false}, + {"empty response", "", false}, + {"missing role", "zk_quorum_size\t3\n", false}, + {"missing membership", "zk_server_state\tfollower\n", false}, + {"missing synchronized count", "zk_server_state\tleader\nzk_quorum_size\t3\n", false}, + {"duplicate role", zooKeeperFixture(zooKeeperLeader) + "zk_server_state\tfollower\n", false}, + {"duplicate empty value", "zk_version\t\n" + zooKeeperFixture(zooKeeperLeader), false}, + {"negative synchronized count", strings.ReplaceAll(zooKeeperFixture(zooKeeperLeader), "zk_synced_followers\t2", "zk_synced_followers\t-1"), false}, + {"invalid membership", strings.ReplaceAll(zooKeeperFixture(zooKeeperFollower), "zk_quorum_size\t3", "zk_quorum_size\tnull"), false}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + observation, err := parseZooKeeper(test.data, "zk-0") + if !test.valid { + require.Error(t, err) + return + } + require.NoError(t, err) + assert.Equal(t, 3, observation.voters) + assert.Equal(t, "zk-0", observation.pod) + }) + } +} + +func TestZooKeeperRequiresFullVotingEnsemble(t *testing.T) { + tests := []struct { + name string + mutate func([]zooKeeperObservation) []zooKeeperObservation + status string + }{ + {"healthy", func(o []zooKeeperObservation) []zooKeeperObservation { return o }, Healthy}, + {"quorum without full recovery", func(o []zooKeeperObservation) []zooKeeperObservation { o[0].synced = 1; return o }, Degraded}, + {"no leader", func(o []zooKeeperObservation) []zooKeeperObservation { o[0].role = zooKeeperFollower; return o }, Degraded}, + {"two leaders", func(o []zooKeeperObservation) []zooKeeperObservation { o[1].role = zooKeeperLeader; return o }, Degraded}, + {"election", func(o []zooKeeperObservation) []zooKeeperObservation { o[1].role = "looking"; return o }, Degraded}, + {"observer", func(o []zooKeeperObservation) []zooKeeperObservation { o[1].role = "observer"; return o }, Degraded}, + {"standalone", func(o []zooKeeperObservation) []zooKeeperObservation { o[1].role = "standalone"; return o }, Degraded}, + {"synchronizing", func(o []zooKeeperObservation) []zooKeeperObservation { + o[1].peerState = "following - synchronization" + return o + }, Degraded}, + {"inconsistent role", func(o []zooKeeperObservation) []zooKeeperObservation { + o[0].peerState = "following - broadcast" + return o + }, Degraded}, + {"unexpected voter configuration", func(o []zooKeeperObservation) []zooKeeperObservation { o[1].voters = 5; return o }, Unknown}, + {"duplicate member", func(o []zooKeeperObservation) []zooKeeperObservation { o[1].pod = o[0].pod; return o }, Unknown}, + {"too few voters", func(o []zooKeeperObservation) []zooKeeperObservation { return o[:2] }, Degraded}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + observations := []zooKeeperObservation{ + {pod: "zk-0", role: zooKeeperLeader, voters: 3, synced: 2}, + {pod: "zk-1", role: zooKeeperFollower, voters: 3}, + {pod: "zk-2", role: zooKeeperFollower, voters: 3}, + } + assert.Equal(t, test.status, evaluateZooKeeper(test.mutate(observations)).Status) + }) + } +} + +func TestZooKeeperQueriesEveryMemberAndRechecksLeader(t *testing.T) { + tests := []struct { + name, status string + finalRole string + finalSynced string + queryError bool + }{ + {name: "healthy", status: Healthy, finalRole: zooKeeperLeader, finalSynced: "2"}, + {name: "election during observation", status: Unknown, finalRole: zooKeeperFollower}, + {name: "follower falls behind during observation", status: Degraded, finalRole: zooKeeperLeader, finalSynced: "1"}, + {name: "query denied", status: Unknown, queryError: true}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + var queried []string + kube := &fakeKubernetes{ + client: fake.NewSimpleClientset(databaseObjects("zookeeper", "zookeeper", "zookeeper", 3)...), + exec: func(_ context.Context, namespace, pod, container string, command []string) ([]byte, error) { + assert.Equal(t, "test", namespace) + assert.Equal(t, "zookeeper", container) + assert.Equal(t, []string{"bash", "-ec", zookeeperQuery}, command) + queried = append(queried, pod) + if test.queryError { + return nil, fmt.Errorf("query denied") + } + if len(queried) == 4 { + return []byte(strings.ReplaceAll(zooKeeperFixture(test.finalRole), "zk_synced_followers\t2", "zk_synced_followers\t"+test.finalSynced)), nil + } + if pod == "zookeeper-0" { + return []byte(zooKeeperFixture(zooKeeperLeader)), nil + } + return []byte(zooKeeperFixture(zooKeeperFollower)), nil + }, + } + options := testOptions() + options.Components = []string{"zookeeper"} + probe, err := New(kube, options) + require.NoError(t, err) + assert.Equal(t, test.status, probe.Check(context.Background()).Status) + if !test.queryError { + assert.Equal(t, []string{"zookeeper-0", "zookeeper-1", "zookeeper-2", "zookeeper-0"}, queried) + } + }) + } +} + +func TestZooKeeperMissingPodCannotPass(t *testing.T) { + objects := databaseObjects("zookeeper", "zookeeper", "zookeeper", 3) + kube := &fakeKubernetes{client: fake.NewSimpleClientset(objects[:len(objects)-1]...)} + options := testOptions() + options.Components = []string{"zookeeper"} + probe, err := New(kube, options) + require.NoError(t, err) + assert.Equal(t, Unknown, probe.Check(context.Background()).Status) + assert.Zero(t, kube.calls) +} From 5109f24462c65ad5dc5f52b2f11d9795bfeef780 Mon Sep 17 00:00:00 2001 From: Vladimir Iliakov Date: Fri, 11 Sep 2026 10:57:13 +0200 Subject: [PATCH 4/4] Show replication wait stability progress and stop cleanly on cancellation Timestamp progress and final table output, and expose the elapsed/remaining stability period, resets and completion. Keep the default 30-second requirement after the first completed healthy observation. Shorten the next polling delay when the stability deadline comes before the configured interval. Discard interrupted observations, preserve the last completed report and stop launching queries after cancellation. Return nonzero with an unknown overall result without flooding output with canceled request URLs. Full Go tests and golangci-lint pass. Virtual-time tests cover the default stability period, resets, long intervals and cancellation. Live Beest validation exited successfully after three healthy rounds (37.4 seconds of observed stability); a separate SIGINT check retained the last completed report and exited nonzero without query URLs. --- cmd/replication/replication.go | 40 ++++++++- cmd/replication/replication_test.go | 47 +++++++++++ docs/replication.md | 27 +++++- internal/orchestration/replication/checker.go | 17 +++- .../orchestration/replication/checker_test.go | 20 +++++ internal/orchestration/replication/wait.go | 47 ++++++++--- .../orchestration/replication/wait_test.go | 83 +++++++++++++++++++ 7 files changed, 262 insertions(+), 19 deletions(-) diff --git a/cmd/replication/replication.go b/cmd/replication/replication.go index 47f61ee..bfcb207 100644 --- a/cmd/replication/replication.go +++ b/cmd/replication/replication.go @@ -118,17 +118,36 @@ func observe(ctx context.Context, check func(context.Context) checker.Report, f if !f.wait { report := check(ctx) if ctx.Err() != nil { - return report, fmt.Errorf("replication check ended: %w", ctx.Err()) + return checker.Report{Namespace: report.Namespace}, fmt.Errorf("replication check ended: %w", ctx.Err()) } return report, nil } - return checker.Wait(ctx, check, f.interval, f.stableFor, func(report checker.Report) { + _, _ = fmt.Fprintf(progress, "[%s] Waiting for all checks to remain healthy for %s (timeout %s).\n", + time.Now().UTC().Format(time.RFC3339), f.stableFor, f.timeout) + return checker.Wait(ctx, check, f.interval, f.stableFor, func(report checker.Report, state checker.WaitProgress) { + timestamp := state.ObservedAt.Format(time.RFC3339) for _, check := range report.Checks { - _, _ = fmt.Fprintf(progress, "%s %s: %s\n", check.Component, check.Status, strings.Join(check.Messages, "; ")) + _, _ = fmt.Fprintf(progress, "[%s] %s %s: %s\n", timestamp, check.Component, check.Status, strings.Join(check.Messages, "; ")) } + _, _ = fmt.Fprintf(progress, "[%s] %s\n", timestamp, stabilityMessage(report, state)) }) } +func stabilityMessage(report checker.Report, state checker.WaitProgress) string { + switch { + case state.Complete: + return fmt.Sprintf("All checks healthy; stability period satisfied (%s/%s).", + state.HealthyFor.Round(time.Millisecond), state.Required) + case report.Status == checker.Healthy: + return fmt.Sprintf("All checks healthy; verifying stability: %s/%s (%s remaining).", + state.HealthyFor.Round(time.Millisecond), state.Required, (state.Required - state.HealthyFor).Round(time.Millisecond)) + case state.Reset: + return "Stability period reset; waiting for all selected checks to become healthy." + default: + return "Waiting for all selected checks to become healthy; stability period has not started." + } +} + func writeReport(writer io.Writer, format string, report checker.Report) error { if format == "json" { if err := json.NewEncoder(writer).Encode(report); err != nil { @@ -136,6 +155,21 @@ func writeReport(writer io.Writer, format string, report checker.Report) error { } return nil } + if _, err := fmt.Fprintf(writer, "[%s] Replication result: %s\n", time.Now().UTC().Format(time.RFC3339), report.Status); err != nil { + return fmt.Errorf("write report status: %w", err) + } + if report.Error != "" { + if _, err := fmt.Fprintln(writer, report.Error); err != nil { + return fmt.Errorf("write report error: %w", err) + } + } + if len(report.Checks) == 0 { + _, err := fmt.Fprintln(writer, "No completed observation.") + return err + } + if _, err := fmt.Fprintf(writer, "Last completed observation started: %s\n", report.CheckedAt.Format(time.RFC3339)); err != nil { + return fmt.Errorf("write observation timestamp: %w", err) + } table := tabwriter.NewWriter(writer, 0, 0, tablePadding, ' ', 0) if _, err := fmt.Fprintln(table, "COMPONENT\tSTATUS\tDETAILS"); err != nil { return fmt.Errorf("write report header: %w", err) diff --git a/cmd/replication/replication_test.go b/cmd/replication/replication_test.go index fa2ade3..c2fd8d7 100644 --- a/cmd/replication/replication_test.go +++ b/cmd/replication/replication_test.go @@ -4,7 +4,9 @@ import ( "bytes" "context" "encoding/json" + "strings" "testing" + "testing/synctest" "time" "github.com/stretchr/testify/assert" @@ -78,3 +80,48 @@ func TestReportAndExitAgree(t *testing.T) { }) } } + +func TestWaitOutputShowsTimestampsAndStabilityProgress(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + var stderr bytes.Buffer + report, err := observe(context.Background(), func(context.Context) checker.Report { + return checker.Report{CheckedAt: time.Now().UTC(), Status: checker.Healthy, + Checks: []checker.Result{{Component: "kafka", Status: checker.Healthy, Messages: []string{"all replicas in sync"}}}} + }, &flags{wait: true, interval: 10 * time.Second, stableFor: 30 * time.Second, timeout: time.Minute}, &stderr) + require.NoError(t, err) + assert.Equal(t, checker.Healthy, report.Status) + for _, line := range strings.Split(strings.TrimSpace(stderr.String()), "\n") { + assert.Regexp(t, `^\[\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z\] `, line) + } + assert.Contains(t, stderr.String(), "verifying stability: 0s/30s (30s remaining)") + assert.Contains(t, stderr.String(), "verifying stability: 10s/30s (20s remaining)") + assert.Contains(t, stderr.String(), "stability period satisfied (30s/30s)") + }) +} + +func TestCancelledTableLabelsLastCompletedObservation(t *testing.T) { + var stdout bytes.Buffer + report := checker.Report{CheckedAt: time.Date(2026, time.September, 11, 12, 0, 0, 0, time.UTC), Status: checker.Healthy, + Checks: []checker.Result{{Component: "kafka", Status: checker.Healthy, Messages: []string{"all replicas in sync"}}}} + err := finishReport(&stdout, "table", report, context.Canceled) + require.ErrorIs(t, err, context.Canceled) + assert.Contains(t, stdout.String(), "Replication result: unknown") + assert.Contains(t, stdout.String(), "context canceled") + assert.Contains(t, stdout.String(), "Last completed observation started: 2026-09-11T12:00:00Z") + assert.Contains(t, stdout.String(), "all replicas in sync") +} + +func TestCancellationBeforeFirstObservation(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + var stdout, stderr bytes.Buffer + report, err := observe(ctx, func(context.Context) checker.Report { + cancel() + return checker.Report{Namespace: "test", Checks: []checker.Result{{Messages: []string{"aborted request URL"}}}} + }, &flags{wait: true, interval: time.Second}, &stderr) + require.ErrorIs(t, err, context.Canceled) + assert.Equal(t, "test", report.Namespace) + assert.Empty(t, report.Checks) + assert.NotContains(t, stderr.String(), "aborted") + require.ErrorIs(t, finishReport(&stdout, "table", report, err), context.Canceled) + assert.Contains(t, stdout.String(), "No completed observation.") +} diff --git a/docs/replication.md b/docs/replication.md index ae24ca6..80a98ca 100644 --- a/docs/replication.md +++ b/docs/replication.md @@ -37,10 +37,29 @@ sts-backup replication check \ ``` Wait mode requires consecutive healthy observations spanning `--stable-for`. -An unsuccessful observation resets that period. Progress goes to stderr; -stdout contains one final report. The overall deadline includes database -queries. A timeout or cancellation returns nonzero, even if an earlier -observation was healthy. `--request-timeout` bounds each API request or query. +The default is 30 seconds, starting when the first fully healthy observation +completes. Earlier rounds with any `unknown` or `degraded` component do not +count. An unsuccessful observation resets the period. + +Progress includes UTC timestamps and shows the elapsed and remaining healthy +period, resets, and successful completion. Seeing every component healthy +does not mean the stability period has already elapsed. Database queries take +time; the checker needs another completed healthy observation to confirm the +period, rather than exiting on a timer alone. + +To finish on the first fully healthy observation, explicitly use +`--wait --stable-for 0s`. + +Progress goes to stderr; stdout contains one final report. Table output includes +the report time and the start time of the last completed observation. JSON +retains its `checkedAt` timestamp. The overall deadline includes database +queries. `--request-timeout` bounds each API request or query. + +Ctrl+C stops further queries. Cancellation or timeout returns nonzero and sets +the overall result to `unknown`, retaining the last completed observation +instead of replacing it with errors from interrupted queries. If no observation +completed, the report says so. Healthy component results in a cancelled report +describe that previous observation; the wait did not finish successfully. Select an explicit subset if a database is intentionally disabled: diff --git a/internal/orchestration/replication/checker.go b/internal/orchestration/replication/checker.go index 7f8cdf4..936cd1b 100644 --- a/internal/orchestration/replication/checker.go +++ b/internal/orchestration/replication/checker.go @@ -34,12 +34,20 @@ func (c *Checker) Check(ctx context.Context) Report { } before, err := c.discover(ctx) for _, component := range c.options.Components { + if ctx.Err() != nil { + report.Status = Unknown + return report + } if err != nil { report.Checks = append(report.Checks, result(component, Unknown, err.Error())) continue } report.Checks = append(report.Checks, c.checkComponent(ctx, before, component)) } + if ctx.Err() != nil { + report.Status = Unknown + return report + } if err == nil { after, afterErr := c.discover(ctx) if afterErr != nil || before.fingerprint() != after.fingerprint() { @@ -58,9 +66,16 @@ func (c *Checker) Check(ctx context.Context) Report { } func (c *Checker) query(ctx context.Context, pod, container string, command []string) ([]byte, error) { + if err := ctx.Err(); err != nil { + return nil, err + } ctx, cancel := context.WithTimeout(ctx, c.options.RequestTimeout) defer cancel() - return c.kube.Exec(ctx, c.options.Namespace, pod, container, command) + data, err := c.kube.Exec(ctx, c.options.Namespace, pod, container, command) + if ctx.Err() != nil { + return nil, ctx.Err() + } + return data, err } func (c *Checker) checkComponent(ctx context.Context, inventory inventory, component string) Result { diff --git a/internal/orchestration/replication/checker_test.go b/internal/orchestration/replication/checker_test.go index 43b7daa..b959c36 100644 --- a/internal/orchestration/replication/checker_test.go +++ b/internal/orchestration/replication/checker_test.go @@ -204,3 +204,23 @@ func TestInvalidScopeRejected(t *testing.T) { require.Error(t, err) } } + +func TestCancellationStopsRemainingQueriesAndDiscovery(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + client := fake.NewSimpleClientset(kafkaObjects()...) + kube := &fakeKubernetes{client: client, exec: func(context.Context, string, string, string, []string) ([]byte, error) { + cancel() + return nil, fmt.Errorf("aborted request with query URL") + }} + options := testOptions() + options.Components = []string{"kafka", "zookeeper", "clickhouse"} + probe, err := New(kube, options) + require.NoError(t, err) + report := probe.Check(ctx) + assert.Equal(t, Unknown, report.Status) + require.Len(t, report.Checks, 1) + assert.Equal(t, []string{"context canceled"}, report.Checks[0].Messages) + assert.Equal(t, 1, kube.calls) + assert.Len(t, client.Actions(), 2, "no second discovery after cancellation") +} diff --git a/internal/orchestration/replication/wait.go b/internal/orchestration/replication/wait.go index a830bf8..a03e9f4 100644 --- a/internal/orchestration/replication/wait.go +++ b/internal/orchestration/replication/wait.go @@ -6,9 +6,18 @@ import ( "time" ) +// WaitProgress describes the stability period after a completed observation. +type WaitProgress struct { + ObservedAt time.Time + HealthyFor time.Duration + Required time.Duration + Reset bool + Complete bool +} + // Wait polls until replication remains healthy for stableFor or the caller's deadline expires. -// The observer receives every report and can write progress to stderr. -func Wait(ctx context.Context, check func(context.Context) Report, interval, stableFor time.Duration, observe func(Report)) (Report, error) { +// Interrupted observations never replace the last completed report. +func Wait(ctx context.Context, check func(context.Context) Report, interval, stableFor time.Duration, observe func(Report, WaitProgress)) (Report, error) { if interval <= 0 || stableFor < 0 { return Report{}, fmt.Errorf("interval must be positive and stable-for cannot be negative") } @@ -18,24 +27,40 @@ func Wait(ctx context.Context, check func(context.Context) Report, interval, sta if err := ctx.Err(); err != nil { return report, fmt.Errorf("replication wait ended: %w", err) } - report = check(ctx) - if observe != nil { - observe(report) - } + next := check(ctx) if err := ctx.Err(); err != nil { + if report.Namespace == "" { + report.Namespace = next.Namespace + } return report, fmt.Errorf("replication wait ended: %w", err) } + report = next + now := time.Now() + progress := WaitProgress{ObservedAt: now.UTC(), Required: stableFor} if report.Status != Healthy { + progress.Reset = !healthySince.IsZero() healthySince = time.Time{} } else { if healthySince.IsZero() { - healthySince = time.Now() - } - if time.Since(healthySince) >= stableFor { - return report, nil + healthySince = now } + progress.HealthyFor = now.Sub(healthySince) + progress.Complete = progress.HealthyFor >= stableFor + } + if observe != nil { + observe(report, progress) + } + if err := ctx.Err(); err != nil { + return report, fmt.Errorf("replication wait ended: %w", err) + } + if progress.Complete { + return report, nil + } + delay := interval + if report.Status == Healthy { + delay = min(delay, stableFor-progress.HealthyFor) } - timer := time.NewTimer(interval) + timer := time.NewTimer(delay) select { case <-ctx.Done(): timer.Stop() diff --git a/internal/orchestration/replication/wait_test.go b/internal/orchestration/replication/wait_test.go index 87b01e9..c11eb86 100644 --- a/internal/orchestration/replication/wait_test.go +++ b/internal/orchestration/replication/wait_test.go @@ -3,6 +3,7 @@ package replication import ( "context" "testing" + "testing/synctest" "time" "github.com/stretchr/testify/assert" @@ -57,3 +58,85 @@ func TestWaitCancellation(t *testing.T) { }, time.Hour, 0, nil) require.ErrorIs(t, err, context.Canceled) } + +func TestWaitExitsAfterDefaultStabilityPeriodWithSlowChecks(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + start := time.Now() + var progress []WaitProgress + ctx, cancel := context.WithTimeout(context.Background(), time.Minute) + defer cancel() + report, err := Wait(ctx, func(context.Context) Report { + time.Sleep(2 * time.Second) + return Report{Status: Healthy} + }, 10*time.Second, 30*time.Second, func(_ Report, state WaitProgress) { + progress = append(progress, state) + }) + require.NoError(t, err) + assert.Equal(t, Healthy, report.Status) + require.Len(t, progress, 4) + assert.Equal(t, time.Duration(0), progress[0].HealthyFor) + assert.Equal(t, 12*time.Second, progress[1].HealthyFor) + assert.Equal(t, 32*time.Second, progress[3].HealthyFor) + assert.True(t, progress[3].Complete) + assert.Equal(t, 34*time.Second, time.Since(start)) + }) +} + +func TestWaitRechecksAtStabilityDeadlineBeforeLongInterval(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + start := time.Now() + ctx, cancel := context.WithTimeout(context.Background(), 45*time.Second) + defer cancel() + calls := 0 + _, err := Wait(ctx, func(context.Context) Report { + calls++ + return Report{Status: Healthy} + }, time.Minute, 30*time.Second, nil) + require.NoError(t, err) + assert.Equal(t, 2, calls) + assert.Equal(t, 30*time.Second, time.Since(start)) + }) +} + +func TestWaitResetsStabilityProgress(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + calls := 0 + var progress []WaitProgress + _, err := Wait(context.Background(), func(context.Context) Report { + calls++ + if calls == 3 { + return Report{Status: Degraded} + } + return Report{Status: Healthy} + }, 10*time.Second, 30*time.Second, func(_ Report, state WaitProgress) { + progress = append(progress, state) + }) + require.NoError(t, err) + require.Len(t, progress, 7) + assert.True(t, progress[2].Reset) + assert.Zero(t, progress[3].HealthyFor) + assert.Equal(t, 30*time.Second, progress[6].HealthyFor) + assert.True(t, progress[6].Complete) + }) +} + +func TestWaitCancellationPreservesLastCompletedObservation(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + calls, observed := 0, 0 + lastCompleted := Report{Namespace: "test", CheckedAt: time.Now(), Status: Healthy, + Checks: []Result{result("kafka", Healthy, "all partitions in sync")}} + report, err := Wait(ctx, func(context.Context) Report { + calls++ + if calls == 3 { + cancel() + return Report{Status: Unknown, Checks: []Result{result("kafka", Unknown, "aborted query URL")}} + } + return lastCompleted + }, 10*time.Second, 30*time.Second, func(Report, WaitProgress) { observed++ }) + require.ErrorIs(t, err, context.Canceled) + assert.Equal(t, lastCompleted, report) + assert.Equal(t, 2, observed) + }) +}