second review

Author: szybia

rest-api-spec/src/yamlRestTest/resources/rest-api-spec/test/ingest/20_tracking.yml

@@ -21,8 +21,8 @@ setup:
   - do:
       ingest.get_pipeline:
         id: "my_pipeline"
-  - match: { my_pipeline.created_date: "/\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}\\.\\d{3}Z/" }
-  - match: { my_pipeline.modified_date: "/\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}\\.\\d{3}Z/" }
+  - match: { my_pipeline.created_date: "/^\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}\\.\\d{3}Z$/" }
+  - match: { my_pipeline.modified_date: "/^\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}\\.\\d{3}Z$/" }
 
 ---
 "Test PUT setting created_date":

server/src/internalClusterTest/java/org/elasticsearch/ingest/PipelineConfigurationSyncIT.java

@@ -87,7 +87,7 @@ public TransformState transform(ProjectId projectId, List<PutPipelineRequest> so
         ProjectMetadata projectMetadata = clusterState.metadata().getProject(projectId);
 
         for (var request : requests) {
-            var nopUpdate = IngestService.isNoOpPipelineUpdate(projectMetadata, request, () -> IngestService.readPipelineConfig(request));
+            var nopUpdate = IngestService.isNoOpPipelineUpdate(projectMetadata, request);
 
             if (nopUpdate) {
                 continue;

server/src/main/java/org/elasticsearch/action/ingest/ReservedPipelineAction.java

@@ -42,8 +42,6 @@ static void executeDocument(
         if (verbose) {
             List<SimulateProcessorResult> processorResultList = new CopyOnWriteArrayList<>();
             CompoundProcessor verbosePipelineProcessor = decorate(pipeline.getCompoundProcessor(), null, processorResultList);
-            long createdDate = pipeline.getCreatedDateMillis().orElse(-1);
-            long modifiedDate = pipeline.getModifiedDateMillis().orElse(-1);
             Pipeline verbosePipeline = new Pipeline(
                 pipeline.getId(),
                 pipeline.getDescription(),
@@ -52,8 +50,8 @@ static void executeDocument(
                 verbosePipelineProcessor,
                 pipeline.getFieldAccessPattern(),
                 pipeline.getDeprecated(),
-                createdDate == -1 ? null : createdDate,
-                modifiedDate == -1 ? null : modifiedDate
+                pipeline.getCreatedDateMillis().orElse(null),
+                pipeline.getModifiedDateMillis().orElse(null)
             );
             ingestDocument.executePipeline(verbosePipeline, (result, e) -> {
                 handler.accept(new SimulateDocumentVerboseResult(processorResultList), e);

server/src/main/java/org/elasticsearch/action/ingest/SimulateExecutionService.java

@@ -49,7 +49,6 @@
 import org.elasticsearch.cluster.service.MasterServiceTaskQueue;
 import org.elasticsearch.common.Priority;
 import org.elasticsearch.common.TriConsumer;
-import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.collect.ImmutableOpenMap;
 import org.elasticsearch.common.logging.DeprecationCategory;
 import org.elasticsearch.common.logging.DeprecationLogger;
@@ -78,10 +77,11 @@
 import org.elasticsearch.script.ScriptService;
 import org.elasticsearch.threadpool.Scheduler;
 import org.elasticsearch.threadpool.ThreadPool;
-import org.elasticsearch.xcontent.XContentBuilder;
 
-import java.io.IOException;
 import java.time.Instant;
+import java.time.InstantSource;
+import java.time.format.DateTimeFormatter;
+import java.time.format.DateTimeFormatterBuilder;
 import java.time.temporal.ChronoUnit;
 import java.util.ArrayList;
 import java.util.Collection;
@@ -104,7 +104,6 @@
 import java.util.function.Function;
 import java.util.function.IntConsumer;
 import java.util.function.Predicate;
-import java.util.function.Supplier;
 import java.util.stream.Collectors;
 
 import static org.elasticsearch.core.Strings.format;
@@ -545,9 +544,7 @@ public void putPipeline(
         ActionListener<AcknowledgedResponse> listener,
         Consumer<ActionListener<NodesInfoResponse>> nodeInfoListener
     ) throws Exception {
-        Map<String, Object> newPipelineConfig = readPipelineConfig(request);
-        validateNoSystemPropertiesInPipelineConfig(newPipelineConfig);
-        if (isNoOpPipelineUpdate(state.metadata().getProject(projectId), request, () -> newPipelineConfig)) {
+        if (isNoOpPipelineUpdate(state.metadata().getProject(projectId), request)) {
             // existing pipeline matches request pipeline -- no need to update
             listener.onResponse(AcknowledgedResponse.TRUE);
             return;
@@ -558,7 +555,7 @@ public void putPipeline(
 
             taskQueue.submitTask(
                 "put-pipeline-" + request.getId(),
-                new PutPipelineClusterStateUpdateTask(projectId, l, request),
+                new PutPipelineClusterStateUpdateTask(projectId, l, request, Instant::now),
                 request.masterNodeTimeout()
             );
         }));
@@ -574,10 +571,6 @@ public void validatePipelineRequest(ProjectId projectId, PutPipelineRequest requ
         validatePipeline(ingestInfos, projectId, request.getId(), config);
     }
 
-    public static Map<String, Object> readPipelineConfig(PutPipelineRequest request) {
-        return XContentHelper.convertToMap(request.getSource(), false, request.getXContentType()).v2();
-    }
-
     public static void validateNoSystemPropertiesInPipelineConfig(final Map<String, Object> pipelineConfig) {
         if (pipelineConfig.containsKey(Pipeline.CREATED_DATE_KEY)) {
             throw new ElasticsearchParseException("Provided a pipeline property which is managed by the system: created_date.");
@@ -586,24 +579,23 @@ public static void validateNoSystemPropertiesInPipelineConfig(final Map<String,
         }
     }
 
-    public static boolean isNoOpPipelineUpdate(
-        ProjectMetadata metadata,
-        PutPipelineRequest request,
-        Supplier<Map<String, Object>> newPipelineConfigSupplier
-    ) {
+    /** Check whether updating a potentially existing pipeline will be a NOP.
+     * Will return <code>false</code> if request contains system-properties like `{created,modified}_date,
+     * these should be rejected later.`*/
+    public static boolean isNoOpPipelineUpdate(ProjectMetadata metadata, PutPipelineRequest request) {
         IngestMetadata currentIngestMetadata = metadata.custom(IngestMetadata.TYPE);
         if (request.getVersion() == null
             && currentIngestMetadata != null
             && currentIngestMetadata.getPipelines().containsKey(request.getId())) {
 
+            var newPipelineConfig = XContentHelper.convertToMap(request.getSource(), false, request.getXContentType()).v2();
+
             Map<String, Object> currentConfigWithoutSystemProps = new HashMap<>(
                 currentIngestMetadata.getPipelines().get(request.getId()).getConfig()
             );
             currentConfigWithoutSystemProps.remove(Pipeline.CREATED_DATE_KEY);
             currentConfigWithoutSystemProps.remove(Pipeline.MODIFIED_DATE_KEY);
 
-            Map<String, Object> newPipelineConfig = newPipelineConfigSupplier.get();
-
             return newPipelineConfig.equals(currentConfigWithoutSystemProps);
         }
 
@@ -701,26 +693,42 @@ private static void collectProcessorMetrics(
      * Used in this class and externally by the {@link org.elasticsearch.action.ingest.ReservedPipelineAction}
      */
     public static class PutPipelineClusterStateUpdateTask extends PipelineClusterStateUpdateTask {
+        // always output millis even if instantSource returns millis == 0
+        private static final DateTimeFormatter ISO8601_WITH_MILLIS_FORMATTER = new DateTimeFormatterBuilder().appendInstant(3)
+            .toFormatter(Locale.ROOT);
+
         private final PutPipelineRequest request;
+        private final InstantSource instantSource;
 
-        PutPipelineClusterStateUpdateTask(ProjectId projectId, ActionListener<AcknowledgedResponse> listener, PutPipelineRequest request) {
+        PutPipelineClusterStateUpdateTask(
+            final ProjectId projectId,
+            final ActionListener<AcknowledgedResponse> listener,
+            final PutPipelineRequest request,
+            final InstantSource instantSource
+        ) {
             super(projectId, listener);
             this.request = request;
+            this.instantSource = instantSource;
         }
 
         /**
          * Used by {@link org.elasticsearch.action.ingest.ReservedPipelineAction}
          */
         public PutPipelineClusterStateUpdateTask(ProjectId projectId, PutPipelineRequest request) {
-            this(projectId, null, request);
+            this(projectId, null, request, Instant::now);
         }
 
         @Override
         public IngestMetadata execute(IngestMetadata currentIngestMetadata, Collection<IndexMetadata> allIndexMetadata) {
-            BytesReference pipelineSource = request.getSource();
+            final Map<String, PipelineConfiguration> existingPipelines = currentIngestMetadata == null
+                ? new HashMap<>(1)
+                : new HashMap<>(currentIngestMetadata.getPipelines());
+            final PipelineConfiguration existingPipeline = existingPipelines.get(request.getId());
+            final Map<String, Object> newPipelineConfig = XContentHelper.convertToMap(request.getSource(), true, request.getXContentType())
+                .v2();
+
             if (request.getVersion() != null) {
-                var currentPipeline = currentIngestMetadata != null ? currentIngestMetadata.getPipelines().get(request.getId()) : null;
-                if (currentPipeline == null) {
+                if (existingPipeline == null) {
                     throw new IllegalArgumentException(
                         String.format(
                             Locale.ROOT,
@@ -731,7 +739,7 @@ public IngestMetadata execute(IngestMetadata currentIngestMetadata, Collection<I
                     );
                 }
 
-                final Integer currentVersion = currentPipeline.getVersion();
+                final Integer currentVersion = existingPipeline.getVersion();
                 if (Objects.equals(request.getVersion(), currentVersion) == false) {
                     throw new IllegalArgumentException(
                         String.format(
@@ -744,9 +752,8 @@ public IngestMetadata execute(IngestMetadata currentIngestMetadata, Collection<I
                     );
                 }
 
-                var pipelineConfig = XContentHelper.convertToMap(request.getSource(), false, request.getXContentType()).v2();
-                final Integer specifiedVersion = (Integer) pipelineConfig.get("version");
-                if (pipelineConfig.containsKey("version") && Objects.equals(specifiedVersion, currentVersion)) {
+                final Integer specifiedVersion = (Integer) newPipelineConfig.get("version");
+                if (newPipelineConfig.containsKey("version") && Objects.equals(specifiedVersion, currentVersion)) {
                     throw new IllegalArgumentException(
                         String.format(
                             Locale.ROOT,
@@ -759,28 +766,15 @@ public IngestMetadata execute(IngestMetadata currentIngestMetadata, Collection<I
 
                 // if no version specified in the pipeline definition, inject a version of [request.getVersion() + 1]
                 if (specifiedVersion == null) {
-                    pipelineConfig.put("version", request.getVersion() == null ? 1 : request.getVersion() + 1);
-                    try {
-                        var builder = XContentBuilder.builder(request.getXContentType().xContent()).map(pipelineConfig);
-                        pipelineSource = BytesReference.bytes(builder);
-                    } catch (IOException e) {
-                        throw new IllegalStateException(e);
-                    }
+                    newPipelineConfig.put("version", request.getVersion() == null ? 1 : request.getVersion() + 1);
                 }
             }
 
-            Map<String, PipelineConfiguration> pipelines;
-            if (currentIngestMetadata != null) {
-                pipelines = new HashMap<>(currentIngestMetadata.getPipelines());
-            } else {
-                pipelines = new HashMap<>();
-            }
-
-            Instant now = Instant.now().truncatedTo(ChronoUnit.MILLIS);
-            Map<String, Object> newPipelineConfig = XContentHelper.convertToMap(pipelineSource, true, request.getXContentType()).v2();
-            PipelineConfiguration existingPipeline = pipelines.get(request.getId());
+            final String iso8601WithMillisNow = ISO8601_WITH_MILLIS_FORMATTER.format(
+                instantSource.instant().truncatedTo(ChronoUnit.MILLIS)
+            );
             if (existingPipeline == null) {
-                newPipelineConfig.put(Pipeline.CREATED_DATE_KEY, now.toString());
+                newPipelineConfig.put(Pipeline.CREATED_DATE_KEY, iso8601WithMillisNow);
             } else {
                 Object existingCreatedAt = existingPipeline.getConfig().get(Pipeline.CREATED_DATE_KEY);
                 // only set/carry over `created_date` if existing pipeline already has it.
@@ -789,10 +783,10 @@ public IngestMetadata execute(IngestMetadata currentIngestMetadata, Collection<I
                     newPipelineConfig.put(Pipeline.CREATED_DATE_KEY, existingCreatedAt);
                 }
             }
-            newPipelineConfig.put(Pipeline.MODIFIED_DATE_KEY, now.toString());
+            newPipelineConfig.put(Pipeline.MODIFIED_DATE_KEY, iso8601WithMillisNow);
 
-            pipelines.put(request.getId(), new PipelineConfiguration(request.getId(), newPipelineConfig));
-            return new IngestMetadata(pipelines);
+            existingPipelines.put(request.getId(), new PipelineConfiguration(request.getId(), newPipelineConfig));
+            return new IngestMetadata(existingPipelines);
         }
     }
 
@@ -803,6 +797,7 @@ void validatePipeline(
         String pipelineId,
         Map<String, Object> pipelineConfig
     ) throws Exception {
+        validateNoSystemPropertiesInPipelineConfig(pipelineConfig);
         if (ingestInfos.isEmpty()) {
             throw new IllegalStateException("Ingest info is empty");
         }

server/src/main/java/org/elasticsearch/ingest/IngestService.java

@@ -19,7 +19,7 @@
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
-import java.util.OptionalLong;
+import java.util.Optional;
 import java.util.function.BiConsumer;
 import java.util.function.LongSupplier;
 import java.util.function.Predicate;
@@ -305,12 +305,12 @@ public boolean isDeprecated() {
         return Boolean.TRUE.equals(deprecated);
     }
 
-    public OptionalLong getCreatedDateMillis() {
-        return createdDateMillis == null ? OptionalLong.empty() : OptionalLong.of(createdDateMillis);
+    public Optional<Long> getCreatedDateMillis() {
+        return Optional.ofNullable(createdDateMillis);
     }
 
-    public OptionalLong getModifiedDateMillis() {
-        return modifiedDateMillis == null ? OptionalLong.empty() : OptionalLong.of(modifiedDateMillis);
+    public Optional<Long> getModifiedDateMillis() {
+        return Optional.ofNullable(modifiedDateMillis);
     }
 
     @Override
@@ -325,10 +325,9 @@ public String toString() {
         sb.append(", relativeTimeProvider=").append(relativeTimeProvider);
         sb.append(", fieldAccessPattern=").append(fieldAccessPattern);
         sb.append(", deprecated=").append(deprecated);
-        sb.append(", createdDateMillis=").append(createdDateMillis);
-        sb.append(", modifiedDateMillis=").append(modifiedDateMillis);
+        sb.append(", createdDateMillis='").append(createdDateMillis).append('\'');
+        sb.append(", modifiedDateMillis='").append(modifiedDateMillis).append('\'');
         sb.append('}');
         return sb.toString();
     }
-
 }

server/src/main/java/org/elasticsearch/ingest/Pipeline.java

@@ -210,8 +210,6 @@ private void decorateAndExecutePipeline(
                 conditionalWithResult
             )
         );
-        long createdDate = pipeline.getCreatedDateMillis().orElse(-1);
-        long modifiedDate = pipeline.getModifiedDateMillis().orElse(-1);
         Pipeline verbosePipeline = new Pipeline(
             pipeline.getId(),
             pipeline.getDescription(),
@@ -220,8 +218,8 @@ private void decorateAndExecutePipeline(
             verbosePipelineProcessor,
             pipeline.getFieldAccessPattern(),
             pipeline.getDeprecated(),
-            createdDate == -1 ? null : createdDate,
-            modifiedDate == -1 ? null : modifiedDate
+            pipeline.getCreatedDateMillis().orElse(null),
+            pipeline.getModifiedDateMillis().orElse(null)
         );
         ingestDocument.executePipeline(verbosePipeline, handler);
     }