Db partition analytics cmeyers2

awx/api/filters.py

@@ -133,7 +133,7 @@ class FieldLookupBackend(BaseFilterBackend):
     Filter using field lookups provided via query string parameters.
     """
 
-    RESERVED_NAMES = ('page', 'page_size', 'format', 'order', 'order_by', 'search', 'type', 'host_filter', 'count_disabled', 'no_truncate')
+    RESERVED_NAMES = ('page', 'page_size', 'format', 'order', 'order_by', 'search', 'type', 'host_filter', 'count_disabled', 'no_truncate', 'limit')
 
     SUPPORTED_LOOKUPS = (
         'exact',

awx/api/pagination.py

@@ -1,12 +1,16 @@
 # Copyright (c) 2015 Ansible, Inc.
 # All Rights Reserved.
 
+from collections import OrderedDict
+
 # Django REST Framework
 from django.conf import settings
 from django.core.paginator import Paginator as DjangoPaginator
 from rest_framework import pagination
 from rest_framework.response import Response
 from rest_framework.utils.urls import replace_query_param
+from rest_framework.settings import api_settings
+from django.utils.translation import gettext_lazy as _
 
 
 class DisabledPaginator(DjangoPaginator):
@@ -65,3 +69,65 @@ def get_paginated_response(self, data):
         if self.count_disabled:
             return Response({'results': data})
         return super(Pagination, self).get_paginated_response(data)
+
+
+class LimitPagination(pagination.BasePagination):
+    default_limit = api_settings.PAGE_SIZE
+    limit_query_param = 'limit'
+    limit_query_description = _('Number of results to return per page.')
+    max_page_size = settings.MAX_PAGE_SIZE
+
+    def paginate_queryset(self, queryset, request, view=None):
+        self.limit = self.get_limit(request)
+        self.request = request
+
+        return list(queryset[0 : self.limit])
+
+    def get_paginated_response(self, data):
+        return Response(OrderedDict([('results', data)]))
+
+    def get_paginated_response_schema(self, schema):
+        return {
+            'type': 'object',
+            'properties': {
+                'results': schema,
+            },
+        }
+
+    def get_limit(self, request):
+        try:
+            return pagination._positive_int(request.query_params[self.limit_query_param], strict=True)
+        except (KeyError, ValueError):
+            pass
+
+        return self.default_limit
+
+
+class UnifiedJobEventPagination(Pagination):
+    """
+    By default, use Pagination for all operations.
+    If `limit` query parameter specified use LimitPagination
+    """
+
+    def __init__(self, *args, **kwargs):
+        self.use_limit_paginator = False
+        self.limit_pagination = LimitPagination()
+        return super().__init__(*args, **kwargs)
+
+    def paginate_queryset(self, queryset, request, view=None):
+        if 'limit' in request.query_params:
+            self.use_limit_paginator = True
+
+        if self.use_limit_paginator:
+            return self.limit_pagination.paginate_queryset(queryset, request, view=view)
+        return super().paginate_queryset(queryset, request, view=view)
+
+    def get_paginated_response(self, data):
+        if self.use_limit_paginator:
+            return self.limit_pagination.get_paginated_response(data)
+        return super().get_paginated_response(data)
+
+    def get_paginated_response_schema(self, schema):
+        if self.use_limit_paginator:
+            return self.limit_pagination.get_paginated_response_schema(schema)
+        return super().get_paginated_response_schema(schema)

awx/api/serializers.py

@@ -3044,7 +3044,7 @@ def get_related(self, obj):
         res = super(JobSerializer, self).get_related(obj)
         res.update(
             dict(
-                job_events=self.reverse('api:job_job_events_list', kwargs={'pk': obj.pk}),
+                job_events=self.reverse('api:job_job_events_list', kwargs={'pk': obj.pk}),  # TODO: consider adding job_created
                 job_host_summaries=self.reverse('api:job_job_host_summaries_list', kwargs={'pk': obj.pk}),
                 activity_stream=self.reverse('api:job_activity_stream_list', kwargs={'pk': obj.pk}),
                 notifications=self.reverse('api:job_notifications_list', kwargs={'pk': obj.pk}),
@@ -3111,16 +3111,16 @@ class Meta:
         fields = ('*', 'host_status_counts', 'playbook_counts', 'custom_virtualenv')
 
     def get_playbook_counts(self, obj):
-        task_count = obj.job_events.filter(event='playbook_on_task_start').count()
-        play_count = obj.job_events.filter(event='playbook_on_play_start').count()
+        task_count = obj.get_event_queryset().filter(event='playbook_on_task_start').count()
+        play_count = obj.get_event_queryset().filter(event='playbook_on_play_start').count()
 
         data = {'play_count': play_count, 'task_count': task_count}
 
         return data
 
     def get_host_status_counts(self, obj):
         try:
-            counts = obj.job_events.only('event_data').get(event='playbook_on_stats').get_host_status_counts()
+            counts = obj.get_event_queryset().only('event_data').get(event='playbook_on_stats').get_host_status_counts()
         except JobEvent.DoesNotExist:
             counts = {}
 

awx/api/urls/job_event.py

@@ -3,14 +3,11 @@
 
 from django.conf.urls import url
 
-from awx.api.views import JobEventList, JobEventDetail, JobEventChildrenList, JobEventHostsList
-
+from awx.api.views import JobEventDetail, JobEventChildrenList
 
 urls = [
-    url(r'^$', JobEventList.as_view(), name='job_event_list'),
     url(r'^(?P<pk>[0-9]+)/$', JobEventDetail.as_view(), name='job_event_detail'),
     url(r'^(?P<pk>[0-9]+)/children/$', JobEventChildrenList.as_view(), name='job_event_children_list'),
-    url(r'^(?P<pk>[0-9]+)/hosts/$', JobEventHostsList.as_view(), name='job_event_hosts_list'),
 ]
 
 __all__ = ['urls']

awx/api/views/__init__.py

@@ -21,7 +21,7 @@
 from django.conf import settings
 from django.core.exceptions import FieldError, ObjectDoesNotExist
 from django.db.models import Q, Sum
-from django.db import IntegrityError, transaction, connection
+from django.db import IntegrityError, ProgrammingError, transaction, connection
 from django.shortcuts import get_object_or_404
 from django.utils.safestring import mark_safe
 from django.utils.timezone import now
@@ -172,11 +172,21 @@
     ApiV2AttachView,
 )
 from awx.api.views.webhooks import WebhookKeyView, GithubWebhookReceiver, GitlabWebhookReceiver  # noqa
+from awx.api.pagination import UnifiedJobEventPagination
 
 
 logger = logging.getLogger('awx.api.views')
 
 
+def unpartitioned_event_horizon(cls):
+    with connection.cursor() as cursor:
+        try:
+            cursor.execute(f'SELECT MAX(id) FROM _unpartitioned_{cls._meta.db_table}')
+            return cursor.fetchone()[0] or -1
+        except ProgrammingError:
+            return 0
+
+
 def api_exception_handler(exc, context):
     """
     Override default API exception handler to catch IntegrityError exceptions.
