reconcile redis services

Author: odacremolbap

pkg/apis/eventing/v1alpha1/redisbroker_lifecycle.go

@@ -4,10 +4,12 @@
 package v1alpha1
 
 import (
+	"context"
 	"sync"
 
+	appsv1 "k8s.io/api/apps/v1"
+	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
-
 	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 )
@@ -97,6 +99,39 @@ func (bs *RedisBrokerStatus) InitializeConditions() {
 	bs.GetConditionSet().Manage(bs).InitializeConditions()
 }
 
-func (bs *RedisBrokerStatus) MarkRedisBrokerFailed(reason, messageFormat string, messageA ...interface{}) {
+func (bs *RedisBrokerStatus) MarkRedisDeploymentFailed(reason, messageFormat string, messageA ...interface{}) {
 	redisBrokerCondSet.Manage(bs).MarkFalse(RedisBrokerRedisDeployment, reason, messageFormat, messageA...)
 }
+
+func (bs *RedisBrokerStatus) MarkRedisDeploymentUnknown(reason, messageFormat string, messageA ...interface{}) {
+	redisBrokerCondSet.Manage(bs).MarkUnknown(RedisBrokerRedisDeployment, reason, messageFormat, messageA...)
+}
+
+func (bs *RedisBrokerStatus) PropagateRedisDeploymentAvailability(ctx context.Context, ds *appsv1.DeploymentStatus) {
+	for _, cond := range ds.Conditions {
+
+		if cond.Type == appsv1.DeploymentAvailable {
+			switch cond.Status {
+			case corev1.ConditionTrue:
+				redisBrokerCondSet.Manage(bs).MarkTrue(RedisBrokerRedisDeployment)
+			case corev1.ConditionFalse:
+				bs.MarkRedisDeploymentFailed("RedisDeploymentFalse", "The status of Redis Deployment is False: %s : %s", cond.Reason, cond.Message)
+			default:
+				// expected corev1.ConditionUnknown
+				bs.MarkRedisDeploymentUnknown("RedisDeploymentUnknown", "The status of Redis Deployment is Unknown: %s : %s", cond.Reason, cond.Message)
+			}
+		}
+	}
+}
+
+func (bs *RedisBrokerStatus) MarkRedisServiceFailed(reason, messageFormat string, messageA ...interface{}) {
+	redisBrokerCondSet.Manage(bs).MarkFalse(RedisBrokerRedisService, reason, messageFormat, messageA...)
+}
+
+func (bs *RedisBrokerStatus) MarkRedisServiceUnknown(reason, messageFormat string, messageA ...interface{}) {
+	redisBrokerCondSet.Manage(bs).MarkUnknown(RedisBrokerRedisService, reason, messageFormat, messageA...)
+}
+
+func (bs *RedisBrokerStatus) MarkRedisServiceReady() {
+	redisBrokerCondSet.Manage(bs).MarkTrue(RedisBrokerRedisService)
+}

pkg/reconciler/events.go

@@ -15,19 +15,16 @@ const (
 	// ReasonFailedRBACUpdate = "FailedRBACUpdate"
 
 	ReasonDeploymentCreate       = "CreateDeployment"
-	ReasonDeploymentGet          = "GetDeployment"
 	ReasonDeploymentUpdate       = "UpdateDeployment"
+	ReasonFailedDeploymentGet    = "FailedDeploymentGet"
 	ReasonFailedDeploymentCreate = "FailedDeploymentCreate"
 	ReasonFailedDeploymentUpdate = "FailedDeploymentUpdate"
 
-	// // ReasonServiceCreate indicates that an Service object was successfully created.
-	// ReasonServiceCreate = "CreateService"
-	// // ReasonServiceUpdate indicates that an Service object was successfully updated.
-	// ReasonServiceUpdate = "UpdateService"
-	// // ReasonFailedServiceCreate indicates that the creation of an Service object failed.
-	// ReasonFailedServiceCreate = "FailedServiceCreate"
-	// // ReasonFailedServiceUpdate indicates that the update of an Service object failed.
-	// ReasonFailedServiceUpdate = "FailedServiceUpdate"
+	ReasonServiceCreate       = "CreateService"
+	ReasonServiceUpdate       = "UpdateService"
+	ReasonFailedServiceGet    = "FailedServiceGet"
+	ReasonFailedServiceCreate = "FailedServiceCreate"
+	ReasonFailedServiceUpdate = "FailedServiceUpdate"
 
 	// // ReasonBadSinkURI indicates that the URI of a sink can't be determined.
 	// ReasonBadSinkURI = "BadSinkURI"

pkg/reconciler/redisbroker/controller.go

@@ -35,7 +35,7 @@ func NewController(
 
 	r := &Reconciler{
 		kubeClientSet:        kubeclient.Get(ctx),
-		redisReconciler:      newRedisReconciler(ctx, deploymentInformer.Lister()),
+		redisReconciler:      newRedisReconciler(ctx, deploymentInformer.Lister(), serviceInformer.Lister()),
 		serviceLister:        serviceInformer.Lister(),
 		serviceAccountLister: serviceAccountInformer.Lister(),
 		roleBindingLister:    roleBindingInformer.Lister(),

pkg/reconciler/redisbroker/reconcile_redis.go

@@ -7,55 +7,47 @@ import (
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	apierrs "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/kubernetes"
 	appsv1listers "k8s.io/client-go/listers/apps/v1"
+	corev1listers "k8s.io/client-go/listers/core/v1"
 	k8sclient "knative.dev/pkg/client/injection/kube/client"
 	"knative.dev/pkg/logging"
 	pkgreconciler "knative.dev/pkg/reconciler"
 
 	eventingv1alpha1 "github.com/triggermesh/triggermesh-core/pkg/apis/eventing/v1alpha1"
 	"github.com/triggermesh/triggermesh-core/pkg/reconciler"
 	"github.com/triggermesh/triggermesh-core/pkg/reconciler/resources"
+	"github.com/triggermesh/triggermesh-core/pkg/reconciler/semantic"
 )
 
 type redisReconciler struct {
 	client           kubernetes.Interface
 	deploymentLister appsv1listers.DeploymentLister
+	serviceLister    corev1listers.ServiceLister
 }
 
-func newRedisReconciler(ctx context.Context, deploymentLister appsv1listers.DeploymentLister) redisReconciler {
+func newRedisReconciler(ctx context.Context, deploymentLister appsv1listers.DeploymentLister, serviceLister corev1listers.ServiceLister) redisReconciler {
 	return redisReconciler{
 		client:           k8sclient.Get(ctx),
 		deploymentLister: deploymentLister,
+		serviceLister:    serviceLister,
 	}
 }
 
-func (r *redisReconciler) Reconcile(ctx context.Context, rb *eventingv1alpha1.RedisBroker) (*appsv1.Deployment, error) {
-
-	desired := buildRedisDeployment(rb)
-	current, err := r.deploymentLister.Deployments(desired.Namespace).Get(desired.Name)
-	switch {
-	case err == nil:
-		// TODO compare
-		// TODO if equal return
-		// continue
-	case apierrs.IsNotFound(err):
-		// continue
-	default:
-		fullname := types.NamespacedName{Namespace: desired.Namespace, Name: desired.Name}
-		logging.FromContext(ctx).Error("Unable to get the deployment", zap.String("deployment", fullname.String()), zap.Error(err))
-		rb.Status.MarkRedisBrokerFailed("DeploymentGetFailed", "Failed to get Redis deployment")
-
-		return nil, pkgreconciler.NewEvent(corev1.EventTypeWarning, reconciler.ReasonDeploymentGet,
-			"Failed to get Redis deployment %s: %w", fullname, err)
+func (r *redisReconciler) reconcile(ctx context.Context, rb *eventingv1alpha1.RedisBroker) (*appsv1.Deployment, *corev1.Service, error) {
+	d, err := r.reconcileDeployment(ctx, rb)
+	if err != nil {
+		return nil, nil, err
 	}
 
-	// Create
-
-	// TODO Update statuses
+	svc, err := r.reconcileService(ctx, rb)
+	if err != nil {
+		return d, nil, err
+	}
 
-	return current, nil
+	return d, svc, nil
 }
 
 func buildRedisDeployment(rb *eventingv1alpha1.RedisBroker) *appsv1.Deployment {
@@ -72,3 +64,112 @@ func buildRedisDeployment(rb *eventingv1alpha1.RedisBroker) *appsv1.Deployment {
 					resources.ContainerAddEnvFromValue("REDIS_ARGS", "--appendonly yes"),
 					resources.ContainerAddPort("redis", 6379)))))
 }
+
+func (r *redisReconciler) reconcileDeployment(ctx context.Context, rb *eventingv1alpha1.RedisBroker) (*appsv1.Deployment, error) {
+	desired := buildRedisDeployment(rb)
+	current, err := r.deploymentLister.Deployments(desired.Namespace).Get(desired.Name)
+	switch {
+	case err == nil:
+		// Compare current object with desired, update if needed.
+		if !semantic.Semantic.DeepEqual(desired, current) {
+			desired.Status = current.Status
+			desired.ResourceVersion = current.ResourceVersion
+
+			current, err = r.client.AppsV1().Deployments(desired.Namespace).Update(ctx, desired, metav1.UpdateOptions{})
+			if err != nil {
+				fullname := types.NamespacedName{Namespace: desired.Namespace, Name: desired.Name}
+				logging.FromContext(ctx).Error("Unable to update the deployment", zap.String("deployment", fullname.String()), zap.Error(err))
+				rb.Status.MarkRedisDeploymentFailed(reconciler.ReasonFailedDeploymentUpdate, "Failed to update Redis deployment")
+
+				return nil, pkgreconciler.NewEvent(corev1.EventTypeWarning, reconciler.ReasonFailedDeploymentUpdate,
+					"Failed to get Redis deployment %s: %w", fullname, err)
+			}
+		}
+
+	case !apierrs.IsNotFound(err):
+		// An error ocurred retrieving current deployment.
+		fullname := types.NamespacedName{Namespace: desired.Namespace, Name: desired.Name}
+		logging.FromContext(ctx).Error("Unable to get the deployment", zap.String("deployment", fullname.String()), zap.Error(err))
+		rb.Status.MarkRedisDeploymentFailed(reconciler.ReasonFailedDeploymentGet, "Failed to get Redis deployment")
+
+		return nil, pkgreconciler.NewEvent(corev1.EventTypeWarning, reconciler.ReasonFailedDeploymentGet,
+			"Failed to get Redis deployment %s: %w", fullname, err)
+
+	default:
+		// The deployment has not been found, create it.
+		current, err = r.client.AppsV1().Deployments(desired.Namespace).Create(ctx, desired, metav1.CreateOptions{})
+		if err != nil {
+			fullname := types.NamespacedName{Namespace: desired.Namespace, Name: desired.Name}
+			logging.FromContext(ctx).Error("Unable to create the deployment", zap.String("deployment", fullname.String()), zap.Error(err))
+			rb.Status.MarkRedisDeploymentFailed(reconciler.ReasonFailedDeploymentCreate, "Failed to create Redis deployment")
+
+			return nil, pkgreconciler.NewEvent(corev1.EventTypeWarning, reconciler.ReasonFailedDeploymentCreate,
+				"Failed to create Redis deployment %s: %w", fullname, err)
+		}
+	}
+
+	// Update status based on deployment
+	rb.Status.PropagateRedisDeploymentAvailability(ctx, &current.Status)
+
+	return current, nil
+}
+
+func buildRedisService(rb *eventingv1alpha1.RedisBroker) *corev1.Service {
+	return resources.NewService(rb.Namespace, rb.Name+"-redis-server",
+		resources.ServiceWithMetaOptions(
+			resources.MetaAddLabel("app", "redis-server"),
+			resources.MetaAddLabel("eventing.triggermesh.io/redis-name", rb.Name+"-redis-server"),
+			resources.MetaAddOwner(rb, rb.GetGroupVersionKind())),
+		resources.ServiceSetType(corev1.ServiceTypeClusterIP),
+		resources.ServiceAddSelectorLabel("eventing.triggermesh.io/redis-name", rb.Name+"-redis-server"),
+		resources.ServiceAddPort("redis", 6379, 6379))
+}
+
+func (r *redisReconciler) reconcileService(ctx context.Context, rb *eventingv1alpha1.RedisBroker) (*corev1.Service, error) {
+	desired := buildRedisService(rb)
+	current, err := r.serviceLister.Services(desired.Namespace).Get(desired.Name)
+	switch {
+	case err == nil:
+		// Compare current object with desired, update if needed.
+		if !semantic.Semantic.DeepEqual(desired, current) {
+			desired.Status = current.Status
+			desired.ResourceVersion = current.ResourceVersion
+
+			current, err = r.client.CoreV1().Services(desired.Namespace).Update(ctx, desired, metav1.UpdateOptions{})
+			if err != nil {
+				fullname := types.NamespacedName{Namespace: desired.Namespace, Name: desired.Name}
+				logging.FromContext(ctx).Error("Unable to update the service", zap.String("service", fullname.String()), zap.Error(err))
+				rb.Status.MarkRedisServiceFailed(reconciler.ReasonFailedServiceUpdate, "Failed to update Redis service")
+
+				return nil, pkgreconciler.NewEvent(corev1.EventTypeWarning, reconciler.ReasonFailedServiceUpdate,
+					"Failed to get Redis service %s: %w", fullname, err)
+			}
+		}
+
+	case !apierrs.IsNotFound(err):
+		// An error ocurred retrieving current object.
+		fullname := types.NamespacedName{Namespace: desired.Namespace, Name: desired.Name}
+		logging.FromContext(ctx).Error("Unable to get the service", zap.String("service", fullname.String()), zap.Error(err))
+		rb.Status.MarkRedisServiceFailed(reconciler.ReasonFailedServiceGet, "Failed to get Redis service")
+
+		return nil, pkgreconciler.NewEvent(corev1.EventTypeWarning, reconciler.ReasonFailedServiceGet,
+			"Failed to get Redis service %s: %w", fullname, err)
+
+	default:
+		// The object has not been found, create it.
+		current, err = r.client.CoreV1().Services(desired.Namespace).Create(ctx, desired, metav1.CreateOptions{})
+		if err != nil {
+			fullname := types.NamespacedName{Namespace: desired.Namespace, Name: desired.Name}
+			logging.FromContext(ctx).Error("Unable to create the service", zap.String("service", fullname.String()), zap.Error(err))
+			rb.Status.MarkRedisServiceFailed(reconciler.ReasonFailedServiceCreate, "Failed to create Redis service")
+
+			return nil, pkgreconciler.NewEvent(corev1.EventTypeWarning, reconciler.ReasonFailedServiceCreate,
+				"Failed to create Redis service %s: %w", fullname, err)
+		}
+	}
+
+	// Service exists and is up to date.
+	rb.Status.MarkRedisServiceReady()
+
+	return current, nil
+}

pkg/reconciler/redisbroker/reconciler.go

@@ -33,7 +33,7 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, rb *eventingv1alpha1.Red
 	// Iterate triggers and create secret
 
 	// Make sure the Redis deployment exists and propagate the status to the Channel
-	_, err := r.redisReconciler.Reconcile(ctx, rb)
+	_, _, err := r.redisReconciler.reconcile(ctx, rb)
 	if err != nil {
 		return err
 	}

pkg/reconciler/resources/service.go

@@ -0,0 +1,67 @@
+// Copyright 2022 TriggerMesh Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+package resources
+
+import (
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/intstr"
+)
+
+type ServiceOption func(*corev1.Service)
+
+func NewService(namespace, name string, opts ...ServiceOption) *corev1.Service {
+	meta := NewMeta(namespace, name)
+	d := &corev1.Service{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       "Service",
+			APIVersion: corev1.SchemeGroupVersion.String(),
+		},
+		ObjectMeta: *meta,
+	}
+
+	for _, opt := range opts {
+		opt(d)
+	}
+
+	return d
+}
+
+func ServiceWithMetaOptions(opts ...MetaOption) ServiceOption {
+	return func(s *corev1.Service) {
+		for _, opt := range opts {
+			opt(&s.ObjectMeta)
+		}
+	}
+}
+
+func ServiceAddSelectorLabel(key, value string) ServiceOption {
+	return func(s *corev1.Service) {
+		if s.Spec.Selector == nil {
+			s.Spec.Selector = make(map[string]string, 1)
+		}
+
+		s.Spec.Selector[key] = value
+	}
+}
+
+func ServiceAddPort(name string, port int32, targetPort int) ServiceOption {
+	return func(s *corev1.Service) {
+		if s.Spec.Ports == nil {
+			s.Spec.Ports = make([]corev1.ServicePort, 0, 1)
+		}
+
+		s.Spec.Ports = append(s.Spec.Ports, corev1.ServicePort{
+			Name:       name,
+			Port:       port,
+			TargetPort: intstr.FromInt(targetPort),
+		})
+	}
+}
+
+func ServiceSetType(st corev1.ServiceType) ServiceOption {
+	return func(s *corev1.Service) {
+		s.Spec.Type = st
+	}
+}