k8s: set Status.Version when all pods have that version

src/go/k8s/controllers/redpanda/cluster_controller.go

@@ -669,7 +669,15 @@ func (r *ClusterReconciler) reportStatus(
 			cluster.Status.Nodes = *nodeList
 			cluster.Status.ReadyReplicas = sts.LastObservedState.Status.ReadyReplicas
 			cluster.Status.Replicas = sts.LastObservedState.Status.Replicas
-			cluster.Status.Version = sts.Version()
+			version, err := sts.CurrentVersion(ctx)
+			if err != nil {
+				// this is non-fatal error, it will return error even if e.g.
+				// the rollout is not finished because then the currentversion
+				// of the cluster cannot be determined
+				r.Log.Info(fmt.Sprintf("cannot set Status.Version, %s", err))
+			} else {
+				cluster.Status.Version = version
+			}
 
 			err = r.Status().Update(ctx, &cluster)
 			if err == nil {

src/go/k8s/pkg/resources/statefulset.go

@@ -864,26 +864,55 @@ func (r *StatefulSetResource) fullConfiguratorImage() string {
 	return fmt.Sprintf("%s:%s", r.configuratorSettings.ConfiguratorBaseImage, r.configuratorSettings.ConfiguratorTag)
 }
 
-// Version returns the cluster version specified in the image tag.
+// Version returns the cluster version specified in the image tag of the
+// statefulset spec. Depending on the rollout status it might not be the same as
+// version of the pods
 func (r *StatefulSetResource) Version() string {
 	lastObservedSts := r.LastObservedState
 	if lastObservedSts != nil {
-		cc := lastObservedSts.Spec.Template.Spec.Containers
-		for i := range cc {
-			c := cc[i]
-			if c.Name != redpandaContainerName {
-				continue
-			}
-			// Will always have tag even for latest because of pandaCluster.FullImageName().
-			if s := strings.Split(c.Image, ":"); len(s) > 1 {
-				version := s[len(s)-1]
-				// Image uses registry with port and no tag (e.g. localhost:5000/redpanda)
-				if strings.Contains(version, "/") {
-					version = ""
-				}
-				return version
+		return redpandaContainerVersion(lastObservedSts.Spec.Template.Spec.Containers)
+	}
+	return ""
+}
+
+func redpandaContainerVersion(containers []corev1.Container) string {
+	for i := range containers {
+		c := containers[i]
+		if c.Name != redpandaContainerName {
+			continue
+		}
+		// Will always have tag even for latest because of pandaCluster.FullImageName().
+		if s := strings.Split(c.Image, ":"); len(s) > 1 {
+			version := s[len(s)-1]
+			// Image uses registry with port and no tag (e.g. localhost:5000/redpanda)
+			if strings.Contains(version, "/") {
+				version = ""
 			}
+			return version
 		}
 	}
 	return ""
 }
+
+// CurrentVersion is the version that's rolled out to all nodes (pods) of the cluster
+func (r *StatefulSetResource) CurrentVersion(ctx context.Context) (string, error) {
+	stsVersion := r.Version()
+	if stsVersion == "" {
+		return "", nil
+	}
+	replicas := *r.LastObservedState.Spec.Replicas
+	pods, err := r.getPodList(ctx)
+	if err != nil {
+		return "", err
+	}
+	if int32(len(pods.Items)) != replicas {
+		return stsVersion, fmt.Errorf("rollout incomplete: pods count %d does not match expected replicas %d", len(pods.Items), replicas)
+	}
+	for _, p := range pods.Items {
+		podVersion := redpandaContainerVersion(p.Spec.Containers)
+		if podVersion != stsVersion {
+			return stsVersion, fmt.Errorf("rollout incomplete: at least one pod has version %s not %s", podVersion, stsVersion)
+		}
+	}
+	return stsVersion, nil
+}

src/go/k8s/pkg/resources/statefulset_test.go

@@ -29,6 +29,7 @@ import (
 
 	redpandav1alpha1 "github.com/redpanda-data/redpanda/src/go/k8s/apis/redpanda/v1alpha1"
 	adminutils "github.com/redpanda-data/redpanda/src/go/k8s/pkg/admin"
+	"github.com/redpanda-data/redpanda/src/go/k8s/pkg/labels"
 	res "github.com/redpanda-data/redpanda/src/go/k8s/pkg/resources"
 	resourcetypes "github.com/redpanda-data/redpanda/src/go/k8s/pkg/resources/types"
 )
@@ -343,3 +344,94 @@ func TestVersion(t *testing.T) {
 		assert.Equal(t, tt.ExpectedVersion, sts.Version())
 	}
 }
+
+func TestCurrentVersion(t *testing.T) {
+	redpandaContainerName := "redpanda"
+	redpanda := pandaCluster()
+	tests := []struct {
+		name             string
+		pods             []corev1.Pod
+		expectedReplicas int32
+		expectedVersion  string
+		shouldError      bool
+	}{
+		{"one pod with matching version", []corev1.Pod{
+			{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test",
+					Namespace: redpanda.Namespace,
+				},
+				Spec: corev1.PodSpec{Containers: []corev1.Container{{Name: redpandaContainerName, Image: "vectorized/redpanda:v21.11.11"}}},
+			},
+		}, 1, "v21.11.11", false},
+		{"one pod with matching version but expecting two replicas", []corev1.Pod{
+			{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test",
+					Namespace: redpanda.Namespace,
+				},
+				Spec: corev1.PodSpec{Containers: []corev1.Container{{Name: redpandaContainerName, Image: "vectorized/redpanda:v21.11.11"}}},
+			},
+		}, 2, "v21.11.11", true},
+		{"one pod with matching and one pod with non-matching version", []corev1.Pod{
+			{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test",
+					Namespace: redpanda.Namespace,
+				},
+				Spec: corev1.PodSpec{Containers: []corev1.Container{{Name: redpandaContainerName, Image: "vectorized/redpanda:v21.11.11"}}},
+			},
+			{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test2",
+					Namespace: redpanda.Namespace,
+				},
+				Spec: corev1.PodSpec{Containers: []corev1.Container{{Name: redpandaContainerName, Image: "vectorized/redpanda:v22.2.2"}}},
+			},
+		}, 2, "v21.11.11", true},
+	}
+
+	for _, tt := range tests {
+		c := fake.NewClientBuilder().Build()
+		for _, p := range tt.pods {
+			p.Labels = labels.ForCluster(redpanda)
+			assert.NoError(t, c.Create(context.TODO(), &p))
+		}
+		sts := res.NewStatefulSet(c, redpanda, scheme.Scheme,
+			"cluster.local",
+			"servicename",
+			types.NamespacedName{Name: "test", Namespace: "test"},
+			TestStatefulsetTLSVolumeProvider{},
+			TestAdminTLSConfigProvider{},
+			"",
+			res.ConfiguratorSettings{
+				ConfiguratorBaseImage: "vectorized/configurator",
+				ConfiguratorTag:       "latest",
+				ImagePullPolicy:       "Always",
+			},
+			func(ctx context.Context) (string, error) { return hash, nil },
+			func(ctx context.Context, k8sClient client.Reader, redpandaCluster *redpandav1alpha1.Cluster, fqdn string, adminTLSProvider resourcetypes.AdminTLSConfigProvider, ordinals ...int32) (adminutils.AdminAPIClient, error) {
+				return nil, nil
+			},
+			time.Second,
+			ctrl.Log.WithName("test"),
+			0)
+		sts.LastObservedState = &v1.StatefulSet{
+			Spec: v1.StatefulSetSpec{
+				Replicas: &tt.expectedReplicas,
+				Template: corev1.PodTemplateSpec{
+					Spec: corev1.PodSpec{
+						Containers: []corev1.Container{{Name: redpandaContainerName, Image: "vectorized/redpanda:v21.11.11"}},
+					},
+				},
+			},
+		}
+		v, err := sts.CurrentVersion(context.TODO())
+		assert.Equal(t, tt.expectedVersion, v)
+		if tt.shouldError {
+			assert.Error(t, err)
+		} else {
+			assert.NoError(t, err)
+		}
+	}
+}

