Repository: vshn/k8up
Pull request #218 (2020-12-21 11:05:55)

Add JobSucceeded Condition to Status fields
Based on #204

Implements and closes https://github.com/vshn/k8up/issues/212

## Checklist
<!--
Remove items that do not apply. For completed items, change [ ] to [x].
-->

- [x] Keep pull requests small so they can be easily reviewed.
- [x] Update the documentation.
- [x] Update tests.
- [x] Link this PR to related issues.

<!--
NOTE: these things are not required to open a PR and can be done afterwards,
while the PR is open.
-->


cimnine: I will leave this issue in draft state until #204 is merged.

Patch:
diff --git a/docs/modules/ROOT/pages/how-tos/check-status.adoc b/docs/modules/ROOT/pages/how-tos/check-status.adoc
index 38e45109..c4370a7e 100644
--- a/docs/modules/ROOT/pages/how-tos/check-status.adoc
+++ b/docs/modules/ROOT/pages/how-tos/check-status.adoc
@@ -40,6 +40,10 @@ Status:
     Status:                True
     Type:                  JobCreated
     Last Transition Time:  2020-13-32T25:61:61Z
+    Message:               the job 'default/demo-backup' ended successfully
+    Status:                True
+    Type:                  JobSucceeded
+    Last Transition Time:  2020-13-32T25:61:61Z
     Message:               no container definitions found
     Status:                False
     Type:                  PreBackupPodsAvailable
