Add Logstash Telemetry (#6562)

This commit adds information about Logstash to the set of ECK telemetry 
Co-authored-by: Thibault Richard <[email protected]>

pkg/telemetry/fixtures.go

@@ -89,6 +89,14 @@ const expectedTelemetryTemplate = `eck:
       helm_resource_count: 0
       pod_count: 0
       resource_count: 1
+    logstashes:
+      pipeline_count: 0
+      pipeline_ref_count: 0
+      pod_count: 0
+      resource_count: 0
+      service_count: 0
+      stack_monitoring_logs_count: 0
+      stack_monitoring_metrics_count: 0
     maps:
       pod_count: 0
       resource_count: 0

pkg/telemetry/telemetry.go

@@ -24,6 +24,7 @@ import (
 	esv1 "github.com/elastic/cloud-on-k8s/v2/pkg/apis/elasticsearch/v1"
 	entv1 "github.com/elastic/cloud-on-k8s/v2/pkg/apis/enterprisesearch/v1"
 	kbv1 "github.com/elastic/cloud-on-k8s/v2/pkg/apis/kibana/v1"
+	logstashv1alpha1 "github.com/elastic/cloud-on-k8s/v2/pkg/apis/logstash/v1alpha1"
 	mapsv1alpha1 "github.com/elastic/cloud-on-k8s/v2/pkg/apis/maps/v1alpha1"
 	policyv1alpha1 "github.com/elastic/cloud-on-k8s/v2/pkg/apis/stackconfigpolicy/v1alpha1"
 	"github.com/elastic/cloud-on-k8s/v2/pkg/controller/common/reconciler"
@@ -40,8 +41,7 @@ const (
 	resourceCount            = "resource_count"
 	podCount                 = "pod_count"
 	helmManagedResourceCount = "helm_resource_count"
-
-	timestampFieldName = "timestamp"
+	timestampFieldName       = "timestamp"
 )
 
 type ECKTelemetry struct {
@@ -123,6 +123,7 @@ func (r *Reporter) getResourceStats(ctx context.Context) (map[string]interface{}
 		agentStats,
 		mapsStats,
 		scpStats,
+		logstashStats,
 	} {
 		key, statsPart, err := f(r.client, r.managedNamespaces)
 		if err != nil {
@@ -398,6 +399,43 @@ func agentStats(k8sClient k8s.Client, managedNamespaces []string) (string, inter
 	return "agents", stats, nil
 }
 
+func logstashStats(k8sClient k8s.Client, managedNamespaces []string) (string, interface{}, error) {
+	const (
+		pipelineCount               = "pipeline_count"
+		pipelineRefCount            = "pipeline_ref_count"
+		serviceCount                = "service_count"
+		stackMonitoringLogsCount    = "stack_monitoring_logs_count"
+		stackMonitoringMetricsCount = "stack_monitoring_metrics_count"
+	)
+	stats := map[string]int32{resourceCount: 0, podCount: 0, stackMonitoringLogsCount: 0,
+		stackMonitoringMetricsCount: 0, serviceCount: 0, pipelineCount: 0, pipelineRefCount: 0}
+
+	var logstashList logstashv1alpha1.LogstashList
+	for _, ns := range managedNamespaces {
+		if err := k8sClient.List(context.Background(), &logstashList, client.InNamespace(ns)); err != nil {
+			return "", nil, err
+		}
+
+		for _, ls := range logstashList.Items {
+			ls := ls
+			stats[resourceCount]++
+			stats[serviceCount] += int32(len(ls.Spec.Services))
+			stats[podCount] += ls.Status.AvailableNodes
+			stats[pipelineCount] += int32(len(ls.Spec.Pipelines))
+			if ls.Spec.PipelinesRef != nil {
+				stats[pipelineRefCount]++
+			}
+			if monitoring.IsLogsDefined(&ls) {
+				stats[stackMonitoringLogsCount]++
+			}
+			if monitoring.IsMetricsDefined(&ls) {
+				stats[stackMonitoringMetricsCount]++
+			}
+		}
+	}
+	return "logstashes", stats, nil
+}
+
 func mapsStats(k8sClient k8s.Client, managedNamespaces []string) (string, interface{}, error) {
 	stats := map[string]int32{resourceCount: 0, podCount: 0}
 

pkg/telemetry/telemetry_test.go

@@ -24,6 +24,7 @@ import (
 	esv1 "github.com/elastic/cloud-on-k8s/v2/pkg/apis/elasticsearch/v1"
 	entv1 "github.com/elastic/cloud-on-k8s/v2/pkg/apis/enterprisesearch/v1"
 	kbv1 "github.com/elastic/cloud-on-k8s/v2/pkg/apis/kibana/v1"
+	logstashv1alpha1 "github.com/elastic/cloud-on-k8s/v2/pkg/apis/logstash/v1alpha1"
 	mapsv1alpha1 "github.com/elastic/cloud-on-k8s/v2/pkg/apis/maps/v1alpha1"
 	policyv1alpha1 "github.com/elastic/cloud-on-k8s/v2/pkg/apis/stackconfigpolicy/v1alpha1"
 	"github.com/elastic/cloud-on-k8s/v2/pkg/controller/kibana"
@@ -227,6 +228,69 @@ func TestNewReporter(t *testing.T) {
 				},
 			},
 		},
+		&logstashv1alpha1.Logstash{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: "ns1",
+			},
+			Spec: logstashv1alpha1.LogstashSpec{
+				Count: 3,
+				Monitoring: commonv1.Monitoring{
+					Logs:    commonv1.LogsMonitoring{ElasticsearchRefs: []commonv1.ObjectSelector{{Name: "monitoring"}}},
+					Metrics: commonv1.MetricsMonitoring{ElasticsearchRefs: []commonv1.ObjectSelector{{Name: "monitoring"}}},
+				},
+				Pipelines: []commonv1.Config{
+					{Data: map[string]interface{}{"pipeline.id": "main"}},
+				},
+				Services: []logstashv1alpha1.LogstashService{
+					{
+						Name: "test1",
+						Service: commonv1.ServiceTemplate{
+							Spec: corev1.ServiceSpec{
+								Ports: []corev1.ServicePort{
+									{Port: 9200},
+								},
+							},
+						},
+					},
+					{
+						Name: "test2",
+						Service: commonv1.ServiceTemplate{
+							Spec: corev1.ServiceSpec{
+								Ports: []corev1.ServicePort{
+									{Port: 9201},
+								},
+							},
+						},
+					},
+				},
+			},
+			Status: logstashv1alpha1.LogstashStatus{
+				AvailableNodes: 3,
+			},
+		},
+		&logstashv1alpha1.Logstash{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: "ns2",
+			},
+			Spec: logstashv1alpha1.LogstashSpec{
+				Count: 1,
+				Services: []logstashv1alpha1.LogstashService{
+					{
+						Name: "test1",
+						Service: commonv1.ServiceTemplate{
+							Spec: corev1.ServiceSpec{
+								Ports: []corev1.ServicePort{
+									{Port: 9200},
+								},
+							},
+						},
+					},
+				},
+			},
+			Status: logstashv1alpha1.LogstashStatus{
+				AvailableNodes: 1,
+			},
+		},
 		&beatv1beta1.Beat{
 			ObjectMeta: metav1.ObjectMeta{
 				Name:      "beat1",
@@ -417,6 +481,14 @@ func TestNewReporter(t *testing.T) {
       helm_resource_count: 1
       pod_count: 0
       resource_count: 3
+    logstashes:
+      pipeline_count: 1
+      pipeline_ref_count: 0
+      pod_count: 4
+      resource_count: 2
+      service_count: 3
+      stack_monitoring_logs_count: 1
+      stack_monitoring_metrics_count: 1
     maps:
       pod_count: 1
       resource_count: 1