Add stream class events (#211)

* Add stream class events

* Fix unit tests

* Fix main.py

* More fix

* Add more unit tests

Author: s-vitaliy

main.go

@@ -68,7 +68,7 @@ func main() {
 	probesService := health.NewProbesService(appConfig.ProbesConfiguration)
 	go func() {
 		err := probesService.ListenAndServe(ctx)
-		if err != nil {
+		if err != nil && !errors.Is(err, context.Canceled) {
 			setupLog.V(0).Error(err, "unable to start health probes server")
 			panic(err)
 		}
@@ -103,7 +103,7 @@ func main() {
 		mgr,
 		eventRecorder,
 	)
-	err = stream_class.NewStreamClassReconciler(mgr.GetClient(), controllerFactory, reporter).SetupWithManager(mgr)
+	err = stream_class.NewStreamClassReconciler(mgr.GetClient(), controllerFactory, reporter, eventRecorder).SetupWithManager(mgr)
 
 	if err != nil {
 		setupLog.V(0).Error(err, "unable to create controller", "controller", "StreamClass")

services/controllers/event_func.go

@@ -0,0 +1,5 @@
+package controllers
+
+// EventFunc is a function type for event handling.
+// Used to abstract event handling logic and simplify function signatures.
+type EventFunc func()

services/controllers/stream/stream_reconciler.go

@@ -297,7 +297,7 @@ func (s *streamReconciler) moveFsm(ctx context.Context, definition Definition, j
 	)
 }
 
-func (s *streamReconciler) stopStream(ctx context.Context, definition Definition, nextPhase Phase, eventFunc func()) (reconcile.Result, error) {
+func (s *streamReconciler) stopStream(ctx context.Context, definition Definition, nextPhase Phase, eventFunc controllers.EventFunc) (reconcile.Result, error) {
 	j := &batchv1.Job{}
 	j.SetName(definition.NamespacedName().Name)
 	j.SetNamespace(definition.NamespacedName().Namespace)
@@ -309,7 +309,7 @@ func (s *streamReconciler) stopStream(ctx context.Context, definition Definition
 	return s.updateStreamPhase(ctx, definition, nil, nextPhase, eventFunc)
 }
 
-func (s *streamReconciler) startBackfill(ctx context.Context, definition Definition, nextPhase Phase, eventFunc func()) (reconcile.Result, error) {
+func (s *streamReconciler) startBackfill(ctx context.Context, definition Definition, nextPhase Phase, eventFunc controllers.EventFunc) (reconcile.Result, error) {
 
 	logger := s.getLogger(ctx, definition.NamespacedName())
 	logger.V(2).Info("starting backfill by creating a backfill request")
@@ -334,7 +334,7 @@ func (s *streamReconciler) startBackfill(ctx context.Context, definition Definit
 	return s.updateStreamPhase(ctx, definition, backfillRequest, nextPhase, eventFunc)
 }
 
-func (s *streamReconciler) reconcileJob(ctx context.Context, definition Definition, backfillRequest *v1.BackfillRequest, nextPhase Phase, eventFunc func()) (reconcile.Result, error) {
+func (s *streamReconciler) reconcileJob(ctx context.Context, definition Definition, backfillRequest *v1.BackfillRequest, nextPhase Phase, eventFunc controllers.EventFunc) (reconcile.Result, error) {
 	logger := s.getLogger(ctx, definition.NamespacedName())
 	v1job := batchv1.Job{}
 	err := s.client.Get(ctx, definition.NamespacedName(), &v1job)
@@ -401,7 +401,7 @@ func (s *streamReconciler) compareConfigurations(ctx context.Context, v1job batc
 	return jobConfiguration == definitionConfiguration, nil
 }
 
-func (s *streamReconciler) completeBackfill(ctx context.Context, job *StreamingJob, definition Definition, request *v1.BackfillRequest, nextStatus Phase, eventFunc func()) (reconcile.Result, error) {
+func (s *streamReconciler) completeBackfill(ctx context.Context, job *StreamingJob, definition Definition, request *v1.BackfillRequest, nextStatus Phase, eventFunc controllers.EventFunc) (reconcile.Result, error) {
 	if job != nil {
 		err := s.client.Delete(ctx, job.ToV1Job())
 		if client.IgnoreNotFound(err) != nil {
@@ -494,7 +494,7 @@ func (s *streamReconciler) getLogger(_ context.Context, request types.Namespaced
 		WithValues("namespace", request.Namespace, "name", request.Name, "kind", s.gvk.Kind)
 }
 
-func (s *streamReconciler) updateStreamPhase(ctx context.Context, definition Definition, backfillRequest *v1.BackfillRequest, next Phase, eventFunc func()) (reconcile.Result, error) {
+func (s *streamReconciler) updateStreamPhase(ctx context.Context, definition Definition, backfillRequest *v1.BackfillRequest, next Phase, eventFunc controllers.EventFunc) (reconcile.Result, error) {
 	logger := s.getLogger(ctx, definition.NamespacedName())
 	if definition.GetPhase() == next { // coverage-ignore
 		logger.V(0).Info("Stream phase is already set to", "phase", definition.GetPhase())

services/controllers/stream_class/stream_class.go

@@ -5,8 +5,11 @@ import (
 	"errors"
 	"fmt"
 	v1 "github.com/SneaksAndData/arcane-operator/pkg/apis/streaming/v1"
+	"github.com/SneaksAndData/arcane-operator/services/controllers"
+	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/tools/record"
 	"k8s.io/klog/v2"
 	runtime "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -22,14 +25,16 @@ type StreamClassReconciler struct {
 	streamControllers       map[types.NamespacedName]*StreamControllerHandle
 	streamControllerFactory UnmanagedControllerFactory
 	reporter                StreamClassMetricsReporter
+	eventRecorder           record.EventRecorder
 }
 
-func NewStreamClassReconciler(client client.Client, streamControllerFactory UnmanagedControllerFactory, reporter StreamClassMetricsReporter) *StreamClassReconciler {
+func NewStreamClassReconciler(client client.Client, streamControllerFactory UnmanagedControllerFactory, reporter StreamClassMetricsReporter, eventRecorder record.EventRecorder) *StreamClassReconciler {
 	return &StreamClassReconciler{
 		client:                  client,
 		streamControllers:       make(map[types.NamespacedName]*StreamControllerHandle),
 		streamControllerFactory: streamControllerFactory,
 		reporter:                reporter,
+		eventRecorder:           eventRecorder,
 	}
 }
 
@@ -40,14 +45,14 @@ func (s *StreamClassReconciler) Reconcile(ctx context.Context, request reconcile
 	sc := &v1.StreamClass{}
 	err := s.client.Get(ctx, request.NamespacedName, sc)
 	deleted := apierrors.IsNotFound(err)
-	if client.IgnoreNotFound(err) != nil {
+	if client.IgnoreNotFound(err) != nil { // coverage-ignore
 		logger.V(0).Error(err, "unable to get stream class")
 	}
 
 	return s.moveFsm(ctx, sc, deleted, request.NamespacedName)
 }
 
-func (s *StreamClassReconciler) SetupWithManager(mgr runtime.Manager) error {
+func (s *StreamClassReconciler) SetupWithManager(mgr runtime.Manager) error { // coverage-ignore (should be tested in e2e)
 	return runtime.NewControllerManagedBy(mgr).For(&v1.StreamClass{}).Complete(s)
 }
 
@@ -58,13 +63,41 @@ func (s *StreamClassReconciler) getLogger(ctx context.Context, request types.Nam
 }
 
 func (s *StreamClassReconciler) moveFsm(ctx context.Context, sc *v1.StreamClass, deleted bool, name types.NamespacedName) (reconcile.Result, error) {
+	logger := s.getLogger(ctx, name)
 	switch {
-	case !deleted && sc.Status.Phase == "":
-		return s.updatePhase(ctx, sc, name, v1.PhasePending)
-	case !deleted && (sc.Status.Phase == v1.PhasePending || sc.Status.Phase == v1.PhaseReady):
-		return s.tryStartStreamController(ctx, sc, name, v1.PhaseReady)
 	case deleted:
-		return s.tryStopStreamController(ctx, name)
+		return s.tryStopStreamController(ctx, name, nil)
+
+	case sc.Status.Phase == "":
+		return s.updatePhase(ctx, sc, name, v1.PhasePending, func() {
+			s.eventRecorder.Event(sc,
+				corev1.EventTypeNormal,
+				"StreamClassCreated",
+				"New StreamClass has been created and is pending processing")
+		})
+	case sc.Status.Phase == v1.PhasePending:
+		return s.tryStartStreamController(ctx, sc, name, v1.PhaseReady, func() {
+			s.eventRecorder.Event(sc,
+				corev1.EventTypeNormal,
+				"StreamClassReady",
+				"StreamClass is ready and stream controller has been started")
+		})
+	case sc.Status.Phase == v1.PhaseFailed:
+		logger.V(0).Info("Found StreamClass in Failed state, attempting to recover")
+		return s.tryStartStreamController(ctx, sc, name, v1.PhaseReady, func() {
+			s.eventRecorder.Event(sc,
+				corev1.EventTypeWarning,
+				"StreamClassRecovered",
+				"StreamClass was found in Failed state and recovery was attempted")
+		})
+
+	case sc.Status.Phase == v1.PhaseReady:
+		return s.tryStartStreamController(ctx, sc, name, v1.PhaseReady, func() {
+			s.eventRecorder.Event(sc,
+				corev1.EventTypeNormal,
+				"StreamClassReconciled",
+				"StreamClass is reconciled and stream controller is running")
+		})
 	}
 
 	return reconcile.Result{}, fmt.Errorf("failed to reconcile StreamClass FSM for %s. Current state: %s",
@@ -73,7 +106,7 @@ func (s *StreamClassReconciler) moveFsm(ctx context.Context, sc *v1.StreamClass,
 	)
 }
 
-func (s *StreamClassReconciler) tryStartStreamController(ctx context.Context, sc *v1.StreamClass, name types.NamespacedName, nextPhase v1.Phase) (reconcile.Result, error) {
+func (s *StreamClassReconciler) tryStartStreamController(ctx context.Context, sc *v1.StreamClass, name types.NamespacedName, nextPhase v1.Phase, eventFunc controllers.EventFunc) (reconcile.Result, error) {
 	s.rwLock.Lock()
 	defer s.rwLock.Unlock()
 
@@ -82,14 +115,14 @@ func (s *StreamClassReconciler) tryStartStreamController(ctx context.Context, sc
 	_, ok := s.streamControllers[name]
 	if ok {
 		logger.V(0).Info("Stream controller is already running")
-		return s.updatePhase(ctx, sc, name, nextPhase)
+		return s.updatePhase(ctx, sc, name, nextPhase, eventFunc)
 	}
 
 	controller, err := s.streamControllerFactory.CreateStreamController(ctx, sc.TargetResourceGvk(), sc)
 
 	if err != nil {
 		logger.V(0).Error(err, "unable to create stream reconciler")
-		return s.updatePhase(ctx, sc, name, v1.PhaseFailed)
+		return s.updatePhase(ctx, sc, name, v1.PhaseFailed, eventFunc)
 	}
 
 	controllerContext, cancelFunc := context.WithCancel(ctx)
@@ -103,60 +136,71 @@ func (s *StreamClassReconciler) tryStartStreamController(ctx context.Context, sc
 			logger := s.getLogger(ctx, name)
 			logger.V(0).Error(err, "stream controller exited with error")
 
-			_, err = s.updatePhase(ctx, sc, name, v1.PhaseFailed)
+			_, err = s.updatePhase(ctx, sc, name, v1.PhaseFailed, func() {
+				s.eventRecorder.Event(sc,
+					corev1.EventTypeWarning,
+					"StreamControllerError",
+					"Stream controller exited with error, StreamClass moved to Failed state")
+			})
 			if err != nil {
 				logger := s.getLogger(ctx, name)
 				logger.V(0).Error(err, "unable to update StreamClass phase to Failed after stream controller exited with error")
 			}
 		}
 	}()
 
-	logger.V(1).Info("Stream controller is started")
+	logger.V(0).Info("Stream controller is started")
 	s.streamControllers[name] = &StreamControllerHandle{
 		cancelFunc: cancelFunc,
 		gvk:        sc.TargetResourceGvk(),
 	}
 	s.reporter.AddStreamClass(sc.TargetResourceGvk().Kind, "stream_class", sc.MetricsTags())
 
-	return s.updatePhase(ctx, sc, name, nextPhase)
+	return s.updatePhase(ctx, sc, name, nextPhase, eventFunc)
 }
 
-func (s *StreamClassReconciler) tryStopStreamController(ctx context.Context, name types.NamespacedName) (reconcile.Result, error) {
+func (s *StreamClassReconciler) tryStopStreamController(ctx context.Context, name types.NamespacedName, eventFunc controllers.EventFunc) (reconcile.Result, error) {
 	s.rwLock.Lock()
 	defer s.rwLock.Unlock()
 
 	logger := s.getLogger(ctx, name)
 	_, ok := s.streamControllers[name]
 	if !ok {
-		logger.V(2).Info("Stream controller is not running")
+		logger.V(0).Info("Stream controller is not running")
 		return reconcile.Result{}, nil
 	}
 	s.streamControllers[name].cancelFunc()
-	logger.V(2).Info("Stream controller is stopped")
+	logger.V(0).Info("Stream controller is stopped")
 	s.reporter.RemoveStreamClass(s.streamControllers[name].gvk.Kind)
 
 	delete(s.streamControllers, name)
-	return s.updatePhase(ctx, nil, name, v1.PhaseStopped)
+	return s.updatePhase(ctx, nil, name, v1.PhaseStopped, eventFunc)
 }
 
-func (s *StreamClassReconciler) updatePhase(ctx context.Context, sc *v1.StreamClass, name types.NamespacedName, nextPhase v1.Phase) (reconcile.Result, error) {
+func (s *StreamClassReconciler) updatePhase(ctx context.Context, sc *v1.StreamClass, name types.NamespacedName, nextPhase v1.Phase, eventFunc controllers.EventFunc) (reconcile.Result, error) {
 	logger := s.getLogger(ctx, name)
 	if sc == nil {
 		logger.V(0).Info("Stream class is deleted, skipping phase update")
 		return reconcile.Result{}, nil
 	}
 
 	if sc.Status.Phase == nextPhase {
-		logger.V(2).Info("StreamClass phase is already set to", sc.Status.Phase)
+		logger.V(0).Info("StreamClass phase is already set to", sc.Status.Phase)
 		return reconcile.Result{}, nil
 	}
 
-	logger.V(1).Info("Updating StreamClass phase", "from", sc.Status.Phase, "to", nextPhase)
+	logger.V(0).Info("Updating StreamClass phase", "from", sc.Status.Phase, "to", nextPhase)
 	sc.Status.Phase = nextPhase
 	err := s.client.Status().Update(ctx, sc)
+
+	if eventFunc != nil {
+		eventFunc()
+	}
+
 	if client.IgnoreNotFound(err) != nil {
-		logger.V(1).Error(err, "unable to update Stream Class status")
+		logger.V(0).Error(err, "unable to update Stream Class status")
 		return reconcile.Result{}, err
 	}
+
 	return reconcile.Result{}, nil
 }