[Ingest Manager] Fix timestamp format to be processable by filebeat (#…

…24230)

[Ingest Manager] Fix timestamp format to be processable by filebeat (#24230)

libbeat/common/jsontransform/jsonhelper.go

@@ -18,6 +18,7 @@
 package jsontransform
 
 import (
+	"errors"
 	"fmt"
 	"time"
 
@@ -26,6 +27,16 @@ import (
 	"github.com/elastic/beats/v7/libbeat/logp"
 )
 
+const (
+	iso8601 = "2006-01-02T15:04:05.000Z0700"
+)
+
+var (
+	// ErrInvalidTimestamp is returned when parsing of a @timestamp field fails.
+	// Supported formats: ISO8601, RFC3339
+	ErrInvalidTimestamp = errors.New("failed to parse @timestamp, unknown format")
+)
+
 // WriteJSONKeys writes the json keys to the given event based on the overwriteKeys option and the addErrKey
 func WriteJSONKeys(event *beat.Event, keys map[string]interface{}, expandKeys, overwriteKeys, addErrKey bool) {
 	logger := logp.NewLogger("jsonhelper")
@@ -56,8 +67,8 @@ func WriteJSONKeys(event *beat.Event, keys map[string]interface{}, expandKeys, o
 				continue
 			}
 
-			// @timestamp must be of format RFC3339
-			ts, err := time.Parse(time.RFC3339, vstr)
+			// @timestamp must be of format RFC3339 or ISO8601
+			ts, err := parseTimestamp(vstr)
 			if err != nil {
 				logger.Errorf("JSON: Won't overwrite @timestamp because of parsing error: %v", err)
 				event.SetErrorWithOption(createJSONError(fmt.Sprintf("@timestamp not overwritten (parse error on %s)", vstr)), addErrKey)
@@ -110,3 +121,21 @@ func removeKeys(keys map[string]interface{}, names ...string) {
 		delete(keys, name)
 	}
 }
+
+func parseTimestamp(timestamp string) (time.Time, error) {
+	validFormats := []string{
+		time.RFC3339,
+		iso8601,
+	}
+
+	for _, f := range validFormats {
+		ts, parseErr := time.Parse(f, timestamp)
+		if parseErr != nil {
+			continue
+		}
+
+		return ts, nil
+	}
+
+	return time.Time{}, ErrInvalidTimestamp
+}

libbeat/common/jsontransform/jsonhelper_test.go

@@ -89,6 +89,41 @@ func TestWriteJSONKeys(t *testing.T) {
 				"top_c": "COMPLETELY_NEW_c",
 			},
 		},
+		"overwrite_true_ISO8601": {
+			overwriteKeys: true,
+			keys: map[string]interface{}{
+				"@metadata": map[string]interface{}{
+					"foo": "NEW_bar",
+					"baz": map[string]interface{}{
+						"qux":   "NEW_qux",
+						"durrr": "COMPLETELY_NEW",
+					},
+				},
+				"@timestamp": now.Format(iso8601),
+				"top_b": map[string]interface{}{
+					"inner_d": "NEW_dee",
+					"inner_e": "COMPLETELY_NEW_e",
+				},
+				"top_c": "COMPLETELY_NEW_c",
+			},
+			expectedMetadata: common.MapStr{
+				"foo": "NEW_bar",
+				"baz": common.MapStr{
+					"qux":   "NEW_qux",
+					"durrr": "COMPLETELY_NEW",
+				},
+			},
+			expectedTimestamp: now,
+			expectedFields: common.MapStr{
+				"top_a": 23,
+				"top_b": common.MapStr{
+					"inner_c": "see",
+					"inner_d": "NEW_dee",
+					"inner_e": "COMPLETELY_NEW_e",
+				},
+				"top_c": "COMPLETELY_NEW_c",
+			},
+		},
 		"overwrite_false": {
 			overwriteKeys: false,
 			keys: map[string]interface{}{