move dataflow affinity logic to fluidapp controller (#4138)

* use fluidapp to support data flow affinity Signed-off-by: xliuqq <[email protected]> * make fluidapp dataop controller optional Signed-off-by: xliuqq <[email protected]> * fix check Signed-off-by: xliuqq <[email protected]> * add unit test for injectPodNodeLabelsToJob Signed-off-by: xliuqq <[email protected]> * fix: use library.fluid.labels instead of fluid.io/managed-by Signed-off-by: xliuqq <[email protected]> * fix comment error and reslove conflicts Signed-off-by: xliuqq <[email protected]> * fix annotation nil error Signed-off-by: xliuqq <[email protected]> * fix e2e test, serverless pod use serverless.fluid.io/inject and fluid.io/managed-by two labels Signed-off-by: xliuqq <[email protected]> --------- Signed-off-by: xliuqq <[email protected]>
fluid-cloudnative · Jun 24, 2024 · 620babc · 620babc
1 parent d42a2e5
commit 620babc
Show file tree

Hide file tree

Showing 59 changed files with 1,193 additions and 506 deletions.
diff --git a/api/v1alpha1/common.go b/api/v1alpha1/common.go
@@ -246,15 +246,6 @@ type Condition struct {
 	LastTransitionTime metav1.Time `json:"lastTransitionTime,omitempty"`
 }
 
-type OperationType string
-
-const (
-	DataLoadType    OperationType = "DataLoad"
-	DataBackupType  OperationType = "DataBackup"
-	DataMigrateType OperationType = "DataMigrate"
-	DataProcessType OperationType = "DataProcess"
-)
-
 // AffinityPolicy the strategy for the affinity between Data Operation Pods.
 type AffinityPolicy string
 

diff --git a/charts/fluid-dataloader/alluxio/templates/job.yaml b/charts/fluid-dataloader/alluxio/templates/job.yaml
@@ -49,6 +49,7 @@ spec:
         role: dataload-pod
         app: alluxio
         targetDataset: {{ required "targetDataset should be set" .Values.dataloader.targetDataset }}
+        {{- include "library.fluid.labels" . | nindent 8 }}
       {{- if .Values.dataloader.labels }}
       {{- range $key, $val := .Values.dataloader.labels }}
         {{ $key | quote }}: {{ $val | quote }}

diff --git a/charts/fluid-dataloader/goosefs/templates/job.yaml b/charts/fluid-dataloader/goosefs/templates/job.yaml
@@ -41,6 +41,7 @@ spec:
         role: dataload-pod
         app: goosefs
         targetDataset: {{ required "targetDataset should be set" .Values.dataloader.targetDataset }}
+        {{- include "library.fluid.labels" . | nindent 8 }}
     spec:
       {{- if .Values.dataloader.schedulerName }}
       schedulerName: {{ .Values.dataloader.schedulerName }}

diff --git a/charts/fluid-dataloader/jindo/templates/job.yaml b/charts/fluid-dataloader/jindo/templates/job.yaml
@@ -44,6 +44,7 @@ spec:
         role: dataload-pod
         app: jindofs
         targetDataset: {{ required "targetDataset should be set" .Values.dataloader.targetDataset }}
+        {{- include "library.fluid.labels" . | nindent 8 }}
     spec:
       {{- if .Values.dataloader.schedulerName }}
       schedulerName: {{ .Values.dataloader.schedulerName }}

diff --git a/charts/fluid-dataloader/jindocache/templates/job.yaml b/charts/fluid-dataloader/jindocache/templates/job.yaml
@@ -49,6 +49,7 @@ spec:
         role: dataload-pod
         app: jindocache
         targetDataset: {{ required "targetDataset should be set" .Values.dataloader.targetDataset }}
+        {{- include "library.fluid.labels" . | nindent 8 }}
       {{- if .Values.dataloader.labels }}
       {{- range $key, $val := .Values.dataloader.labels }}
         {{ $key | quote }}: {{ $val | quote }}

diff --git a/charts/fluid-dataloader/jindofsx/templates/job.yaml b/charts/fluid-dataloader/jindofsx/templates/job.yaml
@@ -49,6 +49,7 @@ spec:
         role: dataload-pod
         app: jindofsx
         targetDataset: {{ required "targetDataset should be set" .Values.dataloader.targetDataset }}
+        {{- include "library.fluid.labels" . | nindent 8 }}
       {{- if .Values.dataloader.labels }}
       {{- range $key, $val := .Values.dataloader.labels }}
         {{ $key | quote }}: {{ $val | quote }}

diff --git a/charts/fluid-dataloader/juicefs/templates/job.yaml b/charts/fluid-dataloader/juicefs/templates/job.yaml
@@ -49,6 +49,7 @@ spec:
         role: dataload-pod
         app: juicefs
         targetDataset: {{ required "targetDataset should be set" .Values.dataloader.targetDataset }}
+        {{- include "library.fluid.labels" . | nindent 8 }}
       {{- if .Values.dataloader.labels }}
       {{- range $key, $val := .Values.dataloader.labels }}
         {{ $key | quote }}: {{ $val | quote }}

diff --git a/charts/fluid-datamigrate/juicefs/templates/job.yaml b/charts/fluid-datamigrate/juicefs/templates/job.yaml
@@ -9,6 +9,8 @@ metadata:
     app: juicefs
     targetDataset: {{ required "targetDataset should be set" .Values.datamigrate.targetDataset }}
     {{- include "library.fluid.labels" . | nindent 4 }}
+    # indicates the parallel task number
+    parallelism: {{ .Values.datamigrate.parallelism | quote }}
   ownerReferences:
   {{- if .Values.owner.enabled }}
     - apiVersion: {{ .Values.owner.apiVersion }}
@@ -38,6 +40,7 @@ spec:
         app: juicefs
         targetDataset: {{ required "targetDataset should be set" .Values.datamigrate.targetDataset }}
         fluid.io/operation: migrate-{{ .Release.Namespace }}-{{ .Release.Name }}
+        {{- include "library.fluid.labels" . | nindent 8 }}
       {{- if .Values.datamigrate.labels }}
       {{- range $key, $val := .Values.datamigrate.labels }}
         {{ $key | quote }}: {{ $val | quote }}

diff --git a/charts/fluid-dataprocess/common/templates/job.yaml b/charts/fluid-dataprocess/common/templates/job.yaml
@@ -34,6 +34,7 @@ spec:
         role: dataprocess-pod
         app: fluid-dataprocess
         targetDataset: {{ required "targetDataset should be set" .Values.dataProcess.targetDataset }}
+        {{- include "library.fluid.labels" . | nindent 8 }}
         {{- if .Values.dataProcess.labels }}
         {{ toYaml .Values.dataProcess.labels | nindent 8 }}
         {{- end }}

diff --git a/charts/fluid/fluid/templates/_helpers.tpl b/charts/fluid/fluid/templates/_helpers.tpl
@@ -99,7 +99,7 @@ Create the name of the service account to use
 Check if feature gate DataflowAffinity is enabled in the featureGates.
 */}}
 {{- define "fluid.dataflowAffinity.enabled" -}}
