diff --git a/pkg/driver/gcp-pd/gcp_pd.go b/pkg/driver/gcp-pd/gcp_pd.go index 7cb686e18..32000730b 100644 --- a/pkg/driver/gcp-pd/gcp_pd.go +++ b/pkg/driver/gcp-pd/gcp_pd.go @@ -67,6 +67,14 @@ func GetGCPPDOperatorConfig() *config.OperatorConfig { AssetDir: generatedAssetBase, OperatorControllerConfigBuilder: GetGCPPDOperatorControllerConfig, Removable: false, + PrerequisiteAssets: []string{ + "controller_sa.yaml", + "node_sa.yaml", + "hostnetwork_role.yaml", + "controller_hostnetwork_binding.yaml", + "privileged_role.yaml", + "node_privileged_binding.yaml", + }, } } diff --git a/pkg/operator/config/config.go b/pkg/operator/config/config.go index 8004f7b8d..398ab0160 100644 --- a/pkg/operator/config/config.go +++ b/pkg/operator/config/config.go @@ -34,6 +34,8 @@ type OperatorConfig struct { CloudConfigNamespace string // Removable should be true if the operator and its operand can be removed Removable bool + // Prerequisite (static) assets + PrerequisiteAssets []string } // OperatorControllerConfig is configuration of controllers that are used to deploy CSI drivers. diff --git a/pkg/operator/prerequisites.go b/pkg/operator/prerequisites.go new file mode 100644 index 000000000..5dbbd5e2f --- /dev/null +++ b/pkg/operator/prerequisites.go @@ -0,0 +1,71 @@ +package operator + +import ( + "context" + "fmt" + "path/filepath" + "time" + + "github.com/openshift/csi-operator/assets" + "github.com/openshift/library-go/pkg/operator/events" + "github.com/openshift/library-go/pkg/operator/resource/resourceapply" + kubeclient "k8s.io/client-go/kubernetes" + "k8s.io/klog/v2" +) + +var ( + numIterations = 10 + delayIteration = 5 * time.Second +) + +func applyPrerequisites(ctx context.Context, kubeClient kubeclient.Interface, recorder events.Recorder, assetDir string, assetNames []string) error { + files := make([]string, len(assetNames)) + for i, name := range assetNames { + files[i] = filepath.Join(assetDir, name) + + } + + var errs []error + for range numIterations { + if len(files) == 0 { + klog.Infof("All prerequisite assets are applied") + return nil + } + results := resourceapply.ApplyDirectly( + ctx, + resourceapply.NewKubeClientHolder(kubeClient), + recorder, + resourceapply.NewResourceCache(), + assets.ReadFile, + files..., + ) + + errs = errs[:0] + for _, result := range results { + if result.Error != nil { + errs = append(errs, fmt.Errorf("%s: %w", result.File, result.Error)) + continue + } + klog.V(2).Infof("Applied prerequisite asset %s (changed=%v)", result.File, result.Changed) + // Remove successfully applied assets from the list + for i, file := range files { + if file == result.File { + files = append(files[:i], files[i+1:]...) + break + } + } + } + if len(errs) != 0 { + klog.Warningf("Failed to apply some prerequisites: %v", errs) + } + if len(files) != 0 { + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(delayIteration): + } + } + } + return fmt.Errorf("failed to apply some prerequisites: %v", errs) + +} diff --git a/pkg/operator/prerequisites_test.go b/pkg/operator/prerequisites_test.go new file mode 100644 index 000000000..80b62d563 --- /dev/null +++ b/pkg/operator/prerequisites_test.go @@ -0,0 +1,150 @@ +package operator + +import ( + "context" + "fmt" + "strings" + "testing" + + "github.com/openshift/library-go/pkg/operator/events" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + fakecore "k8s.io/client-go/kubernetes/fake" + k8stesting "k8s.io/client-go/testing" + "k8s.io/utils/clock" +) + +func TestApplyPrerequisites(t *testing.T) { + assetDir := "overlays/gcp-pd/generated/standalone" + + cases := []struct { + name string + assetDir string + assetNames []string + setupReactor func(client *fakecore.Clientset) + expectErr bool + errContains string + }{ + { + name: "should apply all prerequisite assets successfully", + assetDir: assetDir, + assetNames: []string{"controller_sa.yaml", "node_sa.yaml"}, + expectErr: false, + }, + { + name: "should succeed with empty asset list", + assetDir: assetDir, + assetNames: []string{}, + expectErr: false, + }, + { + name: "should apply all GCP PD prerequisite assets", + assetDir: assetDir, + assetNames: []string{ + "controller_sa.yaml", + "node_sa.yaml", + "hostnetwork_role.yaml", + "controller_hostnetwork_binding.yaml", + "privileged_role.yaml", + "node_privileged_binding.yaml", + }, + expectErr: false, + }, + { + name: "should return error for non-existent asset file", + assetDir: assetDir, + assetNames: []string{"nonexistent.yaml"}, + expectErr: true, + errContains: "nonexistent.yaml", + }, + { + name: "should return error when some assets do not exist", + assetDir: assetDir, + assetNames: []string{"controller_sa.yaml", "nonexistent.yaml"}, + expectErr: true, + errContains: "nonexistent.yaml", + }, + { + name: "should retry failed assets and eventually succeed", + assetDir: assetDir, + assetNames: []string{ + "controller_sa.yaml", + "node_sa.yaml", + }, + setupReactor: func(client *fakecore.Clientset) { + callCount := 0 + client.PrependReactor("create", "serviceaccounts", func(action k8stesting.Action) (bool, runtime.Object, error) { + createAction := action.(k8stesting.CreateAction) + sa := createAction.GetObject().(*corev1.ServiceAccount) + if sa.Name == "gcp-pd-csi-driver-node-sa" { + callCount++ + if callCount <= 1 { + return true, nil, fmt.Errorf("transient error") + } + } + return false, nil, nil + }) + }, + expectErr: false, + }, + { + name: "should return error for non-existent asset directory", + assetDir: "no/such/dir", + assetNames: []string{"controller_sa.yaml"}, + expectErr: true, + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + origDelay := delayIteration + origIterations := numIterations + delayIteration = 0 + numIterations = 3 + defer func() { + delayIteration = origDelay + numIterations = origIterations + }() + + kubeClient := fakecore.NewClientset() + if tc.setupReactor != nil { + tc.setupReactor(kubeClient) + } + recorder := events.NewInMemoryRecorder("test", &clock.RealClock{}) + ctx := context.Background() + + err := applyPrerequisites(ctx, kubeClient, recorder, tc.assetDir, tc.assetNames) + if tc.expectErr && err == nil { + t.Fatalf("expected error but got nil") + } + if !tc.expectErr && err != nil { + t.Fatalf("unexpected error: %v", err) + } + if tc.errContains != "" && err != nil && !strings.Contains(err.Error(), tc.errContains) { + t.Fatalf("expected error to contain %q, got %q", tc.errContains, err.Error()) + } + + if !tc.expectErr { + verifyAppliedAssets(t, ctx, kubeClient, tc.assetDir, tc.assetNames) + } + }) + } +} + +func verifyAppliedAssets(t *testing.T, ctx context.Context, kubeClient *fakecore.Clientset, assetDir string, assetNames []string) { + t.Helper() + for _, name := range assetNames { + switch { + case strings.Contains(name, "_sa.yaml"): + namespace := "openshift-cluster-csi-drivers" + saList, err := kubeClient.CoreV1().ServiceAccounts(namespace).List(ctx, metav1.ListOptions{}) + if err != nil { + t.Fatalf("failed to list service accounts: %v", err) + } + if len(saList.Items) == 0 { + t.Fatalf("expected service accounts to be created, but found none") + } + } + } +} diff --git a/pkg/operator/starter.go b/pkg/operator/starter.go index 5094526d3..975300444 100644 --- a/pkg/operator/starter.go +++ b/pkg/operator/starter.go @@ -254,6 +254,11 @@ func RunOperator(ctx context.Context, controllerConfig *controllercmd.Controller c.WaitForCacheSync(ctx) klog.V(2).Infof("Informers synced") + // Apply some assets earlier. Some controllers won't be happy until they are applied. See, for example, OCPBUGS-99490. + if err = applyPrerequisites(ctx, c.ControlPlaneKubeClient, controllerConfig.EventRecorder, assetDir, opConfig.PrerequisiteAssets); err != nil { + return err + } + if csiOperatorControllerConfig.Precondition != nil { starterController := StarterController( c.OperatorClient,