prevent duplicate logs in cloudrun job (datacommonsorg#1478)

Author: rohitkumarbhagat

import-automation/executor/main.py

@@ -35,6 +35,7 @@
 sys.path.append(os.path.join(REPO_DIR, 'util'))
 
 from log_util import log_metric, configure_cloud_logging
+from cloudrun_util import running_on_cloudrun
 
 FLAGS = flags.FLAGS
 flags.DEFINE_string('import_name', '', 'Absoluate import name.')
@@ -101,7 +102,11 @@ def run_import_job(absolute_import_name: str, import_config: str):
 
 
 def main(_):
-    if FLAGS.enable_cloud_logging:
+    running_on_cloudrun_result = running_on_cloudrun()
+    if running_on_cloudrun_result:
+        logging.info("Running under Cloud Run detected.")
+
+    if FLAGS.enable_cloud_logging or running_on_cloudrun_result:
         configure_cloud_logging()
         logging.info("Google Cloud Logging configured.")
     return run_import_job(FLAGS.import_name, FLAGS.import_config)

tools/statvar_importer/stat_var_processor.py

@@ -66,6 +66,8 @@
 sys.path.append(os.path.join(_SCRIPT_DIR, 'schema'))
 
 import eval_functions
+from log_util import configure_cloud_logging
+from cloudrun_util import running_on_cloudrun
 from utils import (capitalize_first_char, is_place_dcid,
                    get_observation_date_format, get_observation_period_for_date,
                    pvs_has_any_prop, str_from_number, prepare_input_data)
@@ -2788,6 +2790,14 @@ def process(
 
 
 def main(_):
+    # Configure cloud logging if running on CloudRun
+    if running_on_cloudrun():
+        logging.info("Running under Cloud Run detected.")
+        configure_cloud_logging()
+        logging.info("Google Cloud Logging configured.")
+    else:
+        logging.info("Not running under Cloud Run")
+
     # uncomment to run pprof
     # start_pprof_server(port=8123)
 

util/cloudrun_util.py

@@ -0,0 +1,25 @@
+# Copyright 2023 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.
+"""Utility functions for Cloud Run environment."""
+
+import os
+
+
+def running_on_cloudrun() -> bool:
+    """Check if running on Cloud Run.
+    
+    Returns:
+        bool: True if running on Cloud Run services, False otherwise.
+    """
+    return bool(os.getenv('K_SERVICE'))

util/cloudrun_util_test.py

@@ -0,0 +1,42 @@
+# Copyright 2023 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.
+"""Tests for util.cloudrun_util module."""
+
+import os
+import unittest
+from unittest import mock
+
+from util import cloudrun_util
+
+
+class CloudRunUtilTest(unittest.TestCase):
+    """Tests for Cloud Run utility functions."""
+
+    def test_running_on_cloudrun_when_k_service_is_set(self):
+        """Test that running_on_cloudrun returns True when K_SERVICE is set."""
+        with mock.patch.dict(os.environ, {'K_SERVICE': 'my-service'}):
+            self.assertTrue(cloudrun_util.running_on_cloudrun())
+
+    def test_running_on_cloudrun_when_k_service_is_not_set(self):
+        """Test that running_on_cloudrun returns False when K_SERVICE is not set."""
+        with mock.patch.dict(os.environ, {}, clear=True):
+            # Ensure K_SERVICE is not in environment
+            if 'K_SERVICE' in os.environ:
+                del os.environ['K_SERVICE']
+            self.assertFalse(cloudrun_util.running_on_cloudrun())
+
+    def test_running_on_cloudrun_with_empty_k_service_value(self):
+        """Test that running_on_cloudrun returns False when K_SERVICE is empty."""
+        with mock.patch.dict(os.environ, {'K_SERVICE': ''}):
+            self.assertFalse(cloudrun_util.running_on_cloudrun())

util/log_util.py

@@ -14,7 +14,9 @@
 """Utility functions for logging."""
 
 import json
+import logging
 import google.cloud.logging
+from absl import logging as absl_logging
 
 
 def log_struct(level: str, message: str, labels: dict):
@@ -67,12 +69,17 @@ def log_metric(log_type: str, level: str, message: str, metric_labels: dict):
 def configure_cloud_logging():
     """Configure Google Cloud Logging handler for structured logging with proper severity.
     
-    Handles both standard Python logging and absl logging since absl uses Python's 
-    logging system internally. Maps log levels to GCP severity (INFO→INFO, ERROR→ERROR).
-    
-    Note: Prefer calling this after app.run() begins so that absl's logging handlers 
-    are already set up. Both handlers will coexist - ABSLHandler for console output 
-    and CloudLoggingHandler for structured logs with proper severity.
+    Removes ABSL handler if present to prevent log duplication between stderr and Cloud Logging.
+    This ensures logs appear only in Cloud Logging with proper severity levels.
     """
+    # Remove ABSL handler if present to prevent duplication
+    absl_handler = absl_logging.get_absl_handler()
+    if absl_handler in logging.root.handlers:
+        logging.root.handlers.remove(absl_handler)
+        logging.info(
+            "ABSL handler found and removed to prevent log duplication")
+    else:
+        logging.info("ABSL handler not found in root handlers")
+
     client = google.cloud.logging.Client()
     client.setup_logging()