[Logs+] Default pipeline for logs data streams (#95971)

Author: eyalkoren

docs/changelog/95971.yaml

@@ -0,0 +1,7 @@
+pr: 95971
+summary: "Set `@timestamp` for documents in logs data streams if missing and add support for custom pipeline"
+area: Data streams
+type: enhancement
+issues:
+ - 95537
+ - 95551

modules/data-streams/src/yamlRestTest/resources/rest-api-spec/test/data_stream/220_logs_default_pipeline.yml

@@ -0,0 +1,95 @@
+---
+Test default logs-*-* pipeline:
+ - do:
+ # setting up a custom field mapping, to test custom pipeline
+ cluster.put_component_template:
+ name: logs@custom
+ body:
+ template:
+ mappings:
+ properties:
+ custom_timestamp:
+ type: date
+
+ - do:
+ ingest.put_pipeline:
+ # testing custom pipeline - setting a custom timestamp with the same value used to set the `@timestamp` field when missing
+ id: "logs@custom"
+ body: >
+ {
+ "processors": [
+ {
+ "set" : {
+ "field": "custom_timestamp",
+ "copy_from": "_ingest.timestamp"
+ }
+ }
+ ]
+ }
+
+ - do:
+ indices.create_data_stream:
+ name: logs-generic-default
+ - is_true: acknowledged
+
+ - do:
+ indices.get_data_stream:
+ name: logs-generic-default
+ - set: { data_streams.0.indices.0.index_name: idx0name }
+
+ - do:
+ indices.get_mapping:
+ index: logs-generic-default
+ - match: { .$idx0name.mappings.properties.@timestamp.type: "date" }
+
+ - do:
+ index:
+ index: logs-generic-default
+ refresh: true
+ body:
+ # no timestamp - testing default pipeline's @timestamp set processor
+ message: 'no_timestamp'
+ - match: {result: "created"}
+
+ - do:
+ search:
+ index: logs-generic-default
+ body:
+ query:
+ term:
+ message:
+ value: 'no_timestamp'
+ fields:
+ - field: '@timestamp'
+ - field: 'custom_timestamp'
+ - length: { hits.hits: 1 }
+ - match: { hits.hits.0._source.@timestamp: '/[0-9-]+T[0-9:.]+Z/' }
+ - set: {hits.hits.0._source.custom_timestamp: custom_timestamp_source }
+ - match: { hits.hits.0._source.@timestamp: $custom_timestamp_source }
+ - match: { hits.hits.0.fields.@timestamp.0: '/[0-9-]+T[0-9:.]+Z/' }
+ - set: {hits.hits.0.fields.custom_timestamp.0: custom_timestamp_field }
+ - match: { hits.hits.0.fields.@timestamp.0: $custom_timestamp_field }
+
+ # verify that when a document is ingested with a timestamp, it does not get overridden
+ - do:
+ index:
+ index: logs-generic-default
+ refresh: true
+ body:
+ '@timestamp': '2023-05-10'
+ message: 'with_timestamp'
+ - match: {result: "created"}
+
+ - do:
+ search:
+ index: logs-generic-default
+ body:
+ query:
+ term:
+ message:
+ value: 'with_timestamp'
+ fields:
+ - field: '@timestamp'
+ - length: { hits.hits: 1 }
+ - match: { hits.hits.0.fields.@timestamp.0: '2023-05-10T00:00:00.000Z' }
+

x-pack/plugin/core/src/main/resources/logs-default-pipeline.json

@@ -0,0 +1,24 @@
+{
+ "processors": [
+ {
+ "set": {
+ "description": "If '@timestamp' is missing, set it with the ingest timestamp",
+ "field": "@timestamp",
+ "override": false,
+ "copy_from": "_ingest.timestamp"
+ }
+ },
+ {
+ "pipeline" : {
+ "name": "logs@custom",
+ "ignore_missing_pipeline": true,
+ "description": "A custom pipeline for logs data streams, which does not exist by default, but can be added if additional processing is required"
+ }
+ }
+ ],
+ "_meta": {
+ "description": "default pipeline for the logs index template installed by x-pack",
+ "managed": true
+ },
+ "version": ${xpack.stack.template.version}
+}

x-pack/plugin/core/src/main/resources/logs-settings.json

@@ -11,7 +11,8 @@
  },
  "mapping": {
  "ignore_malformed": true
- }
+ },
+ "default_pipeline": "logs-default-pipeline"
  }
  }
  },

x-pack/plugin/stack/src/main/java/org/elasticsearch/xpack/stack/StackTemplateRegistry.java

@@ -23,6 +23,7 @@
 import org.elasticsearch.xpack.core.ilm.LifecyclePolicy;
 import org.elasticsearch.xpack.core.template.IndexTemplateConfig;
 import org.elasticsearch.xpack.core.template.IndexTemplateRegistry;
