[Connectors API] Unify timestamp field parsing (#104416)

x-pack/plugin/ent-search/qa/rest/src/yamlRestTest/resources/rest-api-spec/test/entsearch/334_connector_update_last_sync_stats.yml

@@ -30,6 +30,24 @@ setup:
   - match: { last_sync_error: "oh no error" }
   - match: { last_access_control_sync_scheduled_at: "2023-05-25T12:30:00.000Z" }
 
+---
+"Update Connector Last Sync Stats - Supports different datetime format":
+  - do:
+      connector.last_sync:
+        connector_id: test-connector
+        body:
+          last_sync_error: "oh no error"
+          last_access_control_sync_scheduled_at: "2023-05-25T12:30:00.000Z"
+
+  - match: { result: updated }
+
+  - do:
+      connector.get:
+        connector_id: test-connector
+
+  - match: { last_sync_error: "oh no error" }
+  - match: { last_access_control_sync_scheduled_at: "2023-05-25T12:30:00.000Z" }
+
 ---
 "Update Connector Last Sync Stats - Connector doesn't exist":
   - do:

x-pack/plugin/ent-search/src/main/java/org/elasticsearch/xpack/application/connector/Connector.java

@@ -273,15 +273,18 @@ public Connector(StreamInput in) throws IOException {
         PARSER.declareStringOrNull(optionalConstructorArg(), LANGUAGE_FIELD);
         PARSER.declareField(
             optionalConstructorArg(),
-            (p, c) -> p.currentToken() == XContentParser.Token.VALUE_NULL ? null : Instant.parse(p.text()),
+            (p, c) -> ConnectorUtils.parseNullableInstant(p, Connector.LAST_SEEN_FIELD.getPreferredName()),
             Connector.LAST_SEEN_FIELD,
             ObjectParser.ValueType.STRING_OR_NULL
         );
 
         PARSER.declareStringOrNull(optionalConstructorArg(), ConnectorSyncInfo.LAST_ACCESS_CONTROL_SYNC_ERROR);
         PARSER.declareField(
             optionalConstructorArg(),
-            (p, c) -> p.currentToken() == XContentParser.Token.VALUE_NULL ? null : Instant.parse(p.text()),
+            (p, c) -> ConnectorUtils.parseNullableInstant(
+                p,
+                ConnectorSyncInfo.LAST_ACCESS_CONTROL_SYNC_SCHEDULED_AT_FIELD.getPreferredName()
+            ),
             ConnectorSyncInfo.LAST_ACCESS_CONTROL_SYNC_SCHEDULED_AT_FIELD,
             ObjectParser.ValueType.STRING_OR_NULL
         );
@@ -294,15 +297,15 @@ public Connector(StreamInput in) throws IOException {
         PARSER.declareLong(optionalConstructorArg(), ConnectorSyncInfo.LAST_DELETED_DOCUMENT_COUNT_FIELD);
         PARSER.declareField(
             optionalConstructorArg(),
-            (p, c) -> p.currentToken() == XContentParser.Token.VALUE_NULL ? null : Instant.parse(p.text()),
+            (p, c) -> ConnectorUtils.parseNullableInstant(p, ConnectorSyncInfo.LAST_INCREMENTAL_SYNC_SCHEDULED_AT_FIELD.getPreferredName()),
             ConnectorSyncInfo.LAST_INCREMENTAL_SYNC_SCHEDULED_AT_FIELD,
             ObjectParser.ValueType.STRING_OR_NULL
         );
         PARSER.declareLong(optionalConstructorArg(), ConnectorSyncInfo.LAST_INDEXED_DOCUMENT_COUNT_FIELD);
         PARSER.declareStringOrNull(optionalConstructorArg(), ConnectorSyncInfo.LAST_SYNC_ERROR_FIELD);
         PARSER.declareField(
             optionalConstructorArg(),
-            (p, c) -> p.currentToken() == XContentParser.Token.VALUE_NULL ? null : Instant.parse(p.text()),
+            (p, c) -> ConnectorUtils.parseNullableInstant(p, ConnectorSyncInfo.LAST_SYNC_SCHEDULED_AT_FIELD.getPreferredName()),
             ConnectorSyncInfo.LAST_SYNC_SCHEDULED_AT_FIELD,
             ObjectParser.ValueType.STRING_OR_NULL
         );
@@ -314,7 +317,7 @@ public Connector(StreamInput in) throws IOException {
         );
         PARSER.declareField(
             optionalConstructorArg(),
-            (p, c) -> p.currentToken() == XContentParser.Token.VALUE_NULL ? null : Instant.parse(p.text()),
+            (p, c) -> ConnectorUtils.parseNullableInstant(p, ConnectorSyncInfo.LAST_SYNCED_FIELD.getPreferredName()),
             ConnectorSyncInfo.LAST_SYNCED_FIELD,
             ObjectParser.ValueType.STRING_OR_NULL
         );

x-pack/plugin/ent-search/src/main/java/org/elasticsearch/xpack/application/connector/ConnectorCustomSchedule.java

@@ -101,8 +101,8 @@ public ConnectorCustomSchedule(StreamInput in) throws IOException {
         PARSER.declareString(constructorArg(), INTERVAL_FIELD);
         PARSER.declareField(
             optionalConstructorArg(),
-            (p, c) -> p.currentToken() == XContentParser.Token.VALUE_NULL ? null : Instant.parse(p.text()),
-            ConnectorSyncInfo.LAST_SYNCED_FIELD,
+            (p, c) -> ConnectorUtils.parseNullableInstant(p, LAST_SYNCED_FIELD.getPreferredName()),
+            LAST_SYNCED_FIELD,
             ObjectParser.ValueType.STRING_OR_NULL
         );
         PARSER.declareString(constructorArg(), NAME_FIELD);

x-pack/plugin/ent-search/src/main/java/org/elasticsearch/xpack/application/connector/ConnectorUtils.java

@@ -0,0 +1,40 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.application.connector;
+
+import org.elasticsearch.xcontent.XContentParser;
+import org.elasticsearch.xpack.core.common.time.TimeUtils;
+
+import java.io.IOException;
+import java.time.Instant;
+
+public class ConnectorUtils {
+
+    /**
+     * Parses a field from the XContentParser to an Instant. This method should be used for parsing
+     * all datetime fields related to Connector APIs. It utilizes the parseTimeFieldToInstant method from {@link TimeUtils}
+     * to parse the date-time string to an Instant.
+     *
+     * @param p         the XContentParser instance from which to parse the date-time string.
+     * @param fieldName the name of the field whose value is to be parsed.
+     */
+    public static Instant parseInstant(XContentParser p, String fieldName) throws IOException {
+        return TimeUtils.parseTimeFieldToInstant(p, fieldName);
+    }
+
+    /**
+     * Parses a nullable field from the XContentParser to an Instant. This method is useful
+     * when parsing datetime fields that might have null values.
+     *
+     * @param p         the XContentParser instance from which to parse the date-time string.
+     * @param fieldName the name of the field whose value is to be parsed.
+     */
+    public static Instant parseNullableInstant(XContentParser p, String fieldName) throws IOException {
+        return p.currentToken() == XContentParser.Token.VALUE_NULL ? null : parseInstant(p, fieldName);
+    }
+}

x-pack/plugin/ent-search/src/main/java/org/elasticsearch/xpack/application/connector/action/UpdateConnectorLastSyncStatsAction.java

@@ -25,6 +25,7 @@
 import org.elasticsearch.xcontent.XContentType;
 import org.elasticsearch.xpack.application.connector.ConnectorSyncInfo;
 import org.elasticsearch.xpack.application.connector.ConnectorSyncStatus;
+import org.elasticsearch.xpack.application.connector.ConnectorUtils;
 
 import java.io.IOException;
 import java.time.Instant;
@@ -101,7 +102,10 @@ public ActionRequestValidationException validate() {
             PARSER.declareStringOrNull(optionalConstructorArg(), ConnectorSyncInfo.LAST_ACCESS_CONTROL_SYNC_ERROR);
             PARSER.declareField(
                 optionalConstructorArg(),
-                (p, c) -> p.currentToken() == XContentParser.Token.VALUE_NULL ? null : Instant.parse(p.text()),
+                (p, c) -> ConnectorUtils.parseNullableInstant(
+                    p,
+                    ConnectorSyncInfo.LAST_ACCESS_CONTROL_SYNC_SCHEDULED_AT_FIELD.getPreferredName()
+                ),
                 ConnectorSyncInfo.LAST_ACCESS_CONTROL_SYNC_SCHEDULED_AT_FIELD,
                 ObjectParser.ValueType.STRING_OR_NULL
             );
@@ -114,15 +118,18 @@ public ActionRequestValidationException validate() {
             PARSER.declareLong(optionalConstructorArg(), ConnectorSyncInfo.LAST_DELETED_DOCUMENT_COUNT_FIELD);
             PARSER.declareField(
                 optionalConstructorArg(),
-                (p, c) -> p.currentToken() == XContentParser.Token.VALUE_NULL ? null : Instant.parse(p.text()),
+                (p, c) -> ConnectorUtils.parseNullableInstant(
+                    p,
+                    ConnectorSyncInfo.LAST_INCREMENTAL_SYNC_SCHEDULED_AT_FIELD.getPreferredName()
+                ),
                 ConnectorSyncInfo.LAST_INCREMENTAL_SYNC_SCHEDULED_AT_FIELD,
                 ObjectParser.ValueType.STRING_OR_NULL
             );
             PARSER.declareLong(optionalConstructorArg(), ConnectorSyncInfo.LAST_INDEXED_DOCUMENT_COUNT_FIELD);
             PARSER.declareStringOrNull(optionalConstructorArg(), ConnectorSyncInfo.LAST_SYNC_ERROR_FIELD);
             PARSER.declareField(
                 optionalConstructorArg(),
-                (p, c) -> p.currentToken() == XContentParser.Token.VALUE_NULL ? null : Instant.parse(p.text()),
+                (p, c) -> ConnectorUtils.parseNullableInstant(p, ConnectorSyncInfo.LAST_SYNC_SCHEDULED_AT_FIELD.getPreferredName()),
                 ConnectorSyncInfo.LAST_SYNC_SCHEDULED_AT_FIELD,
                 ObjectParser.ValueType.STRING_OR_NULL
             );
@@ -134,7 +141,7 @@ public ActionRequestValidationException validate() {
             );
             PARSER.declareField(
                 optionalConstructorArg(),
-                (p, c) -> p.currentToken() == XContentParser.Token.VALUE_NULL ? null : Instant.parse(p.text()),
+                (p, c) -> ConnectorUtils.parseNullableInstant(p, ConnectorSyncInfo.LAST_SYNCED_FIELD.getPreferredName()),
                 ConnectorSyncInfo.LAST_SYNCED_FIELD,
                 ObjectParser.ValueType.STRING_OR_NULL
             );

x-pack/plugin/ent-search/src/main/java/org/elasticsearch/xpack/application/connector/filtering/FilteringAdvancedSnippet.java

@@ -16,6 +16,7 @@
 import org.elasticsearch.xcontent.ToXContentObject;
 import org.elasticsearch.xcontent.XContentBuilder;
 import org.elasticsearch.xcontent.XContentParser;
+import org.elasticsearch.xpack.application.connector.ConnectorUtils;
 
 import java.io.IOException;
 import java.time.Instant;
@@ -71,8 +72,18 @@ public FilteringAdvancedSnippet(StreamInput in) throws IOException {
     );
 
     static {
-        PARSER.declareField(constructorArg(), (p, c) -> Instant.parse(p.text()), CREATED_AT_FIELD, ObjectParser.ValueType.STRING);
-        PARSER.declareField(constructorArg(), (p, c) -> Instant.parse(p.text()), UPDATED_AT_FIELD, ObjectParser.ValueType.STRING);
+        PARSER.declareField(
+            constructorArg(),
+            (p, c) -> ConnectorUtils.parseInstant(p, CREATED_AT_FIELD.getPreferredName()),
+            CREATED_AT_FIELD,
+            ObjectParser.ValueType.STRING
+        );
+        PARSER.declareField(
+            constructorArg(),
+            (p, c) -> ConnectorUtils.parseInstant(p, UPDATED_AT_FIELD.getPreferredName()),
+            UPDATED_AT_FIELD,
+            ObjectParser.ValueType.STRING
+        );
         PARSER.declareField(constructorArg(), (p, c) -> p.map(), VALUE_FIELD, ObjectParser.ValueType.OBJECT);
     }
 

x-pack/plugin/ent-search/src/main/java/org/elasticsearch/xpack/application/connector/filtering/FilteringRule.java

@@ -16,6 +16,7 @@
 import org.elasticsearch.xcontent.ToXContentObject;
 import org.elasticsearch.xcontent.XContentBuilder;
 import org.elasticsearch.xcontent.XContentParser;
+import org.elasticsearch.xpack.application.connector.ConnectorUtils;
 
 import java.io.IOException;
 import java.time.Instant;
@@ -108,7 +109,12 @@ public FilteringRule(StreamInput in) throws IOException {
     );
 
     static {
-        PARSER.declareField(constructorArg(), (p, c) -> Instant.parse(p.text()), CREATED_AT_FIELD, ObjectParser.ValueType.STRING);
+        PARSER.declareField(
+            constructorArg(),
+            (p, c) -> ConnectorUtils.parseInstant(p, CREATED_AT_FIELD.getPreferredName()),
+            CREATED_AT_FIELD,
+            ObjectParser.ValueType.STRING
+        );
         PARSER.declareString(constructorArg(), FIELD_FIELD);
         PARSER.declareString(constructorArg(), ID_FIELD);
         PARSER.declareInt(constructorArg(), ORDER_FIELD);
@@ -124,7 +130,12 @@ public FilteringRule(StreamInput in) throws IOException {
             RULE_FIELD,
             ObjectParser.ValueType.STRING
         );
-        PARSER.declareField(constructorArg(), (p, c) -> Instant.parse(p.text()), UPDATED_AT_FIELD, ObjectParser.ValueType.STRING);
+        PARSER.declareField(
+            constructorArg(),
+            (p, c) -> ConnectorUtils.parseInstant(p, UPDATED_AT_FIELD.getPreferredName()),
+            UPDATED_AT_FIELD,
+            ObjectParser.ValueType.STRING
+        );
         PARSER.declareString(constructorArg(), VALUE_FIELD);
     }
 

x-pack/plugin/ent-search/src/main/java/org/elasticsearch/xpack/application/connector/syncjob/ConnectorSyncJob.java

@@ -28,6 +28,7 @@
 import org.elasticsearch.xpack.application.connector.ConnectorFiltering;
 import org.elasticsearch.xpack.application.connector.ConnectorIngestPipeline;
 import org.elasticsearch.xpack.application.connector.ConnectorSyncStatus;
+import org.elasticsearch.xpack.application.connector.ConnectorUtils;
 
 import java.io.IOException;
 import java.time.Instant;
@@ -265,19 +266,19 @@ public ConnectorSyncJob(StreamInput in) throws IOException {
     static {
         PARSER.declareField(
             optionalConstructorArg(),
-            (p, c) -> parseNullableInstant(p),
+            (p, c) -> ConnectorUtils.parseNullableInstant(p, CANCELATION_REQUESTED_AT_FIELD.getPreferredName()),
             CANCELATION_REQUESTED_AT_FIELD,
             ObjectParser.ValueType.STRING_OR_NULL
         );
         PARSER.declareField(
             optionalConstructorArg(),
-            (p, c) -> parseNullableInstant(p),
+            (p, c) -> ConnectorUtils.parseNullableInstant(p, CANCELED_AT_FIELD.getPreferredName()),
             CANCELED_AT_FIELD,
             ObjectParser.ValueType.STRING_OR_NULL
         );
         PARSER.declareField(
             optionalConstructorArg(),
-            (p, c) -> parseNullableInstant(p),
+            (p, c) -> ConnectorUtils.parseNullableInstant(p, COMPLETED_AT_FIELD.getPreferredName()),
             COMPLETED_AT_FIELD,
             ObjectParser.ValueType.STRING_OR_NULL
         );
@@ -287,7 +288,12 @@ public ConnectorSyncJob(StreamInput in) throws IOException {
             CONNECTOR_FIELD,
             ObjectParser.ValueType.OBJECT
         );
-        PARSER.declareField(constructorArg(), (p, c) -> Instant.parse(p.text()), CREATED_AT_FIELD, ObjectParser.ValueType.STRING);
+        PARSER.declareField(
+            constructorArg(),
+            (p, c) -> ConnectorUtils.parseInstant(p, CREATED_AT_FIELD.getPreferredName()),
+            CREATED_AT_FIELD,
+            ObjectParser.ValueType.STRING
+        );
         PARSER.declareLong(constructorArg(), DELETED_DOCUMENT_COUNT_FIELD);
         PARSER.declareStringOrNull(optionalConstructorArg(), ERROR_FIELD);
         PARSER.declareString(constructorArg(), ID_FIELD);
@@ -299,11 +305,16 @@ public ConnectorSyncJob(StreamInput in) throws IOException {
             JOB_TYPE_FIELD,
             ObjectParser.ValueType.STRING
         );
-        PARSER.declareField(constructorArg(), (p, c) -> parseNullableInstant(p), LAST_SEEN_FIELD, ObjectParser.ValueType.STRING_OR_NULL);
+        PARSER.declareField(
+            constructorArg(),
+            (p, c) -> ConnectorUtils.parseNullableInstant(p, LAST_SEEN_FIELD.getPreferredName()),
+            LAST_SEEN_FIELD,
+            ObjectParser.ValueType.STRING_OR_NULL
+        );
         PARSER.declareField(constructorArg(), (p, c) -> p.map(), METADATA_FIELD, ObjectParser.ValueType.OBJECT);
         PARSER.declareField(
             optionalConstructorArg(),
-            (p, c) -> parseNullableInstant(p),
+            (p, c) -> ConnectorUtils.parseNullableInstant(p, STARTED_AT_FIELD.getPreferredName()),
             STARTED_AT_FIELD,
             ObjectParser.ValueType.STRING_OR_NULL
         );
@@ -323,10 +334,6 @@ public ConnectorSyncJob(StreamInput in) throws IOException {
         PARSER.declareStringOrNull(optionalConstructorArg(), WORKER_HOSTNAME_FIELD);
     }
 
-    private static Instant parseNullableInstant(XContentParser p) throws IOException {
-        return p.currentToken() == XContentParser.Token.VALUE_NULL ? null : Instant.parse(p.text());
-    }
-
     @SuppressWarnings("unchecked")
     private static final ConstructingObjectParser<Connector, String> SYNC_JOB_CONNECTOR_PARSER = new ConstructingObjectParser<>(
         "sync_job_connector",

x-pack/plugin/ent-search/src/main/java/org/elasticsearch/xpack/application/connector/syncjob/action/UpdateConnectorSyncJobIngestionStatsAction.java

@@ -24,6 +24,8 @@
 import org.elasticsearch.xcontent.XContentParser;
 import org.elasticsearch.xcontent.XContentParserConfiguration;
 import org.elasticsearch.xcontent.XContentType;
+import org.elasticsearch.xpack.application.connector.Connector;
+import org.elasticsearch.xpack.application.connector.ConnectorUtils;
 import org.elasticsearch.xpack.application.connector.action.ConnectorUpdateActionResponse;
 import org.elasticsearch.xpack.application.connector.syncjob.ConnectorSyncJob;
 
@@ -166,7 +168,7 @@ public ActionRequestValidationException validate() {
             PARSER.declareLong(optionalConstructorArg(), ConnectorSyncJob.TOTAL_DOCUMENT_COUNT_FIELD);
             PARSER.declareField(
                 optionalConstructorArg(),
-                (p, c) -> Instant.parse(p.text()),
+                (p, c) -> ConnectorUtils.parseInstant(p, Connector.LAST_SEEN_FIELD.getPreferredName()),
                 ConnectorSyncJob.LAST_SEEN_FIELD,
                 ObjectParser.ValueType.OBJECT_OR_STRING
             );

x-pack/plugin/ent-search/src/test/java/org/elasticsearch/xpack/application/connector/ConnectorTestUtils.java

@@ -87,15 +87,15 @@ public static ConnectorIngestPipeline getRandomConnectorIngestPipeline() {
 
     public static ConnectorSyncInfo getRandomConnectorSyncInfo() {
         return new ConnectorSyncInfo.Builder().setLastAccessControlSyncError(randomFrom(new String[] { null, randomAlphaOfLength(10) }))
-            .setLastAccessControlSyncScheduledAt(randomFrom(new Instant[] { null, Instant.ofEpochMilli(randomLong()) }))
+            .setLastAccessControlSyncScheduledAt(randomFrom(new Instant[] { null, ConnectorTestUtils.randomInstant() }))
             .setLastAccessControlSyncStatus(randomFrom(new ConnectorSyncStatus[] { null, getRandomSyncStatus() }))
             .setLastDeletedDocumentCount(randomLong())
-            .setLastIncrementalSyncScheduledAt(randomFrom(new Instant[] { null, Instant.ofEpochMilli(randomLong()) }))
+            .setLastIncrementalSyncScheduledAt(randomFrom(new Instant[] { null, ConnectorTestUtils.randomInstant() }))
             .setLastIndexedDocumentCount(randomLong())
             .setLastSyncError(randomFrom(new String[] { null, randomAlphaOfLength(10) }))
-            .setLastSyncScheduledAt(randomFrom(new Instant[] { null, Instant.ofEpochMilli(randomLong()) }))
+            .setLastSyncScheduledAt(randomFrom(new Instant[] { null, ConnectorTestUtils.randomInstant() }))
             .setLastSyncStatus(randomFrom(new ConnectorSyncStatus[] { null, getRandomSyncStatus() }))
-            .setLastSynced(randomFrom(new Instant[] { null, Instant.ofEpochMilli(randomLong()) }))
+            .setLastSynced(randomFrom(new Instant[] { null, ConnectorTestUtils.randomInstant() }))
             .build();
     }
 
@@ -249,7 +249,7 @@ public static Connector getRandomConnector() {
             .setIndexName(randomAlphaOfLength(10))
             .setIsNative(randomBoolean())
             .setLanguage(randomFrom(new String[] { null, randomAlphaOfLength(10) }))
-            .setLastSeen(randomFrom(new Instant[] { null, Instant.ofEpochMilli(randomLong()) }))
+            .setLastSeen(randomFrom(new Instant[] { null, ConnectorTestUtils.randomInstant() }))
             .setSyncInfo(getRandomConnectorSyncInfo())
             .setName(randomFrom(new String[] { null, randomAlphaOfLength(10) }))
             .setPipeline(randomBoolean() ? getRandomConnectorIngestPipeline() : null)
@@ -287,6 +287,21 @@ private static Cron getRandomCronExpression() {
         );
     }
 
+    /**
+     * Generate a random Instant between:
+     * - 1 January 1970 00:00:00+00:00
+     * - 24 January 2065 05:20:00+00:00
+     */
+    public static Instant randomInstant() {
+        Instant lowerBoundInstant = Instant.ofEpochSecond(0L);
+        Instant upperBoundInstant = Instant.ofEpochSecond(3000000000L);
+
+        return Instant.ofEpochSecond(
+            randomLongBetween(lowerBoundInstant.getEpochSecond(), upperBoundInstant.getEpochSecond()),
+            randomLongBetween(0, 999999999)
+        );
+    }
+
     public static ConnectorSyncStatus getRandomSyncStatus() {
         ConnectorSyncStatus[] values = ConnectorSyncStatus.values();
         return values[randomInt(values.length - 1)];