[Logs+] Add pipeline that parses JSON log events into top-level fields

docs/changelog/96083.yaml

@@ -0,0 +1,6 @@
+pr: 96083
+summary: Automatically parse log events in logs data streams, if their `message` field contains JSON content
+area: Data streams
+type: enhancement
+issues:
+ - 95522

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

@@ -92,4 +92,3 @@ Test default logs-*-* pipeline:
  - field: '@timestamp'
  - length: { hits.hits: 1 }
  - match: { hits.hits.0.fields.@timestamp.0: '2023-05-10T00:00:00.000Z' }
-

modules/data-streams/src/yamlRestTest/resources/rest-api-spec/test/data_stream/230_logs_message_pipeline.yml

@@ -0,0 +1,114 @@
+---
+Test log message JSON-parsing pipeline:
+ - do:
+ ingest.put_pipeline:
+ # opting in to use the JSON parsing pipeline for message field
+ id: "logs@custom"
+ body: >
+ {
+ "processors": [
+ {
+ "pipeline" : {
+ "name": "logs@json-message",
+ "description": "A pipeline that automatically parses JSON log events into top-level fields if they are such"
+ }
+ }
+ ]
+ }
+
+ - do:
+ indices.create_data_stream:
+ name: logs-generic-default
+ - is_true: acknowledged
+
+ - do:
+ index:
+ index: logs-generic-default
+ refresh: true
+ body:
+ '@timestamp': '2023-05-10'
+ message: |-
+ {
+ "@timestamp":"2023-05-09T16:48:34.135Z",
+ "message":"json",
+ "log.level": "INFO",
+ "ecs.version": "1.6.0",
+ "service.name":"my-app",
+ "event.dataset":"my-app.RollingFile",
+ "process.thread.name":"main",
+ "log.logger":"root.pkg.MyApp"
+ }
+ - match: {result: "created"}
+
+ - do:
+ search:
+ index: logs-generic-default
+ body:
+ query:
+ term:
+ message:
+ value: 'json'
+ fields:
+ - field: 'message'
+ - length: { hits.hits: 1 }
+ # root field parsed from JSON should win
+ - match: { hits.hits.0._source.@timestamp: '2023-05-09T16:48:34.135Z' }
+ - match: { hits.hits.0._source.message: 'json' }
+ - match: { hits.hits.0.fields.message.0: 'json' }
+ # successful access to subfields verifies that dot expansion is part of the pipeline
+ - match: { hits.hits.0._source.log.level: 'INFO' }
+ - match: { hits.hits.0._source.ecs.version: '1.6.0' }
+ - match: { hits.hits.0._source.service.name: 'my-app' }
+ - match: { hits.hits.0._source.event.dataset: 'my-app.RollingFile' }
+ - match: { hits.hits.0._source.process.thread.name: 'main' }
+ - match: { hits.hits.0._source.log.logger: 'root.pkg.MyApp' }
+ # _tmp_json_message should be removed by the pipeline
+ - match: { hits.hits.0._source._tmp_json_message: null }
+
+ # test malformed-JSON parsing - parsing error should be ignored and the document should be indexed with original message
+ - do:
+ index:
+ index: logs-generic-default
+ refresh: true
+ body:
+ '@timestamp': '2023-05-10'
+ test: 'malformed_json'
+ message: '{"@timestamp":"2023-05-09T16:48:34.135Z", "message":"malformed_json"}}'
+ - match: {result: "created"}
+
+ - do:
+ search:
+ index: logs-generic-default
+ body:
+ query:
+ term:
+ test:
+ value: 'malformed_json'
+ - length: { hits.hits: 1 }
+ - match: { hits.hits.0._source.@timestamp: '2023-05-10' }
+ - match: { hits.hits.0._source.message: '{"@timestamp":"2023-05-09T16:48:34.135Z", "message":"malformed_json"}}' }
+ - match: { hits.hits.0._source._tmp_json_message: null }
+
+ # test non-string message field
+ - do:
+ index:
+ index: logs-generic-default
+ refresh: true
+ body:
+ test: 'numeric_message'
+ message: 42
+ - match: {result: "created"}
+
+ - do:
+ search:
+ index: logs-generic-default
+ body:
+ query:
+ term:
+ test:
+ value: 'numeric_message'
+ fields:
+ - field: 'message'
+ - length: { hits.hits: 1 }
+ - match: { hits.hits.0._source.message: 42 }
+ - match: { hits.hits.0.fields.message.0: '42' }

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/template/IndexTemplateRegistry.java

