Cloud Run Jobs container exits before all the logging is complete

[ankitarya1019]

Environment details

• OS type and version: Cloud Run
• Python version: 3.10
• Cloud Run with google-cloud-logging==3.9.0

Steps to reproduce

https://stackoverflow.com/questions/77937865/cloud-run-jobs-container-exits-before-all-the-logging-is-complete?noredirect=1#comment137411974_77937865

I have been struggling with the above for a while, not sure if it is a cloud run issue or google cloud logging.

[gkevinzheng]

@ankitarya1019 Sorry for the late reply. How are you viewing the logs, with Cloud Run Jobs's built in web UI, or on Logs Explorer? There's a known issue where a missing label on Cloud Run Jobs environment makes the web UI miss all CloudLoggingHandler logs.

[gkevinzheng]

@ankitarya1019 Reading through your StackExchange post, I'm not sure what get_task_params and run_task do, but I ran this code on Cloud Run Jobs:

import logging
import os
import time
import sys

import logging.handlers

import google.cloud.logging

# Retrieve Job-defined env vars
TASK_INDEX = os.getenv("CLOUD_RUN_TASK_INDEX", 0)
TASK_ATTEMPT = os.getenv("CLOUD_RUN_TASK_ATTEMPT", 0)

def flush_logs():
    logger = logging.getLogger()
    handlers = logger.handlers[:]
    for handler in handlers:
        handler.flush()
        handler.close()
        logger.removeHandler(handler)

    logger.info("All logging handlers have been flushed and closed.")

    logging.shutdown()


def setup_logging():
    # Instantiates a client
    client = google.cloud.logging.Client()
    # Retrieves a Cloud Logging handler based on the environment
    # you're running in and integrates the handler with the
    # Python logging module. By default this captures all logs
    # at INFO level and higher
    client.setup_logging()

    if not os.path.exists(os.path.dirname('/var/log/foo')):
        os.makedirs(os.path.dirname('/var/log/foo'))

    log_formatter = logging.Formatter(
        "%(asctime)s %(levelname)s: %(message)s [in %(pathname)s:%(lineno)d]"
    )

    # Configure the file handler
    file_handler = logging.handlers.RotatingFileHandler(
        '/var/log/foo', maxBytes=1024 * 1024 * 10, backupCount=10
    )
    file_handler.setFormatter(log_formatter)
    file_handler.setLevel(logging.INFO)

    # Configure the stream handler
    stream_handler = logging.StreamHandler()
    stream_handler.setFormatter(log_formatter)
    stream_handler.setLevel(logging.INFO)

    # Set the log level and handlers for the logger
    logger = logging.getLogger()  # Define logger at the module level
    logger.setLevel(logging.INFO)
    logger.addHandler(file_handler)
    logger.addHandler(stream_handler)

    # Log the initialization
    logger.info("Logging is set up")


# Initialize logging with the desired configuration
setup_logging()

logger = logging.getLogger()

try:
    logger.info("TESTS")
    time.sleep(2) # Replacement for run_tasks
    logger.info("Tasks are DONE; Waiting for 5 seconds before exiting to allow logs to be flushed.")
    flush_logs()
    time.sleep(5)  # Wait 5 seconds before exiting to allow logs to be flushed
except Exception as err:
    message = f"Task #{TASK_INDEX}, Attempt #{TASK_ATTEMPT} failed: {str(err)}"
    logger.info(message)
    sys.exit(1)  # Non-zero exit code for failure

and I wasn't able to reproduce the issue on my side.