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

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

CHANGELOG.next.asciidoc

@@ -221,6 +221,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Update Logstash module's Grok patterns to support Logstash 7.4 logs. {pull}14743[14743]
 - Fix a problem in Filebeat input httpjson where interval is not used as time.Duration. {issue}14752[14752] {pull}14753[14753]
 - Fix SSL config in input.yml for Filebeat httpjson input in the MISP module. {pull}14767[14767]
+- Check content-type when creating new reader in s3 input. {pull}15252[15252] {issue}15225[15225]
 - Fix session reset detection and a crash in Netflow input. {pull}14904[14904]
 
 *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
 }