Watch cinder endpoint in WatcherDecisionEngine

Watch the cinder KeystoneEndpoint in the WatcherDecisionEngine to
detect changes and trigger reconciliation when endpoints are
modified. This ensures the decision engine is reconciled if Cinder
is enabled or disabled to ensure the storage collector is properly
configured.

The endpoint URL hashing logic detects when cached endpoint URLs
need to be updated and triggers deployment restarts accordingly.

Assisted-By: Cursor (claude-4-sonnet)

Resolves: OSPRH-20349

Author: cescgina

controllers/watcher_common.go

@@ -36,6 +36,8 @@ const (
 	tlsAPIPublicField       = ".spec.tls.api.public.secretName"
 	topologyField           = ".spec.topologyRef.Name"
 	memcachedInstanceField  = ".spec.memcachedInstance"
+	// service label for cinder endpoint
+	endpointCinder = "cinder"
 )
 
 var (
@@ -66,6 +68,9 @@ var (
 		topologyField,
 		memcachedInstanceField,
 	}
+	endpointList = []string{
+		endpointCinder,
+	}
 )
 
 const (

controllers/watcherdecisionengine_controller.go

@@ -53,6 +53,7 @@ import (
 	k8s_errors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/utils/ptr"
 )
 
 // WatcherDecisionEngineReconciler reconciles a WatcherDecisionEngine object
@@ -229,6 +230,25 @@ func (r *WatcherDecisionEngineReconciler) Reconcile(ctx context.Context, req ctr
 		return ctrl.Result{}, err
 	}
 
+	// hash the endpoint URLs of the services this depends on
+	// By adding the hash to the hash of hashes being added to the deployment
+	// allows it to get restarted, in case the endpoint changes and it requires
+	// the current cached ones to be updated.
+
+	endpointUrlsHash, err := keystonev1.GetHashforKeystoneEndpointUrlsForServices(
+		ctx,
+		helper,
+		instance.Namespace,
+		ptr.To(string(endpoint.EndpointInternal)),
+		endpointList,
+	)
+
+	if err != nil {
+		return ctrl.Result{}, err
+	}
+
+	configVars["endpointUrlsHash"] = env.SetValue(endpointUrlsHash)
+
 	Log.Info(fmt.Sprintf("[DecisionEngine] Getting input hash '%s'", instance.Name))
 	//
 	// create hash over all the different input resources to identify if any those changed
@@ -403,6 +423,9 @@ func (r *WatcherDecisionEngineReconciler) SetupWithManager(mgr ctrl.Manager) err
 		Watches(&keystonev1.KeystoneAPI{},
 			handler.EnqueueRequestsFromMapFunc(r.findObjectForSrc),
 			builder.WithPredicates(keystonev1.KeystoneAPIStatusChangedPredicate)).
+		Watches(&keystonev1.KeystoneEndpoint{},
+			handler.EnqueueRequestsFromMapFunc(r.findObjectsWithAppSelectorLabelInNamespace),
+			builder.WithPredicates(keystonev1.KeystoneEndpointStatusChangedPredicate)).
 		Complete(r)
 }
 
@@ -700,3 +723,36 @@ func getDecisionEngineServiceLabels() map[string]string {
 		common.AppSelector: WatcherDecisionEngineLabelPrefix,
 	}
 }
+
+func (r *WatcherDecisionEngineReconciler) findObjectsWithAppSelectorLabelInNamespace(ctx context.Context, src client.Object) []reconcile.Request {
+	requests := []reconcile.Request{}
+
+	l := log.FromContext(ctx).WithName("Controllers").WithName("WatcherDecisionEngine")
+
+	// if the endpoint has the service label and its in our endpointList, reconcile the CR in the namespace
+	if svc, ok := src.GetLabels()[common.AppSelector]; ok && util.StringInSlice(svc, endpointList) {
+		crList := &watcherv1beta1.WatcherDecisionEngineList{}
+		listOps := &client.ListOptions{
+			Namespace: src.GetNamespace(),
+		}
+		err := r.Client.List(ctx, crList, listOps)
+		if err != nil {
+			l.Error(err, fmt.Sprintf("listing %s for namespace: %s", crList.GroupVersionKind().Kind, src.GetNamespace()))
+			return requests
+		}
+
+		for _, item := range crList.Items {
+			l.Info(fmt.Sprintf("input source %s changed, reconcile: %s - %s", src.GetName(), item.GetName(), item.GetNamespace()))
+
+			requests = append(requests,
+				reconcile.Request{
+					NamespacedName: types.NamespacedName{
+						Name:      item.GetName(),
+						Namespace: item.GetNamespace(),
+					},
+				},
+			)
+		}
+	}
+	return requests
+}

