todo

Signed-off-by: Raghavendra Talur <[email protected]>

api/v1alpha1/volumereplicationgroup_types.go

@@ -319,6 +319,8 @@ type KubeObjectsCaptureIdentifier struct {
 type KubeObjectProtectionStatus struct {
 	//+optional
 	CaptureToRecoverFrom *KubeObjectsCaptureIdentifier `json:"captureToRecoverFrom,omitempty"`
+	//+optional
+	CurrentCaptureStartTime *metav1.Time `json:"currentCaptureStartTime,omitempty"`
 }
 
 // VolumeReplicationGroupStatus defines the observed state of VolumeReplicationGroup

internal/controller/vrg_kubeobjects.go

@@ -229,6 +229,24 @@ const (
 	vrgGenerationNumberBitCount = 64
 )
 
+func getFirstGroup(groups []kubeobjects.CaptureSpec) string {
+	firstGroup := ""
+
+	for _, group := range groups {
+		cg := group
+
+		if cg.IsHook {
+			continue
+		}
+
+		firstGroup = cg.Name
+
+		break
+	}
+
+	return firstGroup
+}
+
 func (v *VRGInstance) kubeObjectsCaptureStartOrResume(
 	result *ctrl.Result,
 	captureStartConditionally captureStartConditionally,
@@ -270,7 +288,9 @@ func (v *VRGInstance) kubeObjectsCaptureStartOrResume(
 		}
 	}
 
-	request0, ok := requests[kubeObjectsCaptureName(namePrefix, groups[0].Name, v.s3StoreAccessors[0].S3ProfileName)]
+	firstGroup := getFirstGroup(groups)
+
+	request0, ok := requests[kubeObjectsCaptureName(namePrefix, firstGroup, v.s3StoreAccessors[0].S3ProfileName)]
 
 	if ok {
 		v.kubeObjectsCaptureComplete(
@@ -280,7 +300,7 @@ func (v *VRGInstance) kubeObjectsCaptureStartOrResume(
 			veleroNamespaceName,
 			interval,
 			labels,
-			request0.StartTime(),
+			*v.instance.Status.KubeObjectProtection.CurrentCaptureStartTime,
 			request0.Object().GetAnnotations(),
 		)
 	}
@@ -631,6 +651,16 @@ func (v *VRGInstance) getRecoverRequests() (map[string]kubeobjects.Request, erro
 	return kubeobjects.RequestsMapKeyedByName(recoverRequestsStruct), nil
 }
 
+func getRequestsStartTime(requests []kubeobjects.Request) (metav1.Time, error) {
+	for _, request := range requests {
+		if request != nil {
+			return request.StartTime(), nil
+		}
+	}
+
+	return metav1.Time{}, fmt.Errorf("no requests found")
+}
+
 func (v *VRGInstance) kubeObjectsRecoveryStartOrResume(
 	result *ctrl.Result, s3ProfileName string,
 	captureToRecoverFromIdentifier *ramen.KubeObjectsCaptureIdentifier,
@@ -674,7 +704,11 @@ func (v *VRGInstance) kubeObjectsRecoveryStartOrResume(
 		}
 	}
 
-	startTime := requests[0].StartTime()
+	startTime, err := getRequestsStartTime(requests)
+	if err != nil {
+		return fmt.Errorf("kube objects recovery couldn't get start time: %v", err)
+	}
+
 	duration := time.Since(startTime.Time)
 	log.Info("Kube objects recovered", "groups", len(groups), "start", startTime, "duration", duration)
 
@@ -799,19 +833,56 @@ func kubeObjectsRequestsWatch(
 	return b
 }
 
-func getCaptureGroups(recipe Recipe.Recipe) ([]kubeobjects.CaptureSpec, error) {
-	var workflow *Recipe.Workflow
-
+func getBackupWorkflow(recipe Recipe.Recipe) (*Recipe.Workflow, error) {
 	for _, w := range recipe.Spec.Workflows {
 		if w.Name == Recipe.BackupWorkflowName {
-			workflow = w
+			return w, nil
+		}
+	}
 
-			break
+	return nil, ErrWorkflowNotFound
+}
+
+func getRestoreWorkflow(recipe Recipe.Recipe) (*Recipe.Workflow, error) {
+	for _, w := range recipe.Spec.Workflows {
+		if w.Name == Recipe.RestoreWorkflowName {
+			return w, nil
 		}
 	}
 
-	if workflow == nil {
-		return nil, ErrWorkflowNotFound
+	return nil, ErrWorkflowNotFound
+}
+
+func validateWorkflow(workflow *Recipe.Workflow) error {
+	if len(workflow.Sequence) == 0 {
+		return fmt.Errorf("a workflow cannot be empty")
+	}
+
+	var workflowHasResourceTypeGroup bool
+
+	for _, resource := range workflow.Sequence {
+		for resourceType := range resource {
+			if resourceType == "group" {
+				workflowHasResourceTypeGroup = true
+			}
+		}
+	}
+
+	if !workflowHasResourceTypeGroup {
+		return fmt.Errorf("a workflow must contain at least one group")
+	}
+
+	return nil
+}
+
+func getCaptureGroups(recipe Recipe.Recipe) ([]kubeobjects.CaptureSpec, error) {
+	workflow, err := getBackupWorkflow(recipe)
+	if err != nil {
+		return nil, err
+	}
+
+	if err := validateWorkflow(workflow); err != nil {
+		return nil, err
 	}
 
 	resources := make([]kubeobjects.CaptureSpec, len(workflow.Sequence))
@@ -837,18 +908,13 @@ func getCaptureGroups(recipe Recipe.Recipe) ([]kubeobjects.CaptureSpec, error) {
 }
 
 func getRecoverGroups(recipe Recipe.Recipe) ([]kubeobjects.RecoverSpec, error) {
-	var workflow *Recipe.Workflow
-
-	for _, w := range recipe.Spec.Workflows {
-		if w.Name == Recipe.RestoreWorkflowName {
-			workflow = w
-
-			break
-		}
+	workflow, err := getRestoreWorkflow(recipe)
+	if err != nil {
+		return nil, err
 	}
 
-	if workflow == nil {
-		return nil, ErrWorkflowNotFound
+	if err := validateWorkflow(workflow); err != nil {
+		return nil, err
 	}
 
 	resources := make([]kubeobjects.RecoverSpec, len(workflow.Sequence))