Skip to content

Commit 133ab6a

Browse files
khushijain21khushijain21
authored
[beatreceiver] Add e2e tests for tcp and udp inputs (#51705)
* [beatreceiver] Add e2e tests for tcp and udp inputs
1 parent 4a6a42b commit 133ab6a

1 file changed

Lines changed: 292 additions & 0 deletions

File tree

‎x-pack/filebeat/tests/integration/otel_tcp_udp_test.go‎

Lines changed: 292 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,292 @@
1+
// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
2+
// or more contributor license agreements. Licensed under the Elastic License;
3+
// you may not use this file except in compliance with the Elastic License.
4+
5+
//go:build integration
6+
7+
package integration
8+
9+
import (
10+
"bytes"
11+
"context"
12+
"fmt"
13+
"strings"
14+
"testing"
15+
"text/template"
16+
"time"
17+
18+
"github.com/gofrs/uuid/v5"
19+
"github.com/stretchr/testify/assert"
20+
"github.com/stretchr/testify/require"
21+
22+
"github.com/elastic/beats/v7/filebeat/input/net/nettest"
23+
"github.com/elastic/beats/v7/libbeat/tests/integration"
24+
"github.com/elastic/beats/v7/x-pack/otel/oteltest"
25+
"github.com/elastic/beats/v7/x-pack/otel/oteltestcol"
26+
"github.com/elastic/elastic-agent-libs/testing/estools"
27+
)
28+
29+
const (
30+
tcpInputTestMsg = "tcp-input-otel-e2e-test-event"
31+
udpInputTestMsg = "udp-input-otel-e2e-test-event"
32+
)
33+
34+
func TestTCPInputOTelE2E(t *testing.T) {
35+
// TODO: change this to use port from log lines
36+
// See https://github.com/elastic/beats/pull/51617
37+
otelServerAddr := "127.0.0.1:9042"
38+
fbServerAddr := "127.0.0.1:9043"
39+
40+
runSocketInputOTelE2E(
41+
t,
42+
"tcp",
43+
tcpInputTestMsg,
44+
otelServerAddr,
45+
fbServerAddr,
46+
nettest.RunTCPClient,
47+
)
48+
}
49+
50+
func TestUDPInputOTelE2E(t *testing.T) {
51+
// TODO: change this to use port from log lines
52+
// See https://github.com/elastic/beats/pull/51617
53+
otelServerAddr := "127.0.0.1:9042"
54+
fbServerAddr := "127.0.0.1:9043"
55+
56+
runSocketInputOTelE2E(
57+
t,
58+
"udp",
59+
udpInputTestMsg,
60+
otelServerAddr,
61+
fbServerAddr,
62+
nettest.RunUDPClient,
63+
)
64+
}
65+
66+
type socketClientFn func(t *testing.T, address string, data []string)
67+
68+
func runSocketInputOTelE2E(
69+
t *testing.T,
70+
inputType, testMessage, otelAddress, fbAddress string,
71+
runClient socketClientFn,
72+
) {
73+
t.Helper()
74+
integration.EnsureESIsRunning(t)
75+
76+
otelHome := t.TempDir()
77+
78+
host := integration.GetESURL(t, "http")
79+
user := host.User.Username()
80+
password, _ := host.User.Password()
81+
82+
otelNamespace := strings.ReplaceAll(uuid.Must(uuid.NewV4()).String(), "-", "")
83+
fbNamespace := strings.ReplaceAll(uuid.Must(uuid.NewV4()).String(), "-", "")
84+
85+
otelIndex := "logs-integration-" + otelNamespace
86+
fbIndex := "logs-integration-" + fbNamespace
87+
88+
data := []string{testMessage}
89+
90+
type options struct {
91+
InputType string
92+
Index string
93+
ESURL string
94+
Username string
95+
Password string
96+
Host string
97+
PathHome string
98+
}
99+
100+
filebeatConfig := `filebeat.inputs:
101+
- type: {{ .InputType }}
102+
id: {{ .InputType }}-input-e2e
103+
host: {{ .Host }}
104+
105+
output:
106+
elasticsearch:
107+
hosts:
108+
- {{ .ESURL }}
109+
username: {{ .Username }}
110+
password: {{ .Password }}
111+
index: {{ .Index }}
112+
113+
queue.mem.flush.timeout: 0s
114+
setup.template.enabled: false
115+
processors:
116+
- add_host_metadata: ~
117+
- add_cloud_metadata: ~
118+
- add_docker_metadata: ~
119+
- add_kubernetes_metadata: ~
120+
`
121+
122+
otelConfig := `exporters:
123+
elasticsearch:
124+
auth:
125+
authenticator: beatsauth
126+
compression: gzip
127+
compression_params:
128+
level: 1
129+
endpoints:
130+
- {{ .ESURL }}
131+
logs_index: {{ .Index }}
132+
max_conns_per_host: 1
133+
password: {{ .Password }}
134+
retry:
135+
enabled: true
136+
initial_interval: 1s
137+
max_interval: 1m0s
138+
max_retries: 3
139+
sending_queue:
140+
batch:
141+
flush_timeout: 10s
142+
max_size: 1600
143+
min_size: 0
144+
sizer: items
145+
block_on_overflow: true
146+
enabled: true
147+
num_consumers: 1
148+
queue_size: 3200
149+
wait_for_result: true
150+
user: {{ .Username }}
151+
extensions:
152+
beatsauth:
153+
idle_connection_timeout: 3s
154+
proxy_disable: false
155+
timeout: 1m30s
156+
receivers:
157+
filebeatreceiver:
158+
filebeat:
159+
inputs:
160+
- type: {{ .InputType }}
161+
id: {{ .InputType }}-input-e2e
162+
host: {{ .Host }}
163+
processors:
164+
- add_host_metadata: ~
165+
- add_cloud_metadata: ~
166+
- add_docker_metadata: ~
167+
- add_kubernetes_metadata: ~
168+
queue.mem.flush.timeout: 0s
169+
setup.template.enabled: false
170+
path.home: {{ .PathHome }}
171+
service:
172+
extensions:
173+
- beatsauth
174+
pipelines:
175+
logs:
176+
exporters:
177+
- elasticsearch
178+
receivers:
179+
- filebeatreceiver
180+
telemetry:
181+
metrics:
182+
level: none
183+
`
184+
185+
optionsValue := options{
186+
InputType: inputType,
187+
ESURL: fmt.Sprintf("%s://%s", host.Scheme, host.Host),
188+
Username: user,
189+
Password: password,
190+
PathHome: otelHome,
191+
}
192+
193+
var configBuffer bytes.Buffer
194+
optionsValue.Host = otelAddress
195+
optionsValue.Index = otelIndex
196+
require.NoError(t, template.Must(template.New("config").Parse(otelConfig)).Execute(&configBuffer, optionsValue))
197+
198+
oteltestcol.New(t, configBuffer.String())
199+
200+
configBuffer.Reset()
201+
202+
optionsValue.Host = fbAddress
203+
optionsValue.Index = fbIndex
204+
require.NoError(t, template.Must(template.New("config").Parse(filebeatConfig)).Execute(&configBuffer, optionsValue))
205+
206+
filebeat := integration.NewBeat(
207+
t,
208+
"filebeat",
209+
"../../filebeat.test",
210+
)
211+
filebeat.WriteConfigFile(configBuffer.String())
212+
filebeat.Start()
213+
defer filebeat.Stop()
214+
215+
filebeat.WaitLogsContainsAnyOrder(
216+
[]string{
217+
"filebeat start running",
218+
},
219+
20*time.Second,
220+
"filebeat did not run",
221+
)
222+
223+
go runClient(t, otelAddress, data)
224+
go runClient(t, fbAddress, data)
225+
226+
es := integration.GetESClient(t, "http")
227+
228+
t.Cleanup(func() {
229+
_, err := es.Indices.DeleteDataStream([]string{
230+
otelIndex,
231+
fbIndex,
232+
})
233+
require.NoError(t, err, "failed to delete data streams")
234+
})
235+
236+
rawQuery := map[string]any{
237+
"query": map[string]any{
238+
"bool": map[string]any{
239+
"must": []map[string]any{
240+
{
241+
"match_phrase": map[string]any{
242+
"input.type": inputType,
243+
},
244+
},
245+
{
246+
"match_phrase": map[string]any{
247+
"message": testMessage,
248+
},
249+
},
250+
},
251+
},
252+
},
253+
"sort": []map[string]any{
254+
{"@timestamp": map[string]any{"order": "asc"}},
255+
},
256+
}
257+
258+
var filebeatDocs estools.Documents
259+
var otelDocs estools.Documents
260+
var err error
261+
262+
require.EventuallyWithTf(t,
263+
func(ct *assert.CollectT) {
264+
findCtx, findCancel := context.WithTimeout(t.Context(), 900*time.Millisecond)
265+
defer findCancel()
266+
267+
otelDocs, err = estools.PerformQueryForRawQuery(findCtx, rawQuery, ".ds-"+otelIndex+"*", es)
268+
assert.NoError(ct, err)
269+
assert.GreaterOrEqual(ct, otelDocs.Hits.Total.Value, 1, "expected at least 1 otel document, got %d", otelDocs.Hits.Total.Value)
270+
271+
filebeatDocs, err = estools.PerformQueryForRawQuery(findCtx, rawQuery, ".ds-"+fbIndex+"*", es)
272+
assert.NoError(ct, err)
273+
assert.GreaterOrEqual(ct, filebeatDocs.Hits.Total.Value, 1, "expected at least 1 filebeat document, got %d", filebeatDocs.Hits.Total.Value)
274+
},
275+
3*time.Minute, 1*time.Second, "expected at least 1 document for both filebeat and otel modes")
276+
277+
filebeatDoc := filebeatDocs.Hits.Hits[0].Source
278+
otelDoc := otelDocs.Hits.Hits[0].Source
279+
ignoredFields := []string{
280+
"@timestamp",
281+
"agent.ephemeral_id",
282+
"agent.id",
283+
"log.source.address",
284+
}
285+
286+
oteltest.AssertMapsEqual(t, filebeatDoc, otelDoc, ignoredFields, "expected documents to be equal")
287+
}
288+
289+
// HostAddress returns the host:port address used by net input integration tests.
290+
func hostAddress(port uint16) string {
291+
return fmt.Sprintf("127.0.0.1:%d", port)
292+
}

0 commit comments

Comments
 (0)