Add support for consuming notifications

This patch is adding a new parameter `notificationsBusInstance` to the
Watcher CRD. The value should be the name of a rabbitmq instance which
is created specifically for notifications, not RPCs.

When it has a value, the watcher-operator will create a TransportURL
object for credentials and will configure it as transport_url in the
`oslo_messaging_notifications` section so that the dedision engine and
potentially any watcher service can consume Nova or other services
notifications.

Note that, althoug watcher can also send notifications, this patch is
disabling sending notifications by default as there is no known
consumers of those messages.

The patch is also adding functional and kuttl tests.

Jira: OSPRH-18694

Signed-off-by: Alfredo Moralejo <amoralej@redhat.com>

Author: amoralej

api/bases/watcher.openstack.org_watchers.yaml

@@ -603,6 +603,15 @@ spec:
                   NodeSelector to target subset of worker nodes running this component. Setting here overrides
                   any global NodeSelector settings within the Watcher CR.
                 type: object
+              notificationsBusInstance:
+                description: |-
+                  NotificationsBusInstance is the name of the RabbitMqCluster CR to select
+                  the Message Bus Service instance used by the Watcher service to publish and consume notifications
+                  from other services.
+                  If undefined, the value will be inherited from OpenStackControlPlane.
+                  An empty value "" leaves the notification drivers unconfigured and emitting no notifications at all.
+                  Avoid colocating it with RabbitMqClusterName or other message bus instances used for RPC.
+                type: string
               passwordSelectors:
                 default:
                   service: WatcherPassword

api/v1beta1/common_types.go

@@ -128,6 +128,15 @@ type WatcherSpecCore struct {
 	// +kubebuilder:validation:Minimum=10
 	// APITimeout for Route and Apache
 	APITimeout *int `json:"apiTimeout"`
+
+	// +kubebuilder:validation:Optional
+	// NotificationsBusInstance is the name of the RabbitMqCluster CR to select
+	// the Message Bus Service instance used by the Watcher service to publish and consume notifications
+	// from other services.
+	// If undefined, the value will be inherited from OpenStackControlPlane.
+	// An empty value "" leaves the notification drivers unconfigured and emitting no notifications at all.
+	// Avoid colocating it with RabbitMqClusterName or other message bus instances used for RPC.
+	NotificationsBusInstance *string `json:"notificationsBusInstance,omitempty"`
 }
 
 // PasswordSelector to identify the DB and AdminUser password from the Secret

api/v1beta1/conditions.go

