core: reference all containers in a spec by name instead of index

This changeset excludes converting _test.go code as there may be a
legitimate reasons (E.g. convenience) for unit test to be sensitive to
ordering.

Related to:
- https://github.com/rook/rook/issues/15291
- https://github.com/rook/rook/pull/16969

Signed-off-by: Joshua Hoblitt <josh@hoblitt.com>
(cherry picked from commit d059c351fc)

PendingReleaseNotes.md

@@ -14,3 +14,4 @@
 - SSE-S3 with Vault Agent: Added support for server-side encryption with SSE-S3 using HashiCorp Vault Agent authentication. See the [CephObjectStore Security Settings](Documentation/CRDs/Object-Storage/ceph-object-store-crd.md#sse-s3-with-vault-agent) for more details.
 - Unused CRUSH rule cleanup: Rook now deletes unused CRUSH rules by default after the Ceph mgr starts. Set `ROOK_DELETE_UNUSED_CRUSH_RULES` to `false` in the operator config to disable this cleanup.
 - Declare stable the feature to concurrently reconcile multiple Ceph Clusters with the setting `ROOK_RECONCILE_CONCURRENT_CLUSTERS`.
+- Containers within a pod are now consistently reconciled by name instead of relying on the order in which they are declared.  This is a defensive measure against the declaration order changing due to manipulation by a mutating webhook.

cmd/rook/ceph/operator.go

@@ -64,7 +64,7 @@ func startOperator(cmd *cobra.Command, args []string) error {
 		rook.TerminateFatal(errors.Errorf("rook operator namespace is not provided. expose it via downward API in the rook operator manifest file using environment variable %q", k8sutil.PodNamespaceEnvVar))
 	}
 
-	rook.CheckOperatorResources(cmd.Context(), context.Clientset)
+	rook.CheckOperatorResources(cmd.Context(), context.Clientset, containerName)
 	rookImage := rook.GetOperatorImage(cmd.Context(), context.Clientset, containerName)
 	rookBaseImageCephVersion, err := rook.GetOperatorBaseImageCephVersion(context)
 	if err != nil {

cmd/rook/rook/rook.go

@@ -165,13 +165,14 @@ func GetOperatorServiceAccount(ctx context.Context, clientset kubernetes.Interfa
 	return pod.Spec.ServiceAccountName
 }
 
-func CheckOperatorResources(ctx context.Context, clientset kubernetes.Interface) {
+func CheckOperatorResources(ctx context.Context, clientset kubernetes.Interface, containerName string) {
 	// Getting the info of the operator pod
 	pod, err := k8sutil.GetRunningPod(ctx, clientset)
 	TerminateOnError(err, "failed to get pod")
-	resource := pod.Spec.Containers[0].Resources
+	container, err := k8sutil.GetContainerByName(pod.Spec.Containers, containerName)
+	TerminateOnError(err, "failed to find operator container in pod")
 	// set env var if operator pod resources are set
-	if !reflect.DeepEqual(resource, (v1.ResourceRequirements{})) {
+	if !reflect.DeepEqual(container.Resources, (v1.ResourceRequirements{})) {
 		os.Setenv("OPERATOR_RESOURCES_SPECIFIED", "true")
 	}
 }

pkg/operator/ceph/cluster/mon/mon.go

@@ -735,10 +735,14 @@ func scheduleMonitor(c *Cluster, mon *monConfig) (*apps.Deployment, error) {
 	// avoid issues with the real deployment, the canary should be careful not
 	// to modify the storage by instead running an innocuous command.
 	d.Spec.Template.Spec.InitContainers = []corev1.Container{}
-	d.Spec.Template.Spec.Containers[0].Image = c.rookImage
+	monContainer, err := k8sutil.GetContainerByName(d.Spec.Template.Spec.Containers, monContainerName)
+	if err != nil {
+		return nil, errors.Wrapf(err, "failed to find %q container in mon canary deployment", monContainerName)
+	}
+	monContainer.Image = c.rookImage
 	// As PID 1, sleep will not capture the SIGTERM signal, so use a bash
 	// trap to exit cleanly and quickly when the pod is deleted.
-	d.Spec.Template.Spec.Containers[0].Command = []string{
+	monContainer.Command = []string{
 		"/bin/bash",
 		"-c",
 		`trap 'exit' SIGTERM
@@ -746,8 +750,8 @@ func scheduleMonitor(c *Cluster, mon *monConfig) (*apps.Deployment, error) {
 	}
 
 	// remove the startup and liveness probes on the canary pod
-	d.Spec.Template.Spec.Containers[0].StartupProbe = nil
-	d.Spec.Template.Spec.Containers[0].LivenessProbe = nil
+	monContainer.StartupProbe = nil
+	monContainer.LivenessProbe = nil
 
 	// setup affinity settings for pod scheduling
 	p := c.getMonPlacement(mon.Zone)

pkg/operator/ceph/cluster/mon/spec.go

@@ -48,6 +48,8 @@ var floatingMonDeploymentDrbdTemplatePath string
 const (
 	// Full path of the command used to invoke the Ceph mon daemon
 	cephMonCommand = "ceph-mon"
+	// Name of the main mon container in the mon deployment pod spec
+	monContainerName = "mon"
 )
 
 func (c *Cluster) getLabels(monConfig *monConfig, canary, includeNewLabels bool) map[string]string {
@@ -309,7 +311,7 @@ func (c *Cluster) makeMonDaemonContainer(monConfig *monConfig) corev1.Container
 	podIPEnvVar := "ROOK_POD_IP"
 
 	container := corev1.Container{
-		Name: "mon",
+		Name: monContainerName,
 		Command: []string{
 			cephMonCommand,
 		},

pkg/operator/ceph/controller/network.go

@@ -275,7 +275,11 @@ func discoverAddressRanges(
 	// set up net status vol from downward api, plus init container to wait for net status to be available
 	netStatusVol, netStatusMount := networkStatusVolumeAndMount()
 	job.Spec.Template.Spec.Volumes = append(job.Spec.Template.Spec.Volumes, netStatusVol)
-	job.Spec.Template.Spec.Containers[0].VolumeMounts = append(job.Spec.Template.Spec.Containers[0].VolumeMounts, netStatusMount)
+	canaryContainer, err := k8sutil.GetContainerByName(job.Spec.Template.Spec.Containers, cmdreporter.CmdReporterContainerName)
+	if err != nil {
+		return ranges, errors.Wrapf(err, "failed to find %q container in ceph %q network canary job", cmdreporter.CmdReporterContainerName, cephNetwork)
+	}
+	canaryContainer.VolumeMounts = append(canaryContainer.VolumeMounts, netStatusMount)
 	job.Spec.Template.Spec.InitContainers = append(job.Spec.Template.Spec.InitContainers, containerWaitForNetworkStatus(clusterSpec, rookImage))
 
 	stdout, stderr, retcode, err := networkCanary.Run(ctx, detectNetworkCIDRTimeout)

pkg/operator/ceph/nfs/security.go

@@ -19,8 +19,10 @@ package nfs
 import (
 	"fmt"
 
+	"github.com/pkg/errors"
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	"github.com/rook/rook/pkg/operator/ceph/controller"
+	"github.com/rook/rook/pkg/operator/k8sutil"
 	"github.com/rook/rook/pkg/util/log"
 	v1 "k8s.io/api/core/v1"
 )
@@ -35,18 +37,22 @@ func (r *ReconcileCephNFS) addSecurityConfigsToPod(nfs *cephv1.CephNFS, pod *v1.
 
 	if sec.SSSD != nil {
 		log.NamedDebug(nsName, logger, "configuring system security services daemon (SSSD) for CephNFS")
-		addSSSDConfigsToPod(r, nfs, pod)
+		if err := addSSSDConfigsToPod(r, nfs, pod); err != nil {
+			return err
+		}
 	}
 
 	if sec.Kerberos != nil {
 		log.NamedDebug(nsName, logger, "configuring Kerberos for CephNFS")
-		addKerberosConfigsToPod(r, nfs, pod)
+		if err := addKerberosConfigsToPod(r, nfs, pod); err != nil {
+			return err
+		}
 	}
 
 	return nil
 }
 
-func addSSSDConfigsToPod(r *ReconcileCephNFS, nfs *cephv1.CephNFS, pod *v1.PodSpec) {
+func addSSSDConfigsToPod(r *ReconcileCephNFS, nfs *cephv1.CephNFS, pod *v1.PodSpec) error {
 	nsName := controller.NsName(nfs.Namespace, nfs.Name)
 
 	// generate /etc/nsswitch.conf file for the nfs-ganesha pod
@@ -54,8 +60,11 @@ func addSSSDConfigsToPod(r *ReconcileCephNFS, nfs *cephv1.CephNFS, pod *v1.PodSp
 
 	pod.InitContainers = append(pod.InitContainers, *nssCfgInitContainer)
 	pod.Volumes = append(pod.Volumes, *nssCfgVol)
-	// assume the first container is the NFS-Ganesha container
-	pod.Containers[0].VolumeMounts = append(pod.Containers[0].VolumeMounts, *nssCfgMount)
+	ganesha, err := k8sutil.GetContainerByName(pod.Containers, ganeshaContainerName)
+	if err != nil {
+		return errors.Wrapf(err, "failed to add SSSD config")
+	}
+	ganesha.VolumeMounts = append(ganesha.VolumeMounts, *nssCfgMount)
 
 	sidecarCfg := nfs.Spec.Security.SSSD.Sidecar
 	if sidecarCfg != nil {
@@ -65,19 +74,27 @@ func addSSSDConfigsToPod(r *ReconcileCephNFS, nfs *cephv1.CephNFS, pod *v1.PodSp
 		pod.InitContainers = append(pod.InitContainers, *init)
 		pod.Containers = append(pod.Containers, *sidecar)
 		pod.Volumes = append(pod.Volumes, vols...)
-		// assume the first container is the NFS-Ganesha container
-		pod.Containers[0].VolumeMounts = append(pod.Containers[0].VolumeMounts, mounts...)
+		// re-look up the container; appending the sidecar may have reallocated pod.Containers
+		ganesha, err = k8sutil.GetContainerByName(pod.Containers, ganeshaContainerName)
+		if err != nil {
+			return errors.Wrapf(err, "failed to add SSSD sidecar mounts")
+		}
+		ganesha.VolumeMounts = append(ganesha.VolumeMounts, mounts...)
 	}
+	return nil
 }
 
-func addKerberosConfigsToPod(r *ReconcileCephNFS, nfs *cephv1.CephNFS, pod *v1.PodSpec) {
+func addKerberosConfigsToPod(r *ReconcileCephNFS, nfs *cephv1.CephNFS, pod *v1.PodSpec) error {
 	init, volume, ganeshaMounts := generateKrbConfResources(r, nfs)
 
 	pod.InitContainers = append(pod.InitContainers, *init)
 	pod.Volumes = append(pod.Volumes, *volume)
-	// assume the first container is the NFS-Ganesha container
+	ganesha, err := k8sutil.GetContainerByName(pod.Containers, ganeshaContainerName)
+	if err != nil {
+		return errors.Wrapf(err, "failed to add Kerberos config")
+	}
 	for _, m := range ganeshaMounts {
-		pod.Containers[0].VolumeMounts = append(pod.Containers[0].VolumeMounts, *m)
+		ganesha.VolumeMounts = append(ganesha.VolumeMounts, *m)
 	}
 
 	configVolSrc := nfs.Spec.Security.Kerberos.ConfigFiles.VolumeSource
@@ -85,7 +102,7 @@ func addKerberosConfigsToPod(r *ReconcileCephNFS, nfs *cephv1.CephNFS, pod *v1.P
 		vol, mnt := kerberosConfigFilesVolAndMount(*configVolSrc)
 
 		pod.Volumes = append(pod.Volumes, vol)
-		pod.Containers[0].VolumeMounts = append(pod.Containers[0].VolumeMounts, mnt)
+		ganesha.VolumeMounts = append(ganesha.VolumeMounts, mnt)
 	}
 
 	keytabVolSrc := nfs.Spec.Security.Kerberos.KeytabFile.VolumeSource
@@ -93,8 +110,9 @@ func addKerberosConfigsToPod(r *ReconcileCephNFS, nfs *cephv1.CephNFS, pod *v1.P
 		vol, mnt := keytabVolAndMount(*keytabVolSrc)
 
 		pod.Volumes = append(pod.Volumes, vol)
-		pod.Containers[0].VolumeMounts = append(pod.Containers[0].VolumeMounts, mnt)
+		ganesha.VolumeMounts = append(ganesha.VolumeMounts, mnt)
 	}
+	return nil
 }
 
 func generateSssdSidecarResources(nfs *cephv1.CephNFS, sidecarCfg *cephv1.SSSDSidecar) (

pkg/operator/ceph/nfs/spec.go

@@ -39,6 +39,7 @@ import (
 const (
 	// AppName is the name of the app
 	AppName               = "rook-ceph-nfs"
+	ganeshaContainerName  = "nfs-ganesha"
 	ganeshaConfigVolume   = "ganesha-config"
 	nfsPort               = 2049
 	ganeshaPid            = "/var/run/ganesha/ganesha.pid"
@@ -233,7 +234,7 @@ func (r *ReconcileCephNFS) daemonContainer(nfs *cephv1.CephNFS, cfg daemonConfig
 	}
 
 	container := v1.Container{
-		Name: "nfs-ganesha",
+		Name: ganeshaContainerName,
 		Command: []string{
 			"ganesha.nfsd",
 		},

pkg/operator/ceph/object/spec.go

@@ -47,6 +47,7 @@ import (
 
 const (
 	serviceAccountName = "rook-ceph-rgw"
+	rgwContainerName   = "rgw"
 	sseKMS             = "ssekms"
 	sseS3              = "sses3"
 	vaultPrefix        = "/v1/"
@@ -331,7 +332,11 @@ func (c *clusterConfig) makeRGWPodSpec(rgwConfig *rgwConfig) (v1.PodTemplateSpec
 
 	addVols, addMounts := c.store.Spec.Gateway.AdditionalVolumeMounts.GenerateVolumesAndMounts("/var/rgw/")
 	podTemplateSpec.Spec.Volumes = append(podTemplateSpec.Spec.Volumes, addVols...)
-	podTemplateSpec.Spec.Containers[0].VolumeMounts = append(podTemplateSpec.Spec.Containers[0].VolumeMounts, addMounts...)
+	rgwContainer, err := k8sutil.GetContainerByName(podTemplateSpec.Spec.Containers, rgwContainerName)
+	if err != nil {
+		return podTemplateSpec, errors.Wrapf(err, "failed to find %q container in RGW pod spec", rgwContainerName)
+	}
+	rgwContainer.VolumeMounts = append(rgwContainer.VolumeMounts, addMounts...)
 
 	return podTemplateSpec, nil
 }
@@ -420,7 +425,7 @@ func (c *clusterConfig) makeDaemonContainer(rgwConfig *rgwConfig) (v1.Container,
 	}
 
 	container := v1.Container{
-		Name:            "rgw",
+		Name:            rgwContainerName,
 		Image:           c.clusterSpec.CephVersion.Image,
 		ImagePullPolicy: controller.GetContainerImagePullPolicy(c.clusterSpec.CephVersion.ImagePullPolicy),
 		Command: []string{

pkg/operator/discover/discover.go

@@ -236,8 +236,11 @@ func (d *Discover) createDiscoverDaemonSet(ctx context.Context, namespace, disco
 	}
 
 	if opcontroller.LoopDevicesAllowed() {
-		ds.Spec.Template.Spec.Containers[0].Env = append(ds.Spec.Template.Spec.Containers[0].Env,
-			v1.EnvVar{Name: "CEPH_VOLUME_ALLOW_LOOP_DEVICES", Value: "true"})
+		c, err := k8sutil.GetContainerByName(ds.Spec.Template.Spec.Containers, discoverDaemonsetName)
+		if err != nil {
+			return fmt.Errorf("failed to find rook-discover container in daemonset spec. %+v", err)
+		}
+		c.Env = append(c.Env, v1.EnvVar{Name: "CEPH_VOLUME_ALLOW_LOOP_DEVICES", Value: "true"})
 	}
 
 	_, err = d.clientset.AppsV1().DaemonSets(namespace).Create(ctx, ds, metav1.CreateOptions{})

pkg/operator/k8sutil/pod.go

@@ -96,28 +96,22 @@ func ConfigDirEnvVar(dataDir string) v1.EnvVar {
 	return v1.EnvVar{Name: "ROOK_CONFIG_DIR", Value: dataDir}
 }
 
-// GetContainerImage returns the container image
-// matching the given name for a pod. If the pod
-// only has a single container, the name argument
-// is ignored.
+// GetContainerImage returns the container image matching the given name for a pod.
 func GetContainerImage(pod *v1.Pod, name string) (string, error) {
 	return GetSpecContainerImage(pod.Spec, name, false)
 }
 
-// GetSpecContainerImage returns the container image
-// for a podspec, given a container name. The name is
-// ignored if the podspec has a single container, in
-// which case the image for that container is returned.
+// GetSpecContainerImage returns the container image for a podspec, given a container name.
 func GetSpecContainerImage(spec v1.PodSpec, name string, initContainer bool) (string, error) {
 	containers := spec.Containers
 	if initContainer {
 		containers = spec.InitContainers
 	}
-	image, err := GetMatchingContainer(containers, name)
+	container, err := GetContainerByName(containers, name)
 	if err != nil {
 		return "", err
 	}
-	return image.Image, nil
+	return container.Image, nil
 }
 
 // Replaces the pod default toleration of 300s used when the node controller
@@ -174,32 +168,17 @@ func GetRunningPod(ctx context.Context, clientset kubernetes.Interface) (*v1.Pod
 	return pod, nil
 }
 
-// GetMatchingContainer takes a list of containers and a name,
-// and returns the first container in the list matching the
-// name. If the list contains a single container it is always
-// returned, even if the name does not match.
-func GetMatchingContainer(containers []v1.Container, name string) (v1.Container, error) {
-	var result *v1.Container
-	if len(containers) == 1 {
-		// if there is only one pod, use its image rather than require a set container name
-		// #nosec G602 -- len(containers) == 1 is checked above so index 0 is always valid
-		result = &containers[0]
-	} else {
-		// if there are multiple pods, we require the container to have the expected name
-		for _, container := range containers {
-			if container.Name == name {
-				localcontainer := container
-				result = &localcontainer
-				break
-			}
+// GetContainerByName returns a pointer to the container in the slice whose
+// Name matches the given name. The returned pointer references the slice
+// element, so callers may mutate the container in place. Returns an error
+// if no container matches.
+func GetContainerByName(containers []v1.Container, name string) (*v1.Container, error) {
+	for i := range containers {
+		if containers[i].Name == name {
+			return &containers[i], nil
 		}
 	}
-
-	if result == nil {
-		return v1.Container{}, fmt.Errorf("failed to find image for container %s", name)
-	}
-
-	return *result, nil
+	return nil, fmt.Errorf("container %q not found", name)
 }
 
 // PodsRunningWithLabel returns the number of running pods with the given label

pkg/operator/k8sutil/pod_test.go

@@ -27,31 +27,37 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
-func TestGetContainerInPod(t *testing.T) {
+func TestGetContainerByName(t *testing.T) {
 	expectedName := "mycontainer"
 	imageName := "myimage"
 
-	// no container fails
-	_, err := GetMatchingContainer([]v1.Container{}, expectedName)
+	// empty list returns an error
+	_, err := GetContainerByName([]v1.Container{}, expectedName)
 	assert.NotNil(t, err)
 
-	// one container will allow any name
-	container, err := GetMatchingContainer([]v1.Container{{Name: "foo", Image: imageName}}, expectedName)
-	assert.Nil(t, err)
-	assert.Equal(t, imageName, container.Image)
+	// single non-matching container returns an error
+	_, err = GetContainerByName([]v1.Container{{Name: "foo", Image: imageName}}, expectedName)
+	assert.NotNil(t, err)
 
-	// multiple container fails if we don't find the correct name
-	container, err = GetMatchingContainer(
+	// multiple containers with no match returns an error
+	_, err = GetContainerByName(
 		[]v1.Container{{Name: "foo", Image: imageName}, {Name: "bar", Image: imageName}},
 		expectedName)
 	assert.NotNil(t, err)
 
-	// multiple container succeeds if we find the correct name
-	container, err = GetMatchingContainer(
+	// match returns the container
+	container, err := GetContainerByName(
 		[]v1.Container{{Name: "foo", Image: imageName}, {Name: expectedName, Image: imageName}},
 		expectedName)
 	assert.Nil(t, err)
 	assert.Equal(t, imageName, container.Image)
+
+	// returned pointer references the slice element, so mutations are visible
+	containers := []v1.Container{{Name: expectedName, Image: imageName}}
+	c, err := GetContainerByName(containers, expectedName)
+	assert.Nil(t, err)
+	c.Image = "newimage"
+	assert.Equal(t, "newimage", containers[0].Image)
 }
 
 func TestGetPodPhaseMap(t *testing.T) {