+import org.elasticsearch.xpack.core.template.IngestPipelineConfig;
 import org.elasticsearch.xpack.core.template.LifecyclePolicyConfig;
 
 import java.io.IOException;
@@ -221,6 +222,15 @@ protected Map<String, ComposableIndexTemplate> getComposableTemplateConfigs() {
  }
  }
 
+ private static final List<IngestPipelineConfig> INGEST_PIPELINE_CONFIGS = List.of(
+ new IngestPipelineConfig("logs-default-pipeline", "/logs-default-pipeline.json", REGISTRY_VERSION, TEMPLATE_VERSION_VARIABLE)
+ );
+
+ @Override
+ protected List<IngestPipelineConfig> getIngestPipelines() {
+ return INGEST_PIPELINE_CONFIGS;
+ }
+
  @Override
  protected String getOrigin() {
  return ClientHelper.STACK_ORIGIN;

x-pack/plugin/stack/src/test/java/org/elasticsearch/xpack/stack/StackTemplateRegistryTests.java

@@ -13,6 +13,7 @@
 import org.elasticsearch.action.ActionType;
 import org.elasticsearch.action.admin.indices.template.put.PutComponentTemplateAction;
 import org.elasticsearch.action.admin.indices.template.put.PutComposableIndexTemplateAction;
+import org.elasticsearch.action.ingest.PutPipelineAction;
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
 import org.elasticsearch.cluster.ClusterChangedEvent;
 import org.elasticsearch.cluster.ClusterName;
@@ -27,6 +28,8 @@
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.TriFunction;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.ingest.IngestMetadata;
+import org.elasticsearch.ingest.PipelineConfiguration;
 import org.elasticsearch.test.ClusterServiceUtils;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.test.client.NoOpClient;
@@ -45,6 +48,7 @@
 import org.elasticsearch.xpack.core.ilm.LifecyclePolicyMetadata;
 import org.elasticsearch.xpack.core.ilm.OperationMode;
 import org.elasticsearch.xpack.core.ilm.action.PutLifecycleAction;
+import org.elasticsearch.xpack.core.template.IngestPipelineConfig;
 import org.junit.After;
 import org.junit.Before;
 
@@ -190,10 +194,40 @@ public void testPolicyAlreadyExists() {
  return null;
  });
 
- ClusterChangedEvent event = createClusterChangedEvent(Collections.emptyMap(), policyMap, nodes);
+ ClusterChangedEvent event = createClusterChangedEvent(Collections.emptyMap(), policyMap, nodes, true);
  registry.clusterChanged(event);
  }
 
+ public void testThatRequiredPipelinesAreAdded() throws Exception {
+ DiscoveryNode node = TestDiscoveryNode.create("node");
+ DiscoveryNodes nodes = DiscoveryNodes.builder().localNodeId("node").masterNodeId("node").add(node).build();
+
+ AtomicInteger calledTimes = new AtomicInteger(0);
+ client.setVerifier((action, request, listener) -> {
+ if (action instanceof PutPipelineAction) {
+ calledTimes.incrementAndGet();
+ return AcknowledgedResponse.TRUE;
+ }
+ if (action instanceof PutComponentTemplateAction) {
+ // Ignore this, it's verified in another test
+ return AcknowledgedResponse.TRUE;
+ } else if (action instanceof PutLifecycleAction) {
+ // Ignore this, it's verified in another test
+ return AcknowledgedResponse.TRUE;
+ } else if (action instanceof PutComposableIndexTemplateAction) {
+ // Ignore this, it's verified in another test
+ return AcknowledgedResponse.TRUE;
+ } else {
+ fail("client called with unexpected request: " + request.toString());
+ }
+ return null;
+ });
+
+ ClusterChangedEvent event = createInitialClusterChangedEvent(nodes);
+ registry.clusterChanged(event);
+ assertBusy(() -> assertThat(calledTimes.get(), equalTo(registry.getIngestPipelines().size())));
+ }
+
  public void testPolicyAlreadyExistsButDiffers() throws IOException {
  DiscoveryNode node = TestDiscoveryNode.create("node");
  DiscoveryNodes nodes = DiscoveryNodes.builder().localNodeId("node").masterNodeId("node").add(node).build();
@@ -235,7 +269,7 @@ public void testPolicyAlreadyExistsButDiffers() throws IOException {
  ) {
  LifecyclePolicy different = LifecyclePolicy.parse(parser, policies.get(0).getName());
  policyMap.put(policies.get(0).getName(), different);
- ClusterChangedEvent event = createClusterChangedEvent(Collections.emptyMap(), policyMap, nodes);
+ ClusterChangedEvent event = createClusterChangedEvent(Collections.emptyMap(), policyMap, nodes, true);
  registry.clusterChanged(event);
  }
  }
