Telemetry webhook validation

.github/workflows/test_and_build_image.yaml

@@ -16,7 +16,6 @@ jobs:
   unit_integration_tests:
     name: Run unit/integration tests
     runs-on: ubuntu-latest
-    env:
     steps:
       - name: Set up Go 1.x
         uses: actions/setup-go@v2

apis/k8ssandra/v1alpha1/k8ssandracluster_webhook.go

@@ -19,10 +19,13 @@ package v1alpha1
 import (
 	"fmt"
 
+	telemetryapi "github.com/k8ssandra/k8ssandra-operator/apis/telemetry/v1alpha1"
 	"github.com/k8ssandra/k8ssandra-operator/pkg/clientcache"
+	validationpkg "github.com/k8ssandra/k8ssandra-operator/pkg/validation"
 	"github.com/pkg/errors"
 	"k8s.io/apimachinery/pkg/runtime"
 	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 	logf "sigs.k8s.io/controller-runtime/pkg/log"
 	"sigs.k8s.io/controller-runtime/pkg/webhook"
 )
@@ -67,6 +70,7 @@ func (r *K8ssandraCluster) ValidateCreate() error {
 func (r *K8ssandraCluster) validateK8ssandraCluster() error {
 	hasClusterStorageConfig := r.Spec.Cassandra.StorageConfig != nil
 	// Verify given k8s-contexts are correct
+
 	for _, dc := range r.Spec.Cassandra.Datacenters {
 		_, err := clientCache.GetRemoteClient(dc.K8sContext)
 		if err != nil {
@@ -85,14 +89,107 @@ func (r *K8ssandraCluster) validateK8ssandraCluster() error {
 			}
 		}
 	}
+	if err := TelemetrySpecsAreValid(r, clientCache); err != nil {
+		return err
+	}
+	return nil
+}
+
+// clientGetter is an interface here purely to enable DI. It is anything that implements `GetRemoteClient()` mechanism to get remote clients by name.
+type clientGetter interface {
+	GetRemoteClient(k8sContextName string) (client.Client, error)
+}
+
+func TelemetrySpecsAreValid(kCluster *K8ssandraCluster, cGetter clientGetter) error {
+	// Iterate through the DCs, checking Stargate and Cassandra telemetry
+	if kCluster.Spec.Cassandra.Datacenters != nil {
+		for _, dc := range kCluster.Spec.Cassandra.Datacenters {
+			//Get client and determine if prom installed.
+			dcClient, err := cGetter.GetRemoteClient(dc.K8sContext)
+			if err != nil {
+				return err
+			}
+			promInstalled, err := validationpkg.IsPromInstalled(dcClient, webhookLog)
+			if err != nil {
+				return err
+			}
+
+			cassToValidate := &telemetryapi.TelemetrySpec{}
+			if kCluster.Spec.Cassandra.Telemetry != nil {
+				cassToValidate = kCluster.Spec.Cassandra.Telemetry.Merge(dc.Telemetry)
+			} else if dc.Telemetry != nil {
+				cassToValidate = dc.Telemetry
+			}
+			if cassToValidate != nil {
+				webhookLog.Info("validating cass telemetry in webhook", "cassToValidate", cassToValidate)
+				cassIsValid, err := validationpkg.TelemetrySpecIsValid(cassToValidate, promInstalled)
+				if err != nil {
+					return err
+				}
+				if !cassIsValid {
+					webhookLog.Info("throwing an error, cass telemetry spec invalid")
+					return errors.New(fmt.Sprint("Cassandra telemetry specification was incorrect in context", dc.K8sContext))
+				}
 
+			}
+
+			sgToValidate := &telemetryapi.TelemetrySpec{}
+			if kCluster.Spec.Stargate != nil && kCluster.Spec.Stargate.Telemetry != nil {
+				if dc.Stargate != nil {
+					sgToValidate = kCluster.Spec.Stargate.Telemetry.Merge(dc.Stargate.Telemetry)
+				} else {
+					sgToValidate = kCluster.Spec.Stargate.Telemetry
+				}
+			} else if dc.Stargate != nil && dc.Stargate.Telemetry != nil {
+				sgToValidate = dc.Stargate.Telemetry
+			}
+			if sgToValidate != nil {
+				sgIsValid, err := validationpkg.TelemetrySpecIsValid(sgToValidate, promInstalled)
+				if err != nil {
+					return err
+				}
+				if !sgIsValid {
+					return errors.New(fmt.Sprint("Stargate telemetry specification was incorrect in context", dc.K8sContext))
+				}
+			}
+		}
+	}
+	// Deal with the case that datacenters is completely undefined and the above for loop is empty.
+	if len(kCluster.Spec.Cassandra.Datacenters) == 0 {
+		localClient, err := cGetter.GetRemoteClient("")
+		if err != nil {
+			return err
+		}
+		promInstalled, err := validationpkg.IsPromInstalled(localClient, webhookLog)
+		if err != nil {
+			return err
+		}
+		if kCluster.Spec.Cassandra.Telemetry != nil {
+			cassIsValid, err := validationpkg.TelemetrySpecIsValid(kCluster.Spec.Cassandra.Telemetry, promInstalled)
+			if err != nil {
+				return err
+			}
+			if !cassIsValid {
+				return errors.New("Cassandra telemetry specification was incorrect in control plane")
+			}
+		}
+		if kCluster.Spec.Stargate != nil && kCluster.Spec.Stargate.Telemetry != nil {
+			sgIsValid, err := validationpkg.TelemetrySpecIsValid(kCluster.Spec.Stargate.Telemetry, promInstalled)
+			if err != nil {
+				return err
+			}
+			if !sgIsValid {
+				return errors.New("Stargate telemetry specification was incorrect in control plane")
+			}
+		}
+
+	}
 	return nil
 }
 
 // ValidateUpdate implements webhook.Validator so a webhook will be registered for the type
 func (r *K8ssandraCluster) ValidateUpdate(old runtime.Object) error {
 	webhookLog.Info("validate K8ssandraCluster update", "K8ssandraCluster", r.Name)
-
 	if err := r.validateK8ssandraCluster(); err != nil {
 		return err
 	}

apis/telemetry/v1alpha1/telemetry_methods.go

@@ -36,11 +36,17 @@ func (a *PrometheusTelemetrySpec) Merge(b *PrometheusTelemetrySpec) *PrometheusT
 	}
 	if a.CommonLabels != nil {
 		for k, v := range a.CommonLabels {
+			if out.CommonLabels == nil {
+				out.CommonLabels = make(map[string]string)
+			}
 			out.CommonLabels[k] = v
 		}
 	}
 	if b.CommonLabels != nil {
 		for k, v := range b.CommonLabels {
+			if out.CommonLabels == nil {
+				out.CommonLabels = make(map[string]string)
+			}
 			out.CommonLabels[k] = v
 		}
 	}

controllers/k8ssandra/cassandra_telemetry_reconciler.go

@@ -5,6 +5,7 @@ package k8ssandra
 import (
 	"context"
 	"errors"
+	"github.com/k8ssandra/k8ssandra-operator/pkg/validation"
 
 	"github.com/go-logr/logr"
 	cassdcapi "github.com/k8ssandra/cass-operator/apis/cassandra/v1beta1"
@@ -34,11 +35,11 @@ func (r *K8ssandraClusterReconciler) reconcileCassandraDCTelemetry(
 	}
 	logger.Info("merged TelemetrySpec constructed", "mergedSpec", mergedSpec, "cluster", kc.Name)
 	// Confirm telemetry config is valid (e.g. Prometheus is installed if it is requested.)
-	promInstalled, err := telemetry.IsPromInstalled(remoteClient, logger)
+	promInstalled, err := validation.IsPromInstalled(remoteClient, logger)
 	if err != nil {
 		return result.Error(err)
 	}
-	validConfig, err := telemetry.SpecIsValid(mergedSpec, promInstalled)
+	validConfig, err := validation.TelemetrySpecIsValid(mergedSpec, promInstalled)
 	if err != nil {
 		return result.Error(errors.New("could not determine if telemetry config is valid"))
 	}

controllers/k8ssandra/cassandra_telemetry_reconciler_test.go

@@ -30,7 +30,7 @@ func Test_reconcileCassandraDCTelemetry_TracksNamespaces(t *testing.T) {
 	// Test fixtures
 	r := newDummyK8ssandraClusterReconciler()
 	ctx := context.Background()
-	fakeClient := test.NewFakeClientWRestMapper()
+	fakeClient, _ := test.NewFakeClientWithProm()
 	testLogger := testlogr.NewTestLogger(t)
 	// Resources to create
 	cfg := telemetry.PrometheusResourcer{

controllers/k8ssandra/stop_dc_test.go

@@ -176,17 +176,21 @@ func stopExistingDc(t *testing.T, f *framework.Framework, ctx context.Context, k
 
 	replication := map[string]int{"dc1": 3, "dc2": 3}
 	stopDcManagementApiReset(replication)
-
 	t.Log("stop dc1")
-	patch := client.MergeFromWithOptions(kc.DeepCopy(), client.MergeFromWithOptimisticLock{})
-	kc.Spec.Cassandra.Datacenters[0].Stopped = true
-	err := f.Client.Patch(ctx, kc, patch)
-	require.NoError(t, err, "failed to patch kc")
+	require.Eventually(t, func() bool {
+		kc = &api.K8ssandraCluster{}
+		f.Client.Get(ctx, kcKey, kc)
+		patch := client.MergeFromWithOptions(kc.DeepCopy(), client.MergeFromWithOptimisticLock{})
+		kc.Spec.Cassandra.Datacenters[0].Stopped = true
+		err := f.Client.Patch(ctx, kc, patch)
+		t.Log(err)
+		return err == nil
+	}, timeout, interval, "failed to patch dc1 status to stopped")
 	withDc1 := f.NewWithDatacenter(ctx, dc1Key)
 	require.Eventually(t, withDc1(func(dc1 *cassdcapi.CassandraDatacenter) bool {
 		return assert.True(t, dc1.Spec.Stopped)
 	}), timeout, interval, "timeout waiting for dc1 to be stopped")
-	err = f.SetDatacenterStatusStopped(ctx, dc1Key)
+	err := f.SetDatacenterStatusStopped(ctx, dc1Key)
 	require.NoError(t, err, "failed to set dc1 status stopped")
 
 	t.Log("wait for the dc conditions to be met")
@@ -211,7 +215,7 @@ func stopExistingDc(t *testing.T, f *framework.Framework, ctx context.Context, k
 	require.Eventually(t, f.ReaperExists(ctx, reaper2Key), timeout, interval, "failed to verify reaper reaper2 created")
 
 	t.Log("start dc1")
-	patch = client.MergeFromWithOptions(kc.DeepCopy(), client.MergeFromWithOptimisticLock{})
+	patch := client.MergeFromWithOptions(kc.DeepCopy(), client.MergeFromWithOptimisticLock{})
 	kc.Spec.Cassandra.Datacenters[0].Stopped = false
 	err = f.Client.Patch(ctx, kc, patch)
 	require.NoError(t, err, "failed to patch kc")

controllers/stargate/stargate_telemetry_reconciler.go

@@ -5,6 +5,7 @@ package stargate
 import (
 	"context"
 	"errors"
+	"github.com/k8ssandra/k8ssandra-operator/pkg/validation"
 
 	k8ssandraapi "github.com/k8ssandra/k8ssandra-operator/apis/k8ssandra/v1alpha1"
 
@@ -35,11 +36,11 @@ func (r *StargateReconciler) reconcileStargateTelemetry(
 		cfg.CommonLabels[k8ssandraapi.K8ssandraClusterNameLabel] = klusterName
 	}
 	// Confirm telemetry config is valid (e.g. Prometheus is installed if it is requested.)
-	promInstalled, err := telemetry.IsPromInstalled(remoteClient, logger)
+	promInstalled, err := validation.IsPromInstalled(remoteClient, logger)
 	if err != nil {
 		return ctrl.Result{}, err
 	}
-	validConfig, err := telemetry.SpecIsValid(thisStargate.Spec.Telemetry, promInstalled)
+	validConfig, err := validation.TelemetrySpecIsValid(thisStargate.Spec.Telemetry, promInstalled)
 	if err != nil {
 		return ctrl.Result{}, errors.New("could not determine if telemetry config is valid")
 	}

controllers/stargate/stargate_telemetry_reconciler_test.go

@@ -29,7 +29,7 @@ func Test_reconcileStargateTelemetry_succeeds(t *testing.T) {
 	// Test fixtures
 	r := newDummyK8ssandraClusterReconciler()
 	ctx := context.Background()
-	fakeClient := test.NewFakeClientWRestMapper()
+	fakeClient, _ := test.NewFakeClientWithProm()
 	testLogger := testlogr.NewTestLogger(t)
 	// Resources to create
 	stargate := test.NewStargate("test-stargate", "test-stargate-namespace")

pkg/telemetry/prometheus_resourcer_test.go

@@ -19,10 +19,7 @@ import (
 
 // Test_PrometheusResourcer_UpdateResources_Create_CassDC tests that a serviceMonitor is created if one does not exist.
 func Test_PrometheusResourcer_UpdateResources_Create_CassDC(t *testing.T) {
-	fakeClient, err := test.NewFakeClient()
-	if err != nil {
-		assert.Fail(t, "could not create fake client", err)
-	}
+	fakeClient, _ := test.NewFakeClientWithProm()
 	ctx := context.Background()
 	// Create k8ssandra cluster and pass through to PrometheusResourcer.UpdateResources()
 	logger := testlogr.NewTestLogger(t)
@@ -54,7 +51,7 @@ func Test_PrometheusResourcer_UpdateResources_Create_CassDC(t *testing.T) {
 // Test_PrometheusResourcer_Cleanup_CassDC tests that the Cleanup method on PrometheusResourcer correctly
 // cleans up all resources it creates using its UpdateResources method.
 func Test_PrometheusResourcer_Cleanup_CassDC(t *testing.T) {
-	fakeClient, err := test.NewFakeClient()
+	fakeClient, err := test.NewFakeClientWithProm()
 	if err != nil {
 		assert.Fail(t, "could not create fake client", err)
 	}