PWX-30419: Use reconcile loop to check pre-flight status. Timeout after 15m.

drivers/storage/portworx/portworx.go

@@ -98,70 +98,116 @@ func (p *portworx) Init(
 }
 
 func (p *portworx) Validate(cluster *corev1.StorageCluster) error {
+
+	condition := &corev1.ClusterCondition{
+		Source: pxutil.PortworxComponentName,
+		Type:   corev1.ClusterConditionTypePreflight,
+	}
+
+	setClusterCondition := func(status corev1.ClusterConditionStatus, msg string) {
+		condition.Status = status
+		condition.Message = msg
+		util.UpdateStorageClusterCondition(cluster, condition)
+	}
+
 	podSpec, err := p.GetStoragePodSpec(cluster, "")
 	if err != nil {
+		err = fmt.Errorf("pre-flight: get storage pod spec: %v", err)
+		setClusterCondition(corev1.ClusterConditionStatusFailed, err.Error())
 		return err
 	}
 
 	preFlighter := NewPreFlighter(cluster, p.k8sClient, podSpec)
 
+	deletePreflight := func() {
+		_, _, err := GetPreFlightPodsFromNamespace(p.k8sClient, cluster.Namespace)
+		if err == nil {
+			// Clean up the pre-flight pods
+			logrus.Infof("pre-flight: cleaning pre-flight ds...")
+			if derr := preFlighter.DeletePreFlight(); derr != nil {
+				logrus.Errorf("pre-flight: error deleting pre-flight: %v", derr)
+			}
+		}
+	}
+
+	check, ok := cluster.Annotations[pxutil.AnnotationPreflightCheck]
+	check = strings.TrimSpace(strings.ToLower(check))
+	if !ok || check == "skip" {
+		setClusterCondition(corev1.ClusterConditionStatusDisabled, "pre-flight: skipped...")
+		logrus.Infof(condition.Message)
+		deletePreflight()
+		return nil
+	}
+
 	// Start the pre-flight container. The pre-flight checks at this time are specific to enabling DMthin
 	err = preFlighter.RunPreFlight()
 	if err != nil {
 		if !errors.IsAlreadyExists(err) {
+			err = fmt.Errorf("pre-flight: run error: %v", err)
+			setClusterCondition(corev1.ClusterConditionStatusFailed, err.Error())
+			deletePreflight()
 			return err
 		}
 		logrus.Debugf("pre-flight: container already running...")
 	}
 
-	defer func() {
-		// Clean up the pre-flight pods
-		logrus.Infof("pre-flight: cleaning pre-flight ds...")
-		if derr := preFlighter.DeletePreFlight(); derr != nil {
-			logrus.Errorf("pre-flight: error deleting pre-flight: %v", derr)
-		}
-	}()
-
-	cnt := 0
-	//  Wait for all the pre-flight pods to finish
-	for {
-		time.Sleep(3 * time.Second) // Pause before status check
-		completed, inProgress, total, err := preFlighter.GetPreFlightStatus()
-		if err != nil {
-			logrus.Errorf("pre-flight: error getting pre-flight status: %v", err)
-			return err
-		}
-		logrus.Infof("pre-flight: Completed [%v] In Progress [%v] Total [%v]", completed, inProgress, total)
-
-		if total != 0 && completed == total {
-			logrus.Infof("pre-flight: completed...")
-			break
+	completed, inProgress, total, age, err := preFlighter.GetPreFlightStatus()
+	if err != nil {
+		if errors.IsNotFound(err) {
+			setClusterCondition(corev1.ClusterConditionStatusInProgress, "pre-flight: has not started yet...")
+			logrus.Infof(condition.Message)
+			return nil
 		}
+		err = fmt.Errorf("pre-flight: error getting pre-flight status: %v", err)
+		setClusterCondition(corev1.ClusterConditionStatusFailed, err.Error())
+		deletePreflight()
+		return err
+	}
 
-		// Add five minute timeout.  If we do reconcile loop check we will need a different way.
-		cnt++
-		if cnt == 200 { // 3s * 100 = 300s (10 mins)
-			err = fmt.Errorf("pre-flight: pre-flight status check timed out")
-			logrus.Errorf("%v", err)
+	logrus.Infof("pre-flight: Completed [%v] In Progress [%v] Total [%v]", completed, inProgress, total)
+	if total != 0 && completed == total {
+		logrus.Infof("pre-flight: checks completed...")
+	} else {
+		if age >= 15*time.Minute {
+			err = fmt.Errorf("pre-flight: pre-flight check timed out")
+			setClusterCondition(corev1.ClusterConditionStatusTimeout, err.Error())
+			deletePreflight()
 			return err
 		}
+		setClusterCondition(corev1.ClusterConditionStatusInProgress,
+			fmt.Sprintf("pre-flight: Operation still in progress: Completed [%v] In Progress [%v] Total [%v]",
+				completed, inProgress, total))
+		logrus.Infof(condition.Message)
+		return nil
 	}
 
 	// Process all the StorageNode.Status.Checks
 	var storageNodes []*corev1.StorageNode
 
 	storageNodes, err = p.storageNodesList(cluster)
-	if err == nil {
-		err = preFlighter.ProcessPreFlightResults(p.recorder, storageNodes)
-		if err != nil {
-			logrus.Errorf("pre-flight: Error processing results: %v", err)
-		}
-	} else {
-		logrus.Errorf("pre-flight incomplete: Error getting storage node list: %v", err)
+	if err != nil {
+		err = fmt.Errorf("pre-flight incomplete: Error getting storage node list: %v", err)
+		setClusterCondition(corev1.ClusterConditionStatusFailed, err.Error())
+		deletePreflight()
+		return err
 	}
 
-	return err
+	logrus.Infof("pre-flight: process pre-flight results...")
+	err = preFlighter.ProcessPreFlightResults(p.recorder, storageNodes)
+	if err != nil {
+		err = fmt.Errorf("pre-flight: Error processing results: %v", err)
+		setClusterCondition(corev1.ClusterConditionStatusFailed, err.Error())
+		deletePreflight()
+		return err
+	}
+
+	setClusterCondition(corev1.ClusterConditionStatusCompleted, "pre-flight: done...")
+	logrus.Infof(condition.Message)
+	deletePreflight()
+
+	return nil
 }
+
 func (p *portworx) initializeComponents() {
 	for _, comp := range component.GetAll() {
 		comp.Initialize(p.k8sClient, *p.k8sVersion, p.scheme, p.recorder)

drivers/storage/portworx/portworx_test.go

@@ -91,6 +91,9 @@ func TestValidate(t *testing.T) {
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      "px-cluster",
 			Namespace: "kube-test",
+			Annotations: map[string]string{
+				pxutil.AnnotationPreflightCheck: "true",
+			},
 		},
 	}
 
@@ -151,6 +154,7 @@ func TestValidate(t *testing.T) {
 		},
 	}
 
+	coreops.SetInstance(coreops.New(fakek8sclient.NewSimpleClientset()))
 	k8sClient := testutil.FakeK8sClient(preflightDS)
 
 	err := k8sClient.Create(context.TODO(), preFlightPod1)
@@ -222,6 +226,9 @@ func TestValidateCheckFailure(t *testing.T) {
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      "px-cluster",
 			Namespace: "kube-test",
+			Annotations: map[string]string{
+				pxutil.AnnotationPreflightCheck: "true",
+			},
 		},
 	}
 
@@ -287,6 +294,7 @@ func TestValidateCheckFailure(t *testing.T) {
 		},
 	}
 
+	coreops.SetInstance(coreops.New(fakek8sclient.NewSimpleClientset()))
 	k8sClient := testutil.FakeK8sClient(preflightDS)
 
 	err := k8sClient.Create(context.TODO(), preFlightPod1)
@@ -323,6 +331,9 @@ func TestValidateMissingRequiredCheck(t *testing.T) {
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      "px-cluster",
 			Namespace: "kube-test",
+			Annotations: map[string]string{
+				pxutil.AnnotationPreflightCheck: "true",
+			},
 		},
 	}
 
@@ -379,6 +390,7 @@ func TestValidateMissingRequiredCheck(t *testing.T) {
 		},
 	}
 