@@ -442,7 +476,6 @@ private ActionResponse verifyComponentTemplateInstalled(
  ) {
  if (action instanceof PutComponentTemplateAction) {
  calledTimes.incrementAndGet();
- assertThat(action, instanceOf(PutComponentTemplateAction.class));
  assertThat(request, instanceOf(PutComponentTemplateAction.Request.class));
  final PutComponentTemplateAction.Request putRequest = (PutComponentTemplateAction.Request) request;
  assertThat(putRequest.componentTemplate().version(), equalTo((long) StackTemplateRegistry.REGISTRY_VERSION));
@@ -461,15 +494,20 @@ private ActionResponse verifyComponentTemplateInstalled(
  }
 
  private ClusterChangedEvent createClusterChangedEvent(Map<String, Integer> existingTemplates, DiscoveryNodes nodes) {
- return createClusterChangedEvent(existingTemplates, Collections.emptyMap(), nodes);
+ return createClusterChangedEvent(existingTemplates, Collections.emptyMap(), nodes, true);
+ }
+
+ private ClusterChangedEvent createInitialClusterChangedEvent(DiscoveryNodes nodes) {
+ return createClusterChangedEvent(Collections.emptyMap(), Collections.emptyMap(), nodes, false);
  }
 
  private ClusterChangedEvent createClusterChangedEvent(
  Map<String, Integer> existingTemplates,
  Map<String, LifecyclePolicy> existingPolicies,
- DiscoveryNodes nodes
+ DiscoveryNodes nodes,
+ boolean addRegistryPipelines
  ) {
- ClusterState cs = createClusterState(Settings.EMPTY, existingTemplates, existingPolicies, nodes);
+ ClusterState cs = createClusterState(Settings.EMPTY, existingTemplates, existingPolicies, nodes, addRegistryPipelines);
  ClusterChangedEvent realEvent = new ClusterChangedEvent(
  "created-from-test",
  cs,
@@ -485,7 +523,8 @@ private ClusterState createClusterState(
  Settings nodeSettings,
  Map<String, Integer> existingComponentTemplates,
  Map<String, LifecyclePolicy> existingPolicies,
- DiscoveryNodes nodes
+ DiscoveryNodes nodes,
+ boolean addRegistryPipelines
  ) {
  Map<String, ComponentTemplate> componentTemplates = new HashMap<>();
  for (Map.Entry<String, Integer> template : existingComponentTemplates.entrySet()) {
@@ -499,12 +538,26 @@ private ClusterState createClusterState(
  .collect(Collectors.toMap(Map.Entry::getKey, e -> new LifecyclePolicyMetadata(e.getValue(), Collections.emptyMap(), 1, 1)));
  IndexLifecycleMetadata ilmMeta = new IndexLifecycleMetadata(existingILMMeta, OperationMode.RUNNING);
 
+ // adding the registry pipelines, as they may be dependencies for index templates
+ Map<String, PipelineConfiguration> ingestPipelines = new HashMap<>();
+ if (addRegistryPipelines) {
+ for (IngestPipelineConfig ingestPipelineConfig : registry.getIngestPipelines()) {
+ // we cannot mock PipelineConfiguration as it is a final class
+ ingestPipelines.put(
+ ingestPipelineConfig.getId(),
+ new PipelineConfiguration(ingestPipelineConfig.getId(), ingestPipelineConfig.loadConfig(), XContentType.JSON)
+ );
+ }
+ }
+ IngestMetadata ingestMetadata = new IngestMetadata(ingestPipelines);
+
  return ClusterState.builder(new ClusterName("test"))
  .metadata(
  Metadata.builder()
  .componentTemplates(componentTemplates)
  .transientSettings(nodeSettings)
  .putCustom(IndexLifecycleMetadata.TYPE, ilmMeta)
+ .putCustom(IngestMetadata.TYPE, ingestMetadata)
  .build()
  )
  .blocks(new ClusterBlocks.Builder().build())

x-pack/qa/core-rest-tests-with-security/build.gradle

@@ -4,6 +4,7 @@ dependencies {
  testImplementation project(':x-pack:qa')
  clusterModules project(':modules:mapper-extras')
  clusterModules project(':modules:rank-eval')
+ clusterModules project(':modules:ingest-common')
  clusterModules project(xpackModule('stack'))
  clusterModules project(xpackModule('ilm'))
  clusterModules project(xpackModule('mapper-constant-keyword'))

x-pack/qa/core-rest-tests-with-security/src/yamlRestTest/java/org/elasticsearch/xpack/security/CoreWithSecurityClientYamlTestSuiteIT.java

@@ -35,6 +35,7 @@ public class CoreWithSecurityClientYamlTestSuiteIT extends ESClientYamlSuiteTest
  .module("rank-eval")
  .module("x-pack-ilm")
  .module("x-pack-stack")
+ .module("ingest-common")
  .setting("xpack.security.enabled", "true")
  .setting("xpack.watcher.enabled", "false")
  .setting("xpack.ml.enabled", "false")