Catch publisher eventual errors
@@ -3,6 +3,7 @@ from __future__ import absolute_import, division, print_function
import logging
+from amqp import AMQPError
from django.conf import settings
from polyaxon.config_settings import RoutingKeys
@@ -41,6 +42,7 @@ def publish_log(log_line,
logger.info("Streaming new log event for experiment: {}".format(experiment_uuid))
with celery_app.producer_or_acquire(None) as producer:
+ try:
producer.publish(
{
'experiment_uuid': experiment_uuid,
@@ -55,3 +57,6 @@ def publish_log(log_line,
job_uuid),
exchange=settings.INTERNAL_EXCHANGE,
)
+ except (TimeoutError, AMQPError):
+ pass
+
