Allow users to suspend Elasticsearch Pods for debugging purposes

pkg/apis/elasticsearch/v1/elasticsearch_types.go

@@ -6,9 +6,10 @@ package v1
 
 import (
 	"fmt"
-	"github.com/elastic/cloud-on-k8s/pkg/utils/set"
 	"strings"
 
+	"github.com/elastic/cloud-on-k8s/pkg/utils/set"
+
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"

pkg/apis/elasticsearch/v1/elasticsearch_types_test.go

@@ -6,12 +6,13 @@ package v1
 
 import (
 	"fmt"
-	"github.com/elastic/cloud-on-k8s/pkg/utils/set"
 	"reflect"
 	"sort"
 	"testing"
 	"time"
 
+	"github.com/elastic/cloud-on-k8s/pkg/utils/set"
+
 	"github.com/elastic/cloud-on-k8s/pkg/utils/pointer"
 
 	"github.com/stretchr/testify/assert"
@@ -215,37 +216,35 @@ func Test_GetMaxUnavailableOrDefault(t *testing.T) {
 
 func TestElasticsearch_SuspendedPodNames(t *testing.T) {
 	tests := []struct {
-		name   string
+		name       string
 		ObjectMeta metav1.ObjectMeta
-		want   set.StringSet
+		want       set.StringSet
 	}{
 		{
-			name:   "no annotation",
-			ObjectMeta: metav1.ObjectMeta{Annotations: map[string]string{
-
-			}},
-			want:   nil,
+			name:       "no annotation",
+			ObjectMeta: metav1.ObjectMeta{Annotations: map[string]string{}},
+			want:       nil,
 		},
 		{
-			name:       "single value",
+			name: "single value",
 			ObjectMeta: metav1.ObjectMeta{Annotations: map[string]string{
 				SuspendAnnotation: "a",
 			}},
-			want:       set.Make("a"),
+			want: set.Make("a"),
 		},
 		{
-			name:       "multi value",
+			name: "multi value",
 			ObjectMeta: metav1.ObjectMeta{Annotations: map[string]string{
 				SuspendAnnotation: "a,b,c",
 			}},
-			want:       set.Make("a", "b", "c"),
+			want: set.Make("a", "b", "c"),
 		},
 		{
-			name:       "multi value with whitespace",
+			name: "multi value with whitespace",
 			ObjectMeta: metav1.ObjectMeta{Annotations: map[string]string{
 				SuspendAnnotation: "a , b , c",
 			}},
-			want:       set.Make("a", "b", "c"),
+			want: set.Make("a", "b", "c"),
 		},
 	}
 	for _, tt := range tests {

pkg/controller/elasticsearch/configmap/configmap.go

@@ -49,8 +49,8 @@ func ReconcileScriptsConfigMap(ctx context.Context, c k8s.Client, es esv1.Elasti
 			nodespec.ReadinessProbeScriptConfigKey: nodespec.ReadinessProbeScript,
 			nodespec.PreStopHookScriptConfigKey:    nodespec.PreStopHookScript,
 			initcontainer.PrepareFsScriptConfigKey: fsScript,
-			initcontainer.SuspendScriptConfigKey: initcontainer.SuspendScript,
-			initcontainer.SuspendedHostsFile: es.Annotations[esv1.SuspendAnnotation],
+			initcontainer.SuspendScriptConfigKey:   initcontainer.SuspendScript,
+			initcontainer.SuspendedHostsFile:       es.Annotations[esv1.SuspendAnnotation],
 		},
 	)
 

pkg/controller/elasticsearch/driver/driver.go

@@ -267,7 +267,7 @@ func (d *defaultDriver) Reconcile(ctx context.Context) *reconciler.Results {
 	// we want to reconcile suspended Pods before we start reconciling node specs as this is considered a debugging and
 	// troubleshooting tool that does not follow the change budget restrictions
 	// TODO: does this need to happen after all expectations are met?
-	if err := reconcileSuspendedPods(d.Client, d.ES); err != nil {
+	if err := reconcileSuspendedPods(d.Client, d.ES, d.Expectations); err != nil {
 		return results.WithError(err)
 	}
 

pkg/controller/elasticsearch/driver/esstate.go

@@ -6,9 +6,10 @@ package driver
 
 import (
 	"context"
-	"github.com/elastic/cloud-on-k8s/pkg/utils/set"
 	"sync"
 
+	"github.com/elastic/cloud-on-k8s/pkg/utils/set"
+
 	esclient "github.com/elastic/cloud-on-k8s/pkg/controller/elasticsearch/client"
 	"github.com/elastic/cloud-on-k8s/pkg/utils/stringsutil"
 )

pkg/controller/elasticsearch/driver/esstate_test.go

@@ -6,9 +6,10 @@ package driver
 
 import (
 	"context"
-	"github.com/elastic/cloud-on-k8s/pkg/utils/set"
 	"testing"
 
+	"github.com/elastic/cloud-on-k8s/pkg/utils/set"
+
 	"github.com/stretchr/testify/require"
 
 	esv1 "github.com/elastic/cloud-on-k8s/pkg/apis/elasticsearch/v1"
@@ -141,7 +142,6 @@ func Test_memoizingNodes_NodesInCluster(t *testing.T) {
 	inCluster, err = memoizingNodes.NodesInCluster([]string{"a", "b", "c", "e"})
 	require.NoError(t, err)
 	require.True(t, inCluster)
-
 }
 
 func Test_memoizingShardsAllocationEnabled_ShardAllocationsEnabled(t *testing.T) {

pkg/controller/elasticsearch/driver/suspend.go

@@ -6,40 +6,62 @@ package driver
 
 import (
 	"context"
+	"github.com/elastic/cloud-on-k8s/pkg/controller/common/annotation"
+	"github.com/elastic/cloud-on-k8s/pkg/controller/common/expectations"
+	"github.com/elastic/cloud-on-k8s/pkg/controller/elasticsearch/initcontainer"
+
 	esv1 "github.com/elastic/cloud-on-k8s/pkg/apis/elasticsearch/v1"
 	"github.com/elastic/cloud-on-k8s/pkg/controller/elasticsearch/sset"
 	"github.com/elastic/cloud-on-k8s/pkg/utils/k8s"
 	corev1 "k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
-func reconcileSuspendedPods(c k8s.Client, es esv1.Elasticsearch) error {
+func reconcileSuspendedPods(c k8s.Client, es esv1.Elasticsearch, e *expectations.Expectations) error {
+	if satisfied, err := e.DeletionsSatisfied(); err != nil || !satisfied {
+		if !satisfied {
+			log.Info("Not reconciling Pod suspensions as deletion expectations are not satisfied")
+		}
+		return err
+	}
+
 	suspendedPodNames := es.SuspendedPodNames()
 
 	statefulSets, err := sset.RetrieveActualStatefulSets(c, k8s.ExtractNamespacedName(&es))
 	if err != nil {
 		return err
 	}
-	knownPodNames := statefulSets.PodNames()
+	knownPods, err := statefulSets.GetActualPods(c)
+	if err != nil {
+		return err
+	}
 
-	for _, podName := range knownPodNames {
-		if suspendedPodNames.Has(podName) {
-			var pod corev1.Pod
-			if err := c.Get(context.Background(), types.NamespacedName{Namespace: es.Namespace, Name: podName}, &pod); err != nil {
-				return err
-			}
+	for _, pod := range knownPods {
+		if suspendedPodNames.Has(pod.Name) {
 			for _, s := range pod.Status.ContainerStatuses {
 				// delete the Pod without grace period if the main container is running
 				if s.Name == esv1.ElasticsearchContainerName && s.State.Running != nil {
 					log.Info("Deleting suspended pod", "pod_name", pod.Name, "pod_uid", pod.UID,
 						"namespace", es.Namespace, "es_name", es.Name)
+					e.ExpectDeletion(pod)
 					if err := c.Delete(context.Background(), &pod, client.GracePeriodSeconds(0)); err != nil {
 						return err
 					}
 				}
 			}
+		} else if isSuspended(pod) {
+			// try to speed up propagation of configmap entries 
+			annotation.MarkPodAsUpdated(c, pod)
 		}
 	}
 	return nil
 }
+
+func isSuspended(pod corev1.Pod) bool {
+	for _, s := range pod.Status.InitContainerStatuses {
+		if s.Name == initcontainer.SuspendContainerName && s.State.Running != nil {
+			return true
+		}
+	}
+	return false
+}

pkg/controller/elasticsearch/driver/upgrade_predicates.go

@@ -6,6 +6,7 @@ package driver
 
 import (
 	"context"
+
 	"github.com/elastic/cloud-on-k8s/pkg/utils/set"
 
 	esv1 "github.com/elastic/cloud-on-k8s/pkg/apis/elasticsearch/v1"

pkg/controller/elasticsearch/initcontainer/initcontainer.go

@@ -15,7 +15,6 @@ const (
 	PrepareFilesystemContainerName = "elastic-internal-init-filesystem"
 	// SuspendContainerName is the name of the container that is used to suspend Elasticsearch if requested by the user.
 	SuspendContainerName = "elastic-internal-suspend"
-
 )
 
 // NewInitContainers creates init containers according to the given parameters

pkg/controller/elasticsearch/initcontainer/initcontainer_test.go

@@ -14,7 +14,7 @@ import (
 
 func TestNewInitContainers(t *testing.T) {
 	type args struct {
-		keystoreResources  *keystore.Resources
+		keystoreResources *keystore.Resources
 	}
 	tests := []struct {
 		name                       string
@@ -24,14 +24,14 @@ func TestNewInitContainers(t *testing.T) {
 		{
 			name: "with keystore resources",
 			args: args{
-				keystoreResources:  &keystore.Resources{},
+				keystoreResources: &keystore.Resources{},
 			},
 			expectedNumberOfContainers: 3,
 		},
 		{
-			name:                       "without keystore resources",
-			args:                       args{
-				keystoreResources:  nil,
+			name: "without keystore resources",
+			args: args{
+				keystoreResources: nil,
 			},
 			expectedNumberOfContainers: 2,
 		},

pkg/controller/elasticsearch/initcontainer/suspend.go

@@ -6,12 +6,13 @@ package initcontainer
 
 import (
 	"fmt"
+	"path"
+
 	esv1 "github.com/elastic/cloud-on-k8s/pkg/apis/elasticsearch/v1"
 	"github.com/elastic/cloud-on-k8s/pkg/controller/common/defaults"
 	esvolume "github.com/elastic/cloud-on-k8s/pkg/controller/elasticsearch/volume"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
-	"path"
 )
 
 const (
@@ -45,6 +46,6 @@ func NewSuspendInitContainer() corev1.Container {
 		Name:            SuspendContainerName,
 		Env:             defaults.PodDownwardEnvVars(),
 		Command:         []string{"bash", "-c", path.Join(esvolume.ScriptsVolumeMountPath, SuspendScriptConfigKey)},
-		Resources: suspendContainerResources,
+		Resources:       suspendContainerResources,
 	}
 }

pkg/utils/set/set.go

@@ -34,7 +34,7 @@ func (set StringSet) MergeWith(other StringSet) {
 	}
 }
 
-func (set StringSet) Diff( other StringSet) StringSet {
+func (set StringSet) Diff(other StringSet) StringSet {
 	res := Make()
 	res.MergeWith(set)
 	for str := range other {