diff --git a/executor/archive.go b/executor/archive.go
index 790a031f..7d1a772f 100644
--- a/executor/archive.go
+++ b/executor/archive.go
@@ -58,8 +58,8 @@ func (a *ArchiveExecutor) Execute() error {
 }
 
 func (a *ArchiveExecutor) startArchive(archiveJob *batchv1.Job, archive *k8upv1alpha1.Archive) {
-	name := types.NamespacedName{Namespace: a.Obj.GetMetaObject().GetNamespace(), Name: a.Obj.GetMetaObject().GetName()}
-	a.setArchiveCallback(name, archive)
+	a.registerArchiveCallback(archive)
+	a.RegisterJobSucceededConditionCallback()
 
 	archiveJob.Spec.Template.Spec.Containers[0].Env = a.setupEnvVars(archive)
 	archiveJob.Spec.Template.Spec.Containers[0].Args = a.setupArgs(archive)
@@ -76,8 +76,9 @@ func (a *ArchiveExecutor) startArchive(archiveJob *batchv1.Job, archive *k8upv1a
 	a.SetStarted(ConditionJobCreated, "the job '%v/%v' was created", archiveJob.Namespace, archiveJob.Name)
 }
 
-func (a *ArchiveExecutor) setArchiveCallback(name types.NamespacedName, archive *k8upv1alpha1.Archive) {
-	observer.GetObserver().RegisterCallback(name.String(), func() {
+func (a *ArchiveExecutor) registerArchiveCallback(archive *k8upv1alpha1.Archive) {
+	name := a.GetJobNamespacedName()
+	observer.GetObserver().RegisterCallback(name.String(), func(_ observer.ObservableJob) {
 		a.cleanupOldArchives(name, archive)
 	})
 }
diff --git a/executor/backup.go b/executor/backup.go
index c745b8f5..fc5fa2f4 100644
--- a/executor/backup.go
+++ b/executor/backup.go
@@ -168,9 +168,8 @@ func (b *BackupExecutor) startBackup(backupJob *batchv1.Job) {
 		return
 	}
 
-	name := types.NamespacedName{Namespace: b.Obj.GetMetaObject().GetNamespace(), Name: b.Obj.GetMetaObject().GetName()}
-
-	b.setBackupCallback(name, preBackup)
+	b.registerBackupCallback(preBackup)
+	b.RegisterJobSucceededConditionCallback()
 
 	volumes := b.listPVCs(cfg.Config.BackupAnnotation)
 
@@ -191,8 +190,9 @@ func (b *BackupExecutor) startBackup(backupJob *batchv1.Job) {
 	b.SetStarted(ConditionJobCreated, "the job '%v/%v' was created", backupJob.Namespace, backupJob.Name)
 }
 
-func (b *BackupExecutor) setBackupCallback(name types.NamespacedName, preBackup *prebackup.PreBackup) {
-	observer.GetObserver().RegisterCallback(name.String(), func() {
+func (b *BackupExecutor) registerBackupCallback(preBackup *prebackup.PreBackup) {
+	name := b.GetJobNamespacedName()
+	observer.GetObserver().RegisterCallback(name.String(), func(_ observer.ObservableJob) {
 		preBackup.Stop()
 		b.cleanupOldBackups(name)
 	})
diff --git a/executor/generic.go b/executor/generic.go
index a8ed8fdc..c5dce3e9 100644
--- a/executor/generic.go
+++ b/executor/generic.go
@@ -8,9 +8,12 @@ import (
 	"fmt"
 	"sort"
 
+	"k8s.io/apimachinery/pkg/types"
+
 	"github.com/vshn/k8up/api/v1alpha1"
 	"github.com/vshn/k8up/cfg"
 	"github.com/vshn/k8up/job"
+	"github.com/vshn/k8up/observer"
 	"github.com/vshn/k8up/queue"
 
 	"github.com/go-logr/logr"
@@ -44,6 +47,9 @@ const (
 	// ConditionJobCreated indicates whether the relevant job was created or not and perhaps why
 	ConditionJobCreated status.ConditionType = "JobCreated"
 
+	// ConditionJobSucceeded indicates that the relevant job has ended and was successful
+	ConditionJobSucceeded status.ConditionType = "JobSucceeded"
+
 	// ConditionCleanupSucceeded indicates whether the cleanup job succeeded
 	ConditionCleanupSucceeded status.ConditionType = "CleanupSucceeded"
 )
@@ -124,10 +130,32 @@ func (g *generic) GetJobNamespace() string {
 	return g.Obj.GetMetaObject().GetNamespace()
 }
 
+func (g *generic) GetJobNamespacedName() types.NamespacedName {
+	return types.NamespacedName{Namespace: g.Obj.GetMetaObject().GetNamespace(), Name: g.Obj.GetMetaObject().GetName()}
+}
+
 func (g *generic) GetJobType() v1alpha1.JobType {
 	return g.Obj.GetType()
 }
 
+// RegisterJobSucceededConditionCallback registers an observer on the job which updates ConditionJobSucceeded when
+// the job succeeds or fails, respectively.
+func (g *generic) RegisterJobSucceededConditionCallback() {
+	name := g.GetJobNamespacedName()
+	observer.GetObserver().RegisterCallback(name.String(), func(event observer.ObservableJob) {
+		switch event.Event {
+		case observer.Suceeded:
+			g.SetConditionTrueWithMessage(ConditionJobSucceeded,
+				"the job '%v/%v' ended successfully",
+				event.Job.Namespace, event.Job.Name)
+		case observer.Failed:
+			g.SetConditionFalse(ConditionJobSucceeded,
+				"the job '%v/%v' failed, please check its log for details",
+				event.Job.Namespace, event.Job.Name)
+		}
+	})
+}
+
 // NewExecutor will return the right Executor for the given job object.
 func NewExecutor(config job.Config) queue.Executor {
 	switch config.Obj.GetType() {
diff --git a/executor/prune.go b/executor/prune.go
index 7e57122b..08b32177 100644
--- a/executor/prune.go
+++ b/executor/prune.go
@@ -63,8 +63,8 @@ func (p *PruneExecutor) Exclusive() bool {
 }
 
 func (p *PruneExecutor) startPrune(pruneJob *batchv1.Job, prune *k8upv1alpha1.Prune) {
-	name := types.NamespacedName{Namespace: p.Obj.GetMetaObject().GetNamespace(), Name: p.Obj.GetMetaObject().GetName()}
-	p.setPruneCallback(name, prune)
+	p.registerPruneCallback(prune)
+	p.RegisterJobSucceededConditionCallback()
 
 	pruneJob.Spec.Template.Spec.Containers[0].Env = p.setupEnvVars(prune)
 	pruneJob.Spec.Template.Spec.Containers[0].Args = []string{"-prune"}
@@ -80,8 +80,9 @@ func (p *PruneExecutor) startPrune(pruneJob *batchv1.Job, prune *k8upv1alpha1.Pr
 	p.SetStarted(ConditionJobCreated, "the job '%v/%v' was created", pruneJob.Namespace, pruneJob.Name)
 }
 
-func (p *PruneExecutor) setPruneCallback(name types.NamespacedName, prune *k8upv1alpha1.Prune) {
-	observer.GetObserver().RegisterCallback(name.String(), func() {
+func (p *PruneExecutor) registerPruneCallback(prune *k8upv1alpha1.Prune) {
+	name := p.GetJobNamespacedName()
+	observer.GetObserver().RegisterCallback(name.String(), func(_ observer.ObservableJob) {
 		p.cleanupOldPrunes(name, prune)
 	})
 }
diff --git a/executor/restore.go b/executor/restore.go
index ceb51146..0e197baa 100644
--- a/executor/restore.go
+++ b/executor/restore.go
@@ -52,8 +52,8 @@ func (r *RestoreExecutor) Execute() error {
 }
 
 func (r *RestoreExecutor) startRestore(restore *k8upv1alpha1.Restore) {
-	name := types.NamespacedName{Namespace: r.Obj.GetMetaObject().GetNamespace(), Name: r.Obj.GetMetaObject().GetName()}
-	r.setRestoreCallback(name, restore)
+	r.registerRestoreCallback(restore)
+	r.RegisterJobSucceededConditionCallback()
 
 	restoreJob, err := r.buildRestoreObject(restore)
 	if err != nil {
@@ -73,8 +73,9 @@ func (r *RestoreExecutor) startRestore(restore *k8upv1alpha1.Restore) {
 	r.SetStarted(ConditionJobCreated, "the job '%v/%v' was created", restoreJob.Namespace, restoreJob.Name)
 }
 
-func (r *RestoreExecutor) setRestoreCallback(name types.NamespacedName, restore *k8upv1alpha1.Restore) {
-	observer.GetObserver().RegisterCallback(name.String(), func() {
+func (r *RestoreExecutor) registerRestoreCallback(restore *k8upv1alpha1.Restore) {
+	name := r.GetJobNamespacedName()
+	observer.GetObserver().RegisterCallback(name.String(), func(_ observer.ObservableJob) {
 		r.cleanupOldRestores(name, restore)
 	})
 }
diff --git a/observer/observer.go b/observer/observer.go
index 73fb57b1..cd5f0473 100644
--- a/observer/observer.go
+++ b/observer/observer.go
@@ -9,10 +9,11 @@ import (
 	"github.com/go-logr/logr"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
-	"github.com/vshn/k8up/api/v1alpha1"
 	batchv1 "k8s.io/api/batch/v1"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/metrics"
+
+	"github.com/vshn/k8up/api/v1alpha1"
 )
 
 const (
@@ -64,9 +65,15 @@ type ObservableJob struct {
 	Event      EventType
 	Exclusive  bool
 	Repository string
-	callback   func()
+	callbacks  []ObservableJobCallback
 }
 
+// ObservableJobCallback is invoked on certain events of the ObservableJob.
+// The related ObservableJob is passed as argument.
+// Check the ObservableJob.Event property if you need to know the exact event on which
+// your ObservableJobCallback was invoked.
+type ObservableJobCallback func(ObservableJob)
+
 // EventType describes an event for the observer
 type EventType string
 
@@ -92,49 +99,50 @@ func GetObserver() *Observer {
 
 func (o *Observer) run() {
 	for event := range o.events {
+		o.handleEvent(event)
+	}
+}
 
-		jobName := o.getJobName(event.Job)
+func (o *Observer) handleEvent(event ObservableJob) {
+	jobName := o.getJobName(event.Job)
 
-		o.mutex.Lock()
+	o.mutex.Lock()
+	defer o.mutex.Unlock()
 
-		existingJob, exists := o.observedJobs[jobName]
+	existingJob, exists := o.observedJobs[jobName]
 
-		// we need to add the callbacks to the new event so they won't get lost
-		if exists {
-			event.callback = existingJob.callback
-		}
+	// we need to add the callbacks to the new event so they won't get lost
+	if exists {
+		event.callbacks = existingJob.callbacks
+	}
 
-		o.log.Info("new event on observed job", "event", event.Event, "jobName", jobName)
+	o.log.Info("new event on observed job", "event", event.Event, "jobName", jobName)
 
-		switch event.Event {
-		case Failed:
-			if event.callback != nil {
-				event.callback()
-			}
-			incFailureCounters(event.Job.Namespace)
-		case Suceeded:
-			// only report back succeeded jobs we've already seen. Will prevent
-			// reporting succeeded jobs on operator restart
-			if exists {
-				o.log.Info("job succeeded", "jobName", jobName)
-				o.observedJobs[jobName] = event
-				incSuccessCounters(event.Job.Namespace)
-				if event.callback != nil {
-					event.callback()
-				}
-			}
-		case Delete:
-			o.log.Info("deleting job from observer", "jobName", jobName)
-			delete(o.observedJobs, jobName)
-			if event.callback != nil {
-				event.callback()
-			}
-		default:
+	switch event.Event {
+	case Failed:
+		incFailureCounters(event.Job.Namespace)
+		invokeCallbacks(event)
+	case Suceeded:
+		// Only report succeeded jobs we've already seen to prevent
+		// reporting succeeded jobs on operator restart
+		if exists {
+			o.log.Info("job succeeded", "jobName", jobName)
 			o.observedJobs[jobName] = event
+			incSuccessCounters(event.Job.Namespace)
+			invokeCallbacks(event)
 		}
+	case Delete:
+		o.log.Info("deleting job from observer", "jobName", jobName)
+		delete(o.observedJobs, jobName)
+		invokeCallbacks(event)
+	default:
+		o.observedJobs[jobName] = event
+	}
+}
 
-		o.mutex.Unlock()
-
+func invokeCallbacks(event ObservableJob) {
+	for _, callback := range event.callbacks {
+		callback(event)
 	}
 }
 
@@ -234,14 +242,14 @@ func (o *Observer) IsConcurrentJobsLimitReached(jobType v1alpha1.JobType, limit
 }
 
 // RegisterCallback will register a function to the given observed job.
-// The callback will be executed if the job is, successful, failed or deleted.
-func (o *Observer) RegisterCallback(name string, callback func()) {
+// The callbacks will be executed if the job is successful, failed or deleted.
+func (o *Observer) RegisterCallback(name string, callback ObservableJobCallback) {
 	o.mutex.Lock()
 	defer o.mutex.Unlock()
 	if event, exists := o.observedJobs[name]; !exists {
-		o.observedJobs[name] = ObservableJob{callback: callback}
+		o.observedJobs[name] = ObservableJob{callbacks: []ObservableJobCallback{callback}}
 	} else {
-		event.callback = callback
+		event.callbacks = append(event.callbacks, callback)
 		o.observedJobs[name] = event
 	}
 }


Test patch:
diff --git a/observer/observer_test.go b/observer/observer_test.go
index a70845d6..5a96082f 100644
--- a/observer/observer_test.go
+++ b/observer/observer_test.go
@@ -6,10 +6,11 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
-	"github.com/vshn/k8up/api/v1alpha1"
 	batchv1 "k8s.io/api/batch/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	ctrl "sigs.k8s.io/controller-runtime"
+
+	"github.com/vshn/k8up/api/v1alpha1"
 )
 
 func TestObserver_IsConcurrentJobsLimitReached(t *testing.T) {
@@ -35,13 +36,11 @@ func TestObserver_IsConcurrentJobsLimitReached(t *testing.T) {
 		Event:      Running,
 		Exclusive:  true,
 		Repository: "some-repo",
-		callback:   func() {},
+		callbacks:  []ObservableJobCallback{},
 	}
 
 	ojs := make(map[string]ObservableJob)
 	ojs["some-name"] = oj
-	fmt.Println(len(ojs))
-
 	o.observedJobs = ojs
 
 	// 1 observedJob and limit = 1
@@ -56,3 +55,123 @@ func TestObserver_IsConcurrentJobsLimitReached(t *testing.T) {
 	isLimitReached = o.IsConcurrentJobsLimitReached(v1alpha1.ArchiveType, 0)
 	assert.False(t, isLimitReached)
 }
+
+func TestObserver_AreAllCallbacksInvoked(t *testing.T) {
+	for expectedInvocations := 0; expectedInvocations < 3; expectedInvocations++ {
+		t.Run(fmt.Sprintf("Given%dCallbacks_WhenHandleEvent_ThenExpect%[1]vInvocations", expectedInvocations), func(t *testing.T) {
+			// Given
+			var actualInvocations = new(int)
+			ojc := func(job ObservableJob) {
+				*actualInvocations++
+			}
+
+			callbacks := make([]ObservableJobCallback, 0)
+			for i := 0; i < expectedInvocations; i++ {
+				callbacks = append(callbacks, ojc)
+			}
+
+			oj := ObservableJob{
+				Job: &batchv1.Job{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "some-name",
+						Namespace: "default",
+					},
+				},
+				JobType:    v1alpha1.BackupType,
+				Event:      Suceeded,
+				Exclusive:  true,
+				Repository: "some-repo",
+				callbacks:  callbacks,
+			}
+
+			observedJobs := map[string]ObservableJob{
+				"default/some-name": oj,
+			}
+
+			o := &Observer{
+				events:       make(chan ObservableJob, 10),
+				observedJobs: observedJobs,
+				log:          ctrl.Log.WithName("observer-test"),
+				mutex:        sync.Mutex{},
+			}
+
+			// When
+			o.handleEvent(oj)
+
+			// Then
+			assert.Equal(t, expectedInvocations, *actualInvocations)
+		})
+	}
+}
+
+func TestObserver_AreOnlyExpectedCallbacksInvoked(t *testing.T) {
+	tests := map[string]struct {
+		givenEventType   EventType
+		expectInvocation bool
+	}{
+		"GivenStatusSucceeded_WhenHandleEvent_ThenExpectInvocation": {
+			givenEventType:   Suceeded,
+			expectInvocation: true,
+		},
+		"GivenStatusFailed_WhenHandleEvent_ThenExpectInvocation": {
+			givenEventType:   Failed,
+			expectInvocation: true,
+		},
+		"GivenStatusDelete_WhenHandleEvent_ThenExpectInvocation": {
+			givenEventType:   Delete,
+			expectInvocation: true,
+		},
+		"GivenStatusRunning_WhenHandleEvent_ThenExpectNoInvocation": {
+			givenEventType:   Running,
+			expectInvocation: false,
+		},
+		"GivenStatusCreate_WhenHandleEvent_ThenExpectNoInvocation": {
+			givenEventType:   Create,
+			expectInvocation: false,
+		},
+		"GivenStatusUpdate_WhenHandleEvent_ThenExpectNoInvocation": {
+			givenEventType:   Update,
+			expectInvocation: false,
+		},
+	}
+	for testCase, testParameter := range tests {
+		t.Run(testCase, func(t *testing.T) {
+			// Given
+			var actualInvocations = new(int)
+			ojc := func(job ObservableJob) {
+				*actualInvocations++
+			}
+
+			oj := ObservableJob{
+				Job: &batchv1.Job{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "some-name",
+						Namespace: "default",
+					},
+				},
+				JobType:    v1alpha1.BackupType,
+				Event:      testParameter.givenEventType,
+				Exclusive:  true,
+				Repository: "some-repo",
+				callbacks:  []ObservableJobCallback{ojc},
+			}
+
+			observedJobs := map[string]ObservableJob{
+				"default/some-name": oj,
+			}
+
+			o := &Observer{
+				events:       make(chan ObservableJob, 10),
+				observedJobs: observedJobs,
+				log:          ctrl.Log.WithName("observer-test"),
+				mutex:        sync.Mutex{},
+			}
+
+			// When
+			o.handleEvent(oj)
+
+			// Then
+			assert.Equal(t, testParameter.expectInvocation, *actualInvocations == 1)
+		})
+	}
+}