Add a controller that will watch events and Fire Metrics (#91)

Author: engedaam

events/controller.go

@@ -0,0 +1,65 @@
+package events
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"time"
+
+	pmetrics "github.com/awslabs/operatorpkg/metrics"
+	"github.com/awslabs/operatorpkg/object"
+	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/utils/clock"
+	controllerruntime "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/builder"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
+	"sigs.k8s.io/controller-runtime/pkg/predicate"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+)
+
+type Controller[T client.Object] struct {
+	gvk        schema.GroupVersionKind
+	startTime  time.Time
+	kubeClient client.Client
+	EventCount pmetrics.CounterMetric
+}
+
+func NewController[T client.Object](client client.Client, clock clock.Clock) *Controller[T] {
+	gvk := object.GVK(object.New[T]())
+	return &Controller[T]{
+		gvk:        gvk,
+		startTime:  clock.Now(),
+		kubeClient: client,
+		EventCount: eventTotalMetric(strings.ToLower(gvk.Kind)),
+	}
+}
+
+func (c *Controller[T]) Register(_ context.Context, m manager.Manager) error {
+	return controllerruntime.NewControllerManagedBy(m).
+		For(&v1.Event{}, builder.WithPredicates(predicate.NewTypedPredicateFuncs(func(o client.Object) bool {
+			// Only reconcile on the object kind we care about
+			event := o.(*v1.Event)
+			return event.InvolvedObject.Kind == c.gvk.Kind && event.InvolvedObject.APIVersion == c.gvk.GroupVersion().String()
+		}))).
+		WithOptions(controller.Options{MaxConcurrentReconciles: 10}).
+		Named(fmt.Sprintf("operatorpkg.%s.events", strings.ToLower(c.gvk.Kind))).
+		Complete(reconcile.AsReconciler(m.GetClient(), c))
+}
+
+func (c *Controller[T]) Reconcile(ctx context.Context, event *v1.Event) (reconcile.Result, error) {
+	// We check if the event was created in the lifetime of this controller
+	// since we don't duplicate metrics on controller restart or lease handover
+	if c.startTime.Before(event.LastTimestamp.Time) {
+		c.EventCount.Inc(map[string]string{
+			pmetrics.LabelGroup:  c.gvk.Group,
+			pmetrics.LabelKind:   event.InvolvedObject.Kind,
+			pmetrics.LabelType:   event.Type,
+			pmetrics.LabelReason: event.Reason,
+		})
+	}
+
+	return reconcile.Result{}, nil
+}

events/metrics.go

@@ -0,0 +1,25 @@
+package events
+
+import (
+	pmetrics "github.com/awslabs/operatorpkg/metrics"
+	"github.com/prometheus/client_golang/prometheus"
+	"sigs.k8s.io/controller-runtime/pkg/metrics"
+)
+
+func eventTotalMetric(objectName string) pmetrics.CounterMetric {
+	return pmetrics.NewPrometheusCounter(
+		metrics.Registry,
+		prometheus.CounterOpts{
+			Namespace: pmetrics.Namespace,
+			Subsystem: objectName,
+			Name:      "event_total",
+			Help:      "The total of events of a given type for an object.",
+		},
+		[]string{
+			pmetrics.LabelGroup,
+			pmetrics.LabelKind,
+			pmetrics.LabelType,
+			pmetrics.LabelReason,
+		},
+	)
+}

events/suite_test.go