@@ -562,34 +562,57 @@ private void addIngestPipelinesIfMissing(ClusterState state) {
  );
 
  if (creationCheck.compareAndSet(false, true)) {
- PipelineConfiguration existingPipeline = findInstalledPipeline(state, requiredPipeline.getId());
- if (existingPipeline != null) {
- Integer existingPipelineVersion = existingPipeline.getVersion();
- if (existingPipelineVersion == null || existingPipelineVersion < requiredPipeline.getVersion()) {
- logger.info(
- "upgrading ingest pipeline [{}] for [{}] from version [{}] to version [{}]",
- requiredPipeline.getId(),
- getOrigin(),
- existingPipelineVersion,
- requiredPipeline.getVersion()
- );
- putIngestPipeline(requiredPipeline, creationCheck);
+ List<String> pipelineDependencies = requiredPipeline.getPipelineDependencies();
+ if (pipelineDependencies != null && pipelineDependenciesExist(state, pipelineDependencies) == false) {
+ creationCheck.set(false);
+ logger.trace(
+ "not adding ingest pipeline [{}] for [{}] because its dependencies do not exist",
+ requiredPipeline.getId(),
+ getOrigin()
+ );
+ } else {
+ PipelineConfiguration existingPipeline = findInstalledPipeline(state, requiredPipeline.getId());
+ if (existingPipeline != null) {
+ Integer existingPipelineVersion = existingPipeline.getVersion();
+ if (existingPipelineVersion == null || existingPipelineVersion < requiredPipeline.getVersion()) {
+ logger.info(
+ "upgrading ingest pipeline [{}] for [{}] from version [{}] to version [{}]",
+ requiredPipeline.getId(),
+ getOrigin(),
+ existingPipelineVersion,
+ requiredPipeline.getVersion()
+ );
+ putIngestPipeline(requiredPipeline, creationCheck);
+ } else {
+ creationCheck.set(false);
+ logger.debug(
+ "not adding ingest pipeline [{}] for [{}], because it already exists",
+ requiredPipeline.getId(),
+ getOrigin()
+ );
+ }
  } else {
  logger.debug(
- "not adding ingest pipeline [{}] for [{}], because it already exists",
+ "adding ingest pipeline [{}] for [{}], because it doesn't exist",
  requiredPipeline.getId(),
  getOrigin()
  );
- creationCheck.set(false);
+ putIngestPipeline(requiredPipeline, creationCheck);
  }
- } else {
- logger.debug("adding ingest pipeline [{}] for [{}], because it doesn't exist", requiredPipeline.getId(), getOrigin());
- putIngestPipeline(requiredPipeline, creationCheck);
  }
  }
  }
  }
 
