Check when response content type is nil (elastic#28457)

CHANGELOG.next.asciidoc

@@ -322,6 +322,7 @@ for a few releases. Please use other tools provided by Elastic to fetch data fro
 - Relax time parsing and capture group and session type in Cisco ASA module {issue}24710[24710] {pull}28325[28325]
 - Correctly track bytes read when max_bytes is exceeded. {issue}28317[28317] {pull}28352[28352]
 - Fix initialization of http client in Cloudfoundry input. {issue}28271[28271] {pull}28277[28277]
+- Fix aws-s3 input by checking if GetObject API call response content type exists. {pull}28457[28457]
 
 *Heartbeat*
 

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

@@ -174,10 +174,13 @@ func (p *s3ObjectProcessor) download() (contentType string, metadata map[string]
 	}
 
 	if resp == nil {
-		return "", nil, nil, errors.New("empty reponse from s3 get object")
+		return "", nil, nil, errors.New("empty response from s3 get object")
 	}
 
 	meta := s3Metadata(resp, p.readerConfig.IncludeS3Metadata...)
+	if resp.ContentType == nil {
+		return "", meta, resp.Body, nil
+	}
 	return *resp.ContentType, meta, resp.Body, nil
 }
 

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

@@ -40,9 +40,13 @@ func newS3GetObjectResponse(filename string, data []byte, contentType string) *s
 		GetObjectOutput: &s3.GetObjectOutput{
 			Body:          ioutil.NopCloser(r),
 			ContentLength: &contentLen,
-			ContentType:   &contentType,
 		},
 	}
+
+	if contentType != "" {
+		resp.ContentType = &contentType
+	}
+
 	switch strings.ToLower(filepath.Ext(filename)) {
 	case ".gz":
 		gzipEncoding := "gzip"
@@ -169,6 +173,33 @@ func TestS3ObjectProcessor(t *testing.T) {
 		err := s3ObjProc.Create(ctx, logp.NewLogger(inputName), ack, s3Event).ProcessS3Object()
 		require.Error(t, err)
 	})
+
+	t.Run("no content type in GetObject response", func(t *testing.T) {
+		ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
+		defer cancel()
+
+		ctrl, ctx := gomock.WithContext(ctx, t)
+		defer ctrl.Finish()
+		mockS3API := NewMockS3API(ctrl)
+		mockPublisher := NewMockBeatClient(ctrl)
+		s3Event, s3Resp := newS3Object(t, "testdata/log.txt", "")
+
+		var events []beat.Event
+		gomock.InOrder(
+			mockS3API.EXPECT().
+				GetObject(gomock.Any(), gomock.Eq(s3Event.S3.Bucket.Name), gomock.Eq(s3Event.S3.Object.Key)).
+				Return(s3Resp, nil),
+			mockPublisher.EXPECT().
+				Publish(gomock.Any()).
+				Do(func(event beat.Event) { events = append(events, event) }).
+				Times(2),
+		)
+
+		s3ObjProc := newS3ObjectProcessorFactory(logp.NewLogger(inputName), nil, mockS3API, mockPublisher, nil)
+		ack := newEventACKTracker(ctx)
+		err := s3ObjProc.Create(ctx, logp.NewLogger(inputName), ack, s3Event).ProcessS3Object()
+		require.NoError(t, err)
+	})
 }
 
 func testProcessS3Object(t testing.TB, file, contentType string, numEvents int, selectors ...fileSelectorConfig) []beat.Event {