add debug views for task manager(s)

implement https://github.com/ansible/awx/issues/12446
in development environment, enable set of views that run
the task manager(s).

Also introduce a setting that disables any calls to schedule()
that do not originate from the debug views when in the development
environment. With guards around both if we are in the development
environment and the setting, I think we're pretty safe this won't get
triggered unintentionally.

awx/api/urls/debug.py

@@ -0,0 +1,17 @@
+from django.urls import re_path
+
+from awx.api.views.debug import (
+    DebugRootView,
+    TaskManagerDebugView,
+    DependencyManagerDebugView,
+    WorkflowManagerDebugView,
+)
+
+urls = [
+    re_path(r'^$', DebugRootView.as_view(), name='debug'),
+    re_path(r'^task_manager/$', TaskManagerDebugView.as_view(), name='task_manager'),
+    re_path(r'^dependency_manager/$', DependencyManagerDebugView.as_view(), name='dependency_manager'),
+    re_path(r'^workflow_manager/$', WorkflowManagerDebugView.as_view(), name='workflow_manager'),
+]
+
+__all__ = ['urls']

awx/api/urls/urls.py

@@ -149,3 +149,7 @@ if settings.SETTINGS_MODULE == 'awx.settings.development':
     from awx.api.swagger import SwaggerSchemaView
 
     urlpatterns += [re_path(r'^swagger/$', SwaggerSchemaView.as_view(), name='swagger_view')]
+
+    from awx.api.urls.debug import urls as debug_urls
+
+    urlpatterns += [re_path(r'^debug/', include(debug_urls))]

awx/api/views/debug.py

@@ -0,0 +1,68 @@
+from collections import OrderedDict
+
+from django.conf import settings
+
+from rest_framework.permissions import AllowAny
+from rest_framework.response import Response
+from rest_framework.views import APIView
+
+from awx.main.scheduler import TaskManager, DependencyManager, WorkflowManager
+
+
+class TaskManagerDebugView(APIView):
+    _ignore_model_permissions = True
+    exclude_from_schema = True
+    permission_classes = [AllowAny]
+    prefix = 'Task'
+
+    def get(self, request):
+        TaskManager().schedule(debug=True)
+        if not settings.AWX_DISABLE_TASK_MANAGERS:
+            msg = f"Running {self.prefix} manager. To disable other triggers to the {self.prefix} manager, set AWX_DISABLE_TASK_MANAGERS to True"
+        else:
+            msg = f"AWX_DISABLE_TASK_MANAGERS is True, this view is the only way to trigger the {self.prefix} manager"
+        return Response(msg)
+
+
+class DependencyManagerDebugView(APIView):
+    _ignore_model_permissions = True
+    exclude_from_schema = True
+    permission_classes = [AllowAny]
+    prefix = 'Dependency'
+
+    def get(self, request):
+        DependencyManager().schedule(debug=True)
+        if not settings.AWX_DISABLE_TASK_MANAGERS:
+            msg = f"Running {self.prefix} manager. To disable other triggers to the {self.prefix} manager, set AWX_DISABLE_TASK_MANAGERS to True"
+        else:
+            msg = f"AWX_DISABLE_TASK_MANAGERS is True, this view is the only way to trigger the {self.prefix} manager"
+        return Response(msg)
+
+
+class WorkflowManagerDebugView(APIView):
+    _ignore_model_permissions = True
+    exclude_from_schema = True
+    permission_classes = [AllowAny]
+    prefix = 'Workflow'
+
+    def get(self, request):
+        WorkflowManager().schedule(debug=True)
+        if not settings.AWX_DISABLE_TASK_MANAGERS:
+            msg = f"Running {self.prefix} manager. To disable other triggers to the {self.prefix} manager, set AWX_DISABLE_TASK_MANAGERS to True"
+        else:
+            msg = f"AWX_DISABLE_TASK_MANAGERS is True, this view is the only way to trigger the {self.prefix} manager"
+        return Response(msg)
+
+
+class DebugRootView(APIView):
+    _ignore_model_permissions = True
+    exclude_from_schema = True
+    permission_classes = [AllowAny]
+
+    def get(self, request, format=None):
+        '''List of available debug urls'''
+        data = OrderedDict()
+        data['task_manager'] = '/api/debug/task_manager/'
+        data['dependency_manager'] = '/api/debug/dependency_manager/'
+        data['workflow_manager'] = '/api/debug/workflow_manager/'
+        return Response(data)

awx/main/scheduler/task_manager.py

@@ -109,7 +109,12 @@ class TaskBase:
         self.record_aggregate_metrics()
         sys.exit(1)
 
-    def schedule(self):
+    def schedule(self, debug=False):
+
+        if settings.SETTINGS_MODULE == 'awx.settings.development' and settings.AWX_DISABLE_TASK_MANAGERS and not debug:
+            logger.debug(f"Not running {self.prefix} scheduler, AWX_DISABLE_TASK_MANAGERS is True. Trigger with GET to /api/{self.prefix}_manager_debug/")
+            return
+
         # Lock
         with advisory_lock(f"{self.prefix}_lock", wait=False) as acquired:
             with transaction.atomic():

awx/settings/development.py

@@ -110,5 +110,12 @@ CLUSTER_HOST_ID = socket.gethostname()
 
 AWX_CALLBACK_PROFILE = True
 
+# ======================!!!!!!! FOR DEVELOPMENT ONLY !!!!!!!=================================
+# Disable normal scheduled/triggered task managers (DependencyManager, TaskManager, WorkflowManager).
+# Allows user to trigger task managers directly for debugging and profiling purposes.
+# Only works in combination with settings.SETTINGS_MODULE == 'awx.settings.development'
+AWX_DISABLE_TASK_MANAGERS = os.getenv('AWX_DISABLE_TASK_MANAGERS', False)
+# ======================!!!!!!! FOR DEVELOPMENT ONLY !!!!!!!=================================
+
 if 'sqlite3' not in DATABASES['default']['ENGINE']:  # noqa
     DATABASES['default'].setdefault('OPTIONS', dict()).setdefault('application_name', f'{CLUSTER_HOST_ID}-{os.getpid()}-{" ".join(sys.argv)}'[:63])  # noqa