Cherry-pick #15252 to 7.5: [Filebeat] Check content type when reading…

… s3 files (#15369)

* [Filebeat] Check content type when reading s3 files (#15252)

* Check resp.ContentType and filename
* Remove case "text/plain" to use default instead

(cherry picked from commit 034e719)

* Fix changelog

CHANGELOG.next.asciidoc

@@ -52,6 +52,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 
 - cisco/asa fileset: Fix parsing of 302021 message code. {pull}14519[14519]
 - Fix filebeat azure dashboards, event category should be `Alert`. {pull}14668[14668]
+- Check content-type when creating new reader in s3 input. {pull}15252[15252] {issue}15225[15225]
 
 *Heartbeat*
 

x-pack/filebeat/input/s3/input.go

@@ -440,16 +440,28 @@ func (p *s3Input) newS3BucketReader(svc s3iface.ClientAPI, s3Info s3Info) (*bufi
 		return nil, errors.New("s3 get object response body is empty")
 	}
 
+	// Check content-type
+	if resp.ContentType != nil {
+		switch *resp.ContentType {
+		case "application/x-gzip":
+			reader, err := gzip.NewReader(resp.Body)
+			if err != nil {
+				return nil, errors.Wrapf(err, "Failed to decompress gzipped file %v", s3Info.key)
+			}
+			return bufio.NewReader(reader), nil
+		default:
+			return bufio.NewReader(resp.Body), nil
+		}
+	}
+
+	// If there is no content-type, check file name instead.
 	if strings.HasSuffix(s3Info.key, ".gz") {
 		gzipReader, err := gzip.NewReader(resp.Body)
-
 		if err != nil {
 			return nil, errors.Wrapf(err, "Failed to decompress gzipped file %v", s3Info.key)
 		}
-
 		return bufio.NewReader(gzipReader), nil
 	}
-
 	return bufio.NewReader(resp.Body), nil
 }