-{{- $featureGates := splitList "," .Values.dataset.featureGates }}
+{{- $featureGates := splitList "," .Values.fluidapp.featureGates }}
 {{- $found := false -}}
 {{- range $idx, $featureGate := $featureGates }}
     {{- $featureGateKV := splitList "=" $featureGate }}

diff --git a/charts/fluid/fluid/templates/controller/dataset_controller.yaml b/charts/fluid/fluid/templates/controller/dataset_controller.yaml
@@ -50,7 +50,6 @@ spec:
           - --pprof-addr=:6060
           - --enable-leader-election
           - --leader-election-namespace={{ include "fluid.namespace" . }}
-          - --feature-gates={{ .Values.dataset.featureGates }}
         env:
           {{- if .Values.workdir }}
           - name: FLUID_WORKDIR

diff --git a/charts/fluid/fluid/templates/controller/fluidapp_controller.yaml b/charts/fluid/fluid/templates/controller/fluidapp_controller.yaml
@@ -48,6 +48,7 @@ spec:
           - --pprof-addr=:6060
           - --enable-leader-election
           - --leader-election-namespace={{ include "fluid.namespace" . }}
+          - --feature-gates={{ .Values.fluidapp.featureGates }}
         env:
         {{- if .Values.workdir }}
           - name: FLUID_WORKDIR

