[OTel] Add TLS test to Logstash Exporter (#47199)

This commit verifies that key TLS features in Logstash Exporter still work as expected

Add TLS tests
- TLS connections
- Self-signed certificates
- Keypass support

Add E2E TestLogstashExporterProxyURL to ensure proxy_url config is respected
---------

Co-authored-by: Tiago Queiroz <[email protected]>

Author: kaisecheng

libbeat/otelbeat/oteltest/beatsauth_test.go

@@ -53,8 +53,6 @@ import (
 	"go.opentelemetry.io/otel/sdk/metric/metricdata"
 	"go.uber.org/zap/zaptest"
 
-	"github.com/elastic/pkcs8"
-
 	"gopkg.in/yaml.v2"
 
 	"github.com/elastic/beats/v7/libbeat/otelbeat/beatconverter"
@@ -101,7 +99,7 @@ func TestMTLS(t *testing.T) {
 	}
 
 	// get client certificates paths
-	clientCertificate, clientKey := getClientCerts(t, caCert, "")
+	clientCertificate, clientKey := GetClientCerts(t, caCert, "")
 
 	// start test server with given server and root certs
 	certPool := x509.NewCertPool()
@@ -117,11 +115,11 @@ func TestMTLS(t *testing.T) {
 
 	inputConfig := `
 receivers:
-  filebeatreceiver:	
+  filebeatreceiver:
     output:
       elasticsearch:
         hosts: {{ .Host }}
-        ssl: 
+        ssl:
           enabled: true
           certificate_authorities:
             - {{ .CACertificate }}
@@ -175,7 +173,7 @@ func TestKeyPassPhrase(t *testing.T) {
 	}
 
 	// get client certificates paths with key file encrypted in PKCS#8 format
-	clientCertificate, clientKey := getClientCerts(t, caCert, "your-password")
+	clientCertificate, clientKey := GetClientCerts(t, caCert, "your-password")
 
 	// start test server with given server and root certs
 	certPool := x509.NewCertPool()
@@ -191,17 +189,17 @@ func TestKeyPassPhrase(t *testing.T) {
 
 	inputConfig := `
 receivers:
-  filebeatreceiver:	
+  filebeatreceiver:
     output:
       elasticsearch:
         hosts: {{ .Host }}
-        ssl: 
+        ssl:
           enabled: true
           certificate_authorities:
             - {{ .CACertificate }}
           certificate: {{ .ClientCert }}
           key: {{ .ClientKey }}
-          key_passphrase: your-password		  
+          key_passphrase: your-password
 `
 
 	var otelConfigBuffer bytes.Buffer
@@ -262,13 +260,13 @@ func TestCATrustedFingerPrint(t *testing.T) {
 
 	inputConfig := `
 receivers:
-  filebeatreceiver:	
+  filebeatreceiver:
     output:
       elasticsearch:
         hosts: {{ .Host }}
-        ssl: 
+        ssl:
           enabled: true
-          ca_trusted_fingerprint: {{ .CATrustedFingerPrint }} 
+          ca_trusted_fingerprint: {{ .CATrustedFingerPrint }}
 `
 
 	var otelConfigBuffer bytes.Buffer
@@ -440,11 +438,11 @@ func TestVerificationMode(t *testing.T) {
 
 			inputConfig := `
 receivers:
-  filebeatreceiver:	
+  filebeatreceiver:
     output:
       elasticsearch:
         hosts: {{ .Host }}
-        ssl: 
+        ssl:
           enabled: true
           certificate_authorities:
             - {{ .CACertificate }}
@@ -536,7 +534,7 @@ func TestProxyHTTP(t *testing.T) {
 				proxytest.WithRequestLog("https", t.Logf)},
 			inputConfig: `
 receivers:
-  filebeatreceiver:	
+  filebeatreceiver:
     output:
       elasticsearch:
         hosts: {{ .Host }}
@@ -569,7 +567,7 @@ receivers:
 				})},
 			inputConfig: `
 receivers:
-  filebeatreceiver:	
+  filebeatreceiver:
     output:
       elasticsearch:
         hosts: {{ .Host }}
@@ -587,7 +585,7 @@ receivers:
 				proxytest.WithRequestLog("https", t.Logf)},
 			inputConfig: `
 receivers:
-  filebeatreceiver:	
+  filebeatreceiver:
     output:
       elasticsearch:
         hosts: {{ .Host }}
@@ -762,54 +760,6 @@ func mustSendLogs(t *testing.T, exporter exporter.Logs, logs plog.Logs) error {
 	return err
 }
 
-// getClientCerts creates client certificates, writes them to a file and return the path of certificate and key
-// if passphrase is passed, it is used to encrypt the key file
-func getClientCerts(t *testing.T, caCert tls.Certificate, passphrase string) (certificate string, key string) {
-	// create client certificates
-	clientCerts, err := tlscommontest.GenSignedCert(caCert, x509.KeyUsageCertSign, false, "client", []string{"localhost"}, []net.IP{net.IPv4(127, 0, 0, 1)}, false)
-	if err != nil {
-		t.Fatalf("could not generate certificates: %s", err)
-	}
-
-	tempDir := t.TempDir()
-	clientCertPath := filepath.Join(tempDir, "client-cert.pem")
-	clientKeyPath := filepath.Join(tempDir, "client-key.pem")
-
-	if passphrase != "" {
-		clientKey, err := pkcs8.MarshalPrivateKey(clientCerts.PrivateKey, []byte(passphrase), pkcs8.DefaultOpts)
-		if err != nil {
-			t.Fatalf("could not marshal private key: %v", err)
-		}
-
-		if err = os.WriteFile(clientKeyPath, pem.EncodeToMemory(&pem.Block{
-			Type:  "ENCRYPTED PRIVATE KEY",
-			Bytes: clientKey,
-		}), 0400); err != nil {
-			t.Fatalf("could not write client key to file")
-		}
-	} else {
-		clientKey, err := x509.MarshalPKCS8PrivateKey(clientCerts.PrivateKey)
-		if err != nil {
-			t.Fatalf("could not marshal private key: %v", err)
-		}
-		if err = os.WriteFile(clientKeyPath, pem.EncodeToMemory(&pem.Block{
-			Type:  "RSA PRIVATE KEY",
-			Bytes: clientKey,
-		}), 0400); err != nil {
-			t.Fatalf("could not write client key to file")
-		}
-	}
-
-	if err = os.WriteFile(clientCertPath, pem.EncodeToMemory(&pem.Block{
-		Type:  "CERTIFICATE",
-		Bytes: clientCerts.Leaf.Raw,
-	}), 0400); err != nil {
-		t.Fatalf("could not write client certificate to file")
-	}
-
-	return clientCertPath, clientKeyPath
-}
-
 func getTranslatedConf(t *testing.T, input []byte) *confmap.Conf {
 	c := beatconverter.Converter{}
 

libbeat/otelbeat/oteltest/tls_helper.go

@@ -0,0 +1,79 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package oteltest
+
+import (
+	"crypto/tls"
+	"crypto/x509"
+	"encoding/pem"
+	"net"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/elastic/elastic-agent-libs/transport/tlscommontest"
+	"github.com/elastic/pkcs8"
+)
+
+// GetClientCerts creates client certificates, writes them to a file and return the path of certificate and key
+// if passphrase is passed, it is used to encrypt the key file
+func GetClientCerts(t *testing.T, caCert tls.Certificate, passphrase string) (certificate string, key string) {
+	// create client certificates
+	clientCerts, err := tlscommontest.GenSignedCert(caCert, x509.KeyUsageCertSign, false, "client", []string{"localhost"}, []net.IP{net.IPv4(127, 0, 0, 1)}, false)
+	if err != nil {
+		t.Fatalf("could not generate certificates: %s", err)
+	}
+
+	tempDir := t.TempDir()
+	clientCertPath := filepath.Join(tempDir, "client-cert.pem")
+	clientKeyPath := filepath.Join(tempDir, "client-key.pem")
+
+	if passphrase != "" {
+		clientKey, err := pkcs8.MarshalPrivateKey(clientCerts.PrivateKey, []byte(passphrase), pkcs8.DefaultOpts)
+		if err != nil {
+			t.Fatalf("could not marshal private key: %v", err)
+		}
+
+		if err = os.WriteFile(clientKeyPath, pem.EncodeToMemory(&pem.Block{
+			Type:  "ENCRYPTED PRIVATE KEY",
+			Bytes: clientKey,
+		}), 0400); err != nil {
+			t.Fatalf("could not write client key to file")
+		}
+	} else {
+		clientKey, err := x509.MarshalPKCS8PrivateKey(clientCerts.PrivateKey)
+		if err != nil {
+			t.Fatalf("could not marshal private key: %v", err)
+		}
+		if err = os.WriteFile(clientKeyPath, pem.EncodeToMemory(&pem.Block{
+			Type:  "RSA PRIVATE KEY",
+			Bytes: clientKey,
+		}), 0400); err != nil {
+			t.Fatalf("could not write client key to file")
+		}
+	}
+
+	if err = os.WriteFile(clientCertPath, pem.EncodeToMemory(&pem.Block{
+		Type:  "CERTIFICATE",
+		Bytes: clientCerts.Leaf.Raw,
+	}), 0400); err != nil {
+		t.Fatalf("could not write client certificate to file")
+	}
+
+	return clientCertPath, clientKeyPath
+}

x-pack/filebeat/docker-compose.yml

@@ -58,6 +58,14 @@ services:
       - --addr=:8090
       - --config=/files/config.yaml
 
+  socks5-proxy:
+    image: serjs/go-socks5-proxy:latest
+    container_name: socks5-proxy
+    ports:
+      - "1080:1080"
+    environment:
+      PROXY_USER: elastic
+      PROXY_PASSWORD: changeme
   init-logstash:
     image: busybox:latest
     user: "0:0"
@@ -68,6 +76,7 @@ services:
   logstash:
     depends_on:
       - init-logstash
+      - socks5-proxy
     extends:
       file: ${ES_BEATS}/testing/environments/${STACK_ENVIRONMENT}.yml
       service: logstash

x-pack/filebeat/tests/integration/otel_lsexporter_test.go

@@ -149,6 +149,81 @@ processors:
 	compareOutputFiles(t, fbFilePath, otelFilePath, ignoredFields)
 }
 
+// TestLogstashExporterProxyURL verifies that Filebeat OTel mode can send data to Logstash via a SOCKS5 proxy.
+// Filebeat otel mode sends events to "logstash" via a socks5-proxy container running on localhost:1080
+func TestLogstashExporterProxyURL(t *testing.T) {
+	// ensure the size of events is big enough
+	numEvents := 3
+
+	var beatsCfgFile = `
+filebeat.inputs:
+  - type: filestream
+    id: filestream-input-id
+    enabled: true
+    paths:
+      - %s
+output.logstash:
+  hosts:
+    - "logstash:%s"
+  pipelining: 0
+  worker: 1
+  proxy_url: "socks5://elastic:changeme@localhost:1080"
+queue.mem.flush.timeout: 0s
+processors:
+  - add_host_metadata: ~
+  - add_fields:
+      target: ""
+      fields:
+        testcase: %s
+`
+	testCaseName := uuid.Must(uuid.NewV4()).String()
+	events := generateEvents(numEvents)
+
+	// start filebeat in otel mode
+	fbOTel := integration.NewBeat(
+		t,
+		"filebeat-otel",
+		"../../filebeat.test",
+		"otel",
+	)
+
+	inputFilePath := filepath.Join(fbOTel.TempDir(), "event.json")
+	writeEvents(t, inputFilePath, events)
+
+	fbOTel.WriteConfigFile(fmt.Sprintf(beatsCfgFile, inputFilePath, "5055", testCaseName))
+	fbOTel.Start()
+	defer fbOTel.Stop()
+
+	// Nginx endpoint URLs
+	baseURL := "http://localhost:8082"
+	outOTelFileURL := fmt.Sprintf("%s/%s_otel.json", baseURL, testCaseName)
+
+	// Logstash is outputting to a file inside its container, to access
+	// this file we use Nginx to serve the output folder via HTTP
+	// (see docker-compose.yml for Logstash and Nginx configuration).
+	// Wait to ensure the file can be downloaded via HTTP, the file
+	// being available indicates that Filebeat successfully sent data
+	// to Logstash
+	require.EventuallyWithTf(t,
+		func(ct *assert.CollectT) {
+			for _, url := range []string{outOTelFileURL} {
+				checkURLHasContent(ct, url)
+			}
+		},
+		30*time.Second,
+		1*time.Second,
+		"did not find Logstash output file served via Nginx")
+
+	// download files from Nginx
+	otelFilePath := downloadToTestData(t, outOTelFileURL, fmt.Sprintf("%s_otel.json", testCaseName))
+	otelEvents, err := readAllEvents(otelFilePath)
+
+	require.NoError(t, err, "failed to read otel events")
+	require.Equal(t, numEvents, len(otelEvents),
+		"different number of events: sent=%d, get=%d", numEvents, len(otelEvents))
+
+}
+
 func generateEvents(numEvents int) []string {
 	gofakeit.Seed(time.Now().UnixNano())