@@ -878,11 +888,17 @@ class ProjectUpdateEventsList(SubListAPIView):
     relationship = 'project_update_events'
     name = _('Project Update Events List')
     search_fields = ('stdout',)
+    pagination_class = UnifiedJobEventPagination
 
     def finalize_response(self, request, response, *args, **kwargs):
         response['X-UI-Max-Events'] = settings.MAX_UI_JOB_EVENTS
         return super(ProjectUpdateEventsList, self).finalize_response(request, response, *args, **kwargs)
 
+    def get_queryset(self):
+        pu = self.get_parent_object()
+        self.check_parent_access(pu)
+        return pu.get_event_queryset()
+
 
 class SystemJobEventsList(SubListAPIView):
 
@@ -892,11 +908,17 @@ class SystemJobEventsList(SubListAPIView):
     relationship = 'system_job_events'
     name = _('System Job Events List')
     search_fields = ('stdout',)
+    pagination_class = UnifiedJobEventPagination
 
     def finalize_response(self, request, response, *args, **kwargs):
         response['X-UI-Max-Events'] = settings.MAX_UI_JOB_EVENTS
         return super(SystemJobEventsList, self).finalize_response(request, response, *args, **kwargs)
 
+    def get_queryset(self):
+        job = self.get_parent_object()
+        self.check_parent_access(job)
+        return job.get_event_queryset()
+
 
 class ProjectUpdateCancel(RetrieveAPIView):
 
