Pipelines: Add created_date and modified_date

Author: szybia

docs/changelog/130847.yaml

@@ -0,0 +1,12 @@
+pr: 130847
+summary: "Pipelines: Add `created_date` and `modified_date`"
+area: Ingest Node
+type: enhancement
+issues: [108754]
+highlight:
+ title: "Pipelines: Add `created_date` and `modified_date`"
+ body: |-
+ Pipelines now have extra properties to indicate when they were created, and updated, to improve visibility
+ and debugging.
+ These properties are managed by the system and cannot be updated by the user.
+ notable: true

qa/mixed-cluster/build.gradle

@@ -71,6 +71,12 @@ excludeList.add('aggregations/percentiles_hdr_metric/Negative values test')
 // sync_id is removed in 9.0
 excludeList.add("cat.shards/10_basic/Help")
 
+// new optional properties only available on latest cluster
+excludeList.add("ingest/10_basic/Test creating and getting pipeline returns created_date and modified_date")
+excludeList.add("ingest/10_basic/Test PUT setting created_date")
+excludeList.add("ingest/10_basic/Test PUT setting modified_date")
+excludeList.add("simulate.ingest/10_basic/Test simulate with pipeline with created_date")
+
 def clusterPath = getPath()
 
 buildParams.bwcVersions.withWireCompatible { bwcVersion, baseName ->

rest-api-spec/src/yamlRestTest/resources/rest-api-spec/test/ingest/10_basic.yml

@@ -184,3 +184,49 @@
  - is_false: first_pipeline.description
  - is_true: second_pipeline
  - is_false: second_pipeline.description
+
+
+---
+"Test creating and getting pipeline returns created_date and modified_date":
+ - do:
+ ingest.put_pipeline:
+ id: "my_pipeline"
+ body: >
+ {
+ "processors": []
+ }
+ - match: { acknowledged: true }
+
+ - 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/" }
+
+---
+"Test PUT setting created_date":
+ - do:
+ catch: bad_request
+ ingest.put_pipeline:
+ id: "my_pipeline"
+ body: >
+ {
+ "processors": [],
+ "created_date": "2025-07-04T12:50:48.415Z"
+ }
+ - match: { status: 400 }
+ - match: { error.reason: "Provided a pipeline property which is managed by the system, e.g. `created_date`." }
+
+---
+"Test PUT setting modified_date":
+ - do:
+ catch: bad_request
+ ingest.put_pipeline:
+ id: "my_pipeline"
+ body: >
+ {
+ "processors": [],
+ "modified_date": "2025-07-04T12:50:48.415Z"
+ }
+ - match: { status: 400 }
+ - match: { error.reason: "Provided a pipeline property which is managed by the system, e.g. `created_date`." }

rest-api-spec/src/yamlRestTest/resources/rest-api-spec/test/simulate.ingest/10_basic.yml

@@ -748,3 +748,33 @@ setup:
  - match: { docs.1.doc._index: "index-2" }
  - match: { docs.1.doc._source.foo: "rab" }
  - match: { docs.1.doc.executed_pipelines: ["my-pipeline"] }
+
+---
+"Test simulate with pipeline with created_date":
+ - skip:
+ features: headers
+ - do:
+ catch: request
+ headers:
+ Content-Type: application/json
+ simulate.ingest:
+ pipeline: "my_pipeline"
+ body: >
+ {
+ "docs": [
+ {
+ "_index": "index-1",
+ "_source": {
+ "foo": "bar"
+ }
+ }
+ ],
+ "pipeline_substitutions": {
+ "my_pipeline": {
+ "processors": [],
+ "created_date": "asd"
+ }
+ }
+ }
+ - match: { status: 500 }
+ - match: { error.reason: "/(runtime_exception:\\ )?org\\.elasticsearch\\.ElasticsearchParseException:\\ Provided\\ a\\ pipeline\\ property\\ which\\ is\\ managed\\ by\\ the\\ system,\\ e\\.g\\.\\ `created_date`\\./" }

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

@@ -0,0 +1,77 @@
+/*
+ * 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", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.ingest;
+
+import org.elasticsearch.action.ingest.GetPipelineResponse;
+import org.elasticsearch.cluster.metadata.Metadata;
+import org.elasticsearch.node.NodeService;
+import org.elasticsearch.test.ESIntegTestCase;
+
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.hamcrest.Matchers.lessThanOrEqualTo;
+
+@ESIntegTestCase.ClusterScope(numDataNodes = 3)
+public class PipelineConfigurationSyncIT extends ESIntegTestCase {
+
+ public void testAllNodesGetPipelineTrackingClusterState() throws Exception {
+ final String pipelineId = "_id";
+ GetPipelineResponse getResponse = getPipelines(pipelineId);
+ assertFalse(getResponse.isFound());
+
+ final long timeBeforePut = System.currentTimeMillis();
+ putJsonPipeline(
+ "_id",
+ (builder, params) -> builder.field("description", "my_pipeline").startArray("processors").startObject().endObject().endArray()
+ );
+ final Long timeAfterPut = System.currentTimeMillis();
+
+ getResponse = getPipelines(pipelineId);
+ assertTrue(getResponse.isFound());
+
+ final Pipeline node1Pipeline = internalCluster().getInstance(NodeService.class, "node_s0")
+ .getIngestService()
+ .getPipeline(Metadata.DEFAULT_PROJECT_ID, "_id");
+ final Pipeline node2Pipeline = internalCluster().getInstance(NodeService.class, "node_s1")
+ .getIngestService()
+ .getPipeline(Metadata.DEFAULT_PROJECT_ID, "_id");
+ final Pipeline node3Pipeline = internalCluster().getInstance(NodeService.class, "node_s2")
+ .getIngestService()
+ .getPipeline(Metadata.DEFAULT_PROJECT_ID, "_id");
+
+ assertNotSame(node1Pipeline, node2Pipeline);
+ assertNotSame(node2Pipeline, node3Pipeline);
+
+ assertThat(node1Pipeline.getDescription(), equalTo(node2Pipeline.getDescription()));
+ assertThat(node2Pipeline.getDescription(), equalTo(node3Pipeline.getDescription()));
+
+ // created_date
+ final long node1CreatedDate = node1Pipeline.getCreatedDate().orElseThrow();
+ final long node2CreatedDate = node2Pipeline.getCreatedDate().orElseThrow();
+ final long node3CreatedDate = node3Pipeline.getCreatedDate().orElseThrow();
+
+ assertThat(node1CreatedDate, equalTo(node2CreatedDate));
+ assertThat(node2CreatedDate, equalTo(node3CreatedDate));
+
+ assertThat(node1CreatedDate, greaterThanOrEqualTo(timeBeforePut));
+ assertThat(node1CreatedDate, lessThanOrEqualTo(timeAfterPut));
+
+ // modified_date
+ final long node1ModifiedDate = node1Pipeline.getModifiedDate().orElseThrow();
+ final long node2ModifiedDate = node2Pipeline.getModifiedDate().orElseThrow();
+ final long node3ModifiedDate = node3Pipeline.getModifiedDate().orElseThrow();
+
+ assertThat(node1ModifiedDate, equalTo(node2ModifiedDate));
+ assertThat(node2ModifiedDate, equalTo(node3ModifiedDate));
+
+ assertThat(node1ModifiedDate, greaterThanOrEqualTo(timeBeforePut));
+ assertThat(node1ModifiedDate, lessThanOrEqualTo(timeAfterPut));
+ }
+}

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -338,6 +338,7 @@ static TransportVersion def(int id) {
  public static final TransportVersion ESQL_FIXED_INDEX_LIKE = def(9_119_0_00);
  public static final TransportVersion LOOKUP_JOIN_CCS = def(9_120_0_00);
  public static final TransportVersion NODE_USAGE_STATS_FOR_THREAD_POOLS_IN_CLUSTER_INFO = def(9_121_0_00);
+ public static final TransportVersion PIPELINE_TRACKING_INFO = def(9_122_0_00);
 
  /*
  * STOP! READ THIS FIRST! No, really,

server/src/main/java/org/elasticsearch/action/ingest/ReservedPipelineAction.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);
+ var nopUpdate = IngestService.isNoOpPipelineUpdate(projectMetadata, request, () -> IngestService.readPipelineConfig(request));
 
  if (nopUpdate) {
  continue;

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

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

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

@@ -81,6 +81,8 @@
 import org.elasticsearch.xcontent.XContentBuilder;
 
 import java.io.IOException;
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
@@ -102,6 +104,7 @@
 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;
@@ -542,7 +545,9 @@ public void putPipeline(
  ActionListener<AcknowledgedResponse> listener,
  Consumer<ActionListener<NodesInfoResponse>> nodeInfoListener
  ) throws Exception {
- if (isNoOpPipelineUpdate(state.metadata().getProject(projectId), request)) {
+ Map<String, Object> newPipelineConfig = readPipelineConfig(request);
+ validateNoSystemPropertiesInPipelineConfig(newPipelineConfig);
+ if (isNoOpPipelineUpdate(state.metadata().getProject(projectId), request, () -> newPipelineConfig)) {
  // existing pipeline matches request pipeline -- no need to update
  listener.onResponse(AcknowledgedResponse.TRUE);
  return;
@@ -569,16 +574,35 @@ public void validatePipelineRequest(ProjectId projectId, PutPipelineRequest requ
  validatePipeline(ingestInfos, projectId, request.getId(), config);
  }
 
- public static boolean isNoOpPipelineUpdate(ProjectMetadata metadata, PutPipelineRequest request) {
+ 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) || pipelineConfig.containsKey(Pipeline.MODIFIED_DATE_KEY)) {
+ throw new ElasticsearchParseException("Provided a pipeline property which is managed by the system, e.g. `created_date`.");
+ }
+ }
+
+ public static boolean isNoOpPipelineUpdate(
+ ProjectMetadata metadata,
+ PutPipelineRequest request,
+ Supplier<Map<String, Object>> newPipelineConfigSupplier
+ ) {
  IngestMetadata currentIngestMetadata = metadata.custom(IngestMetadata.TYPE);
  if (request.getVersion() == null
  && currentIngestMetadata != null
  && currentIngestMetadata.getPipelines().containsKey(request.getId())) {
- var pipelineConfig = XContentHelper.convertToMap(request.getSource(), false, request.getXContentType()).v2();
- var currentPipeline = currentIngestMetadata.getPipelines().get(request.getId());
- if (currentPipeline.getConfig().equals(pipelineConfig)) {
- return true;
- }
+
+ 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);
  }
 
  return false;
@@ -750,7 +774,22 @@ public IngestMetadata execute(IngestMetadata currentIngestMetadata, Collection<I
  pipelines = new HashMap<>();
  }
 
- pipelines.put(request.getId(), new PipelineConfiguration(request.getId(), pipelineSource, request.getXContentType()));
+ Instant now = Instant.now().truncatedTo(ChronoUnit.MILLIS);
+ Map<String, Object> newPipelineConfig = XContentHelper.convertToMap(pipelineSource, true, request.getXContentType()).v2();
+ PipelineConfiguration existingPipeline = pipelines.get(request.getId());
+ if (existingPipeline == null) {
+ newPipelineConfig.put(Pipeline.CREATED_DATE_KEY, now.toString());
+ } else {
+ Object existingCreatedAt = existingPipeline.getConfig().get(Pipeline.CREATED_DATE_KEY);
+ // only set/carry over `created_date` if existing pipeline already has it.
+ // would be confusing if existing pipelines were all updated to have `created_date` set to now.
+ if (existingCreatedAt != null) {
+ newPipelineConfig.put(Pipeline.CREATED_DATE_KEY, existingCreatedAt);
+ }
+ }
+ newPipelineConfig.put(Pipeline.MODIFIED_DATE_KEY, now.toString());
+
+ pipelines.put(request.getId(), new PipelineConfiguration(request.getId(), newPipelineConfig));
  return new IngestMetadata(pipelines);
  }
  }