add a per-request GUID and log as it travels through background services

see: https://github.com/ansible/awx/issues/9329

awx/main/dispatch/periodic.py

@@ -6,6 +6,7 @@ from multiprocessing import Process
 from django.conf import settings
 from django.db import connections
 from schedule import Scheduler
+from django_guid.middleware import GuidMiddleware
 
 from awx.main.dispatch.worker import TaskWorker
 
@@ -35,6 +36,7 @@ class Scheduler(Scheduler):
                         # If the database connection has a hiccup, re-establish a new
                         # connection
                         conn.close_if_unusable_or_obsolete()
+                    GuidMiddleware.set_guid(GuidMiddleware._generate_guid())
                     self.run_pending()
                 except Exception:
                     logger.exception(

awx/main/dispatch/pool.py

@@ -16,6 +16,7 @@ from queue import Full as QueueFull, Empty as QueueEmpty
 from django.conf import settings
 from django.db import connection as django_connection, connections
 from django.core.cache import cache as django_cache
+from django_guid.middleware import GuidMiddleware
 from jinja2 import Template
 import psutil
 
@@ -445,6 +446,8 @@ class AutoscalePool(WorkerPool):
             return super(AutoscalePool, self).up()
 
     def write(self, preferred_queue, body):
+        if 'guid' in body:
+            GuidMiddleware.set_guid(body['guid'])
         try:
             # when the cluster heartbeat occurs, clean up internally
             if isinstance(body, dict) and 'cluster_node_heartbeat' in body['task']:

awx/main/dispatch/publish.py

@@ -5,6 +5,7 @@ import json
 from uuid import uuid4
 
 from django.conf import settings
+from django_guid.middleware import GuidMiddleware
 
 from . import pg_bus_conn
 
@@ -83,6 +84,9 @@ class task:
                     'kwargs': kwargs,
                     'task': cls.name
                 }
+                guid = GuidMiddleware.get_guid()
+                if guid:
+                    obj['guid'] = guid
                 obj.update(**kw)
                 if callable(queue):
                     queue = queue()

awx/main/dispatch/worker/callback.py

@@ -9,6 +9,7 @@ from django.conf import settings
 from django.utils.timezone import now as tz_now
 from django.db import DatabaseError, OperationalError, connection as django_connection
 from django.db.utils import InterfaceError, InternalError
+from django_guid.middleware import GuidMiddleware
 
 import psutil
 
@@ -152,6 +153,8 @@ class CallbackBrokerWorker(BaseWorker):
 
                 if body.get('event') == 'EOF':
                     try:
+                        if 'guid' in body:
+                            GuidMiddleware.set_guid(body['guid'])
                         final_counter = body.get('final_counter', 0)
                         logger.info('Event processing is finished for Job {}, sending notifications'.format(job_identifier))
                         # EOF events are sent when stdout for the running task is
@@ -176,6 +179,8 @@ class CallbackBrokerWorker(BaseWorker):
                             handle_success_and_failure_notifications.apply_async([uj.id])
                     except Exception:
                         logger.exception('Worker failed to emit notifications: Job {}'.format(job_identifier))
+                    finally:
+                        GuidMiddleware.set_guid('')
                     return
 
                 event = cls.create_from_data(**body)

awx/main/dispatch/worker/task.py

@@ -6,6 +6,8 @@ import traceback
 
 from kubernetes.config import kube_config
 
+from django_guid.middleware import GuidMiddleware
+
 from awx.main.tasks import dispatch_startup, inform_cluster_of_shutdown
 
 from .base import BaseWorker
@@ -52,6 +54,8 @@ class TaskWorker(BaseWorker):
         uuid = body.get('uuid', '<unknown>')
         args = body.get('args', [])
         kwargs = body.get('kwargs', {})
+        if 'guid' in body:
+            GuidMiddleware.set_guid(body.pop('guid'))
         _call = TaskWorker.resolve_callable(task)
         if inspect.isclass(_call):
             # the callable is a class, e.g., RunJob; instantiate and

awx/main/tasks.py