src/go/k8s/pkg/resources/statefulset_update.go

@@ -136,21 +136,29 @@ func (r *StatefulSetResource) isClusterHealthy(ctx context.Context) error {
 	return nil
 }
 
-func (r *StatefulSetResource) rollingUpdate(
-	ctx context.Context, template *corev1.PodTemplateSpec,
-) error {
+func (r *StatefulSetResource) getPodList(ctx context.Context) (*corev1.PodList, error) {
 	var podList corev1.PodList
 	err := r.List(ctx, &podList, &k8sclient.ListOptions{
 		Namespace:     r.pandaCluster.Namespace,
 		LabelSelector: labels.ForCluster(r.pandaCluster).AsClientSelector(),
 	})
 	if err != nil {
-		return fmt.Errorf("unable to list panda pods: %w", err)
+		return nil, fmt.Errorf("unable to list panda pods: %w", err)
 	}
 
 	sort.Slice(podList.Items, func(i, j int) bool {
 		return podList.Items[i].Name < podList.Items[j].Name
 	})
+	return &podList, nil
+}
+
+func (r *StatefulSetResource) rollingUpdate(
+	ctx context.Context, template *corev1.PodTemplateSpec,
+) error {
+	podList, err := r.getPodList(ctx)
+	if err != nil {
+		return fmt.Errorf("error getting pods %w", err)
+	}
 
 	var artificialPod corev1.Pod
 	artificialPod.Annotations = template.Annotations