Add ExecutionProgress() method to function interface to report exec progress

pkg/apis/cr/v1alpha1/types.go

@@ -147,7 +147,8 @@ type ActionStatus struct {
 	DeferPhase Phase `json:"deferPhase,omitempty"`
 }
 
-// ActionProgress provides information on the progress of an action.
+// ActionProgress provides information on the combined progress
+// of all the phases in the action.
 type ActionProgress struct {
 	// RunningPhase represents which phase of the action is being run
 	RunningPhase string `json:"runningPhase,omitempty"`
@@ -187,6 +188,28 @@ type Phase struct {
 	State State `json:"state"`
 	// Output is the map of output artifacts produced by the Blueprint phase.
 	Output map[string]interface{} `json:"output,omitempty"`
+	// Progress represents the phase execution progress.
+	Progress PhaseProgress `json:"progress,omitempty"`
+}
+
+// PhaseProgress represents the execution state of the phase.
+type PhaseProgress struct {
+	// ProgressPercent represents the execution progress in percentage.
+	ProgressPercent string `json:"progressPercent,omitempty"`
+	// SizeUploadedB represents the size of data uploaded in Bytes at a given time during phase execution.
+	// This field will be empty for phases which do not involve data movement.
+	SizeUploadedB int64 `json:"sizeUploadedB,omitempty"`
+	// EstimatedUploadSizeB represents the total estimated size of data in Bytes
+	// that will be uploaded during the phase execution.
+	// This field will be empty for phases which do not involve data movement.
+	EstimatedUploadSizeB int64 `json:"estinatedUploadSizeB,omitempty"`
+	// EstimatedTimeSeconds is the estimated time required in seconds to upload the
+	// remaining data estimated with EstimatedUploadSizeB.
+	// This field will be empty for phases which do not involve data movement.
+	EstimatedTimeSeconds int64 `json:"estinatedTimeSeconds,omitempty"`
+	// LastTransitionTime represents the last date time when the progress status
+	// was received.
+	LastTransitionTime *metav1.Time `json:"lastTransitionTime,omitempty"`
 }
 
 // k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object

pkg/blueprint/validate/validate_test.go

@@ -438,7 +438,9 @@ func blueprint() *crv1alpha1.Blueprint {
 	}
 }
 