@@ -34,6 +34,7 @@ from django.contrib.auth.models import User
 from django.utils.translation import ugettext_lazy as _, gettext_noop
 from django.core.cache import cache
 from django.core.exceptions import ObjectDoesNotExist
+from django_guid.middleware import GuidMiddleware
 
 # Kubernetes
 from kubernetes.client.rest import ApiException
@@ -839,6 +840,7 @@ class BaseTask(object):
         self.cleanup_paths = []
         self.parent_workflow_job_id = None
         self.host_map = {}
+        self.guid = GuidMiddleware.get_guid()
 
     def update_model(self, pk, _attempt=0, **updates):
         """Reload the model instance from the database and update the
@@ -1274,6 +1276,9 @@ class BaseTask(object):
             except json.JSONDecodeError:
                 pass
 
+        if 'event_data' in event_data:
+            event_data['event_data']['guid'] = self.guid
+
         event_data.setdefault(self.event_data_key, self.instance.id)
         self.dispatcher.dispatch(event_data)
         self.event_ct += 1
@@ -1310,6 +1315,7 @@ class BaseTask(object):
         event_data = {
             'event': 'EOF',
             'final_counter': self.event_ct,
+            'guid': self.guid,
         }
         event_data.setdefault(self.event_data_key, self.instance.id)
         self.dispatcher.dispatch(event_data)

awx/main/tests/unit/test_tasks.py

@@ -527,7 +527,7 @@ class TestGenericRun():
         task.instance = Job(pk=1, id=1)
         task.event_ct = 17
         task.finished_callback(None)
-        task.dispatcher.dispatch.assert_called_with({'event': 'EOF', 'final_counter': 17, 'job_id': 1})
+        task.dispatcher.dispatch.assert_called_with({'event': 'EOF', 'final_counter': 17, 'job_id': 1, 'guid': None})
 
     def test_save_job_metadata(self, job, update_model_wrapper):
         class MockMe():

awx/main/utils/filters.py

@@ -15,6 +15,10 @@ from django.apps import apps
 from django.db import models
 from django.conf import settings
 
+from django_guid.log_filters import CorrelationId
+from django_guid.middleware import GuidMiddleware
+
+from awx import MODE
 from awx.main.constants import LOGGER_BLOCKLIST
 from awx.main.utils.common import get_search_fields
 
@@ -364,3 +368,14 @@ class SmartFilter(object):
             return res[0].result
 
         raise RuntimeError("Parsing the filter_string %s went terribly wrong" % filter_string)
+
+
+
+class DefaultCorrelationId(CorrelationId):
+
+    def filter(self, record):
+        guid = GuidMiddleware.get_guid() or '-'
+        if MODE == 'development':
+            guid = guid[:8]
+        record.guid = guid
+        return True

awx/main/utils/formatters.py

@@ -154,6 +154,9 @@ class LogstashFormatter(LogstashFormatterBase):
 
         if kind == 'job_events' and raw_data.get('python_objects', {}).get('job_event'):
             job_event = raw_data['python_objects']['job_event']
+            guid = job_event.event_data.pop('guid', None)
+            if guid:
+                data_for_log['guid'] = guid
             for field_object in job_event._meta.fields:
 
                 if not field_object.__class__ or not field_object.__class__.__name__:

awx/settings/defaults.py

@@ -285,6 +285,7 @@ INSTALLED_APPS = [
     'polymorphic',
     'taggit',
     'social_django',
+    'django_guid',
     'corsheaders',
     'awx.conf',
     'awx.main',
@@ -828,11 +829,14 @@ LOGGING = {
         },
         'dynamic_level_filter': {
             '()': 'awx.main.utils.filters.DynamicLevelFilter'
-        }
+        },
+        'guid': {
+            '()': 'awx.main.utils.filters.DefaultCorrelationId'
+        },
     },
     'formatters': {
         'simple': {
-            'format': '%(asctime)s %(levelname)-8s %(name)s %(message)s',
+            'format': '%(asctime)s %(levelname)-8s [%(guid)s] %(name)s %(message)s',
         },
         'json': {
             '()': 'awx.main.utils.formatters.LogstashFormatter'
@@ -842,7 +846,7 @@ LOGGING = {
             'format': '%(relativeSeconds)9.3f %(levelname)-8s %(message)s'
         },
         'dispatcher': {
-            'format': '%(asctime)s %(levelname)-8s %(name)s PID:%(process)d %(message)s',
+            'format': '%(asctime)s %(levelname)-8s [%(guid)s] %(name)s PID:%(process)d %(message)s',
         },
         'job_lifecycle': {
             '()': 'awx.main.utils.formatters.JobLifeCycleFormatter',
@@ -852,7 +856,7 @@ LOGGING = {
         'console': {
             '()': 'logging.StreamHandler',
             'level': 'DEBUG',
-            'filters': ['require_debug_true_or_test'],
+            'filters': ['require_debug_true_or_test', 'guid'],
             'formatter': 'simple',
         },
         'null': {
@@ -872,33 +876,33 @@ LOGGING = {
             'class': 'awx.main.utils.handlers.RSysLogHandler',
             'formatter': 'json',
             'address': '/var/run/awx-rsyslog/rsyslog.sock',
-            'filters': ['external_log_enabled', 'dynamic_level_filter'],
+            'filters': ['external_log_enabled', 'dynamic_level_filter', 'guid'],
         },
         'tower_warnings': {
             # don't define a level here, it's set by settings.LOG_AGGREGATOR_LEVEL
             'class': 'logging.handlers.WatchedFileHandler',
-            'filters': ['require_debug_false', 'dynamic_level_filter'],
+            'filters': ['require_debug_false', 'dynamic_level_filter', 'guid'],
             'filename': os.path.join(LOG_ROOT, 'tower.log'),
             'formatter':'simple',
         },
         'callback_receiver': {
             # don't define a level here, it's set by settings.LOG_AGGREGATOR_LEVEL
             'class': 'logging.handlers.WatchedFileHandler',
-            'filters': ['require_debug_false', 'dynamic_level_filter'],
+            'filters': ['require_debug_false', 'dynamic_level_filter', 'guid'],
             'filename': os.path.join(LOG_ROOT, 'callback_receiver.log'),
             'formatter':'simple',
         },
         'dispatcher': {
             # don't define a level here, it's set by settings.LOG_AGGREGATOR_LEVEL
             'class': 'logging.handlers.WatchedFileHandler',
-            'filters': ['require_debug_false', 'dynamic_level_filter'],
+            'filters': ['require_debug_false', 'dynamic_level_filter', 'guid'],
             'filename': os.path.join(LOG_ROOT, 'dispatcher.log'),
             'formatter':'dispatcher',
         },
         'wsbroadcast': {
             # don't define a level here, it's set by settings.LOG_AGGREGATOR_LEVEL
             'class': 'logging.handlers.WatchedFileHandler',
-            'filters': ['require_debug_false', 'dynamic_level_filter'],
+            'filters': ['require_debug_false', 'dynamic_level_filter', 'guid'],
             'filename': os.path.join(LOG_ROOT, 'wsbroadcast.log'),
             'formatter':'simple',
         },
@@ -914,7 +918,7 @@ LOGGING = {
         'task_system': {
             # don't define a level here, it's set by settings.LOG_AGGREGATOR_LEVEL
             'class': 'logging.handlers.WatchedFileHandler',
-            'filters': ['require_debug_false', 'dynamic_level_filter'],
+            'filters': ['require_debug_false', 'dynamic_level_filter', 'guid'],
             'filename': os.path.join(LOG_ROOT, 'task_system.log'),
             'formatter':'simple',
         },
@@ -1094,6 +1098,7 @@ AWX_CALLBACK_PROFILE = False
 AWX_CLEANUP_PATHS = True
 
 MIDDLEWARE = [
+    'django_guid.middleware.GuidMiddleware',
     'awx.main.middleware.TimingMiddleware',
     'django.contrib.sessions.middleware.SessionMiddleware',
     'awx.main.middleware.MigrationRanCheckMiddleware',
@@ -1134,3 +1139,7 @@ BROADCAST_WEBSOCKET_NEW_INSTANCE_POLL_RATE_SECONDS = 10
 
 # How often websocket process will generate stats
 BROADCAST_WEBSOCKET_STATS_POLL_RATE_SECONDS = 5
+
+DJANGO_GUID = {
+    'GUID_HEADER_NAME': 'X-API-Request-Id',
+}

docs/licenses/django-guid.txt

@@ -0,0 +1,27 @@
+Copyright (c) 2019 Jonas Krüger Svensson
+All rights reserved.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are met:
+
+    * Redistributions of source code must retain the above copyright
+    notice, this list of conditions and the following disclaimer.
+
+    * Redistributions in binary form must reproduce the above copyright
+    notice, this list of conditions and the following disclaimer in the
+    documentation and/or other materials provided with the distribution.
+
+    * Neither the name of the authors nor the
+    names of its contributors may be used to endorse or promote products
+    derived from this software without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
+ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDERS BE LIABLE FOR ANY
+DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
+(INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
+ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

requirements/requirements.in

@@ -13,6 +13,7 @@ django-auth-ldap
 django-cors-headers
 django-crum
 django-extensions>=2.2.9  # https://github.com/ansible/awx/pull/6441
+django-guid==2.2.0  # pinned to match Django 2.2
 django-jsonfield==1.2.0  # see UPGRADE BLOCKERs
 django-oauth-toolkit==1.1.3  # see UPGRADE BLOCKERs
 django-polymorphic

requirements/requirements.txt

@@ -19,7 +19,7 @@ channels-redis==3.1.0     # via -r /awx_devel/requirements/requirements.in
 channels==2.4.0           # via -r /awx_devel/requirements/requirements.in, channels-redis
 chardet==3.0.4            # via aiohttp, requests
 constantly==15.1.0        # via twisted
-cryptography==2.9.2       # via adal, autobahn, azure-keyvault, pyopenssl, service-identity, social-auth-core
+cryptography==2.9.2       # via -r /awx_devel/requirements/requirements.in, adal, autobahn, azure-keyvault, pyopenssl, service-identity, social-auth-core
 daphne==2.4.1             # via -r /awx_devel/requirements/requirements.in, channels
 defusedxml==0.6.0         # via python3-openid, python3-saml, social-auth-core
 dictdiffer==0.8.1         # via openshift
@@ -27,6 +27,7 @@ django-auth-ldap==2.1.0   # via -r /awx_devel/requirements/requirements.in
 django-cors-headers==3.2.1  # via -r /awx_devel/requirements/requirements.in
 django-crum==0.7.5        # via -r /awx_devel/requirements/requirements.in
 django-extensions==2.2.9  # via -r /awx_devel/requirements/requirements.in
+django-guid==2.2.0        # via -r /awx_devel/requirements/requirements.in
 django-jsonfield==1.2.0   # via -r /awx_devel/requirements/requirements.in
 django-oauth-toolkit==1.1.3  # via -r /awx_devel/requirements/requirements.in
 django-pglocks==1.0.4     # via -r /awx_devel/requirements/requirements.in
@@ -37,7 +38,7 @@ django-redis==4.5.0       # via -r /awx_devel/requirements/requirements.in
 django-solo==1.1.3        # via -r /awx_devel/requirements/requirements.in
 django-split-settings==1.0.0  # via -r /awx_devel/requirements/requirements.in
 django-taggit==1.2.0      # via -r /awx_devel/requirements/requirements.in
-django==2.2.16            # via -r /awx_devel/requirements/requirements.in, channels, django-auth-ldap, django-cors-headers, django-crum, django-jsonfield, django-oauth-toolkit, django-polymorphic, django-taggit, djangorestframework
+django==2.2.16            # via -r /awx_devel/requirements/requirements.in, channels, django-auth-ldap, django-cors-headers, django-crum, django-guid, django-jsonfield, django-oauth-toolkit, django-polymorphic, django-taggit, djangorestframework
 djangorestframework-yaml==1.0.3  # via -r /awx_devel/requirements/requirements.in
 djangorestframework==3.12.1  # via -r /awx_devel/requirements/requirements.in
 docutils==0.16            # via python-daemon