Check expand_event_list_from_field when json in map[string]interface{} format (#20370)

kaiyan-sheng · web-flow · commit 2ced45488460 · 2020-07-30T21:00:34.000-06:00
diff --git a/CHANGELOG.next.asciidoc b/CHANGELOG.next.asciidoc
@@ -226,7 +226,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Ignore missing in Zeek module when dropping unecessary fields. {pull}19984[19984]
 - Fix auditd module syscall table for ppc64 and ppc64le. {pull}20052[20052]
 - Fix Filebeat OOMs on very long lines {issue}19500[19500], {pull}19552[19552]
-- Fix s3 input parsing json file without expand_event_list_from_field. {issue}19902[19902] {pull}19962[19962]
+- 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]
diff --git a/x-pack/filebeat/input/s3/input.go b/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)
