Repository: k8ssandra/k8ssandra-operator (go)

Issue:
K8SSAND-1568 ⁃ Add CDC integration to k8ssandra-operator
Once we have CDC integrated into cass-operator, we should ensure that the relevant API fields are exposed via k8ssandra-operator also.

(I was hoping to have this under epic [K8SSAND-1547](https://app.zenhub.com/workspace/o/k8ssandra/cass-operator/issues/343) but it lives in cass-operator, and you can't span an epic over multiple repos.)

┆Issue is synchronized with this [Jira Task](https://k8ssandra.atlassian.net/browse/K8SSAND-1568) by [Unito](https://www.unito.io)
┆friendlyId: K8SSAND-1568
┆priority: Medium


Pull request description:
Integrate CDC into k8ssandra-operator reconciliation logic and APIs
**What this PR does**:

What it says on the tin.

Note that I've modified go.mod to use my recent changes to cass-operator. We need to revert this back to using a non-snapshot release prior to merge, but this is required because the cass-operator changes aren't yet merged.

**Which issue(s) this PR fixes**:
Fixes #570 

**Checklist**
- [ ] Changes manually tested
- [ ] Automated Tests added/updated
- [ ] Documentation added/updated
- [ ] CHANGELOG.md updated (not required for documentation PRs)
- [ ] CLA Signed:  [DataStax CLA](https://cla.datastax.com/)


Patch:
diff --git a/CHANGELOG/CHANGELOG-1.2.md b/CHANGELOG/CHANGELOG-1.2.md
index 80f1628b..f3ada820 100644
--- a/CHANGELOG/CHANGELOG-1.2.md
+++ b/CHANGELOG/CHANGELOG-1.2.md
@@ -15,6 +15,7 @@ When cutting a new release, update the `unreleased` heading to the tag being gen
 
 ## unreleased
 
+* [CHANGE] Update to Reaper v3.2.0
 * [CHANGE] Update to cass-operator v1.12.0
 * [CHANGE] Update to Medusa v0.13.3
 * [FEATURE] [#620](https://github.com/k8ssandra/k8ssandra-operator/issues/620) Enable injecting volumes in the Cassandra pods
@@ -22,6 +23,7 @@ When cutting a new release, update the `unreleased` heading to the tag being gen
 * [FEATURE] [#496](https://github.com/k8ssandra/k8ssandra-operator/issues/496) Allow overriding Cassandra cluster names
 * [FEATURE] [#460](https://github.com/k8ssandra/k8ssandra-operator/pull/460) Add support for scheduling backups.
 * [FEATURE] [#531](https://github.com/k8ssandra/k8ssandra-operator/issues/531) Add support for JWT authentication in Stargate
+* [ENHANCEMENT] [#578](https://github.com/k8ssandra/k8ssandra-operator/issues/578) Expose Reaper Metrics through a ServiceMonitor
 * [ENHANCEMENT] [#553](https://github.com/k8ssandra/k8ssandra-operator/issues/531) Add resources config settings for Reaper and Medusa containers
 * [ENHANCEMENT] [#573](https://github.com/k8ssandra/k8ssandra-operator/issues/573) Add the ability to set metrics filters in MCAC
 * [ENHANCEMENT] [#584](https://github.com/k8ssandra/k8ssandra-operator/issues/584) Make Medusa use the Cassandra data PV for restore downloads
diff --git a/apis/k8ssandra/v1alpha1/k8ssandracluster_types.go b/apis/k8ssandra/v1alpha1/k8ssandracluster_types.go
index eccfda3c..7bcac6f1 100644
--- a/apis/k8ssandra/v1alpha1/k8ssandracluster_types.go
+++ b/apis/k8ssandra/v1alpha1/k8ssandracluster_types.go
@@ -314,6 +314,11 @@ type DatacenterOptions struct {
 	// +optional
 	Telemetry *telemetryapi.TelemetrySpec `json:"telemetry,omitempty"`
 
+	// CDC defines the desired state for CDC integrations. It can be used to feed mutation events from Cassandra into an Apache Pulsar cluster,
+	// from where they can be expored to external systems.
+	// +optional
+	CDC *cassdcapi.CDCConfiguration `json:"cdc,omitempty"`
+
 	// Containers defines containers to be deployed in each Cassandra pod.
 	// K8ssandra-operator and cass-operator will create their own containers, which can be referenced here to override specific settings,
 	// such as mounts or resources request/limits for example.
diff --git a/apis/k8ssandra/v1alpha1/zz_generated.deepcopy.go b/apis/k8ssandra/v1alpha1/zz_generated.deepcopy.go
index b782c7ae..f56eeb82 100644
--- a/apis/k8ssandra/v1alpha1/zz_generated.deepcopy.go
+++ b/apis/k8ssandra/v1alpha1/zz_generated.deepcopy.go
@@ -1432,6 +1432,11 @@ func (in *DatacenterOptions) DeepCopyInto(out *DatacenterOptions) {
 		*out = new(telemetryv1alpha1.TelemetrySpec)
 		(*in).DeepCopyInto(*out)
 	}
+	if in.CDC != nil {
+		in, out := &in.CDC, &out.CDC
+		*out = new(v1beta1.CDCConfiguration)
+		(*in).DeepCopyInto(*out)
+	}
 	if in.Containers != nil {
 		in, out := &in.Containers, &out.Containers
 		*out = make([]v1.Container, len(*in))
diff --git a/apis/reaper/v1alpha1/reaper_types.go b/apis/reaper/v1alpha1/reaper_types.go
index 3cd49758..bb46194c 100644
--- a/apis/reaper/v1alpha1/reaper_types.go
+++ b/apis/reaper/v1alpha1/reaper_types.go
@@ -17,6 +17,7 @@ limitations under the License.
 package v1alpha1
 
 import (
+	telemetryapi "github.com/k8ssandra/k8ssandra-operator/apis/telemetry/v1alpha1"
 	"github.com/k8ssandra/k8ssandra-operator/pkg/encryption"
 	"github.com/k8ssandra/k8ssandra-operator/pkg/images"
 	corev1 "k8s.io/api/core/v1"
@@ -61,15 +62,15 @@ type ReaperTemplate struct {
 	UiUserSecretRef corev1.LocalObjectReference `json:"uiUserSecretRef,omitempty"`
 
 	// The image to use for the Reaper pod main container.
-	// The default is "thelastpickle/cassandra-reaper:3.1.1".
+	// The default is "thelastpickle/cassandra-reaper:3.2.0".
 	// +optional
-	// +kubebuilder:default={repository:"thelastpickle",name:"cassandra-reaper",tag:"3.1.1"}
+	// +kubebuilder:default={repository:"thelastpickle",name:"cassandra-reaper",tag:"3.2.0"}
 	ContainerImage *images.Image `json:"containerImage,omitempty"`
 
 	// The image to use for the Reaper pod init container (that performs schema migrations).
-	// The default is "thelastpickle/cassandra-reaper:3.1.1".
+	// The default is "thelastpickle/cassandra-reaper:3.2.0".
 	// +optional
-	// +kubebuilder:default={repository:"thelastpickle",name:"cassandra-reaper",tag:"3.1.1"}
+	// +kubebuilder:default={repository:"thelastpickle",name:"cassandra-reaper",tag:"3.2.0"}
 	InitContainerImage *images.Image `json:"initContainerImage,omitempty"`
 
 	// +kubebuilder:default="default"
@@ -124,6 +125,11 @@ type ReaperTemplate struct {
 	// Init Container resources.
 	// +optional
 	InitContainerResources *corev1.ResourceRequirements `json:"initContainerResources,omitempty"`
+
+	// Telemetry defines the desired telemetry integrations to deploy targeting the Reaper pods for all DCs in this cluster
+	// (unless overriden by DC specific settings)
+	// +optional
+	Telemetry *telemetryapi.TelemetrySpec `json:"telemetry,omitempty"`
 }
 
 // AutoScheduling includes options to configure the auto scheduling of repairs for new clusters.
diff --git a/apis/reaper/v1alpha1/zz_generated.deepcopy.go b/apis/reaper/v1alpha1/zz_generated.deepcopy.go
index 1110c80b..0edf8afb 100644
--- a/apis/reaper/v1alpha1/zz_generated.deepcopy.go
+++ b/apis/reaper/v1alpha1/zz_generated.deepcopy.go
@@ -22,6 +22,7 @@ limitations under the License.
 package v1alpha1
 
 import (
+	telemetryv1alpha1 "github.com/k8ssandra/k8ssandra-operator/apis/telemetry/v1alpha1"
 	"github.com/k8ssandra/k8ssandra-operator/pkg/encryption"
 	"github.com/k8ssandra/k8ssandra-operator/pkg/images"
 	"k8s.io/api/core/v1"
@@ -275,6 +276,11 @@ func (in *ReaperTemplate) DeepCopyInto(out *ReaperTemplate) {
 		*out = new(v1.ResourceRequirements)
 		(*in).DeepCopyInto(*out)
 	}
+	if in.Telemetry != nil {
+		in, out := &in.Telemetry, &out.Telemetry
+		*out = new(telemetryv1alpha1.TelemetrySpec)
+		(*in).DeepCopyInto(*out)
+	}
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ReaperTemplate.
diff --git a/apis/telemetry/v1alpha1/telemetry_types.go b/apis/telemetry/v1alpha1/telemetry_types.go
index a6d203f0..2c00ca29 100644
--- a/apis/telemetry/v1alpha1/telemetry_types.go
+++ b/apis/telemetry/v1alpha1/telemetry_types.go
@@ -16,14 +16,22 @@ type PrometheusTelemetrySpec struct {
 
 type McacTelemetrySpec struct {
 	// MetricFilters allows passing filters to MCAC in order to reduce the amount of extracted metrics.
-	// Not setting this field will result in the default filters being used.
-	// Setting it to an empty list will result in all metrics being extracted.
-	// Examples:
+	// Not setting this field will result in the default filters being used:
 	// - "deny:org.apache.cassandra.metrics.Table"
 	// - "deny:org.apache.cassandra.metrics.table"
 	// - "allow:org.apache.cassandra.metrics.table.live_ss_table_count"
 	// - "allow:org.apache.cassandra.metrics.Table.LiveSSTableCount"
 	// - "allow:org.apache.cassandra.metrics.table.live_disk_space_used"
+	// - "allow:org.apache.cassandra.metrics.table.LiveDiskSpaceUsed"
+	// - "allow:org.apache.cassandra.metrics.Table.Pending"
+	// - "allow:org.apache.cassandra.metrics.Table.Memtable"
+	// - "allow:org.apache.cassandra.metrics.Table.Compaction"
+	// - "allow:org.apache.cassandra.metrics.table.read"
+	// - "allow:org.apache.cassandra.metrics.table.write"
+	// - "allow:org.apache.cassandra.metrics.table.range"
+	// - "allow:org.apache.cassandra.metrics.table.coordinator"
+	// - "allow:org.apache.cassandra.metrics.table.dropped_mutations"
+	// Setting it to an empty list will result in all metrics being extracted.
 	// +optional
 	MetricFilters *[]string `json:"metricFilters,omitempty"`
 }
diff --git a/config/crd/bases/k8ssandra.io_k8ssandraclusters.yaml b/config/crd/bases/k8ssandra.io_k8ssandraclusters.yaml
index 2dc0c60b..08099dce 100644
--- a/config/crd/bases/k8ssandra.io_k8ssandraclusters.yaml
+++ b/config/crd/bases/k8ssandra.io_k8ssandraclusters.yaml
@@ -67,6 +67,60 @@ spec:
                     items:
                       type: string
                     type: array
+                  cdc:
+                    description: CDC defines the desired state for CDC integrations.
+                      It can be used to feed mutation events from Cassandra into an
+                      Apache Pulsar cluster, from where they can be expored to external
+                      systems.
+                    properties:
+                      cdcConcurrentProcessors:
+                        type: integer
+                      cdcPollIntervalM:
+                        type: integer
+                      cdcWorkingDir:
+                        type: string
+                      errorCommitLogReprocessEnabled:
+                        type: boolean
+                      pulsarAuthParams:
+                        type: string
+                      pulsarAuthPluginClassName:
+                        type: string
+                      pulsarBatchDelayInMs:
+                        type: integer
+                      pulsarKeyBasedBatcher:
+                        type: boolean
+                      pulsarMaxPendingMessages:
+                        type: integer
+                      pulsarMaxPendingMessagesAcrossPartitions:
+                        type: integer
+                      pulsarServiceUrl:
+                        minLength: 1
+                        type: string
+                      sslAllowInsecureConnection:
+                        type: string
+                      sslCipherSuites:
+                        type: string
+                      sslEnabledProtocols:
+                        type: string
+                      sslHostnameVerificationEnable:
+                        type: string
+                      sslKeystorePassword:
+                        type: string
+                      sslKeystorePath:
+                        type: string
+                      sslProvider:
+                        type: string
+                      sslTruststorePassword:
+                        type: string
+                      sslTruststorePath:
+                        type: string
+                      sslTruststoreType:
+                        type: string
+                      topicPrefix:
+                        type: string
+                    required:
+                    - pulsarServiceUrl
+                    type: object
                   clientEncryptionStores:
                     description: Client encryption stores which are used by Cassandra
                       and Reaper.
@@ -2834,6 +2888,60 @@ spec:
                     description: Datacenters a list of the DCs in the cluster.
                     items:
                       properties:
+                        cdc:
+                          description: CDC defines the desired state for CDC integrations.
+                            It can be used to feed mutation events from Cassandra
+                            into an Apache Pulsar cluster, from where they can be
+                            expored to external systems.
+                          properties:
+                            cdcConcurrentProcessors:
+                              type: integer
+                            cdcPollIntervalM:
+                              type: integer
+                            cdcWorkingDir:
+                              type: string
+                            errorCommitLogReprocessEnabled:
+                              type: boolean
+                            pulsarAuthParams:
+                              type: string
+                            pulsarAuthPluginClassName:
+                              type: string
+                            pulsarBatchDelayInMs:
+                              type: integer
+                            pulsarKeyBasedBatcher:
+                              type: boolean
+                            pulsarMaxPendingMessages:
+                              type: integer
+                            pulsarMaxPendingMessagesAcrossPartitions:
+                              type: integer
+                            pulsarServiceUrl:
+                              minLength: 1
+                              type: string
+                            sslAllowInsecureConnection:
+                              type: string
+                            sslCipherSuites:
+                              type: string
+                            sslEnabledProtocols:
+                              type: string
+                            sslHostnameVerificationEnable:
+                              type: string
+                            sslKeystorePassword:
+                              type: string
+                            sslKeystorePath:
+                              type: string
+                            sslProvider:
+                              type: string
+                            sslTruststorePassword:
+                              type: string
+                            sslTruststorePath:
+                              type: string
+                            sslTruststoreType:
+                              type: string
+                            topicPrefix:
+                              type: string
+                          required:
+                          - pulsarServiceUrl
+                          type: object
                         config:
                           description: CassandraConfig is configuration settings that
                             are applied to cassandra.yaml and the various jvm*.options
@@ -17996,9 +18104,9 @@ spec:
                     default:
                       name: cassandra-reaper
                       repository: thelastpickle
-                      tag: 3.1.1
+                      tag: 3.2.0
                     description: The image to use for the Reaper pod main container.
-                      The default is "thelastpickle/cassandra-reaper:3.1.1".
+                      The default is "thelastpickle/cassandra-reaper:3.2.0".
                     properties:
                       name:
                         description: The image name to use.
@@ -18054,9 +18162,9 @@ spec:
                     default:
                       name: cassandra-reaper
                       repository: thelastpickle
-                      tag: 3.1.1
+                      tag: 3.2.0
                     description: The image to use for the Reaper pod init container
-                      (that performs schema migrations). The default is "thelastpickle/cassandra-reaper:3.1.1".
+                      (that performs schema migrations). The default is "thelastpickle/cassandra-reaper:3.2.0".
                     properties:
                       name:
                         description: The image name to use.
@@ -18974,6 +19082,40 @@ spec:
                             type: string
                         type: object
                     type: object
+                  telemetry:
+                    description: Telemetry defines the desired telemetry integrations
+                      to deploy targeting the Reaper pods for all DCs in this cluster
+                      (unless overriden by DC specific settings)
+                    properties:
+                      mcac:
+                        properties:
+                          metricFilters:
+                            description: 'MetricFilters allows passing filters to
+                              MCAC in order to reduce the amount of extracted metrics.
+                              Not setting this field will result in the default filters
+                              being used. Setting it to an empty list will result
+                              in all metrics being extracted. Examples: - "deny:org.apache.cassandra.metrics.Table"
+                              - "deny:org.apache.cassandra.metrics.table" - "allow:org.apache.cassandra.metrics.table.live_ss_table_count"
+                              - "allow:org.apache.cassandra.metrics.Table.LiveSSTableCount"
+                              - "allow:org.apache.cassandra.metrics.table.live_disk_space_used"'
+                            items:
+                              type: string
+                            type: array
+                        type: object
+                      prometheus:
+                        properties:
+                          commonLabels:
+                            additionalProperties:
+                              type: string
+                            description: CommonLabels are applied to all serviceMonitors
+                              created.
+                            type: object
+                          enabled:
+                            description: Enable the creation of Prometheus serviceMonitors
+                              for this resource (Cassandra or Stargate).
+                            type: boolean
+                        type: object
+                    type: object
                   tolerations:
                     description: Tolerations applied to the Reaper pods.
                     items:
diff --git a/config/crd/bases/reaper.k8ssandra.io_reapers.yaml b/config/crd/bases/reaper.k8ssandra.io_reapers.yaml
index cde7b0a0..a33711af 100644
--- a/config/crd/bases/reaper.k8ssandra.io_reapers.yaml
+++ b/config/crd/bases/reaper.k8ssandra.io_reapers.yaml
@@ -998,9 +998,9 @@ spec:
                 default:
                   name: cassandra-reaper
                   repository: thelastpickle
-                  tag: 3.1.1
+                  tag: 3.2.0
                 description: The image to use for the Reaper pod main container. The
-                  default is "thelastpickle/cassandra-reaper:3.1.1".
+                  default is "thelastpickle/cassandra-reaper:3.2.0".
                 properties:
                   name:
                     description: The image name to use.
@@ -1082,9 +1082,9 @@ spec:
                 default:
                   name: cassandra-reaper
                   repository: thelastpickle
-                  tag: 3.1.1
+                  tag: 3.2.0
                 description: The image to use for the Reaper pod init container (that
-                  performs schema migrations). The default is "thelastpickle/cassandra-reaper:3.1.1".
+                  performs schema migrations). The default is "thelastpickle/cassandra-reaper:3.2.0".
                 properties:
                   name:
                     description: The image name to use.
@@ -1972,6 +1972,40 @@ spec:
                   already up-to-date, or if you know upfront that QUORUM cannot be
                   achieved (for example, because a DC is down).
                 type: boolean
+              telemetry:
+                description: Telemetry defines the desired telemetry integrations
+                  to deploy targeting the Reaper pods for all DCs in this cluster
+                  (unless overriden by DC specific settings)
+                properties:
+                  mcac:
+                    properties:
+                      metricFilters:
+                        description: 'MetricFilters allows passing filters to MCAC
+                          in order to reduce the amount of extracted metrics. Not
+                          setting this field will result in the default filters being
+                          used. Setting it to an empty list will result in all metrics
+                          being extracted. Examples: - "deny:org.apache.cassandra.metrics.Table"
+                          - "deny:org.apache.cassandra.metrics.table" - "allow:org.apache.cassandra.metrics.table.live_ss_table_count"
+                          - "allow:org.apache.cassandra.metrics.Table.LiveSSTableCount"
+                          - "allow:org.apache.cassandra.metrics.table.live_disk_space_used"'
+                        items:
+                          type: string
+                        type: array
+                    type: object
+                  prometheus:
+                    properties:
+                      commonLabels:
+                        additionalProperties:
+                          type: string
+                        description: CommonLabels are applied to all serviceMonitors
+                          created.
+                        type: object
+                      enabled:
+                        description: Enable the creation of Prometheus serviceMonitors
+                          for this resource (Cassandra or Stargate).
+                        type: boolean
+                    type: object
+                type: object
               tolerations:
                 description: Tolerations applied to the Reaper pods.
                 items:
diff --git a/controllers/reaper/reaper_controller.go b/controllers/reaper/reaper_controller.go
index 6101d1f8..042c92b8 100644
--- a/controllers/reaper/reaper_controller.go
+++ b/controllers/reaper/reaper_controller.go
@@ -209,6 +209,11 @@ func (r *ReaperReconciler) reconcileDeployment(
 		}
 	}
 
+	if err = r.reconcileReaperTelemetry(ctx, actualReaper, logger, r.Client); err != nil {
+		logger.Error(err, "reconcileReaperTelemetry failed")
+		return ctrl.Result{}, err
+	}
+
 	actualDeployment = actualDeployment.DeepCopy()
 
 	// Check if the deployment needs to be updated
diff --git a/controllers/reaper/reaper_telemetry_reconciler.go b/controllers/reaper/reaper_telemetry_reconciler.go
new file mode 100644
index 00000000..f7b8f0f5
--- /dev/null
+++ b/controllers/reaper/reaper_telemetry_reconciler.go
@@ -0,0 +1,101 @@
+/*
+
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package reaper
+
+import (
+	"context"
+	"errors"
+
+	k8ssandraapi "github.com/k8ssandra/k8ssandra-operator/apis/k8ssandra/v1alpha1"
+
+	reaperapi "github.com/k8ssandra/k8ssandra-operator/apis/reaper/v1alpha1"
+
+	"github.com/go-logr/logr"
+	"github.com/k8ssandra/k8ssandra-operator/pkg/telemetry"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+func (r *ReaperReconciler) reconcileReaperTelemetry(
+	ctx context.Context,
+	thisReaper *reaperapi.Reaper,
+	logger logr.Logger,
+	remoteClient client.Client,
+) error {
+	logger.Info("reconciling telemetry", "reaper", thisReaper.Name)
+	cfg := telemetry.PrometheusResourcer{
+		MonitoringTargetNS:   thisReaper.Namespace,
+		MonitoringTargetName: thisReaper.Name,
+		ServiceMonitorName:   GetReaperPromSMName(thisReaper.Name),
+		Logger:               logger,
+		CommonLabels:         mustLabels(thisReaper.Name),
+	}
+	klusterName, ok := thisReaper.Labels[k8ssandraapi.K8ssandraClusterNameLabel]
+	if ok {
+		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)
+	if err != nil {
+		return err
+	}
+	validConfig, err := telemetry.SpecIsValid(thisReaper.Spec.Telemetry, promInstalled)
+	if err != nil {
+		return errors.New("could not determine if telemetry config is valid")
+	}
+	if !validConfig {
+		return errors.New("telemetry spec was invalid for this cluster - is Prometheus installed if you have requested it")
+	}
+	// If Prometheus not installed bail here.
+	if !promInstalled {
+		return nil
+	}
+	// If Reaper is attached
+	// Determine if we want a cleanup or a resource update.
+	switch {
+	case thisReaper.Spec.Telemetry == nil || thisReaper.Spec.Telemetry.Prometheus == nil:
+		logger.Info("Telemetry not present for Reaper, will delete resources", "TelemetrySpec", thisReaper.Spec.Telemetry)
+		if err := cfg.CleanupResources(ctx, remoteClient); err != nil {
+			return err
+		}
+	case thisReaper.Spec.Telemetry.Prometheus.Enabled:
+		logger.Info("Prometheus config found", "TelemetrySpec", thisReaper.Spec.Telemetry)
+		desiredSM, err := cfg.NewReaperServiceMonitor()
+		if err != nil {
+			return err
+		}
+		if err := cfg.UpdateResources(ctx, remoteClient, thisReaper, &desiredSM); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// mustLabels() returns the set of labels essential to managing the Prometheus resources. These should not be overwritten by the user.
+func mustLabels(reaperName string) map[string]string {
+	return map[string]string{
+		k8ssandraapi.ManagedByLabel: k8ssandraapi.NameLabelValue,
+		k8ssandraapi.PartOfLabel:    k8ssandraapi.PartOfLabelValue,
+		reaperapi.ReaperLabel:       reaperName,
+		k8ssandraapi.ComponentLabel: k8ssandraapi.ComponentLabelTelemetry,
+		k8ssandraapi.CreatedByLabel: k8ssandraapi.CreatedByLabelValueK8ssandraClusterController,
+	}
+}
+
+// GetReaperPromSMName gets the name for our ServiceMonitors based on cluster and DC name.
+func GetReaperPromSMName(reaperName string) string {
+	return reaperName + "-" + "reaper-servicemonitor"
+}
diff --git a/go.sum b/go.sum
index bdcdbec8..fa8716d6 100644
--- a/go.sum
+++ b/go.sum
@@ -106,6 +106,8 @@ github.com/Microsoft/go-winio v0.4.14/go.mod h1:qXqCSQ3Xa7+6tgxaGTIe4Kpcdsi+P8jB
 github.com/Microsoft/go-winio v0.4.15/go.mod h1:tTuCMEN+UleMWgg9dVx4Hu52b1bJo+59jBh3ajtinzw=
 github.com/Microsoft/go-winio v0.4.17/go.mod h1:JPGBdM1cNvN/6ISo+n8V5iA4v8pBzdOpzfwIujj1a84=
 github.com/Microsoft/hcsshim v0.8.22/go.mod h1:91uVCVzvX2QD16sMCenoxxXo6L1wJnLMX2PSufFMtF0=
+github.com/Miles-Garnsey/cass-operator v1.8.0-rc.1.0.20220617033052-cec71ffc7fac h1:W1AVgA7mScYU26n69sA+VJJCo4NyREaCp1T+7bykgCU=
+github.com/Miles-Garnsey/cass-operator v1.8.0-rc.1.0.20220617033052-cec71ffc7fac/go.mod h1:vOjZ8NP9RNua8mbrZMkXFPBO9YbwVVfnJQIS8Yf8kHU=
 github.com/NYTimes/gziphandler v0.0.0-20170623195520-56545f4a5d46/go.mod h1:3wb06e3pkSAbeQ52E9H9iFoQsEEwGN64994WTCIhntQ=
 github.com/NYTimes/gziphandler v1.1.1/go.mod h1:n/CVRwUEOgIxrgPvAQhUUr9oeUtvrhMomdKFjzJNB0c=
 github.com/OneOfOne/xxhash v1.2.2/go.mod h1:HSdplMjZKSmBqAxg5vPj2TmRDmfkzw+cTzAElWljhcU=
diff --git a/pkg/cassandra/datacenter.go b/pkg/cassandra/datacenter.go
index 5a5163bc..d874e312 100644
--- a/pkg/cassandra/datacenter.go
+++ b/pkg/cassandra/datacenter.go
@@ -113,6 +113,7 @@ type DatacenterConfig struct {
 	Containers               []corev1.Container
 	InitContainers           []corev1.Container
 	ExtraVolumes             *api.K8ssandraVolumes
+	CDC                      *cassdcapi.CDCConfiguration
 }
 
 const (
@@ -175,6 +176,7 @@ func NewDatacenter(klusterKey types.NamespacedName, template *DatacenterConfig)
 			Users:               template.Users,
 			Networking:          template.Networking,
 			PodTemplateSpec:     template.PodTemplateSpec,
+			CDC:                 template.CDC,
 		},
 	}
 
@@ -334,6 +336,10 @@ func Coalesce(clusterName string, clusterTemplate *api.CassandraClusterTemplate,
 		dcConfig.MgmtAPIHeap = dcTemplate.DatacenterOptions.MgmtAPIHeap
 	}
 
+	if dcTemplate.CDC != nil {
+		dcConfig.CDC = dcTemplate.CDC
+	}
+
 	if dcTemplate.DatacenterOptions.SoftPodAntiAffinity == nil {
 		dcConfig.SoftPodAntiAffinity = clusterTemplate.DatacenterOptions.SoftPodAntiAffinity
 	} else {
diff --git a/pkg/reaper/deployment.go b/pkg/reaper/deployment.go
index 97cb222a..7b85c8e8 100644
--- a/pkg/reaper/deployment.go
+++ b/pkg/reaper/deployment.go
@@ -22,7 +22,7 @@ import (
 const (
 	DefaultImageRepository = "thelastpickle"
 	DefaultImageName       = "cassandra-reaper"
-	DefaultVersion         = "3.1.1"
+	DefaultVersion         = "3.2.0"
 	// When changing the default version above, please also change the kubebuilder markers in
 	// apis/reaper/v1alpha1/reaper_types.go accordingly.
 
diff --git a/pkg/telemetry/prom_reaper_servicemonitor.go b/pkg/telemetry/prom_reaper_servicemonitor.go
new file mode 100644
index 00000000..d20f6a21
--- /dev/null
+++ b/pkg/telemetry/prom_reaper_servicemonitor.go
@@ -0,0 +1,67 @@
+package telemetry
+
+import (
+	"fmt"
+	reaperapi "github.com/k8ssandra/k8ssandra-operator/apis/reaper/v1alpha1"
+	"os"
+
+	k8ssandraapi "github.com/k8ssandra/k8ssandra-operator/apis/k8ssandra/v1alpha1"
+	"github.com/k8ssandra/k8ssandra-operator/pkg/annotations"
+	promapi "github.com/prometheus-operator/prometheus-operator/pkg/apis/monitoring/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes/scheme"
+)
+
+// Static configuration for ServiceMonitor's endpoints.
+const reaperEndpointString = `
+apiVersion: monitoring.coreos.com/v1
+kind: ServiceMonitor
+spec:
+  endpoints:
+    - interval: 15s
+      path: /prometheusMetrics
+      port: health
+      scheme: http
+      scrapeTimeout: 15s
+`
+
+var reaperServiceMonitorTemplate = &promapi.ServiceMonitor{}
+
+func init() {
+	decode := scheme.Codecs.UniversalDeserializer().Decode
+	_, _, err := decode([]byte(reaperEndpointString), nil, reaperServiceMonitorTemplate)
+	if err != nil {
+		fmt.Println("Fatal error initialising EndpointHolder in pks/telemetry/prom_reaper_servicemonitor.go", err)
+		os.Exit(1)
+	}
+}
+
+// NewReaperServiceMonitor returns a Prometheus operator ServiceMonitor resource.
+func (cfg PrometheusResourcer) NewReaperServiceMonitor() (promapi.ServiceMonitor, error) {
+	// validate the object we're being passed.
+	if err := cfg.validate(); err != nil {
+		return promapi.ServiceMonitor{}, err
+	}
+	// Overwrite any CommonLabels the user has asked for if they conflict with the labels essential for the functioning of the operator.
+	sm := promapi.ServiceMonitor{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      cfg.ServiceMonitorName,
+			Namespace: cfg.MonitoringTargetNS,
+			Labels:    cfg.CommonLabels,
+		},
+		Spec: promapi.ServiceMonitorSpec{
+			Selector: metav1.LabelSelector{
+				MatchLabels: map[string]string{
+					k8ssandraapi.CreatedByLabel: k8ssandraapi.CreatedByLabelValueReaperController,
+					reaperapi.ReaperLabel:       cfg.MonitoringTargetName,
+				},
+			},
+			NamespaceSelector: promapi.NamespaceSelector{
+				MatchNames: []string{cfg.MonitoringTargetNS},
+			},
+			Endpoints: reaperServiceMonitorTemplate.Spec.Endpoints,
+		},
+	}
+	annotations.AddHashAnnotation(&sm)
+	return sm, nil
+}


Test patch:
diff --git a/controllers/k8ssandra/k8ssandracluster_controller_test.go b/controllers/k8ssandra/k8ssandracluster_controller_test.go
index c6304655..0262d103 100644
--- a/controllers/k8ssandra/k8ssandracluster_controller_test.go
+++ b/controllers/k8ssandra/k8ssandracluster_controller_test.go
@@ -676,6 +676,9 @@ func applyClusterTemplateAndDatacenterTemplateConfigs(t *testing.T, ctx context.
 									MaxHeapSize: parseQuantity("1024Mi"),
 								},
 							},
+							CDC: &cassdcapi.CDCConfiguration{
+								PulsarServiceUrl: pointer.String("pulsar://test-url"),
+							},
 						},
 					},
 				},
@@ -734,6 +737,7 @@ func applyClusterTemplateAndDatacenterTemplateConfigs(t *testing.T, ctx context.
 	assert.Equal(*kc.Spec.Cassandra.Datacenters[1].DatacenterOptions.StorageConfig, dc2.Spec.StorageConfig)
 	assert.Equal(kc.Spec.Cassandra.Datacenters[1].DatacenterOptions.Networking, dc2.Spec.Networking)
 	assert.Equal(dc2Size, dc2.Spec.Size)
+	assert.Equal(*dc2.Spec.CDC.PulsarServiceUrl, "pulsar://test-url")
 
 	actualConfig, err = gabs.ParseJSON(dc2.Spec.Config)
 	require.NoError(err, fmt.Sprintf("failed to parse dc2 config %s", dc2.Spec.Config))
diff --git a/controllers/reaper/reaper_telemetry_reconciler_test.go b/controllers/reaper/reaper_telemetry_reconciler_test.go
new file mode 100644
index 00000000..1a6412d7
--- /dev/null
+++ b/controllers/reaper/reaper_telemetry_reconciler_test.go
@@ -0,0 +1,56 @@
+// Some of the tests in this package target functions from pkg/telemetryapi, because they need to be envtests. We prefer to keep envtests in the controller
+// packages.
+package reaper
+
+import (
+	"context"
+	"testing"
+
+	k8ssandraapi "github.com/k8ssandra/k8ssandra-operator/apis/k8ssandra/v1alpha1"
+
+	testlogr "github.com/go-logr/logr/testing"
+	telemetryapi "github.com/k8ssandra/k8ssandra-operator/apis/telemetry/v1alpha1"
+	"github.com/k8ssandra/k8ssandra-operator/pkg/config"
+	"github.com/k8ssandra/k8ssandra-operator/pkg/telemetry"
+	"github.com/k8ssandra/k8ssandra-operator/pkg/test"
+	promapi "github.com/prometheus-operator/prometheus-operator/pkg/apis/monitoring/v1"
+	"k8s.io/apimachinery/pkg/types"
+
+	"github.com/stretchr/testify/assert"
+)
+
+// new_DummyK8ssandraClusterReconciler gives us back a `K8ssandraClusterReconciler` with just the fields that we need to test `reconcileCassandraDCTelemetry()`.
+func newDummyK8ssandraClusterReconciler() ReaperReconciler {
+	return ReaperReconciler{ReconcilerConfig: &config.ReconcilerConfig{DefaultDelay: interval}}
+}
+
+// Test_reconcilereaperTelemetry_TracksNamespaces tests that the servicemonitor is created
+func Test_reconcilereaperTelemetry_succeeds(t *testing.T) {
+	// Test fixtures
+	r := newDummyK8ssandraClusterReconciler()
+	ctx := context.Background()
+	fakeClient := test.NewFakeClientWRestMapper()
+	testLogger := testlogr.NewTestLogger(t)
+	// Resources to create
+	reaper := test.NewReaper("test-reaper", "test-reaper-namespace")
+	reaper.Spec.Telemetry = &telemetryapi.TelemetrySpec{
+		Prometheus: &telemetryapi.PrometheusTelemetrySpec{
+			Enabled: true,
+		},
+	}
+	cfg := telemetry.PrometheusResourcer{
+		MonitoringTargetNS:   reaper.Namespace,
+		MonitoringTargetName: reaper.Name,
+		Logger:               testLogger,
+		ServiceMonitorName:   GetReaperPromSMName(reaper.Name),
+		CommonLabels:         map[string]string{k8ssandraapi.K8ssandraClusterNameLabel: "test-cluster-name"},
+	}
+	if err := r.reconcileReaperTelemetry(ctx, &reaper, testLogger, fakeClient); err != nil {
+		assert.Fail(t, "reconciliation failed", err)
+	}
+	currentSM := &promapi.ServiceMonitor{}
+	if err := fakeClient.Get(ctx, types.NamespacedName{Name: cfg.ServiceMonitorName, Namespace: cfg.MonitoringTargetNS}, currentSM); err != nil {
+		assert.Fail(t, "could not get actual ServiceMonitor after reconciling k8ssandra cluster", err)
+	}
+	assert.NotEmpty(t, currentSM.Spec.Endpoints)
+}
diff --git a/pkg/cassandra/datacenter_test.go b/pkg/cassandra/datacenter_test.go
index 22809117..8fb61726 100644
--- a/pkg/cassandra/datacenter_test.go
+++ b/pkg/cassandra/datacenter_test.go
@@ -573,6 +573,19 @@ func TestNewDatacenter_Fail_NoServerVersion(t *testing.T) {
 	assert.IsType(t, DCConfigIncomplete{}, err)
 }
 
+func TestCDC(t *testing.T) {
+	template := GetDatacenterConfig()
+	template.CDC = &cassdcapi.CDCConfiguration{
+		PulsarServiceUrl: pointer.String("pulsar://test-url"),
+	}
+	cassDC, err := NewDatacenter(
+		types.NamespacedName{Name: "testdc", Namespace: "test-namespace"},
+		&template,
+	)
+	assert.NoError(t, err)
+	assert.Equal(t, cassDC.Spec.CDC, template.CDC)
+}
+
 func TestDatacentersReplication(t *testing.T) {
 	assert := assert.New(t)
 
diff --git a/pkg/reaper/deployment_test.go b/pkg/reaper/deployment_test.go
index 403c138a..747b139e 100644
--- a/pkg/reaper/deployment_test.go
+++ b/pkg/reaper/deployment_test.go
@@ -282,8 +282,8 @@ func TestImages(t *testing.T) {
 		reaper.Spec.InitContainerImage = nil
 		reaper.Spec.ContainerImage = nil
 		deployment := NewDeployment(reaper, newTestDatacenter(), nil, nil)
-		assert.Equal(t, "docker.io/thelastpickle/cassandra-reaper:3.1.1", deployment.Spec.Template.Spec.InitContainers[0].Image)
-		assert.Equal(t, "docker.io/thelastpickle/cassandra-reaper:3.1.1", deployment.Spec.Template.Spec.Containers[0].Image)
+		assert.Equal(t, "docker.io/thelastpickle/cassandra-reaper:3.2.0", deployment.Spec.Template.Spec.InitContainers[0].Image)
+		assert.Equal(t, "docker.io/thelastpickle/cassandra-reaper:3.2.0", deployment.Spec.Template.Spec.Containers[0].Image)
 		assert.Equal(t, corev1.PullIfNotPresent, deployment.Spec.Template.Spec.InitContainers[0].ImagePullPolicy)
 		assert.Equal(t, corev1.PullIfNotPresent, deployment.Spec.Template.Spec.Containers[0].ImagePullPolicy)
 		assert.Empty(t, deployment.Spec.Template.Spec.ImagePullSecrets)
@@ -297,8 +297,8 @@ func TestImages(t *testing.T) {
 		}
 		reaper.Spec.ContainerImage = nil
 		deployment := NewDeployment(reaper, newTestDatacenter(), nil, nil)
-		assert.Equal(t, "docker.io/thelastpickle/cassandra-reaper:3.1.1", deployment.Spec.Template.Spec.InitContainers[0].Image)
-		assert.Equal(t, "docker.io/thelastpickle/cassandra-reaper:3.1.1", deployment.Spec.Template.Spec.Containers[0].Image)
+		assert.Equal(t, "docker.io/thelastpickle/cassandra-reaper:3.2.0", deployment.Spec.Template.Spec.InitContainers[0].Image)
+		assert.Equal(t, "docker.io/thelastpickle/cassandra-reaper:3.2.0", deployment.Spec.Template.Spec.Containers[0].Image)
 		assert.Equal(t, corev1.PullIfNotPresent, deployment.Spec.Template.Spec.InitContainers[0].ImagePullPolicy)
 		assert.Equal(t, corev1.PullIfNotPresent, deployment.Spec.Template.Spec.Containers[0].ImagePullPolicy)
 		assert.Empty(t, deployment.Spec.Template.Spec.ImagePullSecrets)
diff --git a/pkg/telemetry/prom_reaper_servicemonitor_test.go b/pkg/telemetry/prom_reaper_servicemonitor_test.go
new file mode 100644
index 00000000..3c144bfc
--- /dev/null
+++ b/pkg/telemetry/prom_reaper_servicemonitor_test.go
@@ -0,0 +1,27 @@
+package telemetry
+
+import (
+	"testing"
+
+	testlogr "github.com/go-logr/logr/testing"
+	k8ssandraapi "github.com/k8ssandra/k8ssandra-operator/apis/k8ssandra/v1alpha1"
+
+	"github.com/stretchr/testify/assert"
+)
+
+//Test_NewServiceMonitor_SUCCESS tests that a new service monitor is successfully returned.
+func Test_PrometheusResourcer_NewReaperServiceMonitor_SUCCESS(t *testing.T) {
+	logger := testlogr.NewTestLogger(t)
+	cfg := PrometheusResourcer{
+		MonitoringTargetNS:   "test-namespace",
+		MonitoringTargetName: "test-dc-name",
+		Logger:               logger,
+		ServiceMonitorName:   "test-servicemonitor",
+		CommonLabels:         map[string]string{k8ssandraapi.K8ssandraClusterNameLabel: "test-cluster-name"},
+	}
+	actualSM, err := cfg.NewReaperServiceMonitor()
+	if err != nil {
+		assert.Fail(t, "error creating new service monitor", err)
+	}
+	assert.Equal(t, "health", actualSM.Spec.Endpoints[0].Port)
+}
diff --git a/pkg/test/test_objects.go b/pkg/test/test_objects.go
index 96d9a1d0..93211f1b 100644
--- a/pkg/test/test_objects.go
+++ b/pkg/test/test_objects.go
@@ -4,6 +4,7 @@ package test
 import (
 	cassdcapi "github.com/k8ssandra/cass-operator/apis/cassandra/v1beta1"
 	k8ssandraapi "github.com/k8ssandra/k8ssandra-operator/apis/k8ssandra/v1alpha1"
+	reaperapi "github.com/k8ssandra/k8ssandra-operator/apis/reaper/v1alpha1"
 	stargateapi "github.com/k8ssandra/k8ssandra-operator/apis/stargate/v1alpha1"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
@@ -97,3 +98,16 @@ func NewStargate(name string, namespace string) stargateapi.Stargate {
 		},
 	}
 }
+
+func NewReaper(name string, namespace string) reaperapi.Reaper {
+	return reaperapi.Reaper{
+		TypeMeta: metav1.TypeMeta{
+			APIVersion: "reaper.k8ssandra.io/v1alpha",
+			Kind:       "Reaper",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: namespace,
+		},
+	}
+}
diff --git a/test/kuttl/test-servicemonitors/03-assert.yaml b/test/kuttl/test-servicemonitors/03-assert.yaml
index 909b8b69..dec9529a 100644
--- a/test/kuttl/test-servicemonitors/03-assert.yaml
+++ b/test/kuttl/test-servicemonitors/03-assert.yaml
@@ -17,4 +17,10 @@ kind: ServiceMonitor
 metadata:
   name: test-dc1-stargate-stargate-servicemonitor
   namespace: k8ssandra-operator
+---
+apiVersion: monitoring.coreos.com/v1
+kind: ServiceMonitor
+metadata:
+  name: test-dc1-reaper-reaper-servicemonitor
+  namespace: k8ssandra-operator
   
\ No newline at end of file
diff --git a/test/kuttl/test-servicemonitors/config/single-node-k8ssandra/kustomization.yaml b/test/kuttl/test-servicemonitors/config/single-node-k8ssandra/kustomization.yaml
index cc43fe1a..f91508fc 100644
--- a/test/kuttl/test-servicemonitors/config/single-node-k8ssandra/kustomization.yaml
+++ b/test/kuttl/test-servicemonitors/config/single-node-k8ssandra/kustomization.yaml
@@ -3,7 +3,7 @@ kind: Kustomization
 namespace: k8ssandra-operator
 resources:
 - ../../../config/fixtures/k8ssandra-base
-# take basic deployment, add stargate, and add telemetry for both stargate and cassandra.
+# take basic deployment, add stargate and reaper, and add telemetry for stargate, cassandra and reaper.
 patches:
 - patch: |-
     - op: add
@@ -19,6 +19,12 @@ patches:
       value:
         prometheus:
             enabled: true
+    - op: add
+      path: /spec/reaper
+      value:
+        telemetry:
+          prometheus:
+            enabled: true
   target:
     kind: K8ssandraCluster
     name: test
\ No newline at end of file