feature: support dataflow (#3384)

* Move OperationType to api package Signed-off-by: trafalgarzzz <[email protected]> * Init data operation API Signed-off-by: trafalgarzzz <[email protected]> * Implement framework for DataFlow reconciler Signed-off-by: trafalgarzzz <[email protected]> * Implement dataflow (DataLoad kind) Signed-off-by: trafalgarzzz <[email protected]> * Implement dataflow Signed-off-by: trafalgarzzz <[email protected]> * Generate openapi Signed-off-by: trafalgarzzz <[email protected]> --------- Signed-off-by: trafalgarzzz <[email protected]>
fluid-cloudnative · Aug 10, 2023 · 2549d23 · 2549d23
1 parent 94a8588
commit 2549d23
Show file tree

Hide file tree

Showing 37 changed files with 920 additions and 51 deletions.
diff --git a/api/v1alpha1/common.go b/api/v1alpha1/common.go
@@ -233,3 +233,32 @@ type Condition struct {
 	// LastTransitionTime describes last time the condition transitioned from one status to another.
 	LastTransitionTime metav1.Time `json:"lastTransitionTime,omitempty"`
 }
+
+type OperationType string
+
+const (
+	DataLoadType    OperationType = "DataLoad"
+	DataBackupType  OperationType = "DataBackup"
+	DataMigrateType OperationType = "DataMigrate"
+	DataProcessType OperationType = "DataProcess"
+)
+
+type OperationRef struct {
+	// OperationKind specifies the type of the data operation
+	// +required
+	// +kubebuilder:validation:Enum=DataLoad;DataBackup;DataMigrate;DataProcess
+	OperationKind OperationType `json:"operationKind"`
+
+	// Name specifies the name of the referred data operation
+	// +required
+	Name string `json:"name"`
+
+	// Namespace specifies the namespace of the referred data operation
+	// +required
+	Namespace string `json:"namespace"`
+}
+
+type WaitingStatus struct {
+	// OperationComplete indicates if the preceding operation is complete
+	OperationComplete *bool `json:"operationComplete,omitempty"`
+}
diff --git a/api/v1alpha1/databackup_types.go b/api/v1alpha1/databackup_types.go
@@ -30,6 +30,9 @@ type DataBackupSpec struct {
 	BackupPath string `json:"backupPath,omitempty"`
 	// Manage the user to run Alluxio DataBackup
 	RunAs *User `json:"runAs,omitempty"`
+	// Specifies that the preceding operation in a workflow
+	// +optional
+	RunAfter *OperationRef `json:"runAfter,omitempty"`
 }
 
 // +kubebuilder:printcolumn:name="Dataset",type="string",JSONPath=`.spec.dataset`

diff --git a/api/v1alpha1/dataload_types.go b/api/v1alpha1/dataload_types.go
@@ -85,6 +85,10 @@ type DataLoadSpec struct {
 	// The schedule in Cron format, only set when policy is cron, see https://en.wikipedia.org/wiki/Cron.
 	// +optional
 	Schedule string `json:"schedule,omitempty"`
+
+	// Specifies that the preceding operation in a workflow
+	// +optional
+	RunAfter *OperationRef `json:"runAfter,omitempty"`
 }
 
 // +kubebuilder:printcolumn:name="Dataset",type="string",JSONPath=`.spec.dataset.name`

diff --git a/api/v1alpha1/datamigrate_types.go b/api/v1alpha1/datamigrate_types.go
@@ -76,6 +76,10 @@ type DataMigrateSpec struct {
 	// +optional
 	// SchedulerName sets the scheduler to be used for DataLoad pod
 	SchedulerName string `json:"schedulerName,omitempty"`
+
+	// Specifies that the preceding operation in a workflow
+	// +optional
+	RunAfter *OperationRef `json:"runAfter,omitempty"`
 }
 
 type DataToMigrate struct {

diff --git a/api/v1alpha1/dataprocess_types.go b/api/v1alpha1/dataprocess_types.go
@@ -94,6 +94,10 @@ type DataProcessSpec struct {
 	// Processor specify how to process data.
 	// +required
 	Processor Processor `json:"processor"`
+
+	// Specifies that the preceding operation in a workflow
+	// +optional
+	RunAfter *OperationRef `json:"runAfter,omitempty"`
 }
 
 // +kubebuilder:printcolumn:name="Dataset",type="string",JSONPath=`.spec.dataset.name`

diff --git a/api/v1alpha1/openapi_generated.go b/api/v1alpha1/openapi_generated.go
diff --git a/api/v1alpha1/status.go b/api/v1alpha1/status.go
@@ -146,6 +146,8 @@ type OperationStatus struct {
 	LastScheduleTime *metav1.Time `json:"lastScheduleTime,omitempty"`
 	// LastSuccessfulTime is the last time the cron operation successfully completed
 	LastSuccessfulTime *metav1.Time `json:"lastSuccessfulTime,omitempty"`
+	// WaitingStatus stores information about waiting operation.
+	WaitingFor WaitingStatus `json:"waitingFor,omitempty"`
 }
 
 type RuntimePhase string

diff --git a/api/v1alpha1/zz_generated.deepcopy.go b/api/v1alpha1/zz_generated.deepcopy.go