test(longhaul): add kill-operator-pod and kill-primary-pod chaos operations

test/longhaul/README.md

@@ -125,6 +125,7 @@ All configuration is via environment variables.
 | `LONGHAUL_DOCUMENTDB_URI` | Yes | — | Connection string to the DocumentDB gateway. |
 | `LONGHAUL_CLUSTER_NAME` | Yes | — | Name of the target DocumentDB cluster CR. |
 | `LONGHAUL_NAMESPACE` | No | `default` | Kubernetes namespace of the target cluster. |
+| `LONGHAUL_OPERATOR_NAMESPACE` | No | `documentdb-operator` | Namespace of the DocumentDB operator Deployment (target of the `kill-operator-pod` chaos op). |
 | `LONGHAUL_MAX_DURATION` | No | `30m` | Max test duration. Use `0s` for run-until-failure. |
 | `LONGHAUL_NUM_WRITERS` | No | `5` | Number of concurrent writers. |
 | `LONGHAUL_OP_COOLDOWN` | No | `5m` | Cooldown between management operations. |
@@ -134,6 +135,44 @@ All configuration is via environment variables.
 | `LONGHAUL_REPORT_INTERVAL` | No | `1h` | How often to write checkpoint reports to ConfigMap. |
 | `LONGHAUL_RESET_DATA` | No | `false` | If `true`, drop the workload collection on startup. Off by default so a Deployment pod restart preserves durability history. |
 
+## Operations
+
+The scheduler runs one disruptive operation at a time, gated by steady state and
+a global cooldown. Current operations:
+
+| Operation | Kind | Notes |
+|-----------|------|-------|
+| `scale-up` / `scale-down` | Topology | Adjusts `spec.instancesPerNode` within `[MIN, MAX]`. Only adds/removes a standby, so the primary write path is untouched (near-zero outage budget). |
+| `upgrade-documentdb` | Topology | In-place version upgrade; requires HA (`instancesPerNode>=2`). |
+| `kill-operator-pod` | Chaos | Deletes the operator pod; asserts the data plane keeps serving (near-zero outage budget). |
+| `kill-primary-pod` | Chaos | Deletes the CNPG primary pod to exercise automatic failover; requires HA (`instancesPerNode>=2`). |
+
+Operations that keep the write path up throughout — the scale ops and
+`kill-operator-pod` — share the near-zero `journal.NoOutagePolicy` budget instead
+of ad-hoc per-op numbers, so a regression that unexpectedly disrupts writes
+during a "safe" operation trips the policy.
+
+Outage budgets are expressed as **wall-clock write-outage durations**
+(`OutagePolicy.MaxWriteOutage`), not raw write-failure counts. The journal
+converts the observed failure count into an estimated outage using the workload's
+aggregate write rate (`workload.AggregateWriteRate(NumWriters)`), so the budgets
+are independent of `LONGHAUL_NUM_WRITERS`: `NoOutagePolicy` ≈ 300ms (noise
+cushion), while `kill-primary-pod` and `upgrade-documentdb` share the
+`journal.PrimaryHandoverPolicy` budget of 30s — both interrupt writes for a
+single primary handover (an ungraceful failover vs. a graceful switchover), and
+an upgrade's longer whole-topology restart is bounded by `MustRecoverWithin`,
+not the write-outage budget.
+
+### RBAC for chaos operations
+
+Beyond the base RBAC the driver already needs, the chaos operations require the
+driver ServiceAccount to be granted (in `deploy/rbac.yaml`):
+
+- **`kill-primary-pod`** — `get`/`list` on `clusters.postgresql.cnpg.io` (to read
+  `status.currentPrimary`) and `delete` on `pods` in the cluster namespace.
+- **`kill-operator-pod`** — `get` on `deployments` and `get`/`list`/`delete` on
+  `pods` in the operator namespace (`LONGHAUL_OPERATOR_NAMESPACE`).
+
 ## CI Safety
 
 The long haul test binary is deployed as a Kubernetes Deployment on a dedicated AKS