diff --git a/charts/fluid/fluid/templates/role/dataset/rbac.yaml b/charts/fluid/fluid/templates/role/dataset/rbac.yaml
@@ -13,16 +13,6 @@ rules:
     - get
     - list
     - watch
-  {{- if eq (include "fluid.dataflowAffinity.enabled" . ) "true" }}
-  - apiGroups:
-      - ""
-    resources:
-      - nodes
-    verbs:
-      - get
-      - list
-      - watch
-  {{- end }}
   - apiGroups:
     - ""
     resources:

diff --git a/charts/fluid/fluid/templates/role/fluidapp/rbac.yaml b/charts/fluid/fluid/templates/role/fluidapp/rbac.yaml
@@ -12,6 +12,25 @@ rules:
       - get
       - watch
       - create
+  {{- if eq (include "fluid.dataflowAffinity.enabled" . ) "true" }}
+  - apiGroups:
+      - ""
+    resources:
+      - nodes
+    verbs:
+      - get
+      - list
+      - watch
+  - apiGroups:
+      - batch
+    resources:
+      - jobs
+    verbs:
+      - get
+      - list
+      - watch
+      - update
+  {{- end }}
   - apiGroups:
       - ""
     resources:

diff --git a/charts/fluid/fluid/values.yaml b/charts/fluid/fluid/values.yaml
@@ -26,7 +26,6 @@ dataset:
     - operator: Exists
   controller:
     image: fluidcloudnative/dataset-controller:v1.0.1-b33c500
-  featureGates: "DataflowAffinity=false"
 
 csi:
   tolerations:
@@ -204,10 +203,12 @@ webhook:
           required:
             - fluid.io/node
 
+
 fluidapp:
   enabled: true
   replicas: 1
   tolerations:
     - operator: Exists
   controller:
     image: fluidcloudnative/application-controller:v1.0.1-b33c500