+	coreops.SetInstance(coreops.New(fakek8sclient.NewSimpleClientset()))
 	k8sClient := testutil.FakeK8sClient(preflightDS)
 
 	err := k8sClient.Create(context.TODO(), preFlightPod1)

drivers/storage/portworx/preflight.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"regexp"
 	"strings"
+	"time"
 
 	"github.com/hashicorp/go-version"
 	"github.com/sirupsen/logrus"
@@ -43,8 +44,9 @@ type PreFlightPortworx interface {
 	// RunPreFlight runs the pre-flight  daemonset
 	RunPreFlight() error
 	// GetPreFlightStatus returns the status of the pre-flight daemonset
-	// returns the no. of completed, in progress and total pods
-	GetPreFlightStatus() (int32, int32, int32, error)
+	// returns the no. of completed, in progress and total pods, plus how
+	// long the daemonset  has been running.
+	GetPreFlightStatus() (int32, int32, int32, time.Duration, error)
 	// GetPreFlightPods returns the pods of the pre-flight daemonset
 	GetPreFlightPods() ([]*v1.Pod, error)
 	// ProcessPreFlightResults process StorageNode status checks
@@ -77,7 +79,7 @@ func NewPreFlighter(
 	}
 }
 
-func getPreFlightPodsFromNamespace(k8sClient client.Client, namespace string) (*appsv1.DaemonSet, []*v1.Pod, error) {
+func GetPreFlightPodsFromNamespace(k8sClient client.Client, namespace string) (*appsv1.DaemonSet, []*v1.Pod, error) {
 	ds := &appsv1.DaemonSet{}
 	err := k8sClient.Get(
 		context.TODO(),
@@ -227,14 +229,14 @@ func (u *preFlightPortworx) CreatePreFlightDaemonsetSpec(ownerRef *metav1.OwnerR
 
 // GetPreFlightPods returns the pods of the pre-flight daemonset
 func (u *preFlightPortworx) GetPreFlightPods() ([]*v1.Pod, error) {
-	_, pods, err := getPreFlightPodsFromNamespace(u.k8sClient, u.cluster.Namespace)
+	_, pods, err := GetPreFlightPodsFromNamespace(u.k8sClient, u.cluster.Namespace)
 	return pods, err
 }
 
-func (u *preFlightPortworx) GetPreFlightStatus() (int32, int32, int32, error) {
-	ds, pods, err := getPreFlightPodsFromNamespace(u.k8sClient, u.cluster.Namespace)
+func (u *preFlightPortworx) GetPreFlightStatus() (int32, int32, int32, time.Duration, error) {
+	ds, pods, err := GetPreFlightPodsFromNamespace(u.k8sClient, u.cluster.Namespace)
 	if err != nil {
-		return -1, -1, -1, err
+		return -1, -1, -1, 0, err
 	}
 	totalPods := ds.Status.DesiredNumberScheduled
 	completedPods := 0
@@ -247,8 +249,11 @@ func (u *preFlightPortworx) GetPreFlightStatus() (int32, int32, int32, error) {
 			}
 		}
 	}
+
+	statusAge := time.Since(ds.CreationTimestamp.Time)
+	logrus.Infof("Pre-flight status running time: %v", statusAge)
 	logrus.Infof("Pre-flight Status: Completed [%v] InProgress [%v] Total Pods [%v]", completedPods, totalPods-int32(completedPods), totalPods)
-	return int32(completedPods), totalPods - int32(completedPods), totalPods, nil
+	return int32(completedPods), totalPods - int32(completedPods), totalPods, statusAge, nil
 }
 
 func (u *preFlightPortworx) RunPreFlight() error {
@@ -288,7 +293,11 @@ func (u *preFlightPortworx) RunPreFlight() error {
 
 	err = u.k8sClient.Create(context.TODO(), preflightDS)
 	if err != nil {
-		logrus.Errorf("runPreFlight: error creating: %v", err)
+		if errors.IsAlreadyExists(err) {
+			logrus.Infof("runPreFlight: daemonset already exists")
+		} else {
+			logrus.Errorf("RunPreFlight: error creating: %v", err)
+		}
 	}
 
 	return err

pkg/controller/storagecluster/controller_test.go

@@ -9114,15 +9114,21 @@ func TestEKSPreflightCheck(t *testing.T) {
 
 	// TestCase: fix the permission then rerun preflight
 	cluster.Annotations[pxutil.AnnotationPreflightCheck] = "true"
+	// Cloud drive permission check will only be checked once when the status condition is not set.
+	cluster.Status.Conditions = []corev1.ClusterCondition{}
 	preflightOps.EXPECT().CheckCloudDrivePermission(cluster).Return(nil)
 	err = controller.runPreflightCheck(cluster)
 	require.NoError(t, err)
 
+	// Pre-flight checks now contain DMthin checks which will not be executed unless cloud permission checks passes.
+	// Since the DMthin checks will not be  executed the condition returned will be 'ClusterConditionStatusInProgress'.
+	// This is good enough to determine if cloud permission is working. If it had failed DMthin would not be run and
+	// the condition would have been ClusterConditionStatusFailed.
 	err = testutil.Get(k8sClient, cluster, cluster.Name, cluster.Namespace)
 	require.NoError(t, err)
 	condition = util.GetStorageClusterCondition(cluster, pxutil.PortworxComponentName, corev1.ClusterConditionTypePreflight)
 	require.NotNil(t, condition)
-	require.Equal(t, corev1.ClusterConditionStatusCompleted, condition.Status)
+	require.Equal(t, corev1.ClusterConditionStatusInProgress, condition.Status)
 }
 
 func TestStorageClusterStateDuringValidation(t *testing.T) {