fix: structured log handler formatting issues (#319)

Author: daniel-sanche

packages/google-cloud-logging/google/cloud/logging_v2/handlers/structured_log.py

@@ -14,13 +14,13 @@
 
 """Logging handler for printing formatted structured logs to standard output.
 """
-
+import json
 import logging.handlers
 
 from google.cloud.logging_v2.handlers.handlers import CloudLoggingFilter
 
 GCP_FORMAT = (
-    '{"message": "%(_msg_str)s", '
+    '{"message": %(_formatted_msg)s, '
     '"severity": "%(levelname)s", '
     '"logging.googleapis.com/labels": %(_labels_str)s, '
     '"logging.googleapis.com/trace": "%(_trace_str)s", '
@@ -50,7 +50,7 @@ def __init__(self, *, labels=None, stream=None, project_id=None):
         self.addFilter(log_filter)
 
         # make logs appear in GCP structured logging format
-        self.formatter = logging.Formatter(GCP_FORMAT)
+        self._gcp_formatter = logging.Formatter(GCP_FORMAT)
 
     def format(self, record):
         """Format the message into structured log JSON.
@@ -59,6 +59,12 @@ def format(self, record):
         Returns:
             str: A JSON string formatted for GKE fluentd.
         """
-
-        payload = self.formatter.format(record)
-        return payload
+        # let other formatters alter the message
+        super_payload = None
+        if record.msg:
+            super_payload = super(StructuredLogHandler, self).format(record)
+        # properly break any formatting in string to make it json safe
+        record._formatted_msg = json.dumps(super_payload or "")
+        # convert to GCP structred logging format
+        gcp_payload = self._gcp_formatter.format(record)
+        return gcp_payload

packages/google-cloud-logging/tests/environment/deployable/nodejs/Dockerfile

@@ -19,13 +19,14 @@ FROM node:12-slim
 # Create and change to the app directory.
 WORKDIR /usr/src/app
 
-# Copy test script and dependencies to the container image.
+# Copy test script and local dependencies to the container image.
 COPY package*.json ./
 COPY app.js ./
 COPY tests.js ./
+# Assumption: local file is already built
 COPY nodejs-logging ./nodejs-logging
 
-# Install dependencies.
+# Install test app's dependencies.
 RUN npm install --production
 
 # Environment variable denoting whether to run an app server

packages/google-cloud-logging/tests/environment/deployable/nodejs/app.js

@@ -14,9 +14,30 @@
 
 var tests = require('./tests.js');
 
+/** ****************** GAE, GKE, GCE ******************
+ * Enable app subscriber for all environments, except GCR, GCF.
+ */
+async function enableSubscriber() {
+  if (process.env.ENABLE_SUBSCRIBER) {
+    const gcpMetadata = require('gcp-metadata');
+    const projectId = await gcpMetadata.project('project-id');
+    const topicId = process.env.PUBSUB_TOPIC || 'logging-test';
+    const subscriptionId = `${topicId}-subscriber`;
+    const topicName = `projects/${projectId}/topics/${topicId}`;
+    const subscriptionName = `projects/${projectId}/subscriptions/${subscriptionId}`
+
+    const {PubSub} = require('@google-cloud/pubsub');
+    const pubSubClient = new PubSub();
+    // Creates a new subscription
+    pubSubClient.topic(topicName).createSubscription(subscriptionName);
+    listenForMessages(pubSubClient, subscriptionName).catch(console.error);
+  }
+}
+enableSubscriber().catch(console.error);
+
 /**
- * Only triggers for GCP services that require a running app server.
- * For instance, Cloud Functions does not execute this block.
+ * ****************** GCR, GKE, GCE ******************
+ * For GCP services that require a running app server, except GAE and GCF.
  * RUNSERVER env var is set in the Dockerfile.
  */
 if (process.env.RUNSERVER) {
@@ -29,25 +50,27 @@ if (process.env.RUNSERVER) {
   /**
    * Cloud Run to be triggered by Pub/Sub.
    */
-  app.post('/', (req, res) => {
-    if (!req.body) {
-      const msg = 'no Pub/Sub message received';
-      console.error(`error: ${msg}`);
-      res.status(400).send(`Bad Request: ${msg}`);
-      return;
-    }
-    if (!req.body.message) {
-      const msg = 'invalid Pub/Sub message format';
-      console.error(`error: ${msg}`);
-      res.status(400).send(`Bad Request: ${msg}`);
-      return;
-    }
-
-    const message = req.body.message;
-    triggerTest(message);
+  if (process.env.K_CONFIGURATION) {
+    app.post('/', (req, res) => {
+      if (!req.body) {
+        const msg = 'no Pub/Sub message received';
+        console.error(`error: ${msg}`);
+        res.status(400).send(`Bad Request: ${msg}`);
+        return;
+      }
+      if (!req.body.message) {
+        const msg = 'invalid Pub/Sub message format';
+        console.error(`error: ${msg}`);
+        res.status(400).send(`Bad Request: ${msg}`);
+        return;
+      }
 
-    res.status(204).send();
-  });
+      const message = req.body.message;
+      triggerTest(message);
+
+      res.status(204).send();
+    });
+  };
 
   // Start app server
   const PORT = process.env.PORT || 8080;
@@ -68,6 +91,31 @@ exports.pubsubFunction = (message, context) => {
   triggerTest(message);
 };
 
+/**
+ * ****************** GAE, GKE, GCE ******************
+ * Asynchronously listens for pubsub messages until a TIMEOUT is reached.
+ * @param pubSubClient
+ * @param subscriptionName
+ */
+async function listenForMessages(pubSubClient, subscriptionName) {
+  // References an existing subscription
+  const subscription = pubSubClient.subscription(subscriptionName);
+
+  // Handles incoming messages and triggers tests.
+  const messageHandler = message => {
+    triggerTest(message);
+    // "Ack" (acknowledge receipt of) the message
+    message.ack();
+  };
+
+  // Listen for new messages until timeout is hit or test is done.
+  subscription.on('message', messageHandler);
+
+  setTimeout(() => {
+    subscription.removeListener('message', messageHandler);
+  }, 600000); // max 10 minutes timeout
+}
+
 function triggerTest(message) {
   const testName = message.data
       ? Buffer.from(message.data, 'base64').toString()

packages/google-cloud-logging/tests/environment/deployable/nodejs/package.json

@@ -5,9 +5,11 @@
   "description": "",
   "main": "app.js",
   "dependencies": {
+    "@google-cloud/logging": "file:nodejs-logging",
+    "@google-cloud/pubsub": "^2.12.0",
     "body-parser": "^1.19.0",
     "express": "^4.17.1",
-    "@google-cloud/logging": "file:nodejs-logging"
+    "gcp-metadata": "^4.2.1"
   },
   "devDependencies": {},
   "engines": {

packages/google-cloud-logging/tests/environment/deployable/python/snippets.py

@@ -100,6 +100,21 @@ def pylogging(log_text="pylogging", severity="WARNING", **kwargs):
     else:
         logging.critical(log_text, extra=kwargs)
 
+def pylogging_multiline(log_text="pylogging", second_line="line 2", **kwargs):
+    logging.error(f"{log_text}\n{second_line}")
+
+def pylogging_complex_chars(**kwargs):
+    logging.error('}"{!@[')
+
+def pylogging_with_formatter(log_text="pylogging", format_str="%(name)s :: %(levelname)s :: %(message)s", **kwargs):
+    root_logger = logging.getLogger()
+    handler = root_logger.handlers[0]
+    handler.setFormatter(logging.Formatter(fmt=format_str))
+    logging.error(log_text)
+    handler.setFormatter(None)
+
+def pylogging_with_arg(log_text="my_arg", **kwargs):
+    logging.error("Arg: %s", log_text)
 
 def pylogging_flask(
     log_text="pylogging_flask",

packages/google-cloud-logging/tests/environment/envctl/env_scripts/go/kubernetes.sh

@@ -82,9 +82,6 @@ build_go_container(){
 }
 
 deploy() {
-#   local SCRIPT="${1:-router.py}"
-#   TODO: double check this doesn't print impt otkens/pws
-  set -x
   attach_or_create_gke_cluster
   build_go_container
   cat <<EOF > $TMP_DIR/gke.yaml

packages/google-cloud-logging/tests/environment/envctl/env_scripts/nodejs/appengine_standard.sh

@@ -0,0 +1,89 @@
+#!/bin/bash
+# Copyright 2021 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+set -e # exit on any failure
+set -o pipefail # any step in pipe caused failure
+set -u # undefined variables cause exit
+
+SERVICE_NAME="log-node-gae-$(echo $ENVCTL_ID | head -c 8)"
+LIBRARY_NAME="nodejs-logging"
+
+destroy() {
+  set +e
+  # delete pubsub resources
+  gcloud pubsub topics delete $SERVICE_NAME -q 2> /dev/null
+  gcloud pubsub subscriptions delete $SERVICE_NAME-subscriber -q 2> /dev/null
+  # delete service
+  gcloud app services delete $SERVICE_NAME -q 2> /dev/null
+  set -e
+}
+
+verify() {
+  set +e
+  gcloud app services describe $SERVICE_NAME -q > /dev/null 2> /dev/null
+  if [[ $? == 0 ]]; then
+     echo "TRUE"
+     exit 0
+   else
+     echo "FALSE"
+     exit 1
+  fi
+  set -e
+}
+
+deploy() {
+  # create pub/sub topic
+  set +e
+  gcloud pubsub topics create $SERVICE_NAME 2>/dev/null
+  set -ex
+
+  # copy over local version of the library
+  pushd $SUPERREPO_ROOT
+    tar -cvf $TMP_DIR/lib.tar --exclude node_modules --exclude env-tests-logging --exclude test --exclude system-test --exclude .nox --exclude samples --exclude docs .
+  popd
+  mkdir -p $TMP_DIR/$LIBRARY_NAME
+  tar -xvf $TMP_DIR/lib.tar --directory $TMP_DIR/$LIBRARY_NAME
+
+  # Copy over test code and Node dependencies
+  cp $REPO_ROOT/deployable/nodejs/package.json $TMP_DIR
+  cp $REPO_ROOT/deployable/nodejs/*.js $TMP_DIR
+
+  # manual_scaling allows 1 instance to continuously run regardless of the load level.
+  cat <<EOF > $TMP_DIR/app.yaml
+    runtime: nodejs12
+    service: $SERVICE_NAME
+    manual_scaling:
+      instances: 1
+    env_variables:
+      ENABLE_SUBSCRIBER: "true"
+      PUBSUB_TOPIC: $SERVICE_NAME
+EOF
+  # deploy
+  pushd $TMP_DIR
+    gcloud app deploy -q
+  popd
+  # wait for the pub/sub subscriber to start
+  NUM_SUBSCRIBERS=0
+  TRIES=0
+  while [[ "${NUM_SUBSCRIBERS}" -lt 1 && "${TRIES}" -lt 10 ]]; do
+    sleep 30
+    NUM_SUBSCRIBERS=$(gcloud pubsub topics list-subscriptions $SERVICE_NAME 2> /dev/null | wc -l)
+    TRIES=$((TRIES + 1))
+  done
+}
+
+filter-string() {
+  echo "resource.type=\"gae_app\" AND resource.labels.module_id=\"$SERVICE_NAME\""
+}

packages/google-cloud-logging/tests/environment/envctl/env_scripts/nodejs/cloudrun.sh

@@ -87,7 +87,6 @@ build_node_container() {
 }
 
 deploy() {
-  set -x
   build_node_container
   gcloud config set run/platform managed
   gcloud config set run/region us-west1