freeze AWS glue job

[andreisorokin-astrazeneca]

I use WatchTower as workaround for inability to change CloudWatch group for glue jobs written in Python.
But when I add WatchTower handler the glue jobs after successfuly finishing job freeze.

In Spark worker logs I see nothing - just normal finishing and after that final message that the worker had been killed by driver.

In Spark Driver I see repeated attempts to send stdout to CloudWatch by internal AWS mechanism - do not see how is possibly connected with WatchTower:

INFO LogPusher: uploading /tmp/spark-event-logs/ to s3://****
INFO MultipartUploadOutputStream: close closed:false s3://***.inprogress

If I set use_queues=False it finishes normally and in logs I see
INFO MultipartUploadOutputStream: close closed:false s3://***/FilteredOut/run-1705485647329-part-r-00002.gz

[andreisorokin-astrazeneca]

... and in both cases logs were successfully and fully sent to CloudWatch by WatchTower

[andreisorokin-astrazeneca]

I found workaround - if I add that in the end of glue job

logger = logging.getLogger()
logging.shutdown()
for handler in logger.handlers:
    logger.removeHandler(handler)

it finishes normally

[dovy]

Can you provide the code you used and maybe the version of cloudwatch? I can't get it to run via glue properly.

[dovy]

@andreisorokin-astrazeneca

[tor-gu]

I encountered this issue as well, and I believe I understand what is happening.

The issue is triggered by a watchtower handler receiving a message for a new stream_name during shutdown. Since the stream is new, watchtower starts a new daemon thread. Since it happens during shutdown, this creates a deadlock -- the code hangs on thread.start here: https://github.com/kislyuk/watchtower/blob/main/watchtower/__init__.py#L462

The code already checks to see if it is shutting_down before it adds anything to the queue, but not before it starts the thread. This can be solved if the shutting_down check is done before starting the thread, like this:

if self.shutting_down:
    warnings.warn("Received message after logging system shutdown", WatchtowerWarning)
else:
    if stream_name not in self.queues:
        self.queues[stream_name] = queue.Queue()
        thread = threading.Thread(
            target=self._dequeue_batch,
            args=(
                self.queues[stream_name],
                stream_name,
                self.send_interval,
                self.max_batch_size,
                self.max_batch_count,
            ),
        )
        self.threads.append(thread)
        thread.daemon = True
        thread.start()
    self.queues[stream_name].put(cwl_message)

A scenario where this issue will cause a Glue job to hang is the following:

• The root logger level is set to INFO; and
• The watchtower handler is attached to the root logger; and
• The watchtower handler is initialized with a log_stream_name template that includes {logger_name}

For example:

logger = logging.getLogger() # Root logger
handler = ClouwWatchLogHandler(
    # Default log_stream_name includes {logger_name}
    log_group_name="/aws-glue/jobs/output"
)
logger.addHandler(handler)
logger.setLevel("INFO") 

In this scenario, during shutdown of the glue job, the py4j library that pyspark uses will attempt to log "Closing down clientserver connection" at the INFO level with logger name py4j.clientserver. This will be forwarded to watchtower, and will be the first py4j.clientserver message watchtower has seen, so watchtower will create a new queue and new thread (which hangs).

[kislyuk]

Thank you for the analysis @tor-gu, this sounds very plausible to me. Would you like to open a PR with the necessary changes?

[tor-gu]

Sure, I will do that.