tests/functional/watcherdecisionengine_controller_test.go

@@ -1133,4 +1133,124 @@ heartbeat_in_pthread=false`,
 			}, timeout, interval).Should(Succeed())
 		})
 	})
+	When("WatcherDecisionEngine is reconfigured", func() {
+		cinderEndpoint := types.NamespacedName{}
+		BeforeEach(func() {
+			secret := CreateInternalTopLevelSecret()
+			DeferCleanup(k8sClient.Delete, ctx, secret)
+			prometheusSecret := th.CreateSecret(
+				watcherTest.PrometheusSecretName,
+				map[string][]byte{
+					"host": []byte("prometheus.example.com"),
+					"port": []byte("9090"),
+				},
+			)
+			DeferCleanup(k8sClient.Delete, ctx, prometheusSecret)
+			DeferCleanup(
+				mariadb.DeleteDBService,
+				mariadb.CreateDBService(
+					watcherTest.WatcherDecisionEngine.Namespace,
+					"openstack",
+					corev1.ServiceSpec{
+						Ports: []corev1.ServicePort{{Port: 3306}},
+					},
+				),
+			)
+			mariadb.CreateMariaDBAccountAndSecret(
+				watcherTest.WatcherDatabaseAccount,
+				mariadbv1.MariaDBAccountSpec{
+					UserName: "watcher",
+				},
+			)
+			mariadb.CreateMariaDBDatabase(
+				watcherTest.WatcherDecisionEngine.Namespace,
+				"watcher",
+				mariadbv1.MariaDBDatabaseSpec{
+					Name: "watcher",
+				},
+			)
+			mariadb.SimulateMariaDBAccountCompleted(watcherTest.WatcherDatabaseAccount)
+			mariadb.SimulateMariaDBDatabaseCompleted(watcherTest.WatcherDatabaseName)
+
+			DeferCleanup(keystone.DeleteKeystoneAPI, keystone.CreateKeystoneAPI(watcherTest.WatcherDecisionEngine.Namespace))
+
+			memcachedSpec := memcachedv1.MemcachedSpec{
+				MemcachedSpecCore: memcachedv1.MemcachedSpecCore{
+					Replicas: ptr.To(int32(1)),
+				},
+			}
+			DeferCleanup(infra.DeleteMemcached, infra.CreateMemcached(watcherTest.WatcherDecisionEngine.Namespace, MemcachedInstance, memcachedSpec))
+			infra.SimulateMemcachedReady(watcherTest.MemcachedNamespace)
+
+			DeferCleanup(th.DeleteInstance, CreateWatcherDecisionEngine(watcherTest.WatcherDecisionEngine, GetDefaultWatcherDecisionEngineSpec()))
+			// create watcher applier and watcher api to later check that their observed generation has not changed
+			DeferCleanup(th.DeleteInstance, CreateWatcherApplier(watcherTest.WatcherApplier, GetDefaultWatcherApplierSpec()))
+			DeferCleanup(th.DeleteInstance, CreateWatcherAPI(watcherTest.WatcherAPI, GetDefaultWatcherAPISpec()))
+
+			logger.Info("Created cinder endpoint")
+			cinderEndpoint = types.NamespacedName{Name: "cinder", Namespace: watcherTest.WatcherDecisionEngine.Namespace}
+			DeferCleanup(keystone.DeleteKeystoneEndpoint, keystone.CreateKeystoneEndpoint(cinderEndpoint))
+			keystone.SimulateKeystoneEndpointReady(cinderEndpoint)
+
+			th.SimulateStatefulSetReplicaReady(watcherTest.WatcherDecisionEngineStatefulSet)
+			th.SimulateStatefulSetReplicaReady(watcherTest.WatcherApplierStatefulSet)
+			th.SimulateStatefulSetReplicaReady(watcherTest.WatcherAPIStatefulSet)
+
+			th.ExpectCondition(
+				watcherTest.WatcherDecisionEngine,
+				ConditionGetterFunc(WatcherDecisionEngineConditionGetter),
+				condition.ReadyCondition,
+				corev1.ConditionTrue,
+			)
+
+		})
+		It("updates the deployment if cinder public endpoint gets deleted", func() {
+			originalConfigHash := GetEnvVarValue(
+				th.GetStatefulSet(watcherTest.WatcherDecisionEngine).Spec.Template.Spec.Containers[0].Env, "CONFIG_HASH", "")
+			Expect(originalConfigHash).NotTo(Equal(""))
+			decisionEngineObservedOrig := th.GetStatefulSet(watcherTest.WatcherDecisionEngine).Status.ObservedGeneration
+			applierObservedOrig := th.GetStatefulSet(watcherTest.WatcherApplier).Status.ObservedGeneration
+			apiObservedOrig := th.GetStatefulSet(watcherTest.WatcherAPI).Status.ObservedGeneration
+
+			keystone.DeleteKeystoneEndpoint(cinderEndpoint)
+			logger.Info("Deleted cinder endpoint")
+			// Assert that the CONFIG_HASH of the StateFulSet is changed due to this reconfiguration
+			Eventually(func(g Gomega) {
+				currentConfigHash := GetEnvVarValue(
+					th.GetStatefulSet(watcherTest.WatcherDecisionEngine).Spec.Template.Spec.Containers[0].Env, "CONFIG_HASH", "")
+				g.Expect(originalConfigHash).NotTo(Equal(currentConfigHash))
+
+			}, timeout, interval).Should(Succeed())
+
+			// Simulate the StatefulSet replicas to be ready after deleting cinder
+			th.SimulateStatefulSetReplicaReady(watcherTest.WatcherDecisionEngineStatefulSet)
+			th.SimulateStatefulSetReplicaReady(watcherTest.WatcherApplierStatefulSet)
+			th.SimulateStatefulSetReplicaReady(watcherTest.WatcherAPIStatefulSet)
+
+			// check that the StatefulSet replica watcher-decision-engine is updated while watcher-applier and watcher-api are not updated
+			applierObservedNew := th.GetStatefulSet(watcherTest.WatcherApplier).Status.ObservedGeneration
+			Expect(applierObservedNew).Should(Equal(applierObservedOrig))
+			apiObservedNew := th.GetStatefulSet(watcherTest.WatcherAPI).Status.ObservedGeneration
+			Expect(apiObservedNew).Should(Equal(apiObservedOrig))
+			decisionEngineObservedNew := th.GetStatefulSet(watcherTest.WatcherDecisionEngine).Status.ObservedGeneration
+			Expect(decisionEngineObservedNew).Should(Equal(decisionEngineObservedOrig + 1))
+
+		})
+		It("updates the deployment if cinder internal endpoint is modified", func() {
+			originalConfigHash := GetEnvVarValue(
+				th.GetStatefulSet(watcherTest.WatcherDecisionEngine).Spec.Template.Spec.Containers[0].Env, "CONFIG_HASH", "")
+			Expect(originalConfigHash).NotTo(Equal(""))
+
+			keystone.UpdateKeystoneEndpoint(cinderEndpoint, "internal", "https://cinder-test-internal")
+			logger.Info("Reconfigured")
+
+			// Assert that the CONFIG_HASH of the StateFulSet is changed due to this reconfiguration
+			Eventually(func(g Gomega) {
+				currentConfigHash := GetEnvVarValue(
+					th.GetStatefulSet(watcherTest.WatcherDecisionEngine).Spec.Template.Spec.Containers[0].Env, "CONFIG_HASH", "")
+				g.Expect(originalConfigHash).NotTo(Equal(currentConfigHash))
+
+			}, timeout, interval).Should(Succeed())
+		})
+	})
 })

tests/kuttl/test-suites/default/deps/kustomization.yaml

@@ -21,6 +21,7 @@ secretGenerator:
   - NovaAPIDatabasePassword=password
   - NovaCell0DatabasePassword=password
   - NovaCell1DatabasePassword=password
+  - CinderPassword=password
   - MetadataSecret=42
   name: osp-secret
 generatorOptions:

tests/kuttl/test-suites/default/watcher-cinder/00-cleanup-watcher.yaml

@@ -0,0 +1 @@
+../common/cleanup-watcher.yaml

tests/kuttl/test-suites/default/watcher-cinder/01-assert.yaml

@@ -0,0 +1,21 @@
+apiVersion: watcher.openstack.org/v1beta1
+kind: Watcher
+metadata:
+  finalizers:
+  - openstack.org/watcher
+  name: watcher-kuttl
+  namespace: watcher-kuttl-default
+status:
+  # we just want to assert that the watcher is ready in this test
+  apiServiceReadyCount: 1
+  applierServiceReadyCount: 1
+  decisionengineServiceReadyCount: 1
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestAssert
+namespaced: true
+commands:
+  - script: |
+      set -euxo pipefail
+      # check that the decision engine correctly detects that there is no cinder service
+      [ "$(oc logs -n $NAMESPACE watcher-kuttl-decision-engine-0 | grep -c 'Block storage service is not enabled, skipping storage collector')" == 2 ]

tests/kuttl/test-suites/default/watcher-cinder/01-deploy-watcher-no-cinder.yaml

@@ -0,0 +1,10 @@
+apiVersion: watcher.openstack.org/v1beta1
+kind: Watcher
+metadata:
+  name: watcher-kuttl
+  namespace: watcher-kuttl-default
+spec:
+  databaseInstance: "openstack"
+  apiServiceTemplate:
+    tls:
+      caBundleSecretName: "combined-ca-bundle"

tests/kuttl/test-suites/default/watcher-cinder/02-assert.yaml

@@ -0,0 +1,32 @@
+apiVersion: watcher.openstack.org/v1beta1
+kind: Watcher
+metadata:
+  finalizers:
+  - openstack.org/watcher
+  name: watcher-kuttl
+  namespace: watcher-kuttl-default
+status:
+  # we just want to assert that the watcher is ready in this test
+  apiServiceReadyCount: 1
+  applierServiceReadyCount: 1
+  decisionengineServiceReadyCount: 1
+---
+apiVersion: keystone.openstack.org/v1beta1
+kind: KeystoneService
+metadata:
+  name: cinderv3
+---
+apiVersion: keystone.openstack.org/v1beta1
+kind: KeystoneEndpoint
+metadata:
+  name: cinderv3
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestAssert
+namespaced: true
+commands:
+  - script: |
+      set -euxo pipefail
+      # check that the decision detects that there is a cinder service and
+      # does not log that storage collector is skipped
+      [ "$(oc logs -n $NAMESPACE watcher-kuttl-decision-engine-0 |grep -c 'Block storage service is not enabled, skipping storage collector')" == 0 ]

tests/kuttl/test-suites/default/watcher-cinder/02-deploy-cinder.yaml

@@ -0,0 +1,13 @@
+apiVersion: core.openstack.org/v1beta1
+kind: OpenStackControlPlane
+metadata:
+  name: openstack
+spec:
+  cinder:
+    enabled: true
+    template:
+      databaseInstance: openstack
+      databaseAccount: cinder
+      secret: osp-secret
+      cinderAPI:
+        replicas: 1

tests/kuttl/test-suites/default/watcher-cinder/03-assert.yaml

@@ -0,0 +1,21 @@
+apiVersion: watcher.openstack.org/v1beta1
+kind: Watcher
+metadata:
+  finalizers:
+  - openstack.org/watcher
+  name: watcher-kuttl
+  namespace: watcher-kuttl-default
+status:
+  # we just want to assert that the watcher is ready in this test
+  apiServiceReadyCount: 1
+  applierServiceReadyCount: 1
+  decisionengineServiceReadyCount: 1
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestAssert
+namespaced: true
+commands:
+  - script: |
+      set -euxo pipefail
+      # check that the decision engine correctly detects that there is no cinder service
+      [ "$(oc logs -n $NAMESPACE watcher-kuttl-decision-engine-0 |grep -c 'Block storage service is not enabled, skipping storage collector')" == 2 ]