Update publisher to allow to stream only without invoking tasks
@@ -24,9 +24,12 @@ class PublisherService(Service):
log_lines,
experiment_uuid,
experiment_name,
- job_uuid):
+ job_uuid,
+ send_task=True):
log_lines = to_list(log_lines)
self._logger.debug("Publishing log event for task: %s, %s", job_uuid, experiment_name)
+
+ if send_task:
celery_app.send_task(
LogsCeleryTasks.LOGS_HANDLE_EXPERIMENT_JOB,
kwargs={
@@ -52,7 +55,7 @@ class PublisherService(Service):
'log_lines': log_lines,
},
retry=True,
- routing_key='{}.{}.{}'.format(RoutingKeys.LOGS_SIDECARS_EXPERIMENTS,
+ routing_key='{}.{}.{}'.format(RoutingKeys.STREAM_LOGS_SIDECARS_EXPERIMENTS,
experiment_uuid,
job_uuid),
exchange=settings.INTERNAL_EXCHANGE,
@@ -82,10 +85,11 @@ class PublisherService(Service):
except (TimeoutError, AMQPError):
pass
- def publish_build_job_log(self, log_lines, job_uuid, job_name):
+ def publish_build_job_log(self, log_lines, job_uuid, job_name, send_task=True):
log_lines = to_list(log_lines)
self._logger.info("Publishing log event for task: %s", job_uuid)
+ if send_task:
celery_app.send_task(
LogsCeleryTasks.LOGS_HANDLE_BUILD_JOB,
kwargs={
@@ -95,12 +99,13 @@ class PublisherService(Service):
})
self._stream_job_log(job_uuid=job_uuid,
log_lines=log_lines,
- routing_key=RoutingKeys.LOGS_SIDECARS_BUILDS)
+ routing_key=RoutingKeys.STREAM_LOGS_SIDECARS_BUILDS)
- def publish_job_log(self, log_lines, job_uuid, job_name):
+ def publish_job_log(self, log_lines, job_uuid, job_name, send_task=True):
log_lines = to_list(log_lines)
self._logger.info("Publishing log event for task: %s", job_uuid)
+ if send_task:
celery_app.send_task(
LogsCeleryTasks.LOGS_HANDLE_JOB,
kwargs={
@@ -110,7 +115,7 @@ class PublisherService(Service):
})
self._stream_job_log(job_uuid=job_uuid,
log_lines=log_lines,
- routing_key=RoutingKeys.LOGS_SIDECARS_JOBS)
+ routing_key=RoutingKeys.STREAM_LOGS_SIDECARS_JOBS)
def setup(self):
import logging
