[#716] improvement(operator): support specifying imagePullSecrets (#765)

### What changes were proposed in this pull request?
Support specifying imagePullSecrets

### Why are the changes needed?
If the images are stored in a private registry, Kubernetes needs to be provided with the necessary credentials to authenticate with the registry.
Fix: #716

### Does this PR introduce any user-facing change?
Example:

apiVersion: uniffle.apache.org/v1alpha1
kind: RemoteShuffleService
metadata:
  name: rss-demo
  namespace: rss
spec:
  # ConfigMapName indicates configMap name stores configurations of coordinators and shuffle servers.
  configMapName: rss-configuration
  imagePullSecrets:
    - name: "default-secret"
### How was this patch tested?
UT

deploy/kubernetes/operator/api/uniffle/v1alpha1/remoteshuffleservice_types.go

@@ -59,6 +59,9 @@ type RemoteShuffleServiceSpec struct {
 
 	// ConfigMapName indicates configMap name stores configurations of coordinators and shuffle servers.
 	ConfigMapName string `json:"configMapName"`
+
+	// +optional
+	ImagePullSecrets []corev1.LocalObjectReference `json:"imagePullSecrets,omitempty"`
 }
 
 // CoordinatorConfig records configuration used to generate workload of coordination.

deploy/kubernetes/operator/config/crd/bases/uniffle.apache.org_remoteshuffleservices.yaml

@@ -4189,6 +4189,17 @@ spec:
                 - image
                 - xmxSize
                 type: object
+              imagePullSecrets:
+                items:
+                  description: LocalObjectReference contains enough information to
+                    let you locate the referenced object inside the same namespace.
+                  properties:
+                    name:
+                      description: 'Name of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#names
+                        TODO: Add other useful fields. apiVersion, kind, uid?'
+                      type: string
+                  type: object
+                type: array
               shuffleServer:
                 description: ShuffleServer contains configuration of the shuffle servers.
                 properties:

deploy/kubernetes/operator/pkg/controller/sync/coordinator/coordinator.go

@@ -182,6 +182,7 @@ func GenerateDeploy(rss *unifflev1alpha1.RemoteShuffleService, index int) *appsv
 		Volumes:            rss.Spec.Coordinator.Volumes,
 		NodeSelector:       rss.Spec.Coordinator.NodeSelector,
 		Affinity:           rss.Spec.Coordinator.Affinity,
+		ImagePullSecrets:   rss.Spec.ImagePullSecrets,
 	}
 	configurationVolume := corev1.Volume{
 		Name: controllerconstants.ConfigurationVolumeName,

deploy/kubernetes/operator/pkg/controller/sync/coordinator/coordinator_test.go

@@ -128,6 +128,11 @@ var (
 			},
 		},
 	}
+	testImagePullSecrets = []corev1.LocalObjectReference{
+		{
+			Name: "default-secret",
+		},
+	}
 )
 
 func buildRssWithLabels() *uniffleapi.RemoteShuffleService {
@@ -166,6 +171,12 @@ func withCustomAffinity(affinity *corev1.Affinity) *uniffleapi.RemoteShuffleServ
 	return rss
 }
 
+func withCustomImagePullSecrets(imagePullSecrets []corev1.LocalObjectReference) *uniffleapi.RemoteShuffleService {
+	rss := utils.BuildRSSWithDefaultValue()
+	rss.Spec.ImagePullSecrets = imagePullSecrets
+	return rss
+}
+
 func buildRssWithCustomRPCPort() *uniffleapi.RemoteShuffleService {
 	rss := utils.BuildRSSWithDefaultValue()
 	rss.Spec.Coordinator.RPCPort = pointer.Int32(testRPCPort)
@@ -436,6 +447,20 @@ func TestGenerateDeploy(t *testing.T) {
 				return false, fmt.Errorf("generated deploy should include affinity: %v", testAffinity)
 			},
 		},
+		{
+			name: "set custom imagePullSecrets",
+			rss:  withCustomImagePullSecrets(testImagePullSecrets),
+			IsValidDeploy: func(deploy *appsv1.Deployment, rss *uniffleapi.RemoteShuffleService) (bool, error) {
+				if deploy.Spec.Template.Spec.ImagePullSecrets != nil {
+					deploy.Spec.Template.Spec.ImagePullSecrets = rss.Spec.ImagePullSecrets
+					equal := reflect.DeepEqual(deploy.Spec.Template.Spec.ImagePullSecrets, testImagePullSecrets)
+					if equal {
+						return true, nil
+					}
+				}
+				return false, fmt.Errorf("generated deploy should include imagePullSecrets: %v", testImagePullSecrets)
+			},
+		},
 	} {
 		t.Run(tt.name, func(tc *testing.T) {
 			deploy := GenerateDeploy(tt.rss, 0)

deploy/kubernetes/operator/pkg/controller/sync/shuffleserver/shuffleserver.go

@@ -89,6 +89,7 @@ func GenerateSts(rss *unifflev1alpha1.RemoteShuffleService) *appsv1.StatefulSet
 		Volumes:            rss.Spec.ShuffleServer.Volumes,
 		NodeSelector:       rss.Spec.ShuffleServer.NodeSelector,
 		Affinity:           rss.Spec.ShuffleServer.Affinity,
+		ImagePullSecrets:   rss.Spec.ImagePullSecrets,
 	}
 
 	configurationVolume := corev1.Volume{

deploy/kubernetes/operator/pkg/controller/sync/shuffleserver/shuffleserver_test.go

@@ -128,6 +128,11 @@ var (
 			},
 		},
 	}
+	testImagePullSecrets = []corev1.LocalObjectReference{
+		{
+			Name: "default-secret",
+		},
+	}
 )
 
 func buildRssWithLabels() *uniffleapi.RemoteShuffleService {
@@ -183,6 +188,12 @@ func withCustomAffinity(affinity *corev1.Affinity) *uniffleapi.RemoteShuffleServ
 	return rss
 }
 
+func withCustomImagePullSecrets(imagePullSecrets []corev1.LocalObjectReference) *uniffleapi.RemoteShuffleService {
+	rss := utils.BuildRSSWithDefaultValue()
+	rss.Spec.ImagePullSecrets = imagePullSecrets
+	return rss
+}
+
 func buildCommonExpectedENVs(rss *uniffleapi.RemoteShuffleService) []corev1.EnvVar {
 	return []corev1.EnvVar{
 		{
@@ -443,6 +454,20 @@ func TestGenerateSts(t *testing.T) {
 				return false, fmt.Errorf("generated sts should include affinity: %v", testAffinity)
 			},
 		},
+		{
+			name: "set custom imagePullSecrets",
+			rss:  withCustomImagePullSecrets(testImagePullSecrets),
+			IsValidSts: func(deploy *appsv1.StatefulSet, rss *uniffleapi.RemoteShuffleService) (bool, error) {
+				if deploy.Spec.Template.Spec.ImagePullSecrets != nil {
+					deploy.Spec.Template.Spec.ImagePullSecrets = rss.Spec.ImagePullSecrets
+					equal := reflect.DeepEqual(deploy.Spec.Template.Spec.ImagePullSecrets, testImagePullSecrets)
+					if equal {
+						return true, nil
+					}
+				}
+				return false, fmt.Errorf("generated sts should include imagePullSecrets: %v", testImagePullSecrets)
+			},
+		},
 	} {
 		t.Run(tt.name, func(tc *testing.T) {
 			sts := GenerateSts(tt.rss)