@@ -5,6 +5,8 @@ import "github.com/openstack-k8s-operators/lib-common/modules/common/condition"
 const (
 	// WatcherRabbitMQTransportURLReadyCondition -
 	WatcherRabbitMQTransportURLReadyCondition condition.Type = "WatcherRabbitMQTransportURLReady"
+	// WatcherNotificationsBusTransportURLReadyCondition -
+	WatcherNotificationTransportURLReadyCondition condition.Type = "WatcherNotificationTransportURLReady"
 	// WatcherAPIReadyCondition -
 	WatcherAPIReadyCondition condition.Type = "WatcherAPIReady"
 	// WatcherApplierReadyCondition -
@@ -20,6 +22,12 @@ const (
 	WatcherRabbitMQTransportURLReadyMessage = "WatcherRabbitMQTransportURL successfully created"
 	// WatcherRabbitMQTransportURLReadyErrorMessage -
 	WatcherRabbitMQTransportURLReadyErrorMessage = "WatcherRabbitMQTransportURL error occured %s"
+	// WatcherNotificationTransportURLReadyRunningMessage -
+	WatcherNotificationTransportURLReadyRunningMessage = "WatcherNotificationTransportURL creation in progress"
+	// WatcherNotificationTransportURLReadyMessage -
+	WatcherNotificationTransportURLReadyMessage = "WatcherNotificationTransportURL successfully created"
+	// WatcherNotificationTransportURLReadyErrorMessage -
+	WatcherNotificationTransportURLReadyErrorMessage = "WatcherNotificationTransportURL error occured %s"
 	// WatcherAPIReadyInitMessage -
 	WatcherAPIReadyInitMessage = "WatcherAPI creation not started"
 	// WatcherAPIReadyRunningMessage -

api/v1beta1/zz_generated.deepcopy.go

@@ -603,6 +603,15 @@ spec:
                   NodeSelector to target subset of worker nodes running this component. Setting here overrides
                   any global NodeSelector settings within the Watcher CR.
                 type: object
+              notificationsBusInstance:
+                description: |-
+                  NotificationsBusInstance is the name of the RabbitMqCluster CR to select
+                  the Message Bus Service instance used by the Watcher service to publish and consume notifications
+                  from other services.
+                  If undefined, the value will be inherited from OpenStackControlPlane.
+                  An empty value "" leaves the notification drivers unconfigured and emitting no notifications at all.
+                  Avoid colocating it with RabbitMqClusterName or other message bus instances used for RPC.
+                type: string
               passwordSelectors:
                 default:
                   service: WatcherPassword

config/crd/bases/watcher.openstack.org_watchers.yaml

@@ -65,6 +65,8 @@ var (
 const (
 	// TransportURLSelector is the name of key in the secret created by TransportURL
 	TransportURLSelector = "transport_url"
+	// NotificationURLSelector is the name of key in the secret created by the notification TransportURL
+	NotificationURLSelector = "notification_url"
 	// DatabaseAccount is the name of key in the secret for the name of the Database Acount object
 	DatabaseAccount = "database_account"
 	// DatabaseUsername is the name of key in the secret for the user name used to login to the database

controllers/watcher_common.go

@@ -201,11 +201,11 @@ func (r *WatcherReconciler) Reconcile(ctx context.Context, req ctrl.Request) (re
 	// create service DB - end
 
 	//
-	// create RabbitMQ transportURL CR and get the actual URL from the associated secret that is created
+	// create RPC RabbitMQ transportURL CR and get the actual URL from the associated secret that is created
 	// not-ready condition is managed here instead of in ensureMQ to distinguish between Error (when receiving)
 	// an error, or Running when transportURL is empty.
 	//
-	transportURL, op, err := r.ensureMQ(ctx, instance, helper, serviceLabels)
+	transportURL, op, err := r.ensureMQ(ctx, instance, helper, instance.Name+"-watcher-transport", *instance.Spec.RabbitMqClusterName, serviceLabels)
 	if err != nil {
 		instance.Status.Conditions.Set(condition.FalseCondition(
 			watcherv1beta1.WatcherRabbitMQTransportURLReadyCondition,
@@ -226,9 +226,64 @@ func (r *WatcherReconciler) Reconcile(ctx context.Context, req ctrl.Request) (re
 		return ctrl.Result{RequeueAfter: time.Duration(10) * time.Second}, nil
 	}
 
+	instance.Status.Conditions.MarkTrue(watcherv1beta1.WatcherRabbitMQTransportURLReadyCondition, watcherv1beta1.WatcherRabbitMQTransportURLReadyMessage)
+
 	_ = op
 	// end of TransportURL creation
 
+	// create Notification RabbitMQ transportURL CR and get the actual URL from the associated secret that is created
+	notificationURLSecret := &corev1.Secret{}
+
+	if instance.Spec.NotificationsBusInstance != nil && *instance.Spec.NotificationsBusInstance != "" {
+		instance.Status.Conditions.Set(condition.FalseCondition(
+			watcherv1beta1.WatcherNotificationTransportURLReadyCondition,
+			condition.RequestedReason,
+			condition.SeverityInfo,
+			watcherv1beta1.WatcherNotificationTransportURLReadyRunningMessage,
+		))
+		notificationURL, op, err := r.ensureMQ(ctx, instance, helper, instance.Name+"-watcher-notification", *instance.Spec.NotificationsBusInstance, serviceLabels)
+		if err != nil {
+			instance.Status.Conditions.Set(condition.FalseCondition(
+				watcherv1beta1.WatcherNotificationTransportURLReadyCondition,
+				condition.ErrorReason,
+				condition.SeverityWarning,
+				watcherv1beta1.WatcherNotificationTransportURLReadyErrorMessage,
+				err.Error()))
+			return ctrl.Result{}, err
+		}
+
+		if notificationURL == nil {
+			Log.Info(fmt.Sprintf("Waiting for TransportURL for %s to be created", instance.Name))
+			instance.Status.Conditions.Set(condition.FalseCondition(
+				watcherv1beta1.WatcherNotificationTransportURLReadyCondition,
+				condition.RequestedReason,
+				condition.SeverityInfo,
+				watcherv1beta1.WatcherNotificationTransportURLReadyRunningMessage,
+			))
+			return ctrl.Result{RequeueAfter: time.Duration(10) * time.Second}, nil
+		}
+		instance.Status.Conditions.MarkTrue(watcherv1beta1.WatcherNotificationTransportURLReadyCondition, watcherv1beta1.WatcherNotificationTransportURLReadyMessage)
+
+		// NotificationURL Secret
+		hashNotificationURL, _, notificationSecret, err := ensureSecret(
+			ctx,
+			types.NamespacedName{Namespace: instance.Namespace, Name: notificationURL.Status.SecretName},
+			[]string{
+				TransportURLSelector,
+			},
+			helper.GetClient(),
+			&instance.Status.Conditions,
+			r.RequeueTimeout,
+		)
+		if err != nil || hashNotificationURL == "" {
+			// Empty hash means that there is some problem retrieving the key from the secret
+			return ctrl.Result{}, errors.New("error retrieving required data from notificationURL secret")
+		}
+		notificationURLSecret = &notificationSecret
+		_ = op
+	}
+
+	// end of Notification TransportURL creation
 	// Check we have the required inputs
 	// Top level secret
 	hash, _, inputSecret, err := ensureSecret(
@@ -300,7 +355,7 @@ func (r *WatcherReconciler) Reconcile(ctx context.Context, req ctrl.Request) (re
 
 	// End of Prometheus config secret
 
-	subLevelSecretName, err := r.createSubLevelSecret(ctx, helper, instance, transporturlSecret, inputSecret, db)
+	subLevelSecretName, err := r.createSubLevelSecret(ctx, helper, instance, transporturlSecret, notificationURLSecret, inputSecret, db)
 	if err != nil {
 		return ctrl.Result{}, nil
 	}
@@ -605,21 +660,23 @@ func (r *WatcherReconciler) ensureMQ(
 	ctx context.Context,
 	instance *watcherv1beta1.Watcher,
 	h *helper.Helper,
+	transportURLName string,
+	messageBusInstance string,
 	serviceLabels map[string]string,
 ) (*rabbitmqv1.TransportURL, controllerutil.OperationResult, error) {
 	Log := r.GetLogger(ctx)
-	Log.Info(fmt.Sprintf("Reconciling the RabbitMQ TransportURL for '%s'", instance.Name))
+	Log.Info(fmt.Sprintf("Reconciling the RabbitMQ TransportURL '%s' for '%s'", transportURLName, instance.Name))
 
 	transportURL := &rabbitmqv1.TransportURL{
 		ObjectMeta: metav1.ObjectMeta{
-			Name:      fmt.Sprintf("%s-watcher-transport", instance.Name),
+			Name:      transportURLName,
 			Namespace: instance.Namespace,
 			Labels:    serviceLabels,
 		},
 	}
 
 	op, err := controllerutil.CreateOrUpdate(ctx, r.Client, transportURL, func() error {
-		transportURL.Spec.RabbitmqClusterName = *instance.Spec.RabbitMqClusterName
+		transportURL.Spec.RabbitmqClusterName = messageBusInstance
 
 		err := controllerutil.SetControllerReference(instance, transportURL, r.Scheme)
 		return err
@@ -656,7 +713,6 @@ func (r *WatcherReconciler) ensureMQ(
 			"the TransportURL secret %s does not have 'transport_url' field", transportURL.Status.SecretName)
 	}
 
-	instance.Status.Conditions.MarkTrue(watcherv1beta1.WatcherRabbitMQTransportURLReadyCondition, watcherv1beta1.WatcherRabbitMQTransportURLReadyMessage)
 	return transportURL, op, nil
 }
 
@@ -810,13 +866,15 @@ func (r *WatcherReconciler) createSubLevelSecret(
 	helper *helper.Helper,
 	instance *watcherv1beta1.Watcher,
 	transportURLSecret corev1.Secret,
+	notificationURLSecret *corev1.Secret,
 	inputSecret corev1.Secret,
 	db *mariadbv1.Database,
 ) (string, error) {
 	Log := r.GetLogger(ctx)
 	Log.Info(fmt.Sprintf("Creating SubCr Level Secret for '%s'", instance.Name))
 	databaseAccount := db.GetAccount()
 	databaseSecret := db.GetSecret()
+
 	data := map[string]string{
 		*instance.Spec.PasswordSelectors.Service: string(inputSecret.Data[*instance.Spec.PasswordSelectors.Service]),
 		TransportURLSelector:                     string(transportURLSecret.Data[TransportURLSelector]),
@@ -825,6 +883,7 @@ func (r *WatcherReconciler) createSubLevelSecret(
 		DatabasePassword:                         string(databaseSecret.Data[mariadbv1.DatabasePasswordSelector]),
 		DatabaseHostname:                         db.GetDatabaseHostname(),
 		watcher.GlobalCustomConfigFileName:       instance.Spec.CustomServiceConfig,
+		NotificationURLSelector:                  string(notificationURLSecret.Data[TransportURLSelector]),
 	}
 	secretName := instance.Name
 

controllers/watcher_controller.go

@@ -466,6 +466,10 @@ func (r *WatcherAPIReconciler) generateServiceConfigs(
 		"PrometheusCaCertPath":     prometheusCaCertPath,
 	}
 
+	if string(secret.Data[NotificationURLSelector]) != "" {
+		templateParameters["NotificationURL"] = string(secret.Data[NotificationURLSelector])
+	}
+
 	// create httpd  vhost template parameters
 	httpdVhostConfig := map[string]interface{}{}
 	for _, endpt := range []service.Endpoint{service.EndpointInternal, service.EndpointPublic} {

controllers/watcherapi_controller.go

@@ -429,6 +429,10 @@ func (r *WatcherApplierReconciler) generateServiceConfigs(
 		"CaFilePath":               CaFilePath,
 	}
 
+	if string(secret.Data[NotificationURLSelector]) != "" {
+		templateParameters["NotificationURL"] = string(secret.Data[NotificationURLSelector])
+	}
+
 	return GenerateConfigsGeneric(ctx, helper, instance, envVars, templateParameters, customData, labels, false)
 }
 

controllers/watcherapplier_controller.go

@@ -542,6 +542,10 @@ func (r *WatcherDecisionEngineReconciler) generateServiceConfigs(
 		"PrometheusCaCertPath":     prometheusCaCertPath,
 	}
 
+	if string(secret.Data[NotificationURLSelector]) != "" {
+		templateParameters["NotificationURL"] = string(secret.Data[NotificationURLSelector])
+	}
+
 	return GenerateConfigsGeneric(ctx, helper, instance, envVars, templateParameters, customData, labels, false)
 }