feat: Add execution id logging to uniquely identify request logs

Adds an execution id for each http. When the LOG_EXECUTION_ID env var is set, the execution id is added as a new field to the json logs.

Author: maemayve

.gitignore

@@ -20,6 +20,7 @@
 
 # Maven
 target/
+dependency-reduced-pom.xml
 
 # Gradle
 .gradle

invoker/core/src/main/java/com/google/cloud/functions/invoker/BackgroundFunctionExecutor.java

@@ -21,6 +21,7 @@
 import com.google.cloud.functions.CloudEventsFunction;
 import com.google.cloud.functions.Context;
 import com.google.cloud.functions.RawBackgroundFunction;
+import com.google.cloud.functions.invoker.gcf.ExecutionIdUtil;
 import com.google.gson.Gson;
 import com.google.gson.GsonBuilder;
 import com.google.gson.TypeAdapter;
@@ -51,6 +52,7 @@ public final class BackgroundFunctionExecutor extends HttpServlet {
   private static final Logger logger = Logger.getLogger("com.google.cloud.functions.invoker");
 
   private final FunctionExecutor<?> functionExecutor;
+  private final ExecutionIdUtil executionIdUtil = new ExecutionIdUtil();
 
   private BackgroundFunctionExecutor(FunctionExecutor<?> functionExecutor) {
     this.functionExecutor = functionExecutor;
@@ -322,6 +324,7 @@ void serviceCloudEvent(CloudEvent cloudEvent) throws Exception {
   @Override
   public void service(HttpServletRequest req, HttpServletResponse res) throws IOException {
     String contentType = req.getContentType();
+    executionIdUtil.storeExecutionId(req);
     try {
       if ((contentType != null && contentType.startsWith("application/cloudevents+json"))
           || req.getHeader("ce-specversion") != null) {
@@ -334,6 +337,7 @@ public void service(HttpServletRequest req, HttpServletResponse res) throws IOEx
       res.setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR);
       logger.log(Level.SEVERE, "Failed to execute " + functionExecutor.functionName(), t);
     }
+    executionIdUtil.removeExecutionId();
   }
 
   private enum CloudEventKind {
@@ -348,6 +352,7 @@ private enum CloudEventKind {
    *     FunctionExecutor}.
    */
   private <CloudEventT> void serviceCloudEvent(HttpServletRequest req) throws Exception {
+    executionIdUtil.storeExecutionId(req);
     @SuppressWarnings("unchecked")
     FunctionExecutor<CloudEventT> executor = (FunctionExecutor<CloudEventT>) functionExecutor;
     byte[] body = req.getInputStream().readAllBytes();
@@ -362,6 +367,7 @@ private <CloudEventT> void serviceCloudEvent(HttpServletRequest req) throws Exce
     runWithContextClassLoader(() -> executor.serviceCloudEvent(reader.toEvent(data -> data)));
     // The data->data is a workaround for a bug fixed since Milestone 4 of the SDK, in
     // https://github.com/cloudevents/sdk-java/pull/259.
+    executionIdUtil.removeExecutionId();
   }
 
   private static Map<String, List<String>> headerMap(HttpServletRequest req) {

invoker/core/src/main/java/com/google/cloud/functions/invoker/HttpFunctionExecutor.java

@@ -15,6 +15,7 @@
 package com.google.cloud.functions.invoker;
 
 import com.google.cloud.functions.HttpFunction;
+import com.google.cloud.functions.invoker.gcf.ExecutionIdUtil;
 import com.google.cloud.functions.invoker.http.HttpRequestImpl;
 import com.google.cloud.functions.invoker.http.HttpResponseImpl;
 import java.util.logging.Level;
@@ -28,6 +29,7 @@ public class HttpFunctionExecutor extends HttpServlet {
   private static final Logger logger = Logger.getLogger("com.google.cloud.functions.invoker");
 
   private final HttpFunction function;
+  private final ExecutionIdUtil executionIdUtil = new ExecutionIdUtil();
 
   private HttpFunctionExecutor(HttpFunction function) {
     this.function = function;
@@ -64,6 +66,7 @@ public static HttpFunctionExecutor forClass(Class<?> functionClass) {
   /** Executes the user's method, can handle all HTTP type methods. */
   @Override
   public void service(HttpServletRequest req, HttpServletResponse res) {
+    executionIdUtil.storeExecutionId(req);
     HttpRequestImpl reqImpl = new HttpRequestImpl(req);
     HttpResponseImpl respImpl = new HttpResponseImpl(res);
     ClassLoader oldContextLoader = Thread.currentThread().getContextClassLoader();
@@ -77,5 +80,6 @@ public void service(HttpServletRequest req, HttpServletResponse res) {
       Thread.currentThread().setContextClassLoader(oldContextLoader);
       respImpl.flush();
     }
+    executionIdUtil.removeExecutionId();
   }
 }

invoker/core/src/main/java/com/google/cloud/functions/invoker/gcf/ExecutionIdUtil.java

@@ -0,0 +1,62 @@
+package com.google.cloud.functions.invoker.gcf;
+
+import java.util.Base64;
+import java.util.Random;
+import java.util.logging.Handler;
+import java.util.logging.Logger;
+import javax.servlet.http.HttpServletRequest;
+
+/**
+ * A helper class that fetches either fetches a unique execution id from request HTTP headers or
+ * generates a random id.
+ */
+public final class ExecutionIdUtil {
+  private static final Logger rootLogger = Logger.getLogger("");
+  private static final int EXECUTION_ID_LENGTH = 12;
+  private static final String EXECUTION_ID_HTTP_HEADER = "HTTP_FUNCTION_EXECUTION_ID";
+  private static final String LOG_EXECUTION_ID_ENV_NAME = "LOG_EXECUTION_ID";
+
+  private final Random random = new Random();
+
+  /**
+   * Add mapping to root logger from current thread id to execution id. This mapping will be used to
+   * append the execution id to log lines.
+   */
+  public void storeExecutionId(HttpServletRequest request) {
+    if (!executionIdLoggingEnabled()) {
+      return;
+    }
+    for (Handler handler : rootLogger.getHandlers()) {
+      if (handler instanceof JsonLogHandler) {
+        String id = getOrGenerateExecutionId(request);
+        ((JsonLogHandler) handler).addExecutionId(Thread.currentThread().getId(), id);
+      }
+    }
+  }
+
+  /** Remove mapping from curent thread to request execution id */
+  public void removeExecutionId() {
+    if (!executionIdLoggingEnabled()) {
+      return;
+    }
+    for (Handler handler : rootLogger.getHandlers()) {
+      if (handler instanceof JsonLogHandler) {
+        ((JsonLogHandler) handler).removeExecutionId(Thread.currentThread().getId());
+      }
+    }
+  }
+
+  private String getOrGenerateExecutionId(HttpServletRequest request) {
+    String executionId = request.getHeader(EXECUTION_ID_HTTP_HEADER);
+    if (executionId == null) {
+      byte[] array = new byte[EXECUTION_ID_LENGTH];
+      random.nextBytes(array);
+      executionId = Base64.getEncoder().encodeToString(array);
+    }
+    return executionId;
+  }
+
+  private boolean executionIdLoggingEnabled() {
+    return Boolean.parseBoolean(System.getenv().getOrDefault(LOG_EXECUTION_ID_ENV_NAME, "false"));
+  }
+}

invoker/core/src/main/java/com/google/cloud/functions/invoker/gcf/JsonLogHandler.java

@@ -5,6 +5,8 @@
 import java.io.StringWriter;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
 import java.util.logging.Handler;
 import java.util.logging.Level;
 import java.util.logging.LogRecord;
@@ -24,6 +26,13 @@ public final class JsonLogHandler extends Handler {
 
   private final PrintStream out;
   private final boolean closePrintStreamOnClose;
+  // This map is used to track execution id for currently running Jetty requests. Mapping thread
+  // id to request works because of an implementation detail of Jetty thread pool handling.
+  // Jetty worker threads completely handle a request before beginning work on a new request.
+  // NOTE: Store thread id as a string to avoid comparison failures between int and long.
+  //
+  // Jetty Documentation (https://jetty.org/docs/jetty/10/programming-guide/arch/threads.html)
+  private final ConcurrentMap<String, String> executionIdByThreadMap = new ConcurrentHashMap<>();
 
   public JsonLogHandler(PrintStream out, boolean closePrintStreamOnClose) {
     this.out = out;
@@ -38,6 +47,7 @@ public void publish(LogRecord record) {
     StringBuilder json = new StringBuilder("{");
     appendSeverity(json, record);
     appendSourceLocation(json, record);
+    appendExecutionId(json, record);
     appendMessage(json, record); // must be last, see appendMessage
     json.append("}");
     // We must output the log all at once (should only call println once per call to publish)
@@ -96,6 +106,12 @@ private static void appendSourceLocation(StringBuilder json, LogRecord record) {
     json.append(SOURCE_LOCATION_KEY).append("{").append(String.join(", ", entries)).append("}, ");
   }
 
+  private void appendExecutionId(StringBuilder json, LogRecord record) {
+    json.append("\"execution_id\": \"")
+        .append(executionIdByThreadMap.get(Integer.toString(record.getThreadID())))
+        .append("\", ");
+  }
+
   private static String escapeString(String s) {
     return s.replace("\\", "\\\\").replace("\"", "\\\"").replace("\n", "\\n").replace("\r", "\\r");
   }
@@ -117,4 +133,12 @@ public void close() throws SecurityException {
       out.close();
     }
   }
+
+  public void addExecutionId(long threadId, String executionId) {
+    executionIdByThreadMap.put(Long.toString(threadId), executionId);
+  }
+
+  public void removeExecutionId(long threadId) {
+    executionIdByThreadMap.remove(Long.toString(threadId));
+  }
 }

invoker/core/src/test/java/com/google/cloud/functions/invoker/IntegrationTest.java

@@ -89,6 +89,8 @@ public class IntegrationTest {
   @Rule public final TestName testName = new TestName();
 
   private static final String SERVER_READY_STRING = "Started ServerConnector";
+  private static final String EXECUTION_ID_HTTP_HEADER = "HTTP_FUNCTION_EXECUTION_ID";
+  private static final String EXECUTION_ID = "1234abcd";
 
   private static final ExecutorService EXECUTOR = Executors.newCachedThreadPool();
 
@@ -286,14 +288,18 @@ public void exceptionHttp() throws Exception {
     String exceptionExpectedOutput =
         "\"severity\": \"ERROR\", \"logging.googleapis.com/sourceLocation\": {\"file\":"
             + " \"com/google/cloud/functions/invoker/HttpFunctionExecutor.java\", \"method\":"
-            + " \"service\"}, \"message\": \"Failed to execute"
+            + " \"service\"}, \"execution_id\": \""
+            + EXECUTION_ID
+            + "\","
+            + " \"message\": \"Failed to execute"
             + " com.google.cloud.functions.invoker.testfunctions.ExceptionHttp\\n"
             + "java.lang.RuntimeException: exception thrown for test";
     testHttpFunction(
         fullTarget("ExceptionHttp"),
         ImmutableList.of(
             TestCase.builder()
                 .setExpectedResponseCode(500)
+                .setHttpHeaders(ImmutableMap.of(EXECUTION_ID_HTTP_HEADER, EXECUTION_ID))
                 .setExpectedOutput(exceptionExpectedOutput)
                 .build()));
   }
@@ -303,7 +309,10 @@ public void exceptionBackground() throws Exception {
     String exceptionExpectedOutput =
         "\"severity\": \"ERROR\", \"logging.googleapis.com/sourceLocation\": {\"file\":"
             + " \"com/google/cloud/functions/invoker/BackgroundFunctionExecutor.java\", \"method\":"
-            + " \"service\"}, \"message\": \"Failed to execute"
+            + " \"service\"}, \"execution_id\": \""
+            + EXECUTION_ID
+            + "\", "
+            + "\"message\": \"Failed to execute"
             + " com.google.cloud.functions.invoker.testfunctions.ExceptionBackground\\n"
             + "java.lang.RuntimeException: exception thrown for test";
 
@@ -317,6 +326,7 @@ public void exceptionBackground() throws Exception {
         ImmutableList.of(
             TestCase.builder()
                 .setRequestText(gcfRequestText)
+                .setHttpHeaders(ImmutableMap.of(EXECUTION_ID_HTTP_HEADER, EXECUTION_ID))
                 .setExpectedResponseCode(500)
                 .setExpectedOutput(exceptionExpectedOutput)
                 .build()),
@@ -359,25 +369,37 @@ public void stackDriverLogging() throws Exception {
             + "\"logging.googleapis.com/sourceLocation\": "
             + "{\"file\": \"com/google/cloud/functions/invoker/testfunctions/Log.java\","
             + " \"method\": \"service\"},"
+            + " \"execution_id\": \""
+            + EXECUTION_ID
+            + "\","
             + " \"message\": \"blim\"}";
     TestCase simpleTestCase =
-        TestCase.builder().setUrl("/?message=blim").setExpectedOutput(simpleExpectedOutput).build();
+        TestCase.builder()
+            .setUrl("/?message=blim")
+            .setHttpHeaders(ImmutableMap.of(EXECUTION_ID_HTTP_HEADER, EXECUTION_ID))
+            .setExpectedOutput(simpleExpectedOutput)
+            .build();
     String quotingExpectedOutput = "\"message\": \"foo\\nbar\\\"";
     TestCase quotingTestCase =
         TestCase.builder()
             .setUrl("/?message=" + URLEncoder.encode("foo\nbar\"", "UTF-8"))
+            .setHttpHeaders(ImmutableMap.of(EXECUTION_ID_HTTP_HEADER, EXECUTION_ID))
             .setExpectedOutput(quotingExpectedOutput)
             .build();
     String exceptionExpectedOutput =
         "{\"severity\": \"ERROR\", "
             + "\"logging.googleapis.com/sourceLocation\": "
             + "{\"file\": \"com/google/cloud/functions/invoker/testfunctions/Log.java\", "
             + "\"method\": \"service\"}, "
+            + "\"execution_id\": \""
+            + EXECUTION_ID
+            + "\", "
             + "\"message\": \"oops\\njava.lang.Exception: disaster\\n"
             + "	at com.google.cloud.functions.invoker.testfunctions.Log.service(Log.java:";
     TestCase exceptionTestCase =
         TestCase.builder()
             .setUrl("/?message=oops&level=severe&exception=disaster")
+            .setHttpHeaders(ImmutableMap.of(EXECUTION_ID_HTTP_HEADER, EXECUTION_ID))
             .setExpectedOutput(exceptionExpectedOutput)
             .build();
     testHttpFunction(
@@ -842,7 +864,9 @@ private ServerProcess startServer(
             "FUNCTION_SIGNATURE_TYPE",
             signatureType.toString(),
             "FUNCTION_TARGET",
-            target);
+            target,
+            "LOG_EXECUTION_ID",
+            "true");
     processBuilder.environment().putAll(environment);
     processBuilder.environment().putAll(environmentVariables);
     Process serverProcess = processBuilder.start();