@@ -0,0 +1,132 @@
+package events_test
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/awslabs/operatorpkg/events"
+	pmetrics "github.com/awslabs/operatorpkg/metrics"
+	"github.com/awslabs/operatorpkg/object"
+	"github.com/awslabs/operatorpkg/test"
+	. "github.com/awslabs/operatorpkg/test/expectations"
+	"github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	"github.com/samber/lo"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/client-go/kubernetes/scheme"
+	clock "k8s.io/utils/clock/testing"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+	"sigs.k8s.io/controller-runtime/pkg/log"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+)
+
+var (
+	SchemeBuilder = runtime.NewSchemeBuilder(func(scheme *runtime.Scheme) error {
+		scheme.AddKnownTypes(schema.GroupVersion{Group: test.APIGroup, Version: "v1alpha1"}, &test.CustomObject{})
+		return nil
+	})
+)
+
+var ctx context.Context
+var fakeClock *clock.FakeClock
+var controller *events.Controller[*test.CustomObject]
+var kubeClient client.Client
+
+func Test(t *testing.T) {
+	lo.Must0(SchemeBuilder.AddToScheme(scheme.Scheme))
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Events")
+}
+
+var _ = BeforeSuite(func() {
+	fakeClock = clock.NewFakeClock(time.Now())
+	kubeClient = fake.NewClientBuilder().WithScheme(scheme.Scheme).WithIndex(&corev1.Event{}, "involvedObject.kind", func(o client.Object) []string {
+		evt := o.(*corev1.Event)
+		return []string{evt.InvolvedObject.Kind}
+	}).Build()
+	controller = events.NewController[*test.CustomObject](kubeClient, fakeClock)
+	ctx = log.IntoContext(context.Background(), ginkgo.GinkgoLogr)
+})
+
+var _ = Describe("Controller", func() {
+	BeforeEach(func() {
+		controller.EventCount.Reset()
+	})
+	It("should emit metrics on an event", func() {
+		events := []*corev1.Event{}
+
+		for i := range 5 {
+			// create an event for custom object
+			events = append(events, createEvent("test-object", fmt.Sprintf("Test-type-%d", i), fmt.Sprintf("Test-reason-%d", i)))
+			ExpectApplied(ctx, kubeClient, events[i])
+
+			// expect an metrics for custom object to be zero, waiting on controller reconcile
+			Expect(GetMetric("operator_customobject_event_total", conditionLabels(fmt.Sprintf("Test-type-%d", i), fmt.Sprintf("Test-reason-%d", i)))).To(BeNil())
+
+			// reconcile on the event
+			_, err := reconcile.AsReconciler(kubeClient, controller).Reconcile(ctx, reconcile.Request{NamespacedName: client.ObjectKeyFromObject(events[i])})
+			Expect(err).ToNot(HaveOccurred())
+
+			// expect an emitted metric to for the event
+			Expect(GetMetric("operator_customobject_event_total", conditionLabels(fmt.Sprintf("Test-type-%d", i), fmt.Sprintf("Test-reason-%d", i))).GetCounter().GetValue()).To(BeEquivalentTo(1))
+		}
+	})
+	It("should not fire metrics if the last transition was before controller start-up", func() {
+		// create an event for custom object that was produced before the controller start-up time
+		event := createEvent("test-name", corev1.EventTypeNormal, "reason")
+		event.LastTimestamp.Time = time.Now().Add(30 * time.Minute)
+		ExpectApplied(ctx, kubeClient, event)
+
+		// expect an metrics for custom object to be zero, waiting on controller reconcile
+		Expect(GetMetric("operator_ustomobject_event_total", conditionLabels(corev1.EventTypeNormal, "reason"))).To(BeNil())
+
+		// reconcile on the event
+		_, err := reconcile.AsReconciler(kubeClient, controller).Reconcile(ctx, reconcile.Request{NamespacedName: client.ObjectKeyFromObject(event)})
+		Expect(err).ToNot(HaveOccurred())
+
+		// expect not have an emitted metric to for the event
+		Expect(GetMetric("operator_customobject_event_total", conditionLabels(corev1.EventTypeNormal, "reason")).GetCounter().GetValue()).To(BeEquivalentTo(1))
+
+		// create an event for custom object that was produced after the controller start-up time
+		event.LastTimestamp.Time = time.Now().Add(-30 * time.Minute)
+		ExpectApplied(ctx, kubeClient, event)
+
+		// reconcile on the event
+		_, err = reconcile.AsReconciler(kubeClient, controller).Reconcile(ctx, reconcile.Request{NamespacedName: client.ObjectKeyFromObject(event)})
+		Expect(err).ToNot(HaveOccurred())
+
+		// expect an emitted metric to for the event
+		Expect(GetMetric("operator_customobject_event_total", conditionLabels(corev1.EventTypeNormal, "reason")).GetCounter().GetValue()).To(BeEquivalentTo(1))
+	})
+})
+
+func createEvent(name string, eventType string, reason string) *corev1.Event {
+	return &corev1.Event{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: test.RandomName(),
+		},
+		InvolvedObject: corev1.ObjectReference{
+			Namespace: "default",
+			Name:      name,
+			Kind:      object.GVK(&test.CustomObject{}).Kind,
+		},
+		LastTimestamp: metav1.Time{Time: time.Now().Add(30 * time.Minute)},
+		Type:          eventType,
+		Reason:        reason,
+		Count:         5,
+	}
+}
+
+func conditionLabels(eventType string, reason string) map[string]string {
+	return map[string]string{
+		pmetrics.LabelType:   eventType,
+		pmetrics.LabelReason: reason,
+	}
+}

metrics/types.go

@@ -1,5 +1,13 @@
 package metrics
 
+const (
+	Namespace   = "operator"
+	LabelGroup  = "group"
+	LabelKind   = "kind"
+	LabelType   = "type"
+	LabelReason = "reason"
+)
+
 type ObservationMetric interface {
 	Observe(v float64, labels map[string]string)
 	Delete(labels map[string]string)