Check when response content type is nil (#28457)

(cherry picked from commit ccc2206)

CHANGELOG.next.asciidoc

@@ -186,6 +186,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Drop aws.vpcflow.pkt_srcaddr and aws.vpcflow.pkt_dstaddr when equal to "-". {pull}22721[22721] {issue}22716[22716]
 - Improve Cisco ASA/FTD parsing of messages - better support for identity FW messages. Change network.bytes, source.bytes, and destination.bytes to long from integer since value can exceed integer capacity.  Add descriptions for various processors for easier pipeline editing in Kibana UI. {pull}23766[23766]
 - 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 {