Pipelines: Add `created_date` and `modified_date` (#130847)

Add new system-managed properties to pipelines to allow for better tracking of changes.

docs/changelog/130847.yaml

@@ -0,0 +1,5 @@
+pr: 130847
+summary: "Pipelines: Add `created_date` and `modified_date`"
+area: Ingest Node
+type: enhancement
+issues: []

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

@@ -0,0 +1,84 @@
+setup:
+  - requires:
+      test_runner_features: capabilities
+      capabilities:
+        - method: PUT
+          path: /_ingest/pipeline/{id}
+          capabilities: [ pipeline_tracking_info ]
+      reason: "Pipelines have tracking info: modified_date and created_date"
+
+---
+"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:
+        human: true
+        id: "my_pipeline"
+  - gte: { my_pipeline.created_date_millis: 0 }
+  - gte: { my_pipeline.modified_date_millis: 0 }
+  - 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: created_date." }
+
+---
+"Test PUT setting created_date_millis":
+  - do:
+      catch: bad_request
+      ingest.put_pipeline:
+        id: "my_pipeline"
+        body: >
+          {
+            "processors": [],
+            "created_date_millis": 0
+          }
+  - match: { status: 400 }
+  - match: { error.reason: "Provided a pipeline property which is managed by the system: created_date_millis." }
+
+---
+"Test PUT setting modified_date_millis":
+  - do:
+      catch: bad_request
+      ingest.put_pipeline:
+        id: "my_pipeline"
+        body:  >
+          {
+            "processors": [],
+            "modified_date_millis": 0
+          }
+  - match: { status: 400 }
+  - match: { error.reason: "Provided a pipeline property which is managed by the system: modified_date_millis." }
+
+---
+"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: modified_date." }

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

@@ -748,3 +748,42 @@ 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":
+  - requires:
+        test_runner_features: capabilities
+        capabilities:
+          - method: PUT
+            path: /_ingest/pipeline/{id}
+            capabilities: [ pipeline_tracking_info ]
+        reason: "Pipelines have tracking info: modified_date and created_date"
+  - requires:
+      test_runner_features: contains
+  - 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 }
+  - contains: { error.reason: "Provided a pipeline property which is managed by the system: created_date." }

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

