fix(opensearch-operator): stop excluding a node whose drain cannot finish

The yellow restart gate carried in the previous patch lets a permanently yellow cluster through, and with three or more nodes packages/apps/opensearch turns drainDataNodes on. The restart then reaches PreparePodForDelete, which excludes the candidate from allocation before it can know whether the drain will finish. On the cluster that made the gate wait it never finishes: a shard on that node has nowhere else to go, which is the same condition holding the cluster yellow. The exclusion is withdrawn only after a pod is really deleted, so a data node stays out of allocation for as long as the restart is stuck, shedding its shards onto the rest of the cluster and buying nothing. Ask the allocator before excluding anything, and only while nothing is moving, so a drain that is making progress costs nothing. A started shard answering can_move_to_other_node: no means the drain cannot finish, and then no exclusion is written and one left by an earlier pass is withdrawn. The two-data-node path keeps its behaviour, because it never waits for a full drain. Checking after the fact instead would flap: the next reconcile, ten seconds later, would exclude the node again. Upstream as opensearch-project/opensearch-k8s-operator#1585, fixed there by #1586; the patch goes away with the first release carrying it. Backport: the Dockerfile header keeps release-1.6's trace paragraph, placed between the new ones as on main. Assisted-by: LLM Signed-off-by: IvanHunters <xorokhotnikov@gmail.com> (cherry picked from commit 28195d3ef8) Signed-off-by: Myasnikov Daniil <myasnikovdaniil2001@gmail.com>
2026-10-30 08:37:31 +09:00 · 2026-10-22 10:42:41 +00:00
1 parent f78d3a4e1c
commit ce29a18dd4
@@ -14,6 +14,17 @@
 # data role only; carry it as well if that ever changes. Drop the patch on the
 # first operator release that includes the change.
 #
+# patches/skip-unusable-drain-exclusion.diff is the other half of that change,
+# for a release with three or more nodes, where packages/apps/opensearch turns
+# drainDataNodes on. The gate above now lets a permanently yellow cluster reach
+# PreparePodForDelete, which excludes the candidate from allocation before it can
+# know whether the drain will finish. On the cluster that made the gate wait, it
+# never does, and the exclusion is only withdrawn after a pod is really deleted,
+# so a data node stays out of allocation for as long as the restart is stuck. The
+# patch asks the allocator first, and only while nothing is moving. Tracked
+# upstream as opensearch-project/opensearch-k8s-operator#1585, fixed by #1586;
+# drop it on the first release carrying that.
+#
 # patches/trace-stalled-component-reconcilers.diff names the component a
 # reconcile is stuck in. The cluster controller runs nine component reconcilers
 # in sequence and none of them logs unconditionally when it is entered, so a
@@ -25,6 +36,10 @@
 # nothing while a component behaves, so it costs nothing on a healthy cluster.
 # It reports rather than repairs: drop it once the stall itself is understood
 # and fixed.
+#
+# The patches are applied in the order the glob expands, which is alphabetical,
+# and skip-unusable-drain-exclusion builds on restart-on-permanently-yellow. A
+# rename that reorders them fails the build rather than producing a wrong binary.
 FROM golang:1.26@sha256:079e59808d2d252516e27e3f3a9c003740dee7f75e55aa71528766d52bcfc16a AS builder

 # Keep in step with the chart version pulled by `make update`: the chart's