@@ -3602,7 +3624,7 @@ def post(self, request, *args, **kwargs):
                     status=status.HTTP_400_BAD_REQUEST,
                 )
             host_qs = obj.retry_qs(retry_hosts)
-            if not obj.job_events.filter(event='playbook_on_stats').exists():
+            if not obj.get_event_queryset().filter(event='playbook_on_stats').exists():
                 return Response(
                     {'hosts': _('Cannot retry on {status_value} hosts, playbook stats not available.').format(status_value=retry_hosts)},
                     status=status.HTTP_400_BAD_REQUEST,
@@ -3729,17 +3751,21 @@ class JobHostSummaryDetail(RetrieveAPIView):
     serializer_class = serializers.JobHostSummarySerializer
 
 
-class JobEventList(NoTruncateMixin, ListAPIView):
+class JobEventDetail(RetrieveAPIView):
 
-    model = models.JobEvent
     serializer_class = serializers.JobEventSerializer
-    search_fields = ('stdout',)
-
 
-class JobEventDetail(RetrieveAPIView):
+    @property
+    def is_partitioned(self):
+        if 'pk' not in self.kwargs:
+            return True
+        return int(self.kwargs['pk']) > unpartitioned_event_horizon(models.JobEvent)
 
-    model = models.JobEvent
-    serializer_class = serializers.JobEventSerializer
+    @property
+    def model(self):
+        if self.is_partitioned:
+            return models.JobEvent
+        return models.UnpartitionedJobEvent
 
     def get_serializer_context(self):
         context = super().get_serializer_context()
@@ -3749,33 +3775,31 @@ def get_serializer_context(self):
 
 class JobEventChildrenList(NoTruncateMixin, SubListAPIView):
 
-    model = models.JobEvent
     serializer_class = serializers.JobEventSerializer
-    parent_model = models.JobEvent
     relationship = 'children'
     name = _('Job Event Children List')
     search_fields = ('stdout',)
 
-    def get_queryset(self):
-        parent_event = self.get_parent_object()
-        self.check_parent_access(parent_event)
-        qs = self.request.user.get_queryset(self.model).filter(parent_uuid=parent_event.uuid)
-        return qs
-
+    @property
+    def is_partitioned(self):
+        if 'pk' not in self.kwargs:
+            return True
+        return int(self.kwargs['pk']) > unpartitioned_event_horizon(models.JobEvent)
 
-class JobEventHostsList(HostRelatedSearchMixin, SubListAPIView):
+    @property
+    def model(self):
+        if self.is_partitioned:
+            return models.JobEvent
+        return models.UnpartitionedJobEvent
 
-    model = models.Host
-    serializer_class = serializers.HostSerializer
-    parent_model = models.JobEvent
-    relationship = 'hosts'
-    name = _('Job Event Hosts List')
+    @property
+    def parent_model(self):
+        return self.model
 
     def get_queryset(self):
         parent_event = self.get_parent_object()
         self.check_parent_access(parent_event)
-        qs = self.request.user.get_queryset(self.model).filter(job_events_as_primary_host=parent_event)
-        return qs
+        return parent_event.job.get_event_queryset().filter(parent_uuid=parent_event.uuid)
 
 
 class BaseJobEventsList(NoTruncateMixin, SubListAPIView):
@@ -3811,12 +3835,12 @@ class GroupJobEventsList(BaseJobEventsList):
 class JobJobEventsList(BaseJobEventsList):
 
     parent_model = models.Job
+    pagination_class = UnifiedJobEventPagination
 
     def get_queryset(self):
         job = self.get_parent_object()
         self.check_parent_access(job)
-        qs = job.job_events.select_related('host').order_by('start_line')
-        return qs.all()
+        return job.get_event_queryset().select_related('host').order_by('start_line')
 
 
 class AdHocCommandList(ListCreateAPIView):
@@ -3974,6 +3998,11 @@ class AdHocCommandEventList(NoTruncateMixin, ListAPIView):
     serializer_class = serializers.AdHocCommandEventSerializer
     search_fields = ('stdout',)
 
+    def get_queryset(self):
+        adhoc = self.get_parent_object()
+        self.check_parent_access(adhoc)
+        return adhoc.get_event_queryset()
+
 
 class AdHocCommandEventDetail(RetrieveAPIView):
 
@@ -3994,12 +4023,21 @@ class BaseAdHocCommandEventsList(NoTruncateMixin, SubListAPIView):
     relationship = 'ad_hoc_command_events'
     name = _('Ad Hoc Command Events List')
     search_fields = ('stdout',)
+    pagination_class = UnifiedJobEventPagination
+
+    def get_queryset(self):
+        parent = self.get_parent_object()
+        self.check_parent_access(parent)
+        return parent.get_event_queryset()
 
 
 class HostAdHocCommandEventsList(BaseAdHocCommandEventsList):
 
     parent_model = models.Host
 
+    def get_queryset(self):
+        return super(BaseAdHocCommandEventsList, self).get_queryset()
+
 
 # class GroupJobEventsList(BaseJobEventsList):
 #    parent_model = Group

awx/api/views/inventory.py

@@ -38,6 +38,9 @@
 )
 from awx.api.views.mixin import RelatedJobsPreventDeleteMixin, ControlledByScmMixin
 
