diff --git a/pkg/operator/ceph/cluster/cluster.go b/pkg/operator/ceph/cluster/cluster.go index 9ba83f9f6..166ccfad5 100755 --- a/pkg/operator/ceph/cluster/cluster.go +++ b/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" diff --git a/pkg/operator/ceph/cluster/cluster_test.go b/pkg/operator/ceph/cluster/cluster_test.go index fed3ae2ad..cc300188d 100644 --- a/pkg/operator/ceph/cluster/cluster_test.go +++ b/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() }) diff --git a/pkg/operator/ceph/cluster/nodedaemon/add.go b/pkg/operator/ceph/cluster/nodedaemon/add.go index 5fec0dab6..a0e0ff9c5 100644 --- a/pkg/operator/ceph/cluster/nodedaemon/add.go +++ b/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] diff --git a/pkg/operator/ceph/cluster/nodedaemon/crash.go b/pkg/operator/ceph/cluster/nodedaemon/crash.go index 83757853e..c6b9e7e09 100644 --- a/pkg/operator/ceph/cluster/nodedaemon/crash.go +++ b/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(), } diff --git a/pkg/operator/ceph/cluster/nodedaemon/crash_test.go b/pkg/operator/ceph/cluster/nodedaemon/crash_test.go index 46fc3a652..d5602c733 100644 --- a/pkg/operator/ceph/cluster/nodedaemon/crash_test.go +++ b/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) diff --git a/pkg/operator/ceph/cluster/nodedaemon/reconcile.go b/pkg/operator/ceph/cluster/nodedaemon/reconcile.go index 65f8f8ed1..af714bcd7 100644 --- a/pkg/operator/ceph/cluster/nodedaemon/reconcile.go +++ b/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") } diff --git a/pkg/operator/ceph/cluster/telemetry/telemetry.go b/pkg/operator/ceph/cluster/telemetry/telemetry.go index 7fc673f5f..da3a3239b 100644 --- a/pkg/operator/ceph/cluster/telemetry/telemetry.go +++ b/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 diff --git a/pkg/operator/ceph/csi/controller.go b/pkg/operator/ceph/csi/controller.go index fd899a118..64d9a3bb5 100644 --- a/pkg/operator/ceph/csi/controller.go +++ b/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) } diff --git a/pkg/operator/ceph/csi/controller_test.go b/pkg/operator/ceph/csi/controller_test.go index 700edb83d..8f72eb265 100644 --- a/pkg/operator/ceph/csi/controller_test.go +++ b/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, diff --git a/pkg/operator/ceph/csi/spec.go b/pkg/operator/ceph/csi/spec.go index 94743c037..d63a36658 100644 --- a/pkg/operator/ceph/csi/spec.go +++ b/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") } diff --git a/pkg/operator/test/client.go b/pkg/operator/test/client.go index 115144d4d..bf5851594 100644 --- a/pkg/operator/test/client.go +++ b/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) +}