From 25b8f86a4945903b65c27a839231c0a5ede71cff Mon Sep 17 00:00:00 2001 From: Vladimir Iliakov Date: Mon, 9 Nov 2026 19:03:25 +0100 Subject: [PATCH 1/6] Wait for pods selected by each scaled workload's selector on scale-down The restore label is set on the workload metadata only, never on pod templates, so waiting on the label matched no pods and ScaleDown returned while the pods were still terminating. Wait on each workload's spec.selector. Applies to every datastore. Co-Authored-By: Claude Sonnet 5.5 --- internal/clients/k8s/client.go | 30 +++++++++++-- internal/clients/k8s/client_test.go | 27 ++++++++++++ internal/orchestration/scale/scale.go | 46 ++++++++++---------- internal/orchestration/scale/scale_test.go | 50 ++++++++++++++++++++++ 4 files changed, 127 insertions(+), 26 deletions(-) diff --git a/internal/clients/k8s/client.go b/internal/clients/k8s/client.go index e955666..3f515b1 100644 --- a/internal/clients/k8s/client.go +++ b/internal/clients/k8s/client.go @@ -202,6 +202,8 @@ const ( type AppsScale struct { Name string Replicas int32 + // PodSelector selects the workload's pods (from spec.selector); only set by scale-down. + PodSelector string } // RestoreLockInfo holds information about an active restore lock on a resource @@ -376,6 +378,16 @@ func scaleUpStatefulSetFromAnnotation(ctx context.Context, client kubernetes.Int return scaledTo, found, err } +// podSelectorString converts a workload's spec.selector into a pod label selector string. +// The restore label sits on the workload only, so pods must be found through spec.selector. +func podSelectorString(name string, selector *metav1.LabelSelector) (string, error) { + sel, err := metav1.LabelSelectorAsSelector(selector) + if err != nil { + return "", fmt.Errorf("invalid selector on %s: %w", name, err) + } + return sel.String(), nil +} + // ScaleDownDeployments scales down deployments matching a label selector to 0 replicas // Returns a list of deployment names and their original replica counts func (c *Client) ScaleDownDeployments(namespace, labelSelector string) ([]AppsScale, error) { @@ -391,13 +403,18 @@ func (c *Client) ScaleDownDeployments(namespace, labelSelector string) ([]AppsSc var scaledResources []AppsScale for _, dep := range deployments.Items { + podSelector, err := podSelectorString(dep.Name, dep.Spec.Selector) + if err != nil { + return scaledResources, err + } originalReplicas, err := scaleDownDeployment(ctx, c.clientset, namespace, dep.Name) if err != nil { return scaledResources, fmt.Errorf("failed to scale down deployment %s: %w", dep.Name, err) } scaledResources = append(scaledResources, AppsScale{ - Name: dep.Name, - Replicas: originalReplicas, + Name: dep.Name, + Replicas: originalReplicas, + PodSelector: podSelector, }) } @@ -454,13 +471,18 @@ func (c *Client) ScaleDownStatefulSets(namespace, labelSelector string) ([]AppsS var scaledResources []AppsScale for _, sts := range statefulSets.Items { + podSelector, err := podSelectorString(sts.Name, sts.Spec.Selector) + if err != nil { + return scaledResources, err + } originalReplicas, err := scaleDownStatefulSet(ctx, c.clientset, namespace, sts.Name) if err != nil { return scaledResources, fmt.Errorf("failed to scale down statefulset %s: %w", sts.Name, err) } scaledResources = append(scaledResources, AppsScale{ - Name: sts.Name, - Replicas: originalReplicas, + Name: sts.Name, + Replicas: originalReplicas, + PodSelector: podSelector, }) } diff --git a/internal/clients/k8s/client_test.go b/internal/clients/k8s/client_test.go index cefd882..9d9e105 100644 --- a/internal/clients/k8s/client_test.go +++ b/internal/clients/k8s/client_test.go @@ -965,3 +965,30 @@ func TestClient_SetAndClearRestoreLock_Integration(t *testing.T) { require.NoError(t, err) assert.Empty(t, locks) } + +func TestClient_ScaleDownReturnsPodSelector(t *testing.T) { + fakeClient := fake.NewSimpleClientset() + client := &Client{clientset: fakeClient} + ctx := context.Background() + meta := map[string]string{"restore": "true"} + selector := &metav1.LabelSelector{MatchLabels: map[string]string{"app": "x", "tier": "y"}} + + deploy := createDeployment("deploy", "test-ns", meta, 1) + deploy.Spec.Selector = selector + _, err := fakeClient.AppsV1().Deployments("test-ns").Create(ctx, &deploy, metav1.CreateOptions{}) + require.NoError(t, err) + sts := createStatefulSet("sts", "test-ns", meta, 1) + sts.Spec.Selector = selector + _, err = fakeClient.AppsV1().StatefulSets("test-ns").Create(ctx, &sts, metav1.CreateOptions{}) + require.NoError(t, err) + + deployScales, err := client.ScaleDownDeployments("test-ns", "restore=true") + require.NoError(t, err) + require.Len(t, deployScales, 1) + assert.Equal(t, "app=x,tier=y", deployScales[0].PodSelector) + + stsScales, err := client.ScaleDownStatefulSets("test-ns", "restore=true") + require.NoError(t, err) + require.Len(t, stsScales, 1) + assert.Equal(t, "app=x,tier=y", stsScales[0].PodSelector) +} diff --git a/internal/orchestration/scale/scale.go b/internal/orchestration/scale/scale.go index 107a9ae..edea8f1 100644 --- a/internal/orchestration/scale/scale.go +++ b/internal/orchestration/scale/scale.go @@ -3,9 +3,9 @@ package scale import ( "context" "fmt" + "slices" "time" - corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "github.com/stackvista/stackstate-backup-cli/internal/clients/k8s" @@ -19,7 +19,7 @@ const ( ) // ScaleDown scales down deployments matching the label selector and logs the results. -// It waits for all pods to terminate before returning. +// It waits until no pod selected by any scaled workload remains. // //nolint:revive // Package name "scale" with function "ScaleDown" is intentionally verbose for clarity func ScaleDown(k8sClient *k8s.Client, namespace, labelSelector string, log *logger.Logger) ([]k8s.AppsScale, error) { @@ -54,8 +54,16 @@ func ScaleDown(k8sClient *k8s.Client, namespace, labelSelector string, log *logg log.Infof(" - %s (replicas: %d -> 0)", dep.Name, dep.Replicas) } - // Wait for pods to terminate - if err := waitForPodsToTerminate(k8sClient, namespace, labelSelector, log); err != nil { + var podSelectors []string + for _, app := range scaledApps { + if app.PodSelector != "" { + podSelectors = append(podSelectors, app.PodSelector) + } + } + slices.Sort(podSelectors) + podSelectors = slices.Compact(podSelectors) + + if err := waitForPodsToTerminate(k8sClient, namespace, podSelectors, log); err != nil { return scaledApps, fmt.Errorf("failed waiting for pods to terminate: %w", err) } @@ -141,8 +149,8 @@ func ScaleUpAndReleaseLock(k8sClient *k8s.Client, namespace, labelSelector strin return nil } -// waitForPodsToTerminate polls for pod termination until all pods matching the label selector are gone -func waitForPodsToTerminate(k8sClient *k8s.Client, namespace, labelSelector string, log *logger.Logger) error { +// waitForPodsToTerminate polls until no pod matches any of the pod selectors +func waitForPodsToTerminate(k8sClient *k8s.Client, namespace string, podSelectors []string, log *logger.Logger) error { ctx := context.Background() ticker := time.NewTicker(podTerminationCheckInterval) defer ticker.Stop() @@ -156,29 +164,23 @@ func waitForPodsToTerminate(k8sClient *k8s.Client, namespace, labelSelector stri case <-timeout: return fmt.Errorf("timeout waiting for pods to terminate after %v", podTerminationTimeout) case <-ticker.C: - podList, err := k8sClient.Clientset().CoreV1().Pods(namespace).List(ctx, metav1.ListOptions{ - LabelSelector: labelSelector, - }) - if err != nil { - return fmt.Errorf("failed to list pods: %w", err) - } - - // Filter out pods in terminal states (Succeeded, Failed) or being deleted - activePods := 0 - for _, pod := range podList.Items { - if pod.Status.Phase != corev1.PodSucceeded && - pod.Status.Phase != corev1.PodFailed && - pod.DeletionTimestamp == nil { - activePods++ + total := 0 + for _, selector := range podSelectors { + podList, err := k8sClient.Clientset().CoreV1().Pods(namespace).List(ctx, metav1.ListOptions{ + LabelSelector: selector, + }) + if err != nil { + return fmt.Errorf("failed to list pods: %w", err) } + total += len(podList.Items) } - if activePods == 0 && len(podList.Items) == 0 { + if total == 0 { log.Successf("All pods have terminated") return nil } - log.Infof("Waiting for %d pod(s) to terminate...", activePods+len(podList.Items)-activePods) + log.Infof("Waiting for %d pod(s) to terminate...", total) } } } diff --git a/internal/orchestration/scale/scale_test.go b/internal/orchestration/scale/scale_test.go index 2ee83f8..81ad3b1 100644 --- a/internal/orchestration/scale/scale_test.go +++ b/internal/orchestration/scale/scale_test.go @@ -13,6 +13,7 @@ import ( "github.com/stretchr/testify/require" appsv1 "k8s.io/api/apps/v1" corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/client-go/kubernetes/fake" ) @@ -1153,3 +1154,52 @@ func TestScaleDownWithLock_FullCycle(t *testing.T) { require.NoError(t, err) assert.Len(t, scaledApps, 1) } + +// The restore label sits on the workload only; pods must be found through spec.selector. +func TestScaleDown_WaitsForPodsSelectedByWorkload(t *testing.T) { + podLabels := map[string]string{"app": "receiver"} + deploy := createDeployment("receiver", podLabels, 1) + deploy.Labels = map[string]string{"restore": "true"} + sts := createStatefulSet("receiver-sts", podLabels, 1) + sts.Labels = map[string]string{"restore": "true"} + sts.Spec.Selector = &metav1.LabelSelector{MatchLabels: map[string]string{"app": "receiver-sts"}} + + tests := []struct { + name string + kind string + pod string + }{ + {"deployment", "deployment", "receiver-0"}, + {"statefulset", "statefulset", "receiver-sts-0"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + fakeClient := fake.NewSimpleClientset() + client := k8s.NewTestClient(fakeClient) + ctx := context.Background() + podAppLabel := "receiver" + if tt.kind == "deployment" { + _, err := fakeClient.AppsV1().Deployments("test-ns").Create(ctx, &deploy, metav1.CreateOptions{}) + require.NoError(t, err) + } else { + podAppLabel = "receiver-sts" + _, err := fakeClient.AppsV1().StatefulSets("test-ns").Create(ctx, &sts, metav1.CreateOptions{}) + require.NoError(t, err) + } + pod := createPod(tt.pod, map[string]string{"app": podAppLabel}, corev1.PodRunning) + _, err := fakeClient.CoreV1().Pods("test-ns").Create(ctx, &pod, metav1.CreateOptions{}) + require.NoError(t, err) + + go func() { + time.Sleep(3 * time.Second) + _ = fakeClient.CoreV1().Pods("test-ns").Delete(ctx, tt.pod, metav1.DeleteOptions{}) + }() + + _, err = ScaleDown(client, "test-ns", "restore=true", logger.New(true, false)) + require.NoError(t, err) + + _, err = fakeClient.CoreV1().Pods("test-ns").Get(ctx, tt.pod, metav1.GetOptions{}) + assert.True(t, apierrors.IsNotFound(err), "ScaleDown returned while the pod still existed") + }) + } +} From f1c786b0570b9b6b213bf121ab8498022d599bb0 Mon Sep 17 00:00:00 2001 From: Vladimir Iliakov Date: Mon, 9 Nov 2026 19:03:25 +0100 Subject: [PATCH 2/6] Restore the sts_k8s_logs data stream with Elasticsearch restore Cleanup deleted the data stream's backing indices by name after a rollover, and the restore named only backing indices, so they came back as plain indices without a data stream (#126). - Delete the data stream instead of rolling it over; a 404 means only orphaned backing indices remain, which are deleted as plain indices. - Name the data stream in the restore request when the snapshot has it, and reject before scale-down a snapshot that has its backing indices without it. - List again after cleanup and fail before _restore if STS indices reappeared. Co-Authored-By: Claude Sonnet 5.5 --- cmd/elasticsearch/configure_test.go | 4 +- cmd/elasticsearch/list_test.go | 4 +- cmd/elasticsearch/restore.go | 132 ++++++-- cmd/elasticsearch/restore_test.go | 307 ++++++++++++++++-- internal/clients/elasticsearch/client.go | 22 +- internal/clients/elasticsearch/client_test.go | 41 ++- internal/clients/elasticsearch/interface.go | 2 +- .../config/testdata/validConfigMapConfig.yaml | 2 +- .../config/testdata/validConfigMapOnly.yaml | 2 +- .../testdata/validStorageConfigMapConfig.yaml | 2 +- .../testdata/validStorageConfigMapOnly.yaml | 2 +- 11 files changed, 440 insertions(+), 80 deletions(-) diff --git a/cmd/elasticsearch/configure_test.go b/cmd/elasticsearch/configure_test.go index 396209c..7acbaad 100644 --- a/cmd/elasticsearch/configure_test.go +++ b/cmd/elasticsearch/configure_test.go @@ -96,8 +96,8 @@ func (m *mockESClientForConfigure) RestoreSnapshot(_, _, _ string, _ bool) error return fmt.Errorf("not implemented") } -func (m *mockESClientForConfigure) RolloverDatastream(_ string) error { - return fmt.Errorf("not implemented") +func (m *mockESClientForConfigure) DeleteDatastream(_ string) (bool, error) { + return false, fmt.Errorf("not implemented") } // TestConfigureCmd_Unit tests the command structure diff --git a/cmd/elasticsearch/list_test.go b/cmd/elasticsearch/list_test.go index 3a7b45e..375239a 100644 --- a/cmd/elasticsearch/list_test.go +++ b/cmd/elasticsearch/list_test.go @@ -273,8 +273,8 @@ func (m *mockESClient) ConfigureSLMPolicy(_, _, _, _, _, _ string, _, _ int) err return fmt.Errorf("not implemented") } -func (m *mockESClient) RolloverDatastream(_ string) error { - return fmt.Errorf("not implemented") +func (m *mockESClient) DeleteDatastream(_ string) (bool, error) { + return false, fmt.Errorf("not implemented") } // createFakeClientWithConfig creates a fake Kubernetes client with a ConfigMap containing the given config diff --git a/cmd/elasticsearch/restore.go b/cmd/elasticsearch/restore.go index 4c25f98..5be999d 100644 --- a/cmd/elasticsearch/restore.go +++ b/cmd/elasticsearch/restore.go @@ -3,6 +3,7 @@ package elasticsearch import ( "fmt" "path" + "slices" "sort" "strings" "time" @@ -39,7 +40,7 @@ func restoreCmd(globalFlags *config.CLIGlobalFlags) *cobra.Command { cmd := &cobra.Command{ Use: "restore", Short: "Restore Elasticsearch from a snapshot", - Long: `Restore Elasticsearch indices from a snapshot. Deletes existing STS indices before restore. Waits for completion by default; use --background to run asynchronously.`, + Long: `Restore Elasticsearch indices and the logs data stream from a snapshot. Deletes existing STS indices and the logs data stream before restore. Waits for completion by default; use --background to run asynchronously.`, Run: func(_ *cobra.Command, _ []string) { cmdutils.Run(globalFlags, runRestore, cmdutils.StorageIsRequired) }} @@ -104,6 +105,11 @@ func runRestore(appCtx *app.Context) error { return err } + targets, err := restoreTargets(snapshotDetails, restoreIndices, appCtx.Config.Elasticsearch.Restore) + if err != nil { + return err + } + // Confirm with user before starting destructive operation if !skipConfirmation { appCtx.Logger.Println() @@ -135,27 +141,34 @@ func runRestore(appCtx *app.Context) error { return err } - // Delete all STS indices before restore + if err := replaceSTSIndices(esClient, appCtx, repository, snapshotDetails, targets); err != nil { + return err + } + + if runBackground { + restore.PrintAPIRunningRestoreStatus("elasticsearch", selectedSnapshot, appCtx.Namespace, appCtx.Logger) + return nil + } + + return checkAndFinalize(esClient, appCtx, repository, selectedSnapshot, !runBackground, restoreNoProgressIn) +} + +// replaceSTSIndices deletes the target's STS indices and triggers an asynchronous restore of the snapshot's. +func replaceSTSIndices(esClient es.Interface, appCtx *app.Context, repository string, snapshot *es.Snapshot, targets []string) error { appCtx.Logger.Println() if err := deleteAllSTSIndices(esClient, appCtx); err != nil { return err } - // Trigger async restore appCtx.Logger.Println() - isPartial := snapshotDetails.State == "PARTIAL" - appCtx.Logger.Infof("Triggering restore for snapshot: %s", selectedSnapshot) - if err := esClient.RestoreSnapshot(repository, selectedSnapshot, strings.Join(restoreIndices, ","), isPartial); err != nil { + isPartial := snapshot.State == "PARTIAL" + appCtx.Logger.Infof("Triggering restore for snapshot: %s", snapshot.Snapshot) + if err := esClient.RestoreSnapshot(repository, snapshot.Snapshot, strings.Join(targets, ","), isPartial); err != nil { return fmt.Errorf("failed to trigger restore: %w", err) } appCtx.Logger.Successf("Restore triggered successfully") - if runBackground { - restore.PrintAPIRunningRestoreStatus("elasticsearch", selectedSnapshot, appCtx.Namespace, appCtx.Logger) - return nil - } - - return checkAndFinalize(esClient, appCtx, repository, selectedSnapshot, !runBackground, restoreNoProgressIn) + return nil } // getLatestSnapshot retrieves the most recent snapshot from the repository @@ -210,50 +223,99 @@ func validateSnapshotState(snapshot *es.Snapshot, appCtx *app.Context, skipConfi } } -// deleteAllSTSIndices deletes all STS indices including datastream rollover if needed +// deleteAllSTSIndices deletes all STS indices, including the logs data stream and any of its +// backing indices left without one. func deleteAllSTSIndices(esClient es.Interface, appCtx *app.Context) error { - appCtx.Logger.Infof("Fetching current Elasticsearch indices...") - allIndices, err := esClient.ListIndices("*") - if err != nil { - return fmt.Errorf("failed to list indices: %w", err) - } + restoreConfig := appCtx.Config.Elasticsearch.Restore - stsIndices := filterSTSIndices(allIndices, appCtx.Config.Elasticsearch.Restore.IndexPrefix, appCtx.Config.Elasticsearch.Restore.DatastreamIndexPrefix) - stsIndices, err = filterIndicesByPattern(stsIndices, appCtx.Config.Elasticsearch.Restore.IndicesPattern) + appCtx.Logger.Infof("Fetching current Elasticsearch indices...") + stsIndices, err := listSTSIndices(esClient, restoreConfig) if err != nil { - return fmt.Errorf("invalid Elasticsearch restore indicesPattern: %w", err) + return err } if len(stsIndices) == 0 { appCtx.Logger.Infof("No STS indices found to delete") - return nil - } - - appCtx.Logger.Infof("Found %d STS index(es) to delete", len(stsIndices)) - for _, index := range stsIndices { - appCtx.Logger.Debugf(" - %s", index) + } else { + appCtx.Logger.Infof("Found %d STS index(es) to delete", len(stsIndices)) + for _, index := range stsIndices { + appCtx.Logger.Debugf(" - %s", index) + } } - // Check for datastream and rollover if needed - if hasDatastreamIndices(stsIndices, appCtx.Config.Elasticsearch.Restore.DatastreamIndexPrefix) { - appCtx.Logger.Infof("Rolling over datastream '%s'...", appCtx.Config.Elasticsearch.Restore.DatastreamName) - if err := esClient.RolloverDatastream(appCtx.Config.Elasticsearch.Restore.DatastreamName); err != nil { - return fmt.Errorf("failed to rollover datastream: %w", err) + // Rolling over instead would leave a new write index behind that can collide with the snapshot. + if hasDatastreamIndices(stsIndices, restoreConfig.DatastreamIndexPrefix) { + appCtx.Logger.Infof("Deleting datastream '%s'...", restoreConfig.DatastreamName) + deleted, err := esClient.DeleteDatastream(restoreConfig.DatastreamName) + if err != nil { + return fmt.Errorf("failed to delete datastream: %w", err) + } + if deleted { + appCtx.Logger.Successf("Datastream deleted successfully") + if stsIndices, err = listSTSIndices(esClient, restoreConfig); err != nil { + return err + } + } else { + appCtx.Logger.Infof("Datastream '%s' does not exist; deleting its backing indices directly", restoreConfig.DatastreamName) } - appCtx.Logger.Successf("Datastream rolled over successfully") } - // Delete all indices - appCtx.Logger.Infof("Deleting %d index(es)...", len(stsIndices)) + if len(stsIndices) > 0 { + appCtx.Logger.Infof("Deleting %d index(es)...", len(stsIndices)) + } for _, index := range stsIndices { if err := deleteIndexWithVerification(esClient, index, appCtx.Logger); err != nil { return err } } + + // A workload that is still writing recreates indices, which would restore into a dirty target. + remaining, err := listSTSIndices(esClient, restoreConfig) + if err != nil { + return err + } + if len(remaining) > 0 { + return fmt.Errorf("STS indices reappeared after deletion, so a workload is still writing to Elasticsearch: %s. "+ + "Check for writers not matched by scaleDownLabelSelector %q. "+ + "Workloads are still scaled down; after fixing this, release them with: sts-backup elasticsearch check-and-finalize --finalize-only -n %s", + strings.Join(remaining, ", "), restoreConfig.ScaleDownLabelSelector, appCtx.Namespace) + } appCtx.Logger.Successf("All indices deleted successfully") return nil } +func listSTSIndices(esClient es.Interface, restoreConfig config.RestoreConfig) ([]string, error) { + allIndices, err := esClient.ListIndices("*") + if err != nil { + return nil, fmt.Errorf("failed to list indices: %w", err) + } + + stsIndices := filterSTSIndices(allIndices, restoreConfig.IndexPrefix, restoreConfig.DatastreamIndexPrefix) + stsIndices, err = filterIndicesByPattern(stsIndices, restoreConfig.IndicesPattern) + if err != nil { + return nil, fmt.Errorf("invalid Elasticsearch restore indicesPattern: %w", err) + } + + return stsIndices, nil +} + +// restoreTargets adds the logs data stream to the restore request. Naming only its backing indices +// restores them as plain indices and leaves the data stream missing. Naming a data stream the +// snapshot lacks fails the whole restore, so that case is rejected up front. +func restoreTargets(snapshot *es.Snapshot, indices []string, restoreConfig config.RestoreConfig) ([]string, error) { + if !hasDatastreamIndices(indices, restoreConfig.DatastreamIndexPrefix) { + return indices, nil + } + + if !slices.Contains(snapshot.DataStreams, restoreConfig.DatastreamName) { + return nil, fmt.Errorf("snapshot %s contains %s backing indices but not the %s data stream, "+ + "so the logs cannot be restored as a data stream; use a snapshot taken by the backup SLM policy", + snapshot.Snapshot, restoreConfig.DatastreamIndexPrefix, restoreConfig.DatastreamName) + } + + return append(slices.Clone(indices), restoreConfig.DatastreamName), nil +} + // filterSTSIndices filters indices that match the configured STS prefixes func filterSTSIndices(allIndices []string, indexPrefix, datastreamPrefix string) []string { var stsIndices []string diff --git a/cmd/elasticsearch/restore_test.go b/cmd/elasticsearch/restore_test.go index 33531e5..62886bc 100644 --- a/cmd/elasticsearch/restore_test.go +++ b/cmd/elasticsearch/restore_test.go @@ -2,6 +2,8 @@ package elasticsearch import ( "fmt" + "slices" + "strings" "testing" "time" @@ -21,13 +23,30 @@ type mockESClientForRestore struct { indexExistsMap map[string]bool restoreErr error getSnapshotErr error - rolloverErr error deletedIndices []string restoredSnapshot string - rolledOverDS string + restoredTargets string + restoredPartial bool + // datastreams maps a live data stream to its backing indices. + datastreams map[string][]string + datastreamDeleteCalls []string + deleteDatastreamErr error + // listCalls counts ListIndices calls; addOnListCall adds indices on the given call, modelling a writer. + listCalls int + addOnListCall map[int][]string } func (m *mockESClientForRestore) ListIndices(_ string) ([]string, error) { + m.listCalls++ + if added := m.addOnListCall[m.listCalls]; len(added) > 0 { + if m.indexExistsMap == nil { + m.indexExistsMap = make(map[string]bool) + } + for _, index := range added { + m.indices = append(m.indices, index) + m.indexExistsMap[index] = true + } + } return m.indices, nil } @@ -42,7 +61,13 @@ func (m *mockESClientForRestore) DeleteIndex(index string) error { if m.deleteErr != nil { return m.deleteErr } + for _, backing := range m.datastreams { + if len(backing) > 0 && backing[len(backing)-1] == index { + return fmt.Errorf("index [%s] is the write index for data stream and cannot be deleted", index) + } + } m.deletedIndices = append(m.deletedIndices, index) + m.indices = slices.DeleteFunc(m.indices, func(i string) bool { return i == index }) // Update exists map if m.indexExistsMap == nil { m.indexExistsMap = make(map[string]bool) @@ -62,20 +87,43 @@ func (m *mockESClientForRestore) IndexExists(index string) (bool, error) { return exists, nil } -func (m *mockESClientForRestore) RestoreSnapshot(_, snapshotName, _ string, _ bool) error { +// RestoreSnapshot fails like Elasticsearch when a requested snapshot index is still open on the target. +func (m *mockESClientForRestore) RestoreSnapshot(_, snapshotName, targets string, partial bool) error { if m.restoreErr != nil { return m.restoreErr } + for _, target := range strings.Split(targets, ",") { + if m.snapshot != nil && !slices.Contains(m.snapshot.Indices, target) && !slices.Contains(m.snapshot.DataStreams, target) { + return fmt.Errorf("index_not_found_exception: no such index [%s]", target) + } + if slices.Contains(m.indices, target) { + return fmt.Errorf("cannot restore index [%s] because an open index with same name already exists", target) + } + } m.restoredSnapshot = snapshotName + m.restoredTargets = targets + m.restoredPartial = partial return nil } -func (m *mockESClientForRestore) RolloverDatastream(datastreamName string) error { - if m.rolloverErr != nil { - return m.rolloverErr +func (m *mockESClientForRestore) DeleteDatastream(datastreamName string) (bool, error) { + m.datastreamDeleteCalls = append(m.datastreamDeleteCalls, datastreamName) + if m.deleteDatastreamErr != nil { + return false, m.deleteDatastreamErr } - m.rolledOverDS = datastreamName - return nil + backing, ok := m.datastreams[datastreamName] + if !ok { + return false, nil + } + m.indices = slices.DeleteFunc(m.indices, func(i string) bool { return slices.Contains(backing, i) }) + if m.indexExistsMap == nil { + m.indexExistsMap = make(map[string]bool) + } + for _, index := range backing { + m.indexExistsMap[index] = false + } + delete(m.datastreams, datastreamName) + return true, nil } func (m *mockESClientForRestore) ListSnapshots(_ string) ([]elasticsearch.Snapshot, error) { @@ -295,6 +343,229 @@ func TestDeleteAllSTSIndicesHonorsIndicesPattern(t *testing.T) { assert.Equal(t, []string{"sts_topology"}, client.deletedIndices) } +func newLogsRestoreContext() *app.Context { + return &app.Context{ + Config: &config.Config{ + Elasticsearch: config.ElasticsearchConfig{ + Restore: config.RestoreConfig{ + IndexPrefix: "sts", + DatastreamIndexPrefix: ".ds-sts_k8s_logs", + DatastreamName: "sts_k8s_logs", + IndicesPattern: "sts*,.ds-sts_k8s_logs*", + }, + }, + }, + Logger: logger.New(true, false), + } +} + +//nolint:funlen +func TestReplaceSTSIndicesRestoresLogsDatastream(t *testing.T) { + const ( + topo = "sts_topology_events-2026.11.01" + b1 = ".ds-sts_k8s_logs-2026.11.01-000001" + b2 = ".ds-sts_k8s_logs-2026.11.01-000002" + b3 = ".ds-sts_k8s_logs-2026.11.09-000003" + full = topo + "," + b1 + "," + b2 + ",sts_k8s_logs" + ) + dsSnapshot := "sts_k8s_logs" + + tests := []struct { + name string + pattern string + targetIndices []string + targetDatastreams map[string][]string + addOnListCall map[int][]string + deleteDatastreamErr error + snapshotState string + expectDSCalls []string + expectDeleted []string + expectTargets string + expectRemaining []string + expectListCalls int + expectPartial bool + expectErr string + }{ + { + name: "T1 no STS indices", + targetIndices: []string{}, + expectTargets: full, + expectRemaining: []string{}, + expectListCalls: 2, + }, + { + name: "T2 plain indices only", + targetIndices: []string{topo, "other"}, + expectDeleted: []string{topo}, + expectTargets: full, + expectRemaining: []string{"other"}, + }, + { + name: "T3 live data stream write index collides", + targetIndices: []string{topo, b2}, + targetDatastreams: map[string][]string{dsSnapshot: {b2}}, + expectDSCalls: []string{dsSnapshot}, + expectDeleted: []string{topo}, + expectTargets: full, + expectRemaining: []string{}, + expectListCalls: 3, + }, + { + name: "T4 orphaned backing indices, no data stream", + targetIndices: []string{topo, b1}, + expectDSCalls: []string{dsSnapshot}, + expectDeleted: []string{topo, b1}, + expectTargets: full, + }, + { + name: "T5 data stream plus orphans", + targetIndices: []string{topo, b1, b3}, + targetDatastreams: map[string][]string{dsSnapshot: {b3}}, + expectDSCalls: []string{dsSnapshot}, + expectDeleted: []string{topo, b1}, + expectTargets: full, + expectRemaining: []string{}, + }, + { + name: "T7 pattern excludes logs", + pattern: "sts_topology*", + targetIndices: []string{topo, b2}, + targetDatastreams: map[string][]string{dsSnapshot: {b2}}, + expectDeleted: []string{topo}, + expectTargets: topo, + expectRemaining: []string{b2}, + }, + { + name: "T8 writer recreates the data stream", + targetIndices: []string{topo}, + addOnListCall: map[int][]string{2: {".ds-sts_k8s_logs-2026.11.09-000001"}}, + expectDeleted: []string{topo}, + expectErr: "reappeared after deletion", + }, + { + name: "T9 data stream delete fails", + targetIndices: []string{b1}, + deleteDatastreamErr: fmt.Errorf("boom"), + expectDSCalls: []string{dsSnapshot}, + expectErr: "failed to delete datastream", + expectRemaining: []string{b1}, + }, + { + name: "T10 partial is passed through", + targetIndices: []string{topo, b2}, + targetDatastreams: map[string][]string{dsSnapshot: {b2}}, + snapshotState: "PARTIAL", + expectDSCalls: []string{dsSnapshot}, + expectDeleted: []string{topo}, + expectTargets: full, + expectRemaining: []string{}, + expectPartial: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + state := tt.snapshotState + if state == "" { + state = elasticsearch.StatusSuccess + } + snapshot := &elasticsearch.Snapshot{ + Snapshot: "sts-backup-20261101", + State: state, + Indices: []string{topo, b1, b2}, + DataStreams: []string{dsSnapshot}, + } + client := &mockESClientForRestore{ + indices: slices.Clone(tt.targetIndices), + snapshot: snapshot, + datastreams: tt.targetDatastreams, + addOnListCall: tt.addOnListCall, + deleteDatastreamErr: tt.deleteDatastreamErr, + indexExistsMap: map[string]bool{}, + } + for _, index := range tt.targetIndices { + client.indexExistsMap[index] = true + } + appCtx := newLogsRestoreContext() + if tt.pattern != "" { + appCtx.Config.Elasticsearch.Restore.IndicesPattern = tt.pattern + } + restoreIndices, err := restorableSnapshotIndices(snapshot, appCtx.Config.Elasticsearch.Restore) + require.NoError(t, err) + targets, err := restoreTargets(snapshot, restoreIndices, appCtx.Config.Elasticsearch.Restore) + require.NoError(t, err) + + err = replaceSTSIndices(client, appCtx, "backup-repo", snapshot, targets) + + assert.Equal(t, tt.expectDSCalls, client.datastreamDeleteCalls) + assert.Equal(t, tt.expectDeleted, client.deletedIndices) + if tt.expectErr != "" { + require.ErrorContains(t, err, tt.expectErr) + assert.Empty(t, client.restoredTargets, "restore must not be sent") + if tt.name == "T8 writer recreates the data stream" { + assert.ErrorContains(t, err, ".ds-sts_k8s_logs-2026.11.09-000001") + } + if tt.expectRemaining != nil { + assert.Equal(t, tt.expectRemaining, client.indices) + } + return + } + require.NoError(t, err) + assert.Equal(t, tt.expectTargets, client.restoredTargets) + assert.Equal(t, tt.expectPartial, client.restoredPartial) + if tt.expectRemaining != nil { + assert.Equal(t, tt.expectRemaining, client.indices) + } + if tt.expectListCalls != 0 { + assert.Equal(t, tt.expectListCalls, client.listCalls) + } + }) + } +} + +func TestRestoreTargetsRejectsSnapshotWithoutDatastream(t *testing.T) { + restoreConfig := newLogsRestoreContext().Config.Elasticsearch.Restore + const topo = "sts_topology_events-2026.11.01" + + _, err := restoreTargets(&elasticsearch.Snapshot{Snapshot: "snap"}, + []string{topo, ".ds-sts_k8s_logs-2026.11.01-000001"}, restoreConfig) + require.ErrorContains(t, err, "not the sts_k8s_logs data stream") + + targets, err := restoreTargets(&elasticsearch.Snapshot{Snapshot: "snap"}, []string{topo}, restoreConfig) + require.NoError(t, err) + assert.Equal(t, []string{topo}, targets) +} + +func TestRestoreTargetsKeepCompletionExpectation(t *testing.T) { + const ( + topo = "sts_topology_events-2026.11.01" + b1 = ".ds-sts_k8s_logs-2026.11.01-000001" + b2 = ".ds-sts_k8s_logs-2026.11.01-000002" + ) + appCtx := newLogsRestoreContext() + restoreConfig := appCtx.Config.Elasticsearch.Restore + snapshot := &elasticsearch.Snapshot{ + Snapshot: "snap", + Indices: []string{topo, b1, b2}, + DataStreams: []string{"sts_k8s_logs"}, + } + + expected, err := restorableSnapshotIndices(snapshot, restoreConfig) + require.NoError(t, err) + targets, err := restoreTargets(snapshot, expected, restoreConfig) + require.NoError(t, err) + fromCheck, err := expectedRestoredIndices(&snapshotOnlyClient{snapshot: snapshot}, appCtx, "repo", "snap") + require.NoError(t, err) + + assert.Equal(t, expected, fromCheck) + assert.NotContains(t, expected, "sts_k8s_logs") + assert.Equal(t, append(slices.Clone(expected), "sts_k8s_logs"), targets) + + restored := elasticsearch.IndexHealth{NumberOfShards: 1, ActivePrimaryShards: 1} + health := map[string]elasticsearch.IndexHealth{topo: restored, b1: restored, b2: restored} + assert.True(t, measureRestore(expected, restoreConfig.DatastreamIndexPrefix, health).complete()) +} + // TestHasDatastreamIndices tests datastream detection func TestHasDatastreamIndices(t *testing.T) { tests := []struct { @@ -357,10 +628,8 @@ func TestMockESClientForRestore(t *testing.T) { indicesToDelete []string deleteErr error restoreErr error - rolloverErr error expectDeletedCount int expectRestoreOK bool - expectRolloverOK bool }{ { name: "successful restore with index deletion", @@ -398,14 +667,6 @@ func TestMockESClientForRestore(t *testing.T) { expectDeletedCount: 0, expectRestoreOK: false, }, - { - name: "successful rollover", - initialIndices: []string{"sts_k8s_logs-000001"}, - indicesToDelete: []string{}, - rolloverErr: nil, - expectDeletedCount: 0, - expectRolloverOK: true, - }, } for _, tt := range tests { @@ -414,7 +675,6 @@ func TestMockESClientForRestore(t *testing.T) { indices: tt.initialIndices, deleteErr: tt.deleteErr, restoreErr: tt.restoreErr, - rolloverErr: tt.rolloverErr, indexExistsMap: make(map[string]bool), snapshot: &elasticsearch.Snapshot{ Snapshot: "test-snapshot", @@ -441,15 +701,8 @@ func TestMockESClientForRestore(t *testing.T) { assert.Equal(t, tt.expectDeletedCount, len(mockClient.deletedIndices)) - // Test rollover if applicable - if tt.expectRolloverOK { - err := mockClient.RolloverDatastream("sts_k8s_logs") - assert.NoError(t, err) - assert.Equal(t, "sts_k8s_logs", mockClient.rolledOverDS) - } - // Test restore - err := mockClient.RestoreSnapshot("backup-repo", "test-snapshot", "sts_*", false) + err := mockClient.RestoreSnapshot("backup-repo", "test-snapshot", "sts_topology,sts_metrics", false) if tt.expectRestoreOK { assert.NoError(t, err) assert.Equal(t, "test-snapshot", mockClient.restoredSnapshot) diff --git a/internal/clients/elasticsearch/client.go b/internal/clients/elasticsearch/client.go index f18aed7..95543d5 100644 --- a/internal/clients/elasticsearch/client.go +++ b/internal/clients/elasticsearch/client.go @@ -60,6 +60,7 @@ type Snapshot struct { EndTimeMillis int64 `json:"end_time_in_millis"` DurationInMillis int64 `json:"duration_in_millis"` Indices []string `json:"indices"` + DataStreams []string `json:"data_streams"` Failures []SnapshotFailure `json:"failures"` Shards struct { Total int `json:"total"` @@ -240,22 +241,27 @@ func (c *Client) IndexExists(index string) (bool, error) { return true, nil } -// RolloverDatastream performs a rollover on a datastream -func (c *Client) RolloverDatastream(datastreamName string) error { - res, err := c.es.Indices.Rollover( - datastreamName, - c.es.Indices.Rollover.WithContext(context.Background()), +// DeleteDatastream deletes a data stream together with all of its backing indices. +// It reports false when the data stream does not exist. +func (c *Client) DeleteDatastream(datastreamName string) (bool, error) { + res, err := c.es.Indices.DeleteDataStream( + []string{datastreamName}, + c.es.Indices.DeleteDataStream.WithContext(context.Background()), ) if err != nil { - return fmt.Errorf("failed to rollover datastream: %w", err) + return false, fmt.Errorf("failed to delete datastream: %w", err) } defer res.Body.Close() + if res.StatusCode == http.StatusNotFound { + return false, nil + } + if res.IsError() { - return fmt.Errorf("elasticsearch returned error: %s", res.String()) + return false, fmt.Errorf("elasticsearch returned error: %s", res.String()) } - return nil + return true, nil } // DeleteSnapshotRepository deletes a snapshot repository diff --git a/internal/clients/elasticsearch/client_test.go b/internal/clients/elasticsearch/client_test.go index 5392aab..d5c586a 100644 --- a/internal/clients/elasticsearch/client_test.go +++ b/internal/clients/elasticsearch/client_test.go @@ -134,7 +134,8 @@ func TestClient_GetSnapshot(t *testing.T) { "uuid": "uuid-1", "repository": "test-repo", "state": "SUCCESS", - "indices": ["index-1", "index-2"] + "indices": ["index-1", "index-2"], + "data_streams": ["sts_k8s_logs"] } ] }`, @@ -181,6 +182,7 @@ func TestClient_GetSnapshot(t *testing.T) { assert.NotNil(t, snapshot) assert.Equal(t, tt.snapshotName, snapshot.Snapshot) assert.Equal(t, tt.repository, snapshot.Repository) + assert.Equal(t, []string{"sts_k8s_logs"}, snapshot.DataStreams) }) } } @@ -309,6 +311,43 @@ func TestClient_DeleteIndex(t *testing.T) { } } +func TestClient_DeleteDatastream(t *testing.T) { + tests := []struct { + name string + responseStatus int + expectDeleted bool + expectError bool + }{ + {name: "deleted", responseStatus: http.StatusOK, expectDeleted: true}, + {name: "not found", responseStatus: http.StatusNotFound, expectDeleted: false}, + {name: "elasticsearch error", responseStatus: http.StatusInternalServerError, expectError: true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + server := mockESServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + assert.Equal(t, "/_data_stream/sts_k8s_logs", r.URL.Path) + assert.Equal(t, http.MethodDelete, r.Method) + + w.WriteHeader(tt.responseStatus) + })) + defer server.Close() + + client, err := NewClient(server.URL) + require.NoError(t, err) + + deleted, err := client.DeleteDatastream("sts_k8s_logs") + + if tt.expectError { + assert.Error(t, err) + return + } + require.NoError(t, err) + assert.Equal(t, tt.expectDeleted, deleted) + }) + } +} + func TestClient_IndexExists(t *testing.T) { tests := []struct { name string diff --git a/internal/clients/elasticsearch/interface.go b/internal/clients/elasticsearch/interface.go index f8e0089..40758e6 100644 --- a/internal/clients/elasticsearch/interface.go +++ b/internal/clients/elasticsearch/interface.go @@ -16,7 +16,7 @@ type Interface interface { IndexExists(index string) (bool, error) // Datastream operations - RolloverDatastream(datastreamName string) error + DeleteDatastream(datastreamName string) (bool, error) // Repository and SLM operations DeleteSnapshotRepository(name string) error diff --git a/internal/foundation/config/testdata/validConfigMapConfig.yaml b/internal/foundation/config/testdata/validConfigMapConfig.yaml index 4308734..c58b2ee 100644 --- a/internal/foundation/config/testdata/validConfigMapConfig.yaml +++ b/internal/foundation/config/testdata/validConfigMapConfig.yaml @@ -54,7 +54,7 @@ elasticsearch: indexPrefix: sts # Prefix for datastream indices (datastreams use pattern: .ds-{name}-{generation}) datastreamIndexPrefix: .ds-sts_k8s_logs - # Name of the datastream (used for rollover operations) + # Name of the datastream (deleted and restored together with its backing indices) datastreamName: sts_k8s_logs # Pattern for indices to restore from snapshot (comma-separated glob patterns) indicesPattern: sts*,.ds-sts_k8s_logs* diff --git a/internal/foundation/config/testdata/validConfigMapOnly.yaml b/internal/foundation/config/testdata/validConfigMapOnly.yaml index 3003119..072090a 100644 --- a/internal/foundation/config/testdata/validConfigMapOnly.yaml +++ b/internal/foundation/config/testdata/validConfigMapOnly.yaml @@ -61,7 +61,7 @@ elasticsearch: indexPrefix: sts # Prefix for datastream indices (datastreams use pattern: .ds-{name}-{generation}) datastreamIndexPrefix: .ds-sts_k8s_logs - # Name of the datastream (used for rollover operations) + # Name of the datastream (deleted and restored together with its backing indices) datastreamName: sts_k8s_logs # Pattern for indices to restore from snapshot (comma-separated glob patterns) indicesPattern: sts*,.ds-sts_k8s_logs* diff --git a/internal/foundation/config/testdata/validStorageConfigMapConfig.yaml b/internal/foundation/config/testdata/validStorageConfigMapConfig.yaml index dc2f72e..5db8859 100644 --- a/internal/foundation/config/testdata/validStorageConfigMapConfig.yaml +++ b/internal/foundation/config/testdata/validStorageConfigMapConfig.yaml @@ -56,7 +56,7 @@ elasticsearch: indexPrefix: sts # Prefix for datastream indices (datastreams use pattern: .ds-{name}-{generation}) datastreamIndexPrefix: .ds-sts_k8s_logs - # Name of the datastream (used for rollover operations) + # Name of the datastream (deleted and restored together with its backing indices) datastreamName: sts_k8s_logs # Pattern for indices to restore from snapshot (comma-separated glob patterns) indicesPattern: sts*,.ds-sts_k8s_logs* diff --git a/internal/foundation/config/testdata/validStorageConfigMapOnly.yaml b/internal/foundation/config/testdata/validStorageConfigMapOnly.yaml index ae82bcf..b8f3c12 100644 --- a/internal/foundation/config/testdata/validStorageConfigMapOnly.yaml +++ b/internal/foundation/config/testdata/validStorageConfigMapOnly.yaml @@ -63,7 +63,7 @@ elasticsearch: indexPrefix: sts # Prefix for datastream indices (datastreams use pattern: .ds-{name}-{generation}) datastreamIndexPrefix: .ds-sts_k8s_logs - # Name of the datastream (used for rollover operations) + # Name of the datastream (deleted and restored together with its backing indices) datastreamName: sts_k8s_logs # Pattern for indices to restore from snapshot (comma-separated glob patterns) indicesPattern: sts*,.ds-sts_k8s_logs* From 70ef3c4630415b9a0910b6d86d5417bb64874cd7 Mon Sep 17 00:00:00 2001 From: Vladimir Iliakov Date: Tue, 10 Nov 2026 09:27:59 +0100 Subject: [PATCH 3/6] Ignore finished pods in the scale-down wait and keep recorded replicas on retry An evicted pod stays in Failed phase after scale-to-0, so the wait counted it and timed out. The lock was then released with the workload still at 0, and a retry overwrote pre-restore-replicas with 0, so scale-up restored 0 replicas. Count only pods that are not Succeeded or Failed, and do not overwrite an existing annotation when the workload is already at 0. Co-Authored-By: Claude Sonnet 5.5 --- internal/clients/k8s/client.go | 23 ++++++ internal/orchestration/scale/scale.go | 8 +- internal/orchestration/scale/scale_test.go | 89 ++++++++++++++++++---- 3 files changed, 106 insertions(+), 14 deletions(-) diff --git a/internal/clients/k8s/client.go b/internal/clients/k8s/client.go index 3f515b1..fd05b03 100644 --- a/internal/clients/k8s/client.go +++ b/internal/clients/k8s/client.go @@ -10,6 +10,7 @@ import ( "net/url" "os" "path" + "strconv" "time" appsv1 "k8s.io/api/apps/v1" @@ -276,6 +277,10 @@ func scaleDownDeployment(ctx context.Context, client kubernetes.Interface, names if dep.Annotations == nil { dep.Annotations = make(map[string]string) } + if recorded, ok := recordedReplicas(dep.Annotations, originalReplicas); ok { + originalReplicas = recorded + return nil + } dep.Annotations[PreRestoreReplicasAnnotation] = fmt.Sprintf("%d", originalReplicas) zero := int32(0) dep.Spec.Replicas = &zero @@ -299,6 +304,10 @@ func scaleDownStatefulSet(ctx context.Context, client kubernetes.Interface, name if sts.Annotations == nil { sts.Annotations = make(map[string]string) } + if recorded, ok := recordedReplicas(sts.Annotations, originalReplicas); ok { + originalReplicas = recorded + return nil + } sts.Annotations[PreRestoreReplicasAnnotation] = fmt.Sprintf("%d", originalReplicas) zero := int32(0) sts.Spec.Replicas = &zero @@ -388,6 +397,20 @@ func podSelectorString(name string, selector *metav1.LabelSelector) (string, err return sel.String(), nil } +// recordedReplicas returns the replica count saved by an earlier scale-down when the workload is +// already at 0. A retry after a failed scale-down must not overwrite it with 0, or scale-up +// would restore 0 replicas. +func recordedReplicas(annotations map[string]string, currentReplicas int32) (int32, bool) { + if currentReplicas != 0 { + return 0, false + } + recorded, err := strconv.ParseInt(annotations[PreRestoreReplicasAnnotation], 10, 32) + if err != nil { + return 0, false + } + return int32(recorded), true +} + // ScaleDownDeployments scales down deployments matching a label selector to 0 replicas // Returns a list of deployment names and their original replica counts func (c *Client) ScaleDownDeployments(namespace, labelSelector string) ([]AppsScale, error) { diff --git a/internal/orchestration/scale/scale.go b/internal/orchestration/scale/scale.go index edea8f1..712177f 100644 --- a/internal/orchestration/scale/scale.go +++ b/internal/orchestration/scale/scale.go @@ -6,6 +6,7 @@ import ( "slices" "time" + corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "github.com/stackvista/stackstate-backup-cli/internal/clients/k8s" @@ -172,7 +173,12 @@ func waitForPodsToTerminate(k8sClient *k8s.Client, namespace string, podSelector if err != nil { return fmt.Errorf("failed to list pods: %w", err) } - total += len(podList.Items) + for _, pod := range podList.Items { + // Finished pods (e.g. evicted) linger after scale-to-0 and never go away on their own. + if pod.Status.Phase != corev1.PodSucceeded && pod.Status.Phase != corev1.PodFailed { + total++ + } + } } if total == 0 { diff --git a/internal/orchestration/scale/scale_test.go b/internal/orchestration/scale/scale_test.go index 81ad3b1..3725da2 100644 --- a/internal/orchestration/scale/scale_test.go +++ b/internal/orchestration/scale/scale_test.go @@ -123,7 +123,7 @@ func TestScaleDown_PodsInTerminalState(t *testing.T) { ) require.NoError(t, err) - // Create pods in terminal states (should be ignored by waitForPodsToTerminate) + // Create pods in terminal states (ignored by waitForPodsToTerminate) pod1 := createPod("pod1", map[string]string{"app": "test"}, corev1.PodSucceeded) pod2 := createPod("pod2", map[string]string{"app": "test"}, corev1.PodFailed) @@ -137,21 +137,10 @@ func TestScaleDown_PodsInTerminalState(t *testing.T) { ) require.NoError(t, err) - // Delete terminal pods (simulating cleanup) - go func() { - time.Sleep(100 * time.Millisecond) - _ = fakeClient.CoreV1().Pods("test-ns").Delete( - context.Background(), "pod1", metav1.DeleteOptions{}, - ) - _ = fakeClient.CoreV1().Pods("test-ns").Delete( - context.Background(), "pod2", metav1.DeleteOptions{}, - ) - }() - // Execute scale down scaledDeployments, err := ScaleDown(client, "test-ns", "app=test", log) - // Assertions - should succeed quickly since pods are in terminal states + // Finished pods are never deleted here: the wait must not count them require.NoError(t, err) assert.Len(t, scaledDeployments, 1) } @@ -1203,3 +1192,77 @@ func TestScaleDown_WaitsForPodsSelectedByWorkload(t *testing.T) { }) } } + +func TestScaleDown_IgnoresFinishedPodsButWaitsForRunningOnes(t *testing.T) { + fakeClient := fake.NewSimpleClientset() + client := k8s.NewTestClient(fakeClient) + ctx := context.Background() + labels := map[string]string{"app": "test"} + + deploy := createDeployment("deploy1", labels, 1) + _, err := fakeClient.AppsV1().Deployments("test-ns").Create(ctx, &deploy, metav1.CreateOptions{}) + require.NoError(t, err) + for _, pod := range []corev1.Pod{ + createPod("evicted", labels, corev1.PodFailed), + createPod("done", labels, corev1.PodSucceeded), + createPod("running", labels, corev1.PodRunning), + } { + _, err = fakeClient.CoreV1().Pods("test-ns").Create(ctx, &pod, metav1.CreateOptions{}) + require.NoError(t, err) + } + + go func() { + time.Sleep(3 * time.Second) + _ = fakeClient.CoreV1().Pods("test-ns").Delete(ctx, "running", metav1.DeleteOptions{}) + }() + + _, err = ScaleDown(client, "test-ns", "app=test", logger.New(true, false)) + require.NoError(t, err) + + _, err = fakeClient.CoreV1().Pods("test-ns").Get(ctx, "running", metav1.GetOptions{}) + assert.True(t, apierrors.IsNotFound(err), "ScaleDown returned while the running pod still existed") + _, err = fakeClient.CoreV1().Pods("test-ns").Get(ctx, "evicted", metav1.GetOptions{}) + assert.NoError(t, err) +} + +// A retry after a failed scale-down finds the workload already at 0 and must keep the recorded count. +func TestScaleDown_RetryKeepsRecordedReplicas(t *testing.T) { + fakeClient := fake.NewSimpleClientset() + client := k8s.NewTestClient(fakeClient) + ctx := context.Background() + labels := map[string]string{"app": "test"} + + deploy := createDeployment("deploy1", labels, 3) + _, err := fakeClient.AppsV1().Deployments("test-ns").Create(ctx, &deploy, metav1.CreateOptions{}) + require.NoError(t, err) + sts := createStatefulSet("sts1", labels, 2) + _, err = fakeClient.AppsV1().StatefulSets("test-ns").Create(ctx, &sts, metav1.CreateOptions{}) + require.NoError(t, err) + + for range 2 { + scaled, err := ScaleDown(client, "test-ns", "app=test", logger.New(true, false)) + require.NoError(t, err) + require.Len(t, scaled, 2) + assert.Equal(t, int32(3), scaled[0].Replicas) + assert.Equal(t, int32(2), scaled[1].Replicas) + } + + deployAfter, err := fakeClient.AppsV1().Deployments("test-ns").Get(ctx, "deploy1", metav1.GetOptions{}) + require.NoError(t, err) + assert.Equal(t, "3", deployAfter.Annotations[k8s.PreRestoreReplicasAnnotation]) + stsAfter, err := fakeClient.AppsV1().StatefulSets("test-ns").Get(ctx, "sts1", metav1.GetOptions{}) + require.NoError(t, err) + assert.Equal(t, "2", stsAfter.Annotations[k8s.PreRestoreReplicasAnnotation]) + + _, err = client.ScaleUpDeploymentsFromAnnotations("test-ns", "app=test") + require.NoError(t, err) + _, err = client.ScaleUpStatefulSetsFromAnnotations("test-ns", "app=test") + require.NoError(t, err) + + deployUp, err := fakeClient.AppsV1().Deployments("test-ns").Get(ctx, "deploy1", metav1.GetOptions{}) + require.NoError(t, err) + assert.Equal(t, int32(3), *deployUp.Spec.Replicas) + stsUp, err := fakeClient.AppsV1().StatefulSets("test-ns").Get(ctx, "sts1", metav1.GetOptions{}) + require.NoError(t, err) + assert.Equal(t, int32(2), *stsUp.Spec.Replicas) +} From 91d0b917e36812c58738a8577fd18ac5ec313f39 Mon Sep 17 00:00:00 2001 From: Vladimir Iliakov Date: Tue, 10 Nov 2026 09:31:59 +0100 Subject: [PATCH 4/6] Verify the restored logs data stream and print the finalize hint on every failure - After the restore completes, require that sts_k8s_logs exists and that no .ds-sts_k8s_logs-* index sits outside it, before scaling workloads up. - Wrap every error after scale-down, not only the reappearance guard, with the check-and-finalize --finalize-only hint. - Point a failing _restore at workloads not matched by scaleDownLabelSelector. - Validate the snapshot in prepareRestoreTargets so it is tested to run before any scale-down, and let the mock model what _restore brings back. - Document that indicesPattern cannot narrow the data stream's backing indices. Co-Authored-By: Claude Sonnet 5.5 --- README.md | 2 + cmd/elasticsearch/check_and_finalize.go | 53 +++++++- cmd/elasticsearch/configure_test.go | 4 + cmd/elasticsearch/list_test.go | 4 + cmd/elasticsearch/restore.go | 35 +++-- cmd/elasticsearch/restore_test.go | 121 ++++++++++++++++++ internal/clients/elasticsearch/client.go | 46 +++++++ internal/clients/elasticsearch/client_test.go | 49 +++++++ internal/clients/elasticsearch/interface.go | 1 + 9 files changed, 300 insertions(+), 15 deletions(-) diff --git a/README.md b/README.md index 0476693..e17fecb 100644 --- a/README.md +++ b/README.md @@ -346,6 +346,8 @@ elasticsearch: indicesPattern: sts*,.ds-sts_k8s_logs* ``` +`indicesPattern` selects indices to delete and restore. The restore names the `datastreamName` data stream, so it brings back all of the data stream's snapshotted backing indices and a pattern cannot select only some of them. Keep `datastreamIndexPrefix` matched by the pattern. + Apply to Kubernetes: ```bash diff --git a/cmd/elasticsearch/check_and_finalize.go b/cmd/elasticsearch/check_and_finalize.go index c6b6197..b464bf3 100644 --- a/cmd/elasticsearch/check_and_finalize.go +++ b/cmd/elasticsearch/check_and_finalize.go @@ -2,6 +2,7 @@ package elasticsearch import ( "fmt" + "slices" "sort" "strings" "time" @@ -427,7 +428,7 @@ func checkAndFinalize( switch status { case es.StatusSuccess: appCtx.Logger.Successf("Restore completed successfully") - return finalizeRestore(appCtx) + return verifyAndFinalize(esClient, appCtx, expected) default: return fmt.Errorf("restore completed with unexpected status: %s", status) } @@ -438,7 +439,9 @@ func checkAndFinalize( if waitForComplete { appCtx.Logger.Println() - return waitAndFinalize(statusFn, appCtx, snapshotName) + return waitAndFinalize(statusFn, appCtx, snapshotName, func() error { + return verifyAndFinalize(esClient, appCtx, expected) + }) } // Not waiting - print status and exit @@ -448,17 +451,59 @@ func checkAndFinalize( } // waitAndFinalize waits for restore to complete and finalizes (scale up) -func waitAndFinalize(statusFn func() (string, bool, error), appCtx *app.Context, snapshotName string) error { +func waitAndFinalize(statusFn func() (string, bool, error), appCtx *app.Context, snapshotName string, finalize func() error) error { restore.PrintAPIWaitingMessage("elasticsearch", snapshotName, appCtx.Namespace, appCtx.Logger) if err := restore.WaitForAPIRestore(statusFn, 0, appCtx.Logger); err != nil { return err } - // Finalize restore (scale up) + return finalize() +} + +// verifyAndFinalize refuses to scale up over a restore that left the logs outside the data stream. +func verifyAndFinalize(esClient es.Interface, appCtx *app.Context, expected []string) error { + if err := verifyLogsDatastream(esClient, appCtx.Config.Elasticsearch.Restore, expected); err != nil { + return withFinalizeHint(err, appCtx.Namespace) + } + return finalizeRestore(appCtx) } +// verifyLogsDatastream checks that the logs data stream exists and that no logs backing index sits +// outside it. It does nothing when the restore did not include logs backing indices. +func verifyLogsDatastream(esClient es.Interface, restoreConfig config.RestoreConfig, expected []string) error { + if !hasDatastreamIndices(expected, restoreConfig.DatastreamIndexPrefix) { + return nil + } + + backing, exists, err := esClient.GetDatastream(restoreConfig.DatastreamName) + if err != nil { + return fmt.Errorf("failed to verify datastream '%s': %w", restoreConfig.DatastreamName, err) + } + if !exists { + return fmt.Errorf("restore completed but the '%s' data stream does not exist", restoreConfig.DatastreamName) + } + + allIndices, err := esClient.ListIndices(restoreConfig.DatastreamIndexPrefix + "-*") + if err != nil { + return fmt.Errorf("failed to list indices: %w", err) + } + var orphans []string + for _, index := range allIndices { + if isDatastreamBackingIndex(index, restoreConfig.DatastreamIndexPrefix) && !slices.Contains(backing, index) { + orphans = append(orphans, index) + } + } + if len(orphans) > 0 { + sort.Strings(orphans) + return fmt.Errorf("restore completed but %d '%s' backing index(es) are outside the data stream: %s", + len(orphans), restoreConfig.DatastreamName, strings.Join(orphans, ", ")) + } + + return nil +} + // finalizeRestore performs post-restore finalization (scale up deployments and release lock) func finalizeRestore(appCtx *app.Context) error { appCtx.Logger.Println() diff --git a/cmd/elasticsearch/configure_test.go b/cmd/elasticsearch/configure_test.go index 7acbaad..c24cbff 100644 --- a/cmd/elasticsearch/configure_test.go +++ b/cmd/elasticsearch/configure_test.go @@ -100,6 +100,10 @@ func (m *mockESClientForConfigure) DeleteDatastream(_ string) (bool, error) { return false, fmt.Errorf("not implemented") } +func (m *mockESClientForConfigure) GetDatastream(_ string) ([]string, bool, error) { + return nil, false, fmt.Errorf("not implemented") +} + // TestConfigureCmd_Unit tests the command structure func TestConfigureCmd_Unit(t *testing.T) { flags := config.NewCLIGlobalFlags() diff --git a/cmd/elasticsearch/list_test.go b/cmd/elasticsearch/list_test.go index 375239a..9016cc5 100644 --- a/cmd/elasticsearch/list_test.go +++ b/cmd/elasticsearch/list_test.go @@ -277,6 +277,10 @@ func (m *mockESClient) DeleteDatastream(_ string) (bool, error) { return false, fmt.Errorf("not implemented") } +func (m *mockESClient) GetDatastream(_ string) ([]string, bool, error) { + return nil, false, fmt.Errorf("not implemented") +} + // createFakeClientWithConfig creates a fake Kubernetes client with a ConfigMap containing the given config func createFakeClientWithConfig(t *testing.T, storageConfig string) *fake.Clientset { t.Helper() diff --git a/cmd/elasticsearch/restore.go b/cmd/elasticsearch/restore.go index 5be999d..dd68eb7 100644 --- a/cmd/elasticsearch/restore.go +++ b/cmd/elasticsearch/restore.go @@ -100,12 +100,7 @@ func runRestore(appCtx *app.Context) error { // Reject snapshots that cannot be monitored safely before scaling down workloads or deleting // any indices. check-and-finalize repeats this validation for independently resumed operations. - restoreIndices, err := restorableSnapshotIndices(snapshotDetails, appCtx.Config.Elasticsearch.Restore) - if err != nil { - return err - } - - targets, err := restoreTargets(snapshotDetails, restoreIndices, appCtx.Config.Elasticsearch.Restore) + targets, err := prepareRestoreTargets(snapshotDetails, appCtx.Config.Elasticsearch.Restore) if err != nil { return err } @@ -142,7 +137,7 @@ func runRestore(appCtx *app.Context) error { } if err := replaceSTSIndices(esClient, appCtx, repository, snapshotDetails, targets); err != nil { - return err + return withFinalizeHint(err, appCtx.Namespace) } if runBackground { @@ -153,6 +148,23 @@ func runRestore(appCtx *app.Context) error { return checkAndFinalize(esClient, appCtx, repository, selectedSnapshot, !runBackground, restoreNoProgressIn) } +// prepareRestoreTargets validates the snapshot and builds the restore request. It needs no cluster +// access, so it runs before anything is scaled down or deleted. +func prepareRestoreTargets(snapshot *es.Snapshot, restoreConfig config.RestoreConfig) ([]string, error) { + restoreIndices, err := restorableSnapshotIndices(snapshot, restoreConfig) + if err != nil { + return nil, err + } + + return restoreTargets(snapshot, restoreIndices, restoreConfig) +} + +// withFinalizeHint tells the user how to release workloads that stay scaled down after a failure. +func withFinalizeHint(err error, namespace string) error { + return fmt.Errorf("%w. Workloads are still scaled down; after fixing this, release them with: "+ + "sts-backup elasticsearch check-and-finalize --finalize-only -n %s", err, namespace) +} + // replaceSTSIndices deletes the target's STS indices and triggers an asynchronous restore of the snapshot's. func replaceSTSIndices(esClient es.Interface, appCtx *app.Context, repository string, snapshot *es.Snapshot, targets []string) error { appCtx.Logger.Println() @@ -164,7 +176,9 @@ func replaceSTSIndices(esClient es.Interface, appCtx *app.Context, repository st isPartial := snapshot.State == "PARTIAL" appCtx.Logger.Infof("Triggering restore for snapshot: %s", snapshot.Snapshot) if err := esClient.RestoreSnapshot(repository, snapshot.Snapshot, strings.Join(targets, ","), isPartial); err != nil { - return fmt.Errorf("failed to trigger restore: %w", err) + return fmt.Errorf("failed to trigger restore (if the error names an existing index or data stream, "+ + "a workload not matched by scaleDownLabelSelector %q may be writing to Elasticsearch): %w", + appCtx.Config.Elasticsearch.Restore.ScaleDownLabelSelector, err) } appCtx.Logger.Successf("Restore triggered successfully") @@ -276,9 +290,8 @@ func deleteAllSTSIndices(esClient es.Interface, appCtx *app.Context) error { } if len(remaining) > 0 { return fmt.Errorf("STS indices reappeared after deletion, so a workload is still writing to Elasticsearch: %s. "+ - "Check for writers not matched by scaleDownLabelSelector %q. "+ - "Workloads are still scaled down; after fixing this, release them with: sts-backup elasticsearch check-and-finalize --finalize-only -n %s", - strings.Join(remaining, ", "), restoreConfig.ScaleDownLabelSelector, appCtx.Namespace) + "Check for writers not matched by scaleDownLabelSelector %q", + strings.Join(remaining, ", "), restoreConfig.ScaleDownLabelSelector) } appCtx.Logger.Successf("All indices deleted successfully") return nil diff --git a/cmd/elasticsearch/restore_test.go b/cmd/elasticsearch/restore_test.go index 62886bc..239e57c 100644 --- a/cmd/elasticsearch/restore_test.go +++ b/cmd/elasticsearch/restore_test.go @@ -34,6 +34,23 @@ type mockESClientForRestore struct { // listCalls counts ListIndices calls; addOnListCall adds indices on the given call, modelling a writer. listCalls int addOnListCall map[int][]string + // restoredDatastreams and restoredPlain record what _restore brought back: a named data stream + // gets its backing indices, anything else comes back as plain indices. + restoredDatastreams map[string][]string + restoredPlain []string +} + +// settle makes the indices brought back by _restore visible to ListIndices. +func (m *mockESClientForRestore) settle() { + m.indices = slices.Clone(m.restoredPlain) + for _, backing := range m.restoredDatastreams { + m.indices = append(m.indices, backing...) + } +} + +func (m *mockESClientForRestore) GetDatastream(name string) ([]string, bool, error) { + backing, ok := m.restoredDatastreams[name] + return backing, ok, nil } func (m *mockESClientForRestore) ListIndices(_ string) ([]string, error) { @@ -103,6 +120,23 @@ func (m *mockESClientForRestore) RestoreSnapshot(_, snapshotName, targets string m.restoredSnapshot = snapshotName m.restoredTargets = targets m.restoredPartial = partial + m.restoredDatastreams = map[string][]string{} + m.restoredPlain = nil + for _, target := range strings.Split(targets, ",") { + if !slices.Contains(m.snapshot.DataStreams, target) { + m.restoredPlain = append(m.restoredPlain, target) + } + } + for _, name := range m.snapshot.DataStreams { + if slices.Contains(strings.Split(targets, ","), name) { + m.restoredDatastreams[name] = slices.DeleteFunc(slices.Clone(m.restoredPlain), func(i string) bool { + return !strings.HasPrefix(i, ".ds-"+name+"-") + }) + m.restoredPlain = slices.DeleteFunc(m.restoredPlain, func(i string) bool { + return strings.HasPrefix(i, ".ds-"+name+"-") + }) + } + } return nil } @@ -536,6 +570,93 @@ func TestRestoreTargetsRejectsSnapshotWithoutDatastream(t *testing.T) { assert.Equal(t, []string{topo}, targets) } +func TestPrepareRestoreTargetsRejectsBeforeAnyClusterAccess(t *testing.T) { + restoreConfig := newLogsRestoreContext().Config.Elasticsearch.Restore + snapshot := &elasticsearch.Snapshot{ + Snapshot: "snap", + State: elasticsearch.StatusSuccess, + Indices: []string{"sts_topology_events-2026.11.01", ".ds-sts_k8s_logs-2026.11.01-000001"}, + } + + targets, err := prepareRestoreTargets(snapshot, restoreConfig) + + require.ErrorContains(t, err, "not the sts_k8s_logs data stream") + assert.Nil(t, targets) +} + +func TestVerifyLogsDatastream(t *testing.T) { + const ( + topo = "sts_topology_events-2026.11.01" + b1 = ".ds-sts_k8s_logs-2026.11.01-000001" + b2 = ".ds-sts_k8s_logs-2026.11.01-000002" + ) + appCtx := newLogsRestoreContext() + restoreConfig := appCtx.Config.Elasticsearch.Restore + snapshot := &elasticsearch.Snapshot{ + Snapshot: "snap", State: elasticsearch.StatusSuccess, + Indices: []string{topo, b1, b2}, DataStreams: []string{"sts_k8s_logs"}, + } + expected, err := restorableSnapshotIndices(snapshot, restoreConfig) + require.NoError(t, err) + + t.Run("restored as a data stream", func(t *testing.T) { + client := &mockESClientForRestore{snapshot: snapshot, indexExistsMap: map[string]bool{}} + targets, err := restoreTargets(snapshot, expected, restoreConfig) + require.NoError(t, err) + require.NoError(t, client.RestoreSnapshot("repo", "snap", strings.Join(targets, ","), false)) + client.settle() + + assert.NoError(t, verifyLogsDatastream(client, restoreConfig, expected)) + }) + + t.Run("restored as plain indices", func(t *testing.T) { + client := &mockESClientForRestore{snapshot: snapshot, indexExistsMap: map[string]bool{}} + require.NoError(t, client.RestoreSnapshot("repo", "snap", strings.Join(expected, ","), false)) + client.settle() + + err := verifyLogsDatastream(client, restoreConfig, expected) + + require.ErrorContains(t, err, "'sts_k8s_logs' data stream does not exist") + }) + + t.Run("backing index outside the data stream", func(t *testing.T) { + client := &mockESClientForRestore{snapshot: snapshot, indexExistsMap: map[string]bool{}} + targets, err := restoreTargets(snapshot, expected, restoreConfig) + require.NoError(t, err) + require.NoError(t, client.RestoreSnapshot("repo", "snap", strings.Join(targets, ","), false)) + client.settle() + client.indices = append(client.indices, ".ds-sts_k8s_logs-2026.11.09-000001") + + err = verifyLogsDatastream(client, restoreConfig, expected) + + require.ErrorContains(t, err, "outside the data stream: .ds-sts_k8s_logs-2026.11.09-000001") + }) + + t.Run("restore without logs indices is not checked", func(t *testing.T) { + client := &mockESClientForRestore{snapshot: snapshot} + + assert.NoError(t, verifyLogsDatastream(client, restoreConfig, []string{topo})) + }) +} + +func TestVerifyAndFinalizeFailsWithFinalizeHint(t *testing.T) { + appCtx := newLogsRestoreContext() + appCtx.Namespace = "ns" + snapshot := &elasticsearch.Snapshot{ + Snapshot: "snap", State: elasticsearch.StatusSuccess, + Indices: []string{"sts_topology_events-2026.11.01", ".ds-sts_k8s_logs-2026.11.01-000001"}, + DataStreams: []string{"sts_k8s_logs"}, + } + client := &mockESClientForRestore{snapshot: snapshot, indexExistsMap: map[string]bool{}} + require.NoError(t, client.RestoreSnapshot("repo", "snap", strings.Join(snapshot.Indices, ","), false)) + client.settle() + + err := verifyAndFinalize(client, appCtx, snapshot.Indices) + + require.ErrorContains(t, err, "does not exist") + assert.ErrorContains(t, err, "check-and-finalize --finalize-only -n ns") +} + func TestRestoreTargetsKeepCompletionExpectation(t *testing.T) { const ( topo = "sts_topology_events-2026.11.01" diff --git a/internal/clients/elasticsearch/client.go b/internal/clients/elasticsearch/client.go index 95543d5..b689e54 100644 --- a/internal/clients/elasticsearch/client.go +++ b/internal/clients/elasticsearch/client.go @@ -264,6 +264,52 @@ func (c *Client) DeleteDatastream(datastreamName string) (bool, error) { return true, nil } +// GetDatastream returns the names of a data stream's backing indices. +// It reports false when the data stream does not exist. +func (c *Client) GetDatastream(datastreamName string) ([]string, bool, error) { + res, err := c.es.Indices.GetDataStream( + c.es.Indices.GetDataStream.WithName(datastreamName), + c.es.Indices.GetDataStream.WithContext(context.Background()), + ) + if err != nil { + return nil, false, fmt.Errorf("failed to get datastream: %w", err) + } + defer res.Body.Close() + + if res.StatusCode == http.StatusNotFound { + return nil, false, nil + } + + if res.IsError() { + return nil, false, fmt.Errorf("elasticsearch returned error: %s", res.String()) + } + + var body struct { + DataStreams []struct { + Name string `json:"name"` + Indices []struct { + IndexName string `json:"index_name"` + } `json:"indices"` + } `json:"data_streams"` + } + if err := json.NewDecoder(res.Body).Decode(&body); err != nil { + return nil, false, fmt.Errorf("failed to decode response: %w", err) + } + + for _, ds := range body.DataStreams { + if ds.Name != datastreamName { + continue + } + indices := make([]string, len(ds.Indices)) + for i, idx := range ds.Indices { + indices[i] = idx.IndexName + } + return indices, true, nil + } + + return nil, false, nil +} + // DeleteSnapshotRepository deletes a snapshot repository func (c *Client) DeleteSnapshotRepository(name string) error { res, err := c.es.Snapshot.DeleteRepository( diff --git a/internal/clients/elasticsearch/client_test.go b/internal/clients/elasticsearch/client_test.go index d5c586a..82b7d6a 100644 --- a/internal/clients/elasticsearch/client_test.go +++ b/internal/clients/elasticsearch/client_test.go @@ -348,6 +348,55 @@ func TestClient_DeleteDatastream(t *testing.T) { } } +func TestClient_GetDatastream(t *testing.T) { + tests := []struct { + name string + responseStatus int + responseBody string + expectIndices []string + expectExists bool + expectError bool + }{ + { + name: "exists", + responseStatus: http.StatusOK, + responseBody: `{"data_streams":[{"name":"sts_k8s_logs","indices":[{"index_name":".ds-sts_k8s_logs-1"},{"index_name":".ds-sts_k8s_logs-2"}]}]}`, + expectIndices: []string{".ds-sts_k8s_logs-1", ".ds-sts_k8s_logs-2"}, + expectExists: true, + }, + {name: "not found", responseStatus: http.StatusNotFound, responseBody: `{}`}, + {name: "empty result", responseStatus: http.StatusOK, responseBody: `{"data_streams":[]}`}, + {name: "elasticsearch error", responseStatus: http.StatusInternalServerError, responseBody: `{}`, expectError: true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + server := mockESServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + assert.Equal(t, "/_data_stream/sts_k8s_logs", r.URL.Path) + assert.Equal(t, http.MethodGet, r.Method) + + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(tt.responseStatus) + _, _ = w.Write([]byte(tt.responseBody)) + })) + defer server.Close() + + client, err := NewClient(server.URL) + require.NoError(t, err) + + indices, exists, err := client.GetDatastream("sts_k8s_logs") + + if tt.expectError { + assert.Error(t, err) + return + } + require.NoError(t, err) + assert.Equal(t, tt.expectExists, exists) + assert.Equal(t, tt.expectIndices, indices) + }) + } +} + func TestClient_IndexExists(t *testing.T) { tests := []struct { name string diff --git a/internal/clients/elasticsearch/interface.go b/internal/clients/elasticsearch/interface.go index 40758e6..cd66417 100644 --- a/internal/clients/elasticsearch/interface.go +++ b/internal/clients/elasticsearch/interface.go @@ -17,6 +17,7 @@ type Interface interface { // Datastream operations DeleteDatastream(datastreamName string) (bool, error) + GetDatastream(datastreamName string) ([]string, bool, error) // Repository and SLM operations DeleteSnapshotRepository(name string) error From 8171f9d172b8b806ca061fc2b22447dc4d8fff16 Mon Sep 17 00:00:00 2001 From: Vladimir Iliakov Date: Tue, 10 Nov 2026 10:41:28 +0100 Subject: [PATCH 5/6] Run the post-restore verification through the reconnecting client The verification used the caller's Elasticsearch client, but polling replaces a port-forward that dropped during a long restore. After a successful restore the verification then failed on a dead connection and left workloads scaled down. Use the reconnecting client and retry transient request failures. Co-Authored-By: Claude Sonnet 5.5 --- cmd/elasticsearch/check_and_finalize.go | 87 ++++++++++++++++++-- cmd/elasticsearch/check_and_finalize_test.go | 27 ++++++ cmd/elasticsearch/restore_test.go | 51 +++++++++++- 3 files changed, 153 insertions(+), 12 deletions(-) diff --git a/cmd/elasticsearch/check_and_finalize.go b/cmd/elasticsearch/check_and_finalize.go index b464bf3..f605254 100644 --- a/cmd/elasticsearch/check_and_finalize.go +++ b/cmd/elasticsearch/check_and_finalize.go @@ -105,7 +105,7 @@ const ( // health check. It starts from the caller's client so the happy path opens no extra port-forward. type reconnectingHealthClient struct { appCtx *app.Context - client indicesHealthGetter + client es.Interface pf *portforward.Conn } @@ -127,6 +127,40 @@ func (r *reconnectingHealthClient) GetIndicesHealth() (map[string]es.IndexHealth return health, nil } +// GetDatastream and ListIndices let the post-restore verification survive a tunnel that dropped +// during a long restore, the same way polling does. +func (r *reconnectingHealthClient) GetDatastream(name string) ([]string, bool, error) { + if r.client == nil { + if err := r.connect(); err != nil { + return nil, false, err + } + } + + indices, exists, err := r.client.GetDatastream(name) + if err != nil { + r.disconnect() + return nil, false, err + } + + return indices, exists, nil +} + +func (r *reconnectingHealthClient) ListIndices(pattern string) ([]string, error) { + if r.client == nil { + if err := r.connect(); err != nil { + return nil, err + } + } + + indices, err := r.client.ListIndices(pattern) + if err != nil { + r.disconnect() + return nil, err + } + + return indices, nil +} + func (r *reconnectingHealthClient) connect() error { pf, err := portforward.SetupPortForward( r.appCtx.K8sClient, @@ -428,7 +462,7 @@ func checkAndFinalize( switch status { case es.StatusSuccess: appCtx.Logger.Successf("Restore completed successfully") - return verifyAndFinalize(esClient, appCtx, expected) + return verifyAndFinalize(healthClient, appCtx, expected) default: return fmt.Errorf("restore completed with unexpected status: %s", status) } @@ -440,7 +474,7 @@ func checkAndFinalize( if waitForComplete { appCtx.Logger.Println() return waitAndFinalize(statusFn, appCtx, snapshotName, func() error { - return verifyAndFinalize(esClient, appCtx, expected) + return verifyAndFinalize(healthClient, appCtx, expected) }) } @@ -461,9 +495,37 @@ func waitAndFinalize(statusFn func() (string, bool, error), appCtx *app.Context, return finalize() } +// logsDatastreamReader is the part of the Elasticsearch client the post-restore verification needs. +type logsDatastreamReader interface { + GetDatastream(datastreamName string) ([]string, bool, error) + ListIndices(pattern string) ([]string, error) +} + +const verifyAttempts = 3 + +// verifyRetryInterval is a variable so tests do not wait. +var verifyRetryInterval = 2 * time.Second + +// withRetries retries transient request failures, such as a port-forward that dropped. It does not +// apply to verification verdicts, which are returned by the caller after the data is fetched. +func withRetries(log *logger.Logger, what string, fn func() error) error { + var err error + for attempt := 1; attempt <= verifyAttempts; attempt++ { + if err = fn(); err == nil { + return nil + } + if attempt < verifyAttempts { + log.Warningf("%s failed (%d/%d), retrying: %v", what, attempt, verifyAttempts, err) + time.Sleep(verifyRetryInterval) + } + } + + return err +} + // verifyAndFinalize refuses to scale up over a restore that left the logs outside the data stream. -func verifyAndFinalize(esClient es.Interface, appCtx *app.Context, expected []string) error { - if err := verifyLogsDatastream(esClient, appCtx.Config.Elasticsearch.Restore, expected); err != nil { +func verifyAndFinalize(client logsDatastreamReader, appCtx *app.Context, expected []string) error { + if err := verifyLogsDatastream(client, appCtx.Logger, appCtx.Config.Elasticsearch.Restore, expected); err != nil { return withFinalizeHint(err, appCtx.Namespace) } @@ -472,12 +534,17 @@ func verifyAndFinalize(esClient es.Interface, appCtx *app.Context, expected []st // verifyLogsDatastream checks that the logs data stream exists and that no logs backing index sits // outside it. It does nothing when the restore did not include logs backing indices. -func verifyLogsDatastream(esClient es.Interface, restoreConfig config.RestoreConfig, expected []string) error { +func verifyLogsDatastream(client logsDatastreamReader, log *logger.Logger, restoreConfig config.RestoreConfig, expected []string) error { if !hasDatastreamIndices(expected, restoreConfig.DatastreamIndexPrefix) { return nil } - backing, exists, err := esClient.GetDatastream(restoreConfig.DatastreamName) + var backing []string + var exists bool + err := withRetries(log, "Data stream check", func() (err error) { + backing, exists, err = client.GetDatastream(restoreConfig.DatastreamName) + return err + }) if err != nil { return fmt.Errorf("failed to verify datastream '%s': %w", restoreConfig.DatastreamName, err) } @@ -485,7 +552,11 @@ func verifyLogsDatastream(esClient es.Interface, restoreConfig config.RestoreCon return fmt.Errorf("restore completed but the '%s' data stream does not exist", restoreConfig.DatastreamName) } - allIndices, err := esClient.ListIndices(restoreConfig.DatastreamIndexPrefix + "-*") + var allIndices []string + err = withRetries(log, "Index listing", func() (err error) { + allIndices, err = client.ListIndices(restoreConfig.DatastreamIndexPrefix + "-*") + return err + }) if err != nil { return fmt.Errorf("failed to list indices: %w", err) } diff --git a/cmd/elasticsearch/check_and_finalize_test.go b/cmd/elasticsearch/check_and_finalize_test.go index ad1c185..74aea1f 100644 --- a/cmd/elasticsearch/check_and_finalize_test.go +++ b/cmd/elasticsearch/check_and_finalize_test.go @@ -22,6 +22,7 @@ type healthResult struct { // fakeHealthClient replays results in order, repeating the last one once exhausted. type fakeHealthClient struct { + es.Interface results []healthResult calls int } @@ -541,6 +542,20 @@ func TestReconnectingHealthClient(t *testing.T) { assert.Nil(t, client.pf, "must not adopt a port-forward it did not open") }) + t.Run("verification calls drop a broken client so the next call reconnects", func(t *testing.T) { + broken := &brokenDatastreamClient{} + client := &reconnectingHealthClient{client: broken} + + _, _, err := client.GetDatastream("sts_k8s_logs") + require.Error(t, err) + assert.Nil(t, client.client) + + client.client = broken + _, err = client.ListIndices("*") + require.Error(t, err) + assert.Nil(t, client.client) + }) + t.Run("disconnect is safe to call repeatedly", func(t *testing.T) { client := &reconnectingHealthClient{client: &fakeHealthClient{}} @@ -550,3 +565,15 @@ func TestReconnectingHealthClient(t *testing.T) { }) }) } + +type brokenDatastreamClient struct { + es.Interface +} + +func (brokenDatastreamClient) GetDatastream(string) ([]string, bool, error) { + return nil, false, errors.New("connection refused") +} + +func (brokenDatastreamClient) ListIndices(string) ([]string, error) { + return nil, errors.New("connection refused") +} diff --git a/cmd/elasticsearch/restore_test.go b/cmd/elasticsearch/restore_test.go index 239e57c..b99a2b9 100644 --- a/cmd/elasticsearch/restore_test.go +++ b/cmd/elasticsearch/restore_test.go @@ -606,7 +606,7 @@ func TestVerifyLogsDatastream(t *testing.T) { require.NoError(t, client.RestoreSnapshot("repo", "snap", strings.Join(targets, ","), false)) client.settle() - assert.NoError(t, verifyLogsDatastream(client, restoreConfig, expected)) + assert.NoError(t, verifyLogsDatastream(client, appCtx.Logger, restoreConfig, expected)) }) t.Run("restored as plain indices", func(t *testing.T) { @@ -614,7 +614,7 @@ func TestVerifyLogsDatastream(t *testing.T) { require.NoError(t, client.RestoreSnapshot("repo", "snap", strings.Join(expected, ","), false)) client.settle() - err := verifyLogsDatastream(client, restoreConfig, expected) + err := verifyLogsDatastream(client, appCtx.Logger, restoreConfig, expected) require.ErrorContains(t, err, "'sts_k8s_logs' data stream does not exist") }) @@ -627,7 +627,7 @@ func TestVerifyLogsDatastream(t *testing.T) { client.settle() client.indices = append(client.indices, ".ds-sts_k8s_logs-2026.11.09-000001") - err = verifyLogsDatastream(client, restoreConfig, expected) + err = verifyLogsDatastream(client, appCtx.Logger, restoreConfig, expected) require.ErrorContains(t, err, "outside the data stream: .ds-sts_k8s_logs-2026.11.09-000001") }) @@ -635,7 +635,50 @@ func TestVerifyLogsDatastream(t *testing.T) { t.Run("restore without logs indices is not checked", func(t *testing.T) { client := &mockESClientForRestore{snapshot: snapshot} - assert.NoError(t, verifyLogsDatastream(client, restoreConfig, []string{topo})) + assert.NoError(t, verifyLogsDatastream(client, appCtx.Logger, restoreConfig, []string{topo})) + }) +} + +type flakyDatastreamReader struct { + failures int + calls int + backing []string +} + +func (f *flakyDatastreamReader) GetDatastream(string) ([]string, bool, error) { + f.calls++ + if f.calls <= f.failures { + return nil, false, fmt.Errorf("connection refused") + } + return f.backing, true, nil +} + +func (f *flakyDatastreamReader) ListIndices(string) ([]string, error) { + return f.backing, nil +} + +func TestVerifyLogsDatastreamRetriesTransientErrors(t *testing.T) { + verifyRetryInterval = 0 + t.Cleanup(func() { verifyRetryInterval = 2 * time.Second }) + appCtx := newLogsRestoreContext() + restoreConfig := appCtx.Config.Elasticsearch.Restore + expected := []string{"sts_topology_events-2026.11.01", ".ds-sts_k8s_logs-2026.11.01-000001"} + backing := []string{".ds-sts_k8s_logs-2026.11.01-000001"} + + t.Run("recovers after a dropped connection", func(t *testing.T) { + reader := &flakyDatastreamReader{failures: verifyAttempts - 1, backing: backing} + + assert.NoError(t, verifyLogsDatastream(reader, appCtx.Logger, restoreConfig, expected)) + assert.Equal(t, verifyAttempts, reader.calls) + }) + + t.Run("gives up after the last attempt", func(t *testing.T) { + reader := &flakyDatastreamReader{failures: verifyAttempts, backing: backing} + + err := verifyLogsDatastream(reader, appCtx.Logger, restoreConfig, expected) + + require.ErrorContains(t, err, "failed to verify datastream") + assert.Equal(t, verifyAttempts, reader.calls) }) } From da7190b5d3632e615b46181fa893b3f706f440dd Mon Sep 17 00:00:00 2001 From: Vladimir Iliakov Date: Tue, 10 Nov 2026 13:54:36 +0100 Subject: [PATCH 6/6] Reject an indicesPattern that selects only some logs backing indices Naming the sts_k8s_logs data stream restores all of its snapshotted backing indices, and deleting it removes all live ones. With a pattern selecting only some of them, completion tracking and verification would not cover the rest, so the CLI could scale workloads up while an omitted backing index still had unavailable primary shards. Fail before scale-down and require the pattern to select all of them or none. The restore mock now expands a named data stream to all of its snapshotted backing indices, as Elasticsearch does. Co-Authored-By: Claude Sonnet 5.5 --- README.md | 2 +- cmd/elasticsearch/restore.go | 15 ++++++++ cmd/elasticsearch/restore_test.go | 62 +++++++++++++++++++++++++++++++ 3 files changed, 78 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index e17fecb..266f1b5 100644 --- a/README.md +++ b/README.md @@ -346,7 +346,7 @@ elasticsearch: indicesPattern: sts*,.ds-sts_k8s_logs* ``` -`indicesPattern` selects indices to delete and restore. The restore names the `datastreamName` data stream, so it brings back all of the data stream's snapshotted backing indices and a pattern cannot select only some of them. Keep `datastreamIndexPrefix` matched by the pattern. +`indicesPattern` selects indices to delete and restore. The restore names the `datastreamName` data stream, which brings back all of its snapshotted backing indices, so the pattern must select all of the logs backing indices or none of them; a pattern that selects only some is rejected before anything is scaled down or deleted. Apply to Kubernetes: diff --git a/cmd/elasticsearch/restore.go b/cmd/elasticsearch/restore.go index dd68eb7..f4ec0e0 100644 --- a/cmd/elasticsearch/restore.go +++ b/cmd/elasticsearch/restore.go @@ -326,6 +326,21 @@ func restoreTargets(snapshot *es.Snapshot, indices []string, restoreConfig confi snapshot.Snapshot, restoreConfig.DatastreamIndexPrefix, restoreConfig.DatastreamName) } + // Naming the data stream restores all of its snapshotted backing indices, which completion + // tracking and the deletion would not cover if the pattern selected only some of them. + var omitted []string + for _, index := range snapshot.Indices { + if isDatastreamBackingIndex(index, restoreConfig.DatastreamIndexPrefix) && !slices.Contains(indices, index) { + omitted = append(omitted, index) + } + } + if len(omitted) > 0 { + slices.Sort(omitted) + return nil, fmt.Errorf("indicesPattern %q selects only some %s backing indices, but restoring the %s data stream "+ + "restores all of them; omitted: %s. The pattern must select all of them or none", + restoreConfig.IndicesPattern, restoreConfig.DatastreamIndexPrefix, restoreConfig.DatastreamName, strings.Join(omitted, ", ")) + } + return append(slices.Clone(indices), restoreConfig.DatastreamName), nil } diff --git a/cmd/elasticsearch/restore_test.go b/cmd/elasticsearch/restore_test.go index b99a2b9..e084f12 100644 --- a/cmd/elasticsearch/restore_test.go +++ b/cmd/elasticsearch/restore_test.go @@ -129,6 +129,12 @@ func (m *mockESClientForRestore) RestoreSnapshot(_, snapshotName, targets string } for _, name := range m.snapshot.DataStreams { if slices.Contains(strings.Split(targets, ","), name) { + // Like Elasticsearch, naming the stream restores all of its snapshotted backing indices. + for _, index := range m.snapshot.Indices { + if strings.HasPrefix(index, ".ds-"+name+"-") && !slices.Contains(m.restoredPlain, index) { + m.restoredPlain = append(m.restoredPlain, index) + } + } m.restoredDatastreams[name] = slices.DeleteFunc(slices.Clone(m.restoredPlain), func(i string) bool { return !strings.HasPrefix(i, ".ds-"+name+"-") }) @@ -570,6 +576,62 @@ func TestRestoreTargetsRejectsSnapshotWithoutDatastream(t *testing.T) { assert.Equal(t, []string{topo}, targets) } +func TestPrepareRestoreTargetsRequiresAllOrNoLogsBackingIndices(t *testing.T) { + const ( + topo = "sts_topology_events-2026.11.01" + b1 = ".ds-sts_k8s_logs-2026.11.01-000001" + b2 = ".ds-sts_k8s_logs-2026.11.01-000002" + ) + snapshot := &elasticsearch.Snapshot{ + Snapshot: "snap", State: elasticsearch.StatusSuccess, + Indices: []string{topo, b1, b2}, DataStreams: []string{"sts_k8s_logs"}, + } + + tests := []struct { + name string + pattern string + expectErr string + expectTargets []string + }{ + {name: "one of two backing indices", pattern: "sts*,.ds-sts_k8s_logs-2026.11.01-000002", expectErr: "omitted: " + b1}, + {name: "exclusion of a backing index", pattern: "sts*,.ds-sts_k8s_logs*,-" + b1, expectErr: "omitted: " + b1}, + {name: "all backing indices", pattern: "sts*,.ds-sts_k8s_logs*", expectTargets: []string{topo, b1, b2, "sts_k8s_logs"}}, + {name: "no backing indices", pattern: "sts*", expectTargets: []string{topo}}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + restoreConfig := newLogsRestoreContext().Config.Elasticsearch.Restore + restoreConfig.IndicesPattern = tt.pattern + + targets, err := prepareRestoreTargets(snapshot, restoreConfig) + + if tt.expectErr != "" { + require.ErrorContains(t, err, tt.expectErr) + assert.ErrorContains(t, err, "must select all of them or none") + return + } + require.NoError(t, err) + assert.Equal(t, tt.expectTargets, targets) + }) + } +} + +func TestMockRestoreExpandsNamedDatastream(t *testing.T) { + const ( + b1 = ".ds-sts_k8s_logs-2026.11.01-000001" + b2 = ".ds-sts_k8s_logs-2026.11.01-000002" + ) + client := &mockESClientForRestore{ + snapshot: &elasticsearch.Snapshot{Indices: []string{b1, b2}, DataStreams: []string{"sts_k8s_logs"}}, + indexExistsMap: map[string]bool{}, + } + + require.NoError(t, client.RestoreSnapshot("repo", "snap", b2+",sts_k8s_logs", false)) + + assert.Equal(t, []string{b2, b1}, client.restoredDatastreams["sts_k8s_logs"]) + assert.Empty(t, client.restoredPlain) +} + func TestPrepareRestoreTargetsRejectsBeforeAnyClusterAccess(t *testing.T) { restoreConfig := newLogsRestoreContext().Config.Elasticsearch.Restore snapshot := &elasticsearch.Snapshot{