chore(operator): Support Progressing state in every phase + refactori…

…ng + speed improvements (#236)

Signed-off-by: odubajDT <ondrej.dubaj@dynatrace.com>

operator/api/v1alpha1/common/common.go

@@ -51,6 +51,10 @@ func (k KeptnState) IsFailed() bool {
 	return k == StateFailed
 }
 
+func (k KeptnState) IsPending() bool {
+	return k == StatePending
+}
+
 type StatusSummary struct {
 	Total       int
 	progressing int

operator/api/v1alpha1/keptnappversion_types.go

@@ -17,6 +17,7 @@ limitations under the License.
 package v1alpha1
 
 import (
+	"fmt"
 	"time"
 
 	"github.com/keptn/lifecycle-controller/operator/api/v1alpha1/common"
@@ -205,3 +206,31 @@ func (v KeptnAppVersion) GetDurationMetricsAttributes() []attribute.KeyValue {
 		common.AppPreviousVersion.String(v.Spec.PreviousVersion),
 	}
 }
+
+func (v KeptnAppVersion) GetState() common.KeptnState {
+	return v.Status.Status
+}
+
+func (v *KeptnAppVersion) SetState(state common.KeptnState) {
+	v.Status.Status = state
+}
+
+func (v KeptnAppVersion) GetCurrentPhase() string {
+	return v.Status.CurrentPhase
+}
+
+func (v *KeptnAppVersion) SetCurrentPhase(phase string) {
+	v.Status.CurrentPhase = phase
+}
+
+func (v *KeptnAppVersion) Complete() {
+	v.SetEndTime()
+}
+
+func (v KeptnAppVersion) GetVersion() string {
+	return v.Spec.Version
+}
+
+func (v KeptnAppVersion) GetSpanName(phase string) string {
+	return fmt.Sprintf("%s.%s.%s.%s", v.Spec.TraceId, v.Spec.AppName, v.Spec.Version, phase)
+}

operator/api/v1alpha1/keptnworkloadinstance_types.go

@@ -17,6 +17,7 @@ limitations under the License.
 package v1alpha1
 
 import (
+	"fmt"
 	"time"
 
 	"github.com/keptn/lifecycle-controller/operator/api/v1alpha1/common"
@@ -242,3 +243,31 @@ func (i KeptnWorkloadInstance) GetIntervalMetricsAttributes() []attribute.KeyVal
 		common.WorkloadPreviousVersion.String(i.Spec.PreviousVersion),
 	}
 }
+
+func (i KeptnWorkloadInstance) GetState() common.KeptnState {
+	return i.Status.Status
+}
+
+func (i *KeptnWorkloadInstance) SetState(state common.KeptnState) {
+	i.Status.Status = state
+}
+
+func (i KeptnWorkloadInstance) GetCurrentPhase() string {
+	return i.Status.CurrentPhase
+}
+
+func (i *KeptnWorkloadInstance) SetCurrentPhase(phase string) {
+	i.Status.CurrentPhase = phase
+}
+
+func (i *KeptnWorkloadInstance) Complete() {
+	i.SetEndTime()
+}
+
+func (i KeptnWorkloadInstance) GetVersion() string {
+	return i.Spec.Version
+}
+
+func (v KeptnWorkloadInstance) GetSpanName(phase string) string {
+	return fmt.Sprintf("%s.%s.%s.%s", v.Spec.TraceId, v.Spec.AppName, v.Spec.Version, phase)
+}

operator/controllers/common/helperfunctions.go

@@ -0,0 +1,37 @@
+package common
+
+import (
+	klcv1alpha1 "github.com/keptn/lifecycle-controller/operator/api/v1alpha1"
+	apicommon "github.com/keptn/lifecycle-controller/operator/api/v1alpha1/common"
+	"k8s.io/apimachinery/pkg/types"
+)
+
+func GetTaskStatus(taskName string, instanceStatus []klcv1alpha1.TaskStatus) klcv1alpha1.TaskStatus {
+	for _, status := range instanceStatus {
+		if status.TaskDefinitionName == taskName {
+			return status
+		}
+	}
+	return klcv1alpha1.TaskStatus{
+		TaskDefinitionName: taskName,
+		Status:             apicommon.StatePending,
+		TaskName:           "",
+	}
+}
+
+func GetEvaluationStatus(evaluationName string, instanceStatus []klcv1alpha1.EvaluationStatus) klcv1alpha1.EvaluationStatus {
+	for _, status := range instanceStatus {
+		if status.EvaluationDefinitionName == evaluationName {
+			return status
+		}
+	}
+	return klcv1alpha1.EvaluationStatus{
+		EvaluationDefinitionName: evaluationName,
+		Status:                   apicommon.StatePending,
+		EvaluationName:           "",
+	}
+}
+
+func GetAppVersionName(namespace string, appName string, version string) types.NamespacedName {
+	return types.NamespacedName{Namespace: namespace, Name: appName + "-" + version}
+}

operator/controllers/common/interfaces.go

@@ -0,0 +1,65 @@
+package common
+
+import (
+	"errors"
+
+	"github.com/keptn/lifecycle-controller/operator/api/v1alpha1/common"
+	"go.opentelemetry.io/otel/attribute"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+//go:generate moq -pkg common_mock --skip-ensure -out ./fake/phaseitem_mock.go . PhaseItem
+type PhaseItem interface {
+	GetState() common.KeptnState
+	SetState(common.KeptnState)
+	GetCurrentPhase() string
+	SetCurrentPhase(string)
+	GetVersion() string
+	GetMetricsAttributes() []attribute.KeyValue
+	GetSpanName(phase string) string
+	Complete()
+}
+
+type PhaseItemWrapper struct {
+	Obj PhaseItem
+}
+
+func NewPhaseItemWrapperFromClientObject(object client.Object) (*PhaseItemWrapper, error) {
+	pi, ok := object.(PhaseItem)
+	if !ok {
+		return nil, errors.New("provided object does not implement PhaseItem interface")
+	}
+	return &PhaseItemWrapper{Obj: pi}, nil
+}
+
+func (pw PhaseItemWrapper) GetState() common.KeptnState {
+	return pw.Obj.GetState()
+}
+
+func (pw *PhaseItemWrapper) SetState(state common.KeptnState) {
+	pw.Obj.SetState(state)
+}
+
+func (pw PhaseItemWrapper) GetCurrentPhase() string {
+	return pw.Obj.GetCurrentPhase()
+}
+
+func (pw *PhaseItemWrapper) SetCurrentPhase(phase string) {
+	pw.Obj.SetCurrentPhase(phase)
+}
+
+func (pw PhaseItemWrapper) GetMetricsAttributes() []attribute.KeyValue {
+	return pw.Obj.GetMetricsAttributes()
+}
+
+func (pw *PhaseItemWrapper) Complete() {
+	pw.Obj.Complete()
+}
+
+func (pw PhaseItemWrapper) GetVersion() string {
+	return pw.Obj.GetVersion()
+}
+
+func (pw PhaseItemWrapper) GetSpanName(phase string) string {
+	return pw.Obj.GetSpanName(phase)
+}

operator/controllers/common/interfaces_test.go

@@ -0,0 +1,26 @@
+package common
+
+import (
+	"github.com/keptn/lifecycle-controller/operator/api/v1alpha1"
+	"github.com/keptn/lifecycle-controller/operator/api/v1alpha1/common"
+	"github.com/stretchr/testify/require"
+	"testing"
+)
+
+func TestPhaseItemWrapper_GetState(t *testing.T) {
+	appVersion := &v1alpha1.KeptnAppVersion{
+		Status: v1alpha1.KeptnAppVersionStatus{
+			Status:       common.StateFailed,
+			CurrentPhase: "test",
+		},
+	}
+
+	object, err := NewPhaseItemWrapperFromClientObject(appVersion)
+	require.Nil(t, err)
+
+	require.Equal(t, "test", object.GetCurrentPhase())
+
+	object.Complete()
+
+	require.NotZero(t, appVersion.Status.EndTime)
+}

operator/controllers/common/phasehandler.go

@@ -0,0 +1,104 @@
+package common
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/go-logr/logr"
+	"github.com/keptn/lifecycle-controller/operator/api/v1alpha1/common"
+	"go.opentelemetry.io/otel/codes"
+	"go.opentelemetry.io/otel/trace"
+	"k8s.io/client-go/tools/record"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+type PhaseHandler struct {
+	client.Client
+	Recorder    record.EventRecorder
+	Log         logr.Logger
+	SpanHandler SpanHandler
+}
+
+type PhaseResult struct {
+	Continue bool
+	ctrl.Result
+}
+
+func RecordEvent(recorder record.EventRecorder, phase common.KeptnPhaseType, eventType string, reconcileObject client.Object, shortReason string, longReason string, version string) {
+	recorder.Event(reconcileObject, eventType, fmt.Sprintf("%s%s", phase.ShortName, shortReason), fmt.Sprintf("%s %s / Namespace: %s, Name: %s, Version: %s ", phase.LongName, longReason, reconcileObject.GetNamespace(), reconcileObject.GetName(), version))
+}
+
+func (r PhaseHandler) HandlePhase(ctx context.Context, ctxAppTrace context.Context, tracer trace.Tracer, reconcileObject client.Object, phase common.KeptnPhaseType, span trace.Span, reconcilePhase func() (common.KeptnState, error)) (*PhaseResult, error) {
+	requeueResult := ctrl.Result{Requeue: true, RequeueAfter: 5 * time.Second}
+	piWrapper, err := NewPhaseItemWrapperFromClientObject(reconcileObject)
+	if err != nil {
+		return &PhaseResult{Continue: false, Result: ctrl.Result{Requeue: true}}, err
+	}
+	oldStatus := piWrapper.GetState()
+	oldPhase := piWrapper.GetCurrentPhase()
+	piWrapper.SetCurrentPhase(phase.ShortName)
+
+	r.Log.Info(phase.LongName + " not finished")
+	ctxAppTrace, spanAppTrace, err := r.SpanHandler.GetSpan(ctxAppTrace, tracer, reconcileObject, phase.ShortName)
+	if err != nil {
+		r.Log.Error(err, "could not get span")
+	}
+
+	state, err := reconcilePhase()
+	if err != nil {
+		spanAppTrace.AddEvent(phase.LongName + " could not get reconciled")
+		RecordEvent(r.Recorder, phase, "Warning", reconcileObject, "ReconcileErrored", "could not get reconciled", piWrapper.GetVersion())
+		span.SetStatus(codes.Error, err.Error())
+		return &PhaseResult{Continue: false, Result: requeueResult}, err
+	}
+
+	if state.IsPending() {
+		state = common.StateProgressing
+	}
+
+	defer func(oldStatus common.KeptnState, oldPhase string, reconcileObject client.Object) {
+		piWrapper, _ := NewPhaseItemWrapperFromClientObject(reconcileObject)
+		if oldStatus != piWrapper.GetState() || oldPhase != piWrapper.GetCurrentPhase() {
+			ctx, spanAppTrace, err = r.SpanHandler.GetSpan(ctxAppTrace, tracer, reconcileObject, piWrapper.GetCurrentPhase())
+			if err != nil {
+				r.Log.Error(err, "could not get span")
+			}
+			if err := r.Status().Update(ctx, reconcileObject); err != nil {
+				r.Log.Error(err, "could not update status")
+			}
+		}
+	}(oldStatus, oldPhase, reconcileObject)
+
+	if state.IsCompleted() {
+		if state.IsFailed() {
+			piWrapper.Complete()
+			piWrapper.SetState(common.StateFailed)
+			spanAppTrace.AddEvent(phase.LongName + " has failed")
+			spanAppTrace.SetStatus(codes.Error, "Failed")
+			spanAppTrace.End()
+			if err := r.SpanHandler.UnbindSpan(reconcileObject, phase.ShortName); err != nil {
+				r.Log.Error(err, "cannot unbind span")
+			}
+			RecordEvent(r.Recorder, phase, "Warning", reconcileObject, "Failed", "has failed", piWrapper.GetVersion())
+			return &PhaseResult{Continue: false, Result: ctrl.Result{}}, nil
+		}
+
+		piWrapper.SetState(common.StateSucceeded)
+		spanAppTrace.AddEvent(phase.LongName + " has succeeded")
+		spanAppTrace.SetStatus(codes.Ok, "Succeeded")
+		spanAppTrace.End()
+		if err := r.SpanHandler.UnbindSpan(reconcileObject, phase.ShortName); err != nil {
+			r.Log.Error(err, "cannot unbind span")
+		}
+		RecordEvent(r.Recorder, phase, "Normal", reconcileObject, "Succeeded", "has succeeded", piWrapper.GetVersion())
+
+		return &PhaseResult{Continue: true, Result: requeueResult}, nil
+	}
+
+	piWrapper.SetState(common.StateProgressing)
+	RecordEvent(r.Recorder, phase, "Warning", reconcileObject, "NotFinished", "has not finished", piWrapper.GetVersion())
+
+	return &PhaseResult{Continue: false, Result: requeueResult}, nil
+}

operator/controllers/common/spanhandler.go

@@ -0,0 +1,38 @@
+package common
+
+import (
+	"context"
+
+	"go.opentelemetry.io/otel/trace"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+type SpanHandler struct {
+	bindCRDSpan map[string]trace.Span
+}
+
+func (r SpanHandler) GetSpan(ctx context.Context, tracer trace.Tracer, reconcileObject client.Object, phase string) (context.Context, trace.Span, error) {
+	piWrapper, err := NewPhaseItemWrapperFromClientObject(reconcileObject)
+	if err != nil {
+		return nil, nil, err
+	}
+	appvName := piWrapper.GetSpanName(phase)
+	if r.bindCRDSpan == nil {
+		r.bindCRDSpan = make(map[string]trace.Span)
+	}
+	if span, ok := r.bindCRDSpan[appvName]; ok {
+		return ctx, span, nil
+	}
+	ctx, span := tracer.Start(ctx, phase, trace.WithSpanKind(trace.SpanKindConsumer))
+	r.bindCRDSpan[appvName] = span
+	return ctx, span, nil
+}
+
+func (r SpanHandler) UnbindSpan(reconcileObject client.Object, phase string) error {
+	piWrapper, err := NewPhaseItemWrapperFromClientObject(reconcileObject)
+	if err != nil {
+		return err
+	}
+	delete(r.bindCRDSpan, piWrapper.GetSpanName(phase))
+	return nil
+}

operator/controllers/keptnapp/controller.go

@@ -19,6 +19,7 @@ package keptnapp
 import (
 	"context"
 	"fmt"
+
 	"sigs.k8s.io/controller-runtime/pkg/builder"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"