+  featureGates: "DataflowAffinity=false"
diff --git a/cmd/dataset/app/dataset.go b/cmd/dataset/app/dataset.go
@@ -17,7 +17,6 @@ limitations under the License.
 package app
 
 import (
-	utilfeature "github.com/fluid-cloudnative/fluid/pkg/utils/feature"
 	"os"
 	"time"
 
@@ -101,8 +100,6 @@ func init() {
 	datasetCmd.Flags().StringVar(&controllerWorkqueueMaxSyncBackoffStr, "workqueue-max-sync-backoff", "1000s", "max backoff period for failed reconciliation in controller's workqueue")
 	datasetCmd.Flags().IntVar(&controllerWorkqueueQPS, "workqueue-qps", 10, "qps limit value for controller's workqueue")
 	datasetCmd.Flags().IntVar(&controllerWorkqueueBurst, "workqueue-burst", 100, "burst limit value for controller's workqueue")
-
-	utilfeature.DefaultMutableFeatureGate.AddFlag(datasetCmd.Flags())
 }
 
 func handle() {

diff --git a/cmd/fluidapp/app/fluidapp.go b/cmd/fluidapp/app/fluidapp.go
@@ -17,7 +17,14 @@
 package app
 
 import (
+	"github.com/fluid-cloudnative/fluid/pkg/common"
+	"github.com/fluid-cloudnative/fluid/pkg/controllers/v1alpha1/fluidapp/dataflowaffinity"
+	"github.com/fluid-cloudnative/fluid/pkg/dataflow"
+	utilfeature "github.com/fluid-cloudnative/fluid/pkg/utils/feature"
+	batchv1 "k8s.io/api/batch/v1"
+	"k8s.io/apimachinery/pkg/labels"
 	"os"
+	"sigs.k8s.io/controller-runtime/pkg/cache"
 
 	"github.com/fluid-cloudnative/fluid"
 	"github.com/fluid-cloudnative/fluid/pkg/controllers/v1alpha1/fluidapp"
@@ -62,6 +69,8 @@ func init() {
 	fluidAppCmd.Flags().BoolVarP(&development, "development", "", true, "Enable development mode for fluid controller.")
 	fluidAppCmd.Flags().StringVarP(&pprofAddr, "pprof-addr", "", "", "The address for pprof to use while exporting profiling results")
 	fluidAppCmd.Flags().IntVar(&maxConcurrentReconciles, "runtime-workers", 3, "Set max concurrent workers for Fluid App controller")
+
+	utilfeature.DefaultMutableFeatureGate.AddFlag(fluidAppCmd.Flags())
 }
 
 func handle() {
@@ -89,7 +98,7 @@ func handle() {
 		LeaderElectionNamespace: leaderElectionNamespace,
 		LeaderElectionID:        "fluidapp.data.fluid.io",
 		Port:                    9443,
-		NewCache:                fluidapp.NewCache(scheme),
+		NewCache:                NewCache(scheme),
 	})
 	if err != nil {
 		setupLog.Error(err, "unable to start fluid app manager")
@@ -98,7 +107,6 @@ func handle() {
 
 	controllerOptions := controller.Options{
 		MaxConcurrentReconciles: maxConcurrentReconciles,
-		// Log:                     ctrl.Log.WithName("appctrl"),
 	}
 	if err = (fluidapp.NewFluidAppReconciler(
 		mgr.GetClient(),
@@ -109,9 +117,44 @@ func handle() {
 		os.Exit(1)
 	}
 
+	if dataflow.Enabled(dataflow.DataflowAffinity) {
+		if err = (dataflowaffinity.NewDataOpJobReconciler(
+			mgr.GetClient(),
+			ctrl.Log.WithName("dataopctrl"),
+			mgr.GetEventRecorderFor("DataOpJob"),
+		)).SetupWithManager(mgr, controllerOptions); err != nil {
+			setupLog.Error(err, "unable to create controller", "controller", "DataOpJob")
+			os.Exit(1)
+		}
+	}
+
 	setupLog.Info("starting fluidapp-controller")
 	if err := mgr.Start(ctrl.SetupSignalHandler()); err != nil {
 		setupLog.Error(err, "problem running fluidapp-controller")
 		os.Exit(1)
 	}
 }
+
+func NewCache(scheme *runtime.Scheme) cache.NewCacheFunc {
+	options := cache.Options{
+		Scheme: scheme,
+		SelectorsByObject: cache.SelectorsByObject{
+			&corev1.Pod{}: {
+				Label: labels.SelectorFromSet(labels.Set{
+					// watch pods managed by fluid, like data operation pods, serverless app pods.
+					common.LabelAnnotationManagedBy: common.Fluid,
+				}),
+			},
+		},
+	}
+	if dataflow.Enabled(dataflow.DataflowAffinity) {
+		options.SelectorsByObject[&batchv1.Job{}] = cache.ObjectSelector{
+			// watch data operation job
+			Label: labels.SelectorFromSet(labels.Set{
+				// only data operations create job resource and the jobs created by cronjob do not have this label.
+				common.LabelAnnotationManagedBy: common.Fluid,
+			}),
+		}
+	}
+	return cache.BuilderWithOptions(options)
+}
diff --git a/docs/en/samples/application_controller.md b/docs/en/samples/application_controller.md
@@ -45,7 +45,7 @@ jfsdemo   [Calculating]    N/A                       N/A                 Bound
 
 **Create Job**
 
-To use Fluid in a serverless scenario, you need to add the `serverless.fluid.io/inject: "true"` label to the application pod. as follows:
+To use Fluid in a serverless scenario, you need to add the `serverless.fluid.io/inject: "true"` and `fluid.io/managed-by: fluid` label to the application pod. as follows:
 
 ```yaml
 $ cat<<EOF >sample.yaml
@@ -58,6 +58,7 @@ spec:
     metadata:
       labels:
         serverless.fluid.io/inject: "true"
+        fluid.io/managed-by: fluid
     spec:
       containers:
         - name: demo

diff --git a/docs/zh/samples/application_controller.md b/docs/zh/samples/application_controller.md
@@ -39,7 +39,7 @@ jfsdemo   [Calculating]    N/A                       N/A                 Bound
 
 **创建 Job 资源对象**
 
-在 Serverless 场景使用 Fluid，需要在应用 Pod 中添加 `serverless.fluid.io/inject: "true"` label。如下：
+在 Serverless 场景使用 Fluid，需要在应用 Pod 中添加 `serverless.fluid.io/inject: "true"`和`fluid.io/managed-by: fluid` label。如下：
 
 ```yaml
 $ cat<<EOF >sample.yaml
@@ -52,6 +52,7 @@ spec:
     metadata:
       labels:
         serverless.fluid.io/inject: "true"
+        fluid.io/managed-by: fluid
     spec:
       containers:
         - name: demo

diff --git a/pkg/common/label.go b/pkg/common/label.go
@@ -34,7 +34,7 @@ const (
 	// LabelAnnotationManagedByDeprecated is a deprecated label key for LabelAnnotationManagedBy
 	LabelAnnotationManagedByDeprecated = LabelAnnotationPrefix + "wrapped-by"
 
-	// LabelAnnotationManagedBy indicates a pvc that is managed by Fluid
+	// LabelAnnotationManagedBy indicates a resource(like pvc) that is managed by Fluid
 	LabelAnnotationManagedBy = LabelAnnotationPrefix + "managed-by"
 
 	// fluid adminssion webhook inject flag
@@ -48,6 +48,11 @@ const (
 
 	// LabelNodePublishMothod is a pv label that indicates the method nodePuhlishVolume use
 	LabelNodePublishMothod = LabelAnnotationPrefix + "node-publish-method"
+
+	// AnnotationDataFlowAffinityInject is an annotation representing enabled the dataflow affinity injection
+	AnnotationDataFlowAffinityInject = LabelAnnotationPrefix + "dataflow-affinity.inject"
+	// LabelDataFlowAffinityPrefix is a prefix for customized dataflow affinity label name.
+	LabelDataFlowAffinityPrefix = "fluid.io."
 )
 
 var (

diff --git a/pkg/controllers/v1alpha1/databackup/implement.go b/pkg/controllers/v1alpha1/databackup/implement.go
@@ -108,8 +108,8 @@ func (r *dataBackupOperation) SetTargetDatasetStatusInProgress(dataset *datav1al
 func (r *dataBackupOperation) RemoveTargetDatasetStatusInProgress(dataset *datav1alpha1.Dataset) {
 }
 
-func (r *dataBackupOperation) GetOperationType() datav1alpha1.OperationType {
-	return datav1alpha1.DataBackupType
+func (r *dataBackupOperation) GetOperationType() dataoperation.OperationType {
+	return dataoperation.DataBackupType
 }
 
 func (r *dataBackupOperation) GetTargetDataset() (*datav1alpha1.Dataset, error) {

diff --git a/pkg/controllers/v1alpha1/databackup/status_handler.go b/pkg/controllers/v1alpha1/databackup/status_handler.go
@@ -17,8 +17,6 @@
 package databackup
 
 import (
-	"fmt"
-	"github.com/fluid-cloudnative/fluid/pkg/dataflow"
 	"time"
 
 	"github.com/fluid-cloudnative/fluid/api/v1alpha1"
@@ -54,14 +52,7 @@ func (o *OnceHandler) GetOperationStatus(ctx runtime.ReconcileRequestContext, op
 		return
 	}
 
-	// set the node labels in status when job finished
-	if dataflow.Enabled(dataflow.DataflowAffinity) && result.NodeAffinity == nil {
-		// generate the node labels
-		result.NodeAffinity, err = dataflow.GenerateNodeAffinity(ctx.Client, backupPod)
-		if err != nil {
-			return nil, fmt.Errorf("error to generate the node labels: %v", err)
-		}
-	}
+	// TODO: inject nodeaffinity like other data operations when using job instead of pod
 
 	var finishTime time.Time
 	if len(backupPod.Status.Conditions) != 0 {

diff --git a/pkg/controllers/v1alpha1/dataload/implement.go b/pkg/controllers/v1alpha1/dataload/implement.go
@@ -71,8 +71,8 @@ func (r *dataLoadOperation) GetChartsDirectory() string {
 	return utils.GetChartsDirectory() + "/" + cdataload.DataloadChart
 }
 
-func (r *dataLoadOperation) GetOperationType() datav1alpha1.OperationType {
-	return datav1alpha1.DataLoadType
+func (r *dataLoadOperation) GetOperationType() dataoperation.OperationType {
+	return dataoperation.DataLoadType
 }
 
 func (r *dataLoadOperation) UpdateOperationApiStatus(opStatus *datav1alpha1.OperationStatus) error {