Ensure entity events are sent via gateway if enabled

.chloggen/fix-entities-with-gateway.yaml

@@ -0,0 +1,8 @@
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: bug_fix
+# The name of the component, or a single word describing the area of concern, (e.g. agent, clusterReceiver, gateway, operator, chart, other)
+component: agent/gateway
+# A brief description of the change. Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Ensure entity events are sent via gateway if enabled.
+# One or more tracking issues related to the change
+issues: [1732]

examples/collector-all-modes/rendered_manifests/configmap-agent.yaml

@@ -23,9 +23,7 @@ data:
         tls:
           insecure: true
       otlphttp/entities:
-        headers:
-          X-SF-Token: ${SPLUNK_OBSERVABILITY_ACCESS_TOKEN}
-        logs_endpoint: https://ingest.CHANGEME.signalfx.com/v3/event
+        endpoint: http://default-splunk-otel-collector:4318
       signalfx:
         access_token: ${SPLUNK_OBSERVABILITY_ACCESS_TOKEN}
         api_url: http://default-splunk-otel-collector:6060

examples/collector-all-modes/rendered_manifests/configmap-gateway.yaml

@@ -25,6 +25,10 @@ data:
         sending_queue:
           num_consumers: 32
         traces_endpoint: https://ingest.CHANGEME.signalfx.com/v2/trace/otlp
+      otlphttp/entities:
+        headers:
+          X-SF-Token: ${SPLUNK_OBSERVABILITY_ACCESS_TOKEN}
+        logs_endpoint: https://ingest.CHANGEME.signalfx.com/v3/event
       signalfx:
         access_token: ${SPLUNK_OBSERVABILITY_ACCESS_TOKEN}
         api_url: https://api.CHANGEME.signalfx.com
@@ -174,6 +178,14 @@ data:
       - zpages
       - http_forwarder
       pipelines:
+        logs/entities:
+          exporters:
+          - otlphttp/entities
+          processors:
+          - memory_limiter
+          - batch
+          receivers:
+          - otlp
         logs/signalfx-events:
           exporters:
           - signalfx

examples/collector-all-modes/rendered_manifests/daemonset.yaml

@@ -32,7 +32,7 @@ spec:
         component: otel-collector-agent
         release: default
       annotations:
-        checksum/config: e7a0c25de991c8e1aebf9d671f91c44db517edf929904230adf462ea12d4eb80
+        checksum/config: e835fc38cee8d6140fb61aa9604e6faa455485306a89d94b3204d97163dc92e6
         kubectl.kubernetes.io/default-container: otel-collector
     spec:
       hostNetwork: true

examples/collector-all-modes/rendered_manifests/deployment-gateway.yaml

@@ -31,7 +31,7 @@ spec:
         component: otel-collector
         release: default
       annotations:
-        checksum/config: bb199f9676abe8e081c9d1baf013f8765f50cb4a8fbd77e9e3905ddb185fd1da
+        checksum/config: 30b682ff55136ca6229f6159e93640a039cf69e20dbd78497f7f6101f0e3b484
     spec:
       serviceAccountName: default-splunk-otel-collector
       nodeSelector:

examples/collector-gateway-only/rendered_manifests/configmap-gateway.yaml

@@ -25,6 +25,10 @@ data:
         sending_queue:
           num_consumers: 32
         traces_endpoint: https://ingest.CHANGEME.signalfx.com/v2/trace/otlp
+      otlphttp/entities:
+        headers:
+          X-SF-Token: ${SPLUNK_OBSERVABILITY_ACCESS_TOKEN}
+        logs_endpoint: https://ingest.CHANGEME.signalfx.com/v3/event
       signalfx:
         access_token: ${SPLUNK_OBSERVABILITY_ACCESS_TOKEN}
         api_url: https://api.CHANGEME.signalfx.com
@@ -174,6 +178,14 @@ data:
       - zpages
       - http_forwarder
       pipelines:
+        logs/entities:
+          exporters:
+          - otlphttp/entities
+          processors:
+          - memory_limiter
+          - batch
+          receivers:
+          - otlp
         logs/signalfx-events:
           exporters:
           - signalfx

examples/collector-gateway-only/rendered_manifests/deployment-gateway.yaml

@@ -31,7 +31,7 @@ spec:
         component: otel-collector
         release: default
       annotations:
-        checksum/config: bb199f9676abe8e081c9d1baf013f8765f50cb4a8fbd77e9e3905ddb185fd1da
+        checksum/config: 30b682ff55136ca6229f6159e93640a039cf69e20dbd78497f7f6101f0e3b484
     spec:
       serviceAccountName: default-splunk-otel-collector
       nodeSelector:

examples/distribution-eks-fargate/rendered_manifests/configmap-gateway.yaml

@@ -25,6 +25,10 @@ data:
         sending_queue:
           num_consumers: 32
         traces_endpoint: https://ingest.CHANGEME.signalfx.com/v2/trace/otlp
+      otlphttp/entities:
+        headers:
+          X-SF-Token: ${SPLUNK_OBSERVABILITY_ACCESS_TOKEN}
+        logs_endpoint: https://ingest.CHANGEME.signalfx.com/v3/event
       signalfx:
         access_token: ${SPLUNK_OBSERVABILITY_ACCESS_TOKEN}
         api_url: https://api.CHANGEME.signalfx.com
@@ -176,6 +180,14 @@ data:
       - zpages
       - http_forwarder
       pipelines:
+        logs/entities:
+          exporters:
+          - otlphttp/entities
+          processors:
+          - memory_limiter
+          - batch
+          receivers:
+          - otlp
         logs/signalfx-events:
           exporters:
           - signalfx

examples/distribution-eks-fargate/rendered_manifests/deployment-gateway.yaml

@@ -31,7 +31,7 @@ spec:
         component: otel-collector
         release: default
       annotations:
-        checksum/config: 45e83563cc906164f7ae0bfadaa6c5cfea671e742fe7156f3a96805814235100
+        checksum/config: 76886faf0fa12b65bca4c14b3954be0a2b9e452c260b71352c5ac0fab6995132
     spec:
       serviceAccountName: default-splunk-otel-collector
       nodeSelector:

functional_tests/discovery/discovery_test.go