test/longhaul/cmd/longhaul/main.go

@@ -57,6 +57,7 @@ func run(cfg config.Config) int {
 
 	// Initialize components.
 	j := journal.New()
+	j.SetWriteRate(workload.AggregateWriteRate(cfg.NumWriters))
 	metrics := workload.NewMetrics()
 
 	// Connect to DocumentDB.
@@ -140,6 +141,8 @@ func run(cfg config.Config) int {
 		operations.NewScaleUp(clusterClient, healthMon, cfg.MaxInstances, cfg.RecoveryTimeout),
 		operations.NewScaleDown(clusterClient, healthMon, cfg.MinInstances, cfg.RecoveryTimeout),
 		operations.NewUpgradeDocumentDB(clusterClient, k8sClientset, healthMon, j, cfg.Namespace, cfg.RecoveryTimeout),
+		operations.NewKillOperatorPod(k8sClientset, cfg.OperatorNamespace, cfg.RecoveryTimeout),
+		operations.NewKillPrimaryPod(clusterClient, healthMon, cfg.RecoveryTimeout),
 	}
 
 	// Start operation scheduler.

test/longhaul/config/config.go

@@ -18,6 +18,10 @@ const (
 	EnvNamespace   = "LONGHAUL_NAMESPACE"
 	EnvClusterName = "LONGHAUL_CLUSTER_NAME"
 
+	// EnvOperatorNamespace is the namespace where the DocumentDB operator
+	// Deployment runs (target of the kill-operator-pod chaos op).
+	EnvOperatorNamespace = "LONGHAUL_OPERATOR_NAMESPACE"
+
 	// Workload and operation tuning.
 	EnvDocumentDBURI   = "LONGHAUL_DOCUMENTDB_URI"
 	EnvNumWriters      = "LONGHAUL_NUM_WRITERS"
@@ -47,6 +51,10 @@ type Config struct {
 	// ClusterName is the name of the target DocumentDB cluster CR.
 	ClusterName string
 
+	// OperatorNamespace is the namespace of the DocumentDB operator Deployment,
+	// targeted by the kill-operator-pod chaos operation.
+	OperatorNamespace string
+
 	// DocumentDBURI is the DocumentDB connection string for data-plane workload.
 	DocumentDBURI string
 
@@ -82,17 +90,18 @@ type Config struct {
 // DefaultConfig returns a Config with safe defaults for local development.
 func DefaultConfig() Config {
 	return Config{
-		MaxDuration:     30 * time.Minute,
-		Namespace:       "default",
-		ClusterName:     "",
-		DocumentDBURI:   "",
-		NumWriters:      5,
-		OpCooldown:      5 * time.Minute,
-		RecoveryTimeout: 5 * time.Minute,
-		SteadyStateWait: 60 * time.Second,
-		MinInstances:    1,
-		MaxInstances:    3,
-		ReportInterval:  1 * time.Hour,
+		MaxDuration:       30 * time.Minute,
+		Namespace:         "default",
+		ClusterName:       "",
+		OperatorNamespace: "documentdb-operator",
+		DocumentDBURI:     "",
+		NumWriters:        5,
+		OpCooldown:        5 * time.Minute,
+		RecoveryTimeout:   5 * time.Minute,
+		SteadyStateWait:   60 * time.Second,
+		MinInstances:      1,
+		MaxInstances:      3,
+		ReportInterval:    1 * time.Hour,
 	}
 }
 
@@ -117,6 +126,10 @@ func LoadFromEnv() (Config, error) {
 		cfg.ClusterName = v
 	}
 
+	if v := os.Getenv(EnvOperatorNamespace); v != "" {
+		cfg.OperatorNamespace = v
+	}
+
 	if v := os.Getenv(EnvDocumentDBURI); v != "" {
 		cfg.DocumentDBURI = v
 	}
@@ -195,6 +208,9 @@ func (c *Config) Validate() error {
 	if c.ClusterName == "" {
 		return fmt.Errorf("cluster name must not be empty")
 	}
+	if c.OperatorNamespace == "" {
+		return fmt.Errorf("operator namespace must not be empty")
+	}
 	if c.NumWriters < 1 {
 		return fmt.Errorf("num writers must be at least 1, got %d", c.NumWriters)
 	}

test/longhaul/config/config_test.go

@@ -17,6 +17,7 @@ var _ = Describe("Config", func() {
 			Expect(cfg.MaxDuration).To(Equal(30 * time.Minute))
 			Expect(cfg.Namespace).To(Equal("default"))
 			Expect(cfg.ClusterName).To(BeEmpty())
+			Expect(cfg.OperatorNamespace).To(Equal("documentdb-operator"))
 			Expect(cfg.NumWriters).To(Equal(5))
 			Expect(cfg.OpCooldown).To(Equal(5 * time.Minute))
 			Expect(cfg.RecoveryTimeout).To(Equal(5 * time.Minute))
@@ -32,6 +33,7 @@ var _ = Describe("Config", func() {
 		BeforeEach(func() {
 			for _, k := range []string{
 				EnvEnabled, EnvMaxDuration, EnvNamespace, EnvClusterName,
+				EnvOperatorNamespace,
 				EnvDocumentDBURI, EnvNumWriters,
 				EnvOpCooldown, EnvRecoveryTimeout, EnvSteadyStateWait,
 				EnvMinInstances, EnvMaxInstances, EnvReportInterval,
@@ -69,6 +71,13 @@ var _ = Describe("Config", func() {
 			Expect(cfg.ClusterName).To(Equal("my-cluster"))
 		})
 
+		It("parses OperatorNamespace from env", func() {
+			GinkgoT().Setenv(EnvOperatorNamespace, "custom-operator-ns")
+			cfg, err := LoadFromEnv()
+			Expect(err).NotTo(HaveOccurred())
+			Expect(cfg.OperatorNamespace).To(Equal("custom-operator-ns"))
+		})
+
 		It("returns error for invalid MaxDuration", func() {
 			GinkgoT().Setenv(EnvMaxDuration, "not-a-duration")
 			_, err := LoadFromEnv()
@@ -124,6 +133,13 @@ var _ = Describe("Config", func() {
 			Expect(cfg.Validate()).To(MatchError(ContainSubstring("cluster name")))
 		})
 
+		It("fails when OperatorNamespace is empty", func() {
+			cfg := DefaultConfig()
+			cfg.ClusterName = "test"
+			cfg.OperatorNamespace = ""
+			Expect(cfg.Validate()).To(MatchError(ContainSubstring("operator namespace")))
+		})
+
 		It("fails when MaxDuration is negative", func() {
 			cfg := DefaultConfig()
 			cfg.ClusterName = "test"

test/longhaul/go.mod

@@ -3,6 +3,7 @@ module github.com/documentdb/documentdb-operator/test/longhaul
 go 1.25.11
 
 require (
+	github.com/cloudnative-pg/cloudnative-pg v1.28.1
 	github.com/documentdb/documentdb-operator v0.0.0-00010101000000-000000000000
 	github.com/documentdb/documentdb-operator/test/shared v0.0.0-00010101000000-000000000000
 	github.com/onsi/ginkgo/v2 v2.28.1
@@ -24,7 +25,6 @@ require (
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/cloudnative-pg/barman-cloud v0.4.1-0.20260108104508-ced266c145f5 // indirect
-	github.com/cloudnative-pg/cloudnative-pg v1.28.1 // indirect
 	github.com/cloudnative-pg/cnpg-i v0.3.1 // indirect
 	github.com/cloudnative-pg/machinery v0.3.3 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect

test/longhaul/journal/journal.go

@@ -52,13 +52,39 @@ type Journal struct {
 
 	// All closed disruption windows.
 	closedWindows []DisruptionWindow
+
+	// writesPerSecond is the workload's aggregate write rate, stamped onto each
+	// disruption window so ExceededPolicy can convert write-failure counts into
+	// an estimated outage duration. Defaults to DefaultWritesPerSecond; override
+	// with SetWriteRate once the real writer count is known.
+	writesPerSecond float64
 }
 
+// DefaultWritesPerSecond is the assumed aggregate write rate used until
+// SetWriteRate is called. It matches the default workload (5 writers at one
+// write per 100ms = 50 writes/s) so tests and un-configured journals still
+// evaluate write-outage budgets sensibly.
+const DefaultWritesPerSecond = 50.0
+
 // New creates a new empty Journal.
 func New() *Journal {
 	return &Journal{
-		events: make([]Event, 0, 256),
+		events:          make([]Event, 0, 256),
+		writesPerSecond: DefaultWritesPerSecond,
+	}
+}
+
+// SetWriteRate records the workload's aggregate write rate (writes/second across
+// all writers) so disruption windows can translate write-failure counts into an
+// estimated outage duration. Non-positive values are ignored, preserving the
+// default. Safe for concurrent use.
+func (j *Journal) SetWriteRate(writesPerSecond float64) {
+	if writesPerSecond <= 0 {
+		return
 	}
+	j.mu.Lock()
+	defer j.mu.Unlock()
+	j.writesPerSecond = writesPerSecond
 }
 
 // Record appends a new event to the journal. Safe for concurrent use.
@@ -112,9 +138,10 @@ func (j *Journal) OpenDisruptionWindow(operationName string, policy OutagePolicy
 	}
 
 	j.activeWindow = &DisruptionWindow{
-		OperationName: operationName,
-		StartTime:     time.Now(),
-		Policy:        policy,
+		OperationName:   operationName,
+		StartTime:       time.Now(),
+		Policy:          policy,
+		WritesPerSecond: j.writesPerSecond,
 	}
 
 	j.events = append(j.events, Event{

test/longhaul/journal/journal_test.go

@@ -43,7 +43,7 @@ var _ = Describe("Journal", func() {
 	Describe("DisruptionWindow lifecycle", func() {
 		It("opens, records failures, and closes correctly", func() {
 			j := New()
-			policy := OutagePolicy{MustRecoverWithin: time.Minute, AllowedWriteFailures: 10}
+			policy := OutagePolicy{MustRecoverWithin: time.Minute, MaxWriteOutage: time.Second}
 
 			Expect(j.ActiveWindow()).To(BeNil())
 
@@ -89,14 +89,14 @@ var _ = Describe("Journal", func() {
 
 		It("returns false on a closed window within budget", func() {
 			j := New()
-			j.OpenDisruptionWindow("op", OutagePolicy{MustRecoverWithin: time.Minute, AllowedWriteFailures: 10})
+			j.OpenDisruptionWindow("op", OutagePolicy{MustRecoverWithin: time.Minute, MaxWriteOutage: time.Second})
 			j.CloseDisruptionWindow()
 			Expect(j.HasPolicyViolation()).To(BeFalse())
 		})
 
-		It("returns true on a closed window over write-failure budget", func() {
+		It("returns true on a closed window over write-outage budget", func() {
 			j := New()
-			j.OpenDisruptionWindow("op", OutagePolicy{MustRecoverWithin: time.Minute, AllowedWriteFailures: 1})
+			j.OpenDisruptionWindow("op", OutagePolicy{MustRecoverWithin: time.Minute, MaxWriteOutage: 10 * time.Millisecond})
 			j.RecordWriteFailure()
 			j.RecordWriteFailure()
 			j.CloseDisruptionWindow()
@@ -105,7 +105,7 @@ var _ = Describe("Journal", func() {
 
 		It("returns true on an active window over time budget", func() {
 			j := New()
-			j.OpenDisruptionWindow("op", OutagePolicy{MustRecoverWithin: time.Nanosecond, AllowedWriteFailures: 10})
+			j.OpenDisruptionWindow("op", OutagePolicy{MustRecoverWithin: time.Nanosecond, MaxWriteOutage: time.Second})
 			time.Sleep(1 * time.Millisecond)
 			Expect(j.HasPolicyViolation()).To(BeTrue())
 		})