Fix k8s resolver parsing so loadbalancing exporter works with service FQDNs

.chloggen/44472.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# 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. receiver/filelog)
+component: exporter/loadbalancing
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: "Fix k8s resolver parsing so loadbalancing exporter works with service FQDNs"
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [44472]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: []

exporter/loadbalancingexporter/resolver_k8s.go

@@ -87,10 +87,10 @@ func newK8sResolver(clt kubernetes.Interface,
 		timeout = defaultListWatchTimeout
 	}
 
-	nAddr := strings.SplitN(service, ".", 2)
-	name, namespace := nAddr[0], "default"
-	if len(nAddr) > 1 {
-		namespace = nAddr[1]
+	parts := strings.Split(service, ".")
+	name, namespace := parts[0], "default"
+	if len(parts) > 1 && parts[1] != "" {
+		namespace = parts[1]
 	} else {
 		logger.Info("the namespace for the Kubernetes service wasn't provided, trying to determine the current namespace", zap.String("name", name))
 		if ns, err := getInClusterNamespace(); err == nil {

exporter/loadbalancingexporter/resolver_k8s_test.go

@@ -257,6 +257,54 @@ func TestK8sResolve(t *testing.T) {
 	}
 }
 
+func TestK8sResolveWithServiceFQDN(t *testing.T) {
+	serviceName := "lb"
+	namespace := "custom"
+	serviceFQDN := fmt.Sprintf("%s.%s.svc.cluster.local", serviceName, namespace)
+	port := int32(4317)
+	hostname := "pod-0"
+
+	endpointSlice := &discoveryv1.EndpointSlice{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      serviceName,
+			Namespace: namespace,
+			Labels: map[string]string{
+				"kubernetes.io/service-name": serviceName,
+			},
+		},
+		Endpoints: []discoveryv1.Endpoint{
+			{
+				Addresses: []string{"10.0.0.1"},
+				Hostname:  &hostname,
+			},
+		},
+	}
+
+	cl := fake.NewClientset(endpointSlice)
+	_, tb := getTelemetryAssets(t)
+	res, err := newK8sResolver(cl, zap.NewNop(), serviceFQDN, []int32{port}, defaultListWatchTimeout, true, tb)
+	require.NoError(t, err)
+	require.Equal(t, serviceName, res.svcName)
+	require.Equal(t, namespace, res.svcNs)
+
+	require.NoError(t, res.start(t.Context()))
+	t.Cleanup(func() {
+		require.NoError(t, res.shutdown(t.Context()))
+	})
+
+	expected := []string{fmt.Sprintf("%s.%s.%s:%d", hostname, serviceName, namespace, port)}
+
+	cErr := waitForCondition(t, 3*time.Second, 20*time.Millisecond, func(ctx context.Context) (bool, error) {
+		if _, err := res.resolve(ctx); err != nil {
+			return false, err
+		}
+		return slices.Equal(expected, res.Endpoints()), nil
+	})
+	if cErr != nil {
+		t.Fatalf("timed out waiting for resolver endpoints to match expected: %v", cErr)
+	}
+}
+
 // waitForCondition will poll the condition function until it returns true or times out.
 // Any errors returned from the condition are treated as test failures.
 func waitForCondition(t *testing.T, timeout, interval time.Duration, condition func(context.Context) (bool, error)) error {