Add Flows and Actions

[ZITADOPER-4]
2024-06-17 15:46:05 -05:00
parent 1d5364dee5
commit aa5a411251
29 changed files with 2108 additions and 0 deletions
--- a/src/internal/controller/action_controller.go
+++ b/src/internal/controller/action_controller.go
@@ -0,0 +1,191 @@
+/*
+Copyright 2024.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package controller
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"time"
+
+	zitadelv1alpha1 "bitbucket.org/topmanage-software-engineering/zitadel-k8s-operator/src/api/v1alpha1"
+	"bitbucket.org/topmanage-software-engineering/zitadel-k8s-operator/src/pkg/builder"
+	condition "bitbucket.org/topmanage-software-engineering/zitadel-k8s-operator/src/pkg/condition"
+	"bitbucket.org/topmanage-software-engineering/zitadel-k8s-operator/src/pkg/controller/zitadel"
+	"github.com/zitadel/zitadel-go/v2/pkg/client/management"
+	"github.com/zitadel/zitadel-go/v2/pkg/client/middleware"
+	pb "github.com/zitadel/zitadel-go/v2/pkg/client/zitadel/management"
+	durationpb "google.golang.org/protobuf/types/known/durationpb"
+	"k8s.io/client-go/util/workqueue"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	ctrlClient "sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller"
+)
+
+// ActionReconciler reconciles a Action object
+type ActionReconciler struct {
+	client.Client
+	RefResolver     *zitadelv1alpha1.RefResolver
+	ConditionReady  *condition.Ready
+	RequeueInterval time.Duration
+	Builder         *builder.Builder
+}
+
+func NewActionReconciler(client client.Client, refResolver *zitadelv1alpha1.RefResolver, builder *builder.Builder, conditionReady *condition.Ready,
+	requeueInterval time.Duration) *ActionReconciler {
+	return &ActionReconciler{
+		Client:          client,
+		RefResolver:     refResolver,
+		ConditionReady:  conditionReady,
+		RequeueInterval: requeueInterval,
+		Builder:         builder,
+	}
+}
+
+//+kubebuilder:rbac:groups=zitadel.topmanage.com,resources=actions,verbs=get;list;watch;create;update;patch;delete
+//+kubebuilder:rbac:groups=zitadel.topmanage.com,resources=actions/status,verbs=get;update;patch
+//+kubebuilder:rbac:groups=zitadel.topmanage.com,resources=actions/finalizers,verbs=update
+
+// Reconcile is part of the main kubernetes reconciliation loop which aims to
+// move the current state of the cluster closer to the desired state.
+func (r *ActionReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
+	var Action zitadelv1alpha1.Action
+	if err := r.Get(ctx, req.NamespacedName, &Action); err != nil {
+		return ctrl.Result{}, client.IgnoreNotFound(err)
+	}
+	wr := newWrappedActionReconciler(r.Client, r.RefResolver, r.Builder, &Action)
+	wf := newWrappedActionFinalizer(r.Client, &Action, r.RefResolver)
+	tf := zitadel.NewZitadelFinalizer(r.Client, wf)
+	tr := zitadel.NewZitadelReconciler(r.Client, r.ConditionReady, wr, tf, r.RequeueInterval)
+
+	result, err := tr.Reconcile(ctx, &Action)
+	if err != nil {
+		return result, fmt.Errorf("error reconciling in ActionReconciler: %v", err)
+	}
+	return result, nil
+}
+
+type wrappedActionReconciler struct {
+	client.Client
+	refResolver *zitadelv1alpha1.RefResolver
+	Action      *zitadelv1alpha1.Action
+	Builder     *builder.Builder
+}
+
+func newWrappedActionReconciler(client client.Client, refResolver *zitadelv1alpha1.RefResolver, builder *builder.Builder,
+	Action *zitadelv1alpha1.Action) zitadel.WrappedReconciler {
+	return &wrappedActionReconciler{
+		Client:      client,
+		refResolver: refResolver,
+		Action:      Action,
+		Builder:     builder,
+	}
+}
+
+type actionReoncilePhase struct {
+	Name      string
+	Reconcile func(context.Context, *management.Client) error
+}
+
+func (wr *wrappedActionReconciler) Reconcile(ctx context.Context, ztdClient *management.Client) error {
+	phases := []actionReoncilePhase{
+		{
+			Name:      "action",
+			Reconcile: wr.reconcileAction,
+		},
+	}
+	for _, p := range phases {
+		err := p.Reconcile(ctx, ztdClient)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (wr *wrappedActionReconciler) reconcileAction(ctx context.Context, ztdClient *management.Client) error {
+	org, err := wr.refResolver.OrganizationRef(ctx, &wr.Action.Spec.OrganizationRef, wr.Action.Namespace)
+	if err != nil {
+		return err
+	}
+	ctx = middleware.SetOrgID(ctx, org.Status.OrgId)
+
+	if wr.Action.Status.ActionId != "" {
+		p, err := ztdClient.GetAction(ctx, &pb.GetActionRequest{Id: wr.Action.Status.ActionId})
+		if p != nil {
+			_, err := ztdClient.UpdateAction(ctx,
+				&pb.UpdateActionRequest{
+					Id:            p.Action.Id,
+					Name:          wr.Action.Name,
+					Script:        wr.Action.Spec.Script,
+					Timeout:       durationpb.New(wr.Action.Spec.Timeout.Duration),
+					AllowedToFail: wr.Action.Spec.AllowedToFail,
+				},
+			)
+
+			if err != nil {
+				if !strings.Contains(err.Error(), "No changes") {
+					return fmt.Errorf("Error updating Action: %v", err)
+				}
+			}
+			return nil
+		}
+		if err != nil {
+			if !strings.Contains(err.Error(), "not found") {
+				return fmt.Errorf("Error getting Action: %v", err)
+			}
+		}
+	}
+
+	resp, err := ztdClient.CreateAction(ctx,
+		&pb.CreateActionRequest{
+			Name:          wr.Action.Name,
+			Script:        wr.Action.Spec.Script,
+			Timeout:       durationpb.New(wr.Action.Spec.Timeout.Duration),
+			AllowedToFail: wr.Action.Spec.AllowedToFail,
+		},
+	)
+
+	if err != nil {
+		if strings.Contains(err.Error(), "AlreadyExists") {
+			return nil
+		}
+		return fmt.Errorf("error creating action in Zitadel: %v", err)
+	}
+	patch := ctrlClient.MergeFrom(wr.Action.DeepCopy())
+	wr.Action.Status.ActionId = resp.Id
+	return wr.Client.Status().Patch(ctx, wr.Action, patch)
+}
+
+func (wr *wrappedActionReconciler) PatchStatus(ctx context.Context, patcher condition.Patcher) error {
+	patch := client.MergeFrom(wr.Action.DeepCopy())
+	patcher(&wr.Action.Status)
+
+	if err := wr.Client.Status().Patch(ctx, wr.Action, patch); err != nil {
+		return fmt.Errorf("error patching Action status: %v", err)
+	}
+	return nil
+}
+
+// SetupWithManager sets up the controller with the Manager.
+func (r *ActionReconciler) SetupWithManager(mgr ctrl.Manager) error {
+	return ctrl.NewControllerManagedBy(mgr).
+		For(&zitadelv1alpha1.Action{}).
+		WithOptions(controller.Options{RateLimiter: workqueue.NewItemExponentialFailureRateLimiter(time.Millisecond*500, time.Minute*3)}).
+		Complete(r)
+}
--- a/src/internal/controller/action_controller_finalizer.go
+++ b/src/internal/controller/action_controller_finalizer.go
@@ -0,0 +1,94 @@
+package controller
+
+import (
+	"strings"
+
+	zitadelv1alpha1 "bitbucket.org/topmanage-software-engineering/zitadel-k8s-operator/src/api/v1alpha1"
+	"bitbucket.org/topmanage-software-engineering/zitadel-k8s-operator/src/pkg/controller/zitadel"
+
+	"context"
+	"fmt"
+
+	"github.com/zitadel/zitadel-go/v2/pkg/client/management"
+	"github.com/zitadel/zitadel-go/v2/pkg/client/middleware"
+	pb "github.com/zitadel/zitadel-go/v2/pkg/client/zitadel/management"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	ctrlClient "sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
+)
+
+const (
+	actionFinalizerName = "action.zitadel.topmanage.com/action"
+)
+
+type wrappedActionFinalizer struct {
+	client.Client
+	action      *zitadelv1alpha1.Action
+	refresolver *zitadelv1alpha1.RefResolver
+}
+
+func newWrappedActionFinalizer(client client.Client, action *zitadelv1alpha1.Action, refresolver *zitadelv1alpha1.RefResolver) zitadel.WrappedFinalizer {
+	return &wrappedActionFinalizer{
+		Client:      client,
+		action:      action,
+		refresolver: refresolver,
+	}
+}
+
+func (wf *wrappedActionFinalizer) AddFinalizer(ctx context.Context) error {
+	if wf.ContainsFinalizer() {
+		return nil
+	}
+	return wf.patch(ctx, wf.action, func(action *zitadelv1alpha1.Action) {
+		controllerutil.AddFinalizer(action, actionFinalizerName)
+	})
+}
+
+func (wf *wrappedActionFinalizer) RemoveFinalizer(ctx context.Context) error {
+	if !wf.ContainsFinalizer() {
+		return nil
+	}
+	return wf.patch(ctx, wf.action, func(action *zitadelv1alpha1.Action) {
+		controllerutil.RemoveFinalizer(wf.action, actionFinalizerName)
+	})
+}
+
+func (wr *wrappedActionFinalizer) ContainsFinalizer() bool {
+	return controllerutil.ContainsFinalizer(wr.action, actionFinalizerName)
+}
+
+func (wf *wrappedActionFinalizer) Reconcile(ctx context.Context, ztdClient *management.Client) error {
+	if wf.action.Status.ActionId == "" {
+		return nil
+	}
+	org, err := wf.refresolver.OrganizationRef(ctx, &wf.action.Spec.OrganizationRef, wf.action.Namespace)
+	if err != nil {
+		return err
+	}
+	{
+
+		_, err := ztdClient.GetAction(middleware.SetOrgID(ctx, org.Status.OrgId), &pb.GetActionRequest{Id: wf.action.Status.ActionId})
+		if err != nil {
+			if strings.Contains(err.Error(), `doesn't exist`) {
+				return nil
+			}
+			return err
+		}
+	}
+	_, err = ztdClient.DeleteAction(middleware.SetOrgID(ctx, org.Status.OrgId), &pb.DeleteActionRequest{Id: wf.action.Status.ActionId})
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func (wr *wrappedActionFinalizer) patch(ctx context.Context, action *zitadelv1alpha1.Action,
+	patchFn func(*zitadelv1alpha1.Action)) error {
+	patch := ctrlClient.MergeFrom(action.DeepCopy())
+	patchFn(action)
+
+	if err := wr.Client.Patch(ctx, action, patch); err != nil {
+		return fmt.Errorf("error patching Action finalizer: %v", err)
+	}
+	return nil
+}
--- a/src/internal/controller/flow_controller.go
+++ b/src/internal/controller/flow_controller.go
@@ -0,0 +1,164 @@
+/*
+Copyright 2024.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package controller
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	zitadelv1alpha1 "bitbucket.org/topmanage-software-engineering/zitadel-k8s-operator/src/api/v1alpha1"
+	"bitbucket.org/topmanage-software-engineering/zitadel-k8s-operator/src/pkg/builder"
+	condition "bitbucket.org/topmanage-software-engineering/zitadel-k8s-operator/src/pkg/condition"
+	"bitbucket.org/topmanage-software-engineering/zitadel-k8s-operator/src/pkg/controller/zitadel"
+	"github.com/zitadel/zitadel-go/v2/pkg/client/management"
+	"github.com/zitadel/zitadel-go/v2/pkg/client/middleware"
+	pb "github.com/zitadel/zitadel-go/v2/pkg/client/zitadel/management"
+	"k8s.io/client-go/util/workqueue"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller"
+)
+
+// FlowReconciler reconciles a Flow object
+type FlowReconciler struct {
+	client.Client
+	RefResolver     *zitadelv1alpha1.RefResolver
+	ConditionReady  *condition.Ready
+	RequeueInterval time.Duration
+	Builder         *builder.Builder
+}
+
+func NewFlowReconciler(client client.Client, refResolver *zitadelv1alpha1.RefResolver, builder *builder.Builder, conditionReady *condition.Ready,
+	requeueInterval time.Duration) *FlowReconciler {
+	return &FlowReconciler{
+		Client:          client,
+		RefResolver:     refResolver,
+		ConditionReady:  conditionReady,
+		RequeueInterval: requeueInterval,
+		Builder:         builder,
+	}
+}
+
+//+kubebuilder:rbac:groups=zitadel.topmanage.com,resources=flows,verbs=get;list;watch;create;update;patch;delete
+//+kubebuilder:rbac:groups=zitadel.topmanage.com,resources=flows/status,verbs=get;update;patch
+//+kubebuilder:rbac:groups=zitadel.topmanage.com,resources=flows/finalizers,verbs=update
+
+// Reconcile is part of the main kubernetes reconciliation loop which aims to
+// move the current state of the cluster closer to the desired state.
+func (r *FlowReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
+	var Flow zitadelv1alpha1.Flow
+	if err := r.Get(ctx, req.NamespacedName, &Flow); err != nil {
+		return ctrl.Result{}, client.IgnoreNotFound(err)
+	}
+	wr := newWrappedFlowReconciler(r.Client, r.RefResolver, r.Builder, &Flow)
+	wf := newWrappedFlowFinalizer(r.Client, &Flow, r.RefResolver)
+	tf := zitadel.NewZitadelFinalizer(r.Client, wf)
+	tr := zitadel.NewZitadelReconciler(r.Client, r.ConditionReady, wr, tf, r.RequeueInterval)
+
+	result, err := tr.Reconcile(ctx, &Flow)
+	if err != nil {
+		return result, fmt.Errorf("error reconciling in FlowReconciler: %v", err)
+	}
+	return result, nil
+}
+
+type wrappedFlowReconciler struct {
+	client.Client
+	refResolver *zitadelv1alpha1.RefResolver
+	Flow        *zitadelv1alpha1.Flow
+	Builder     *builder.Builder
+}
+
+func newWrappedFlowReconciler(client client.Client, refResolver *zitadelv1alpha1.RefResolver, builder *builder.Builder,
+	Flow *zitadelv1alpha1.Flow) zitadel.WrappedReconciler {
+	return &wrappedFlowReconciler{
+		Client:      client,
+		refResolver: refResolver,
+		Flow:        Flow,
+		Builder:     builder,
+	}
+}
+
+type flowReoncilePhase struct {
+	Name      string
+	Reconcile func(context.Context, *management.Client) error
+}
+
+func (wr *wrappedFlowReconciler) Reconcile(ctx context.Context, ztdClient *management.Client) error {
+	phases := []flowReoncilePhase{
+		{
+			Name:      "flow",
+			Reconcile: wr.reconcileFlow,
+		},
+	}
+	for _, p := range phases {
+		err := p.Reconcile(ctx, ztdClient)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (wr *wrappedFlowReconciler) reconcileFlow(ctx context.Context, ztdClient *management.Client) error {
+	org, err := wr.refResolver.OrganizationRef(ctx, &wr.Flow.Spec.OrganizationRef, wr.Flow.Namespace)
+	if err != nil {
+		return err
+	}
+	ctx = middleware.SetOrgID(ctx, org.Status.OrgId)
+
+	actionIds := []string{}
+	for _, actionRef := range wr.Flow.Spec.ActionRefs {
+		action, err := wr.refResolver.ActionRef(ctx, &actionRef, wr.Flow.Namespace)
+		if err != nil {
+			return fmt.Errorf("Error resolving action reference: %v", err)
+		}
+		if action.Status.ActionId == "" {
+			return fmt.Errorf("Action with name: %s not ready for trigger", action.Name)
+		}
+		actionIds = append(actionIds, action.Status.ActionId)
+	}
+
+	_, err = ztdClient.SetTriggerActions(ctx, &pb.SetTriggerActionsRequest{
+		FlowType:    wr.Flow.Spec.FlowType,
+		TriggerType: wr.Flow.Spec.TriggerType,
+		ActionIds:   actionIds,
+	})
+	if err != nil {
+		return fmt.Errorf("Error triggering action flow: %v", err)
+	}
+	return nil
+}
+
+func (wr *wrappedFlowReconciler) PatchStatus(ctx context.Context, patcher condition.Patcher) error {
+	patch := client.MergeFrom(wr.Flow.DeepCopy())
+	patcher(&wr.Flow.Status)
+
+	if err := wr.Client.Status().Patch(ctx, wr.Flow, patch); err != nil {
+		return fmt.Errorf("error patching Flow status: %v", err)
+	}
+	return nil
+}
+
+// SetupWithManager sets up the controller with the Manager.
+func (r *FlowReconciler) SetupWithManager(mgr ctrl.Manager) error {
+	return ctrl.NewControllerManagedBy(mgr).
+		For(&zitadelv1alpha1.Flow{}).
+		WithOptions(controller.Options{RateLimiter: workqueue.NewItemExponentialFailureRateLimiter(time.Millisecond*500, time.Minute*3)}).
+		Complete(r)
+}
--- a/src/internal/controller/flow_controller_finalizer.go
+++ b/src/internal/controller/flow_controller_finalizer.go
@@ -0,0 +1,82 @@
+package controller
+
+import (
+	zitadelv1alpha1 "bitbucket.org/topmanage-software-engineering/zitadel-k8s-operator/src/api/v1alpha1"
+	"bitbucket.org/topmanage-software-engineering/zitadel-k8s-operator/src/pkg/controller/zitadel"
+
+	"context"
+	"fmt"
+
+	"github.com/zitadel/zitadel-go/v2/pkg/client/management"
+	"github.com/zitadel/zitadel-go/v2/pkg/client/middleware"
+	pb "github.com/zitadel/zitadel-go/v2/pkg/client/zitadel/management"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	ctrlClient "sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
+)
+
+const (
+	flowFinalizerName = "flow.zitadel.topmanage.com/flow"
+)
+
+type wrappedFlowFinalizer struct {
+	client.Client
+	flow        *zitadelv1alpha1.Flow
+	refresolver *zitadelv1alpha1.RefResolver
+}
+
+func newWrappedFlowFinalizer(client client.Client, flow *zitadelv1alpha1.Flow, refresolver *zitadelv1alpha1.RefResolver) zitadel.WrappedFinalizer {
+	return &wrappedFlowFinalizer{
+		Client:      client,
+		flow:        flow,
+		refresolver: refresolver,
+	}
+}
+
+func (wf *wrappedFlowFinalizer) AddFinalizer(ctx context.Context) error {
+	if wf.ContainsFinalizer() {
+		return nil
+	}
+	return wf.patch(ctx, wf.flow, func(flow *zitadelv1alpha1.Flow) {
+		controllerutil.AddFinalizer(flow, flowFinalizerName)
+	})
+}
+
+func (wf *wrappedFlowFinalizer) RemoveFinalizer(ctx context.Context) error {
+	if !wf.ContainsFinalizer() {
+		return nil
+	}
+	return wf.patch(ctx, wf.flow, func(flow *zitadelv1alpha1.Flow) {
+		controllerutil.RemoveFinalizer(wf.flow, flowFinalizerName)
+	})
+}
+
+func (wr *wrappedFlowFinalizer) ContainsFinalizer() bool {
+	return controllerutil.ContainsFinalizer(wr.flow, flowFinalizerName)
+}
+
+func (wf *wrappedFlowFinalizer) Reconcile(ctx context.Context, ztdClient *management.Client) error {
+	org, err := wf.refresolver.OrganizationRef(ctx, &wf.flow.Spec.OrganizationRef, wf.flow.Namespace)
+	if err != nil {
+		return err
+	}
+
+	_, err = ztdClient.ClearFlow(middleware.SetOrgID(ctx, org.Status.OrgId), &pb.ClearFlowRequest{
+		Type: wf.flow.Spec.FlowType,
+	})
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func (wr *wrappedFlowFinalizer) patch(ctx context.Context, flow *zitadelv1alpha1.Flow,
+	patchFn func(*zitadelv1alpha1.Flow)) error {
+	patch := ctrlClient.MergeFrom(flow.DeepCopy())
+	patchFn(flow)
+
+	if err := wr.Client.Patch(ctx, flow, patch); err != nil {
+		return fmt.Errorf("error patching Flow finalizer: %v", err)
+	}
+	return nil
+}