mirrors/velero
1
0
Fork 0
mirror of https://github.com/vmware-tanzu/velero.git synced 2026-08-21 22:56:05 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

data mover ms smoke testing

Browse Source 
Signed-off-by: Lyndon-Li <lyonghui@vmware.com>
This commit is contained in:
Lyndon-Li
2024-08-14 17:29:04 +08:00
parent 3c0948c9be
commit ed0ef67c16
10 changed files with 50 additions and 117 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/cmd/cli/datamover/backup.go
+1
View File
@@ -224,6 +224,7 @@ func (s *dataMoverBackup) runDataPath() {
err = dpService.Init()
if err != nil {
dpService.Shutdown()
s.cancelFunc()
funcExitWithMessage(s.logger, false, "Failed to init data path service for DataUpload %s: %v", s.config.duName, err)
return
pkg/cmd/cli/datamover/restore.go
+2 -1
View File
@@ -215,6 +215,7 @@ func (s *dataMoverRestore) runDataPath() {
err = dpService.Init()
if err != nil {
dpService.Shutdown()
s.cancelFunc()
funcExitWithMessage(s.logger, false, "Failed to init data path service for DataDownload %s: %v", s.config.ddName, err)
return
@@ -222,8 +223,8 @@ func (s *dataMoverRestore) runDataPath() {
result, err := dpService.RunCancelableDataPath(s.ctx)
if err != nil {
s.cancelFunc()
dpService.Shutdown()
s.cancelFunc()
funcExitWithMessage(s.logger, false, "Failed to run data path service for DataDownload %s: %v", s.config.ddName, err)
return
}
pkg/cmd/cli/nodeagent/server.go
+2 -4
View File
@@ -292,12 +292,12 @@ func (s *nodeAgentServer) run() {
if s.dataPathConfigs != nil && len(s.dataPathConfigs.LoadAffinity) > 0 {
loadAffinity = s.dataPathConfigs.LoadAffinity[0]
}
dataUploadReconciler := controller.NewDataUploadReconciler(s.mgr.GetClient(), s.mgr, s.kubeClient, s.csiSnapshotClient.SnapshotV1(), s.dataPathMgr, loadAffinity, repoEnsurer, clock.RealClock{}, credentialGetter, s.nodeName, s.fileSystem, s.config.dataMoverPrepareTimeout, s.logger, s.metrics)
dataUploadReconciler := controller.NewDataUploadReconciler(s.mgr.GetClient(), s.mgr, s.kubeClient, s.csiSnapshotClient.SnapshotV1(), s.dataPathMgr, loadAffinity, clock.RealClock{}, s.nodeName, s.config.dataMoverPrepareTimeout, s.logger, s.metrics)
if err = dataUploadReconciler.SetupWithManager(s.mgr); err != nil {
s.logger.WithError(err).Fatal("Unable to create the data upload controller")
}
dataDownloadReconciler := controller.NewDataDownloadReconciler(s.mgr.GetClient(), s.mgr, s.kubeClient, s.dataPathMgr, repoEnsurer, credentialGetter, s.nodeName, s.config.dataMoverPrepareTimeout, s.logger, s.metrics)
dataDownloadReconciler := controller.NewDataDownloadReconciler(s.mgr.GetClient(), s.mgr, s.kubeClient, s.dataPathMgr, s.nodeName, s.config.dataMoverPrepareTimeout, s.logger, s.metrics)
if err = dataDownloadReconciler.SetupWithManager(s.mgr); err != nil {
s.logger.WithError(err).Fatal("Unable to create the data download controller")
}
@@ -312,8 +312,6 @@ func (s *nodeAgentServer) run() {
if err := dataDownloadReconciler.AttemptDataDownloadResume(s.ctx, s.mgr.GetClient(), s.logger.WithField("node", s.nodeName), s.namespace); err != nil {
s.logger.WithError(errors.WithStack(err)).Error("failed to attempt data download resume")
}
s.logger.Info("Attempt complete to resume dataUploads and dataDownloads")
}()
s.logger.Info("Controllers starting...")
pkg/controller/data_download_controller.go
+25 -33
View File
@@ -39,7 +39,6 @@ import (
"sigs.k8s.io/controller-runtime/pkg/predicate"
"sigs.k8s.io/controller-runtime/pkg/reconcile"
"github.com/vmware-tanzu/velero/internal/credentials"
"github.com/vmware-tanzu/velero/pkg/apis/velero/shared"
velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
velerov2alpha1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v2alpha1"
@@ -47,45 +46,37 @@ import (
"github.com/vmware-tanzu/velero/pkg/datapath"
"github.com/vmware-tanzu/velero/pkg/exposer"
"github.com/vmware-tanzu/velero/pkg/metrics"
repository "github.com/vmware-tanzu/velero/pkg/repository"
"github.com/vmware-tanzu/velero/pkg/uploader"
"github.com/vmware-tanzu/velero/pkg/util/filesystem"
"github.com/vmware-tanzu/velero/pkg/util/kube"
)
// DataDownloadReconciler reconciles a DataDownload object
type DataDownloadReconciler struct {
client client.Client
kubeClient kubernetes.Interface
mgr manager.Manager
logger logrus.FieldLogger
credentialGetter *credentials.CredentialGetter
fileSystem filesystem.Interface
Clock clock.WithTickerAndDelayedExecution
restoreExposer exposer.GenericRestoreExposer
nodeName string
repositoryEnsurer *repository.Ensurer
dataPathMgr *datapath.Manager
preparingTimeout time.Duration
metrics *metrics.ServerMetrics
client client.Client
kubeClient kubernetes.Interface
mgr manager.Manager
logger logrus.FieldLogger
Clock clock.WithTickerAndDelayedExecution
restoreExposer exposer.GenericRestoreExposer
nodeName string
dataPathMgr *datapath.Manager
preparingTimeout time.Duration
metrics *metrics.ServerMetrics
}
func NewDataDownloadReconciler(client client.Client, mgr manager.Manager, kubeClient kubernetes.Interface, dataPathMgr *datapath.Manager,
repoEnsurer *repository.Ensurer, credentialGetter *credentials.CredentialGetter, nodeName string, preparingTimeout time.Duration, logger logrus.FieldLogger, metrics *metrics.ServerMetrics) *DataDownloadReconciler {
nodeName string, preparingTimeout time.Duration, logger logrus.FieldLogger, metrics *metrics.ServerMetrics) *DataDownloadReconciler {
return &DataDownloadReconciler{
client: client,
kubeClient: kubeClient,
mgr: mgr,
logger: logger.WithField("controller", "DataDownload"),
credentialGetter: credentialGetter,
fileSystem: filesystem.NewFileSystem(),
Clock: &clock.RealClock{},
nodeName: nodeName,
repositoryEnsurer: repoEnsurer,
restoreExposer: exposer.NewGenericRestoreExposer(kubeClient, logger),
dataPathMgr: dataPathMgr,
preparingTimeout: preparingTimeout,
metrics: metrics,
client: client,
kubeClient: kubeClient,
mgr: mgr,
logger: logger.WithField("controller", "DataDownload"),
Clock: &clock.RealClock{},
nodeName: nodeName,
restoreExposer: exposer.NewGenericRestoreExposer(kubeClient, logger),
dataPathMgr: dataPathMgr,
preparingTimeout: preparingTimeout,
metrics: metrics,
}
}
@@ -282,7 +273,7 @@ func (r *DataDownloadReconciler) Reconcile(ctx context.Context, req ctrl.Request
}
if err := r.initCancelableDataPath(ctx, asyncBR, result, log); err != nil {
log.WithError(err).Warnf("Failed to init cancelable data path for %s, will close and retry", dd.Name)
log.WithError(err).Errorf("Failed to init cancelable data path for %s", dd.Name)
r.closeDataPath(ctx, dd.Name)
return r.errorOut(ctx, dd, err, "error initializing data path", log)
@@ -372,7 +363,7 @@ func (r *DataDownloadReconciler) startCancelableDataPath(asyncBR datapath.AsyncB
return errors.Wrapf(err, "error starting async restore for pod %s, volume %s", res.ByPod.HostingPod.Name, res.ByPod.VolumeName)
}
log.Info("Async restore started for pod %s, volume %s", res.ByPod.HostingPod.Name, res.ByPod.VolumeName)
log.Infof("Async restore started for pod %s, volume %s", res.ByPod.HostingPod.Name, res.ByPod.VolumeName)
return nil
}
@@ -420,7 +411,7 @@ func (r *DataDownloadReconciler) OnDataDownloadFailed(ctx context.Context, names
if getErr := r.client.Get(ctx, types.NamespacedName{Name: ddName, Namespace: namespace}, &dd); getErr != nil {
log.WithError(getErr).Warn("Failed to get data download on failure")
} else {
r.errorOut(ctx, &dd, err, "data path restore failed", log)
_, _ = r.errorOut(ctx, &dd, err, "data path restore failed", log)
}
}
@@ -797,6 +788,7 @@ func (r *DataDownloadReconciler) AttemptDataDownloadResume(ctx context.Context,
err := funcResumeCancellableDataRestore(r, ctx, dd, logger)
if err == nil {
logger.WithField("dd", dd.Name).WithField("current node", r.nodeName).Info("Completed to resume in progress DD")
continue
}
pkg/controller/data_download_controller_test.go
+1 -12
View File
@@ -43,7 +43,6 @@ import (
"sigs.k8s.io/controller-runtime/pkg/client/fake"
"github.com/vmware-tanzu/velero/internal/credentials"
velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
velerov2alpha1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v2alpha1"
"github.com/vmware-tanzu/velero/pkg/builder"
@@ -139,19 +138,9 @@ func initDataDownloadReconcilerWithError(objects []runtime.Object, needError ...
return nil, err
}
credentialFileStore, err := credentials.NewNamespacedFileStore(
fakeClient,
velerov1api.DefaultNamespace,
"/tmp/credentials",
fakeFS,
)
if err != nil {
return nil, err
}
dataPathMgr := datapath.NewManager(1)
return NewDataDownloadReconciler(fakeClient, nil, fakeKubeClient, dataPathMgr, nil, &credentials.CredentialGetter{FromFile: credentialFileStore}, "test-node", time.Minute*5, velerotest.NewLogger(), metrics.NewServerMetrics()), nil
return NewDataDownloadReconciler(fakeClient, nil, fakeKubeClient, dataPathMgr, "test-node", time.Minute*5, velerotest.NewLogger(), metrics.NewServerMetrics()), nil
}
func TestDataDownloadReconcile(t *testing.T) {
pkg/controller/data_upload_controller.go
+8 -16
View File
@@ -41,7 +41,6 @@ import (
snapshotter "github.com/kubernetes-csi/external-snapshotter/client/v7/clientset/versioned/typed/volumesnapshot/v1"
"github.com/vmware-tanzu/velero/internal/credentials"
"github.com/vmware-tanzu/velero/pkg/apis/velero/shared"
velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
velerov2alpha1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v2alpha1"
@@ -50,9 +49,7 @@ import (
"github.com/vmware-tanzu/velero/pkg/exposer"
"github.com/vmware-tanzu/velero/pkg/metrics"
"github.com/vmware-tanzu/velero/pkg/nodeagent"
"github.com/vmware-tanzu/velero/pkg/repository"
"github.com/vmware-tanzu/velero/pkg/uploader"
"github.com/vmware-tanzu/velero/pkg/util/filesystem"
"github.com/vmware-tanzu/velero/pkg/util/kube"
)
@@ -69,11 +66,8 @@ type DataUploadReconciler struct {
kubeClient kubernetes.Interface
csiSnapshotClient snapshotter.SnapshotV1Interface
mgr manager.Manager
repoEnsurer *repository.Ensurer
Clock clocks.WithTickerAndDelayedExecution
credentialGetter *credentials.CredentialGetter
nodeName string
fileSystem filesystem.Interface
logger logrus.FieldLogger
snapshotExposerList map[velerov2alpha1api.SnapshotType]exposer.SnapshotExposer
dataPathMgr *datapath.Manager
@@ -83,19 +77,16 @@ type DataUploadReconciler struct {
}
func NewDataUploadReconciler(client client.Client, mgr manager.Manager, kubeClient kubernetes.Interface, csiSnapshotClient snapshotter.SnapshotV1Interface,
dataPathMgr *datapath.Manager, loadAffinity *nodeagent.LoadAffinity, repoEnsurer *repository.Ensurer, clock clocks.WithTickerAndDelayedExecution,
cred *credentials.CredentialGetter, nodeName string, fs filesystem.Interface, preparingTimeout time.Duration, log logrus.FieldLogger, metrics *metrics.ServerMetrics) *DataUploadReconciler {
dataPathMgr *datapath.Manager, loadAffinity *nodeagent.LoadAffinity, clock clocks.WithTickerAndDelayedExecution, nodeName string, preparingTimeout time.Duration,
log logrus.FieldLogger, metrics *metrics.ServerMetrics) *DataUploadReconciler {
return &DataUploadReconciler{
client: client,
mgr: mgr,
kubeClient: kubeClient,
csiSnapshotClient: csiSnapshotClient,
Clock: clock,
credentialGetter: cred,
nodeName: nodeName,
fileSystem: fs,
logger: log,
repoEnsurer: repoEnsurer,
snapshotExposerList: map[velerov2alpha1api.SnapshotType]exposer.SnapshotExposer{velerov2alpha1api.SnapshotTypeCSI: exposer.NewCSISnapshotExposer(kubeClient, csiSnapshotClient, log)},
dataPathMgr: dataPathMgr,
loadAffinity: loadAffinity,
@@ -293,7 +284,7 @@ func (r *DataUploadReconciler) Reconcile(ctx context.Context, req ctrl.Request)
}
if err := r.initCancelableDataPath(ctx, asyncBR, res, log); err != nil {
log.WithError(err).Warnf("Failed to init cancelable data path for %s, will close and retry", du.Name)
log.WithError(err).Errorf("Failed to init cancelable data path for %s", du.Name)
r.closeDataPath(ctx, du.Name)
return r.errorOut(ctx, du, err, "error initializing data path", log)
@@ -383,7 +374,7 @@ func (r *DataUploadReconciler) startCancelableDataPath(asyncBR datapath.AsyncBR,
return errors.Wrapf(err, "error starting async backup for pod %s, volume %s", res.ByPod.HostingPod.Name, res.ByPod.VolumeName)
}
log.Info("Async backup started for pod %s, volume %s", res.ByPod.HostingPod.Name, res.ByPod.VolumeName)
log.Infof("Async backup started for pod %s, volume %s", res.ByPod.HostingPod.Name, res.ByPod.VolumeName)
return nil
}
@@ -442,7 +433,7 @@ func (r *DataUploadReconciler) OnDataUploadFailed(ctx context.Context, namespace
if getErr := r.client.Get(ctx, types.NamespacedName{Name: duName, Namespace: namespace}, &du); getErr != nil {
log.WithError(getErr).Warn("Failed to get dataupload on failure")
} else {
r.errorOut(ctx, &du, err, "data path backup failed", log)
_, _ = r.errorOut(ctx, &du, err, "data path backup failed", log)
}
}
@@ -502,7 +493,7 @@ func (r *DataUploadReconciler) tryCancelAcceptedDataUpload(ctx context.Context,
r.cleanUp(ctx, du, log)
}
func (r *DataUploadReconciler) cleanUp(ctx context.Context, du *velerov2alpha1api.DataUpload, log *logrus.Entry) {
func (r *DataUploadReconciler) cleanUp(ctx context.Context, du *velerov2alpha1api.DataUpload, log logrus.FieldLogger) {
ep, ok := r.snapshotExposerList[du.Spec.SnapshotType]
if !ok {
log.WithError(fmt.Errorf("%v type of snapshot exposer is not exist", du.Spec.SnapshotType)).
@@ -650,7 +641,7 @@ func (r *DataUploadReconciler) errorOut(ctx context.Context, du *velerov2alpha1a
}
se.CleanUp(ctx, getOwnerObject(du), volumeSnapshotName, du.Spec.SourceNamespace)
} else {
log.Warnf("failed to clean up exposed snapshot could not find %s snapshot exposer", du.Spec.SnapshotType)
log.Errorf("failed to clean up exposed snapshot could not find %s snapshot exposer", du.Spec.SnapshotType)
}
return ctrl.Result{}, r.updateStatusToFailed(ctx, du, err, msg, log)
@@ -894,6 +885,7 @@ func (r *DataUploadReconciler) AttemptDataUploadResume(ctx context.Context, cli
err := funcResumeCancellableDataBackup(r, ctx, du, logger)
if err == nil {
logger.WithField("du", du.Name).WithField("current node", r.nodeName).Info("Completed to resume in progress DU")
continue
}
pkg/controller/data_upload_controller_test.go
+2 -19
View File
@@ -47,7 +47,6 @@ import (
"sigs.k8s.io/controller-runtime/pkg/manager"
"sigs.k8s.io/controller-runtime/pkg/reconcile"
"github.com/vmware-tanzu/velero/internal/credentials"
velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
velerov2alpha1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v2alpha1"
"github.com/vmware-tanzu/velero/pkg/builder"
@@ -229,24 +228,9 @@ func initDataUploaderReconcilerWithError(needError ...error) (*DataUploadReconci
fakeSnapshotClient := snapshotFake.NewSimpleClientset(vsObject, vscObj)
fakeKubeClient := clientgofake.NewSimpleClientset(daemonSet)
fakeFS := velerotest.NewFakeFileSystem()
pathGlob := fmt.Sprintf("/host_pods/%s/volumes/*/%s", "", dataUploadName)
_, err = fakeFS.Create(pathGlob)
if err != nil {
return nil, err
}
credentialFileStore, err := credentials.NewNamespacedFileStore(
fakeClient,
velerov1api.DefaultNamespace,
"/tmp/credentials",
fakeFS,
)
if err != nil {
return nil, err
}
return NewDataUploadReconciler(fakeClient, nil, fakeKubeClient, fakeSnapshotClient.SnapshotV1(), dataPathMgr, nil, nil,
testclocks.NewFakeClock(now), &credentials.CredentialGetter{FromFile: credentialFileStore}, "test-node", fakeFS, time.Minute*5, velerotest.NewLogger(), metrics.NewServerMetrics()), nil
return NewDataUploadReconciler(fakeClient, nil, fakeKubeClient, fakeSnapshotClient.SnapshotV1(), dataPathMgr, nil,
testclocks.NewFakeClock(now), "test-node", time.Minute*5, velerotest.NewLogger(), metrics.NewServerMetrics()), nil
}
func dataUploadBuilder() *builder.DataUploadBuilder {
@@ -626,7 +610,6 @@ func TestReconcile(t *testing.T) {
if !testCreateFsBR && du.Status.Phase != velerov2alpha1api.DataUploadPhaseInProgress {
assert.Nil(t, r.dataPathMgr.GetAsyncBR(test.du.Name))
}
})
}
}
pkg/controller/pod_volume_backup_controller.go
+3 -3
View File
@@ -202,7 +202,7 @@ func (r *PodVolumeBackupReconciler) Reconcile(ctx context.Context, req ctrl.Requ
}
func (r *PodVolumeBackupReconciler) OnDataPathCompleted(ctx context.Context, namespace string, pvbName string, result datapath.Result) {
defer r.closeDataPath(ctx, pvbName)
defer r.dataPathMgr.RemoveAsyncBR(pvbName)
log := r.logger.WithField("pvb", pvbName)
@@ -240,7 +240,7 @@ func (r *PodVolumeBackupReconciler) OnDataPathCompleted(ctx context.Context, nam
}
func (r *PodVolumeBackupReconciler) OnDataPathFailed(ctx context.Context, namespace, pvbName string, err error) {
defer r.closeDataPath(ctx, pvbName)
defer r.dataPathMgr.RemoveAsyncBR(pvbName)
log := r.logger.WithField("pvb", pvbName)
@@ -255,7 +255,7 @@ func (r *PodVolumeBackupReconciler) OnDataPathFailed(ctx context.Context, namesp
}
func (r *PodVolumeBackupReconciler) OnDataPathCancelled(ctx context.Context, namespace string, pvbName string) {
defer r.closeDataPath(ctx, pvbName)
defer r.dataPathMgr.RemoveAsyncBR(pvbName)
log := r.logger.WithField("pvb", pvbName)
pkg/controller/pod_volume_restore_controller.go
+3 -3
View File
@@ -265,7 +265,7 @@ func getInitContainerIndex(pod *corev1api.Pod) int {
}
func (c *PodVolumeRestoreReconciler) OnDataPathCompleted(ctx context.Context, namespace string, pvrName string, result datapath.Result) {
defer c.closeDataPath(ctx, pvrName)
defer c.dataPathMgr.RemoveAsyncBR(pvrName)
log := c.logger.WithField("pvr", pvrName)
@@ -325,7 +325,7 @@ func (c *PodVolumeRestoreReconciler) OnDataPathCompleted(ctx context.Context, na
}
func (c *PodVolumeRestoreReconciler) OnDataPathFailed(ctx context.Context, namespace string, pvrName string, err error) {
defer c.closeDataPath(ctx, pvrName)
defer c.dataPathMgr.RemoveAsyncBR(pvrName)
log := c.logger.WithField("pvr", pvrName)
@@ -340,7 +340,7 @@ func (c *PodVolumeRestoreReconciler) OnDataPathFailed(ctx context.Context, names
}
func (c *PodVolumeRestoreReconciler) OnDataPathCancelled(ctx context.Context, namespace string, pvrName string) {
defer c.closeDataPath(ctx, pvrName)
defer c.dataPathMgr.RemoveAsyncBR(pvrName)
log := c.logger.WithField("pvr", pvrName)
pkg/datapath/micro_service_watcher.go
+3 -26
View File
@@ -103,8 +103,6 @@ func newMicroServiceBRWatcher(client client.Client, kubeClient kubernetes.Interf
}
func (ms *microServiceBRWatcher) Init(ctx context.Context, param interface{}) error {
succeeded := false
eventInformer, err := ms.mgr.GetCache().GetInformer(ctx, &v1.Event{})
if err != nil {
return errors.Wrap(err, "error getting event informer")
@@ -135,19 +133,10 @@ func (ms *microServiceBRWatcher) Init(ctx context.Context, param interface{}) er
},
},
)
if err != nil {
return errors.Wrap(err, "error registering event handler")
}
defer func() {
if !succeeded {
if err := eventInformer.RemoveEventHandler(eventHandler); err != nil {
ms.log.WithError(err).Warn("Failed to remove event handler")
}
}
}()
podHandler, err := podInformer.AddEventHandler(
cache.ResourceEventHandlerFuncs{
UpdateFunc: func(_, obj interface{}) {
@@ -162,19 +151,10 @@ func (ms *microServiceBRWatcher) Init(ctx context.Context, param interface{}) er
},
},
)
if err != nil {
return errors.Wrap(err, "error registering pod handler")
}
defer func() {
if !succeeded {
if err := podInformer.RemoveEventHandler(podHandler); err != nil {
ms.log.WithError(err).Warn("Failed to remove pod handler")
}
}
}()
if err := ms.reEnsureThisPod(ctx); err != nil {
return err
}
@@ -193,10 +173,7 @@ func (ms *microServiceBRWatcher) Init(ctx context.Context, param interface{}) er
"thisPod": ms.thisPod,
}).Info("MicroServiceBR is initialized")
succeeded = true
return nil
}
func (ms *microServiceBRWatcher) Close(ctx context.Context) {
@@ -301,7 +278,7 @@ func (ms *microServiceBRWatcher) startWatch() {
}
if lastPod == nil {
ms.log.Warn("Watch loop is cancelled on waiting data path pod")
ms.log.Warn("Watch loop is canceled on waiting data path pod")
return
}
@@ -309,7 +286,7 @@ func (ms *microServiceBRWatcher) startWatch() {
for !ms.startedFromEvent || !ms.terminatedFromEvent {
select {
case <-ms.ctx.Done():
ms.log.Warn("Watch loop is cancelled on waiting final event")
ms.log.Warn("Watch loop is canceled on waiting final event")
return
case <-time.After(eventWaitTimeout):
break epilogLoop
@@ -373,7 +350,7 @@ func (ms *microServiceBRWatcher) onEvent(evt *v1.Event) {
case EventReasonCancelling:
ms.log.Infof("Received data path canceling message: %s", evt.Message)
default:
ms.log.Infof("Received event for data path %s,reason: %s, message: %s", ms.taskName, evt.Reason, evt.Message)
ms.log.Infof("Received event for data path %s, reason: %s, message: %s", ms.taskName, evt.Reason, evt.Message)
}
}