@@ -0,0 +1,225 @@
+// Copyright Splunk Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+package k8sevents
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/consumer/consumertest"
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	"helm.sh/helm/v3/pkg/action"
+	"helm.sh/helm/v3/pkg/chart/loader"
+	"helm.sh/helm/v3/pkg/cli"
+	"helm.sh/helm/v3/pkg/registry"
+
+	"github.com/signalfx/splunk-otel-collector-chart/functional_tests/internal"
+)
+
+const (
+	redisReleaseName = "test-redis"
+	redisChartRepo   = "https://charts.bitnami.com/bitnami"
+	redisChart       = "redis"
+)
+
+// Env vars to control the test behavior:
+// KUBECONFIG (required): the path to the kubeconfig file
+// TEARDOWN_BEFORE_SETUP: if set to true, the test will run teardown before setup
+// SKIP_TEARDOWN: if set to true, the test will skip teardown
+// SKIP_SETUP: if set to true, the test will skip setup
+func Test_Discovery(t *testing.T) {
+	testKubeConfig, ok := os.LookupEnv("KUBECONFIG")
+	require.True(t, ok, "the environment variable KUBECONFIG must be set")
+	if os.Getenv("TEARDOWN_BEFORE_SETUP") == "true" {
+		teardown(t, testKubeConfig)
+	}
+	installRedisChart(t, testKubeConfig)
+	t.Cleanup(func() {
+		if os.Getenv("SKIP_TEARDOWN") == "true" {
+			t.Log("Skipping teardown as SKIP_TEARDOWN is set to true")
+			return
+		}
+		teardown(t, testKubeConfig)
+	})
+
+	internal.SetupSignalFxApiServer(t)
+
+	tests := []struct {
+		name       string
+		valuesTmpl string
+	}{
+		{
+			name:       "agent_only",
+			valuesTmpl: "agent_only_values.tmpl",
+		},
+		{
+			name:       "agent_with_gateway",
+			valuesTmpl: "agent_with_gateway_values.tmpl",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			metricsSink := internal.SetupSignalfxReceiver(t, internal.SignalFxReceiverPort)
+			eventsSink := internal.SetupOTLPLogsSink(t)
+			installCollectorChart(t, testKubeConfig, tt.valuesTmpl)
+			t.Cleanup(func() {
+				if os.Getenv("SKIP_TEARDOWN") == "true" {
+					return
+				}
+				internal.ChartUninstall(t, testKubeConfig)
+			})
+			assertRedisEntities(t, eventsSink)
+			assertRedisMetrics(t, metricsSink)
+		})
+	}
+}
+
+func assertRedisEntities(t *testing.T, sink *consumertest.LogsSink) {
+	internal.WaitForLogs(t, 1, sink)
+	rl := sink.AllLogs()[len(sink.AllLogs())-1].ResourceLogs().At(0)
+	assertAttr(t, rl.Resource().Attributes(), "k8s.cluster.name", "test-cluster")
+	assert.Equal(t, 1, rl.ScopeLogs().Len())
+	sl := rl.ScopeLogs().At(0)
+	assertAttr(t, sl.Scope().Attributes(), "otel.entity.event_as_log", true)
+	assert.Equal(t, 1, sl.LogRecords().Len())
+	lrAttrs := sl.LogRecords().At(0).Attributes()
+	assertAttr(t, lrAttrs, "otel.entity.event.type", "entity_state")
+	assertAttr(t, lrAttrs, "otel.entity.event.type", "entity_state")
+	assertAttr(t, lrAttrs, "otel.entity.type", "service")
+	entityAttrsVal, ok := lrAttrs.Get("otel.entity.attributes")
+	assert.True(t, ok)
+	entityAttrs := entityAttrsVal.Map()
+	assertAttr(t, entityAttrs, "k8s.namespace.name", internal.Namespace)
+	assertAttr(t, entityAttrs, "service.type", "redis")
+	assertAttr(t, entityAttrs, "service.name", "redis")
+	assertAttr(t, entityAttrs, "k8s.pod.name", "test-redis-master-0")
+	assertAttr(t, entityAttrs, "discovery.status", "successful")
+}
+
+func assertAttr(t *testing.T, attrs pcommon.Map, name string, val any) {
+	entityType, ok := attrs.Get(name)
+	assert.True(t, ok)
+	if ok {
+		assert.Equal(t, val, entityType.AsRaw())
+	}
+}
+
+func assertRedisMetrics(t *testing.T, sink *consumertest.MetricsSink) {
+	internal.WaitForMetrics(t, 5, sink)
+	foundMetrics := make(map[string]bool)
+	for _, m := range sink.AllMetrics() {
+		for i := 0; i < m.ResourceMetrics().Len(); i++ {
+			sm := m.ResourceMetrics().At(i).ScopeMetrics().At(0)
+			for j := 0; j < sm.Metrics().Len(); j++ {
+				foundMetrics[sm.Metrics().At(j).Name()] = true
+			}
+		}
+	}
+	expectedRedisMetrics := []string{
+		"redis.clients.blocked",
+		"redis.clients.connected",
+		"redis.clients.max_input_buffer",
+		"redis.clients.max_output_buffer",
+		"redis.commands",
+		"redis.commands.processed",
+		"redis.connections.received",
+		"redis.connections.rejected",
+		"redis.cpu.time",
+		"redis.keys.evicted",
+		"redis.keys.expired",
+		"redis.keyspace.hits",
+		"redis.keyspace.misses",
+		"redis.latest_fork",
+		"redis.memory.fragmentation_ratio",
+		"redis.memory.lua",
+		"redis.memory.peak",
+		"redis.memory.rss",
+		"redis.memory.used",
+		"redis.net.input",
+		"redis.net.output",
+		"redis.rdb.changes_since_last_save",
+		"redis.replication.backlog_first_byte_offset",
+		"redis.replication.offset",
+		"redis.slaves.connected",
+		"redis.uptime",
+	}
+	for _, rm := range expectedRedisMetrics {
+		assert.Contains(t, foundMetrics, rm)
+	}
+}
+
+func installCollectorChart(t *testing.T, kubeConfig, valuesTmpl string) {
+	t.Helper()
+	if os.Getenv("SKIP_SETUP") == "true" {
+		t.Log("Skipping collector chart installation as SKIP_SETUP is set to true")
+		return
+	}
+
+	hostEp := internal.HostEndpoint(t)
+	valuesFile, err := filepath.Abs(filepath.Join("testdata", valuesTmpl))
+	require.NoError(t, err)
+	internal.ChartInstallOrUpgrade(t, kubeConfig, valuesFile, map[string]any{
+		"ApiURL":    fmt.Sprintf("http://%s:%d", hostEp, internal.SignalFxAPIPort),
+		"IngestURL": fmt.Sprintf("http://%s:%d", hostEp, internal.SignalFxReceiverPort),
+		"EventsURL": fmt.Sprintf("http://%s:%d", hostEp, internal.OTLPHTTPReceiverPort),
+	})
+}
+
+// installRedisChart deploys a simple Redis server with official helm chart.
+func installRedisChart(t *testing.T, kubeConfig string) {
+	t.Helper()
+	if os.Getenv("SKIP_SETUP") == "true" {
+		t.Log("Skipping redis chart installation as SKIP_SETUP is set to true")
+		return
+	}
+
+	actionConfig := internal.InitHelmActionConfig(t, kubeConfig)
+	rc, err := registry.NewClient()
+	require.NoError(t, err)
+	actionConfig.RegistryClient = rc
+	install := action.NewInstall(actionConfig)
+	install.Namespace = internal.Namespace
+	install.ReleaseName = redisReleaseName
+	install.ChartPathOptions.RepoURL = redisChartRepo
+	install.Wait = true
+	install.Timeout = internal.HelmActionTimeout
+	hCli := cli.New()
+	hCli.KubeConfig = kubeConfig
+	chartPath, err := install.ChartPathOptions.LocateChart(redisChart, hCli)
+	require.NoError(t, err)
+	ch, err := loader.Load(chartPath)
+
+	// Install the redis chart with no replicas and no auth
+	release, err := install.Run(ch, map[string]any{
+		"auth": map[string]any{
+			"enabled": false,
+		},
+		"replica": map[string]any{
+			"replicaCount": 0,
+		},
+	})
+	require.NoError(t, err)
+	t.Logf("Helm chart installed. Release name: %s", release.Name)
+}
+
+func uninstallRedisChart(t *testing.T, kubeConfig string) {
+	t.Helper()
+	uninstallAction := action.NewUninstall(internal.InitHelmActionConfig(t, kubeConfig))
+	uninstallAction.Wait = true
+	uninstallAction.Timeout = internal.HelmActionTimeout
+	uninstallAction.IgnoreNotFound = true
+	_, err := uninstallAction.Run(redisReleaseName)
+	require.NoError(t, err)
+	t.Logf("Helm release %q uninstalled", redisReleaseName)
+}
+
+func teardown(t *testing.T, kubeConfig string) {
+	t.Helper()
+	uninstallRedisChart(t, kubeConfig)
+	internal.ChartUninstall(t, kubeConfig)
+}

functional_tests/discovery/testdata/agent_only_values.tmpl

@@ -0,0 +1,18 @@
+splunkObservability:
+   realm:       CHANGEME
+   accessToken: CHANGEME
+   apiUrl: {{ .ApiURL }}
+   ingestUrl: {{ .IngestURL }}
+
+clusterName: test-cluster
+
+agent:
+  discovery:
+    enabled: true
+  config:
+    exporters:
+      otlphttp/entities:
+        logs_endpoint: {{ .EventsURL }}/v1/logs
+
+clusterReceiver:
+  enabled: false