Cherry-pick elastic#20370 to 7.9: Check expand_event_list_from_field …

…when json in map[string]interface{} format (elastic#20373)

(cherry picked from commit c39dde3)

CHANGELOG.next.asciidoc

@@ -126,6 +126,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Fix S3 input to trim delimiter /n from each log line. {pull}19972[19972]
 - Fix s3 input parsing json file without expand_event_list_from_field. {issue}19902[19902] {pull}19962[19962]
 - Ignore missing in Zeek module when dropping unnecessary fields. {pull}19984[19984]
+- Fix s3 input parsing json file without expand_event_list_from_field. {issue}19902[19902] {pull}19962[19962] {pull}20370[20370]
 - Fix millisecond timestamp normalization issues in CrowdStrike module {issue}20035[20035], {pull}20138[20138]
 - Fix support for message code 106100 in Cisco ASA and FTD. {issue}19350[19350] {pull}20245[20245]
 - Fix `fortinet` setting `event.timezone` to the system one when no `tz` field present {pull}20273[20273]

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

@@ -526,10 +526,11 @@ func (p *s3Input) decodeJSON(decoder *json.Decoder, objectHash string, s3Info s3
 			return nil
 		}
 
-		offset, err = p.jsonFieldsType(jsonFields, offset, objectHash, s3Info, s3Ctx)
+		offsetNew, err := p.jsonFieldsType(jsonFields, offset, objectHash, s3Info, s3Ctx)
 		if err != nil {
 			return err
 		}
+		offset = offsetNew
 	}
 }
 
@@ -554,6 +555,27 @@ func (p *s3Input) jsonFieldsType(jsonFields interface{}, offset int, objectHash
 			return offset, nil
 		}
 	case map[string]interface{}:
+		if p.config.ExpandEventListFromField != "" {
+			textValues, ok := f[p.config.ExpandEventListFromField]
+			if !ok {
+				err := errors.Errorf("key '%s' not found", p.config.ExpandEventListFromField)
+				p.logger.Error(err)
+				return offset, err
+			}
+
+			valuesConverted := textValues.([]interface{})
+			for _, textValue := range valuesConverted {
+				offsetNew, err := p.convertJSONToEvent(textValue, offset, objectHash, s3Info, s3Ctx)
+				if err != nil {
+					err = errors.Wrapf(err, "convertJSONToEvent failed for '%s' from S3 bucket '%s'", s3Info.key, s3Info.name)
+					p.logger.Error(err)
+					return offset, err
+				}
+				offset = offsetNew
+			}
+			return offset, nil
+		}
+
 		offset, err := p.convertJSONToEvent(f, offset, objectHash, s3Info, s3Ctx)
 		if err != nil {
 			err = errors.Wrapf(err, "convertJSONToEvent failed for '%s' from S3 bucket '%s'", s3Info.key, s3Info.name)