@@ -0,0 +1,300 @@
+diff --git a/opensearch-operator/opensearch-gateway/responses/AllocationExplainResponse.go b/opensearch-operator/opensearch-gateway/responses/AllocationExplainResponse.go
+new file mode 100644
+--- /dev/null
+++ b/opensearch-operator/opensearch-gateway/responses/AllocationExplainResponse.go
+@@ -0,0 +1,13 @@
+package responses
+
+// AllocationExplainResponse carries the part of _cluster/allocation/explain this
+// operator reads. The endpoint returns a great deal more; everything else is
+// left out on purpose, so a field the operator never consults cannot break
+// decoding when OpenSearch renames it.
+type AllocationExplainResponse struct {
+	Index string `json:"index"`
+	Shard int    `json:"shard"`
+	// CanMoveToOtherNode is "yes", "no" or "throttled", and is only present for a
+	// shard that is currently assigned.
+	CanMoveToOtherNode string `json:"can_move_to_other_node,omitempty"`
+}
+diff --git a/opensearch-operator/opensearch-gateway/services/drain_progress_test.go b/opensearch-operator/opensearch-gateway/services/drain_progress_test.go
+new file mode 100644
+--- /dev/null
+++ b/opensearch-operator/opensearch-gateway/services/drain_progress_test.go
+@@ -0,0 +1,165 @@
+package services
+
+import (
+	"encoding/json"
+	"io"
+	"net/http"
+	"strings"
+	"testing"
+
+	"github.com/go-logr/logr"
+	"github.com/jarcoal/httpmock"
+)
+
+// drainMockClient serves the endpoints a drain decision reads: health, cluster
+// settings (with the exclude list), the shard table, and one allocation
+// explanation per shard, keyed by "<index>/<shard>".
+func drainMockClient(t *testing.T, health, excluded, catShards string, explains map[string]string) (*OsClusterClient, *httpmock.MockTransport, *[]string) {
+	t.Helper()
+	tr := httpmock.NewMockTransport()
+	for _, m := range []string{http.MethodGet, http.MethodHead} {
+		tr.RegisterResponder(m, `=~^http://os.test:9200/?$`, httpmock.NewStringResponder(200, `{"version":{"number":"2.0.0"}}`))
+	}
+	tr.RegisterResponder(http.MethodGet, `=~/_cluster/health`, httpmock.NewStringResponder(200, health))
+	tr.RegisterResponder(http.MethodGet, `=~/_cluster/settings`, httpmock.NewStringResponder(200,
+		`{"transient":{"cluster":{"routing":{"allocation":{"enable":"all","exclude":{"_name":"`+excluded+`"}}}}},"persistent":{}}`))
+	puts := &[]string{}
+	tr.RegisterResponder(http.MethodPut, `=~/_cluster/settings`, func(r *http.Request) (*http.Response, error) {
+		b, _ := io.ReadAll(r.Body)
+		*puts = append(*puts, string(b))
+		return httpmock.NewStringResponse(200, `{}`), nil
+	})
+	tr.RegisterResponder(http.MethodGet, `=~/_cat/shards`, httpmock.NewStringResponder(200, catShards))
+	tr.RegisterResponder(http.MethodGet, `=~/_cat/indices`, httpmock.NewStringResponder(200, `[]`))
+	tr.RegisterResponder(http.MethodPost, `=~/_cluster/allocation/explain`, func(r *http.Request) (*http.Response, error) {
+		var req struct {
+			Index string `json:"index"`
+			Shard int    `json:"shard"`
+		}
+		b, _ := io.ReadAll(r.Body)
+		if err := json.Unmarshal(b, &req); err != nil {
+			return httpmock.NewStringResponse(400, `{}`), nil
+		}
+		body, ok := explains[req.Index]
+		if !ok {
+			return httpmock.NewStringResponse(404, `{}`), nil
+		}
+		return httpmock.NewStringResponse(200, body), nil
+	})
+	c, err := NewOsClusterClient("http://os.test:9200", "u", "p", WithTransport(tr))
+	if err != nil {
+		t.Fatal(err)
+	}
+	return c, tr, puts
+}
+
+func explainBody(canMove string) string {
+	return `{"index":"a","shard":0,"primary":true,"current_state":"started","can_remain_on_current_node":"no","can_move_to_other_node":"` + canMove + `","node_allocation_decisions":[]}`
+}
+
+func TestDrainCanProceed(t *testing.T) {
+	onNode := `[{"index":"a","shard":"0","prirep":"p","state":"STARTED","node":"n0"}]`
+	tests := []struct {
+		name     string
+		shards   string
+		explains map[string]string
+		want     bool
+	}{
+		{"a shard that can move keeps the drain alive", onNode, map[string]string{"a": explainBody("yes")}, true},
+		{"a shard with nowhere to go stops it", onNode, map[string]string{"a": explainBody("no")}, false},
+		{
+			"one immovable shard is enough to stop it",
+			`[{"index":"a","shard":"0","prirep":"p","state":"STARTED","node":"n0"},{"index":"b","shard":"0","prirep":"p","state":"STARTED","node":"n0"}]`,
+			map[string]string{"a": explainBody("yes"), "b": explainBody("no")},
+			false,
+		},
+		{"an empty node has nothing to block it", `[]`, nil, true},
+		{"shards on other nodes are not this node's problem",
+			`[{"index":"a","shard":"0","prirep":"p","state":"STARTED","node":"n1"}]`,
+			map[string]string{"a": explainBody("no")}, true},
+		{"a relocating shard is already moving",
+			`[{"index":"a","shard":"0","prirep":"p","state":"RELOCATING","node":"n0 -> 10.0.0.2 id n2"}]`,
+			map[string]string{"a": explainBody("no")}, true},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			c, _, _ := drainMockClient(t, `{"status":"yellow"}`, "", tt.shards, tt.explains)
+			got, _, err := drainCanProceed(c, "n0")
+			if err != nil {
+				t.Fatal(err)
+			}
+			if got != tt.want {
+				t.Errorf("got %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+// The node holds a shard nothing else can take, so the drain will not finish.
+// The exclusion has to be released rather than held for as long as the restart
+// is stuck, and it must not be re-applied on the next pass either.
+func TestPreparePodForDeleteReleasesAnUnproductiveExclusion(t *testing.T) {
+	stuck := `[{"index":"a","shard":"0","prirep":"p","state":"STARTED","node":"n0"},{"index":"a","shard":"0","prirep":"r","state":"UNASSIGNED","node":null}]`
+	quietYellow := `{"status":"yellow","number_of_data_nodes":3,"unassigned_shards":1}`
+
+	t.Run("an exclusion already in place is withdrawn", func(t *testing.T) {
+		c, _, puts := drainMockClient(t, quietYellow, "n0", stuck, map[string]string{"a": explainBody("no")})
+		ready, err := PreparePodForDelete(c, logr.Discard(), "n0", true, 3)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if ready {
+			t.Fatal("the pod is not safe to delete, it still holds a shard")
+		}
+		if len(*puts) != 1 {
+			t.Fatalf("expected exactly one settings write, got %d: %v", len(*puts), *puts)
+		}
+		if strings.Contains((*puts)[0], `"n0"`) {
+			t.Errorf("the write should clear the exclusion, got %s", (*puts)[0])
+		}
+	})
+
+	t.Run("no exclusion is applied in the first place", func(t *testing.T) {
+		c, _, puts := drainMockClient(t, quietYellow, "", stuck, map[string]string{"a": explainBody("no")})
+		ready, err := PreparePodForDelete(c, logr.Discard(), "n0", true, 3)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if ready {
+			t.Fatal("the pod is not safe to delete, it still holds a shard")
+		}
+		if len(*puts) != 0 {
+			t.Errorf("nothing should be written when the node was never excluded, got %v", *puts)
+		}
+	})
+}
+
+// A drain that can still make progress keeps the behaviour it had: exclude the
+// node and wait for it to empty.
+func TestPreparePodForDeleteStillDrainsWhenItCan(t *testing.T) {
+	movable := `[{"index":"a","shard":"0","prirep":"p","state":"STARTED","node":"n0"},{"index":"a","shard":"0","prirep":"r","state":"STARTED","node":"n1"}]`
+	c, _, puts := drainMockClient(t, `{"status":"green","number_of_data_nodes":3}`, "", movable, map[string]string{"a": explainBody("yes")})
+	ready, err := PreparePodForDelete(c, logr.Discard(), "n0", true, 3)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if ready {
+		t.Fatal("the node is not empty yet")
+	}
+	if len(*puts) != 1 || !strings.Contains((*puts)[0], `n0`) {
+		t.Fatalf("expected the node to be excluded, writes were %v", *puts)
+	}
+}
+
+// While shards are actually moving there is nothing to decide, and the
+// allocation explanation must not be asked for at all.
+func TestPreparePodForDeleteDoesNotExplainWhileShardsMove(t *testing.T) {
+	moving := `[{"index":"a","shard":"0","prirep":"p","state":"STARTED","node":"n0"},{"index":"a","shard":"0","prirep":"r","state":"UNASSIGNED","node":null}]`
+	c, tr, _ := drainMockClient(t, `{"status":"yellow","number_of_data_nodes":3,"relocating_shards":1}`, "n0", moving, map[string]string{"a": explainBody("no")})
+	if _, err := PreparePodForDelete(c, logr.Discard(), "n0", true, 3); err != nil {
+		t.Fatal(err)
+	}
+	if n := tr.GetCallCountInfo()["POST =~/_cluster/allocation/explain"]; n != 0 {
+		t.Errorf("explain must not be called while the allocator is busy, got %d calls", n)
+	}
+}
+diff --git a/opensearch-operator/opensearch-gateway/services/os_client.go b/opensearch-operator/opensearch-gateway/services/os_client.go
+--- a/opensearch-operator/opensearch-gateway/services/os_client.go
+++ b/opensearch-operator/opensearch-gateway/services/os_client.go
+@@ -171,6 +171,26 @@ func (client *OsClusterClient) CatIndices() ([]responses.CatIndicesResponse, err
+ 	return response, err
+ }
+ 
+func (client *OsClusterClient) GetAllocationExplain(index string, shard int, primary bool) (responses.AllocationExplainResponse, error) {
+	body := map[string]interface{}{
+		"index":   index,
+		"shard":   shard,
+		"primary": primary,
+	}
+	req := opensearchapi.ClusterAllocationExplainRequest{Body: opensearchutil.NewJSONReader(body)}
+	explainRes, err := req.Do(context.Background(), client.client)
+	var response responses.AllocationExplainResponse
+	if err != nil {
+		return response, err
+	}
+	defer helpers.SafeClose(explainRes.Body)
+	if explainRes.IsError() {
+		return response, fmt.Errorf("failed to get allocation explanation: %s", explainRes.String())
+	}
+	err = json.NewDecoder(explainRes.Body).Decode(&response)
+	return response, err
+}
+
+ func (client *OsClusterClient) CatShards(headers []string) ([]responses.CatShardsResponse, error) {
+ 	req := opensearchapi.CatShardsRequest{Format: "json", H: headers}
+ 	indicesRes, err := req.Do(context.Background(), client.client)
+diff --git a/opensearch-operator/opensearch-gateway/services/os_data_service.go b/opensearch-operator/opensearch-gateway/services/os_data_service.go
+--- a/opensearch-operator/opensearch-gateway/services/os_data_service.go
+++ b/opensearch-operator/opensearch-gateway/services/os_data_service.go
+@@ -4,6 +4,7 @@ import (
+ 	"context"
+ 	"encoding/json"
+ 	"fmt"
+	"strconv"
+ 	"strings"
+ 
+ 	"github.com/google/go-cmp/cmp"
+@@ -249,8 +250,70 @@ func ReactivateShardAllocation(service *OsClusterClient) error {
+ 	return nil
+ }
+ 
+// drainCanProceed reports whether the allocator still has somewhere to put what
+// is left on nodeName. A drain waits for the node to empty, so one shard that
+// can be placed nowhere else is what turns that wait into a permanent one; the
+// reason for the first such shard is returned for the caller's log line.
+//
+// Only STARTED copies are asked about: a relocating one is already on its way
+// out, and an unassigned one is not on this node to begin with.
+func drainCanProceed(service *OsClusterClient, nodeName string) (bool, string, error) {
+	shards, err := service.CatShards([]string{"index", "shard", "prirep", "state", "node"})
+	if err != nil {
+		return false, "", err
+	}
+	for _, shard := range shardsOnNodeFromResponse(shards, nodeName) {
+		if shard.State != "STARTED" {
+			continue
+		}
+		shardNum, err := strconv.Atoi(shard.Shard)
+		if err != nil {
+			continue
+		}
+		explain, err := service.GetAllocationExplain(shard.Index, shardNum, shard.PrimaryOrReplica == "p")
+		if err != nil {
+			return false, "", err
+		}
+		if strings.EqualFold(explain.CanMoveToOtherNode, "no") {
+			return false, fmt.Sprintf("%s[%s] has no other node to move to", shard.Index, shard.Shard), nil
+		}
+	}
+	return true, "", nil
+}
+
+ func PreparePodForDelete(service *OsClusterClient, lg logr.Logger, podName string, drainNode bool, nodeCount int32) (bool, error) {
+ 	if drainNode {
+		// Excluding a node is only worth anything while the allocator can move its
+		// shards elsewhere. When it cannot, the node never empties, the restart
+		// waits for something that will not happen, and the exclusion keeps a data
+		// node out of allocation for the whole of it. Two data nodes are exempt:
+		// that path never waits for a full drain, only for system-index primaries.
+		//
+		// Asked only while nothing is moving, so a drain in progress pays nothing:
+		// the cost is one allocation explanation per started shard on the node, and
+		// it stops at the first shard that cannot move.
+		if nodeCount != 2 {
+			health, err := service.GetHealth()
+			if err != nil {
+				return false, err
+			}
+			if !hasShardActivity(health) {
+				canDrain, reason, err := drainCanProceed(service, podName)
+				if err != nil {
+					return false, err
+				}
+				if !canDrain {
+					// Withdraw an exclusion left by an earlier pass. With an empty
+					// exclude list this writes nothing.
+					if _, err := RemoveExcludeNodeHost(service, podName); err != nil {
+						return false, err
+					}
+					lg.Info(fmt.Sprintf("Not draining %s: %s. Leaving allocation as it is until that changes", podName, reason))
+					return false, nil
+				}
+			}
+		}
+
+ 		// If we are draining nodes then drain the working node
+ 		_, err := AppendExcludeNodeHost(service, podName)
+ 		if err != nil {