Fix silent event drop when a parser clears message content #52077

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged

MichaelKatsoulis merged 8 commits into elastic:main from MichaelKatsoulis:fix-awss3-readfile-drops-parser-fields

Jul 22, 2026

+88 −14

changelog/fragments/1784536512-fix-awss3-readfile-drops-parser-fields.yaml

-Original file line number
+Diff line change
@@ -0,0 +1,46 @@
+    # 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 aws-s3 input silently dropping events when a parser clears the message content
+    # 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: >
+      A parser can move the decoded data into the event fields and clear the
+      content, for example an ndjson parser configured without a message_key. When
+      such an object was read through the readFile path (any content-type other than
+      application/json or application/x-ndjson), the input dropped every event with
+      no error logged, because it only published when the message content was
+      non-empty. Events are now published when either the content or the fields are
+      populated.
+    # REQUIRED for all kinds
+    # Affected component; usually one of "elastic-agent", "fleet-server", "filebeat", "metricbeat", "auditbeat", "all", etc.
+    component: filebeat
+    # 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/elastic/beats/pull/52077
+    # 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

x-pack/filebeat/input/awss3/s3_objects.go

            
                      Original file line number
                      Diff line number
                      Diff line change
                  
    @@ -47,13 +47,13 @@ type s3ObjectProcessor struct {
  
    	s3ObjHash     string

    	s3RequestURL  string

    	s3Metadata map[string]interface{} // S3 object metadata.

    	s3Metadata map[string]any // S3 object metadata.

    }

    type s3DownloadedObject struct {

    	body        io.ReadCloser

    	contentType string

    	metadata    map[string]interface{}

    	metadata    map[string]any

    }

    const (

    @@ -394,7 +394,11 @@ func (p *s3ObjectProcessor) readFile(r io.Reader, logger *logp.Logger) error {
  
    	var offset int64

    	for {

    		message, err := reader.Next()

    		if len(message.Content) > 0 {

    		// Publish when the reader produced content OR fields. A parser such as

    		// ndjson without a message_key clears message.Content and moves the

    		// decoded data into message.Fields; guarding on Content alone silently

    		// drops those events.

    		if len(message.Content) > 0 || len(message.Fields) > 0 {

    			event := p.createEvent(string(message.Content), offset)

    			event.Fields.DeepUpdate(message.Fields)

    			offset += int64(message.Bytes)

    @@ -501,7 +505,7 @@ func s3Metadata(resp *s3.GetObjectOutput, keys ...string) mapstr.M {
  
    	// other HTTP headers.

    	const userMetaPrefix = "x-amz-meta-"

    	allMeta := map[string]interface{}{}

    	allMeta := map[string]any{}

    	// Get headers using AWS SDK struct tags.

    	fields := reflect.TypeOf(resp).Elem()

    @@ -525,7 +529,7 @@ func s3Metadata(resp *s3.GetObjectOutput, keys ...string) mapstr.M {
  
    		v := values.Field(i)

    		switch v.Kind() {

    		case reflect.Ptr:

    		case reflect.Pointer:

    			if v.IsNil() {

    				continue

    			}

x-pack/filebeat/input/awss3/s3_objects_test.go

-Original file line number
+Diff line change
@@ Expand Up / @@ -7,7 +7,6 @@ package awss3 @@
     import (
     	"bytes"
     	"context"
-    	"errors"
     	"io"
     	"os"
     	"path/filepath"
@@ Expand Down Expand Up / @@ -68,10 +67,10 @@ func TestS3ObjectProcessor(t *testing.T) { @@
     		// Unfortunately the config structs for the parser package are not
     		// exported to use config parsing.
-    		cfg := conf.MustNewConfigFrom(map[string]interface{}{
-    			"parsers": []map[string]interface{}{
+    		cfg := conf.MustNewConfigFrom(map[string]any{
+    			"parsers": []map[string]any{
     				{
-    					"multiline": map[string]interface{}{
+    					"multiline": map[string]any{
     						"pattern": "^<Event",
     						"negate":  true,
     						"match":   "after",
@@ Expand All / @@ -92,6 +91,31 @@ func TestS3ObjectProcessor(t *testing.T) { @@
     		testProcessS3Object(t, "testdata/log.ndjson", "application/x-ndjson", 2)
     	})
+    	t.Run("ndjson parser without message_key is not dropped in readFile", func(t *testing.T) {
+    		// Regression test: an ndjson parser without a message_key moves the
+    		// decoded data into the event fields and clears the content. When the
+    		// object's content-type routes it through readFile (i.e. it is not
+    		// application/json or application/x-ndjson), the events must still be
+    		// published from their fields rather than silently dropped.
+    		sel := fileSelectorConfig{ReaderConfig: readerConfig{}}
+    		sel.ReaderConfig.InitDefaults()
+    		cfg := conf.MustNewConfigFrom(map[string]any{
+    			"parsers": []map[string]any{
+    				{"ndjson": map[string]any{}},
+    			},
+    		})
+    		require.NoError(t, cfg.Unpack(&sel.ReaderConfig.Parsers))
+    		events := testProcessS3Object(t, "testdata/log.ndjson", "text/plain", 2, sel)
+    		require.Len(t, events, 2)
+    		// The decoded JSON must be present on the published events' fields.
+    		got, err := events[0].GetValue("message")
+    		require.NoError(t, err)
+    		assert.Equal(t, "error making http request", got)
+    	})
     	t.Run("configured content-type", func(t *testing.T) {
     		sel := fileSelectorConfig{ReaderConfig: readerConfig{ContentType: contentTypeJSON}}
     		testProcessS3Object(t, "testdata/multiline.json", "application/octet-stream", 2, sel)
@@ Expand Down Expand Up / @@ -158,7 +182,7 @@ func TestS3ObjectProcessor(t *testing.T) { @@
     		s3ObjProc := newS3ObjectProcessorFactory(nil, mockS3API, nil, backupConfig{}, logp.NewNopLogger())
     		err := s3ObjProc.Create(ctx, s3Event).ProcessS3Object(logptest.NewTestingLogger(t, inputName), func(_ beat.Event) {})
     		require.Error(t, err)
-    		assert.True(t, errors.Is(err, errS3DownloadFailed), "expected errS3DownloadFailed")
+    		assert.ErrorIs(t, err, errS3DownloadFailed, "expected errS3DownloadFailed")
     	})
     	t.Run("s3 object streaming error results in errS3DownloadFailed error", func(t *testing.T) {
@@ Expand Down Expand Up / @@ -227,7 +251,7 @@ func TestS3ObjectProcessor(t *testing.T) { @@
     		err := s3ObjProc.Create(ctx, s3Event).ProcessS3Object(logptest.NewTestingLogger(t, inputName), func(event beat.Event) {
     			events = append(events, event)
     		})
-    		assert.Equal(t, 2, len(events))
+    		assert.Len(t, events, 2)
     		require.NoError(t, err)
     	})
@@ Expand Down Expand Up / @@ -318,7 +342,7 @@ func TestS3ObjectProcessor(t *testing.T) { @@
     }
     func TestProcessObjectMetricCollection(t *testing.T) {
-    	logger := logp.NewLogger("testing-s3-processor-metrics")
+    	logger := logptest.NewTestingLogger(t, "testing-s3-processor-metrics")
     	tests := []struct {
     		name        string
@@ Expand Down Expand Up / @@ -378,7 +402,7 @@ func TestProcessObjectMetricCollection(t *testing.T) { @@
     			require.Equal(t, uint64(0), metricRecorder.s3ObjectsInflight.Get())
     			values := metricRecorder.s3ObjectSizeInBytes.Values()
-    			require.Equal(t, 1, len(values))
+    			require.Len(t, values, 1)
     			// since we processed a single object, total and current process size is same
     			require.Equal(t, test.objectSize, values[0])
@@ Expand Down Expand Up @@
     	if !expectErr {
     		require.NoError(t, err)
-    		assert.Equal(t, numEvents, len(events))
+    		assert.Len(t, events, numEvents)
     	} else {
     		require.Error(t, err)
     	}
@@ Expand Down @@

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Fix silent event drop when a parser clears message content #52077

Diff view

Diff view

There are no files selected for viewing

Uh oh!

Uh oh!