Include pipeline and plugin IDs to the JSON logs. (#18470)

* Include pipeline and plugin IDs to the JSON logs.

* Add unit tests for plugin and pipeline IDs.

Author: mashhurs

logstash-core/src/main/java/org/logstash/log/CustomLogEventSerializer.java

@@ -31,6 +31,7 @@
 import org.apache.logging.log4j.Logger;
 import java.io.IOException;
 import java.util.Map;
+import java.util.Objects;
 
 import static org.logstash.ObjectMappers.LOG4J_JSON_MAPPER;
 
@@ -48,6 +49,17 @@ public void serialize(CustomLogEvent event, JsonGenerator generator, SerializerP
         generator.writeObjectField("loggerName", event.getLoggerName());
         generator.writeObjectField("timeMillis", event.getTimeMillis());
         generator.writeObjectField("thread", event.getThreadName());
+
+        final String pipelineId = event.getContextData().getValue("pipeline.id");
+        if (Objects.nonNull(pipelineId) && !pipelineId.isEmpty()) {
+            generator.writeStringField("pipeline.id", pipelineId);
+        }
+
+        final String pluginId = event.getContextData().getValue("plugin.id");
+        if (Objects.nonNull(pluginId) && !pluginId.isEmpty()) {
+            generator.writeStringField("plugin.id", pluginId);
+        }
+
         generator.writeFieldName("logEvent");
         generator.writeStartObject();
 

logstash-core/src/test/java/org/logstash/log/CustomLogEventTests.java

@@ -47,10 +47,12 @@
 import com.fasterxml.jackson.core.JsonProcessingException;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.apache.logging.log4j.ThreadContext;
 import org.apache.logging.log4j.junit.LoggerContextRule;
 import org.apache.logging.log4j.test.appender.ListAppender;
 import org.jruby.RubyHash;
 import org.jruby.runtime.builtin.IRubyObject;
+import org.junit.After;
 import org.junit.ClassRule;
 import org.junit.Test;
 import org.logstash.ObjectMappers;
@@ -59,7 +61,6 @@
 import static junit.framework.TestCase.assertFalse;
 import static junit.framework.TestCase.assertEquals;
 import static junit.framework.TestCase.assertNotNull;
-import static org.junit.Assert.assertTrue;
 
 public class CustomLogEventTests {
     private static final String CONFIG = "log4j2-test1.xml";
@@ -68,6 +69,11 @@ public class CustomLogEventTests {
     @ClassRule
     public static LoggerContextRule CTX = new LoggerContextRule(CONFIG);
 
+    @After
+    public void tearDown() {
+        ThreadContext.clearAll();
+    }
+
     @Test
     public void testPatternLayout() {
         ListAppender appender = CTX.getListAppender("EventLogger").clear();
@@ -206,4 +212,29 @@ public void testJSONLayoutWhenParamsContainsAnotherMessageField() throws JsonPro
             System.setProperty(STRICT_JSON_PROPERTY_NAME, prevSetting);
         }
     }
+
+    @Test
+    @SuppressWarnings("unchecked")
+    public void testJSONLayoutWithPipelineIdAndPluginIds() throws JsonProcessingException {
+        ListAppender appender = CTX.getListAppender("JSONEventLogger").clear();
+        Logger logger = LogManager.getLogger("JSONEventLogger");
+
+        ThreadContext.put("pipeline.id", "main-pipeline");
+        ThreadContext.put("plugin.id", "elasticsearch-output-xyz");
+        logger.debug("Both context fields test");
+        ThreadContext.remove("pipeline.id");
+        ThreadContext.remove("plugin.id");
+
+        List<String> messages = appender.getMessages();
+        assertEquals(1, messages.size());
+
+        Map<String, Object> result = ObjectMappers.JSON_MAPPER.readValue(messages.get(0), Map.class);
+
+        assertEquals("DEBUG", result.get("level"));
+        assertEquals("main-pipeline", result.get("pipeline.id"));
+        assertEquals("elasticsearch-output-xyz", result.get("plugin.id"));
+
+        Map<String, Object> logEvent = (Map<String, Object>) result.get("logEvent");
+        assertEquals("Both context fields test", logEvent.get("message"));
+    }
 }