+from awx.api.pagination import UnifiedJobEventPagination
+
+
 logger = logging.getLogger('awx.api.views.organization')
 
 
@@ -49,6 +52,12 @@ class InventoryUpdateEventsList(SubListAPIView):
     relationship = 'inventory_update_events'
     name = _('Inventory Update Events List')
     search_fields = ('stdout',)
+    pagination_class = UnifiedJobEventPagination
+
+    def get_queryset(self):
+        iu = self.get_parent_object()
+        self.check_parent_access(iu)
+        return iu.get_event_queryset()
 
     def finalize_response(self, request, response, *args, **kwargs):
         response['X-UI-Max-Events'] = settings.MAX_UI_JOB_EVENTS

awx/api/views/mixin.py

@@ -52,6 +52,11 @@ def destroy(self, request, *args, **kwargs):
             else:
                 # if it has been > 1 minute, events are probably lost
                 logger.warning('Allowing deletion of {} through the API without all events ' 'processed.'.format(obj.log_format))
+
+        # Manually cascade delete events if unpartitioned job
+        if obj.has_unpartitioned_events:
+            obj.get_event_queryset().delete()
+
         obj.delete()
         return Response(status=status.HTTP_204_NO_CONTENT)
 

awx/api/views/root.py

@@ -106,7 +106,6 @@ def get(self, request, format=None):
         data['hosts'] = reverse('api:host_list', request=request)
         data['job_templates'] = reverse('api:job_template_list', request=request)
         data['jobs'] = reverse('api:job_list', request=request)
-        data['job_events'] = reverse('api:job_event_list', request=request)
         data['ad_hoc_commands'] = reverse('api:ad_hoc_command_list', request=request)
         data['system_job_templates'] = reverse('api:system_job_template_list', request=request)
         data['system_jobs'] = reverse('api:system_job_list', request=request)