Fix issues with log sidecar publishing
@@ -5,15 +5,16 @@ import logging
import time
+from django.conf import settings
from polyaxon_schemas.experiment import JobLabelConfig
-from api.config_settings import CeleryPublishTask, RoutingKeys
+from api.config_settings import RoutingKeys
from api.celery_api import app as celery_app
from libs.redis_db import RedisToStream
from events.tasks import handle_events_job_logs
from spawner.utils.constants import PodLifeCycle
-logger = logging.getLogger('polyaxon.monitors.sidecar')
+logger = logging.getLogger('monitors.sidecar')
def run(k8s_manager, pod_id, experiment_uuid, job_uuid, container_job_name, persist):
@@ -22,20 +23,30 @@ def run(k8s_manager, pod_id, experiment_uuid, job_uuid, container_job_name, pers
container=container_job_name,
follow=True,
_preload_content=False)
+ P = None
for log_line in raw.stream():
- logger.info("Publishing event: {}".format(log_line))
+ log_line = log_line.decode('utf-8')
+ logger.info("Publishing log event for experiment: {}, {}".format(job_uuid, experiment_uuid))
handle_events_job_logs.delay(experiment_uuid=experiment_uuid,
job_uuid=job_uuid,
log_line=log_line,
persist=persist)
if (RedisToStream.is_monitored_job_logs(job_uuid) or
RedisToStream.is_monitored_experiment_logs(experiment_uuid)):
- celery_app.send_task(
- CeleryPublishTask.PUBLISH_LOGS_SIDECAR,
- kwargs={'experiment_uuid': experiment_uuid,
+ logger.info("Streaming new log event for experiment: {}".format(experiment_uuid))
+
+ with celery_app.producer_or_acquire(P) as P:
+ P.publish(
+ {
+ 'experiment_uuid': experiment_uuid,
'job_uuid': job_uuid,
- 'log_line': log_line},
- routing_key='{}.{}.{}'.format(RoutingKeys.LOGS_SIDECARS, experiment_uuid, job_uuid))
+ 'log_line': log_line
+ },
+ routing_key='{}.{}.{}'.format(RoutingKeys.LOGS_SIDECARS,
+ experiment_uuid,
+ job_uuid),
+ exchange=settings.INTERNAL_EXCHANGE,
+ )
def can_log(k8s_manager, pod_id, log_sleep_interval):
