core: report node metrics using ceph telemetry

Add this reporting with the cephcluster reconcile,
Similar way we reported other telemetry's

Closes: https://github.com/rook/rook/issues/12344

Signed-off-by: parth-gr <paarora@redhat.com>
(cherry picked from commit f007f2aca1)

pkg/operator/ceph/cluster/cluster.go

@@ -20,6 +20,7 @@ package cluster
 import (
 	"context"
 	"fmt"
+	"os"
 	"os/exec"
 	"path"
 	"strconv"
@@ -528,8 +529,14 @@ func (c *cluster) reportTelemetry() {
 	telemetryMutex.Lock()
 	defer telemetryMutex.Unlock()
 
-	// Identify this as a rook cluster for Ceph telemetry by setting the Rook version.
+	reportClusterTelemetry(c)
+	reportNodeTelemetry(c)
+}
+
+func reportClusterTelemetry(c *cluster) {
 	logger.Info("reporting cluster telemetry")
+
+	// Identify this as a rook cluster for Ceph telemetry by setting the Rook version.
 	telemetry.ReportKeyValue(c.context, c.ClusterInfo, telemetry.RookVersionKey, rookversion.Version)
 
 	// Report the K8s version
@@ -577,6 +584,58 @@ func (c *cluster) reportTelemetry() {
 	telemetry.ReportKeyValue(c.context, c.ClusterInfo, telemetry.ExternalModeEnabledKey, strconv.FormatBool(c.Spec.External.Enable))
 }
 
+func reportNodeTelemetry(c *cluster) {
+	logger.Info("reporting node telemetry")
+	operatorNamespace := os.Getenv(k8sutil.PodNamespaceEnvVar)
+
+	// Report the K8sNodeCount
+	nodelist, err := c.context.Clientset.CoreV1().Nodes().List(c.ClusterInfo.Context, metav1.ListOptions{})
+	if err != nil {
+		logger.Warningf("failed to report the K8s node count. %v", err)
+	} else {
+		telemetry.ReportKeyValue(c.context, c.ClusterInfo, telemetry.K8sNodeCount, strconv.Itoa(len(nodelist.Items)))
+	}
+
+	// Report the cephNodeCount
+	if c.Spec.CrashCollector.Disable {
+		telemetry.ReportKeyValue(c.context, c.ClusterInfo, telemetry.CephNodeCount, "-1")
+	} else {
+		listoption := metav1.ListOptions{LabelSelector: fmt.Sprintf("%s=%s", k8sutil.AppAttr, nodedaemon.CrashCollectorAppName)}
+		cephNodeList, err := c.context.Clientset.CoreV1().Pods(c.Namespace).List(c.ClusterInfo.Context, listoption)
+		if err != nil {
+			logger.Warningf("failed to report the ceph node count. %v", err)
+		} else {
+			telemetry.ReportKeyValue(c.context, c.ClusterInfo, telemetry.CephNodeCount, strconv.Itoa(len(cephNodeList.Items)))
+		}
+	}
+	// Report the csi rbd node count
+	listoption := metav1.ListOptions{LabelSelector: fmt.Sprintf("%s=%s", k8sutil.AppAttr, csi.CsiRBDPlugin)}
+	cephRbdNodelist, err := c.context.Clientset.CoreV1().Pods(operatorNamespace).List(c.ClusterInfo.Context, listoption)
+	if err != nil {
+		logger.Warningf("failed to report the ceph rbd node count. %v", err)
+	} else {
+		telemetry.ReportKeyValue(c.context, c.ClusterInfo, telemetry.RBDNodeCount, strconv.Itoa(len(cephRbdNodelist.Items)))
+	}
+
+	// Report the csi cephfs node count
+	listoption = metav1.ListOptions{LabelSelector: fmt.Sprintf("%s=%s", k8sutil.AppAttr, csi.CsiCephFSPlugin)}
+	cephFSNodelist, err := c.context.Clientset.CoreV1().Pods(operatorNamespace).List(c.ClusterInfo.Context, listoption)
+	if err != nil {
+		logger.Warningf("failed to report the ceph cephfs node count. %v", err)
+	} else {
+		telemetry.ReportKeyValue(c.context, c.ClusterInfo, telemetry.CephFSNodeCount, strconv.Itoa(len(cephFSNodelist.Items)))
+	}
+
+	// Report the csi nfs node count
+	listoption = metav1.ListOptions{LabelSelector: fmt.Sprintf("%s=%s", k8sutil.AppAttr, csi.CsiNFSPlugin)}
+	cephNFSNodelist, err := c.context.Clientset.CoreV1().Pods(operatorNamespace).List(c.ClusterInfo.Context, listoption)
+	if err != nil {
+		logger.Warningf("failed to report the ceph nfs node count. %v", err)
+	} else {
+		telemetry.ReportKeyValue(c.context, c.ClusterInfo, telemetry.NFSNodeCount, strconv.Itoa(len(cephNFSNodelist.Items)))
+	}
+}
+
 func (c *cluster) configureMsgr2() error {
 	encryptionSetting := "secure"
 	rbdMapOptions := "rbd_default_map_options"

pkg/operator/ceph/cluster/cluster_test.go

@@ -29,7 +29,9 @@ import (
 	"github.com/rook/rook/pkg/clusterd"
 	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
 	"github.com/rook/rook/pkg/operator/ceph/cluster/mon"
+	"github.com/rook/rook/pkg/operator/ceph/cluster/nodedaemon"
 	"github.com/rook/rook/pkg/operator/ceph/cluster/telemetry"
+	"github.com/rook/rook/pkg/operator/ceph/csi"
 	cephver "github.com/rook/rook/pkg/operator/ceph/version"
 	testop "github.com/rook/rook/pkg/operator/test"
 	exectest "github.com/rook/rook/pkg/util/exec/test"
@@ -295,7 +297,8 @@ func TestConfigureMsgr2(t *testing.T) {
 
 func TestTelemetry(t *testing.T) {
 	var expectedSettings map[string]string
-	context := &clusterd.Context{Clientset: testop.New(t, 3)}
+	clientset := testop.New(t, 3)
+	context := &clusterd.Context{Clientset: clientset}
 	context.Executor = &exectest.MockExecutor{
 		MockExecuteCommandWithTimeout: func(timeout time.Duration, command string, args ...string) (string, error) {
 			logger.Infof("Command: %s %v", command, args)
@@ -316,9 +319,11 @@ func TestTelemetry(t *testing.T) {
 		context:     context,
 		ClusterInfo: cephclient.AdminTestClusterInfo("cluster"),
 		mons:        &mon.Cluster{},
+		Namespace:   "rook-ceph",
 	}
 
 	t.Run("normal cluster", func(t *testing.T) {
+		ns := "rook-ceph"
 		c.Spec = &cephv1.ClusterSpec{
 			Mon: cephv1.MonSpec{Count: 3, AllowMultiplePerNode: true},
 			Storage: cephv1.StorageScopeSpec{
@@ -328,7 +333,8 @@ func TestTelemetry(t *testing.T) {
 					{Name: "three", Count: 3, Portable: false},
 				},
 			},
-			Network: cephv1.NetworkSpec{Provider: "host"},
+			Network:        cephv1.NetworkSpec{Provider: "host"},
+			CrashCollector: cephv1.CrashCollectorSpec{Disable: false},
 		}
 		expectedSettings = map[string]string{
 			telemetry.K8sVersionKey:              "v0.0.0-master+$Format:%H$",
@@ -342,13 +348,45 @@ func TestTelemetry(t *testing.T) {
 			telemetry.DeviceSetNonPortableKey:    "1",
 			telemetry.NetworkProviderKey:         "host",
 			telemetry.ExternalModeEnabledKey:     "false",
+			telemetry.K8sNodeCount:               "3",
+			telemetry.CephFSNodeCount:            "0",
+			telemetry.RBDNodeCount:               "0",
+			telemetry.NFSNodeCount:               "0",
+			telemetry.CephNodeCount:              "0",
+		}
+		c.reportTelemetry()
+
+		// add some pods to the nodes
+		labels := make(map[string]string)
+		labels["app"] = nodedaemon.CrashCollectorAppName
+		testop.FakeCustomisePodCreate(t, clientset, "crash-collector", ns, labels)
+		labels["app"] = csi.CsiRBDPlugin
+		testop.FakeCustomisePodCreate(t, clientset, "csi-rbdplugin", ns, labels)
+		expectedSettings = map[string]string{
+			telemetry.K8sVersionKey:              "v0.0.0-master+$Format:%H$",
+			telemetry.MonMaxIDKey:                "0",
+			telemetry.MonCountKey:                "3",
+			telemetry.MonAllowMultiplePerNodeKey: "true",
+			telemetry.MonPVCEnabledKey:           "false",
+			telemetry.MonStretchEnabledKey:       "false",
+			telemetry.DeviceSetTotalKey:          "3",
+			telemetry.DeviceSetPortableKey:       "2",
+			telemetry.DeviceSetNonPortableKey:    "1",
+			telemetry.NetworkProviderKey:         "host",
+			telemetry.ExternalModeEnabledKey:     "false",
+			telemetry.K8sNodeCount:               "3",
+			telemetry.CephFSNodeCount:            "0",
+			telemetry.RBDNodeCount:               "1", //should be 1
+			telemetry.NFSNodeCount:               "0",
+			telemetry.CephNodeCount:              "1", //should be 1
 		}
 		c.reportTelemetry()
 	})
 
 	t.Run("external cluster", func(t *testing.T) {
 		c.Spec = &cephv1.ClusterSpec{
-			External: cephv1.ExternalSpec{Enable: true},
+			External:       cephv1.ExternalSpec{Enable: true},
+			CrashCollector: cephv1.CrashCollectorSpec{Disable: true},
 		}
 		expectedSettings = map[string]string{
 			telemetry.K8sVersionKey:              "v0.0.0-master+$Format:%H$",
@@ -362,6 +400,11 @@ func TestTelemetry(t *testing.T) {
 			telemetry.DeviceSetNonPortableKey:    "0",
 			telemetry.NetworkProviderKey:         "",
 			telemetry.ExternalModeEnabledKey:     "true",
+			telemetry.K8sNodeCount:               "3",
+			telemetry.CephFSNodeCount:            "0",
+			telemetry.RBDNodeCount:               "1",
+			telemetry.NFSNodeCount:               "0",
+			telemetry.CephNodeCount:              "-1", // checking the disable scenario
 		}
 		c.reportTelemetry()
 	})

pkg/operator/ceph/cluster/nodedaemon/add.go

@@ -42,7 +42,7 @@ import (
 const (
 	controllerName = "ceph-nodedaemon-controller"
 	// AppName is the value to the "app" label for the ceph-crash pods
-	crashCollectorAppName = "rook-ceph-crashcollector"
+	CrashCollectorAppName = "rook-ceph-crashcollector"
 	cephExporterAppName   = "rook-ceph-exporter"
 	prunerName            = "rook-ceph-crashcollector-pruner"
 	// NodeNameLabel is a node name label
@@ -104,7 +104,7 @@ func add(mgr manager.Manager, r reconcile.Reconciler) error {
 			}
 			labels := deployment.GetLabels()
 			appName, ok := labels[k8sutil.AppAttr]
-			if !ok || appName != crashCollectorAppName {
+			if !ok || appName != CrashCollectorAppName {
 				return []reconcile.Request{}
 			}
 			nodeName, ok := deployment.Spec.Template.ObjectMeta.Labels[NodeNameLabel]

pkg/operator/ceph/cluster/nodedaemon/crash.go

@@ -50,7 +50,7 @@ func (r *ReconcileNode) createOrUpdateCephCrash(node corev1.Node, tolerations []
 	}
 	deploy := &appsv1.Deployment{
 		ObjectMeta: metav1.ObjectMeta{
-			Name:      k8sutil.TruncateNodeName(fmt.Sprintf("%s-%%s", crashCollectorAppName), nodeHostnameLabel),
+			Name:      k8sutil.TruncateNodeName(fmt.Sprintf("%s-%%s", CrashCollectorAppName), nodeHostnameLabel),
 			Namespace: cephCluster.GetNamespace(),
 		},
 	}
@@ -67,7 +67,7 @@ func (r *ReconcileNode) createOrUpdateCephCrash(node corev1.Node, tolerations []
 		// labels for the pod, the deployment, and the deploymentSelector
 		deploymentLabels := map[string]string{
 			corev1.LabelHostname: nodeHostnameLabel,
-			k8sutil.AppAttr:      crashCollectorAppName,
+			k8sutil.AppAttr:      CrashCollectorAppName,
 			NodeNameLabel:        node.GetName(),
 		}
 		deploymentLabels[config.CrashType] = "crash"
@@ -76,7 +76,7 @@ func (r *ReconcileNode) createOrUpdateCephCrash(node corev1.Node, tolerations []
 
 		selectorLabels := map[string]string{
 			corev1.LabelHostname: nodeHostnameLabel,
-			k8sutil.AppAttr:      crashCollectorAppName,
+			k8sutil.AppAttr:      CrashCollectorAppName,
 			NodeNameLabel:        node.GetName(),
 		}
 

pkg/operator/ceph/cluster/nodedaemon/crash_test.go

@@ -75,7 +75,7 @@ func TestCreateOrUpdateCephCrash(t *testing.T) {
 	res, err := r.createOrUpdateCephCrash(node, tolerations, cephCluster, cephVersion)
 	assert.NoError(t, err)
 	assert.Equal(t, controllerutil.OperationResult("created"), res)
-	name := k8sutil.TruncateNodeName(fmt.Sprintf("%s-%%s", crashCollectorAppName), "testnode")
+	name := k8sutil.TruncateNodeName(fmt.Sprintf("%s-%%s", CrashCollectorAppName), "testnode")
 	deploy := appsv1.Deployment{}
 	err = r.client.Get(ctx, types.NamespacedName{Namespace: "rook-ceph", Name: name}, &deploy)
 	assert.NoError(t, err)

pkg/operator/ceph/cluster/nodedaemon/reconcile.go

@@ -287,7 +287,7 @@ func (r *ReconcileNode) createOrUpdateNodeDaemons(node corev1.Node, tolerations
 func (r *ReconcileNode) removeDisabledCrashCollectorDaemons(spec cephv1.ClusterSpec, namespace string) bool {
 	// If the crash daemons are disabled in the spec let's remove them
 	if spec.CrashCollector.Disable {
-		r.deleteNodeDaemon(crashCollectorAppName, namespace)
+		r.deleteNodeDaemon(CrashCollectorAppName, namespace)
 	}
 
 	return spec.CrashCollector.Disable
@@ -362,7 +362,7 @@ func (r *ReconcileNode) cephPodList() ([]corev1.Pod, error) {
 
 func (r *ReconcileNode) listNodeDaemonsAndDelete(nodeName, ns string) error {
 	// delete the crash daemons on the given node
-	if err := r.listDeploymentAndDelete(crashCollectorAppName, nodeName, ns); err != nil {
+	if err := r.listDeploymentAndDelete(CrashCollectorAppName, nodeName, ns); err != nil {
 		return errors.Wrap(err, "failed to delete crash collector")
 	}
 

pkg/operator/ceph/cluster/telemetry/telemetry.go

@@ -41,6 +41,11 @@ const (
 	DeviceSetNonPortableKey    = "rook/cluster/storage/device-set/count/non-portable"
 	NetworkProviderKey         = "rook/cluster/network/provider"
 	ExternalModeEnabledKey     = "rook/cluster/external-mode"
+	K8sNodeCount               = "rook/node/count/kubernetes-total"
+	CephNodeCount              = "rook/node/count/with-ceph-daemons"
+	RBDNodeCount               = "rook/node/count/with-csi-rbd-plugin"
+	CephFSNodeCount            = "rook/node/count/with-csi-cephfs-plugin"
+	NFSNodeCount               = "rook/node/count/with-csi-nfs-plugin"
 )
 
 var CSIVersion string

pkg/operator/ceph/csi/controller.go

@@ -33,6 +33,7 @@ import (
 	addonsv1alpha1 "github.com/csi-addons/kubernetes-csi-addons/apis/csiaddons/v1alpha1"
 	"github.com/pkg/errors"
 	"github.com/rook/rook/pkg/clusterd"
+
 	opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
 	"github.com/rook/rook/pkg/operator/ceph/csi/peermap"
 	"github.com/rook/rook/pkg/operator/k8sutil"
@@ -201,6 +202,7 @@ func (r *ReconcileCSI) reconcile(request reconcile.Request) (reconcile.Result, e
 			logger.Debugf("ceph cluster %q has cleanup policy, the cluster will soon go away, no need to reconcile the csi driver", cluster.Name)
 			return reconcile.Result{}, nil
 		}
+
 		holderEnabled := !csiHostNetworkEnabled || cluster.Spec.Network.IsMultus()
 		// Do we have a multus cluster or csi host network disabled?
 		// If so deploy the plugin holder with the fsid attached
@@ -218,11 +220,9 @@ func (r *ReconcileCSI) reconcile(request reconcile.Request) (reconcile.Result, e
 				return opcontroller.ImmediateRetryResult, errors.Wrapf(err, "failed to load cluster info for cluster %q", cluster.Name)
 			}
 			clusterInfo.OwnerInfo = k8sutil.NewOwnerInfo(&cephClusters.Items[i], r.scheme)
-
 			logger.Debugf("cluster %q is running on multus or CSI_ENABLE_HOST_NETWORK is false, deploying the ceph-csi plugin holder", cluster.Name)
 
 			r.clustersWithHolder = append(r.clustersWithHolder, ClusterDetail{cluster: &cephClusters.Items[i], clusterInfo: clusterInfo})
-
 		} else {
 			logger.Debugf("not a multus cluster %q or CSI_ENABLE_HOST_NETWORK is true, not deploying the ceph-csi plugin holder", request.NamespacedName)
 		}

pkg/operator/ceph/csi/controller_test.go

@@ -216,7 +216,7 @@ func TestCephCSIController(t *testing.T) {
 		// Create a fake client to mock API calls.
 		cl := fake.NewClientBuilder().WithScheme(s).WithRuntimeObjects(object...).Build()
 		c.Client = cl
-		// // Create a ReconcileCSI object with the scheme and fake client.
+		// Create a ReconcileCSI object with the scheme and fake client.
 		r := &ReconcileCSI{
 			scheme:  s,
 			client:  cl,

pkg/operator/ceph/csi/spec.go

@@ -248,9 +248,9 @@ const (
 	onDelete      = "OnDelete"
 
 	// driver daemonset names
-	csiRBDPlugin    = "csi-rbdplugin"
-	csiCephFSPlugin = "csi-cephfsplugin"
-	csiNFSPlugin    = "csi-nfsplugin"
+	CsiRBDPlugin    = "csi-rbdplugin"
+	CsiCephFSPlugin = "csi-cephfsplugin"
+	CsiNFSPlugin    = "csi-nfsplugin"
 
 	// driver deployment names
 	csiRBDProvisioner    = "csi-rbdplugin-provisioner"
@@ -656,7 +656,7 @@ func (r *ReconcileCSI) stopDrivers(ver *version.Info) error {
 
 	if !EnableRBD {
 		logger.Info("CSI Ceph RBD driver disabled")
-		err := r.deleteCSIDriverResources(ver, csiRBDPlugin, csiRBDProvisioner, "csi-rbdplugin-metrics", RBDDriverName)
+		err := r.deleteCSIDriverResources(ver, CsiRBDPlugin, csiRBDProvisioner, "csi-rbdplugin-metrics", RBDDriverName)
 		if err != nil {
 			return errors.Wrap(err, "failed to remove CSI Ceph RBD driver")
 		}
@@ -665,7 +665,7 @@ func (r *ReconcileCSI) stopDrivers(ver *version.Info) error {
 
 	if !EnableCephFS {
 		logger.Info("CSI CephFS driver disabled")
-		err := r.deleteCSIDriverResources(ver, csiCephFSPlugin, csiCephFSProvisioner, "csi-cephfsplugin-metrics", CephFSDriverName)
+		err := r.deleteCSIDriverResources(ver, CsiCephFSPlugin, csiCephFSProvisioner, "csi-cephfsplugin-metrics", CephFSDriverName)
 		if err != nil {
 			return errors.Wrap(err, "failed to remove CSI CephFS driver")
 		}
@@ -674,7 +674,7 @@ func (r *ReconcileCSI) stopDrivers(ver *version.Info) error {
 
 	if !EnableNFS {
 		logger.Info("CSI NFS driver disabled")
-		err := r.deleteCSIDriverResources(ver, csiNFSPlugin, csiNFSProvisioner, "csi-nfsplugin-metrics", NFSDriverName)
+		err := r.deleteCSIDriverResources(ver, CsiNFSPlugin, csiNFSProvisioner, "csi-nfsplugin-metrics", NFSDriverName)
 		if err != nil {
 			return errors.Wrap(err, "failed to remove CSI NFS driver")
 		}

pkg/operator/test/client.go

@@ -300,3 +300,26 @@ func FakeReplicaSet(ns string) *appsv1.ReplicaSet {
 
 	return r
 }
+
+func FakeCustomisePodCreate(t *testing.T, clientset *fake.Clientset, name, ns string, label map[string]string) {
+	pod := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: ns,
+			OwnerReferences: []metav1.OwnerReference{
+				{
+					Kind: "Deployment",
+				},
+			},
+			Labels: label,
+		},
+	}
+	err := clientset.Tracker().Create(podGVR, pod, ns)
+	if err != nil {
+		if errors.IsAlreadyExists(err) {
+			t.Logf("pod %q is already created", pod.GetName())
+		}
+		panic(fmt.Errorf("failed to create Pod %q. %v", pod.Name, err))
+	}
+	t.Logf("job reactor: created pod %q ", pod.Name)
+}