@@ -352,6 +352,7 @@ public class TransportVersions {
     public static final TransportVersion ESQL_TOPN_TIMINGS = def(9_128_0_00);
     public static final TransportVersion NODE_WEIGHTS_ADDED_TO_NODE_BALANCE_STATS = def(9_129_0_00);
     public static final TransportVersion RERANK_SNIPPETS = def(9_130_0_00);
+    public static final TransportVersion PIPELINE_TRACKING_INFO = def(9_131_0_00);
 
     /*
      * STOP! READ THIS FIRST! No, really,

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

@@ -13,6 +13,7 @@ import org.elasticsearch.action.ActionResponse;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.core.UpdateForV10;
+import org.elasticsearch.ingest.Pipeline;
 import org.elasticsearch.ingest.PipelineConfiguration;
 import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.xcontent.ToXContentObject;
@@ -74,7 +75,25 @@ public class GetPipelineResponse extends ActionResponse implements ToXContentObj
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
         builder.startObject();
         for (PipelineConfiguration pipeline : pipelines) {
-            builder.field(pipeline.getId(), summary ? Map.of() : pipeline.getConfig());
+            builder.startObject(pipeline.getId());
+            for (final Map.Entry<String, Object> configProperty : (summary ? Map.<String, Object>of() : pipeline.getConfig()).entrySet()) {
+                if (Pipeline.CREATED_DATE_MILLIS.equals(configProperty.getKey())) {
+                    builder.timestampFieldsFromUnixEpochMillis(
+                        Pipeline.CREATED_DATE_MILLIS,
+                        Pipeline.CREATED_DATE,
+                        (Long) configProperty.getValue()
+                    );
+                } else if (Pipeline.MODIFIED_DATE_MILLIS.equals(configProperty.getKey())) {
+                    builder.timestampFieldsFromUnixEpochMillis(
+                        Pipeline.MODIFIED_DATE_MILLIS,
+                        Pipeline.MODIFIED_DATE,
+                        (Long) configProperty.getValue()
+                    );
+                } else {
+                    builder.field(configProperty.getKey(), configProperty.getValue());
+                }
+            }
+            builder.endObject();
         }
         builder.endObject();
         return builder;

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

@@ -49,7 +49,9 @@ class SimulateExecutionService {
                 pipeline.getMetadata(),
                 verbosePipelineProcessor,
                 pipeline.getFieldAccessPattern(),
-                pipeline.getDeprecated()
+                pipeline.getDeprecated(),
+                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/ingest/IngestService.java

@@ -49,7 +49,6 @@ import org.elasticsearch.cluster.service.ClusterService;
 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,9 +77,9 @@ import org.elasticsearch.plugins.internal.XContentParserDecorator;
 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.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
@@ -569,16 +568,36 @@ public class IngestService implements ClusterStateApplier, ReportingService<Inge
         validatePipeline(ingestInfos, projectId, request.getId(), config);
     }
 
+    public static void validateNoSystemPropertiesInPipelineConfig(final Map<String, Object> pipelineConfig) {
+        if (pipelineConfig.containsKey(Pipeline.CREATED_DATE_MILLIS)) {
+            throw new ElasticsearchParseException("Provided a pipeline property which is managed by the system: created_date_millis.");
+        } else if (pipelineConfig.containsKey(Pipeline.CREATED_DATE)) {
+            throw new ElasticsearchParseException("Provided a pipeline property which is managed by the system: created_date.");
+        } else if (pipelineConfig.containsKey(Pipeline.MODIFIED_DATE_MILLIS)) {
+            throw new ElasticsearchParseException("Provided a pipeline property which is managed by the system: modified_date_millis.");
+        } else if (pipelineConfig.containsKey(Pipeline.MODIFIED_DATE)) {
+            throw new ElasticsearchParseException("Provided a pipeline property which is managed by the system: modified_date.");
+        }
+    }
+
+    /** Check whether updating a potentially existing pipeline will be a NOP.
+     * Will return <code>false</code> if request contains system-properties like created or 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 pipelineConfig = XContentHelper.convertToMap(request.getSource(), false, request.getXContentType()).v2();
-            var currentPipeline = currentIngestMetadata.getPipelines().get(request.getId());
-            if (currentPipeline.getConfig().equals(pipelineConfig)) {
-                return true;
-            }
+
+            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_MILLIS);
+            currentConfigWithoutSystemProps.remove(Pipeline.MODIFIED_DATE_MILLIS);
+
+            return newPipelineConfig.equals(currentConfigWithoutSystemProps);
         }
 
         return false;
@@ -676,10 +695,26 @@ public class IngestService implements ClusterStateApplier, ReportingService<Inge
      */
     public static class PutPipelineClusterStateUpdateTask extends PipelineClusterStateUpdateTask {
         private final PutPipelineRequest request;
-
-        PutPipelineClusterStateUpdateTask(ProjectId projectId, ActionListener<AcknowledgedResponse> listener, PutPipelineRequest request) {
+        private final InstantSource instantSource;
+
+        // constructor allowing for injection of InstantSource/time for testing
+        PutPipelineClusterStateUpdateTask(
+            final ProjectId projectId,
+            final ActionListener<AcknowledgedResponse> listener,
+            final PutPipelineRequest request,
+            final InstantSource instantSource
+        ) {
             super(projectId, listener);
             this.request = request;
+            this.instantSource = instantSource;
+        }
+
+        PutPipelineClusterStateUpdateTask(
+            final ProjectId projectId,
+            final ActionListener<AcknowledgedResponse> listener,
+            final PutPipelineRequest request
+        ) {
+            this(projectId, listener, request, Instant::now);
         }
 
         /**
@@ -691,10 +726,15 @@ public class IngestService implements ClusterStateApplier, ReportingService<Inge
 
         @Override
         public IngestMetadata execute(IngestMetadata currentIngestMetadata, Collection<IndexMetadata> allIndexMetadata) {
-            BytesReference pipelineSource = request.getSource();
+            final Map<String, PipelineConfiguration> pipelines = currentIngestMetadata == null
+                ? new HashMap<>(1)
+                : new HashMap<>(currentIngestMetadata.getPipelines());
+            final PipelineConfiguration existingPipeline = pipelines.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,
@@ -705,7 +745,7 @@ public class IngestService implements ClusterStateApplier, ReportingService<Inge
                     );
                 }
 
-                final Integer currentVersion = currentPipeline.getVersion();
+                final Integer currentVersion = existingPipeline.getVersion();
                 if (Objects.equals(request.getVersion(), currentVersion) == false) {
                     throw new IllegalArgumentException(
                         String.format(
@@ -718,9 +758,8 @@ public class IngestService implements ClusterStateApplier, ReportingService<Inge
                     );
                 }
 
-                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,
@@ -733,24 +772,24 @@ public class IngestService implements ClusterStateApplier, ReportingService<Inge
 
                 // 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());
+            final long nowMillis = instantSource.millis();
+            if (existingPipeline == null) {
+                newPipelineConfig.put(Pipeline.CREATED_DATE_MILLIS, nowMillis);
             } else {
-                pipelines = new HashMap<>();
+                Object existingCreatedAt = existingPipeline.getConfig().get(Pipeline.CREATED_DATE_MILLIS);
+                // 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_MILLIS, existingCreatedAt);
+                }
             }
+            newPipelineConfig.put(Pipeline.MODIFIED_DATE_MILLIS, nowMillis);
 
-            pipelines.put(request.getId(), new PipelineConfiguration(request.getId(), pipelineSource, request.getXContentType()));
+            pipelines.put(request.getId(), new PipelineConfiguration(request.getId(), newPipelineConfig));
             return new IngestMetadata(pipelines);
         }
     }
@@ -762,6 +801,7 @@ public class IngestService implements ClusterStateApplier, ReportingService<Inge
         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/Pipeline.java

@@ -18,6 +18,7 @@ import org.elasticsearch.script.ScriptService;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.function.BiConsumer;
 import java.util.function.LongSupplier;
 import java.util.function.Predicate;
@@ -34,6 +35,10 @@ public final class Pipeline {
     public static final String META_KEY = "_meta";
     public static final String FIELD_ACCESS_PATTERN = "field_access_pattern";
     public static final String DEPRECATED_KEY = "deprecated";
+    public static final String CREATED_DATE_MILLIS = "created_date_millis";
+    public static final String CREATED_DATE = "created_date";
+    public static final String MODIFIED_DATE_MILLIS = "modified_date_millis";
+    public static final String MODIFIED_DATE = "modified_date";
 
     private final String id;
     @Nullable
@@ -48,6 +53,10 @@ public final class Pipeline {
     private final IngestPipelineFieldAccessPattern fieldAccessPattern;
     @Nullable
     private final Boolean deprecated;
+    @Nullable
+    private final Long createdDateMillis;
+    @Nullable
+    private final Long modifiedDateMillis;
 
     public Pipeline(
         String id,
@@ -56,7 +65,7 @@ public final class Pipeline {
         @Nullable Map<String, Object> metadata,
         CompoundProcessor compoundProcessor
     ) {
-        this(id, description, version, metadata, compoundProcessor, IngestPipelineFieldAccessPattern.CLASSIC, null);
+        this(id, description, version, metadata, compoundProcessor, IngestPipelineFieldAccessPattern.CLASSIC, null, null, null);
     }
 
     public Pipeline(
@@ -66,9 +75,22 @@ public final class Pipeline {
         @Nullable Map<String, Object> metadata,
         CompoundProcessor compoundProcessor,
         IngestPipelineFieldAccessPattern fieldAccessPattern,
-        @Nullable Boolean deprecated
+        @Nullable Boolean deprecated,
+        @Nullable Long createdDateMillis,
+        @Nullable Long modifiedDateMillis
     ) {
-        this(id, description, version, metadata, compoundProcessor, System::nanoTime, fieldAccessPattern, deprecated);
+        this(
+            id,
+            description,
+            version,
+            metadata,
+            compoundProcessor,
+            System::nanoTime,
+            fieldAccessPattern,
+            deprecated,
+            createdDateMillis,
+            modifiedDateMillis
+        );
     }
 
     // package private for testing
@@ -80,7 +102,9 @@ public final class Pipeline {
         CompoundProcessor compoundProcessor,
         LongSupplier relativeTimeProvider,
         IngestPipelineFieldAccessPattern fieldAccessPattern,
-        @Nullable Boolean deprecated
+        @Nullable Boolean deprecated,
+        @Nullable Long createdDateMillis,
+        @Nullable Long modifiedDateMillis
     ) {
         this.id = id;
         this.description = description;
@@ -91,6 +115,8 @@ public final class Pipeline {
         this.relativeTimeProvider = relativeTimeProvider;
         this.fieldAccessPattern = fieldAccessPattern;
         this.deprecated = deprecated;
+        this.createdDateMillis = createdDateMillis;
+        this.modifiedDateMillis = modifiedDateMillis;
     }
 
     /**
@@ -147,6 +173,8 @@ public final class Pipeline {
             processorFactories,
             projectId
         );
+        String createdDate = ConfigurationUtils.readOptionalStringOrLongProperty(null, null, config, CREATED_DATE_MILLIS);
+        String modifiedDate = ConfigurationUtils.readOptionalStringOrLongProperty(null, null, config, MODIFIED_DATE_MILLIS);
         if (config.isEmpty() == false) {
             throw new ElasticsearchParseException(
                 "pipeline ["
@@ -159,7 +187,19 @@ public final class Pipeline {
             throw new ElasticsearchParseException("pipeline [" + id + "] cannot have an empty on_failure option defined");
         }
         CompoundProcessor compoundProcessor = new CompoundProcessor(false, processors, onFailureProcessors);
-        return new Pipeline(id, description, version, metadata, compoundProcessor, accessPattern, deprecated);
+        Long createdDateMillis = createdDate == null ? null : Long.valueOf(createdDate);
+        Long modifiedDateMillis = modifiedDate == null ? null : Long.valueOf(modifiedDate);
+        return new Pipeline(
+            id,
+            description,
+            version,
+            metadata,
+            compoundProcessor,
+            accessPattern,
+            deprecated,
+            createdDateMillis,
+            modifiedDateMillis
+        );
     }
 
     /**
@@ -265,4 +305,30 @@ public final class Pipeline {
     public boolean isDeprecated() {
         return Boolean.TRUE.equals(deprecated);
     }
+
+    public Optional<Long> getCreatedDateMillis() {
+        return Optional.ofNullable(createdDateMillis);
+    }
+
+    public Optional<Long> getModifiedDateMillis() {
+        return Optional.ofNullable(modifiedDateMillis);
+    }
+
+    @Override
+    public String toString() {
+        final StringBuilder sb = new StringBuilder("Pipeline{");
+        sb.append("id='").append(id).append('\'');
+        sb.append(", description='").append(description).append('\'');
+        sb.append(", version=").append(version);
+        sb.append(", metadata=").append(metadata);
+        sb.append(", compoundProcessor=").append(compoundProcessor);
+        sb.append(", metrics=").append(metrics);
+        sb.append(", relativeTimeProvider=").append(relativeTimeProvider);
+        sb.append(", fieldAccessPattern=").append(fieldAccessPattern);
+        sb.append(", deprecated=").append(deprecated);
+        sb.append(", createdDateMillis='").append(createdDateMillis).append('\'');
+        sb.append(", modifiedDateMillis='").append(modifiedDateMillis).append('\'');
+        sb.append('}');
+        return sb.toString();
+    }
 }

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

@@ -9,6 +9,7 @@
 
 package org.elasticsearch.ingest;
 
+import org.elasticsearch.TransportVersion;
 import org.elasticsearch.TransportVersions;
 import org.elasticsearch.cluster.Diff;
 import org.elasticsearch.cluster.SimpleDiffable;
@@ -29,6 +30,7 @@ import org.elasticsearch.xcontent.json.JsonXContent;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -189,12 +191,14 @@ public final class PipelineConfiguration implements SimpleDiffable<PipelineConfi
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
+        final TransportVersion transportVersion = out.getTransportVersion();
+        final Map<String, Object> configForTransport = configForTransport(transportVersion);
         out.writeString(id);
-        if (out.getTransportVersion().onOrAfter(TransportVersions.V_8_17_0)) {
-            out.writeGenericMap(config);
+        if (transportVersion.onOrAfter(TransportVersions.V_8_17_0)) {
+            out.writeGenericMap(configForTransport);
         } else {
             XContentBuilder builder = XContentBuilder.builder(JsonXContent.jsonXContent).prettyPrint();
-            builder.map(config);
+            builder.map(configForTransport);
             out.writeBytesReference(BytesReference.bytes(builder));
             XContentHelper.writeTo(out, XContentType.JSON);
         }
@@ -246,4 +250,18 @@ public final class PipelineConfiguration implements SimpleDiffable<PipelineConfi
             return this;
         }
     }
+
+    private Map<String, Object> configForTransport(final TransportVersion transportVersion) {
+        final boolean transportSupportsNewProperties = transportVersion.onOrAfter(TransportVersions.PIPELINE_TRACKING_INFO);
+        final boolean noNewProperties = config.containsKey(Pipeline.CREATED_DATE_MILLIS) == false
+            && config.containsKey(Pipeline.MODIFIED_DATE_MILLIS) == false;
+
+        if (transportSupportsNewProperties || noNewProperties) {
+            return config;
+        }
+        final Map<String, Object> configWithoutNewSystemProperties = new HashMap<>(config);
+        configWithoutNewSystemProperties.remove(Pipeline.CREATED_DATE_MILLIS);
+        configWithoutNewSystemProperties.remove(Pipeline.MODIFIED_DATE_MILLIS);
+        return Collections.unmodifiableMap(configWithoutNewSystemProperties);
+    }
 }

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

@@ -53,9 +53,13 @@ public class SimulateIngestService extends IngestService {
         if (rawPipelineSubstitutions != null) {
             for (Map.Entry<String, Map<String, Object>> entry : rawPipelineSubstitutions.entrySet()) {
                 String pipelineId = entry.getKey();
+                Map<String, Object> pipelineConfig = entry.getValue();
+
+                IngestService.validateNoSystemPropertiesInPipelineConfig(pipelineConfig);
+
                 Pipeline pipeline = Pipeline.create(
                     pipelineId,
-                    entry.getValue(),
+                    pipelineConfig,
                     ingestService.getProcessorFactories(),
                     ingestService.getScriptService(),
                     ingestService.getProjectResolver().getProjectId(),

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

@@ -217,7 +217,9 @@ public final class TrackingResultProcessor implements Processor {
             pipeline.getMetadata(),
             verbosePipelineProcessor,
             pipeline.getFieldAccessPattern(),
-            pipeline.getDeprecated()
+            pipeline.getDeprecated(),
+            pipeline.getCreatedDateMillis().orElse(null),
+            pipeline.getModifiedDateMillis().orElse(null)
         );
         ingestDocument.executePipeline(verbosePipeline, handler);
     }

server/src/main/java/org/elasticsearch/rest/action/ingest/RestPutPipelineAction.java

@@ -25,6 +25,7 @@ import org.elasticsearch.xcontent.XContentType;
 import java.io.IOException;
 import java.util.List;
 import java.util.Locale;
+import java.util.Set;
 
 import static org.elasticsearch.rest.RestRequest.Method.PUT;
 import static org.elasticsearch.rest.RestUtils.getAckTimeout;
@@ -73,4 +74,10 @@ public class RestPutPipelineAction extends BaseRestHandler {
             ActionListener.withRef(new RestToXContentListener<>(channel), content)
         );
     }
+
+    @Override
+    public Set<String> supportedCapabilities() {
+        // pipeline_tracking info: `{created,modified}_date` system properties defined within pipeline definition.
+        return Set.of("pipeline_tracking_info");
+    }
 }

server/src/test/java/org/elasticsearch/ingest/IngestServiceTests.java

@@ -81,6 +81,8 @@ import org.mockito.invocation.InvocationOnMock;
 
 import java.io.OutputStream;
 import java.nio.charset.StandardCharsets;
+import java.time.Instant;
+import java.time.InstantSource;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Comparator;
@@ -814,6 +816,65 @@ public class IngestServiceTests extends ESTestCase {
         assertThat(pipeline.getProcessors().size(), equalTo(0));
     }
 
+    public void testPutWithTracking() {
+        final IngestService ingestService = createWithProcessors(Map.of());
+        final String id = "_id";
+        final ProjectId projectId = randomProjectIdOrDefault();
+        ClusterState clusterState = ClusterState.builder(ClusterName.DEFAULT)
+            .putProjectMetadata(ProjectMetadata.builder(projectId).build())
+            .build(); // Start empty
+        final AtomicInteger instantSourceInvocationCounter = new AtomicInteger();
+        final InstantSource instantSource = () -> Instant.ofEpochMilli(instantSourceInvocationCounter.getAndIncrement());
+
+        // add a new pipeline:
+        PutPipelineRequest putRequest = putJsonPipelineRequest(id, "{\"processors\": []}");
+        ClusterState previousClusterState = clusterState;
+        clusterState = executePutWithInstantSource(projectId, putRequest, clusterState, instantSource);
+        ingestService.applyClusterState(new ClusterChangedEvent("", clusterState, previousClusterState));
+        Pipeline pipeline = ingestService.getPipeline(projectId, id);
+        assertThat(pipeline.getCreatedDateMillis().orElseThrow(), is(0L));
+        assertThat(pipeline.getModifiedDateMillis().orElseThrow(), is(0L));
+
+        // overwrite existing pipeline:
+        putRequest = putJsonPipelineRequest(id, """
+            {"processors": [], "description": "_description"}""");
+        previousClusterState = clusterState;
+        clusterState = executePutWithInstantSource(projectId, putRequest, clusterState, instantSource);
+        ingestService.applyClusterState(new ClusterChangedEvent("", clusterState, previousClusterState));
+        pipeline = ingestService.getPipeline(projectId, id);
+        assertThat(pipeline.getCreatedDateMillis().orElseThrow(), is(0L));
+        assertThat(pipeline.getModifiedDateMillis().orElseThrow(), is(1L));
+    }
+
+    public void testPutWithTrackingExistingPipelineWithoutCreatedAtOnlyHasModifiedAt() {
+        final IngestService ingestService = createWithProcessors();
+        final String id = "_id";
+        final ProjectId projectId = randomProjectIdOrDefault();
+        ClusterState clusterState = ClusterState.builder(ClusterName.DEFAULT)
+            .putProjectMetadata(
+                ProjectMetadata.builder(projectId)
+                    .putCustom(
+                        IngestMetadata.TYPE,
+                        new IngestMetadata(
+                            Map.of(id, new PipelineConfiguration(id, Map.of("description", "existing_processor_description")))
+                        )
+                    )
+                    .build()
+            )
+            .build(); // Start empty
+        final AtomicInteger instantSourceInvocationCounter = new AtomicInteger();
+        final InstantSource instantSource = () -> Instant.ofEpochMilli(instantSourceInvocationCounter.getAndIncrement());
+
+        // update existing pipeline which doesn't have `created_date`
+        PutPipelineRequest putRequest = putJsonPipelineRequest(id, "{\"processors\": []}");
+        ClusterState previousClusterState = clusterState;
+        clusterState = executePutWithInstantSource(projectId, putRequest, clusterState, instantSource);
+        ingestService.applyClusterState(new ClusterChangedEvent("", clusterState, previousClusterState));
+        final Pipeline pipeline = ingestService.getPipeline(projectId, id);
+        assertTrue(pipeline.getCreatedDateMillis().isEmpty());
+        assertThat(pipeline.getModifiedDateMillis().orElseThrow(), is(0L));
+    }
+
     public void testPutWithErrorResponse() throws IllegalAccessException {
         IngestService ingestService = createWithProcessors();
         String id = "_id";
@@ -3451,8 +3512,21 @@ public class IngestServiceTests extends ESTestCase {
     }
 
     private static List<IngestService.PipelineClusterStateUpdateTask> oneTask(ProjectId projectId, PutPipelineRequest request) {
+        return oneTaskWithInstantSource(projectId, request, Instant::now);
+    }
+
+    private static List<IngestService.PipelineClusterStateUpdateTask> oneTaskWithInstantSource(
+        final ProjectId projectId,
+        final PutPipelineRequest request,
+        final InstantSource instantSource
+    ) {
         return List.of(
-            new IngestService.PutPipelineClusterStateUpdateTask(projectId, ActionTestUtils.assertNoFailureListener(t -> {}), request)
+            new IngestService.PutPipelineClusterStateUpdateTask(
+                projectId,
+                ActionTestUtils.assertNoFailureListener(t -> {}),
+                request,
+                instantSource
+            )
         );
     }
 
@@ -3461,8 +3535,21 @@ public class IngestServiceTests extends ESTestCase {
     }
 
     private static ClusterState executePut(ProjectId projectId, PutPipelineRequest request, ClusterState clusterState) {
+        return executePutWithInstantSource(projectId, request, clusterState, Instant::now);
+    }
+
+    private static ClusterState executePutWithInstantSource(
+        final ProjectId projectId,
+        final PutPipelineRequest request,
+        final ClusterState clusterState,
+        final InstantSource instantSource
+    ) {
         try {
-            return executeAndAssertSuccessful(clusterState, IngestService.PIPELINE_TASK_EXECUTOR, oneTask(projectId, request));
+            return executeAndAssertSuccessful(
+                clusterState,
+                IngestService.PIPELINE_TASK_EXECUTOR,
+                oneTaskWithInstantSource(projectId, request, instantSource)
+            );
         } catch (Exception e) {
             throw new AssertionError(e);
         }

server/src/test/java/org/elasticsearch/ingest/PipelineProcessorTests.java

@@ -167,6 +167,8 @@ public class PipelineProcessorTests extends ESTestCase {
             new CompoundProcessor(pipeline1Processor),
             relativeTimeProvider,
             IngestPipelineFieldAccessPattern.CLASSIC,
+            null,
+            null,
             null
         );
 
@@ -183,13 +185,15 @@ public class PipelineProcessorTests extends ESTestCase {
             }), pipeline2Processor), List.of()),
             relativeTimeProvider,
             IngestPipelineFieldAccessPattern.CLASSIC,
+            null,
+            null,
             null
         );
         relativeTimeProvider = mock(LongSupplier.class);
         when(relativeTimeProvider.getAsLong()).thenReturn(0L, TimeUnit.MILLISECONDS.toNanos(2));
         Pipeline pipeline3 = new Pipeline(pipeline3Id, null, null, null, new CompoundProcessor(new TestProcessor(ingestDocument -> {
             throw new RuntimeException("error");
-        })), relativeTimeProvider, IngestPipelineFieldAccessPattern.CLASSIC, null);
+        })), relativeTimeProvider, IngestPipelineFieldAccessPattern.CLASSIC, null, null, null);
         when(ingestService.getPipeline(pipeline1Id)).thenReturn(pipeline1);
         when(ingestService.getPipeline(pipeline2Id)).thenReturn(pipeline2);
         when(ingestService.getPipeline(pipeline3Id)).thenReturn(pipeline3);