Add read-only HA replication checks

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 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

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, ClickHouse and ZooKeeper replication in one namespace:
+
+```bash
+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.
+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.

cmd/replication/replication.go

@@ -0,0 +1,186 @@
+// 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, 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,
+		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.kubeconfig, "kubeconfig", "", "Kubeconfig path; uses normal kubeconfig or in-cluster credentials")
+	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")
+	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")
+	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 checker.Report{Namespace: report.Namespace}, fmt.Errorf("replication check ended: %w", ctx.Err())
+		}
+		return report, nil
+	}
+	_, _ = 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: %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 {
+			return fmt.Errorf("write JSON report: %w", err)
+		}
+		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)
+	}
+	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
+}

cmd/replication/replication_test.go

@@ -0,0 +1,127 @@
+package replication
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"strings"
+	"testing"
+	"testing/synctest"
+	"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", "--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.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) {
+	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)
+			}
+		})
+	}
+}
+
+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.")
+}

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() {