+ private boolean pipelineDependenciesExist(ClusterState state, List<String> dependencies) {
+ for (String dependency : dependencies) {
+ if (findInstalledPipeline(state, dependency) == null) {
+ return false;
+ }
+ }
+ return true;
+ }
+
  @Nullable
  private static PipelineConfiguration findInstalledPipeline(ClusterState state, String pipelineId) {
  Optional<IngestMetadata> maybeMeta = Optional.ofNullable(state.metadata().custom(IngestMetadata.TYPE));

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/template/IngestPipelineConfig.java

@@ -10,6 +10,8 @@
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.bytes.BytesReference;
 
+import java.util.Collections;
+import java.util.List;
 import java.util.Objects;
 
 /**
@@ -21,11 +23,23 @@ public class IngestPipelineConfig {
  private final int version;
  private final String versionProperty;
 
+ /**
+ * A list of this pipeline's dependencies, for example- such referred to through a pipeline processor.
+ * This list is used to enforce proper ordering of pipeline installation, so that a pipeline gets installed only if all its
+ * dependencies are already installed.
+ */
+ private final List<String> dependencies;
+
  public IngestPipelineConfig(String id, String resource, int version, String versionProperty) {
+ this(id, resource, version, versionProperty, Collections.emptyList());
+ }
+
+ public IngestPipelineConfig(String id, String resource, int version, String versionProperty, List<String> dependencies) {
  this.id = Objects.requireNonNull(id);
  this.resource = Objects.requireNonNull(resource);
  this.version = version;
  this.versionProperty = Objects.requireNonNull(versionProperty);
+ this.dependencies = dependencies;
  }
 
  public String getId() {
@@ -40,6 +54,10 @@ public String getVersionProperty() {
  return versionProperty;
  }
 
+ public List<String> getPipelineDependencies() {
+ return dependencies;
+ }
+
  public BytesReference loadConfig() {
  return new BytesArray(TemplateUtils.loadTemplate(resource, String.valueOf(version), versionProperty));
  }

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

@@ -0,0 +1,48 @@
+{
+ "processors": [
+ {
+ "rename": {
+ "if": "ctx.message instanceof String && ctx.message.startsWith('{') && ctx.message.endsWith('}')",
+ "field": "message",
+ "target_field": "_tmp_json_message",
+ "ignore_missing": true
+ }
+ },
+ {
+ "json": {
+ "if": "ctx._tmp_json_message != null",
+ "field": "_tmp_json_message",
+ "add_to_root": true,
+ "add_to_root_conflict_strategy": "merge",
+ "allow_duplicate_keys": true,
+ "on_failure": [
+ {
+ "rename": {
+ "field": "_tmp_json_message",
+ "target_field": "message",
+ "ignore_missing": true
+ }
+ }
+ ]
+ }
+ },
+ {
+ "dot_expander" : {
+ "if": "ctx._tmp_json_message != null",
+ "field": "*",
+ "override": true
+ }
+ },
+ {
+ "remove" : {
+ "field": "_tmp_json_message",
+ "ignore_missing": true
+ }
+ }
+ ],
+ "_meta": {
+ "description": "automatic parsing of JSON log messages",
+ "managed": true
+ },
+ "version": ${xpack.stack.template.version}
+}

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/template/IndexTemplateRegistryTests.java

@@ -83,14 +83,14 @@ public void tearDown() throws Exception {
  threadPool.shutdownNow();
  }
 
- public void testThatPipelinesAreAddedImmediately() throws Exception {
+ public void testThatIndependentPipelinesAreAddedImmediately() 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) {
- assertPutPipelineAction(calledTimes, action, request, listener);
+ assertPutPipelineAction(calledTimes, action, request, listener, "custom-plugin-final_pipeline");
  return AcknowledgedResponse.TRUE;
  } else {
  // the composable template is not expected to be added, as it's dependency is not available in the cluster state
@@ -102,7 +102,34 @@ public void testThatPipelinesAreAddedImmediately() throws Exception {
 
  ClusterChangedEvent event = createClusterChangedEvent(Collections.emptyMap(), nodes);
  registry.clusterChanged(event);
- assertBusy(() -> assertThat(calledTimes.get(), equalTo(2)));
+ assertBusy(() -> assertThat(calledTimes.get(), equalTo(1)));
+ }
+
+ public void testThatDependentPipelinesAreAddedIfDependenciesExist() 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) {
+ assertPutPipelineAction(calledTimes, action, request, listener, "custom-plugin-default_pipeline");
+ return AcknowledgedResponse.TRUE;
+ } else {
+ // the composable template is not expected to be added, as it's dependency is not available in the cluster state
+ // custom-plugin-settings.json is not expected to be added as it contains a dependency on the default_pipeline
+ fail("client called with unexpected request: " + request.toString());
+ return null;
+ }
+ });
+
+ ClusterChangedEvent event = createClusterChangedEvent(
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ Map.of("custom-plugin-final_pipeline", 3),
+ nodes
+ );
+ registry.clusterChanged(event);
+ assertBusy(() -> assertThat(calledTimes.get(), equalTo(1)));
  }
 
  public void testThatTemplateIsAddedIfAllDependenciesExist() throws Exception {
@@ -138,7 +165,7 @@ public void testThatTemplateIsNotAddedIfNotAllDependenciesExist() throws Excepti
  AtomicInteger calledTimes = new AtomicInteger(0);
  client.setVerifier((action, request, listener) -> {
  if (action instanceof PutPipelineAction) {
- assertPutPipelineAction(calledTimes, action, request, listener);
+ assertPutPipelineAction(calledTimes, action, request, listener, "custom-plugin-default_pipeline");
  return AcknowledgedResponse.TRUE;
  } else {
  // the template is not expected to be added, as the final pipeline is missing
@@ -150,7 +177,7 @@ public void testThatTemplateIsNotAddedIfNotAllDependenciesExist() throws Excepti
  ClusterChangedEvent event = createClusterChangedEvent(
  Collections.emptyMap(),
  Collections.emptyMap(),
- Map.of("custom-plugin-default_pipeline", 3),
+ Map.of("custom-plugin-final_pipeline", 3),
  nodes
  );
  registry.clusterChanged(event);
@@ -188,7 +215,14 @@ public void testThatTemplatesAreUpgradedWhenNeeded() throws Exception {
  AtomicInteger calledTimes = new AtomicInteger(0);
  client.setVerifier((action, request, listener) -> {
  if (action instanceof PutPipelineAction) {
- assertPutPipelineAction(calledTimes, action, request, listener);
+ assertPutPipelineAction(
+ calledTimes,
+ action,
+ request,
+ listener,
+ "custom-plugin-default_pipeline",
+ "custom-plugin-final_pipeline"
+ );
  return AcknowledgedResponse.TRUE;
  } else if (action instanceof PutComponentTemplateAction) {
  assertPutComponentTemplate(calledTimes, action, request, listener);
@@ -277,12 +311,13 @@ private static void assertPutPipelineAction(
  AtomicInteger calledTimes,
  ActionType<?> action,
  ActionRequest request,
- ActionListener<?> listener
+ ActionListener<?> listener,
+ String... pipelineIds
  ) {
  assertThat(action, instanceOf(PutPipelineAction.class));
  assertThat(request, instanceOf(PutPipelineRequest.class));
  final PutPipelineRequest putRequest = (PutPipelineRequest) request;
- assertThat(putRequest.getId(), oneOf("custom-plugin-default_pipeline", "custom-plugin-final_pipeline"));
+ assertThat(putRequest.getId(), oneOf(pipelineIds));
  PipelineConfiguration pipelineConfiguration = new PipelineConfiguration(
  putRequest.getId(),
  putRequest.getSource(),