-type nonDefaultVersionFunc struct{}
+type nonDefaultVersionFunc struct {
+	progressPercent string
+}
 
 func (nd *nonDefaultVersionFunc) Name() string {
 	return "NonDefaultVersionFunc"
@@ -453,9 +455,15 @@ func (nd *nonDefaultVersionFunc) Arguments() []string {
 }
 
 func (nd *nonDefaultVersionFunc) Exec(context.Context, param.TemplateParams, map[string]interface{}) (map[string]interface{}, error) {
+	nd.progressPercent = "0"
+	defer func() { nd.progressPercent = "100" }()
 	return nil, nil
 }
 
+func (nd *nonDefaultVersionFunc) ExecutionProgress() (crv1alpha1.PhaseProgress, error) {
+	return crv1alpha1.PhaseProgress{ProgressPercent: nd.progressPercent}, nil
+}
+
 var _ kanister.Func = (*nonDefaultVersionFunc)(nil)
 
 func init() {

pkg/controller/controller.go

@@ -394,14 +394,6 @@ func (c *Controller) handleActionSet(ctx context.Context, t *tomb.Tomb, as *crv1
 		}
 	}
 
-	go func() {
-		// progress update is computed on a best-effort basis.
-		// if it exits with error, we will just log it.
-		if err := progress.TrackActionsProgress(ctx, c.crClient, as.GetName(), as.GetNamespace()); err != nil {
-			log.Error().WithError(err)
-		}
-	}()
-
 	for i, a := range as.Status.Actions {
 		var bp *crv1alpha1.Blueprint
 		if bp, err = c.crClient.CrV1alpha1().Blueprints(as.GetNamespace()).Get(ctx, a.Blueprint, v1.GetOptions{}); err != nil {
@@ -471,7 +463,7 @@ func (c *Controller) runAction(ctx context.Context, t *tomb.Tomb, as *crv1alpha1
 		defer func() {
 			var deferErr error
 			if deferPhase != nil {
-				c.updateActionSetRunningPhase(ctx, as, deferPhase.Name())
+				c.updateActionSetRunningPhase(ctx, aIDX, as, deferPhase.Name())
 				deferErr = c.executeDeferPhase(ctx, deferPhase, tp, bp, action.Name, aIDX, as)
 			}
 			// render artifacts only if all the phases are run successfully
@@ -490,11 +482,22 @@ func (c *Controller) runAction(ctx context.Context, t *tomb.Tomb, as *crv1alpha1
 			var output map[string]interface{}
 			var msg string
 			if err == nil {
-				c.updateActionSetRunningPhase(ctx, as, p.Name())
+				c.updateActionSetRunningPhase(ctx, aIDX, as, p.Name())
+				progressTrackCtx, doneProgressTrack := context.WithCancel(ctx)
+				defer doneProgressTrack()
+				go func() {
+					// progress update is computed on a best-effort basis.
+					// if it exits with error, we will just log it.
+					if err := progress.UpdateActionSetsProgress(progressTrackCtx, aIDX, c.crClient, as.GetName(), as.GetNamespace(), p); err != nil {
+						log.Error().WithError(err)
+					}
+				}()
 				output, err = p.Exec(ctx, *bp, action.Name, *tp)
+				doneProgressTrack()
 			} else {
 				msg = fmt.Sprintf("Failed to init phase params: %#v:", as.Status.Actions[aIDX].Phases[i])
 			}
+
 			var rf func(*crv1alpha1.ActionSet) error
 			if err != nil {
 				coreErr = err
@@ -511,8 +514,17 @@ func (c *Controller) runAction(ctx context.Context, t *tomb.Tomb, as *crv1alpha1
 				coreErr = nil
 				rf = func(ras *crv1alpha1.ActionSet) error {
 					ras.Status.Actions[aIDX].Phases[i].State = crv1alpha1.StateComplete
+					pp, err := p.Progress()
+					if err != nil {
+						log.Error().WithError(err)
+						return nil
+					}
+					ras.Status.Actions[aIDX].Phases[i].Progress = pp
 					// this updates the phase output in the actionset status
 					ras.Status.Actions[aIDX].Phases[i].Output = output
+					if err := progress.SetActionSetPercentCompleted(ras); err != nil {
+						log.Error().WithError(err)
+					}
 					return nil
 				}
 			}
@@ -545,9 +557,15 @@ func (c *Controller) runAction(ctx context.Context, t *tomb.Tomb, as *crv1alpha1
 // updateActionSetRunningPhase updates the actionset's `status.Progress.RunningPhase` with the phase name
 // that is being run currently. It doesn't fail if there was a problem updating the actionset. It just logs
 // the failure.
-func (c *Controller) updateActionSetRunningPhase(ctx context.Context, as *crv1alpha1.ActionSet, phase string) {
+func (c *Controller) updateActionSetRunningPhase(ctx context.Context, aIDX int, as *crv1alpha1.ActionSet, phase string) {
 	err := reconcile.ActionSet(ctx, c.crClient.CrV1alpha1(), as.Namespace, as.Name, func(as *crv1alpha1.ActionSet) error {
 		as.Status.Progress.RunningPhase = phase
+		// Iterate through all the phases and set current phase state to running
+		for i := 0; i < len(as.Status.Actions[aIDX].Phases); i++ {
+			if as.Status.Actions[aIDX].Phases[i].Name == phase {
+				as.Status.Actions[aIDX].Phases[i].State = crv1alpha1.StateRunning
+			}
+		}
 		return nil
 	})
 	if err != nil {

pkg/customresource/actionset.yaml

@@ -267,6 +267,20 @@ spec:
                             type: object
                           state:
                             type: string
+                          progress:
+                            properties:
+                              progressPercent:
+                                type: string
+                              sizeUploadedB:
+                                type: integer
+                              estimatedTimeSeconds:
+                                type: integer
+                              estimatedUploadSizeB:
+                                type: integer
+                              lastTransitionTime:
+                                type: string
+                                format: date-time
+                            type: object
                         type: object
                       phases:
                         description: Phases are sub-actions an are executed sequentially.
@@ -279,6 +293,20 @@ spec:
                               type: object
                             state:
                               type: string
+                            progress:
+                              properties:
+                                progressPercent:
+                                  type: string
+                                sizeUploadedB:
+                                  type: integer
+                                estimatedTimeSeconds:
+                                  type: integer
+                                estimatedUploadSizeB:
+                                  type: integer
+                                lastTransitionTime:
+                                  type: string
+                                  format: date-time
+                              type: object
                           type: object
                         type: array
                     type: object

pkg/function/backup_data.go

@@ -16,18 +16,22 @@ package function
 
 import (
 	"context"
+	"time"
 
 	"github.com/pkg/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/rand"
 	"k8s.io/client-go/kubernetes"
 
 	kanister "github.com/kanisterio/kanister/pkg"
+	crv1alpha1 "github.com/kanisterio/kanister/pkg/apis/cr/v1alpha1"
 	"github.com/kanisterio/kanister/pkg/consts"
 	"github.com/kanisterio/kanister/pkg/field"
 	"github.com/kanisterio/kanister/pkg/format"
 	"github.com/kanisterio/kanister/pkg/kube"
 	"github.com/kanisterio/kanister/pkg/log"
 	"github.com/kanisterio/kanister/pkg/param"
+	"github.com/kanisterio/kanister/pkg/progress"
 	"github.com/kanisterio/kanister/pkg/restic"
 )
 
@@ -64,13 +68,19 @@ func init() {
 
 var _ kanister.Func = (*backupDataFunc)(nil)
 
-type backupDataFunc struct{}
+type backupDataFunc struct {
+	progressPercent string
+}
 
 func (*backupDataFunc) Name() string {
 	return BackupDataFuncName
 }
 
-func (*backupDataFunc) Exec(ctx context.Context, tp param.TemplateParams, args map[string]interface{}) (map[string]interface{}, error) {
+func (b *backupDataFunc) Exec(ctx context.Context, tp param.TemplateParams, args map[string]interface{}) (map[string]interface{}, error) {
+	// Set progress percent
+	b.progressPercent = progress.StartedPercent
+	defer func() { b.progressPercent = progress.CompletedPercent }()
+
 	var namespace, pod, container, includePath, backupArtifactPrefix, encryptionKey string
 	var err error
 	if err = Arg(args, BackupDataNamespaceArg, &namespace); err != nil {
@@ -188,3 +198,11 @@ func backupData(ctx context.Context, cli kubernetes.Interface, namespace, pod, c
 		phySize:    phySize,
 	}, nil
 }
+
+func (b *backupDataFunc) ExecutionProgress() (crv1alpha1.PhaseProgress, error) {
+	metav1Time := metav1.NewTime(time.Now())
+	return crv1alpha1.PhaseProgress{
+		ProgressPercent:    b.progressPercent,
+		LastTransitionTime: &metav1Time,
+	}, nil
+}

pkg/function/backup_data_all.go

@@ -19,15 +19,19 @@ import (
 	"encoding/json"
 	"fmt"
 	"strings"
+	"time"
 
 	"github.com/pkg/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
 
 	kanister "github.com/kanisterio/kanister/pkg"
+	crv1alpha1 "github.com/kanisterio/kanister/pkg/apis/cr/v1alpha1"
 	"github.com/kanisterio/kanister/pkg/consts"
 	"github.com/kanisterio/kanister/pkg/field"
 	"github.com/kanisterio/kanister/pkg/kube"
 	"github.com/kanisterio/kanister/pkg/param"
+	"github.com/kanisterio/kanister/pkg/progress"
 	"github.com/kanisterio/kanister/pkg/restic"
 )
 
@@ -62,13 +66,19 @@ func init() {
 
 var _ kanister.Func = (*backupDataAllFunc)(nil)
 
-type backupDataAllFunc struct{}
+type backupDataAllFunc struct {
+	progressPercent string
+}
 
 func (*backupDataAllFunc) Name() string {
 	return BackupDataAllFuncName
 }
 
-func (*backupDataAllFunc) Exec(ctx context.Context, tp param.TemplateParams, args map[string]interface{}) (map[string]interface{}, error) {
+func (b *backupDataAllFunc) Exec(ctx context.Context, tp param.TemplateParams, args map[string]interface{}) (map[string]interface{}, error) {
+	// Set progress percent
+	b.progressPercent = progress.StartedPercent
+	defer func() { b.progressPercent = progress.CompletedPercent }()
+
 	var namespace, pods, container, includePath, backupArtifactPrefix, encryptionKey string
 	var err error
 	if err = Arg(args, BackupDataAllNamespaceArg, &namespace); err != nil {
@@ -172,3 +182,11 @@ func backupDataAll(ctx context.Context, cli kubernetes.Interface, namespace stri
 		FunctionOutputVersion: kanister.DefaultVersion,
 	}, nil
 }
+
+func (b *backupDataAllFunc) ExecutionProgress() (crv1alpha1.PhaseProgress, error) {
+	metav1Time := metav1.NewTime(time.Now())
+	return crv1alpha1.PhaseProgress{
+		ProgressPercent:    b.progressPercent,
+		LastTransitionTime: &metav1Time,
+	}, nil
+}

pkg/function/backup_data_stats.go

@@ -17,8 +17,10 @@ package function
 import (
 	"bytes"
 	"context"
+	"time"
 
 	"github.com/pkg/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
 
 	kanister "github.com/kanisterio/kanister/pkg"
@@ -27,6 +29,7 @@ import (
 	"github.com/kanisterio/kanister/pkg/format"
 	"github.com/kanisterio/kanister/pkg/kube"
 	"github.com/kanisterio/kanister/pkg/param"
+	"github.com/kanisterio/kanister/pkg/progress"
 	"github.com/kanisterio/kanister/pkg/restic"
 )
 
@@ -56,7 +59,9 @@ func init() {
 
 var _ kanister.Func = (*BackupDataStatsFunc)(nil)
 
-type BackupDataStatsFunc struct{}
+type BackupDataStatsFunc struct {
+	progressPercent string
+}
 
 func (*BackupDataStatsFunc) Name() string {
 	return BackupDataStatsFuncName
@@ -130,7 +135,11 @@ func backupDataStatsPodFunc(
 	}
 }
 
-func (*BackupDataStatsFunc) Exec(ctx context.Context, tp param.TemplateParams, args map[string]interface{}) (map[string]interface{}, error) {
+func (b *BackupDataStatsFunc) Exec(ctx context.Context, tp param.TemplateParams, args map[string]interface{}) (map[string]interface{}, error) {
+	// Set progress percent
+	b.progressPercent = progress.StartedPercent
+	defer func() { b.progressPercent = progress.CompletedPercent }()
+
 	var namespace, backupArtifactPrefix, backupID, mode, encryptionKey string
 	var err error
 	if err = Arg(args, BackupDataStatsNamespaceArg, &namespace); err != nil {
@@ -183,3 +192,11 @@ func (*BackupDataStatsFunc) Arguments() []string {
 		BackupDataStatsEncryptionKeyArg,
 	}
 }
+
+func (b *BackupDataStatsFunc) ExecutionProgress() (crv1alpha1.PhaseProgress, error) {
+	metav1Time := metav1.NewTime(time.Now())
+	return crv1alpha1.PhaseProgress{
+		ProgressPercent:    b.progressPercent,
+		LastTransitionTime: &metav1Time,
+	}, nil
+}