diff --git a/changelog/fragments/1773394006-state-store-extension.yaml b/changelog/fragments/1773394006-state-store-extension.yaml new file mode 100644 index 00000000000..687e784a898 --- /dev/null +++ b/changelog/fragments/1773394006-state-store-extension.yaml @@ -0,0 +1,45 @@ +# REQUIRED +# Kind can be one of: +# - breaking-change: a change to previously-documented behavior +# - deprecation: functionality that is being removed in a later release +# - bug-fix: fixes a problem in a previous version +# - enhancement: extends functionality but does not break or fix existing behavior +# - feature: new functionality +# - known-issue: problems that we are aware of in a given version +# - security: impacts on the security of a product or a user’s deployment. +# - upgrade: important information for someone upgrading from a prior version +# - other: does not fit into any of the other categories +kind: bug-fix + +# REQUIRED for all kinds +# Change summary; a 80ish characters long description of the change. +summary: Fix event ingestion for agentless in otel mode. + +# REQUIRED for breaking-change, deprecation, known-issue +# Long description; in case the summary is not enough to describe the change +# this field accommodate a description without length limits. +# description: + +# REQUIRED for breaking-change, deprecation, known-issue +# impact: + +# REQUIRED for breaking-change, deprecation, known-issue +# action: + +# REQUIRED for all kinds +# Affected component; usually one of "elastic-agent", "fleet-server", "filebeat", "metricbeat", "auditbeat", "all", etc. +component: elastic-agent + +# AUTOMATED +# OPTIONAL to manually add other PR URLs +# PR URL: A link the PR that added the changeset. +# If not present is automatically filled by the tooling finding the PR where this changelog fragment has been added. +# NOTE: the tooling supports backports, so it's able to fill the original PR number instead of the backport PR number. +# Please provide it if you are adding a fragment for a different PR. +# pr: https://github.com/owner/repo/1234 + +# AUTOMATED +# OPTIONAL to manually add other issue URLs +# Issue URL; optional; the GitHub issue related to this changeset (either closes or is part of). +# If not present is automatically filled by the tooling with the issue linked to the PR number. +# issue: https://github.com/owner/repo/1234 diff --git a/internal/edot/otelcol/components.go b/internal/edot/otelcol/components.go index 76bd9a2f56d..a94035f7126 100644 --- a/internal/edot/otelcol/components.go +++ b/internal/edot/otelcol/components.go @@ -101,6 +101,7 @@ import ( filestorage "github.com/open-telemetry/opentelemetry-collector-contrib/extension/storage/filestorage" "go.opentelemetry.io/collector/extension/memorylimiterextension" // for putting backpressure when approach a memory limit + elasticsearchstorage "github.com/elastic/beats/v7/x-pack/otel/extension/elasticsearchstorage" elasticdiagnostics "github.com/elastic/elastic-agent/internal/pkg/otel/extension/elasticdiagnostics" "github.com/elastic/opentelemetry-collector-components/extension/apikeyauthextension" @@ -254,6 +255,7 @@ func components(extensionFactories ...extension.Factory) func() (otelcol.Factori headersetterextension.NewFactory(), beatsauthextension.NewFactory(), elasticdiagnostics.NewFactory(), + elasticsearchstorage.NewFactory(), awslogsencodingextension.NewFactory(), opampextension.NewFactory(), } diff --git a/internal/pkg/otel/translate/otelconfig.go b/internal/pkg/otel/translate/otelconfig.go index 38cb3a0d7f7..8e2ec3e8c3d 100644 --- a/internal/pkg/otel/translate/otelconfig.go +++ b/internal/pkg/otel/translate/otelconfig.go @@ -23,6 +23,7 @@ import ( "go.opentelemetry.io/collector/pipeline" "golang.org/x/exp/maps" + fbfeatures "github.com/elastic/beats/v7/filebeat/features" "github.com/elastic/beats/v7/libbeat/outputs/elasticsearch" "github.com/elastic/beats/v7/x-pack/libbeat/management" "github.com/elastic/beats/v7/x-pack/otel/extension/beatsauthextension" @@ -42,6 +43,7 @@ const ( outputOtelOverrideFieldName = "otel" outputOtelOverrideExporterFieldName = "exporter" outputOtelOverrideExtensionsFieldName = "extensions" + elasticsearchStateStoreExtensionName = "elasticsearch_storage" ) // BeatMonitoringConfigGetter is a function that returns the monitoring configuration for a beat receiver. @@ -390,6 +392,9 @@ func getReceiversConfigForComponent( receiverConfig[beatName] = map[string]any{ "inputs": inputs, } + if fbfeatures.IsElasticsearchStateStoreEnabled() { + receiverConfig["storage"] = elasticsearchStateStoreExtensionName + } case "metricbeat": receiverConfig[beatName] = map[string]any{ "modules": inputs, @@ -570,7 +575,12 @@ func unitToExporterConfig(unit component.Unit, outputName string, exporterType o exporterConfig["auth"] = map[string]any{ "authenticator": extensionID.String(), } - + if fbfeatures.IsElasticsearchStateStoreEnabled() { + // Add elasticsearch state store extension for agentless mode + // We paste the config as is, without any translation. + // The state store extension will pick up relevant settings from it and ignore the rest. + extensionCfg[elasticsearchStateStoreExtensionName] = unitConfigMap + } } return exporterConfig, queueSettings, extensionCfg, processors, nil diff --git a/testing/integration/ess/otel_test.go b/testing/integration/ess/otel_test.go index fd7c81185a2..ff6b360667b 100644 --- a/testing/integration/ess/otel_test.go +++ b/testing/integration/ess/otel_test.go @@ -13,6 +13,8 @@ import ( "encoding/pem" "errors" "fmt" + "net" + "net/http" "os" "path/filepath" "strings" @@ -2537,6 +2539,275 @@ service: cancel() } +func TestOtelElasticsearchStateStore_Agentless(t *testing.T) { + info := define.Require(t, define.Requirements{ + Group: integration.Default, + Local: true, + OS: []define.OS{ + {Type: define.Linux}, + {Type: define.Darwin}, + }, + Stack: &define.Stack{}, + }) + + esEndpoint, err := integration.GetESHost() + require.NoError(t, err, "error getting elasticsearch endpoint") + esApiKey := createESApiKey(t, info.ESClient) + decodedApiKey, err := getDecodedApiKey(esApiKey) + require.NoError(t, err) + + mock := newHTTPJSONCursorMockServer(t) + defer mock.Close() + + type configOptions struct { + ESEndpoint string + ESApiKey string + InputURL string + } + + configTemplate := ` +agent.internal.runtime.filebeat.httpjson: otel +agent.monitoring.enabled: false +inputs: + - type: httpjson + id: httpjson-es-state-store + use_output: default + streams: + - id: httpjson-cursor-stream + data_stream: + dataset: httpjson_statestore + request.url: {{.InputURL}} + request.method: GET + request.transforms: + - set: + target: url.params.since + value: '[[.cursor.published]]' + default: '[[formatDate (now (parseDuration "-24h")) "RFC3339"]]' + cursor: + published: + value: '[[.last_event.published]]' + interval: 5s +outputs: + default: + type: elasticsearch + hosts: [{{.ESEndpoint}}] + api_key: "{{.ESApiKey}}" + preset: balanced +agent.grpc: + # listen address for the GRPC server that spawned processes connect back to. + address: localhost +# # port for the GRPC server that spawned processes connect back to. + port: 6794 + +` + var configBuffer bytes.Buffer + require.NoError(t, + template.Must(template.New("config").Parse(configTemplate)).Execute(&configBuffer, + configOptions{ + ESEndpoint: esEndpoint, + ESApiKey: decodedApiKey, + InputURL: mock.URL, + })) + + fixture, err := define.NewFixtureFromLocalBuild(t, define.Version()) + require.NoError(t, err) + + ctx, cancel := testcontext.WithDeadline(t, t.Context(), time.Now().Add(5*time.Minute)) + err = fixture.Prepare(ctx) + require.NoError(t, err) + + err = fixture.Configure(ctx, configBuffer.Bytes()) + require.NoError(t, err) + + dataIndex := ".ds-logs-httpjson_statestore-*" + + cmd, err := fixture.PrepareAgentCommand(ctx, nil) + require.NoError(t, err) + + cmd.Env = append(os.Environ(), "AGENTLESS_ELASTICSEARCH_STATE_STORE_INPUT_TYPES=httpjson") + + output := &strings.Builder{} + cmd.Stderr = output + cmd.Stdout = output + + err = cmd.Start() + require.NoError(t, err) + + t.Cleanup(func() { + _ = cmd.Wait() + if t.Failed() { + t.Log("Elastic-Agent output (1st run):") + t.Log(output.String()) + } + }) + + // Wait for data to arrive in the data index + require.EventuallyWithT(t, + func(ct *assert.CollectT) { + findCtx, findCancel := context.WithTimeout(ctx, 10*time.Second) + defer findCancel() + + docs, err := estools.GetAllLogsForIndexWithContext(findCtx, info.ESClient, dataIndex) + assert.NoError(ct, err) + assert.GreaterOrEqual(ct, docs.Hits.Total.Value, 1, "expected at least 1 event") + }, + 2*time.Minute, 1*time.Second, "expected data in the data index") + + // Wait for at least 2 polling cycles so the cursor is persisted to ES + require.Eventually(t, func() bool { + return mock.RequestCount() >= 2 + }, 60*time.Second, 1*time.Second, "expected at least 2 httpjson poll cycles") + + // Verify state exists in the agentless-state-* index + require.EventuallyWithT(t, + func(ct *assert.CollectT) { + findCtx, findCancel := context.WithTimeout(ctx, 10*time.Second) + defer findCancel() + + docs, err := estools.GetAllLogsForIndexWithContext(findCtx, info.ESClient, "agentless-state-*") + assert.NoError(ct, err) + assert.GreaterOrEqual(ct, docs.Hits.Total.Value, 1, "expected state document in agentless-state-* index") + }, + 30*time.Second, 1*time.Second, "expected cursor state in agentless-state-* index") + + // Record the cursor timestamp just before stopping + lastCursorBeforeStop := mock.LastSinceParam() + require.False(t, lastCursorBeforeStop.IsZero(), "expected at least one 'since' parameter before stopping") + t.Logf("cursor before stop: %s (request count: %d)", lastCursorBeforeStop.Format(time.RFC3339), mock.RequestCount()) + + // Stop the first agent + requestCountBeforeRestart := mock.RequestCount() + cancel() + cmd.Wait() + + mock.MarkRestart() + + ctx, cancel = testcontext.WithDeadline(t, t.Context(), time.Now().Add(5*time.Minute)) + cmd2, err := fixture.PrepareAgentCommand(ctx, nil) + require.NoError(t, err) + + cmd2.Env = append(os.Environ(), "AGENTLESS_ELASTICSEARCH_STATE_STORE_INPUT_TYPES=httpjson") + + output2 := &strings.Builder{} + cmd2.Stderr = output2 + cmd2.Stdout = output2 + + err = cmd2.Start() + require.NoError(t, err) + t.Cleanup(func() { + if cmd2.Process != nil { + _ = cmd2.Process.Kill() + } + _ = cmd2.Wait() + if t.Failed() { + t.Log("Elastic-Agent output (2nd run):") + t.Log(output2.String()) + } + }) + + // Wait for the httpjson input to resume polling after restart + require.Eventually(t, func() bool { + return mock.RequestCount() > requestCountBeforeRestart + }, 60*time.Second, 1*time.Second, "expected httpjson to resume polling after restart") + + // Verify the restored cursor: the first "since" parameter after restart + // should be recent (within 5 minutes), NOT the default 24h-ago value. + // If the cursor was not restored from ES, the httpjson input would fall + // back to its default: formatDate (now (parseDuration "-24h")). + firstSinceAfterRestart := mock.FirstSinceAfterRestart() + require.False(t, firstSinceAfterRestart.IsZero(), "expected a 'since' parameter after restart") + + timeSinceCursor := time.Since(firstSinceAfterRestart) + t.Logf("first 'since' after restart: %s (age: %s)", firstSinceAfterRestart.Format(time.RFC3339), timeSinceCursor) + + assert.Less(t, timeSinceCursor, 5*time.Minute, + "restored cursor should be recent (within 5 minutes), not the default 24h ago; "+ + "got since=%s which is %s old", firstSinceAfterRestart.Format(time.RFC3339), timeSinceCursor) + + cancel() +} + +// httpJSONCursorMockServer is a test HTTP server for the httpjson input that +// tracks the "since" cursor parameter across requests and restarts. +type httpJSONCursorMockServer struct { + URL string + server *http.Server + + mu sync.Mutex + requestCount int64 + lastSince time.Time + restartMarked bool + firstSinceAfterRestart time.Time +} + +func newHTTPJSONCursorMockServer(t *testing.T) *httpJSONCursorMockServer { + t.Helper() + mock := &httpJSONCursorMockServer{} + + mux := http.NewServeMux() + mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { + mock.mu.Lock() + mock.requestCount++ + + // Parse the "since" query parameter sent by the httpjson cursor + sinceParam := r.URL.Query().Get("since") + if sinceParam != "" { + if parsed, err := time.Parse(time.RFC3339, sinceParam); err == nil { + mock.lastSince = parsed + if mock.restartMarked && mock.firstSinceAfterRestart.IsZero() { + mock.firstSinceAfterRestart = parsed + } + } + } + mock.mu.Unlock() + + // Respond with a JSON body containing "published" — the httpjson + // cursor will store this and send it back as "since" next time. + published := time.Now().UTC() + w.Header().Set("Content-Type", "application/json") + fmt.Fprintf(w, `{"message":"hello","published":"%s"}`, published.Format(time.RFC3339)) + }) + + mock.server = &http.Server{Handler: mux} + listener, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + + go func() { _ = mock.server.Serve(listener) }() + mock.URL = fmt.Sprintf("http://%s", listener.Addr().String()) + + return mock +} + +func (m *httpJSONCursorMockServer) RequestCount() int64 { + m.mu.Lock() + defer m.mu.Unlock() + return m.requestCount +} + +func (m *httpJSONCursorMockServer) LastSinceParam() time.Time { + m.mu.Lock() + defer m.mu.Unlock() + return m.lastSince +} + +func (m *httpJSONCursorMockServer) MarkRestart() { + m.mu.Lock() + defer m.mu.Unlock() + m.restartMarked = true +} + +func (m *httpJSONCursorMockServer) FirstSinceAfterRestart() time.Time { + m.mu.Lock() + defer m.mu.Unlock() + return m.firstSinceAfterRestart +} + +func (m *httpJSONCursorMockServer) Close() { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + _ = m.server.Shutdown(ctx) +} + type ZapWriter struct { logger *zap.Logger level zapcore.Level