Fixing NUL characters in event data

We have both psycopg2 and 3 installed in the AWX venv.

Old versions of Django only used psycopg2 but 4.2 now supports 3

Django 4.2 detects psycopg3 first and will use that over psycopg2

So old migrations needed to be updated to support psycopg3

.github/ISSUE_TEMPLATE/bug_report.yml

@@ -19,8 +19,6 @@ body:
           required: true
         - label: I understand that AWX is open source software provided for free and that I might not receive a timely response.
           required: true
-        - label: I am **NOT** reporting a (potential) security vulnerability. (These should be emailed to `security@ansible.com` instead.)
-          required: true
 
   - type: textarea
     id: summary
@@ -44,7 +42,6 @@ body:
       label: Select the relevant components
       options:
         - label: UI
-        - label: UI (tech preview)
         - label: API
         - label: Docs
         - label: Collection

.github/dependabot.yml

@@ -0,0 +1,19 @@
+version: 2
+updates:
+  - package-ecosystem: "npm"
+    directory: "/awx/ui"
+    schedule:
+      interval: "monthly"
+    open-pull-requests-limit: 5
+    allow:
+      - dependency-type: "production"
+    reviewers:
+      - "AlexSCorey"
+      - "keithjgrant"
+      - "kialam"
+      - "mabashian"
+      - "marshmalien"
+    labels:
+      - "component:ui"
+      - "dependencies"
+    target-branch: "devel"

.github/issue_labeler.yml

@@ -6,8 +6,6 @@ needs_triage:
   - "Feature Summary"
 "component:ui":
   - "\\[X\\] UI"
-"component:ui_next":
-  - "\\[X\\] UI \\(tech preview\\)"
 "component:api":
   - "\\[X\\] API"
 "component:docs":

.github/pr_labeler.yml

@@ -15,5 +15,5 @@
 
 "dependencies":
   - any: ["awx/ui/package.json"]
-  - any: ["requirements/*.txt"]
-  - any: ["requirements/requirements.in"]
+  - any: ["awx/requirements/*.txt"]
+  - any: ["awx/requirements/requirements.in"]

.github/workflows/ci.yml

@@ -3,7 +3,7 @@ name: CI
 env:
   LC_ALL: "C.UTF-8" # prevent ERROR: Ansible could not initialize the preferred locale: unsupported locale setting
   CI_GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
-  DEV_DOCKER_OWNER: ${{ github.repository_owner }}
+  DEV_DOCKER_TAG_BASE: ghcr.io/${{ github.repository_owner }}
   COMPOSE_TAG: ${{ github.base_ref || 'devel' }}
 on:
   pull_request:

.github/workflows/devel_images.yml

@@ -48,11 +48,8 @@ jobs:
           DEV_DOCKER_TAG_BASE=ghcr.io/${OWNER_LC} COMPOSE_TAG=${GITHUB_REF##*/} make awx-kube-dev-build
           DEV_DOCKER_TAG_BASE=ghcr.io/${OWNER_LC} COMPOSE_TAG=${GITHUB_REF##*/} make awx-kube-build
 
-      - name: Push development images
+      - name: Push image
         run: |
           docker push ghcr.io/${OWNER_LC}/awx_devel:${GITHUB_REF##*/}
           docker push ghcr.io/${OWNER_LC}/awx_kube_devel:${GITHUB_REF##*/}
-
-      - name: Push AWX k8s image, only for upstream and feature branches
-        run: docker push ghcr.io/${OWNER_LC}/awx:${GITHUB_REF##*/}
-        if: endsWith(github.repository, '/awx')
+          docker push ghcr.io/${OWNER_LC}/awx:${GITHUB_REF##*/}

.github/workflows/label_issue.yml

@@ -6,9 +6,9 @@ on:
       - opened
       - reopened
 
-permissions:
-  contents: write # to fetch code
-  issues: write # to label issues
+    permissions:
+      contents: read # to fetch code
+      issues: write # to label issues
 
 jobs:
   triage:
@@ -17,7 +17,7 @@ jobs:
 
     steps:
       - name: Label Issue
-        uses: github/issue-labeler@v3.1
+        uses: github/issue-labeler@v2.4.1
         with:
           repo-token: "${{ secrets.GITHUB_TOKEN }}"
           not-before: 2021-12-07T07:00:00Z

.github/workflows/label_pr.yml

@@ -8,7 +8,7 @@ on:
       - synchronize
 
 permissions:
-  contents: write # to determine modified files (actions/labeler)
+  contents: read # to determine modified files (actions/labeler)
   pull-requests: write # to add labels to PRs (actions/labeler)
 
 jobs:

.github/workflows/pr_body_check_jira.yml

@@ -1,35 +0,0 @@
----
-name: Check body for reference to jira
-on:
-  pull_request:
-    branches:
-      - release_**
-jobs:
-  pr-check:
-    if: github.repository_owner == 'ansible' && github.repository != 'awx'
-    name: Scan PR description for JIRA links
-    runs-on: ubuntu-latest
-    permissions:
-      packages: write
-      contents: read
-    steps:
-      - name: Check for JIRA lines
-        env:
-          PR_BODY: ${{ github.event.pull_request.body }}
-        run: |
-          echo "$PR_BODY" | grep "JIRA: None" > no_jira
-          echo "$PR_BODY" | grep "JIRA: https://.*[0-9]+"> jira
-          exit 0
-        # We exit 0 and set the shell to prevent the returns from the greps from failing this step
-        # See https://docs.github.com/en/actions/using-workflows/workflow-syntax-for-github-actions#exit-codes-and-error-action-preference
-        shell: bash {0}
-
-      - name: Check for exactly one item
-        run: |
-          if [ $(cat no_jira jira | wc -l) != 1 ] ; then
-            echo "The PR body must contain exactly one of [ 'JIRA: None' or 'JIRA: <one or more links>' ]"
-            echo "We counted $(cat no_jira jira | wc -l)"
-            exit 255;
-          else
-            exit 0;
-          fi

DATA_MIGRATION.md

@@ -4,6 +4,6 @@
 
 Early versions of AWX did not support seamless upgrades between major versions and required the use of a backup and restore tool to perform upgrades.
 
-As of version 18.0, `awx-operator` is the preferred install/upgrade method. Users who wish to upgrade modern AWX installations should follow the instructions at:
+Users who wish to upgrade modern AWX installations should follow the instructions at:
 
-https://github.com/ansible/awx-operator/blob/devel/docs/upgrade/upgrading.md
+https://github.com/ansible/awx/blob/devel/INSTALL.md#upgrading-from-previous-versions

ISSUES.md

@@ -31,7 +31,7 @@ If your issue isn't considered high priority, then please be patient as it may t
 
 `state:needs_info` The issue needs more information. This could be more debug output, more specifics out the system such as version information. Any detail that is currently preventing this issue from moving forward. This should be considered a blocked state.
 
-`state:needs_review` The issue/pull request needs to be reviewed by other maintainers and contributors. This is usually used when there is a question out to another maintainer or when a person is less familiar with an area of the code base the issue is for.
+`state:needs_review` The issue/pull request needs to be reviewed by other maintainers and contributors. This is usually used when there is a question out to another maintainer or when a person is less familar with an area of the code base the issue is for.
 
 `state:needs_revision` More commonly used on pull requests, this state represents that there are changes that are being waited on.
 

Makefile

@@ -1,7 +1,6 @@
 -include awx/ui_next/Makefile
 
 PYTHON := $(notdir $(shell for i in python3.9 python3; do command -v $$i; done|sed 1q))
-SHELL := bash
 DOCKER_COMPOSE ?= docker-compose
 OFFICIAL ?= no
 NODE ?= node
@@ -9,7 +8,7 @@ NPM_BIN ?= npm
 CHROMIUM_BIN=/tmp/chrome-linux/chrome
 GIT_BRANCH ?= $(shell git rev-parse --abbrev-ref HEAD)
 MANAGEMENT_COMMAND ?= awx-manage
-VERSION ?= $(shell $(PYTHON) tools/scripts/scm_version.py)
+VERSION := $(shell $(PYTHON) tools/scripts/scm_version.py)
 
 # ansible-test requires semver compatable version, so we allow overrides to hack it
 COLLECTION_VERSION ?= $(shell $(PYTHON) tools/scripts/scm_version.py | cut -d . -f 1-3)
@@ -28,8 +27,6 @@ COLLECTION_TEMPLATE_VERSION ?= false
 # NOTE: This defaults the container image version to the branch that's active
 COMPOSE_TAG ?= $(GIT_BRANCH)
 MAIN_NODE_TYPE ?= hybrid
-# If set to true docker-compose will also start a pgbouncer instance and use it
-PGBOUNCER ?= false
 # If set to true docker-compose will also start a keycloak instance
 KEYCLOAK ?= false
 # If set to true docker-compose will also start an ldap instance
@@ -40,24 +37,19 @@ SPLUNK ?= false
 PROMETHEUS ?= false
 # If set to true docker-compose will also start a grafana instance
 GRAFANA ?= false
-# If set to true docker-compose will also start a hashicorp vault instance
-VAULT ?= false
 # If set to true docker-compose will also start a tacacs+ instance
 TACACS ?= false
 
 VENV_BASE ?= /var/lib/awx/venv
 
-DEV_DOCKER_OWNER ?= ansible
-# Docker will only accept lowercase, so github names like Paul need to be paul
-DEV_DOCKER_OWNER_LOWER = $(shell echo $(DEV_DOCKER_OWNER) | tr A-Z a-z)
-DEV_DOCKER_TAG_BASE ?= ghcr.io/$(DEV_DOCKER_OWNER_LOWER)
+DEV_DOCKER_TAG_BASE ?= ghcr.io/ansible
 DEVEL_IMAGE_NAME ?= $(DEV_DOCKER_TAG_BASE)/awx_devel:$(COMPOSE_TAG)
 
 RECEPTOR_IMAGE ?= quay.io/ansible/receptor:devel
 
 # Python packages to install only from source (not from binary wheels)
 # Comma separated list
-SRC_ONLY_PKGS ?= cffi,pycparser,psycopg,twilio
+SRC_ONLY_PKGS ?= cffi,pycparser,psycopg2,twilio
 # These should be upgraded in the AWX and Ansible venv before attempting
 # to install the actual requirements
 VENV_BOOTSTRAP ?= pip==21.2.4 setuptools==65.6.3 setuptools_scm[toml]==7.0.5 wheel==0.38.4
@@ -272,11 +264,11 @@ run-wsrelay:
 	$(PYTHON) manage.py run_wsrelay
 
 ## Start the heartbeat process in background in development environment.
-run-ws-heartbeat:
+run-heartbeet:
 	@if [ "$(VENV_BASE)" ]; then \
 		. $(VENV_BASE)/awx/bin/activate; \
 	fi; \
-	$(PYTHON) manage.py run_ws_heartbeat
+	$(PYTHON) manage.py run_heartbeet
 
 reports:
 	mkdir -p $@
@@ -525,20 +517,15 @@ docker-compose-sources: .git/hooks/pre-commit
 	    -e control_plane_node_count=$(CONTROL_PLANE_NODE_COUNT) \
 	    -e execution_node_count=$(EXECUTION_NODE_COUNT) \
 	    -e minikube_container_group=$(MINIKUBE_CONTAINER_GROUP) \
-	    -e enable_pgbouncer=$(PGBOUNCER) \
 	    -e enable_keycloak=$(KEYCLOAK) \
 	    -e enable_ldap=$(LDAP) \
 	    -e enable_splunk=$(SPLUNK) \
 	    -e enable_prometheus=$(PROMETHEUS) \
 	    -e enable_grafana=$(GRAFANA) \
-	    -e enable_vault=$(VAULT) \
 	    -e enable_tacacs=$(TACACS) \
             $(EXTRA_SOURCES_ANSIBLE_OPTS)
 
 docker-compose: awx/projects docker-compose-sources
-	ansible-galaxy install --ignore-certs -r tools/docker-compose/ansible/requirements.yml;
-	ansible-playbook -i tools/docker-compose/inventory tools/docker-compose/ansible/initialize_containers.yml \
-	  -e enable_vault=$(VAULT);
 	$(DOCKER_COMPOSE) -f tools/docker-compose/_sources/docker-compose.yml $(COMPOSE_OPTS) up $(COMPOSE_UP_OPTS) --remove-orphans
 
 docker-compose-credential-plugins: awx/projects docker-compose-sources
@@ -590,7 +577,7 @@ docker-clean:
 	-$(foreach image_id,$(shell docker images --filter=reference='*/*/*awx_devel*' --filter=reference='*/*awx_devel*' --filter=reference='*awx_devel*' -aq),docker rmi --force $(image_id);)
 
 docker-clean-volumes: docker-compose-clean docker-compose-container-group-clean
-	docker volume rm -f tools_awx_db tools_vault_1 tools_grafana_storage tools_prometheus_storage $(docker volume ls --filter name=tools_redis_socket_ -q)
+	docker volume rm -f tools_awx_db tools_grafana_storage tools_prometheus_storage $(docker volume ls --filter name=tools_redis_socket_ -q)
 
 docker-refresh: docker-clean docker-compose
 
@@ -670,12 +657,10 @@ awx-kube-dev-build: Dockerfile.kube-dev
 ## generate UI .pot file, an empty template of strings yet to be translated
 pot: $(UI_BUILD_FLAG_FILE)
 	$(NPM_BIN) --prefix awx/ui --loglevel warn run extract-template --clean
-	$(NPM_BIN) --prefix awx/ui_next --loglevel warn run extract-template --clean
 
 ## generate UI .po files for each locale (will update translated strings for `en`)
 po: $(UI_BUILD_FLAG_FILE)
 	$(NPM_BIN) --prefix awx/ui --loglevel warn run extract-strings -- --clean
-	$(NPM_BIN) --prefix awx/ui_next --loglevel warn run extract-strings -- --clean
 
 ## generate API django .pot .po
 messages:

awx/api/filters.py

@@ -347,7 +347,7 @@ class FieldLookupBackend(BaseFilterBackend):
                         args.append(Q(**{k: v}))
                 for role_name in role_filters:
                     if not hasattr(queryset.model, 'accessible_pk_qs'):
-                        raise ParseError(_('Cannot apply role_level filter to this list because its model does not use roles for access control.'))
+                        raise ParseError(_('Cannot apply role_level filter to this list because its model ' 'does not use roles for access control.'))
                     args.append(Q(pk__in=queryset.model.accessible_pk_qs(request.user, role_name)))
                 if or_filters:
                     q = Q()

awx/api/generics.py

@@ -169,7 +169,7 @@ class APIView(views.APIView):
             self.__init_request_error__ = exc
         except UnsupportedMediaType as exc:
             exc.detail = _(
-                'You did not use correct Content-Type in your HTTP request. If you are using our REST API, the Content-Type must be application/json'
+                'You did not use correct Content-Type in your HTTP request. ' 'If you are using our REST API, the Content-Type must be application/json'
             )
             self.__init_request_error__ = exc
         return drf_request
@@ -232,8 +232,7 @@ class APIView(views.APIView):
 
         response = super(APIView, self).finalize_response(request, response, *args, **kwargs)
         time_started = getattr(self, 'time_started', None)
-        if request.user.is_authenticated:
-            response['X-API-Product-Version'] = get_awx_version()
+        response['X-API-Product-Version'] = get_awx_version()
         response['X-API-Product-Name'] = server_product_name()
 
         response['X-API-Node'] = settings.CLUSTER_HOST_ID
@@ -523,16 +522,14 @@ class SubListAPIView(ParentMixin, ListAPIView):
     def get_queryset(self):
         parent = self.get_parent_object()
         self.check_parent_access(parent)
+        sublist_qs = self.get_sublist_queryset(parent)
         if not self.filter_read_permission:
-            return optimize_queryset(self.get_sublist_queryset(parent))
-        qs = self.request.user.get_queryset(self.model)
-        if hasattr(self, 'parent_key'):
-            # This is vastly preferable for ReverseForeignKey relationships
-            return qs.filter(**{self.parent_key: parent})
-        return qs.distinct() & self.get_sublist_queryset(parent).distinct()
+            return optimize_queryset(sublist_qs)
+        qs = self.request.user.get_queryset(self.model).distinct()
+        return qs & sublist_qs
 
     def get_sublist_queryset(self, parent):
-        return getattrd(parent, self.relationship)
+        return getattrd(parent, self.relationship).distinct()
 
 
 class DestroyAPIView(generics.DestroyAPIView):
@@ -581,6 +578,15 @@ class SubListCreateAPIView(SubListAPIView, ListCreateAPIView):
         d.update({'parent_key': getattr(self, 'parent_key', None)})
         return d
 
+    def get_queryset(self):
+        if hasattr(self, 'parent_key'):
+            # Prefer this filtering because ForeignKey allows us more assumptions
+            parent = self.get_parent_object()
+            self.check_parent_access(parent)
+            qs = self.request.user.get_queryset(self.model)
+            return qs.filter(**{self.parent_key: parent})
+        return super(SubListCreateAPIView, self).get_queryset()
+
     def create(self, request, *args, **kwargs):
         # If the object ID was not specified, it probably doesn't exist in the
         # DB yet. We want to see if we can create it.  The URL may choose to

awx/api/metadata.py

@@ -71,7 +71,7 @@ class Metadata(metadata.SimpleMetadata):
                 'url': _('URL for this {}.'),
                 'related': _('Data structure with URLs of related resources.'),
                 'summary_fields': _(
-                    'Data structure with name/description for related resources.  The output for some objects may be limited for performance reasons.'
+                    'Data structure with name/description for related resources.  ' 'The output for some objects may be limited for performance reasons.'
                 ),
                 'created': _('Timestamp when this {} was created.'),
                 'modified': _('Timestamp when this {} was last modified.'),

awx/api/serializers.py

@@ -220,7 +220,7 @@ class CopySerializer(serializers.Serializer):
         view = self.context.get('view', None)
         obj = view.get_object()
         if name == obj.name:
-            raise serializers.ValidationError(_('The original object is already named {}, a copy from it cannot have the same name.'.format(name)))
+            raise serializers.ValidationError(_('The original object is already named {}, a copy from' ' it cannot have the same name.'.format(name)))
         return attrs
 
 
@@ -760,7 +760,7 @@ class UnifiedJobTemplateSerializer(BaseSerializer):
 class UnifiedJobSerializer(BaseSerializer):
     show_capabilities = ['start', 'delete']
     event_processing_finished = serializers.BooleanField(
-        help_text=_('Indicates whether all of the events generated by this unified job have been saved to the database.'), read_only=True
+        help_text=_('Indicates whether all of the events generated by this ' 'unified job have been saved to the database.'), read_only=True
     )
 
     class Meta:
@@ -1579,7 +1579,7 @@ class ProjectPlaybooksSerializer(ProjectSerializer):
 
 
 class ProjectInventoriesSerializer(ProjectSerializer):
-    inventory_files = serializers.ReadOnlyField(help_text=_('Array of inventory files and directories available within this project, not comprehensive.'))
+    inventory_files = serializers.ReadOnlyField(help_text=_('Array of inventory files and directories available within this project, ' 'not comprehensive.'))
 
     class Meta:
         model = Project
@@ -1629,8 +1629,8 @@ class ProjectUpdateDetailSerializer(ProjectUpdateSerializer):
         fields = ('*', 'host_status_counts', 'playbook_counts')
 
     def get_playbook_counts(self, obj):
-        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()
+        task_count = obj.project_update_events.filter(event='playbook_on_task_start').count()
+        play_count = obj.project_update_events.filter(event='playbook_on_play_start').count()
 
         data = {'play_count': play_count, 'task_count': task_count}
 
@@ -2905,7 +2905,7 @@ class CredentialSerializer(BaseSerializer):
             ):
                 if getattr(self.instance, related_objects).count() > 0:
                     raise ValidationError(
-                        _('You cannot change the credential type of the credential, as it may break the functionality of the resources using it.')
+                        _('You cannot change the credential type of the credential, as it may break the functionality' ' of the resources using it.')
                     )
 
         return credential_type
@@ -2925,7 +2925,7 @@ class CredentialSerializerCreate(CredentialSerializer):
         default=None,
         write_only=True,
         allow_null=True,
-        help_text=_('Write-only field used to add user to owner role. If provided, do not give either team or organization. Only valid for creation.'),
+        help_text=_('Write-only field used to add user to owner role. If provided, ' 'do not give either team or organization. Only valid for creation.'),
     )
     team = serializers.PrimaryKeyRelatedField(
         queryset=Team.objects.all(),
@@ -2933,14 +2933,14 @@ class CredentialSerializerCreate(CredentialSerializer):
         default=None,
         write_only=True,
         allow_null=True,
-        help_text=_('Write-only field used to add team to owner role. If provided, do not give either user or organization. Only valid for creation.'),
+        help_text=_('Write-only field used to add team to owner role. If provided, ' 'do not give either user or organization. Only valid for creation.'),
     )
     organization = serializers.PrimaryKeyRelatedField(
         queryset=Organization.objects.all(),
         required=False,
         default=None,
         allow_null=True,
-        help_text=_('Inherit permissions from organization roles. If provided on creation, do not give either user or team.'),
+        help_text=_('Inherit permissions from organization roles. If provided on creation, ' 'do not give either user or team.'),
     )
 
     class Meta:
@@ -2962,7 +2962,7 @@ class CredentialSerializerCreate(CredentialSerializer):
         if len(owner_fields) > 1:
             received = ", ".join(sorted(owner_fields))
             raise serializers.ValidationError(
-                {"detail": _("Only one of 'user', 'team', or 'organization' should be provided, received {} fields.".format(received))}
+                {"detail": _("Only one of 'user', 'team', or 'organization' should be provided, " "received {} fields.".format(received))}
             )
 
         if attrs.get('team'):
@@ -3622,7 +3622,7 @@ class SystemJobSerializer(UnifiedJobSerializer):
         try:
             return obj.result_stdout
         except StdoutMaxBytesExceeded as e:
-            return _("Standard Output too large to display ({text_size} bytes), only download supported for sizes over {supported_size} bytes.").format(
+            return _("Standard Output too large to display ({text_size} bytes), " "only download supported for sizes over {supported_size} bytes.").format(
                 text_size=e.total, supported_size=e.supported
             )
 
@@ -4536,7 +4536,7 @@ class JobLaunchSerializer(BaseSerializer):
                 if cred.unique_hash() in provided_mapping.keys():
                     continue  # User replaced credential with new of same type
                 errors.setdefault('credentials', []).append(
-                    _('Removing {} credential at launch time without replacement is not supported. Provided list lacked credential(s): {}.').format(
+                    _('Removing {} credential at launch time without replacement is not supported. ' 'Provided list lacked credential(s): {}.').format(
                         cred.unique_hash(display=True), ', '.join([str(c) for c in removed_creds])
                     )
                 )
@@ -4686,11 +4686,12 @@ class BulkJobNodeSerializer(WorkflowJobNodeSerializer):
     # many-to-many fields
     credentials = serializers.ListField(child=serializers.IntegerField(min_value=1), required=False)
     labels = serializers.ListField(child=serializers.IntegerField(min_value=1), required=False)
-    instance_groups = serializers.ListField(child=serializers.IntegerField(min_value=1), required=False)
+    # TODO: Use instance group role added via PR 13584(once merged), for now everything related to instance group is commented
+    # instance_groups = serializers.ListField(child=serializers.IntegerField(min_value=1), required=False)
 
     class Meta:
         model = WorkflowJobNode
-        fields = ('*', 'credentials', 'labels', 'instance_groups')  # m2m fields are not canonical for WJ nodes
+        fields = ('*', 'credentials', 'labels')  # m2m fields are not canonical for WJ nodes, TODO: add instance_groups once supported
 
     def validate(self, attrs):
         return super(LaunchConfigurationBaseSerializer, self).validate(attrs)
@@ -4750,21 +4751,21 @@ class BulkJobLaunchSerializer(serializers.Serializer):
         requested_use_execution_environments = {job['execution_environment'] for job in attrs['jobs'] if 'execution_environment' in job}
         requested_use_credentials = set()
         requested_use_labels = set()
-        requested_use_instance_groups = set()
+        # requested_use_instance_groups = set()
         for job in attrs['jobs']:
             for cred in job.get('credentials', []):
                 requested_use_credentials.add(cred)
             for label in job.get('labels', []):
                 requested_use_labels.add(label)
-            for instance_group in job.get('instance_groups', []):
-                requested_use_instance_groups.add(instance_group)
+            # for instance_group in job.get('instance_groups', []):
+            #     requested_use_instance_groups.add(instance_group)
 
         key_to_obj_map = {
             "unified_job_template": {obj.id: obj for obj in UnifiedJobTemplate.objects.filter(id__in=requested_ujts)},
             "inventory": {obj.id: obj for obj in Inventory.objects.filter(id__in=requested_use_inventories)},
             "credentials": {obj.id: obj for obj in Credential.objects.filter(id__in=requested_use_credentials)},
             "labels": {obj.id: obj for obj in Label.objects.filter(id__in=requested_use_labels)},
-            "instance_groups": {obj.id: obj for obj in InstanceGroup.objects.filter(id__in=requested_use_instance_groups)},
+            # "instance_groups": {obj.id: obj for obj in InstanceGroup.objects.filter(id__in=requested_use_instance_groups)},
             "execution_environment": {obj.id: obj for obj in ExecutionEnvironment.objects.filter(id__in=requested_use_execution_environments)},
         }
 
@@ -4791,7 +4792,7 @@ class BulkJobLaunchSerializer(serializers.Serializer):
 
         self.check_list_permission(Credential, requested_use_credentials, 'use_role')
         self.check_list_permission(Label, requested_use_labels)
-        self.check_list_permission(InstanceGroup, requested_use_instance_groups)  # TODO: change to use_role for conflict
+        # self.check_list_permission(InstanceGroup, requested_use_instance_groups)  # TODO: change to use_role for conflict
         self.check_list_permission(ExecutionEnvironment, requested_use_execution_environments)  # TODO: change if roles introduced
 
         jobs_object = self.get_objectified_jobs(attrs, key_to_obj_map)
@@ -4838,7 +4839,7 @@ class BulkJobLaunchSerializer(serializers.Serializer):
         node_m2m_object_types_to_through_model = {
             'credentials': WorkflowJobNode.credentials.through,
             'labels': WorkflowJobNode.labels.through,
-            'instance_groups': WorkflowJobNode.instance_groups.through,
+            # 'instance_groups': WorkflowJobNode.instance_groups.through,
         }
         node_deferred_attr_names = (
             'limit',
@@ -4891,9 +4892,9 @@ class BulkJobLaunchSerializer(serializers.Serializer):
                 if field_name in node_m2m_objects[node_identifier] and field_name == 'labels':
                     for label in node_m2m_objects[node_identifier][field_name]:
                         through_model_objects.append(through_model(label=label, workflowjobnode=node_m2m_objects[node_identifier]['node']))
-                if field_name in node_m2m_objects[node_identifier] and field_name == 'instance_groups':
-                    for instance_group in node_m2m_objects[node_identifier][field_name]:
-                        through_model_objects.append(through_model(instancegroup=instance_group, workflowjobnode=node_m2m_objects[node_identifier]['node']))
+                # if obj_type in node_m2m_objects[node_identifier] and obj_type == 'instance_groups':
+                #     for instance_group in node_m2m_objects[node_identifier][obj_type]:
+                #         through_model_objects.append(through_model(instancegroup=instance_group, workflowjobnode=node_m2m_objects[node_identifier]['node']))
             if through_model_objects:
                 through_model.objects.bulk_create(through_model_objects)
 
@@ -5018,7 +5019,7 @@ class NotificationTemplateSerializer(BaseSerializer):
                     for subevent in event_messages:
                         if subevent not in ('running', 'approved', 'timed_out', 'denied'):
                             error_list.append(
-                                _("Workflow Approval event '{}' invalid, must be one of 'running', 'approved', 'timed_out', or 'denied'").format(subevent)
+                                _("Workflow Approval event '{}' invalid, must be one of " "'running', 'approved', 'timed_out', or 'denied'").format(subevent)
                             )
                             continue
                         subevent_messages = event_messages[subevent]
@@ -5435,7 +5436,7 @@ class InstanceSerializer(BaseSerializer):
         res = super(InstanceSerializer, self).get_related(obj)
         res['jobs'] = self.reverse('api:instance_unified_jobs_list', kwargs={'pk': obj.pk})
         res['instance_groups'] = self.reverse('api:instance_instance_groups_list', kwargs={'pk': obj.pk})
-        if obj.node_type in [Instance.Types.EXECUTION, Instance.Types.HOP]:
+        if settings.IS_K8S and obj.node_type in (Instance.Types.EXECUTION,):
             res['install_bundle'] = self.reverse('api:instance_install_bundle', kwargs={'pk': obj.pk})
         res['peers'] = self.reverse('api:instance_peers_list', kwargs={"pk": obj.pk})
         if self.context['request'].user.is_superuser or self.context['request'].user.is_system_auditor:
@@ -5558,7 +5559,7 @@ class InstanceGroupSerializer(BaseSerializer):
     instances = serializers.SerializerMethodField()
     is_container_group = serializers.BooleanField(
         required=False,
-        help_text=_('Indicates whether instances in this group are containerized.Containerized groups have a designated Openshift or Kubernetes cluster.'),
+        help_text=_('Indicates whether instances in this group are containerized.' 'Containerized groups have a designated Openshift or Kubernetes cluster.'),
     )
     # NOTE: help_text is duplicated from field definitions, no obvious way of
     # both defining field details here and also getting the field's help_text
@@ -5569,7 +5570,7 @@ class InstanceGroupSerializer(BaseSerializer):
         required=False,
         initial=0,
         label=_('Policy Instance Percentage'),
-        help_text=_("Minimum percentage of all instances that will be automatically assigned to this group when new instances come online."),
+        help_text=_("Minimum percentage of all instances that will be automatically assigned to " "this group when new instances come online."),
     )
     policy_instance_minimum = serializers.IntegerField(
         default=0,
@@ -5577,7 +5578,7 @@ class InstanceGroupSerializer(BaseSerializer):
         required=False,
         initial=0,
         label=_('Policy Instance Minimum'),
-        help_text=_("Static minimum number of Instances that will be automatically assign to this group when new instances come online."),
+        help_text=_("Static minimum number of Instances that will be automatically assign to " "this group when new instances come online."),
     )
     max_concurrent_jobs = serializers.IntegerField(
         default=0,

awx/api/swagger.py

@@ -1,10 +1,16 @@
+import json
 import warnings
 
-from rest_framework.permissions import AllowAny
-from rest_framework.schemas import SchemaGenerator, AutoSchema as DRFAuthSchema
+from coreapi.document import Object, Link
 
-from drf_yasg.views import get_schema_view
-from drf_yasg import openapi
+from rest_framework import exceptions
+from rest_framework.permissions import AllowAny
+from rest_framework.renderers import CoreJSONRenderer
+from rest_framework.response import Response
+from rest_framework.schemas import SchemaGenerator, AutoSchema as DRFAuthSchema
+from rest_framework.views import APIView
+
+from rest_framework_swagger import renderers
 
 
 class SuperUserSchemaGenerator(SchemaGenerator):
@@ -49,15 +55,43 @@ class AutoSchema(DRFAuthSchema):
         return description
 
 
-schema_view = get_schema_view(
-    openapi.Info(
-        title="Snippets API",
-        default_version='v1',
-        description="Test description",
-        terms_of_service="https://www.google.com/policies/terms/",
-        contact=openapi.Contact(email="contact@snippets.local"),
-        license=openapi.License(name="BSD License"),
-    ),
-    public=True,
-    permission_classes=[AllowAny],
-)
+class SwaggerSchemaView(APIView):
+    _ignore_model_permissions = True
+    exclude_from_schema = True
+    permission_classes = [AllowAny]
+    renderer_classes = [CoreJSONRenderer, renderers.OpenAPIRenderer, renderers.SwaggerUIRenderer]
+
+    def get(self, request):
+        generator = SuperUserSchemaGenerator(title='Ansible Automation Platform controller API', patterns=None, urlconf=None)
+        schema = generator.get_schema(request=request)
+        # python core-api doesn't support the deprecation yet, so track it
+        # ourselves and return it in a response header
+        _deprecated = []
+
+        # By default, DRF OpenAPI serialization places all endpoints in
+        # a single node based on their root path (/api).  Instead, we want to
+        # group them by topic/tag so that they're categorized in the rendered
+        # output
+        document = schema._data.pop('api')
+        for path, node in document.items():
+            if isinstance(node, Object):
+                for action in node.values():
+                    topic = getattr(action, 'topic', None)
+                    if topic:
+                        schema._data.setdefault(topic, Object())
+                        schema._data[topic]._data[path] = node
+
+                    if isinstance(action, Object):
+                        for link in action.links.values():
+                            if link.deprecated:
+                                _deprecated.append(link.url)
+            elif isinstance(node, Link):
+                topic = getattr(node, 'topic', None)
+                if topic:
+                    schema._data.setdefault(topic, Object())
+                    schema._data[topic]._data[path] = node
+
+        if not schema:
+            raise exceptions.ValidationError('The schema generator did not return a schema Document')
+
+        return Response(schema, headers={'X-Deprecated-Paths': json.dumps(_deprecated)})

awx/api/templates/instance_install_bundle/group_vars/all.yml

@@ -9,10 +9,10 @@ receptor_work_commands:
     params: worker
     allowruntimeparams: true
     verifysignature: true
-custom_worksign_public_keyfile: receptor/work_public_key.pem
+custom_worksign_public_keyfile: receptor/work-public-key.pem
 custom_tls_certfile: receptor/tls/receptor.crt
 custom_tls_keyfile: receptor/tls/receptor.key
-custom_ca_certfile: receptor/tls/ca/mesh-CA.crt
+custom_ca_certfile: receptor/tls/ca/receptor-ca.crt
 receptor_protocol: 'tcp'
 receptor_listener: true
 receptor_port: {{ instance.listener_port }}

awx/api/urls/urls.py

@@ -30,7 +30,7 @@ from awx.api.views import (
     OAuth2TokenList,
     ApplicationOAuth2TokenList,
     OAuth2ApplicationDetail,
-    HostMetricSummaryMonthlyList,
+    # HostMetricSummaryMonthlyList, # It will be enabled in future version of the AWX
 )
 
 from awx.api.views.bulk import (
@@ -123,7 +123,8 @@ v2_urls = [
     re_path(r'^constructed_inventories/', include(constructed_inventory_urls)),
     re_path(r'^hosts/', include(host_urls)),
     re_path(r'^host_metrics/', include(host_metric_urls)),
-    re_path(r'^host_metric_summary_monthly/$', HostMetricSummaryMonthlyList.as_view(), name='host_metric_summary_monthly_list'),
+    # It will be enabled in future version of the AWX
+    # re_path(r'^host_metric_summary_monthly/$', HostMetricSummaryMonthlyList.as_view(), name='host_metric_summary_monthly_list'),
     re_path(r'^groups/', include(group_urls)),
     re_path(r'^inventory_sources/', include(inventory_source_urls)),
     re_path(r'^inventory_updates/', include(inventory_update_urls)),
@@ -166,13 +167,10 @@ urlpatterns = [
 ]
 if MODE == 'development':
     # Only include these if we are in the development environment
-    from awx.api.swagger import schema_view
+    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))]
-    urlpatterns += [
-        re_path(r'^swagger(?P<format>\.json|\.yaml)/$', schema_view.without_ui(cache_timeout=0), name='schema-json'),
-        re_path(r'^swagger/$', schema_view.with_ui('swagger', cache_timeout=0), name='schema-swagger-ui'),
-        re_path(r'^redoc/$', schema_view.with_ui('redoc', cache_timeout=0), name='schema-redoc'),
-    ]

awx/api/views/__init__.py

@@ -565,7 +565,7 @@ class LaunchConfigCredentialsBase(SubListAttachDetachAPIView):
         if self.relationship not in ask_mapping:
             return {"msg": _("Related template cannot accept {} on launch.").format(self.relationship)}
         elif sub.passwords_needed:
-            return {"msg": _("Credential that requires user input on launch cannot be used in saved launch configuration.")}
+            return {"msg": _("Credential that requires user input on launch " "cannot be used in saved launch configuration.")}
 
         ask_field_name = ask_mapping[self.relationship]
 
@@ -1564,15 +1564,16 @@ class HostMetricDetail(RetrieveDestroyAPIView):
         return Response(status=status.HTTP_204_NO_CONTENT)
 
 
-class HostMetricSummaryMonthlyList(ListAPIView):
-    name = _("Host Metrics Summary Monthly")
-    model = models.HostMetricSummaryMonthly
-    serializer_class = serializers.HostMetricSummaryMonthlySerializer
-    permission_classes = (IsSystemAdminOrAuditor,)
-    search_fields = ('date',)
-
-    def get_queryset(self):
-        return self.model.objects.all()
+# It will be enabled in future version of the AWX
+# class HostMetricSummaryMonthlyList(ListAPIView):
+#     name = _("Host Metrics Summary Monthly")
+#     model = models.HostMetricSummaryMonthly
+#     serializer_class = serializers.HostMetricSummaryMonthlySerializer
+#     permission_classes = (IsSystemAdminOrAuditor,)
+#     search_fields = ('date',)
+#
+#     def get_queryset(self):
+#         return self.model.objects.all()
 
 
 class HostList(HostRelatedSearchMixin, ListCreateAPIView):
@@ -2500,7 +2501,7 @@ class JobTemplateSurveySpec(GenericAPIView):
                     return Response(
                         dict(
                             error=_(
-                                "$encrypted$ is a reserved keyword for password question defaults, survey question {idx} is type {survey_item[type]}."
+                                "$encrypted$ is a reserved keyword for password question defaults, " "survey question {idx} is type {survey_item[type]}."
                             ).format(**context)
                         ),
                         status=status.HTTP_400_BAD_REQUEST,
@@ -3332,6 +3333,7 @@ class JobLabelList(SubListAPIView):
     serializer_class = serializers.LabelSerializer
     parent_model = models.Job
     relationship = 'labels'
+    parent_key = 'job'
 
 
 class WorkflowJobLabelList(JobLabelList):
@@ -4054,7 +4056,7 @@ class UnifiedJobStdout(RetrieveAPIView):
                 return super(UnifiedJobStdout, self).retrieve(request, *args, **kwargs)
         except models.StdoutMaxBytesExceeded as e:
             response_message = _(
-                "Standard Output too large to display ({text_size} bytes), only download supported for sizes over {supported_size} bytes."
+                "Standard Output too large to display ({text_size} bytes), " "only download supported for sizes over {supported_size} bytes."
             ).format(text_size=e.total, supported_size=e.supported)
             if request.accepted_renderer.format == 'json':
                 return Response({'range': {'start': 0, 'end': 1, 'absolute_end': 1}, 'content': response_message})

awx/api/views/bulk.py

@@ -1,7 +1,5 @@
 from collections import OrderedDict
 
-from django.utils.translation import gettext_lazy as _
-
 from rest_framework.permissions import IsAuthenticated
 from rest_framework.renderers import JSONRenderer
 from rest_framework.reverse import reverse
@@ -20,9 +18,6 @@ from awx.api import (
 
 
 class BulkView(APIView):
-    name = _('Bulk')
-    swagger_topic = 'Bulk'
-
     permission_classes = [IsAuthenticated]
     renderer_classes = [
         renderers.BrowsableAPIRenderer,

awx/api/views/instance_install_bundle.py

@@ -57,11 +57,13 @@ class InstanceInstallBundle(GenericAPIView):
 
         with io.BytesIO() as f:
             with tarfile.open(fileobj=f, mode='w:gz') as tar:
-                # copy /etc/receptor/tls/ca/mesh-CA.crt to receptor/tls/ca in the tar file
-                tar.add(os.path.realpath('/etc/receptor/tls/ca/mesh-CA.crt'), arcname=f"{instance_obj.hostname}_install_bundle/receptor/tls/ca/mesh-CA.crt")
+                # copy /etc/receptor/tls/ca/receptor-ca.crt to receptor/tls/ca in the tar file
+                tar.add(
+                    os.path.realpath('/etc/receptor/tls/ca/receptor-ca.crt'), arcname=f"{instance_obj.hostname}_install_bundle/receptor/tls/ca/receptor-ca.crt"
+                )
 
-                # copy /etc/receptor/work_public_key.pem to receptor/work_public_key.pem
-                tar.add('/etc/receptor/work_public_key.pem', arcname=f"{instance_obj.hostname}_install_bundle/receptor/work_public_key.pem")
+                # copy /etc/receptor/signing/work-public-key.pem to receptor/work-public-key.pem
+                tar.add('/etc/receptor/signing/work-public-key.pem', arcname=f"{instance_obj.hostname}_install_bundle/receptor/work-public-key.pem")
 
                 # generate and write the receptor key to receptor/tls/receptor.key in the tar file
                 key, cert = generate_receptor_tls(instance_obj)
@@ -159,14 +161,14 @@ def generate_receptor_tls(instance_obj):
         .sign(key, hashes.SHA256())
     )
 
-    # sign csr with the receptor ca key from /etc/receptor/ca/mesh-CA.key
-    with open('/etc/receptor/tls/ca/mesh-CA.key', 'rb') as f:
+    # sign csr with the receptor ca key from /etc/receptor/ca/receptor-ca.key
+    with open('/etc/receptor/tls/ca/receptor-ca.key', 'rb') as f:
         ca_key = serialization.load_pem_private_key(
             f.read(),
             password=None,
         )
 
-    with open('/etc/receptor/tls/ca/mesh-CA.crt', 'rb') as f:
+    with open('/etc/receptor/tls/ca/receptor-ca.crt', 'rb') as f:
         ca_cert = x509.load_pem_x509_certificate(f.read())
 
     cert = (

awx/api/views/mixin.py

@@ -50,7 +50,7 @@ class UnifiedJobDeletionMixin(object):
                 return Response({"error": _("Job has not finished processing events.")}, status=status.HTTP_400_BAD_REQUEST)
             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))
+                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:

awx/api/views/root.py

@@ -20,7 +20,6 @@ from rest_framework import status
 
 import requests
 
-from awx import MODE
 from awx.api.generics import APIView
 from awx.conf.registry import settings_registry
 from awx.main.analytics import all_collectors
@@ -55,8 +54,6 @@ class ApiRootView(APIView):
         data['custom_logo'] = settings.CUSTOM_LOGO
         data['custom_login_info'] = settings.CUSTOM_LOGIN_INFO
         data['login_redirect_override'] = settings.LOGIN_REDIRECT_OVERRIDE
-        if MODE == 'development':
-            data['swagger'] = drf_reverse('api:schema-swagger-ui')
         return Response(data)
 
 
@@ -107,7 +104,8 @@ class ApiVersionRootView(APIView):
         data['groups'] = reverse('api:group_list', request=request)
         data['hosts'] = reverse('api:host_list', request=request)
         data['host_metrics'] = reverse('api:host_metric_list', request=request)
-        data['host_metric_summary_monthly'] = reverse('api:host_metric_summary_monthly_list', request=request)
+        # It will be enabled in future version of the AWX
+        # data['host_metric_summary_monthly'] = reverse('api:host_metric_summary_monthly_list', request=request)
         data['job_templates'] = reverse('api:job_template_list', request=request)
         data['jobs'] = reverse('api:job_list', request=request)
         data['ad_hoc_commands'] = reverse('api:ad_hoc_command_list', request=request)

awx/api/views/webhooks.py

@@ -114,7 +114,7 @@ class WebhookReceiverBase(APIView):
         # Ensure that the full contents of the request are captured for multiple uses.
         request.body
 
-        logger.debug("headers: {}\ndata: {}\n".format(request.headers, request.data))
+        logger.debug("headers: {}\n" "data: {}\n".format(request.headers, request.data))
         obj = self.get_object()
         self.check_signature(obj)
 

awx/conf/apps.py

@@ -14,7 +14,7 @@ class ConfConfig(AppConfig):
     def ready(self):
         self.module.autodiscover()
 
-        if not set(sys.argv) & {'migrate', 'check_migrations', 'showmigrations'}:
+        if not set(sys.argv) & {'migrate', 'check_migrations'}:
             from .settings import SettingsWrapper
 
             SettingsWrapper.initialize()

awx/conf/migrations/0010_change_to_JSONField.py

@@ -1,17 +0,0 @@
-# Generated by Django 4.2 on 2023-06-09 19:51
-
-from django.db import migrations, models
-
-
-class Migration(migrations.Migration):
-    dependencies = [
-        ('conf', '0009_rename_proot_settings'),
-    ]
-
-    operations = [
-        migrations.AlterField(
-            model_name='setting',
-            name='value',
-            field=models.JSONField(null=True),
-        ),
-    ]

awx/conf/models.py

@@ -8,6 +8,7 @@ import json
 from django.db import models
 
 # AWX
+from awx.main.fields import JSONBlob
 from awx.main.models.base import CreatedModifiedModel, prevent_search
 from awx.main.utils import encrypt_field
 from awx.conf import settings_registry
@@ -17,7 +18,7 @@ __all__ = ['Setting']
 
 class Setting(CreatedModifiedModel):
     key = models.CharField(max_length=255)
-    value = models.JSONField(null=True)
+    value = JSONBlob(null=True)
     user = prevent_search(models.ForeignKey('auth.User', related_name='settings', default=None, null=True, editable=False, on_delete=models.CASCADE))
 
     def __str__(self):

awx/conf/tests/unit/test_fields.py

@@ -35,7 +35,7 @@ class TestStringListBooleanField:
         field = StringListBooleanField()
         with pytest.raises(ValidationError) as e:
             field.to_internal_value(value)
-        assert e.value.detail[0] == "Expected None, True, False, a string or list of strings but got {} instead.".format(type(value))
+        assert e.value.detail[0] == "Expected None, True, False, a string or list " "of strings but got {} instead.".format(type(value))
 
     @pytest.mark.parametrize("value_in, value_known", FIELD_VALUES)
     def test_to_representation_valid(self, value_in, value_known):
@@ -48,7 +48,7 @@ class TestStringListBooleanField:
         field = StringListBooleanField()
         with pytest.raises(ValidationError) as e:
             field.to_representation(value)
-        assert e.value.detail[0] == "Expected None, True, False, a string or list of strings but got {} instead.".format(type(value))
+        assert e.value.detail[0] == "Expected None, True, False, a string or list " "of strings but got {} instead.".format(type(value))
 
 
 class TestListTuplesField:
@@ -67,7 +67,7 @@ class TestListTuplesField:
         field = ListTuplesField()
         with pytest.raises(ValidationError) as e:
             field.to_internal_value(value)
-        assert e.value.detail[0] == "Expected a list of tuples of max length 2 but got {} instead.".format(t)
+        assert e.value.detail[0] == "Expected a list of tuples of max length 2 " "but got {} instead.".format(t)
 
 
 class TestStringListPathField:

awx/main/access.py

@@ -366,9 +366,9 @@ class BaseAccess(object):
             report_violation = lambda message: None
         else:
             report_violation = lambda message: logger.warning(message)
-        if validation_info.get('trial', False) is True:
+        if validation_info.get('trial', False) is True or validation_info['instance_count'] == 10:  # basic 10 license
 
-            def report_violation(message):  # noqa
+            def report_violation(message):
                 raise PermissionDenied(message)
 
         if check_expiration and validation_info.get('time_remaining', None) is None:
@@ -2234,7 +2234,7 @@ class WorkflowJobAccess(BaseAccess):
             if not node_access.can_add({'reference_obj': node}):
                 wj_add_perm = False
         if not wj_add_perm and self.save_messages:
-            self.messages['workflow_job_template'] = _('You do not have permission to the workflow job resources required for relaunch.')
+            self.messages['workflow_job_template'] = _('You do not have permission to the workflow job ' 'resources required for relaunch.')
         return wj_add_perm
 
     def can_cancel(self, obj):

awx/main/analytics/collectors.py

@@ -613,20 +613,3 @@ def host_metric_table(since, full_path, until, **kwargs):
         since.isoformat(), until.isoformat(), since.isoformat(), until.isoformat()
     )
     return _copy_table(table='host_metric', query=host_metric_query, path=full_path)
-
-
-@register('host_metric_summary_monthly_table', '1.0', format='csv', description=_('HostMetricSummaryMonthly export, full sync'), expensive=trivial_slicing)
-def host_metric_summary_monthly_table(since, full_path, **kwargs):
-    query = '''
-    COPY (SELECT main_hostmetricsummarymonthly.id,
-                 main_hostmetricsummarymonthly.date,
-                 main_hostmetricsummarymonthly.license_capacity,
-                 main_hostmetricsummarymonthly.license_consumed,
-                 main_hostmetricsummarymonthly.hosts_added,
-                 main_hostmetricsummarymonthly.hosts_deleted,
-                 main_hostmetricsummarymonthly.indirectly_managed_hosts
-          FROM main_hostmetricsummarymonthly
-          ORDER BY main_hostmetricsummarymonthly.id ASC) TO STDOUT WITH CSV HEADER
-    '''
-
-    return _copy_table(table='host_metric_summary_monthly', query=query, path=full_path)

awx/main/analytics/subsystem_metrics.py

@@ -209,11 +209,6 @@ class Metrics:
             SetFloatM('workflow_manager_recorded_timestamp', 'Unix timestamp when metrics were last recorded'),
             SetFloatM('workflow_manager_spawn_workflow_graph_jobs_seconds', 'Time spent spawning workflow tasks'),
             SetFloatM('workflow_manager_get_tasks_seconds', 'Time spent loading workflow tasks from db'),
-            # dispatcher subsystem metrics
-            SetIntM('dispatcher_pool_scale_up_events', 'Number of times local dispatcher scaled up a worker since startup'),
-            SetIntM('dispatcher_pool_active_task_count', 'Number of active tasks in the worker pool when last task was submitted'),
-            SetIntM('dispatcher_pool_max_worker_count', 'Highest number of workers in worker pool in last collection interval, about 20s'),
-            SetFloatM('dispatcher_availability', 'Fraction of time (in last collection interval) dispatcher was able to receive messages'),
         ]
         # turn metric list into dictionary with the metric name as a key
         self.METRICS = {}

awx/main/cache.py

@@ -1,87 +0,0 @@
-import functools
-
-from django.conf import settings
-from django.core.cache.backends.base import DEFAULT_TIMEOUT
-from django.core.cache.backends.redis import RedisCache
-
-from redis.exceptions import ConnectionError, ResponseError, TimeoutError
-import socket
-
-# This list comes from what django-redis ignores and the behavior we are trying
-# to retain while dropping the dependency on django-redis.
-IGNORED_EXCEPTIONS = (TimeoutError, ResponseError, ConnectionError, socket.timeout)
-
-CONNECTION_INTERRUPTED_SENTINEL = object()
-
-
-def optionally_ignore_exceptions(func=None, return_value=None):
-    if func is None:
-        return functools.partial(optionally_ignore_exceptions, return_value=return_value)
-
-    @functools.wraps(func)
-    def wrapper(*args, **kwargs):
-        try:
-            return func(*args, **kwargs)
-        except IGNORED_EXCEPTIONS as e:
-            if settings.DJANGO_REDIS_IGNORE_EXCEPTIONS:
-                return return_value
-            raise e.__cause__ or e
-
-    return wrapper
-
-
-class AWXRedisCache(RedisCache):
-    """
-    We just want to wrap the upstream RedisCache class so that we can ignore
-    the exceptions that it raises when the cache is unavailable.
-    """
-
-    @optionally_ignore_exceptions
-    def add(self, key, value, timeout=DEFAULT_TIMEOUT, version=None):
-        return super().add(key, value, timeout, version)
-
-    @optionally_ignore_exceptions(return_value=CONNECTION_INTERRUPTED_SENTINEL)
-    def _get(self, key, default=None, version=None):
-        return super().get(key, default, version)
-
-    def get(self, key, default=None, version=None):
-        value = self._get(key, default, version)
-        if value is CONNECTION_INTERRUPTED_SENTINEL:
-            return default
-        return value
-
-    @optionally_ignore_exceptions
-    def set(self, key, value, timeout=DEFAULT_TIMEOUT, version=None):
-        return super().set(key, value, timeout, version)
-
-    @optionally_ignore_exceptions
-    def touch(self, key, timeout=DEFAULT_TIMEOUT, version=None):
-        return super().touch(key, timeout, version)
-
-    @optionally_ignore_exceptions
-    def delete(self, key, version=None):
-        return super().delete(key, version)
-
-    @optionally_ignore_exceptions
-    def get_many(self, keys, version=None):
-        return super().get_many(keys, version)
-
-    @optionally_ignore_exceptions
-    def has_key(self, key, version=None):
-        return super().has_key(key, version)
-
-    @optionally_ignore_exceptions
-    def incr(self, key, delta=1, version=None):
-        return super().incr(key, delta, version)
-
-    @optionally_ignore_exceptions
-    def set_many(self, data, timeout=DEFAULT_TIMEOUT, version=None):
-        return super().set_many(data, timeout, version)
-
-    @optionally_ignore_exceptions
-    def delete_many(self, keys, version=None):
-        return super().delete_many(keys, version)
-
-    @optionally_ignore_exceptions
-    def clear(self):
-        return super().clear()

awx/main/conf.py

@@ -94,20 +94,6 @@ register(
     category_slug='system',
 )
 
-register(
-    'CSRF_TRUSTED_ORIGINS',
-    default=[],
-    field_class=fields.StringListField,
-    label=_('CSRF Trusted Origins List'),
-    help_text=_(
-        "If the service is behind a reverse proxy/load balancer, use this setting "
-        "to configure the schema://addresses from which the service should trust "
-        "Origin header values. "
-    ),
-    category=_('System'),
-    category_slug='system',
-)
-
 register(
     'LICENSE',
     field_class=fields.DictField,
@@ -698,28 +684,11 @@ register(
     field_class=fields.IntegerField,
     default=1,
     min_value=1,
-    label=_('Maximum disk persistence for external log aggregation (in GB)'),
+    label=_('Maximum disk persistance for external log aggregation (in GB)'),
     help_text=_(
         'Amount of data to store (in gigabytes) during an outage of '
         'the external log aggregator (defaults to 1). '
-        'Equivalent to the rsyslogd queue.maxdiskspace setting for main_queue. '
-        'Notably, this is used for the rsyslogd main queue (for input messages).'
-    ),
-    category=_('Logging'),
-    category_slug='logging',
-)
-register(
-    'LOG_AGGREGATOR_ACTION_MAX_DISK_USAGE_GB',
-    field_class=fields.IntegerField,
-    default=1,
-    min_value=1,
-    label=_('Maximum disk persistence for rsyslogd action queuing (in GB)'),
-    help_text=_(
-        'Amount of data to store (in gigabytes) if an rsyslog action takes time '
-        'to process an incoming message (defaults to 1). '
-        'Equivalent to the rsyslogd queue.maxdiskspace setting on the action (e.g. omhttp). '
-        'Like LOG_AGGREGATOR_MAX_DISK_USAGE_GB, it stores files in the directory specified '
-        'by LOG_AGGREGATOR_MAX_DISK_USAGE_PATH.'
+        'Equivalent to the rsyslogd queue.maxdiskspace setting.'
     ),
     category=_('Logging'),
     category_slug='logging',
@@ -862,55 +831,6 @@ register(
     category_slug='system',
 )
 
-register(
-    'HOST_METRIC_SUMMARY_TASK_LAST_TS',
-    field_class=fields.DateTimeField,
-    label=_('Last computing date of HostMetricSummaryMonthly'),
-    allow_null=True,
-    category=_('System'),
-    category_slug='system',
-)
-
-register(
-    'AWX_CLEANUP_PATHS',
-    field_class=fields.BooleanField,
-    label=_('Enable or Disable tmp dir cleanup'),
-    default=True,
-    help_text=_('Enable or Disable TMP Dir cleanup'),
-    category=('Debug'),
-    category_slug='debug',
-)
-
-register(
-    'AWX_REQUEST_PROFILE',
-    field_class=fields.BooleanField,
-    label=_('Debug Web Requests'),
-    default=False,
-    help_text=_('Debug web request python timing'),
-    category=('Debug'),
-    category_slug='debug',
-)
-
-register(
-    'DEFAULT_CONTAINER_RUN_OPTIONS',
-    field_class=fields.StringListField,
-    label=_('Container Run Options'),
-    default=['--network', 'slirp4netns:enable_ipv6=true'],
-    help_text=_("List of options to pass to podman run example: ['--network', 'slirp4netns:enable_ipv6=true', '--log-level', 'debug']"),
-    category=('Jobs'),
-    category_slug='jobs',
-)
-
-register(
-    'RECEPTOR_RELEASE_WORK',
-    field_class=fields.BooleanField,
-    label=_('Release Receptor Work'),
-    default=True,
-    help_text=_('Release receptor work'),
-    category=('Debug'),
-    category_slug='debug',
-)
-
 
 def logging_validate(serializer, attrs):
     if not serializer.instance or not hasattr(serializer.instance, 'LOG_AGGREGATOR_HOST') or not hasattr(serializer.instance, 'LOG_AGGREGATOR_TYPE'):

awx/main/credential_plugins/aws_secretsmanager.py

@@ -1,65 +0,0 @@
-import boto3
-from botocore.exceptions import ClientError
-
-from .plugin import CredentialPlugin
-from django.utils.translation import gettext_lazy as _
-
-
-secrets_manager_inputs = {
-    'fields': [
-        {
-            'id': 'aws_access_key',
-            'label': _('AWS Access Key'),
-            'type': 'string',
-        },
-        {
-            'id': 'aws_secret_key',
-            'label': _('AWS Secret Key'),
-            'type': 'string',
-            'secret': True,
-        },
-    ],
-    'metadata': [
-        {
-            'id': 'region_name',
-            'label': _('AWS Secrets Manager Region'),
-            'type': 'string',
-            'help_text': _('Region which the secrets manager is located'),
-        },
-        {
-            'id': 'secret_name',
-            'label': _('AWS Secret Name'),
-            'type': 'string',
-        },
-    ],
-    'required': ['aws_access_key', 'aws_secret_key', 'region_name', 'secret_name'],
-}
-
-
-def aws_secretsmanager_backend(**kwargs):
-    secret_name = kwargs['secret_name']
-    region_name = kwargs['region_name']
-    aws_secret_access_key = kwargs['aws_secret_key']
-    aws_access_key_id = kwargs['aws_access_key']
-
-    session = boto3.session.Session()
-    client = session.client(
-        service_name='secretsmanager', region_name=region_name, aws_secret_access_key=aws_secret_access_key, aws_access_key_id=aws_access_key_id
-    )
-
-    try:
-        get_secret_value_response = client.get_secret_value(SecretId=secret_name)
-    except ClientError as e:
-        raise e
-    # Secrets Manager decrypts the secret value using the associated KMS CMK
-    # Depending on whether the secret was a string or binary, only one of these fields will be populated
-    if 'SecretString' in get_secret_value_response:
-        secret = get_secret_value_response['SecretString']
-
-    else:
-        secret = get_secret_value_response['SecretBinary']
-
-    return secret
-
-
-aws_secretmanager_plugin = CredentialPlugin('AWS Secrets Manager lookup', inputs=secrets_manager_inputs, backend=aws_secretsmanager_backend)

awx/main/credential_plugins/hashivault.py

@@ -265,8 +265,6 @@ def kv_backend(**kwargs):
 
     if secret_key:
         try:
-            if (secret_key != 'data') and (secret_key not in json['data']) and ('data' in json['data']):
-                return json['data']['data'][secret_key]
             return json['data'][secret_key]
         except KeyError:
             raise RuntimeError('{} is not present at {}'.format(secret_key, secret_path))

awx/main/credential_plugins/tss.py

@@ -1,10 +1,7 @@
 from .plugin import CredentialPlugin
 from django.utils.translation import gettext_lazy as _
 
-try:
-    from delinea.secrets.server import DomainPasswordGrantAuthorizer, PasswordGrantAuthorizer, SecretServer, ServerSecret
-except ImportError:
-    from thycotic.secrets.server import DomainPasswordGrantAuthorizer, PasswordGrantAuthorizer, SecretServer, ServerSecret
+from thycotic.secrets.server import DomainPasswordGrantAuthorizer, PasswordGrantAuthorizer, SecretServer, ServerSecret
 
 tss_inputs = {
     'fields': [
@@ -53,7 +50,7 @@ tss_inputs = {
 
 
 def tss_backend(**kwargs):
-    if kwargs.get("domain"):
+    if 'domain' in kwargs:
         authorizer = DomainPasswordGrantAuthorizer(kwargs['server_url'], kwargs['username'], kwargs['password'], kwargs['domain'])
     else:
         authorizer = PasswordGrantAuthorizer(kwargs['server_url'], kwargs['username'], kwargs['password'])

awx/main/db/profiled_pg/base.py

@@ -87,7 +87,7 @@ class RecordedQueryLog(object):
             )
             log.commit()
             log.execute(
-                'INSERT INTO queries (pid, version, argv, time, sql, explain, bt) VALUES (?, ?, ?, ?, ?, ?, ?);',
+                'INSERT INTO queries (pid, version, argv, time, sql, explain, bt) ' 'VALUES (?, ?, ?, ?, ?, ?, ?);',
                 (os.getpid(), version, ' '.join(sys.argv), seconds, sql, explain, bt),
             )
             log.commit()

awx/main/dispatch/__init__.py

@@ -1,5 +1,5 @@
 import os
-import psycopg
+import psycopg2
 import select
 
 from contextlib import contextmanager
@@ -40,12 +40,8 @@ def get_task_queuename():
 
 
 class PubSub(object):
-    def __init__(self, conn, select_timeout=None):
+    def __init__(self, conn):
         self.conn = conn
-        if select_timeout is None:
-            self.select_timeout = 5
-        else:
-            self.select_timeout = select_timeout
 
     def listen(self, channel):
         with self.conn.cursor() as cur:
@@ -59,42 +55,25 @@ class PubSub(object):
         with self.conn.cursor() as cur:
             cur.execute('SELECT pg_notify(%s, %s);', (channel, payload))
 
-    @staticmethod
-    def current_notifies(conn):
-        """
-        Altered version of .notifies method from psycopg library
-        This removes the outer while True loop so that we only process
-        queued notifications
-        """
-        with conn.lock:
-            try:
-                ns = conn.wait(psycopg.generators.notifies(conn.pgconn))
-            except psycopg.errors._NO_TRACEBACK as ex:
-                raise ex.with_traceback(None)
-        enc = psycopg._encodings.pgconn_encoding(conn.pgconn)
-        for pgn in ns:
-            n = psycopg.connection.Notify(pgn.relname.decode(enc), pgn.extra.decode(enc), pgn.be_pid)
-            yield n
-
-    def events(self, yield_timeouts=False):
+    def events(self, select_timeout=5, yield_timeouts=False):
         if not self.conn.autocommit:
             raise RuntimeError('Listening for events can only be done in autocommit mode')
 
         while True:
-            if select.select([self.conn], [], [], self.select_timeout) == NOT_READY:
+            if select.select([self.conn], [], [], select_timeout) == NOT_READY:
                 if yield_timeouts:
                     yield None
             else:
-                notification_generator = self.current_notifies(self.conn)
-                for notification in notification_generator:
-                    yield notification
+                self.conn.poll()
+                while self.conn.notifies:
+                    yield self.conn.notifies.pop(0)
 
     def close(self):
         self.conn.close()
 
 
 @contextmanager
-def pg_bus_conn(new_connection=False, select_timeout=None):
+def pg_bus_conn(new_connection=False):
     '''
     Any listeners probably want to establish a new database connection,
     separate from the Django connection used for queries, because that will prevent
@@ -110,8 +89,9 @@ def pg_bus_conn(new_connection=False, select_timeout=None):
         conf['OPTIONS'] = conf.get('OPTIONS', {}).copy()
         # Modify the application name to distinguish from other connections the process might use
         conf['OPTIONS']['application_name'] = get_application_name(settings.CLUSTER_HOST_ID, function='listener')
-        connection_data = f"dbname={conf['NAME']} host={conf['HOST']} user={conf['USER']} password={conf['PASSWORD']} port={conf['PORT']}"
-        conn = psycopg.connect(connection_data, autocommit=True, **conf['OPTIONS'])
+        conn = psycopg2.connect(dbname=conf['NAME'], host=conf['HOST'], user=conf['USER'], password=conf['PASSWORD'], port=conf['PORT'], **conf['OPTIONS'])
+        # Django connection.cursor().connection doesn't have autocommit=True on by default
+        conn.set_session(autocommit=True)
     else:
         if pg_connection.connection is None:
             pg_connection.connect()
@@ -119,7 +99,7 @@ def pg_bus_conn(new_connection=False, select_timeout=None):
             raise RuntimeError('Unexpectedly could not connect to postgres for pg_notify actions')
         conn = pg_connection.connection
 
-    pubsub = PubSub(conn, select_timeout=select_timeout)
+    pubsub = PubSub(conn)
     yield pubsub
     if new_connection:
         conn.close()

awx/main/dispatch/control.py

@@ -40,9 +40,6 @@ class Control(object):
     def cancel(self, task_ids, *args, **kwargs):
         return self.control_with_reply('cancel', *args, extra_data={'task_ids': task_ids}, **kwargs)
 
-    def schedule(self, *args, **kwargs):
-        return self.control_with_reply('schedule', *args, **kwargs)
-
     @classmethod
     def generate_reply_queue_name(cls):
         return f"reply_to_{str(uuid.uuid4()).replace('-','_')}"
@@ -55,14 +52,14 @@ class Control(object):
         if not connection.get_autocommit():
             raise RuntimeError('Control-with-reply messages can only be done in autocommit mode')
 
-        with pg_bus_conn(select_timeout=timeout) as conn:
+        with pg_bus_conn() as conn:
             conn.listen(reply_queue)
             send_data = {'control': command, 'reply_to': reply_queue}
             if extra_data:
                 send_data.update(extra_data)
             conn.notify(self.queuename, json.dumps(send_data))
 
-            for reply in conn.events(yield_timeouts=True):
+            for reply in conn.events(select_timeout=timeout, yield_timeouts=True):
                 if reply is None:
                     logger.error(f'{self.service} did not reply within {timeout}s')
                     raise RuntimeError(f"{self.service} did not reply within {timeout}s")

awx/main/dispatch/periodic.py

@@ -1,142 +1,57 @@
 import logging
+import os
 import time
-import yaml
-from datetime import datetime
+from multiprocessing import Process
 
+from django.conf import settings
+from django.db import connections
+from schedule import Scheduler
+from django_guid import set_guid
+from django_guid.utils import generate_guid
+
+from awx.main.dispatch.worker import TaskWorker
+from awx.main.utils.db import set_connection_name
 
 logger = logging.getLogger('awx.main.dispatch.periodic')
 
 
-class ScheduledTask:
-    """
-    Class representing schedules, very loosely modeled after python schedule library Job
-    the idea of this class is to:
-     - only deal in relative times (time since the scheduler global start)
-     - only deal in integer math for target runtimes, but float for current relative time
+class Scheduler(Scheduler):
+    def run_continuously(self):
+        idle_seconds = max(1, min(self.jobs).period.total_seconds() / 2)
 
-    Missed schedule policy:
-    Invariant target times are maintained, meaning that if interval=10s offset=0
-    and it runs at t=7s, then it calls for next run in 3s.
-    However, if a complete interval has passed, that is counted as a missed run,
-    and missed runs are abandoned (no catch-up runs).
-    """
+        def run():
+            ppid = os.getppid()
+            logger.warning('periodic beat started')
 
-    def __init__(self, name: str, data: dict):
-        # parameters need for schedule computation
-        self.interval = int(data['schedule'].total_seconds())
-        self.offset = 0  # offset relative to start time this schedule begins
-        self.index = 0  # number of periods of the schedule that has passed
+            set_connection_name('periodic')  # set application_name to distinguish from other dispatcher processes
 
-        # parameters that do not affect scheduling logic
-        self.last_run = None  # time of last run, only used for debug
-        self.completed_runs = 0  # number of times schedule is known to run
-        self.name = name
-        self.data = data  # used by caller to know what to run
+            while True:
+                if os.getppid() != ppid:
+                    # if the parent PID changes, this process has been orphaned
+                    # via e.g., segfault or sigkill, we should exit too
+                    pid = os.getpid()
+                    logger.warning(f'periodic beat exiting gracefully pid:{pid}')
+                    raise SystemExit()
+                try:
+                    for conn in connections.all():
+                        # If the database connection has a hiccup, re-establish a new
+                        # connection
+                        conn.close_if_unusable_or_obsolete()
+                    set_guid(generate_guid())
+                    self.run_pending()
+                except Exception:
+                    logger.exception('encountered an error while scheduling periodic tasks')
+                time.sleep(idle_seconds)
 
-    @property
-    def next_run(self):
-        "Time until the next run with t=0 being the global_start of the scheduler class"
-        return (self.index + 1) * self.interval + self.offset
-
-    def due_to_run(self, relative_time):
-        return bool(self.next_run <= relative_time)
-
-    def expected_runs(self, relative_time):
-        return int((relative_time - self.offset) / self.interval)
-
-    def mark_run(self, relative_time):
-        self.last_run = relative_time
-        self.completed_runs += 1
-        new_index = self.expected_runs(relative_time)
-        if new_index > self.index + 1:
-            logger.warning(f'Missed {new_index - self.index - 1} schedules of {self.name}')
-        self.index = new_index
-
-    def missed_runs(self, relative_time):
-        "Number of times job was supposed to ran but failed to, only used for debug"
-        missed_ct = self.expected_runs(relative_time) - self.completed_runs
-        # if this is currently due to run do not count that as a missed run
-        if missed_ct and self.due_to_run(relative_time):
-            missed_ct -= 1
-        return missed_ct
+        process = Process(target=run)
+        process.daemon = True
+        process.start()
 
 
-class Scheduler:
-    def __init__(self, schedule):
-        """
-        Expects schedule in the form of a dictionary like
-        {
-            'job1': {'schedule': timedelta(seconds=50), 'other': 'stuff'}
-        }
-        Only the schedule nearest-second value is used for scheduling,
-        the rest of the data is for use by the caller to know what to run.
-        """
-        self.jobs = [ScheduledTask(name, data) for name, data in schedule.items()]
-        min_interval = min(job.interval for job in self.jobs)
-        num_jobs = len(self.jobs)
-
-        # this is intentionally oppioniated against spammy schedules
-        # a core goal is to spread out the scheduled tasks (for worker management)
-        # and high-frequency schedules just do not work with that
-        if num_jobs > min_interval:
-            raise RuntimeError(f'Number of schedules ({num_jobs}) is more than the shortest schedule interval ({min_interval} seconds).')
-
-        # even space out jobs over the base interval
-        for i, job in enumerate(self.jobs):
-            job.offset = (i * min_interval) // num_jobs
-
-        # internally times are all referenced relative to startup time, add grace period
-        self.global_start = time.time() + 2.0
-
-    def get_and_mark_pending(self):
-        relative_time = time.time() - self.global_start
-        to_run = []
-        for job in self.jobs:
-            if job.due_to_run(relative_time):
-                to_run.append(job)
-                logger.debug(f'scheduler found {job.name} to run, {relative_time - job.next_run} seconds after target')
-                job.mark_run(relative_time)
-        return to_run
-
-    def time_until_next_run(self):
-        relative_time = time.time() - self.global_start
-        next_job = min(self.jobs, key=lambda j: j.next_run)
-        delta = next_job.next_run - relative_time
-        if delta <= 0.1:
-            # careful not to give 0 or negative values to the select timeout, which has unclear interpretation
-            logger.warning(f'Scheduler next run of {next_job.name} is {-delta} seconds in the past')
-            return 0.1
-        elif delta > 20.0:
-            logger.warning(f'Scheduler next run unexpectedly over 20 seconds in future: {delta}')
-            return 20.0
-        logger.debug(f'Scheduler next run is {next_job.name} in {delta} seconds')
-        return delta
-
-    def debug(self, *args, **kwargs):
-        data = dict()
-        data['title'] = 'Scheduler status'
-
-        now = datetime.fromtimestamp(time.time()).strftime('%Y-%m-%d %H:%M:%S UTC')
-        start_time = datetime.fromtimestamp(self.global_start).strftime('%Y-%m-%d %H:%M:%S UTC')
-        relative_time = time.time() - self.global_start
-        data['started_time'] = start_time
-        data['current_time'] = now
-        data['current_time_relative'] = round(relative_time, 3)
-        data['total_schedules'] = len(self.jobs)
-
-        data['schedule_list'] = dict(
-            [
-                (
-                    job.name,
-                    dict(
-                        last_run_seconds_ago=round(relative_time - job.last_run, 3) if job.last_run else None,
-                        next_run_in_seconds=round(job.next_run - relative_time, 3),
-                        offset_in_seconds=job.offset,
-                        completed_runs=job.completed_runs,
-                        missed_runs=job.missed_runs(relative_time),
-                    ),
-                )
-                for job in sorted(self.jobs, key=lambda job: job.interval)
-            ]
-        )
-        return yaml.safe_dump(data, default_flow_style=False, sort_keys=False)
+def run_continuously():
+    scheduler = Scheduler()
+    for task in settings.CELERYBEAT_SCHEDULE.values():
+        apply_async = TaskWorker.resolve_callable(task['task']).apply_async
+        total_seconds = task['schedule'].total_seconds()
+        scheduler.every(total_seconds).seconds.do(apply_async)
+    scheduler.run_continuously()

awx/main/dispatch/pool.py

@@ -339,17 +339,6 @@ class AutoscalePool(WorkerPool):
         # but if the task takes longer than the time defined here, we will force it to stop here
         self.task_manager_timeout = settings.TASK_MANAGER_TIMEOUT + settings.TASK_MANAGER_TIMEOUT_GRACE_PERIOD
 
-        # initialize some things for subsystem metrics periodic gathering
-        # the AutoscalePool class does not save these to redis directly, but reports via produce_subsystem_metrics
-        self.scale_up_ct = 0
-        self.worker_count_max = 0
-
-    def produce_subsystem_metrics(self, metrics_object):
-        metrics_object.set('dispatcher_pool_scale_up_events', self.scale_up_ct)
-        metrics_object.set('dispatcher_pool_active_task_count', sum(len(w.managed_tasks) for w in self.workers))
-        metrics_object.set('dispatcher_pool_max_worker_count', self.worker_count_max)
-        self.worker_count_max = len(self.workers)
-
     @property
     def should_grow(self):
         if len(self.workers) < self.min_workers:
@@ -417,16 +406,16 @@ class AutoscalePool(WorkerPool):
                 # the task manager to never do more work
                 current_task = w.current_task
                 if current_task and isinstance(current_task, dict):
-                    endings = ('tasks.task_manager', 'tasks.dependency_manager', 'tasks.workflow_manager')
+                    endings = ['tasks.task_manager', 'tasks.dependency_manager', 'tasks.workflow_manager']
                     current_task_name = current_task.get('task', '')
-                    if current_task_name.endswith(endings):
+                    if any(current_task_name.endswith(e) for e in endings):
                         if 'started' not in current_task:
                             w.managed_tasks[current_task['uuid']]['started'] = time.time()
                         age = time.time() - current_task['started']
                         w.managed_tasks[current_task['uuid']]['age'] = age
                         if age > self.task_manager_timeout:
-                            logger.error(f'{current_task_name} has held the advisory lock for {age}, sending SIGUSR1 to {w.pid}')
-                            os.kill(w.pid, signal.SIGUSR1)
+                            logger.error(f'{current_task_name} has held the advisory lock for {age}, sending SIGTERM to {w.pid}')
+                            os.kill(w.pid, signal.SIGTERM)
 
         for m in orphaned:
             # if all the workers are dead, spawn at least one
@@ -454,12 +443,7 @@ class AutoscalePool(WorkerPool):
             idx = random.choice(range(len(self.workers)))
             return idx, self.workers[idx]
         else:
-            self.scale_up_ct += 1
-            ret = super(AutoscalePool, self).up()
-            new_worker_ct = len(self.workers)
-            if new_worker_ct > self.worker_count_max:
-                self.worker_count_max = new_worker_ct
-            return ret
+            return super(AutoscalePool, self).up()
 
     def write(self, preferred_queue, body):
         if 'guid' in body:

awx/main/dispatch/publish.py

@@ -73,15 +73,15 @@ class task:
                 return cls.apply_async(args, kwargs)
 
             @classmethod
-            def get_async_body(cls, args=None, kwargs=None, uuid=None, **kw):
-                """
-                Get the python dict to become JSON data in the pg_notify message
-                This same message gets passed over the dispatcher IPC queue to workers
-                If a task is submitted to a multiprocessing pool, skipping pg_notify, this might be used directly
-                """
+            def apply_async(cls, args=None, kwargs=None, queue=None, uuid=None, **kw):
                 task_id = uuid or str(uuid4())
                 args = args or []
                 kwargs = kwargs or {}
+                queue = queue or getattr(cls.queue, 'im_func', cls.queue)
+                if not queue:
+                    msg = f'{cls.name}: Queue value required and may not be None'
+                    logger.error(msg)
+                    raise ValueError(msg)
                 obj = {'uuid': task_id, 'args': args, 'kwargs': kwargs, 'task': cls.name, 'time_pub': time.time()}
                 guid = get_guid()
                 if guid:
@@ -89,16 +89,6 @@ class task:
                 if bind_kwargs:
                     obj['bind_kwargs'] = bind_kwargs
                 obj.update(**kw)
-                return obj
-
-            @classmethod
-            def apply_async(cls, args=None, kwargs=None, queue=None, uuid=None, **kw):
-                queue = queue or getattr(cls.queue, 'im_func', cls.queue)
-                if not queue:
-                    msg = f'{cls.name}: Queue value required and may not be None'
-                    logger.error(msg)
-                    raise ValueError(msg)
-                obj = cls.get_async_body(args=args, kwargs=kwargs, uuid=uuid, **kw)
                 if callable(queue):
                     queue = queue()
                 if not is_testing():
@@ -126,5 +116,4 @@ class task:
         setattr(fn, 'name', cls.name)
         setattr(fn, 'apply_async', cls.apply_async)
         setattr(fn, 'delay', cls.delay)
-        setattr(fn, 'get_async_body', cls.get_async_body)
         return fn

awx/main/dispatch/worker/base.py

@@ -7,21 +7,18 @@ import signal
 import sys
 import redis
 import json
-import psycopg
+import psycopg2
 import time
 from uuid import UUID
 from queue import Empty as QueueEmpty
-from datetime import timedelta
 
 from django import db
 from django.conf import settings
 
 from awx.main.dispatch.pool import WorkerPool
-from awx.main.dispatch.periodic import Scheduler
 from awx.main.dispatch import pg_bus_conn
 from awx.main.utils.common import log_excess_runtime
 from awx.main.utils.db import set_connection_name
-import awx.main.analytics.subsystem_metrics as s_metrics
 
 if 'run_callback_receiver' in sys.argv:
     logger = logging.getLogger('awx.main.commands.run_callback_receiver')
@@ -66,12 +63,10 @@ class AWXConsumerBase(object):
     def control(self, body):
         logger.warning(f'Received control signal:\n{body}')
         control = body.get('control')
-        if control in ('status', 'schedule', 'running', 'cancel'):
+        if control in ('status', 'running', 'cancel'):
             reply_queue = body['reply_to']
             if control == 'status':
                 msg = '\n'.join([self.listening_on, self.pool.debug()])
-            if control == 'schedule':
-                msg = self.scheduler.debug()
             elif control == 'running':
                 msg = []
                 for worker in self.pool.workers:
@@ -97,11 +92,16 @@ class AWXConsumerBase(object):
         else:
             logger.error('unrecognized control message: {}'.format(control))
 
-    def dispatch_task(self, body):
-        """This will place the given body into a worker queue to run method decorated as a task"""
+    def process_task(self, body):
         if isinstance(body, dict):
             body['time_ack'] = time.time()
 
+        if 'control' in body:
+            try:
+                return self.control(body)
+            except Exception:
+                logger.exception(f"Exception handling control message: {body}")
+                return
         if len(self.pool):
             if "uuid" in body and body['uuid']:
                 try:
@@ -115,24 +115,15 @@ class AWXConsumerBase(object):
         self.pool.write(queue, body)
         self.total_messages += 1
 
-    def process_task(self, body):
-        """Routes the task details in body as either a control task or a task-task"""
-        if 'control' in body:
-            try:
-                return self.control(body)
-            except Exception:
-                logger.exception(f"Exception handling control message: {body}")
-                return
-        self.dispatch_task(body)
-
     @log_excess_runtime(logger)
     def record_statistics(self):
         if time.time() - self.last_stats > 1:  # buffer stat recording to once per second
             try:
                 self.redis.set(f'awx_{self.name}_statistics', self.pool.debug())
+                self.last_stats = time.time()
             except Exception:
                 logger.exception(f"encountered an error communicating with redis to store {self.name} statistics")
-            self.last_stats = time.time()
+                self.last_stats = time.time()
 
     def run(self, *args, **kwargs):
         signal.signal(signal.SIGINT, self.stop)
@@ -151,72 +142,29 @@ class AWXConsumerRedis(AWXConsumerBase):
     def run(self, *args, **kwargs):
         super(AWXConsumerRedis, self).run(*args, **kwargs)
         self.worker.on_start()
-        logger.info(f'Callback receiver started with pid={os.getpid()}')
-        db.connection.close()  # logs use database, so close connection
 
         while True:
+            logger.debug(f'{os.getpid()} is alive')
             time.sleep(60)
 
 
 class AWXConsumerPG(AWXConsumerBase):
-    def __init__(self, *args, schedule=None, **kwargs):
+    def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
-        self.pg_max_wait = settings.DISPATCHER_DB_DOWNTIME_TOLERANCE
+        self.pg_max_wait = settings.DISPATCHER_DB_DOWNTOWN_TOLLERANCE
         # if no successful loops have ran since startup, then we should fail right away
         self.pg_is_down = True  # set so that we fail if we get database errors on startup
-        init_time = time.time()
-        self.pg_down_time = init_time - self.pg_max_wait  # allow no grace period
-        self.last_cleanup = init_time
-        self.subsystem_metrics = s_metrics.Metrics(auto_pipe_execute=False)
-        self.last_metrics_gather = init_time
-        self.listen_cumulative_time = 0.0
-        if schedule:
-            schedule = schedule.copy()
-        else:
-            schedule = {}
-        # add control tasks to be ran at regular schedules
-        # NOTE: if we run out of database connections, it is important to still run cleanup
-        # so that we scale down workers and free up connections
-        schedule['pool_cleanup'] = {'control': self.pool.cleanup, 'schedule': timedelta(seconds=60)}
-        # record subsystem metrics for the dispatcher
-        schedule['metrics_gather'] = {'control': self.record_metrics, 'schedule': timedelta(seconds=20)}
-        self.scheduler = Scheduler(schedule)
-
-    def record_metrics(self):
-        current_time = time.time()
-        self.pool.produce_subsystem_metrics(self.subsystem_metrics)
-        self.subsystem_metrics.set('dispatcher_availability', self.listen_cumulative_time / (current_time - self.last_metrics_gather))
-        self.subsystem_metrics.pipe_execute()
-        self.listen_cumulative_time = 0.0
-        self.last_metrics_gather = current_time
+        self.pg_down_time = time.time() - self.pg_max_wait  # allow no grace period
+        self.last_cleanup = time.time()
 
     def run_periodic_tasks(self):
-        """
-        Run general periodic logic, and return maximum time in seconds before
-        the next requested run
-        This may be called more often than that when events are consumed
-        so this should be very efficient in that
-        """
-        try:
-            self.record_statistics()  # maintains time buffer in method
-        except Exception as exc:
-            logger.warning(f'Failed to save dispatcher statistics {exc}')
+        self.record_statistics()  # maintains time buffer in method
 
-        for job in self.scheduler.get_and_mark_pending():
-            if 'control' in job.data:
-                try:
-                    job.data['control']()
-                except Exception:
-                    logger.exception(f'Error running control task {job.data}')
-            elif 'task' in job.data:
-                body = self.worker.resolve_callable(job.data['task']).get_async_body()
-                # bypasses pg_notify for scheduled tasks
-                self.dispatch_task(body)
-
-        self.pg_is_down = False
-        self.listen_start = time.time()
-
-        return self.scheduler.time_until_next_run()
+        if time.time() - self.last_cleanup > 60:  # same as cluster_node_heartbeat
+            # NOTE: if we run out of database connections, it is important to still run cleanup
+            # so that we scale down workers and free up connections
+            self.pool.cleanup()
+            self.last_cleanup = time.time()
 
     def run(self, *args, **kwargs):
         super(AWXConsumerPG, self).run(*args, **kwargs)
@@ -232,21 +180,17 @@ class AWXConsumerPG(AWXConsumerBase):
                     if init is False:
                         self.worker.on_start()
                         init = True
-                    # run_periodic_tasks run scheduled actions and gives time until next scheduled action
-                    # this is saved to the conn (PubSub) object in order to modify read timeout in-loop
-                    conn.select_timeout = self.run_periodic_tasks()
-                    # this is the main operational loop for awx-manage run_dispatcher
                     for e in conn.events(yield_timeouts=True):
-                        self.listen_cumulative_time += time.time() - self.listen_start  # for metrics
                         if e is not None:
                             self.process_task(json.loads(e.payload))
-                        conn.select_timeout = self.run_periodic_tasks()
+                        self.run_periodic_tasks()
+                        self.pg_is_down = False
                     if self.should_stop:
                         return
-            except psycopg.InterfaceError:
+            except psycopg2.InterfaceError:
                 logger.warning("Stale Postgres message bus connection, reconnecting")
                 continue
-            except (db.DatabaseError, psycopg.OperationalError):
+            except (db.DatabaseError, psycopg2.OperationalError):
                 # If we have attained stady state operation, tolerate short-term database hickups
                 if not self.pg_is_down:
                     logger.exception(f"Error consuming new events from postgres, will retry for {self.pg_max_wait} s")
@@ -288,8 +232,8 @@ class BaseWorker(object):
                     break
             except QueueEmpty:
                 continue
-            except Exception:
-                logger.exception("Exception on worker {}, reconnecting: ".format(idx))
+            except Exception as e:
+                logger.error("Exception on worker {}, restarting: ".format(idx) + str(e))
                 continue
             try:
                 for conn in db.connections.all():

awx/main/dispatch/worker/callback.py

@@ -9,6 +9,7 @@ from django.conf import settings
 from django.utils.functional import cached_property
 from django.utils.timezone import now as tz_now
 from django.db import transaction, connection as django_connection
+from django.db.utils import DataError
 from django_guid import set_guid
 
 import psutil
@@ -191,12 +192,16 @@ class CallbackBrokerWorker(BaseWorker):
                             e._retry_count = retry_count
 
                             # special sanitization logic for postgres treatment of NUL 0x00 char
-                            # This used to check the class of the exception but on the postgres3 upgrade it could appear
-                            #   as either DataError or ValueError, so now lets just try if its there.
-                            if (retry_count == 1) and ("\x00" in e.stdout):
-                                e.stdout = e.stdout.replace("\x00", "")
-
-                            if retry_count >= self.INDIVIDUAL_EVENT_RETRIES:
+                            if (retry_count == 1) and isinstance(exc_indv, DataError):
+                                # The easiest place is in stdout. This raises as an error stating that it can't save a NUL character
+                                if "\x00" in e.stdout:
+                                    e.stdout = e.stdout.replace("\x00", "")
+                                # There is also a chance that NUL char is embedded in event data which is part of a JSON blob. In that case we, thankfully, get a different exception
+                                if 'unsupported Unicode escape sequence' in str(exc_indv):
+                                    e.event_data = json.loads(
+                                        json.dumps(e.event_data).replace("\x00", "").replace("\\x00", "").replace("\u0000", "").replace("\\u0000", "")
+                                    )
+                            elif retry_count >= self.INDIVIDUAL_EVENT_RETRIES:
                                 logger.error(f'Hit max retries ({retry_count}) saving individual Event error: {str(exc_indv)}\ndata:\n{e.__dict__}')
                                 events.remove(e)
                             else:

awx/main/fields.py

@@ -67,60 +67,10 @@ def __enum_validate__(validator, enums, instance, schema):
 Draft4Validator.VALIDATORS['enum'] = __enum_validate__
 
 
-import logging
-
-logger = logging.getLogger('awx.main.fields')
-
-
 class JSONBlob(JSONField):
-    # Cringe... a JSONField that is back ended with a TextField.
-    # This field was a legacy custom field type that tl;dr; was a TextField
-    # Over the years, with Django upgrades, we were able to go to a JSONField instead of the custom field
-    # However, we didn't want to have large customers with millions of events to update from text to json during an upgrade
-    # So we keep this field type as backended with TextField.
     def get_internal_type(self):
         return "TextField"
 
-    # postgres uses a Jsonb field as the default backend
-    # with psycopg2 it was using a psycopg2._json.Json class internally
-    # with psycopg3 it uses a psycopg.types.json.Jsonb class internally
-    # The binary class was not compatible with a text field, so we are going to override these next two methods and ensure we are using a string
-
-    def from_db_value(self, value, expression, connection):
-        if value is None:
-            return value
-
-        if isinstance(value, str):
-            try:
-                return json.loads(value)
-            except Exception as e:
-                logger.error(f"Failed to load JSONField {self.name}: {e}")
-
-        return value
-
-    def get_db_prep_value(self, value, connection, prepared=False):
-        if not prepared:
-            value = self.get_prep_value(value)
-        try:
-            # Null characters are not allowed in text fields and JSONBlobs are JSON data but saved as text
-            # So we want to make sure we strip out any null characters also note, these "should" be escaped by the dumps process:
-            #     >>> my_obj = { 'test': '\x00' }
-            #     >>> import json
-            #     >>> json.dumps(my_obj)
-            #     '{"test": "\\u0000"}'
-            # But just to be safe, lets remove them if they are there. \x00 and \u0000 are the same:
-            #     >>> string = "\x00"
-            #     >>> "\u0000" in string
-            #     True
-            dumped_value = json.dumps(value)
-            if "\x00" in dumped_value:
-                dumped_value = dumped_value.replace("\x00", '')
-            return dumped_value
-        except Exception as e:
-            logger.error(f"Failed to dump JSONField {self.name}: {e} value: {value}")
-
-        return value
-
 
 # Based on AutoOneToOneField from django-annoying:
 # https://bitbucket.org/offline/django-annoying/src/a0de8b294db3/annoying/fields.py
@@ -850,7 +800,7 @@ class CredentialTypeInjectorField(JSONSchemaField):
     def validate_env_var_allowed(self, env_var):
         if env_var.startswith('ANSIBLE_'):
             raise django_exceptions.ValidationError(
-                _('Environment variable {} may affect Ansible configuration so its use is not allowed in credentials.').format(env_var),
+                _('Environment variable {} may affect Ansible configuration so its ' 'use is not allowed in credentials.').format(env_var),
                 code='invalid',
                 params={'value': env_var},
             )

awx/main/management/commands/cleanup_activitystream.py

@@ -23,7 +23,7 @@ class Command(BaseCommand):
 
     def add_arguments(self, parser):
         parser.add_argument('--days', dest='days', type=int, default=90, metavar='N', help='Remove activity stream events more than N days old')
-        parser.add_argument('--dry-run', dest='dry_run', action='store_true', default=False, help='Dry run mode (show items that would be removed)')
+        parser.add_argument('--dry-run', dest='dry_run', action='store_true', default=False, help='Dry run mode (show items that would ' 'be removed)')
 
     def init_logging(self):
         log_levels = dict(enumerate([logging.ERROR, logging.INFO, logging.DEBUG, 0]))

awx/main/management/commands/cleanup_host_metrics.py

@@ -17,6 +17,6 @@ class Command(BaseCommand):
         months_ago = options.get('months-ago') or None
 
         if not months_ago:
-            months_ago = getattr(settings, 'CLEANUP_HOST_METRICS_SOFT_THRESHOLD', 12)
+            months_ago = getattr(settings, 'CLEANUP_HOST_METRICS_THRESHOLD', 12)
 
         HostMetric.cleanup_task(months_ago)

awx/main/management/commands/cleanup_jobs.py

@@ -17,7 +17,10 @@ from django.utils.timezone import now
 
 # AWX
 from awx.main.models import Job, AdHocCommand, ProjectUpdate, InventoryUpdate, SystemJob, WorkflowJob, Notification
-from awx.main.utils import unified_job_class_to_event_table_name
+
+
+def unified_job_class_to_event_table_name(job_class):
+    return f'main_{job_class().event_class.__name__.lower()}'
 
 
 def partition_table_name(job_class, dt):
@@ -149,7 +152,7 @@ class Command(BaseCommand):
 
     def add_arguments(self, parser):
         parser.add_argument('--days', dest='days', type=int, default=90, metavar='N', help='Remove jobs/updates executed more than N days ago. Defaults to 90.')
-        parser.add_argument('--dry-run', dest='dry_run', action='store_true', default=False, help='Dry run mode (show items that would be removed)')
+        parser.add_argument('--dry-run', dest='dry_run', action='store_true', default=False, help='Dry run mode (show items that would ' 'be removed)')
         parser.add_argument('--jobs', dest='only_jobs', action='store_true', default=False, help='Remove jobs')
         parser.add_argument('--ad-hoc-commands', dest='only_ad_hoc_commands', action='store_true', default=False, help='Remove ad hoc commands')
         parser.add_argument('--project-updates', dest='only_project_updates', action='store_true', default=False, help='Remove project updates')

awx/main/management/commands/custom_venv_associations.py

@@ -44,7 +44,7 @@ class Command(BaseCommand):
                         '- To list all (now deprecated) custom virtual environments run:',
                         'awx-manage list_custom_venvs',
                         '',
-                        '- To export the contents of a (deprecated) virtual environment, run the following command while supplying the path as an argument:',
+                        '- To export the contents of a (deprecated) virtual environment, ' 'run the following command while supplying the path as an argument:',
                         'awx-manage export_custom_venv /path/to/venv',
                         '',
                         '- Run these commands with `-q` to remove tool tips.',

awx/main/management/commands/deprovision_instance.py

@@ -13,7 +13,7 @@ class Command(BaseCommand):
     Deprovision a cluster node
     """
 
-    help = 'Remove instance from the database. Specify `--hostname` to use this command.'
+    help = 'Remove instance from the database. ' 'Specify `--hostname` to use this command.'
 
     def add_arguments(self, parser):
         parser.add_argument('--hostname', dest='hostname', type=str, help='Hostname used during provisioning')

awx/main/management/commands/host_metric_summary_monthly.py

@@ -1,9 +0,0 @@
-from django.core.management.base import BaseCommand
-from awx.main.tasks.host_metrics import HostMetricSummaryMonthlyTask
-
-
-class Command(BaseCommand):
-    help = 'Computing of HostMetricSummaryMonthly'
-
-    def handle(self, *args, **options):
-        HostMetricSummaryMonthlyTask().execute()

awx/main/management/commands/list_custom_venvs.py

@@ -22,7 +22,7 @@ class Command(BaseCommand):
                     '# Discovered Virtual Environments:',
                     '\n'.join(venvs),
                     '',
-                    '- To export the contents of a (deprecated) virtual environment, run the following command while supplying the path as an argument:',
+                    '- To export the contents of a (deprecated) virtual environment, ' 'run the following command while supplying the path as an argument:',
                     'awx-manage export_custom_venv /path/to/venv',
                     '',
                     '- To view the connections a (deprecated) virtual environment had in the database, run the following command while supplying the path as an argument:',

awx/main/management/commands/precreate_partitions.py

@@ -1,27 +0,0 @@
-from django.utils.timezone import now
-from django.core.management.base import BaseCommand, CommandParser
-from datetime import timedelta
-from awx.main.utils.common import create_partition, unified_job_class_to_event_table_name
-from awx.main.models import Job, SystemJob, ProjectUpdate, InventoryUpdate, AdHocCommand
-
-
-class Command(BaseCommand):
-    """Command used to precreate database partitions to avoid pg_dump locks"""
-
-    def add_arguments(self, parser: CommandParser) -> None:
-        parser.add_argument('--count', dest='count', action='store', help='The amount of hours of partitions to create', type=int, default=1)
-
-    def _create_partitioned_tables(self, count):
-        tables = list()
-        for model in (Job, SystemJob, ProjectUpdate, InventoryUpdate, AdHocCommand):
-            tables.append(unified_job_class_to_event_table_name(model))
-        start = now()
-        while count > 0:
-            for table in tables:
-                create_partition(table, start)
-                print(f'Created partitions for {table} {start}')
-            start = start + timedelta(hours=1)
-            count -= 1
-
-    def handle(self, **options):
-        self._create_partitioned_tables(count=options.get('count'))

awx/main/management/commands/provision_instance.py

@@ -35,7 +35,7 @@ class Command(BaseCommand):
 
             from awx.main.management.commands.register_queue import RegisterQueue
 
-            (changed, instance) = Instance.objects.register(ip_address=os.environ.get('MY_POD_IP'), node_type='control', node_uuid=settings.SYSTEM_UUID)
+            (changed, instance) = Instance.objects.register(ip_address=os.environ.get('MY_POD_IP'), node_type='control', uuid=settings.SYSTEM_UUID)
             RegisterQueue(settings.DEFAULT_CONTROL_PLANE_QUEUE_NAME, 100, 0, [], is_container_group=False).register()
             RegisterQueue(
                 settings.DEFAULT_EXECUTION_QUEUE_NAME,
@@ -48,7 +48,7 @@ class Command(BaseCommand):
                 max_concurrent_jobs=settings.DEFAULT_EXECUTION_QUEUE_MAX_CONCURRENT_JOBS,
             ).register()
         else:
-            (changed, instance) = Instance.objects.register(hostname=hostname, node_type=node_type, node_uuid=uuid)
+            (changed, instance) = Instance.objects.register(hostname=hostname, node_type=node_type, uuid=uuid)
         if changed:
             print("Successfully registered instance {}".format(hostname))
         else:

awx/main/management/commands/run_cache_clear.py

@@ -2,7 +2,6 @@ import logging
 import json
 
 from django.core.management.base import BaseCommand
-
 from awx.main.dispatch import pg_bus_conn
 from awx.main.dispatch.worker.task import TaskWorker
 
@@ -19,7 +18,7 @@ class Command(BaseCommand):
 
     def handle(self, *arg, **options):
         try:
-            with pg_bus_conn() as conn:
+            with pg_bus_conn(new_connection=True) as conn:
                 conn.listen("tower_settings_change")
                 for e in conn.events(yield_timeouts=True):
                     if e is not None:

awx/main/management/commands/run_dispatcher.py

@@ -4,22 +4,28 @@ import logging
 import yaml
 
 from django.conf import settings
+from django.core.cache import cache as django_cache
 from django.core.management.base import BaseCommand
+from django.db import connection as django_connection
 
 from awx.main.dispatch import get_task_queuename
 from awx.main.dispatch.control import Control
 from awx.main.dispatch.pool import AutoscalePool
 from awx.main.dispatch.worker import AWXConsumerPG, TaskWorker
+from awx.main.dispatch import periodic
 
 logger = logging.getLogger('awx.main.dispatch')
 
 
+def construct_bcast_queue_name(common_name):
+    return common_name + '_' + settings.CLUSTER_HOST_ID
+
+
 class Command(BaseCommand):
     help = 'Launch the task dispatcher'
 
     def add_arguments(self, parser):
         parser.add_argument('--status', dest='status', action='store_true', help='print the internal state of any running dispatchers')
-        parser.add_argument('--schedule', dest='schedule', action='store_true', help='print the current status of schedules being ran by dispatcher')
         parser.add_argument('--running', dest='running', action='store_true', help='print the UUIDs of any tasked managed by this dispatcher')
         parser.add_argument(
             '--reload',
@@ -41,9 +47,6 @@ class Command(BaseCommand):
         if options.get('status'):
             print(Control('dispatcher').status())
             return
-        if options.get('schedule'):
-            print(Control('dispatcher').schedule())
-            return
         if options.get('running'):
             print(Control('dispatcher').running())
             return
@@ -60,11 +63,21 @@ class Command(BaseCommand):
             print(Control('dispatcher').cancel(cancel_data))
             return
 
+        # It's important to close these because we're _about_ to fork, and we
+        # don't want the forked processes to inherit the open sockets
+        # for the DB and cache connections (that way lies race conditions)
+        django_connection.close()
+        django_cache.close()
+
+        # spawn a daemon thread to periodically enqueues scheduled tasks
+        # (like the node heartbeat)
+        periodic.run_continuously()
+
         consumer = None
 
         try:
             queues = ['tower_broadcast_all', 'tower_settings_change', get_task_queuename()]
-            consumer = AWXConsumerPG('dispatcher', TaskWorker(), queues, AutoscalePool(min_workers=4), schedule=settings.CELERYBEAT_SCHEDULE)
+            consumer = AWXConsumerPG('dispatcher', TaskWorker(), queues, AutoscalePool(min_workers=4))
             consumer.run()
         except KeyboardInterrupt:
             logger.debug('Terminating Task Dispatcher')

awx/main/management/commands/run_heartbeet.py

@@ -0,0 +1,74 @@
+import json
+import logging
+import os
+import time
+import signal
+import sys
+
+from django.core.management.base import BaseCommand
+from django.conf import settings
+
+from awx.main.dispatch import pg_bus_conn
+
+logger = logging.getLogger('awx.main.commands.run_heartbeet')
+
+
+class Command(BaseCommand):
+    help = 'Launch the web server beacon (heartbeet)'
+
+    def print_banner(self):
+        heartbeet = r"""
+   **********   **********
+ ************* *************
+*****************************
+ ***********HEART***********
+  *************************
+     *******************
+       *************** _._
+         *********** /`._ `'.      __
+           *******   \  .\| \   _'`  `)
+             ***  (``_)  \| ).'` /`- /
+              *   `\ `;\_ `\\//`-'` /
+                    \ `'.'.| /  __/`
+                     `'--v_|/`'`
+                         __||-._
+                      /'` `-``  `'\\
+                     /         .'` )
+                     \  BEET  '    )
+                      \.          /
+                        '.     /'`
+                           `) |
+                            //
+                            '(.
+                             `\`.
+                               ``"""
+        print(heartbeet)
+
+    def construct_payload(self, action='online'):
+        payload = {
+            'hostname': settings.CLUSTER_HOST_ID,
+            'ip': os.environ.get('MY_POD_IP'),
+            'action': action,
+        }
+        return json.dumps(payload)
+
+    def notify_listener_and_exit(self, *args):
+        with pg_bus_conn(new_connection=False) as conn:
+            conn.notify('web_heartbeet', self.construct_payload(action='offline'))
+        sys.exit(0)
+
+    def do_hearbeat_loop(self):
+        with pg_bus_conn(new_connection=True) as conn:
+            while True:
+                logger.debug('Sending heartbeat')
+                conn.notify('web_heartbeet', self.construct_payload())
+                time.sleep(settings.BROADCAST_WEBSOCKET_BEACON_FROM_WEB_RATE_SECONDS)
+
+    def handle(self, *arg, **options):
+        self.print_banner()
+        signal.signal(signal.SIGTERM, self.notify_listener_and_exit)
+        signal.signal(signal.SIGINT, self.notify_listener_and_exit)
+
+        # Note: We don't really try any reconnect logic to pg_notify here,
+        # just let supervisor restart if we fail.
+        self.do_hearbeat_loop()

awx/main/management/commands/run_rsyslog_configurer.py

@@ -22,7 +22,7 @@ class Command(BaseCommand):
 
     def handle(self, *arg, **options):
         try:
-            with pg_bus_conn() as conn:
+            with pg_bus_conn(new_connection=True) as conn:
                 conn.listen("rsyslog_configurer")
                 # reconfigure rsyslog on start up
                 reconfigure_rsyslog()

awx/main/management/commands/run_ws_heartbeat.py

@@ -1,45 +0,0 @@
-import json
-import logging
-import os
-import time
-import signal
-import sys
-
-from django.core.management.base import BaseCommand
-from django.conf import settings
-
-from awx.main.dispatch import pg_bus_conn
-
-logger = logging.getLogger('awx.main.commands.run_ws_heartbeat')
-
-
-class Command(BaseCommand):
-    help = 'Launch the web server beacon (ws_heartbeat)'
-
-    def construct_payload(self, action='online'):
-        payload = {
-            'hostname': settings.CLUSTER_HOST_ID,
-            'ip': os.environ.get('MY_POD_IP'),
-            'action': action,
-        }
-        return json.dumps(payload)
-
-    def notify_listener_and_exit(self, *args):
-        with pg_bus_conn(new_connection=False) as conn:
-            conn.notify('web_ws_heartbeat', self.construct_payload(action='offline'))
-        sys.exit(0)
-
-    def do_heartbeat_loop(self):
-        while True:
-            with pg_bus_conn() as conn:
-                logger.debug('Sending heartbeat')
-                conn.notify('web_ws_heartbeat', self.construct_payload())
-            time.sleep(settings.BROADCAST_WEBSOCKET_BEACON_FROM_WEB_RATE_SECONDS)
-
-    def handle(self, *arg, **options):
-        signal.signal(signal.SIGTERM, self.notify_listener_and_exit)
-        signal.signal(signal.SIGINT, self.notify_listener_and_exit)
-
-        # Note: We don't really try any reconnect logic to pg_notify here,
-        # just let supervisor restart if we fail.
-        self.do_heartbeat_loop()

awx/main/managers.py

@@ -2,7 +2,6 @@
 # All Rights Reserved.
 
 import logging
-import uuid
 from django.db import models
 from django.conf import settings
 from django.db.models.functions import Lower
@@ -115,7 +114,7 @@ class InstanceManager(models.Manager):
             return node[0]
         raise RuntimeError("No instance found with the current cluster host id")
 
-    def register(self, node_uuid=None, hostname=None, ip_address=None, node_type='hybrid', defaults=None):
+    def register(self, uuid=None, hostname=None, ip_address=None, node_type='hybrid', defaults=None):
         if not hostname:
             hostname = settings.CLUSTER_HOST_ID
 
@@ -132,8 +131,8 @@ class InstanceManager(models.Manager):
                         logger.warning("IP address {0} conflict detected, ip address unset for host {1}.".format(ip_address, other_hostname))
 
             # Return existing instance that matches hostname or UUID (default to UUID)
-            if node_uuid is not None and node_uuid != UUID_DEFAULT and self.filter(uuid=node_uuid).exists():
-                instance = self.filter(uuid=node_uuid)
+            if uuid is not None and uuid != UUID_DEFAULT and self.filter(uuid=uuid).exists():
+                instance = self.filter(uuid=uuid)
             else:
                 # if instance was not retrieved by uuid and hostname was, use the hostname
                 instance = self.filter(hostname=hostname)
@@ -171,7 +170,9 @@ class InstanceManager(models.Manager):
             }
             if defaults is not None:
                 create_defaults.update(defaults)
-            uuid_option = {'uuid': node_uuid if node_uuid is not None else uuid.uuid4()}
+            uuid_option = {}
+            if uuid is not None:
+                uuid_option = {'uuid': uuid}
             if node_type == 'execution' and 'version' not in create_defaults:
                 create_defaults['version'] = RECEPTOR_PENDING
             instance = self.create(hostname=hostname, ip_address=ip_address, node_type=node_type, **create_defaults, **uuid_option)

awx/main/middleware.py

@@ -122,7 +122,7 @@ class URLModificationMiddleware(MiddlewareMixin):
             field_class=fields.DictField,
             read_only=True,
             label=_('Formats of all available named urls'),
-            help_text=_('Read-only list of key-value pairs that shows the standard format of all available named URLs.'),
+            help_text=_('Read-only list of key-value pairs that shows the standard format of all ' 'available named URLs.'),
             category=_('Named URL'),
             category_slug='named-url',
         )

awx/main/migrations/0001_initial.py

@@ -9,11 +9,13 @@ from django.db import migrations, models
 import django.utils.timezone
 import django.db.models.deletion
 from django.conf import settings
+import taggit.managers
 import awx.main.fields
 
 
 class Migration(migrations.Migration):
     dependencies = [
+        ('taggit', '0002_auto_20150616_2121'),
         ('contenttypes', '0002_remove_content_type_name'),
         migrations.swappable_dependency(settings.AUTH_USER_MODEL),
     ]
@@ -182,6 +184,12 @@ class Migration(migrations.Migration):
                         null=True,
                     ),
                 ),
+                (
+                    'tags',
+                    taggit.managers.TaggableManager(
+                        to='taggit.Tag', through='taggit.TaggedItem', blank=True, help_text='A comma-separated list of tags.', verbose_name='Tags'
+                    ),
+                ),
             ],
             options={
                 'ordering': ('kind', 'name'),
@@ -521,6 +529,12 @@ class Migration(migrations.Migration):
                         null=True,
                     ),
                 ),
+                (
+                    'tags',
+                    taggit.managers.TaggableManager(
+                        to='taggit.Tag', through='taggit.TaggedItem', blank=True, help_text='A comma-separated list of tags.', verbose_name='Tags'
+                    ),
+                ),
                 ('users', models.ManyToManyField(related_name='organizations', to=settings.AUTH_USER_MODEL, blank=True)),
             ],
             options={
@@ -575,6 +589,12 @@ class Migration(migrations.Migration):
                         null=True,
                     ),
                 ),
+                (
+                    'tags',
+                    taggit.managers.TaggableManager(
+                        to='taggit.Tag', through='taggit.TaggedItem', blank=True, help_text='A comma-separated list of tags.', verbose_name='Tags'
+                    ),
+                ),
             ],
         ),
         migrations.CreateModel(
@@ -624,6 +644,12 @@ class Migration(migrations.Migration):
                         null=True,
                     ),
                 ),
+                (
+                    'tags',
+                    taggit.managers.TaggableManager(
+                        to='taggit.Tag', through='taggit.TaggedItem', blank=True, help_text='A comma-separated list of tags.', verbose_name='Tags'
+                    ),
+                ),
             ],
             options={
                 'ordering': ['-next_run'],
@@ -661,6 +687,12 @@ class Migration(migrations.Migration):
                     ),
                 ),
                 ('organization', models.ForeignKey(related_name='teams', on_delete=django.db.models.deletion.SET_NULL, to='main.Organization', null=True)),
+                (
+                    'tags',
+                    taggit.managers.TaggableManager(
+                        to='taggit.Tag', through='taggit.TaggedItem', blank=True, help_text='A comma-separated list of tags.', verbose_name='Tags'
+                    ),
+                ),
                 ('users', models.ManyToManyField(related_name='teams', to=settings.AUTH_USER_MODEL, blank=True)),
             ],
             options={
@@ -1235,6 +1267,13 @@ class Migration(migrations.Migration):
                 null=True,
             ),
         ),
+        migrations.AddField(
+            model_name='unifiedjobtemplate',
+            name='tags',
+            field=taggit.managers.TaggableManager(
+                to='taggit.Tag', through='taggit.TaggedItem', blank=True, help_text='A comma-separated list of tags.', verbose_name='Tags'
+            ),
+        ),
         migrations.AddField(
             model_name='unifiedjob',
             name='created_by',
@@ -1280,6 +1319,13 @@ class Migration(migrations.Migration):
             name='schedule',
             field=models.ForeignKey(on_delete=django.db.models.deletion.SET_NULL, default=None, editable=False, to='main.Schedule', null=True),
         ),
+        migrations.AddField(
+            model_name='unifiedjob',
+            name='tags',
+            field=taggit.managers.TaggableManager(
+                to='taggit.Tag', through='taggit.TaggedItem', blank=True, help_text='A comma-separated list of tags.', verbose_name='Tags'
+            ),
+        ),
         migrations.AddField(
             model_name='unifiedjob',
             name='unified_job_template',
@@ -1324,6 +1370,13 @@ class Migration(migrations.Migration):
                 help_text='Organization containing this inventory.',
             ),
         ),
+        migrations.AddField(
+            model_name='inventory',
+            name='tags',
+            field=taggit.managers.TaggableManager(
+                to='taggit.Tag', through='taggit.TaggedItem', blank=True, help_text='A comma-separated list of tags.', verbose_name='Tags'
+            ),
+        ),
         migrations.AddField(
             model_name='host',
             name='inventory',
@@ -1354,6 +1407,13 @@ class Migration(migrations.Migration):
                 null=True,
             ),
         ),
+        migrations.AddField(
+            model_name='host',
+            name='tags',
+            field=taggit.managers.TaggableManager(
+                to='taggit.Tag', through='taggit.TaggedItem', blank=True, help_text='A comma-separated list of tags.', verbose_name='Tags'
+            ),
+        ),
         migrations.AddField(
             model_name='group',
             name='hosts',
@@ -1381,6 +1441,13 @@ class Migration(migrations.Migration):
             name='parents',
             field=models.ManyToManyField(related_name='children', to='main.Group', blank=True),
         ),
+        migrations.AddField(
+            model_name='group',
+            name='tags',
+            field=taggit.managers.TaggableManager(
+                to='taggit.Tag', through='taggit.TaggedItem', blank=True, help_text='A comma-separated list of tags.', verbose_name='Tags'
+            ),
+        ),
         migrations.AddField(
             model_name='custominventoryscript',
             name='organization',
@@ -1392,6 +1459,13 @@ class Migration(migrations.Migration):
                 null=True,
             ),
         ),
+        migrations.AddField(
+            model_name='custominventoryscript',
+            name='tags',
+            field=taggit.managers.TaggableManager(
+                to='taggit.Tag', through='taggit.TaggedItem', blank=True, help_text='A comma-separated list of tags.', verbose_name='Tags'
+            ),
+        ),
         migrations.AddField(
             model_name='credential',
             name='team',

awx/main/migrations/0002_squashed_v300_release.py

@@ -12,6 +12,8 @@ import django.db.models.deletion
 from django.conf import settings
 from django.utils.timezone import now
 
+import taggit.managers
+
 
 def create_system_job_templates(apps, schema_editor):
     """
@@ -123,6 +125,7 @@ class Migration(migrations.Migration):
     ]
 
     dependencies = [
+        ('taggit', '0002_auto_20150616_2121'),
         ('contenttypes', '0002_remove_content_type_name'),
         migrations.swappable_dependency(settings.AUTH_USER_MODEL),
         ('main', '0001_initial'),
@@ -253,6 +256,12 @@ class Migration(migrations.Migration):
                     'organization',
                     models.ForeignKey(related_name='notification_templates', on_delete=django.db.models.deletion.SET_NULL, to='main.Organization', null=True),
                 ),
+                (
+                    'tags',
+                    taggit.managers.TaggableManager(
+                        to='taggit.Tag', through='taggit.TaggedItem', blank=True, help_text='A comma-separated list of tags.', verbose_name='Tags'
+                    ),
+                ),
             ],
         ),
         migrations.AddField(
@@ -712,6 +721,12 @@ class Migration(migrations.Migration):
                         help_text='Organization this label belongs to.',
                     ),
                 ),
+                (
+                    'tags',
+                    taggit.managers.TaggableManager(
+                        to='taggit.Tag', through='taggit.TaggedItem', blank=True, help_text='A comma-separated list of tags.', verbose_name='Tags'
+                    ),
+                ),
             ],
             options={
                 'ordering': ('organization', 'name'),

awx/main/migrations/0006_v320_release.py

@@ -5,6 +5,7 @@ from __future__ import unicode_literals
 # Django
 from django.db import connection, migrations, models, OperationalError, ProgrammingError
 from django.conf import settings
+import taggit.managers
 
 # AWX
 import awx.main.fields
@@ -316,6 +317,10 @@ class Migration(migrations.Migration):
             model_name='permission',
             name='project',
         ),
+        migrations.RemoveField(
+            model_name='permission',
+            name='tags',
+        ),
         migrations.RemoveField(
             model_name='permission',
             name='team',
@@ -505,6 +510,12 @@ class Migration(migrations.Migration):
                         null=True,
                     ),
                 ),
+                (
+                    'tags',
+                    taggit.managers.TaggableManager(
+                        to='taggit.Tag', through='taggit.TaggedItem', blank=True, help_text='A comma-separated list of tags.', verbose_name='Tags'
+                    ),
+                ),
             ],
             options={
                 'ordering': ('kind', 'name'),

awx/main/migrations/0067_v350_credential_plugins.py

@@ -4,6 +4,7 @@ from __future__ import unicode_literals
 from django.conf import settings
 from django.db import migrations, models
 import django.db.models.deletion
+import taggit.managers
 
 # AWX
 import awx.main.fields
@@ -19,6 +20,7 @@ def setup_tower_managed_defaults(apps, schema_editor):
 class Migration(migrations.Migration):
     dependencies = [
         migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+        ('taggit', '0002_auto_20150616_2121'),
         ('main', '0066_v350_inventorysource_custom_virtualenv'),
     ]
 
@@ -58,6 +60,12 @@ class Migration(migrations.Migration):
                     'source_credential',
                     models.ForeignKey(null=True, on_delete=django.db.models.deletion.CASCADE, related_name='target_input_sources', to='main.Credential'),
                 ),
+                (
+                    'tags',
+                    taggit.managers.TaggableManager(
+                        blank=True, help_text='A comma-separated list of tags.', through='taggit.TaggedItem', to='taggit.Tag', verbose_name='Tags'
+                    ),
+                ),
                 (
                     'target_credential',
                     models.ForeignKey(null=True, on_delete=django.db.models.deletion.CASCADE, related_name='input_sources', to='main.Credential'),

awx/main/migrations/0113_v370_event_bigint.py

@@ -12,17 +12,20 @@ def migrate_event_data(apps, schema_editor):
     # https://www.postgresql.org/docs/9.1/datatype-numeric.html)
     for tblname in ('main_jobevent', 'main_inventoryupdateevent', 'main_projectupdateevent', 'main_adhoccommandevent', 'main_systemjobevent'):
         with connection.cursor() as cursor:
-            # This loop used to do roughly the following:
-            #     Rename the table to _old_<tablename>
-            #     Create a new table form the old table (it would have no rows)
-            #     Drop the old sequnce and create a new on tied to the new table and set the sequence to the last number from the old table
-            # This used to work with postgres spitting out a NOTICE and DETAIL
-            # With the django 4.2 upgrade that changed to an ERROR and HINT
-            # By the time we hit the 4.2 upgrade, no one should be upgrading a database this old directly to this new schema
-            # So we no longer really care about having to do all of this work, we only need a table with a bigint ID field
-            # And this can be achieved by just changing the id column type...
+            # rename the current event table
+            cursor.execute(f'ALTER TABLE {tblname} RENAME TO _old_{tblname};')
+            # create a *new* table with the same schema
+            cursor.execute(f'CREATE TABLE {tblname} (LIKE _old_{tblname} INCLUDING ALL);')
+            # alter the *new* table so that the primary key is a big int
             cursor.execute(f'ALTER TABLE {tblname} ALTER COLUMN id TYPE bigint USING id::bigint;')
 
+            # recreate counter for the new table's primary key to
+            # start where the *old* table left off (we have to do this because the
+            # counter changed from an int to a bigint)
+            cursor.execute(f'CREATE SEQUENCE IF NOT EXISTS "{tblname}_id_seq";')
+            cursor.execute(f"SELECT setval('{tblname}_id_seq', COALESCE((SELECT MAX(id)+1 FROM _old_{tblname}), 1), false);")
+            cursor.execute(f'DROP TABLE _old_{tblname};')
+
 
 class FakeAlterField(migrations.AlterField):
     def database_forwards(self, *args):

awx/main/migrations/0124_execution_environments.py

@@ -4,10 +4,12 @@ from django.conf import settings
 from django.db import migrations, models
 import django.db.models.deletion
 import django.db.models.expressions
+import taggit.managers
 
 
 class Migration(migrations.Migration):
     dependencies = [
+        ('taggit', '0003_taggeditem_add_unique_index'),
         migrations.swappable_dependency(settings.AUTH_USER_MODEL),
         ('main', '0123_drop_hg_support'),
     ]
@@ -67,6 +69,12 @@ class Migration(migrations.Migration):
                         to='main.Organization',
                     ),
                 ),
+                (
+                    'tags',
+                    taggit.managers.TaggableManager(
+                        blank=True, help_text='A comma-separated list of tags.', through='taggit.TaggedItem', to='taggit.Tag', verbose_name='Tags'
+                    ),
+                ),
             ],
             options={
                 'ordering': (django.db.models.expressions.OrderBy(django.db.models.expressions.F('organization_id'), nulls_first=True), 'image'),

awx/main/migrations/0144_event_partitions.py

@@ -30,7 +30,7 @@ def migrate_event_data(apps, schema_editor):
             # otherwise, the schema changes we would make on the old jobevents table
             # (namely, dropping the primary key constraint) would cause the migration
             # to suffer a serious performance degradation
-            cursor.execute(f'CREATE TABLE tmp_{tblname} (LIKE _unpartitioned_{tblname} INCLUDING ALL)')
+            cursor.execute(f'CREATE TABLE tmp_{tblname} ' f'(LIKE _unpartitioned_{tblname} INCLUDING ALL)')
 
             # drop primary key constraint; in a partioned table
             # constraints must include the partition key itself
@@ -48,7 +48,7 @@ def migrate_event_data(apps, schema_editor):
             cursor.execute(f'DROP TABLE tmp_{tblname}')
 
             # recreate primary key constraint
-            cursor.execute(f'ALTER TABLE ONLY {tblname} ADD CONSTRAINT {tblname}_pkey_new PRIMARY KEY (id, job_created);')
+            cursor.execute(f'ALTER TABLE ONLY {tblname} ' f'ADD CONSTRAINT {tblname}_pkey_new PRIMARY KEY (id, job_created);')
 
     with connection.cursor() as cursor:
         """

awx/main/migrations/0184_django_upgrade.py

@@ -1,4 +1,4 @@
-# Generated by Django 4.2 on 2023-05-09 19:02
+# Generated by Django 4.2 on 2023-04-21 14:43
 
 import awx.main.fields
 import awx.main.utils.polymorphic
@@ -9,102 +9,12 @@ import django.db.models.deletion
 
 class Migration(migrations.Migration):
     dependencies = [
-        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
         ('contenttypes', '0002_remove_content_type_name'),
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
         ('main', '0183_pre_django_upgrade'),
     ]
 
     operations = [
-        migrations.RenameIndex(
-            model_name='adhoccommandevent',
-            new_name='main_adhocc_ad_hoc__1e4d24_idx',
-            old_fields=('ad_hoc_command', 'job_created', 'uuid'),
-        ),
-        migrations.RenameIndex(
-            model_name='adhoccommandevent',
-            new_name='main_adhocc_ad_hoc__e72142_idx',
-            old_fields=('ad_hoc_command', 'job_created', 'event'),
-        ),
-        migrations.RenameIndex(
-            model_name='adhoccommandevent',
-            new_name='main_adhocc_ad_hoc__a57777_idx',
-            old_fields=('ad_hoc_command', 'job_created', 'counter'),
-        ),
-        migrations.RenameIndex(
-            model_name='inventoryupdateevent',
-            new_name='main_invent_invento_f72b21_idx',
-            old_fields=('inventory_update', 'job_created', 'uuid'),
-        ),
-        migrations.RenameIndex(
-            model_name='inventoryupdateevent',
-            new_name='main_invent_invento_364dcb_idx',
-            old_fields=('inventory_update', 'job_created', 'counter'),
-        ),
-        migrations.RenameIndex(
-            model_name='jobevent',
-            new_name='main_jobeve_job_id_40a56d_idx',
-            old_fields=('job', 'job_created', 'parent_uuid'),
-        ),
-        migrations.RenameIndex(
-            model_name='jobevent',
-            new_name='main_jobeve_job_id_3c4a4a_idx',
-            old_fields=('job', 'job_created', 'uuid'),
-        ),
-        migrations.RenameIndex(
-            model_name='jobevent',
-            new_name='main_jobeve_job_id_51c382_idx',
-            old_fields=('job', 'job_created', 'counter'),
-        ),
-        migrations.RenameIndex(
-            model_name='jobevent',
-            new_name='main_jobeve_job_id_0ddc6b_idx',
-            old_fields=('job', 'job_created', 'event'),
-        ),
-        migrations.RenameIndex(
-            model_name='projectupdateevent',
-            new_name='main_projec_project_449bbd_idx',
-            old_fields=('project_update', 'job_created', 'uuid'),
-        ),
-        migrations.RenameIndex(
-            model_name='projectupdateevent',
-            new_name='main_projec_project_69559a_idx',
-            old_fields=('project_update', 'job_created', 'counter'),
-        ),
-        migrations.RenameIndex(
-            model_name='projectupdateevent',
-            new_name='main_projec_project_c44b7c_idx',
-            old_fields=('project_update', 'job_created', 'event'),
-        ),
-        migrations.RenameIndex(
-            model_name='role',
-            new_name='main_rbac_r_content_979bdd_idx',
-            old_fields=('content_type', 'object_id'),
-        ),
-        migrations.RenameIndex(
-            model_name='roleancestorentry',
-            new_name='main_rbac_r_ancesto_b44606_idx',
-            old_fields=('ancestor', 'content_type_id', 'role_field'),
-        ),
-        migrations.RenameIndex(
-            model_name='roleancestorentry',
-            new_name='main_rbac_r_ancesto_22b9f0_idx',
-            old_fields=('ancestor', 'content_type_id', 'object_id'),
-        ),
-        migrations.RenameIndex(
-            model_name='roleancestorentry',
-            new_name='main_rbac_r_ancesto_c87b87_idx',
-            old_fields=('ancestor', 'descendent'),
-        ),
-        migrations.RenameIndex(
-            model_name='systemjobevent',
-            new_name='main_system_system__e39825_idx',
-            old_fields=('system_job', 'job_created', 'uuid'),
-        ),
-        migrations.RenameIndex(
-            model_name='systemjobevent',
-            new_name='main_system_system__73537a_idx',
-            old_fields=('system_job', 'job_created', 'counter'),
-        ),
         migrations.AlterField(
             model_name='activitystream',
             name='unified_job',

awx/main/migrations/0185_djanog_indexes.py

@@ -0,0 +1,102 @@
+# Generated by Django 4.2 on 2023-04-28 19:21
+
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ('main', '0184_django_upgrade'),
+    ]
+
+    operations = [
+        migrations.RenameIndex(
+            model_name='adhoccommandevent',
+            new_name='main_adhocc_ad_hoc__a57777_idx',
+            old_fields=('ad_hoc_command', 'job_created', 'counter'),
+        ),
+        migrations.RenameIndex(
+            model_name='adhoccommandevent',
+            new_name='main_adhocc_ad_hoc__e72142_idx',
+            old_fields=('ad_hoc_command', 'job_created', 'event'),
+        ),
+        migrations.RenameIndex(
+            model_name='adhoccommandevent',
+            new_name='main_adhocc_ad_hoc__1e4d24_idx',
+            old_fields=('ad_hoc_command', 'job_created', 'uuid'),
+        ),
+        migrations.RenameIndex(
+            model_name='inventoryupdateevent',
+            new_name='main_invent_invento_f72b21_idx',
+            old_fields=('inventory_update', 'job_created', 'uuid'),
+        ),
+        migrations.RenameIndex(
+            model_name='inventoryupdateevent',
+            new_name='main_invent_invento_364dcb_idx',
+            old_fields=('inventory_update', 'job_created', 'counter'),
+        ),
+        migrations.RenameIndex(
+            model_name='jobevent',
+            new_name='main_jobeve_job_id_51c382_idx',
+            old_fields=('job', 'job_created', 'counter'),
+        ),
+        migrations.RenameIndex(
+            model_name='jobevent',
+            new_name='main_jobeve_job_id_0ddc6b_idx',
+            old_fields=('job', 'job_created', 'event'),
+        ),
+        migrations.RenameIndex(
+            model_name='jobevent',
+            new_name='main_jobeve_job_id_40a56d_idx',
+            old_fields=('job', 'job_created', 'parent_uuid'),
+        ),
+        migrations.RenameIndex(
+            model_name='jobevent',
+            new_name='main_jobeve_job_id_3c4a4a_idx',
+            old_fields=('job', 'job_created', 'uuid'),
+        ),
+        migrations.RenameIndex(
+            model_name='projectupdateevent',
+            new_name='main_projec_project_c44b7c_idx',
+            old_fields=('project_update', 'job_created', 'event'),
+        ),
+        migrations.RenameIndex(
+            model_name='projectupdateevent',
+            new_name='main_projec_project_449bbd_idx',
+            old_fields=('project_update', 'job_created', 'uuid'),
+        ),
+        migrations.RenameIndex(
+            model_name='projectupdateevent',
+            new_name='main_projec_project_69559a_idx',
+            old_fields=('project_update', 'job_created', 'counter'),
+        ),
+        migrations.RenameIndex(
+            model_name='role',
+            new_name='main_rbac_r_content_979bdd_idx',
+            old_fields=('content_type', 'object_id'),
+        ),
+        migrations.RenameIndex(
+            model_name='roleancestorentry',
+            new_name='main_rbac_r_ancesto_22b9f0_idx',
+            old_fields=('ancestor', 'content_type_id', 'object_id'),
+        ),
+        migrations.RenameIndex(
+            model_name='roleancestorentry',
+            new_name='main_rbac_r_ancesto_b44606_idx',
+            old_fields=('ancestor', 'content_type_id', 'role_field'),
+        ),
+        migrations.RenameIndex(
+            model_name='roleancestorentry',
+            new_name='main_rbac_r_ancesto_c87b87_idx',
+            old_fields=('ancestor', 'descendent'),
+        ),
+        migrations.RenameIndex(
+            model_name='systemjobevent',
+            new_name='main_system_system__e39825_idx',
+            old_fields=('system_job', 'job_created', 'uuid'),
+        ),
+        migrations.RenameIndex(
+            model_name='systemjobevent',
+            new_name='main_system_system__73537a_idx',
+            old_fields=('system_job', 'job_created', 'counter'),
+        ),
+    ]

awx/main/migrations/0185_move_JSONBlob_to_JSONField.py

@@ -1,277 +0,0 @@
-# Generated by Django 4.2.3 on 2023-08-02 13:18
-
-import awx.main.models.notifications
-from django.db import migrations, models
-
-
-class Migration(migrations.Migration):
-    dependencies = [
-        ('main', '0184_django_indexes'),
-        ('conf', '0010_change_to_JSONField'),
-    ]
-
-    operations = [
-        migrations.AlterField(
-            model_name='instancegroup',
-            name='policy_instance_list',
-            field=models.JSONField(
-                blank=True, default=list, help_text='List of exact-match Instances that will always be automatically assigned to this group'
-            ),
-        ),
-        migrations.AlterField(
-            model_name='jobtemplate',
-            name='survey_spec',
-            field=models.JSONField(blank=True, default=dict),
-        ),
-        migrations.AlterField(
-            model_name='notificationtemplate',
-            name='messages',
-            field=models.JSONField(
-                blank=True,
-                default=awx.main.models.notifications.NotificationTemplate.default_messages,
-                help_text='Optional custom messages for notification template.',
-                null=True,
-            ),
-        ),
-        migrations.AlterField(
-            model_name='notificationtemplate',
-            name='notification_configuration',
-            field=models.JSONField(default=dict),
-        ),
-        migrations.AlterField(
-            model_name='project',
-            name='inventory_files',
-            field=models.JSONField(
-                blank=True,
-                default=list,
-                editable=False,
-                help_text='Suggested list of content that could be Ansible inventory in the project',
-                verbose_name='Inventory Files',
-            ),
-        ),
-        migrations.AlterField(
-            model_name='project',
-            name='playbook_files',
-            field=models.JSONField(blank=True, default=list, editable=False, help_text='List of playbooks found in the project', verbose_name='Playbook Files'),
-        ),
-        migrations.AlterField(
-            model_name='schedule',
-            name='char_prompts',
-            field=models.JSONField(blank=True, default=dict),
-        ),
-        migrations.AlterField(
-            model_name='schedule',
-            name='survey_passwords',
-            field=models.JSONField(blank=True, default=dict, editable=False),
-        ),
-        migrations.AlterField(
-            model_name='workflowjobtemplate',
-            name='char_prompts',
-            field=models.JSONField(blank=True, default=dict),
-        ),
-        migrations.AlterField(
-            model_name='workflowjobtemplate',
-            name='survey_spec',
-            field=models.JSONField(blank=True, default=dict),
-        ),
-        migrations.AlterField(
-            model_name='workflowjobtemplatenode',
-            name='char_prompts',
-            field=models.JSONField(blank=True, default=dict),
-        ),
-        migrations.AlterField(
-            model_name='workflowjobtemplatenode',
-            name='survey_passwords',
-            field=models.JSONField(blank=True, default=dict, editable=False),
-        ),
-        # These are potentially a problem.  Move the existing fields
-        # aside while pretending like they've been deleted, then add
-        # in fresh empty fields.  Make the old fields nullable where
-        # needed while we are at it, so that new rows don't hit
-        # IntegrityError.  We'll do the data migration out-of-band
-        # using a task.
-        migrations.RunSQL(  # Already nullable
-            "ALTER TABLE main_activitystream RENAME deleted_actor TO deleted_actor_old;",
-            state_operations=[
-                migrations.RemoveField(
-                    model_name='activitystream',
-                    name='deleted_actor',
-                ),
-            ],
-        ),
-        migrations.AddField(
-            model_name='activitystream',
-            name='deleted_actor',
-            field=models.JSONField(null=True),
-        ),
-        migrations.RunSQL(
-            """
-            ALTER TABLE main_activitystream RENAME setting TO setting_old;
-            ALTER TABLE main_activitystream ALTER COLUMN setting_old DROP NOT NULL;
-            """,
-            state_operations=[
-                migrations.RemoveField(
-                    model_name='activitystream',
-                    name='setting',
-                ),
-            ],
-        ),
-        migrations.AddField(
-            model_name='activitystream',
-            name='setting',
-            field=models.JSONField(blank=True, default=dict),
-        ),
-        migrations.RunSQL(
-            """
-            ALTER TABLE main_job RENAME survey_passwords TO survey_passwords_old;
-            ALTER TABLE main_job ALTER COLUMN survey_passwords_old DROP NOT NULL;
-            """,
-            state_operations=[
-                migrations.RemoveField(
-                    model_name='job',
-                    name='survey_passwords',
-                ),
-            ],
-        ),
-        migrations.AddField(
-            model_name='job',
-            name='survey_passwords',
-            field=models.JSONField(blank=True, default=dict, editable=False),
-        ),
-        migrations.RunSQL(
-            """
-            ALTER TABLE main_joblaunchconfig RENAME char_prompts TO char_prompts_old;
-            ALTER TABLE main_joblaunchconfig ALTER COLUMN char_prompts_old DROP NOT NULL;
-            """,
-            state_operations=[
-                migrations.RemoveField(
-                    model_name='joblaunchconfig',
-                    name='char_prompts',
-                ),
-            ],
-        ),
-        migrations.AddField(
-            model_name='joblaunchconfig',
-            name='char_prompts',
-            field=models.JSONField(blank=True, default=dict),
-        ),
-        migrations.RunSQL(
-            """
-            ALTER TABLE main_joblaunchconfig RENAME survey_passwords TO survey_passwords_old;
-            ALTER TABLE main_joblaunchconfig ALTER COLUMN survey_passwords_old DROP NOT NULL;
-            """,
-            state_operations=[
-                migrations.RemoveField(
-                    model_name='joblaunchconfig',
-                    name='survey_passwords',
-                ),
-            ],
-        ),
-        migrations.AddField(
-            model_name='joblaunchconfig',
-            name='survey_passwords',
-            field=models.JSONField(blank=True, default=dict, editable=False),
-        ),
-        migrations.RunSQL(
-            """
-            ALTER TABLE main_notification RENAME body TO body_old;
-            ALTER TABLE main_notification ALTER COLUMN body_old DROP NOT NULL;
-            """,
-            state_operations=[
-                migrations.RemoveField(
-                    model_name='notification',
-                    name='body',
-                ),
-            ],
-        ),
-        migrations.AddField(
-            model_name='notification',
-            name='body',
-            field=models.JSONField(blank=True, default=dict),
-        ),
-        migrations.RunSQL(
-            """
-            ALTER TABLE main_unifiedjob RENAME job_env TO job_env_old;
-            ALTER TABLE main_unifiedjob ALTER COLUMN job_env_old DROP NOT NULL;
-            """,
-            state_operations=[
-                migrations.RemoveField(
-                    model_name='unifiedjob',
-                    name='job_env',
-                ),
-            ],
-        ),
-        migrations.AddField(
-            model_name='unifiedjob',
-            name='job_env',
-            field=models.JSONField(blank=True, default=dict, editable=False),
-        ),
-        migrations.RunSQL(
-            """
-            ALTER TABLE main_workflowjob RENAME char_prompts TO char_prompts_old;
-            ALTER TABLE main_workflowjob ALTER COLUMN char_prompts_old DROP NOT NULL;
-            """,
-            state_operations=[
-                migrations.RemoveField(
-                    model_name='workflowjob',
-                    name='char_prompts',
-                ),
-            ],
-        ),
-        migrations.AddField(
-            model_name='workflowjob',
-            name='char_prompts',
-            field=models.JSONField(blank=True, default=dict),
-        ),
-        migrations.RunSQL(
-            """
-            ALTER TABLE main_workflowjob RENAME survey_passwords TO survey_passwords_old;
-            ALTER TABLE main_workflowjob ALTER COLUMN survey_passwords_old DROP NOT NULL;
-            """,
-            state_operations=[
-                migrations.RemoveField(
-                    model_name='workflowjob',
-                    name='survey_passwords',
-                ),
-            ],
-        ),
-        migrations.AddField(
-            model_name='workflowjob',
-            name='survey_passwords',
-            field=models.JSONField(blank=True, default=dict, editable=False),
-        ),
-        migrations.RunSQL(
-            """
-            ALTER TABLE main_workflowjobnode RENAME char_prompts TO char_prompts_old;
-            ALTER TABLE main_workflowjobnode ALTER COLUMN char_prompts_old DROP NOT NULL;
-            """,
-            state_operations=[
-                migrations.RemoveField(
-                    model_name='workflowjobnode',
-                    name='char_prompts',
-                ),
-            ],
-        ),
-        migrations.AddField(
-            model_name='workflowjobnode',
-            name='char_prompts',
-            field=models.JSONField(blank=True, default=dict),
-        ),
-        migrations.RunSQL(
-            """
-            ALTER TABLE main_workflowjobnode RENAME survey_passwords TO survey_passwords_old;
-            ALTER TABLE main_workflowjobnode ALTER COLUMN survey_passwords_old DROP NOT NULL;
-            """,
-            state_operations=[
-                migrations.RemoveField(
-                    model_name='workflowjobnode',
-                    name='survey_passwords',
-                ),
-            ],
-        ),
-        migrations.AddField(
-            model_name='workflowjobnode',
-            name='survey_passwords',
-            field=models.JSONField(blank=True, default=dict, editable=False),
-        ),
-    ]

awx/main/migrations/0186_drop_django_taggit.py

@@ -1,27 +0,0 @@
-# -*- coding: utf-8 -*-
-from __future__ import unicode_literals
-
-from django.db import migrations
-
-
-def delete_taggit_contenttypes(apps, schema_editor):
-    ContentType = apps.get_model('contenttypes', 'ContentType')
-    ContentType.objects.filter(app_label='taggit').delete()
-
-
-def delete_taggit_migration_records(apps, schema_editor):
-    recorder = migrations.recorder.MigrationRecorder(connection=schema_editor.connection)
-    recorder.migration_qs.filter(app='taggit').delete()
-
-
-class Migration(migrations.Migration):
-    dependencies = [
-        ('main', '0185_move_JSONBlob_to_JSONField'),
-    ]
-
-    operations = [
-        migrations.RunSQL("DROP TABLE IF EXISTS taggit_tag CASCADE;"),
-        migrations.RunSQL("DROP TABLE IF EXISTS taggit_taggeditem CASCADE;"),
-        migrations.RunPython(delete_taggit_contenttypes),
-        migrations.RunPython(delete_taggit_migration_records),
-    ]

awx/main/migrations/_inventory_source_vars.py

@@ -158,7 +158,7 @@ class ec2(PluginFileInjector):
         return {
             # vars that change
             'ec2_block_devices': (
-                "dict(block_device_mappings | map(attribute='device_name') | list | zip(block_device_mappings | map(attribute='ebs.volume_id') | list))"
+                "dict(block_device_mappings | map(attribute='device_name') | list | zip(block_device_mappings " "| map(attribute='ebs.volume_id') | list))"
             ),
             'ec2_dns_name': 'public_dns_name',
             'ec2_group_name': 'placement.group_name',
@@ -635,7 +635,7 @@ class satellite6(PluginFileInjector):
             "environment": {
                 "prefix": "{}environment_".format(group_prefix),
                 "separator": "",
-                "key": "foreman['environment_name'] | lower | regex_replace(' ', '') | regex_replace('[^A-Za-z0-9_]', '_') | regex_replace('none', '')",
+                "key": "foreman['environment_name'] | lower | regex_replace(' ', '') | " "regex_replace('[^A-Za-z0-9_]', '_') | regex_replace('none', '')",
             },
             "location": {
                 "prefix": "{}location_".format(group_prefix),
@@ -656,7 +656,7 @@ class satellite6(PluginFileInjector):
             "content_view": {
                 "prefix": "{}content_view_".format(group_prefix),
                 "separator": "",
-                "key": "foreman['content_facet_attributes']['content_view_name'] | lower | regex_replace(' ', '') | regex_replace('[^A-Za-z0-9_]', '_')",
+                "key": "foreman['content_facet_attributes']['content_view_name'] | " "lower | regex_replace(' ', '') | regex_replace('[^A-Za-z0-9_]', '_')",
             },
         }
 

awx/main/models/__init__.py

@@ -3,7 +3,6 @@
 
 # Django
 from django.conf import settings  # noqa
-from django.db import connection
 from django.db.models.signals import pre_delete  # noqa
 
 # AWX
@@ -100,58 +99,6 @@ User.add_to_class('can_access_with_errors', check_user_access_with_errors)
 User.add_to_class('accessible_objects', user_accessible_objects)
 
 
-def convert_jsonfields():
-    if connection.vendor != 'postgresql':
-        return
-
-    # fmt: off
-    fields = [
-        ('main_activitystream', 'id', (
-            'deleted_actor',
-            'setting',
-        )),
-        ('main_job', 'unifiedjob_ptr_id', (
-            'survey_passwords',
-        )),
-        ('main_joblaunchconfig', 'id', (
-            'char_prompts',
-            'survey_passwords',
-        )),
-        ('main_notification', 'id', (
-            'body',
-        )),
-        ('main_unifiedjob', 'id', (
-            'job_env',
-        )),
-        ('main_workflowjob', 'unifiedjob_ptr_id', (
-            'char_prompts',
-            'survey_passwords',
-        )),
-        ('main_workflowjobnode', 'id', (
-            'char_prompts',
-            'survey_passwords',
-        )),
-    ]
-    # fmt: on
-
-    with connection.cursor() as cursor:
-        for table, pkfield, columns in fields:
-            # Do the renamed old columns still exist?  If so, run the task.
-            old_columns = ','.join(f"'{column}_old'" for column in columns)
-            cursor.execute(
-                f"""
-                select count(1) from information_schema.columns
-                where
-                  table_name = %s and column_name in ({old_columns});
-                """,
-                (table,),
-            )
-            if cursor.fetchone()[0]:
-                from awx.main.tasks.system import migrate_jsonfield
-
-                migrate_jsonfield.apply_async([table, pkfield, columns])
-
-
 def cleanup_created_modified_by(sender, **kwargs):
     # work around a bug in django-polymorphic that doesn't properly
     # handle cascades for reverse foreign keys on the polymorphic base model

awx/main/models/activity_stream.py

@@ -3,6 +3,7 @@
 
 # AWX
 from awx.api.versioning import reverse
+from awx.main.fields import JSONBlob
 from awx.main.models.base import accepts_json
 
 # Django
@@ -35,7 +36,7 @@ class ActivityStream(models.Model):
     operation = models.CharField(max_length=13, choices=OPERATION_CHOICES)
     timestamp = models.DateTimeField(auto_now_add=True)
     changes = accepts_json(models.TextField(blank=True))
-    deleted_actor = models.JSONField(null=True)
+    deleted_actor = JSONBlob(null=True)
     action_node = models.CharField(
         blank=True,
         default='',
@@ -83,7 +84,7 @@ class ActivityStream(models.Model):
     o_auth2_application = models.ManyToManyField("OAuth2Application", blank=True)
     o_auth2_access_token = models.ManyToManyField("OAuth2AccessToken", blank=True)
 
-    setting = models.JSONField(default=dict, blank=True)
+    setting = JSONBlob(default=dict, blank=True)
 
     def __str__(self):
         operation = self.operation if 'operation' in self.__dict__ else '_delayed_'

awx/main/models/base.py

@@ -7,6 +7,9 @@ from django.core.exceptions import ValidationError, ObjectDoesNotExist
 from django.utils.translation import gettext_lazy as _
 from django.utils.timezone import now
 
+# Django-Taggit
+from taggit.managers import TaggableManager
+
 # Django-CRUM
 from crum import get_current_user
 
@@ -298,6 +301,8 @@ class PrimordialModel(HasEditsMixin, CreatedModifiedModel):
         on_delete=models.SET_NULL,
     )
 
+    tags = TaggableManager(blank=True)
+
     def __init__(self, *args, **kwargs):
         r = super(PrimordialModel, self).__init__(*args, **kwargs)
         if self.pk:

awx/main/models/credential/__init__.py

@@ -91,7 +91,7 @@ class Credential(PasswordFieldsModel, CommonModelNameNotUnique, ResourceMixin):
         related_name='credentials',
         null=False,
         on_delete=models.CASCADE,
-        help_text=_('Specify the type of credential you want to create. Refer to the documentation for details on each type.'),
+        help_text=_('Specify the type of credential you want to create. Refer ' 'to the documentation for details on each type.'),
     )
     managed = models.BooleanField(default=False, editable=False)
     organization = models.ForeignKey(
@@ -103,7 +103,7 @@ class Credential(PasswordFieldsModel, CommonModelNameNotUnique, ResourceMixin):
         related_name='credentials',
     )
     inputs = CredentialInputField(
-        blank=True, default=dict, help_text=_('Enter inputs using either JSON or YAML syntax. Refer to the documentation for example syntax.')
+        blank=True, default=dict, help_text=_('Enter inputs using either JSON or YAML syntax. ' 'Refer to the documentation for example syntax.')
     )
     admin_role = ImplicitRoleField(
         parent_role=[
@@ -346,12 +346,12 @@ class CredentialType(CommonModelNameNotUnique):
     managed = models.BooleanField(default=False, editable=False)
     namespace = models.CharField(max_length=1024, null=True, default=None, editable=False)
     inputs = CredentialTypeInputField(
-        blank=True, default=dict, help_text=_('Enter inputs using either JSON or YAML syntax. Refer to the documentation for example syntax.')
+        blank=True, default=dict, help_text=_('Enter inputs using either JSON or YAML syntax. ' 'Refer to the documentation for example syntax.')
     )
     injectors = CredentialTypeInjectorField(
         blank=True,
         default=dict,
-        help_text=_('Enter injectors using either JSON or YAML syntax. Refer to the documentation for example syntax.'),
+        help_text=_('Enter injectors using either JSON or YAML syntax. ' 'Refer to the documentation for example syntax.'),
     )
 
     @classmethod
@@ -605,7 +605,9 @@ ManagedCredentialType(
                 'id': 'become_method',
                 'label': gettext_noop('Privilege Escalation Method'),
                 'type': 'string',
-                'help_text': gettext_noop('Specify a method for "become" operations. This is equivalent to specifying the --become-method Ansible parameter.'),
+                'help_text': gettext_noop(
+                    'Specify a method for "become" operations. This is ' 'equivalent to specifying the --become-method ' 'Ansible parameter.'
+                ),
             },
             {
                 'id': 'become_username',
@@ -747,7 +749,7 @@ ManagedCredentialType(
                 'id': 'host',
                 'label': gettext_noop('Host (Authentication URL)'),
                 'type': 'string',
-                'help_text': gettext_noop('The host to authenticate with.  For example, https://openstack.business.com/v2.0/'),
+                'help_text': gettext_noop('The host to authenticate with.  For example, ' 'https://openstack.business.com/v2.0/'),
             },
             {
                 'id': 'project',
@@ -798,7 +800,7 @@ ManagedCredentialType(
                 'id': 'host',
                 'label': gettext_noop('VCenter Host'),
                 'type': 'string',
-                'help_text': gettext_noop('Enter the hostname or IP address that corresponds to your VMware vCenter.'),
+                'help_text': gettext_noop('Enter the hostname or IP address that corresponds ' 'to your VMware vCenter.'),
             },
             {'id': 'username', 'label': gettext_noop('Username'), 'type': 'string'},
             {
@@ -823,7 +825,7 @@ ManagedCredentialType(
                 'id': 'host',
                 'label': gettext_noop('Satellite 6 URL'),
                 'type': 'string',
-                'help_text': gettext_noop('Enter the URL that corresponds to your Red Hat Satellite 6 server. For example, https://satellite.example.org'),
+                'help_text': gettext_noop('Enter the URL that corresponds to your Red Hat ' 'Satellite 6 server. For example, https://satellite.example.org'),
             },
             {'id': 'username', 'label': gettext_noop('Username'), 'type': 'string'},
             {
@@ -848,7 +850,7 @@ ManagedCredentialType(
                 'id': 'username',
                 'label': gettext_noop('Service Account Email Address'),
                 'type': 'string',
-                'help_text': gettext_noop('The email address assigned to the Google Compute Engine service account.'),
+                'help_text': gettext_noop('The email address assigned to the Google Compute ' 'Engine service account.'),
             },
             {
                 'id': 'project',
@@ -868,7 +870,7 @@ ManagedCredentialType(
                 'format': 'ssh_private_key',
                 'secret': True,
                 'multiline': True,
-                'help_text': gettext_noop('Paste the contents of the PEM file associated with the service account email.'),
+                'help_text': gettext_noop('Paste the contents of the PEM file associated ' 'with the service account email.'),
             },
         ],
         'required': ['username', 'ssh_key_data'],
@@ -886,7 +888,7 @@ ManagedCredentialType(
                 'id': 'subscription',
                 'label': gettext_noop('Subscription ID'),
                 'type': 'string',
-                'help_text': gettext_noop('Subscription ID is an Azure construct, which is mapped to a username.'),
+                'help_text': gettext_noop('Subscription ID is an Azure construct, which is ' 'mapped to a username.'),
             },
             {'id': 'username', 'label': gettext_noop('Username'), 'type': 'string'},
             {
@@ -907,7 +909,7 @@ ManagedCredentialType(
                 'id': 'cloud_environment',
                 'label': gettext_noop('Azure Cloud Environment'),
                 'type': 'string',
-                'help_text': gettext_noop('Environment variable AZURE_CLOUD_ENVIRONMENT when using Azure GovCloud or Azure stack.'),
+                'help_text': gettext_noop('Environment variable AZURE_CLOUD_ENVIRONMENT when' ' using Azure GovCloud or Azure stack.'),
             },
         ],
         'required': ['subscription'],
@@ -1038,7 +1040,7 @@ ManagedCredentialType(
                 'label': gettext_noop('Username'),
                 'type': 'string',
                 'help_text': gettext_noop(
-                    'Red Hat Ansible Automation Platform username id to authenticate as.This should not be set if an OAuth token is being used.'
+                    'Red Hat Ansible Automation Platform username id to authenticate as.' 'This should not be set if an OAuth token is being used.'
                 ),
             },
             {
@@ -1052,7 +1054,7 @@ ManagedCredentialType(
                 'label': gettext_noop('OAuth Token'),
                 'type': 'string',
                 'secret': True,
-                'help_text': gettext_noop('An OAuth token to use to authenticate with.This should not be set if username/password are being used.'),
+                'help_text': gettext_noop('An OAuth token to use to authenticate with.' 'This should not be set if username/password are being used.'),
             },
             {'id': 'verify_ssl', 'label': gettext_noop('Verify SSL'), 'type': 'boolean', 'secret': False},
         ],
@@ -1163,7 +1165,7 @@ ManagedCredentialType(
                 'id': 'auth_url',
                 'label': gettext_noop('Auth Server URL'),
                 'type': 'string',
-                'help_text': gettext_noop('The URL of a Keycloak server token_endpoint, if using SSO auth.'),
+                'help_text': gettext_noop('The URL of a Keycloak server token_endpoint, if using ' 'SSO auth.'),
             },
             {
                 'id': 'token',

awx/main/models/events.py

@@ -4,7 +4,6 @@ import datetime
 from datetime import timezone
 import logging
 from collections import defaultdict
-import time
 
 from django.conf import settings
 from django.core.exceptions import ObjectDoesNotExist
@@ -384,17 +383,8 @@ class BasePlaybookEvent(CreatedModifiedModel):
                         .distinct()
                     )  # noqa
 
-                    # NOTE: we take a set of changed and failed parent uuids because the subquery
-                    # complicates the plan with large event tables causing very long query execution time
-                    changed_start = time.time()
-                    changed_res = job.get_event_queryset().filter(uuid__in=set(changed)).update(changed=True)
-                    failed_start = time.time()
-                    failed_res = job.get_event_queryset().filter(uuid__in=set(failed)).update(failed=True)
-                    logger.debug(
-                        f'Event propagation for job {job.id}: '
-                        f'marked {changed_res} as changed in {failed_start - changed_start:.4f}s, '
-                        f'{failed_res} as failed in {time.time() - failed_start:.4f}s'
-                    )
+                    job.get_event_queryset().filter(uuid__in=changed).update(changed=True)
+                    job.get_event_queryset().filter(uuid__in=failed).update(failed=True)
 
         for field in ('playbook', 'play', 'task', 'role'):
             value = force_str(event_data.get(field, '')).strip()

awx/main/models/ha.py

@@ -20,7 +20,7 @@ from solo.models import SingletonModel
 # AWX
 from awx import __version__ as awx_application_version
 from awx.api.versioning import reverse
-from awx.main.fields import ImplicitRoleField
+from awx.main.fields import JSONBlob, ImplicitRoleField
 from awx.main.managers import InstanceManager, UUID_DEFAULT
 from awx.main.constants import JOB_FOLDER_PREFIX
 from awx.main.models.base import BaseModel, HasEditsMixin, prevent_search
@@ -406,7 +406,7 @@ class InstanceGroup(HasPolicyEditsMixin, BaseModel, RelatedJobsMixin, ResourceMi
     max_forks = models.IntegerField(default=0, help_text=_("Max forks to execute on this group. Zero means no limit."))
     policy_instance_percentage = models.IntegerField(default=0, help_text=_("Percentage of Instances to automatically assign to this group"))
     policy_instance_minimum = models.IntegerField(default=0, help_text=_("Static minimum number of Instances to automatically assign to this group"))
-    policy_instance_list = models.JSONField(
+    policy_instance_list = JSONBlob(
         default=list, blank=True, help_text=_("List of exact-match Instances that will always be automatically assigned to this group")
     )
 

awx/main/models/inventory.py

@@ -106,28 +106,28 @@ class Inventory(CommonModelNameNotUnique, ResourceMixin, RelatedJobsMixin):
     has_active_failures = models.BooleanField(
         default=False,
         editable=False,
-        help_text=_('This field is deprecated and will be removed in a future release. Flag indicating whether any hosts in this inventory have failed.'),
+        help_text=_('This field is deprecated and will be removed in a future release. ' 'Flag indicating whether any hosts in this inventory have failed.'),
     )
     total_hosts = models.PositiveIntegerField(
         default=0,
         editable=False,
-        help_text=_('This field is deprecated and will be removed in a future release. Total number of hosts in this inventory.'),
+        help_text=_('This field is deprecated and will be removed in a future release. ' 'Total number of hosts in this inventory.'),
     )
     hosts_with_active_failures = models.PositiveIntegerField(
         default=0,
         editable=False,
-        help_text=_('This field is deprecated and will be removed in a future release. Number of hosts in this inventory with active failures.'),
+        help_text=_('This field is deprecated and will be removed in a future release. ' 'Number of hosts in this inventory with active failures.'),
     )
     total_groups = models.PositiveIntegerField(
         default=0,
         editable=False,
-        help_text=_('This field is deprecated and will be removed in a future release. Total number of groups in this inventory.'),
+        help_text=_('This field is deprecated and will be removed in a future release. ' 'Total number of groups in this inventory.'),
     )
     has_inventory_sources = models.BooleanField(
         default=False,
         editable=False,
         help_text=_(
-            'This field is deprecated and will be removed in a future release. Flag indicating whether this inventory has any external inventory sources.'
+            'This field is deprecated and will be removed in a future release. ' 'Flag indicating whether this inventory has any external inventory sources.'
         ),
     )
     total_inventory_sources = models.PositiveIntegerField(
@@ -424,7 +424,7 @@ class Inventory(CommonModelNameNotUnique, ResourceMixin, RelatedJobsMixin):
             for t in tasks:
                 t.task_impact = t._get_task_impact()
             UnifiedJob.objects.bulk_update(tasks, ['task_impact'])
-        logger.debug("Finished updating inventory computed fields, pk={0}, in {1:.3f} seconds".format(self.pk, time.time() - start_time))
+        logger.debug("Finished updating inventory computed fields, pk={0}, in " "{1:.3f} seconds".format(self.pk, time.time() - start_time))
 
     def websocket_emit_status(self, status):
         connection.on_commit(
@@ -899,18 +899,18 @@ class HostMetric(models.Model):
 
             last_automation_before = now() - dateutil.relativedelta.relativedelta(months=months_ago)
 
-            logger.info(f'cleanup_host_metrics: soft-deleting records last automated before {last_automation_before}')
+            logger.info(f'Cleanup [HostMetric]: soft-deleting records last automated before {last_automation_before}')
             HostMetric.active_objects.filter(last_automation__lt=last_automation_before).update(
                 deleted=True, deleted_counter=models.F('deleted_counter') + 1, last_deleted=now()
             )
             settings.CLEANUP_HOST_METRICS_LAST_TS = now()
         except (TypeError, ValueError):
-            logger.error(f"cleanup_host_metrics: months_ago({months_ago}) has to be a positive integer value")
+            logger.error(f"Cleanup [HostMetric]: months_ago({months_ago}) has to be a positive integer value")
 
 
 class HostMetricSummaryMonthly(models.Model):
     """
-    HostMetric summaries computed by scheduled task 'awx.main.tasks.system.host_metric_summary_monthly' monthly
+    HostMetric summaries computed by scheduled task <TODO> monthly
     """
 
     date = models.DateField(unique=True)
@@ -1055,16 +1055,16 @@ class InventorySourceOptions(BaseModel):
             # the actual inventory source being used (Amazon requires Amazon
             # credentials; Rackspace requires Rackspace credentials; etc...)
             if source.replace('ec2', 'aws') != cred.kind:
-                return _('Cloud-based inventory sources (such as %s) require credentials for the matching cloud service.') % source
+                return _('Cloud-based inventory sources (such as %s) require ' 'credentials for the matching cloud service.') % source
         # Allow an EC2 source to omit the credential.  If Tower is running on
         # an EC2 instance with an IAM Role assigned, boto will use credentials
         # from the instance metadata instead of those explicitly provided.
         elif source in CLOUD_PROVIDERS and source != 'ec2':
             return _('Credential is required for a cloud source.')
         elif source == 'custom' and cred and cred.credential_type.kind in ('scm', 'ssh', 'insights', 'vault'):
-            return _('Credentials of type machine, source control, insights and vault are disallowed for custom inventory sources.')
+            return _('Credentials of type machine, source control, insights and vault are ' 'disallowed for custom inventory sources.')
         elif source == 'scm' and cred and cred.credential_type.kind in ('insights', 'vault'):
-            return _('Credentials of type insights and vault are disallowed for scm inventory sources.')
+            return _('Credentials of type insights and vault are ' 'disallowed for scm inventory sources.')
         return None
 
     def get_cloud_credential(self):
@@ -1623,7 +1623,6 @@ class rhv(PluginFileInjector):
     collection = 'ovirt'
     downstream_namespace = 'redhat'
     downstream_collection = 'rhv'
-    use_fqcn = True
 
 
 class satellite6(PluginFileInjector):

awx/main/models/jobs.py

@@ -101,7 +101,7 @@ class JobOptions(BaseModel):
         max_length=1024,
         default='',
         blank=True,
-        help_text=_('Branch to use in job run. Project default used if blank. Only allowed if project allow_override field is set to true.'),
+        help_text=_('Branch to use in job run. Project default used if blank. ' 'Only allowed if project allow_override field is set to true.'),
     )
     forks = models.PositiveIntegerField(
         blank=True,
@@ -253,7 +253,7 @@ class JobTemplate(UnifiedJobTemplate, JobOptions, SurveyJobTemplateMixin, Resour
     job_slice_count = models.PositiveIntegerField(
         blank=True,
         default=1,
-        help_text=_("The number of jobs to slice into at runtime. Will cause the Job Template to launch a workflow if value is greater than 1."),
+        help_text=_("The number of jobs to slice into at runtime. " "Will cause the Job Template to launch a workflow if value is greater than 1."),
     )
 
     admin_role = ImplicitRoleField(parent_role=['organization.job_template_admin_role'])
@@ -596,12 +596,12 @@ class Job(UnifiedJob, JobOptions, SurveyJobMixin, JobNotificationMixin, TaskMana
     job_slice_number = models.PositiveIntegerField(
         blank=True,
         default=0,
-        help_text=_("If part of a sliced job, the ID of the inventory slice operated on. If not part of sliced job, parameter is not used."),
+        help_text=_("If part of a sliced job, the ID of the inventory slice operated on. " "If not part of sliced job, parameter is not used."),
     )
     job_slice_count = models.PositiveIntegerField(
         blank=True,
         default=1,
-        help_text=_("If ran as part of sliced jobs, the total number of slices. If 1, job is not part of a sliced job."),
+        help_text=_("If ran as part of sliced jobs, the total number of slices. " "If 1, job is not part of a sliced job."),
     )
 
     def _get_parent_field_name(self):
@@ -883,7 +883,7 @@ class LaunchTimeConfigBase(BaseModel):
     )
     # All standard fields are stored in this dictionary field
     # This is a solution to the nullable CharField problem, specific to prompting
-    char_prompts = models.JSONField(default=dict, blank=True)
+    char_prompts = JSONBlob(default=dict, blank=True)
 
     # Define fields that are not really fields, but alias to char_prompts lookups
     limit = NullablePromptPseudoField('limit')
@@ -960,7 +960,7 @@ class LaunchTimeConfig(LaunchTimeConfigBase):
     # Special case prompting fields, even more special than the other ones
     extra_data = JSONBlob(default=dict, blank=True)
     survey_passwords = prevent_search(
-        models.JSONField(
+        JSONBlob(
             default=dict,
             editable=False,
             blank=True,

awx/main/models/mixins.py

@@ -24,7 +24,7 @@ from awx.main.utils import parse_yaml_or_json, get_custom_venv_choices, get_lice
 from awx.main.utils.execution_environments import get_default_execution_environment
 from awx.main.utils.encryption import decrypt_value, get_encryption_key, is_encrypted
 from awx.main.utils.polymorphic import build_polymorphic_ctypes_map
-from awx.main.fields import AskForField
+from awx.main.fields import AskForField, JSONBlob
 from awx.main.constants import ACTIVE_STATES
 
 
@@ -103,7 +103,7 @@ class SurveyJobTemplateMixin(models.Model):
     survey_enabled = models.BooleanField(
         default=False,
     )
-    survey_spec = prevent_search(models.JSONField(default=dict, blank=True))
+    survey_spec = prevent_search(JSONBlob(default=dict, blank=True))
 
     ask_inventory_on_launch = AskForField(
         blank=True,
@@ -392,7 +392,7 @@ class SurveyJobMixin(models.Model):
         abstract = True
 
     survey_passwords = prevent_search(
-        models.JSONField(
+        JSONBlob(
             default=dict,
             editable=False,
             blank=True,
@@ -675,4 +675,4 @@ class WebhookMixin(models.Model):
         if response.status_code < 400:
             logger.debug("Webhook status update sent.")
         else:
-            logger.error("Posting webhook status failed, code: {}\n" "{}\nPayload sent: {}".format(response.status_code, response.text, json.dumps(data)))
+            logger.error("Posting webhook status failed, code: {}\n" "{}\n" "Payload sent: {}".format(response.status_code, response.text, json.dumps(data)))

awx/main/models/notifications.py

@@ -17,6 +17,7 @@ from jinja2.exceptions import TemplateSyntaxError, UndefinedError, SecurityError
 
 # AWX
 from awx.api.versioning import reverse
+from awx.main.fields import JSONBlob
 from awx.main.models.base import CommonModelNameNotUnique, CreatedModifiedModel, prevent_search
 from awx.main.utils import encrypt_field, decrypt_field, set_environ
 from awx.main.notifications.email_backend import CustomEmailBackend
@@ -68,12 +69,12 @@ class NotificationTemplate(CommonModelNameNotUnique):
         choices=NOTIFICATION_TYPE_CHOICES,
     )
 
-    notification_configuration = prevent_search(models.JSONField(default=dict))
+    notification_configuration = prevent_search(JSONBlob(default=dict))
 
     def default_messages():
         return {'started': None, 'success': None, 'error': None, 'workflow_approval': None}
 
-    messages = models.JSONField(null=True, blank=True, default=default_messages, help_text=_('Optional custom messages for notification template.'))
+    messages = JSONBlob(null=True, blank=True, default=default_messages, help_text=_('Optional custom messages for notification template.'))
 
     def has_message(self, condition):
         potential_template = self.messages.get(condition, {})
@@ -235,7 +236,7 @@ class Notification(CreatedModifiedModel):
         default='',
         editable=False,
     )
-    body = models.JSONField(default=dict, blank=True)
+    body = JSONBlob(default=dict, blank=True)
 
     def get_absolute_url(self, request=None):
         return reverse('api:notification_detail', kwargs={'pk': self.pk}, request=request)
@@ -283,7 +284,7 @@ class JobNotificationMixin(object):
         'workflow_url',
         'scm_branch',
         'artifacts',
-        {'host_status_counts': ['skipped', 'ok', 'changed', 'failed', 'failures', 'dark', 'processed', 'rescued', 'ignored']},
+        {'host_status_counts': ['skipped', 'ok', 'changed', 'failed', 'failures', 'dark' 'processed', 'rescued', 'ignored']},
         {
             'summary_fields': [
                 {

awx/main/models/projects.py

@@ -33,7 +33,7 @@ from awx.main.models.mixins import ResourceMixin, TaskManagerProjectUpdateMixin,
 from awx.main.utils import update_scm_url, polymorphic
 from awx.main.utils.ansible import skip_directory, could_be_inventory, could_be_playbook
 from awx.main.utils.execution_environments import get_control_plane_execution_environment
-from awx.main.fields import ImplicitRoleField
+from awx.main.fields import ImplicitRoleField, JSONBlob
 from awx.main.models.rbac import (
     ROLE_SINGLETON_SYSTEM_ADMINISTRATOR,
     ROLE_SINGLETON_SYSTEM_AUDITOR,
@@ -74,7 +74,7 @@ class ProjectOptions(models.Model):
             return []
 
     local_path = models.CharField(
-        max_length=1024, blank=True, help_text=_('Local path (relative to PROJECTS_ROOT) containing playbooks and related files for this project.')
+        max_length=1024, blank=True, help_text=_('Local path (relative to PROJECTS_ROOT) containing ' 'playbooks and related files for this project.')
     )
 
     scm_type = models.CharField(
@@ -276,11 +276,11 @@ class Project(UnifiedJobTemplate, ProjectOptions, ResourceMixin, CustomVirtualEn
     scm_update_cache_timeout = models.PositiveIntegerField(
         default=0,
         blank=True,
-        help_text=_('The number of seconds after the last project update ran that a new project update will be launched as a job dependency.'),
+        help_text=_('The number of seconds after the last project update ran that a new ' 'project update will be launched as a job dependency.'),
     )
     allow_override = models.BooleanField(
         default=False,
-        help_text=_('Allow changing the SCM branch or revision in a job template that uses this project.'),
+        help_text=_('Allow changing the SCM branch or revision in a job template ' 'that uses this project.'),
     )
 
     # credential (keys) used to validate content signature
@@ -303,7 +303,7 @@ class Project(UnifiedJobTemplate, ProjectOptions, ResourceMixin, CustomVirtualEn
         help_text=_('The last revision fetched by a project update'),
     )
 
-    playbook_files = models.JSONField(
+    playbook_files = JSONBlob(
         default=list,
         blank=True,
         editable=False,
@@ -311,7 +311,7 @@ class Project(UnifiedJobTemplate, ProjectOptions, ResourceMixin, CustomVirtualEn
         help_text=_('List of playbooks found in the project'),
     )
 
-    inventory_files = models.JSONField(
+    inventory_files = JSONBlob(
         default=list,
         blank=True,
         editable=False,
@@ -479,7 +479,7 @@ class Project(UnifiedJobTemplate, ProjectOptions, ResourceMixin, CustomVirtualEn
         RunProjectUpdate/RunInventoryUpdate.
         """
 
-        if self.status not in ('error', 'failed') or self.scm_update_on_launch:
+        if self.status not in ('error', 'failed'):
             return None
 
         latest_update = self.project_updates.last()

awx/main/models/unified_jobs.py

@@ -55,7 +55,7 @@ from awx.main.utils import polymorphic
 from awx.main.constants import ACTIVE_STATES, CAN_CANCEL, JOB_VARIABLE_PREFIXES
 from awx.main.redact import UriCleaner, REPLACE_STR
 from awx.main.consumers import emit_channel_notification
-from awx.main.fields import AskForField, OrderedManyToManyField
+from awx.main.fields import AskForField, OrderedManyToManyField, JSONBlob
 
 __all__ = ['UnifiedJobTemplate', 'UnifiedJob', 'StdoutMaxBytesExceeded']
 
@@ -668,7 +668,7 @@ class UnifiedJob(
         editable=False,
     )
     job_env = prevent_search(
-        models.JSONField(
+        JSONBlob(
             default=dict,
             blank=True,
             editable=False,
@@ -1137,6 +1137,9 @@ class UnifiedJob(
                     if total > max_supported:
                         raise StdoutMaxBytesExceeded(total, max_supported)
 
+                # psycopg3's copy writes bytes, but callers of this
+                # function assume a str-based fd will be returned; decode
+                # .write() calls on the fly to maintain this interface
                 tbl = self._meta.db_table + 'event'
                 created_by_cond = ''
                 if self.has_unpartitioned_events:
@@ -1145,9 +1148,6 @@ class UnifiedJob(
                     created_by_cond = f"job_created='{self.created.isoformat()}' AND "
 
                 sql = f"copy (select stdout from {tbl} where {created_by_cond}{self.event_parent_key}={self.id} and stdout != '' order by start_line) to stdout"  # nosql
-                # psycopg3's copy writes bytes, but callers of this
-                # function assume a str-based fd will be returned; decode
-                # .write() calls on the fly to maintain this interface
                 with cursor.copy(sql) as copy:
                     while data := copy.read():
                         fd.write(smart_str(bytes(data)))

awx/main/models/workflow.py

@@ -82,7 +82,7 @@ class WorkflowNodeBase(CreatedModifiedModel, LaunchTimeConfig):
         related_name='%(class)ss_always',
     )
     all_parents_must_converge = models.BooleanField(
-        default=False, help_text=_("If enabled then the node will only run if all of the parent nodes have met the criteria to reach this node")
+        default=False, help_text=_("If enabled then the node will only run if all of the parent nodes " "have met the criteria to reach this node")
     )
     unified_job_template = models.ForeignKey(
         'UnifiedJobTemplate',
@@ -181,7 +181,7 @@ class WorkflowJobTemplateNode(WorkflowNodeBase):
         max_length=512,
         default=uuid4,
         blank=False,
-        help_text=_('An identifier for this node that is unique within its workflow. It is copied to workflow job nodes corresponding to this node.'),
+        help_text=_('An identifier for this node that is unique within its workflow. ' 'It is copied to workflow job nodes corresponding to this node.'),
     )
     instance_groups = OrderedManyToManyField(
         'InstanceGroup',
@@ -334,7 +334,7 @@ class WorkflowJobNode(WorkflowNodeBase):
             accepted_fields, ignored_fields, errors = ujt_obj._accept_or_ignore_job_kwargs(**node_prompts_data)
             if errors:
                 logger.info(
-                    _('Bad launch configuration starting template {template_pk} as part of workflow {workflow_pk}. Errors:\n{error_text}').format(
+                    _('Bad launch configuration starting template {template_pk} as part of ' 'workflow {workflow_pk}. Errors:\n{error_text}').format(
                         template_pk=ujt_obj.pk, workflow_pk=self.pk, error_text=errors
                     )
                 )
@@ -647,7 +647,7 @@ class WorkflowJob(UnifiedJob, WorkflowJobOptions, SurveyJobMixin, JobNotificatio
         null=True,
         default=None,
         on_delete=models.SET_NULL,
-        help_text=_("If automatically created for a sliced job run, the job template the workflow job was created from."),
+        help_text=_("If automatically created for a sliced job run, the job template " "the workflow job was created from."),
     )
     is_sliced_job = models.BooleanField(default=False)
     is_bulk_job = models.BooleanField(default=False)
@@ -661,11 +661,7 @@ class WorkflowJob(UnifiedJob, WorkflowJobOptions, SurveyJobMixin, JobNotificatio
 
     @property
     def event_processing_finished(self):
-        return True  # workflow jobs do not have events
-
-    @property
-    def has_unpartitioned_events(self):
-        return False  # workflow jobs do not have events
+        return True
 
     def _get_parent_field_name(self):
         if self.job_template_id:
@@ -718,7 +714,7 @@ class WorkflowJob(UnifiedJob, WorkflowJobOptions, SurveyJobMixin, JobNotificatio
         wj = self.get_workflow_job()
         while wj and wj.workflow_job_template_id:
             if wj.pk in wj_ids:
-                logger.critical('Cycles detected in the workflow jobs graph, this is not normal and suggests task manager degeneracy.')
+                logger.critical('Cycles detected in the workflow jobs graph, ' 'this is not normal and suggests task manager degeneracy.')
                 break
             wj_ids.add(wj.pk)
             ancestors.append(wj.workflow_job_template)
@@ -918,11 +914,7 @@ class WorkflowApproval(UnifiedJob, JobNotificationMixin):
 
     @property
     def event_processing_finished(self):
-        return True  # approval jobs do not have events
-
-    @property
-    def has_unpartitioned_events(self):
-        return False  # approval jobs do not have events
+        return True
 
     def send_approval_notification(self, approval_status):
         from awx.main.tasks.system import send_notifications  # avoid circular import

awx/main/notifications/custom_notification_base.py

@@ -8,7 +8,7 @@ class CustomNotificationBase(object):
 
     DEFAULT_APPROVAL_RUNNING_MSG = 'The approval node "{{ approval_node_name }}" needs review. This node can be viewed at: {{ workflow_url }}'
     DEFAULT_APPROVAL_RUNNING_BODY = (
-        'The approval node "{{ approval_node_name }}" needs review. This approval node can be viewed at: {{ workflow_url }}\n\n{{ job_metadata }}'
+        'The approval node "{{ approval_node_name }}" needs review. ' 'This approval node can be viewed at: {{ workflow_url }}\n\n{{ job_metadata }}'
     )
 
     DEFAULT_APPROVAL_APPROVED_MSG = 'The approval node "{{ approval_node_name }}" was approved. {{ workflow_url }}'

awx/main/notifications/webhook_backend.py

@@ -32,7 +32,7 @@ class WebhookBackend(AWXBaseEmailBackend, CustomNotificationBase):
         "success": {"body": DEFAULT_BODY},
         "error": {"body": DEFAULT_BODY},
         "workflow_approval": {
-            "running": {"body": '{"body": "The approval node \\"{{ approval_node_name }}\\" needs review. This node can be viewed at: {{ workflow_url }}"}'},
+            "running": {"body": '{"body": "The approval node \\"{{ approval_node_name }}\\" needs review. ' 'This node can be viewed at: {{ workflow_url }}"}'},
             "approved": {"body": '{"body": "The approval node \\"{{ approval_node_name }}\\" was approved. {{ workflow_url }}"}'},
             "timed_out": {"body": '{"body": "The approval node \\"{{ approval_node_name }}\\" has timed out. {{ workflow_url }}"}'},
             "denied": {"body": '{"body": "The approval node \\"{{ approval_node_name }}\\" was denied. {{ workflow_url }}"}'},

awx/main/registrar.py

@@ -3,6 +3,8 @@
 
 from django.db.models.signals import pre_save, post_save, pre_delete, m2m_changed
 
+from taggit.managers import TaggableManager
+
 
 class ActivityStreamRegistrar(object):
     def __init__(self):
@@ -19,6 +21,8 @@ class ActivityStreamRegistrar(object):
             pre_delete.connect(activity_stream_delete, sender=model, dispatch_uid=str(self.__class__) + str(model) + "_delete")
 
             for m2mfield in model._meta.many_to_many:
+                if isinstance(m2mfield, TaggableManager):
+                    continue  # Special case for taggit app
                 try:
                     m2m_attr = getattr(model, m2mfield.name)
                     m2m_changed.connect(

awx/main/scheduler/task_manager.py

@@ -25,6 +25,7 @@ from awx.main.models import (
     InventoryUpdate,
     Job,
     Project,
+    ProjectUpdate,
     UnifiedJob,
     WorkflowApproval,
     WorkflowJob,
@@ -101,33 +102,27 @@ class TaskBase:
 
     def record_aggregate_metrics(self, *args):
         if not is_testing():
-            try:
-                # increment task_manager_schedule_calls regardless if the other
-                # metrics are recorded
-                s_metrics.Metrics(auto_pipe_execute=True).inc(f"{self.prefix}__schedule_calls", 1)
-                # Only record metrics if the last time recording was more
-                # than SUBSYSTEM_METRICS_TASK_MANAGER_RECORD_INTERVAL ago.
-                # Prevents a short-duration task manager that runs directly after a
-                # long task manager to override useful metrics.
-                current_time = time.time()
-                time_last_recorded = current_time - self.subsystem_metrics.decode(f"{self.prefix}_recorded_timestamp")
-                if time_last_recorded > settings.SUBSYSTEM_METRICS_TASK_MANAGER_RECORD_INTERVAL:
-                    logger.debug(f"recording {self.prefix} metrics, last recorded {time_last_recorded} seconds ago")
-                    self.subsystem_metrics.set(f"{self.prefix}_recorded_timestamp", current_time)
-                    self.subsystem_metrics.pipe_execute()
-                else:
-                    logger.debug(f"skipping recording {self.prefix} metrics, last recorded {time_last_recorded} seconds ago")
-            except Exception:
-                logger.exception(f"Error saving metrics for {self.prefix}")
+            # increment task_manager_schedule_calls regardless if the other
+            # metrics are recorded
+            s_metrics.Metrics(auto_pipe_execute=True).inc(f"{self.prefix}__schedule_calls", 1)
+            # Only record metrics if the last time recording was more
+            # than SUBSYSTEM_METRICS_TASK_MANAGER_RECORD_INTERVAL ago.
+            # Prevents a short-duration task manager that runs directly after a
+            # long task manager to override useful metrics.
+            current_time = time.time()
+            time_last_recorded = current_time - self.subsystem_metrics.decode(f"{self.prefix}_recorded_timestamp")
+            if time_last_recorded > settings.SUBSYSTEM_METRICS_TASK_MANAGER_RECORD_INTERVAL:
+                logger.debug(f"recording {self.prefix} metrics, last recorded {time_last_recorded} seconds ago")
+                self.subsystem_metrics.set(f"{self.prefix}_recorded_timestamp", current_time)
+                self.subsystem_metrics.pipe_execute()
+            else:
+                logger.debug(f"skipping recording {self.prefix} metrics, last recorded {time_last_recorded} seconds ago")
 
     def record_aggregate_metrics_and_exit(self, *args):
         self.record_aggregate_metrics()
         sys.exit(1)
 
     def schedule(self):
-        # Always be able to restore the original signal handler if we finish
-        original_sigusr1 = signal.getsignal(signal.SIGUSR1)
-
         # Lock
         with task_manager_bulk_reschedule():
             with advisory_lock(f"{self.prefix}_lock", wait=False) as acquired:
@@ -136,14 +131,9 @@ class TaskBase:
                         logger.debug(f"Not running {self.prefix} scheduler, another task holds lock")
                         return
                     logger.debug(f"Starting {self.prefix} Scheduler")
-                    # if sigusr1 due to timeout, still record metrics
-                    signal.signal(signal.SIGUSR1, self.record_aggregate_metrics_and_exit)
-                    try:
-                        self._schedule()
-                    finally:
-                        # Reset the signal handler back to the default just in case anything
-                        # else uses the same signal for other purposes
-                        signal.signal(signal.SIGUSR1, original_sigusr1)
+                    # if sigterm due to timeout, still record metrics
+                    signal.signal(signal.SIGTERM, self.record_aggregate_metrics_and_exit)
+                    self._schedule()
                     commit_start = time.time()
 
                 if self.prefix == "task_manager":
@@ -164,6 +154,7 @@ class WorkflowManager(TaskBase):
                 logger.warning("Workflow manager has reached time out while processing running workflows, exiting loop early")
                 ScheduleWorkflowManager().schedule()
                 # Do not process any more workflow jobs. Stop here.
+                # Maybe we should schedule another WorkflowManager run
                 break
             dag = WorkflowDAG(workflow_job)
             status_changed = False
@@ -178,8 +169,8 @@ class WorkflowManager(TaskBase):
                     workflow_job.save(update_fields=['status', 'start_args'])
                     status_changed = True
             else:
-                dnr_nodes = dag.mark_dnr_nodes()
-                WorkflowJobNode.objects.bulk_update(dnr_nodes, ['do_not_run'])
+                workflow_nodes = dag.mark_dnr_nodes()
+                WorkflowJobNode.objects.bulk_update(workflow_nodes, ['do_not_run'])
                 # If workflow is now done, we do special things to mark it as done.
                 is_done = dag.is_workflow_done()
                 if is_done:
@@ -259,7 +250,6 @@ class WorkflowManager(TaskBase):
                         job.status = 'failed'
                         job.save(update_fields=['status', 'job_explanation'])
                         job.websocket_emit_status('failed')
-                        ScheduleWorkflowManager().schedule()
 
                     # TODO: should we emit a status on the socket here similar to tasks.py awx_periodic_scheduler() ?
                     # emit_websocket_notification('/socket.io/jobs', '', dict(id=))
@@ -280,115 +270,184 @@ class WorkflowManager(TaskBase):
 class DependencyManager(TaskBase):
     def __init__(self):
         super().__init__(prefix="dependency_manager")
-        self.all_projects = {}
-        self.all_inventory_sources = {}
 
-    def cache_projects_and_sources(self, task_list):
-        project_ids = set()
+    def create_project_update(self, task, project_id=None):
+        if project_id is None:
+            project_id = task.project_id
+        project_task = Project.objects.get(id=project_id).create_project_update(_eager_fields=dict(launch_type='dependency'))
+
+        # Project created 1 seconds behind
+        project_task.created = task.created - timedelta(seconds=1)
+        project_task.status = 'pending'
+        project_task.save()
+        logger.debug('Spawned {} as dependency of {}'.format(project_task.log_format, task.log_format))
+        return project_task
+
+    def create_inventory_update(self, task, inventory_source_task):
+        inventory_task = InventorySource.objects.get(id=inventory_source_task.id).create_inventory_update(_eager_fields=dict(launch_type='dependency'))
+
+        inventory_task.created = task.created - timedelta(seconds=2)
+        inventory_task.status = 'pending'
+        inventory_task.save()
+        logger.debug('Spawned {} as dependency of {}'.format(inventory_task.log_format, task.log_format))
+
+        return inventory_task
+
+    def add_dependencies(self, task, dependencies):
+        with disable_activity_stream():
+            task.dependent_jobs.add(*dependencies)
+
+    def get_inventory_source_tasks(self):
         inventory_ids = set()
-        for task in task_list:
+        for task in self.all_tasks:
             if isinstance(task, Job):
-                if task.project_id:
-                    project_ids.add(task.project_id)
-                if task.inventory_id:
-                    inventory_ids.add(task.inventory_id)
-            elif isinstance(task, InventoryUpdate):
-                if task.inventory_source and task.inventory_source.source_project_id:
-                    project_ids.add(task.inventory_source.source_project_id)
+                inventory_ids.add(task.inventory_id)
+        self.all_inventory_sources = [invsrc for invsrc in InventorySource.objects.filter(inventory_id__in=inventory_ids, update_on_launch=True)]
 
-        for proj in Project.objects.filter(id__in=project_ids, scm_update_on_launch=True):
-            self.all_projects[proj.id] = proj
+    def get_latest_inventory_update(self, inventory_source):
+        latest_inventory_update = InventoryUpdate.objects.filter(inventory_source=inventory_source).order_by("-created")
+        if not latest_inventory_update.exists():
+            return None
+        return latest_inventory_update.first()
 
-        for invsrc in InventorySource.objects.filter(inventory_id__in=inventory_ids, update_on_launch=True):
-            self.all_inventory_sources.setdefault(invsrc.inventory_id, [])
-            self.all_inventory_sources[invsrc.inventory_id].append(invsrc)
+    def should_update_inventory_source(self, job, latest_inventory_update):
+        now = tz_now()
 
-    @staticmethod
-    def should_update_again(update, cache_timeout):
-        '''
-        If it has never updated, we need to update
-        If there is already an update in progress then we do not need to a new create one
-        If the last update failed, we always need to try and update again
-        If current time is more than cache_timeout after last update, then we need a new one
-        '''
-        if (update is None) or (update.status in ['failed', 'canceled', 'error']):
+        if latest_inventory_update is None:
             return True
-        if update.status in ['waiting', 'pending', 'running']:
+        '''
+        If there's already a inventory update utilizing this job that's about to run
+        then we don't need to create one
+        '''
+        if latest_inventory_update.status in ['waiting', 'pending', 'running']:
             return False
 
-        return bool(((update.finished + timedelta(seconds=cache_timeout))) < tz_now())
+        timeout_seconds = timedelta(seconds=latest_inventory_update.inventory_source.update_cache_timeout)
+        if (latest_inventory_update.finished + timeout_seconds) < now:
+            return True
+        if latest_inventory_update.inventory_source.update_on_launch is True and latest_inventory_update.status in ['failed', 'canceled', 'error']:
+            return True
+        return False
 
-    def get_or_create_project_update(self, project_id):
-        project = self.all_projects.get(project_id, None)
-        if project is not None:
-            latest_project_update = project.project_updates.filter(job_type='check').order_by("-created").first()
-            if self.should_update_again(latest_project_update, project.scm_update_cache_timeout):
-                project_task = project.create_project_update(_eager_fields=dict(launch_type='dependency'))
-                project_task.signal_start()
-                return [project_task]
-            else:
-                return [latest_project_update]
-        return []
+    def get_latest_project_update(self, project_id):
+        latest_project_update = ProjectUpdate.objects.filter(project=project_id, job_type='check').order_by("-created")
+        if not latest_project_update.exists():
+            return None
+        return latest_project_update.first()
+
+    def should_update_related_project(self, job, latest_project_update):
+        now = tz_now()
+
+        if latest_project_update is None:
+            return True
+
+        if latest_project_update.status in ['failed', 'canceled']:
+            return True
+
+        '''
+        If there's already a project update utilizing this job that's about to run
+        then we don't need to create one
+        '''
+        if latest_project_update.status in ['waiting', 'pending', 'running']:
+            return False
+
+        '''
+        If the latest project update has a created time == job_created_time-1
+        then consider the project update found. This is so we don't enter an infinite loop
+        of updating the project when cache timeout is 0.
+        '''
+        if (
+            latest_project_update.project.scm_update_cache_timeout == 0
+            and latest_project_update.launch_type == 'dependency'
+            and latest_project_update.created == job.created - timedelta(seconds=1)
+        ):
+            return False
+        '''
+        Normal Cache Timeout Logic
+        '''
+        timeout_seconds = timedelta(seconds=latest_project_update.project.scm_update_cache_timeout)
+        if (latest_project_update.finished + timeout_seconds) < now:
+            return True
+        return False
 
     def gen_dep_for_job(self, task):
-        dependencies = self.get_or_create_project_update(task.project_id)
+        created_dependencies = []
+        dependencies = []
+        # TODO: Can remove task.project None check after scan-job-default-playbook is removed
+        if task.project is not None and task.project.scm_update_on_launch is True:
+            latest_project_update = self.get_latest_project_update(task.project_id)
+            if self.should_update_related_project(task, latest_project_update):
+                latest_project_update = self.create_project_update(task)
+                created_dependencies.append(latest_project_update)
+            dependencies.append(latest_project_update)
 
+        # Inventory created 2 seconds behind job
         try:
             start_args = json.loads(decrypt_field(task, field_name="start_args"))
         except ValueError:
             start_args = dict()
-        # generator for update-on-launch inventory sources related to this task
-        for inventory_source in self.all_inventory_sources.get(task.inventory_id, []):
+        # generator for inventory sources related to this task
+        task_inv_sources = (invsrc for invsrc in self.all_inventory_sources if invsrc.inventory_id == task.inventory_id)
+        for inventory_source in task_inv_sources:
             if "inventory_sources_already_updated" in start_args and inventory_source.id in start_args['inventory_sources_already_updated']:
                 continue
-            latest_inventory_update = inventory_source.inventory_updates.order_by("-created").first()
-            if self.should_update_again(latest_inventory_update, inventory_source.update_cache_timeout):
-                inventory_task = inventory_source.create_inventory_update(_eager_fields=dict(launch_type='dependency'))
-                inventory_task.signal_start()
+            if not inventory_source.update_on_launch:
+                continue
+            latest_inventory_update = self.get_latest_inventory_update(inventory_source)
+            if self.should_update_inventory_source(task, latest_inventory_update):
+                inventory_task = self.create_inventory_update(task, inventory_source)
+                created_dependencies.append(inventory_task)
                 dependencies.append(inventory_task)
             else:
                 dependencies.append(latest_inventory_update)
 
-        return dependencies
+        if dependencies:
+            self.add_dependencies(task, dependencies)
+
+        return created_dependencies
 
     def gen_dep_for_inventory_update(self, inventory_task):
+        created_dependencies = []
         if inventory_task.source == "scm":
             invsrc = inventory_task.inventory_source
-            if invsrc:
-                return self.get_or_create_project_update(invsrc.source_project_id)
-        return []
+            if not invsrc.source_project.scm_update_on_launch:
+                return created_dependencies
+
+            latest_src_project_update = self.get_latest_project_update(invsrc.source_project_id)
+            if self.should_update_related_project(inventory_task, latest_src_project_update):
+                latest_src_project_update = self.create_project_update(inventory_task, project_id=invsrc.source_project_id)
+                created_dependencies.append(latest_src_project_update)
+            self.add_dependencies(inventory_task, [latest_src_project_update])
+            latest_src_project_update.scm_inventory_updates.add(inventory_task)
+        return created_dependencies
 
     @timeit
     def generate_dependencies(self, undeped_tasks):
-        dependencies = []
-        self.cache_projects_and_sources(undeped_tasks)
+        created_dependencies = []
         for task in undeped_tasks:
             task.log_lifecycle("acknowledged")
             if type(task) is Job:
-                job_deps = self.gen_dep_for_job(task)
+                created_dependencies += self.gen_dep_for_job(task)
             elif type(task) is InventoryUpdate:
-                job_deps = self.gen_dep_for_inventory_update(task)
+                created_dependencies += self.gen_dep_for_inventory_update(task)
             else:
                 continue
-            if job_deps:
-                dependencies += job_deps
-                with disable_activity_stream():
-                    task.dependent_jobs.add(*dependencies)
-                logger.debug(f'Linked {[dep.log_format for dep in dependencies]} as dependencies of {task.log_format}')
-
         UnifiedJob.objects.filter(pk__in=[task.pk for task in undeped_tasks]).update(dependencies_processed=True)
 
-        return dependencies
+        return created_dependencies
+
+    def process_tasks(self):
+        deps = self.generate_dependencies(self.all_tasks)
+        self.generate_dependencies(deps)
+        self.subsystem_metrics.inc(f"{self.prefix}_pending_processed", len(self.all_tasks) + len(deps))
 
     @timeit
     def _schedule(self):
         self.get_tasks(dict(status__in=["pending"], dependencies_processed=False))
 
         if len(self.all_tasks) > 0:
-            deps = self.generate_dependencies(self.all_tasks)
-            undeped_deps = [dep for dep in deps if dep.dependencies_processed is False]
-            self.generate_dependencies(undeped_deps)
-            self.subsystem_metrics.inc(f"{self.prefix}_pending_processed", len(self.all_tasks) + len(undeped_deps))
+            self.get_inventory_source_tasks()
+            self.process_tasks()
             ScheduleTaskManager().schedule()
 
 

awx/main/tasks/__init__.py

@@ -1 +1 @@
-from . import host_metrics, jobs, receptor, system  # noqa
+from . import jobs, receptor, system  # noqa

awx/main/tasks/callback.py

@@ -29,9 +29,8 @@ class RunnerCallback:
         self.safe_env = {}
         self.event_ct = 0
         self.model = model
-        self.update_attempts = int(settings.DISPATCHER_DB_DOWNTIME_TOLERANCE / 5)
+        self.update_attempts = int(settings.DISPATCHER_DB_DOWNTOWN_TOLLERANCE / 5)
         self.wrapup_event_dispatched = False
-        self.artifacts_processed = False
         self.extra_update_fields = {}
 
     def update_model(self, pk, _attempt=0, **updates):
@@ -212,9 +211,6 @@ class RunnerCallback:
             if result_traceback:
                 self.delay_update(result_traceback=result_traceback)
 
-    def artifacts_handler(self, artifact_dir):
-        self.artifacts_processed = True
-
 
 class RunnerCallbackForProjectUpdate(RunnerCallback):
     def __init__(self, *args, **kwargs):

awx/main/tasks/facts.py

@@ -9,7 +9,6 @@ from django.conf import settings
 from django.db.models.query import QuerySet
 from django.utils.encoding import smart_str
 from django.utils.timezone import now
-from django.db import OperationalError
 
 # AWX
 from awx.main.utils.common import log_excess_runtime
@@ -58,28 +57,6 @@ def start_fact_cache(hosts, destination, log_data, timeout=None, inventory_id=No
     return None
 
 
-def raw_update_hosts(host_list):
-    Host.objects.bulk_update(host_list, ['ansible_facts', 'ansible_facts_modified'])
-
-
-def update_hosts(host_list, max_tries=5):
-    if not host_list:
-        return
-    for i in range(max_tries):
-        try:
-            raw_update_hosts(host_list)
-        except OperationalError as exc:
-            # Deadlocks can happen if this runs at the same time as another large query
-            # inventory updates and updating last_job_host_summary are candidates for conflict
-            # but these would resolve easily on a retry
-            if i + 1 < max_tries:
-                logger.info(f'OperationalError (suspected deadlock) saving host facts retry {i}, message: {exc}')
-                continue
-            else:
-                raise
-        break
-
-
 @log_excess_runtime(
     logger,
     debug_cutoff=0.01,
@@ -134,6 +111,7 @@ def finish_fact_cache(hosts, destination, facts_write_time, log_data, job_id=Non
             system_tracking_logger.info('Facts cleared for inventory {} host {}'.format(smart_str(host.inventory.name), smart_str(host.name)))
             log_data['cleared_ct'] += 1
         if len(hosts_to_update) > 100:
-            update_hosts(hosts_to_update)
+            Host.objects.bulk_update(hosts_to_update, ['ansible_facts', 'ansible_facts_modified'])
             hosts_to_update = []
-    update_hosts(hosts_to_update)
+    if hosts_to_update:
+        Host.objects.bulk_update(hosts_to_update, ['ansible_facts', 'ansible_facts_modified'])

awx/main/tasks/host_metrics.py

@@ -1,205 +0,0 @@
-import datetime
-from dateutil.relativedelta import relativedelta
-import logging
-
-from django.conf import settings
-from django.db.models import Count
-from django.db.models.functions import TruncMonth
-from django.utils.timezone import now
-from rest_framework.fields import DateTimeField
-from awx.main.dispatch import get_task_queuename
-from awx.main.dispatch.publish import task
-from awx.main.models.inventory import HostMetric, HostMetricSummaryMonthly
-from awx.conf.license import get_license
-
-logger = logging.getLogger('awx.main.tasks.host_metric_summary_monthly')
-
-
-@task(queue=get_task_queuename)
-def host_metric_summary_monthly():
-    """Run cleanup host metrics summary monthly task each week"""
-    if _is_run_threshold_reached(
-        getattr(settings, 'HOST_METRIC_SUMMARY_TASK_LAST_TS', None), getattr(settings, 'HOST_METRIC_SUMMARY_TASK_INTERVAL', 7) * 86400
-    ):
-        logger.info(f"Executing host_metric_summary_monthly, last ran at {getattr(settings, 'HOST_METRIC_SUMMARY_TASK_LAST_TS', '---')}")
-        HostMetricSummaryMonthlyTask().execute()
-        logger.info("Finished host_metric_summary_monthly")
-
-
-def _is_run_threshold_reached(setting, threshold_seconds):
-    last_time = DateTimeField().to_internal_value(setting) if setting else DateTimeField().to_internal_value('1970-01-01')
-
-    return (now() - last_time).total_seconds() > threshold_seconds
-
-
-class HostMetricSummaryMonthlyTask:
-    """
-    This task computes last [threshold] months of HostMetricSummaryMonthly table
-    [threshold] is setting CLEANUP_HOST_METRICS_HARD_THRESHOLD
-    Each record in the table represents changes in HostMetric table in one month
-    It always overrides all the months newer than <threshold>, never updates older months
-    Algorithm:
-    - hosts_added are HostMetric records with first_automation in given month
-    - hosts_deleted are HostMetric records with deleted=True and last_deleted in given month
-    - - HostMetrics soft-deleted before <threshold> also increases hosts_deleted in their last_deleted month
-    - license_consumed is license_consumed(previous month) + hosts_added - hosts_deleted
-    - - license_consumed for HostMetricSummaryMonthly.date < [threshold] is computed also from
-        all HostMetrics.first_automation < [threshold]
-    - license_capacity is set only for current month, and it's never updated (value taken from current subscription)
-    """
-
-    def __init__(self):
-        self.host_metrics = {}
-        self.processed_month = self._get_first_month()
-        self.existing_summaries = None
-        self.existing_summaries_idx = 0
-        self.existing_summaries_cnt = 0
-        self.records_to_create = []
-        self.records_to_update = []
-
-    def execute(self):
-        self._load_existing_summaries()
-        self._load_hosts_added()
-        self._load_hosts_deleted()
-
-        # Get first month after last hard delete
-        month = self._get_first_month()
-        license_consumed = self._get_license_consumed_before(month)
-
-        # Fill record for each month
-        while month <= datetime.date.today().replace(day=1):
-            summary = self._find_or_create_summary(month)
-            # Update summary and update license_consumed by hosts added/removed this month
-            self._update_summary(summary, month, license_consumed)
-            license_consumed = summary.license_consumed
-
-            month = month + relativedelta(months=1)
-
-        # Create/Update stats
-        HostMetricSummaryMonthly.objects.bulk_create(self.records_to_create, batch_size=1000)
-        HostMetricSummaryMonthly.objects.bulk_update(self.records_to_update, ['license_consumed', 'hosts_added', 'hosts_deleted'], batch_size=1000)
-
-        # Set timestamp of last run
-        settings.HOST_METRIC_SUMMARY_TASK_LAST_TS = now()
-
-    def _get_license_consumed_before(self, month):
-        license_consumed = 0
-        for metric_month, metric in self.host_metrics.items():
-            if metric_month < month:
-                hosts_added = metric.get('hosts_added', 0)
-                hosts_deleted = metric.get('hosts_deleted', 0)
-                license_consumed = license_consumed + hosts_added - hosts_deleted
-            else:
-                break
-        return license_consumed
-
-    def _load_existing_summaries(self):
-        """Find all summaries newer than host metrics delete threshold"""
-        self.existing_summaries = HostMetricSummaryMonthly.objects.filter(date__gte=self._get_first_month()).order_by('date')
-        self.existing_summaries_idx = 0
-        self.existing_summaries_cnt = len(self.existing_summaries)
-
-    def _load_hosts_added(self):
-        """Aggregates hosts added each month, by the 'first_automation' timestamp"""
-        #
-        # -- SQL translation (for better code readability)
-        # SELECT date_trunc('month', first_automation) as month,
-        #        count(first_automation) AS hosts_added
-        # FROM main_hostmetric
-        # GROUP BY month
-        # ORDER by month;
-        result = (
-            HostMetric.objects.annotate(month=TruncMonth('first_automation'))
-            .values('month')
-            .annotate(hosts_added=Count('first_automation'))
-            .values('month', 'hosts_added')
-            .order_by('month')
-        )
-
-        for host_metric in list(result):
-            month = host_metric['month']
-            if month:
-                beginning_of_month = datetime.date(month.year, month.month, 1)
-                if self.host_metrics.get(beginning_of_month) is None:
-                    self.host_metrics[beginning_of_month] = {}
-                self.host_metrics[beginning_of_month]['hosts_added'] = host_metric['hosts_added']
-
-    def _load_hosts_deleted(self):
-        """
-        Aggregates hosts deleted each month, by the 'last_deleted' timestamp.
-        Host metrics have to be deleted NOW to be counted as deleted before
-        (by intention - statistics can change retrospectively by re-automation of previously deleted host)
-        """
-        #
-        # -- SQL translation (for better code readability)
-        # SELECT date_trunc('month', last_deleted) as month,
-        #       count(last_deleted) AS hosts_deleted
-        # FROM main_hostmetric
-        # WHERE deleted = True
-        # GROUP BY 1 # equal to "GROUP BY month"
-        # ORDER by month;
-        result = (
-            HostMetric.objects.annotate(month=TruncMonth('last_deleted'))
-            .values('month')
-            .annotate(hosts_deleted=Count('last_deleted'))
-            .values('month', 'hosts_deleted')
-            .filter(deleted=True)
-            .order_by('month')
-        )
-        for host_metric in list(result):
-            month = host_metric['month']
-            if month:
-                beginning_of_month = datetime.date(month.year, month.month, 1)
-                if self.host_metrics.get(beginning_of_month) is None:
-                    self.host_metrics[beginning_of_month] = {}
-                self.host_metrics[beginning_of_month]['hosts_deleted'] = host_metric['hosts_deleted']
-
-    def _find_or_create_summary(self, month):
-        summary = self._find_summary(month)
-
-        if not summary:
-            summary = HostMetricSummaryMonthly(date=month)
-            self.records_to_create.append(summary)
-        else:
-            self.records_to_update.append(summary)
-        return summary
-
-    def _find_summary(self, month):
-        """
-        Existing summaries are ordered by month ASC.
-        This method is called with month in ascending order too => only 1 traversing is enough
-        """
-        summary = None
-        while not summary and self.existing_summaries_idx < self.existing_summaries_cnt:
-            tmp = self.existing_summaries[self.existing_summaries_idx]
-            if tmp.date < month:
-                self.existing_summaries_idx += 1
-            elif tmp.date == month:
-                summary = tmp
-            elif tmp.date > month:
-                break
-        return summary
-
-    def _update_summary(self, summary, month, license_consumed):
-        """Updates the metric with hosts added and deleted and set license info for current month"""
-        # Get month counts from host metrics, zero if not found
-        hosts_added, hosts_deleted = 0, 0
-        if metric := self.host_metrics.get(month, None):
-            hosts_added = metric.get('hosts_added', 0)
-            hosts_deleted = metric.get('hosts_deleted', 0)
-
-        summary.license_consumed = license_consumed + hosts_added - hosts_deleted
-        summary.hosts_added = hosts_added
-        summary.hosts_deleted = hosts_deleted
-
-        # Set subscription count for current month
-        if month == datetime.date.today().replace(day=1):
-            license_info = get_license()
-            summary.license_capacity = license_info.get('instance_count', 0)
-        return summary
-
-    @staticmethod
-    def _get_first_month():
-        """Returns first month after host metrics hard delete threshold"""
-        threshold = getattr(settings, 'CLEANUP_HOST_METRICS_HARD_THRESHOLD', 36)
-        return datetime.date.today().replace(day=1) - relativedelta(months=int(threshold) - 1)

awx/main/tasks/jobs.py

@@ -112,7 +112,7 @@ class BaseTask(object):
 
     def __init__(self):
         self.cleanup_paths = []
-        self.update_attempts = int(settings.DISPATCHER_DB_DOWNTIME_TOLERANCE / 5)
+        self.update_attempts = int(settings.DISPATCHER_DB_DOWNTOWN_TOLLERANCE / 5)
         self.runner_callback = self.callback_class(model=self.model)
 
     def update_model(self, pk, _attempt=0, **updates):
@@ -290,6 +290,13 @@ class BaseTask(object):
             content = safe_dump(vars, safe_dict)
         return self.write_private_data_file(private_data_dir, 'extravars', content, sub_dir='env')
 
+    def add_awx_venv(self, env):
+        env['VIRTUAL_ENV'] = settings.AWX_VENV_PATH
+        if 'PATH' in env:
+            env['PATH'] = os.path.join(settings.AWX_VENV_PATH, "bin") + ":" + env['PATH']
+        else:
+            env['PATH'] = os.path.join(settings.AWX_VENV_PATH, "bin")
+
     def build_env(self, instance, private_data_dir, private_data_files=None):
         """
         Build environment dictionary for ansible-playbook.
@@ -919,7 +926,6 @@ class RunJob(SourceControlMixin, BaseTask):
         path_vars = (
             ('ANSIBLE_COLLECTIONS_PATHS', 'collections_paths', 'requirements_collections', '~/.ansible/collections:/usr/share/ansible/collections'),
             ('ANSIBLE_ROLES_PATH', 'roles_path', 'requirements_roles', '~/.ansible/roles:/usr/share/ansible/roles:/etc/ansible/roles'),
-            ('ANSIBLE_COLLECTIONS_PATH', 'collections_path', 'requirements_collections', '~/.ansible/collections:/usr/share/ansible/collections'),
         )
 
         config_values = read_ansible_config(os.path.join(private_data_dir, 'project'), list(map(lambda x: x[1], path_vars)))
@@ -1094,7 +1100,7 @@ class RunJob(SourceControlMixin, BaseTask):
             # actual `run()` call; this _usually_ means something failed in
             # the pre_run_hook method
             return
-        if self.should_use_fact_cache() and self.runner_callback.artifacts_processed:
+        if self.should_use_fact_cache():
             job.log_lifecycle("finish_job_fact_cache")
             finish_fact_cache(
                 job.get_hosts_for_fact_cache(),
@@ -1262,7 +1268,7 @@ class RunProjectUpdate(BaseTask):
 
         galaxy_creds_are_defined = project_update.project.organization and project_update.project.organization.galaxy_credentials.exists()
         if not galaxy_creds_are_defined and (settings.AWX_ROLES_ENABLED or settings.AWX_COLLECTIONS_ENABLED):
-            logger.warning('Galaxy role/collection syncing is enabled, but no credentials are configured for {project_update.project.organization}.')
+            logger.warning('Galaxy role/collection syncing is enabled, but no ' f'credentials are configured for {project_update.project.organization}.')
 
         extra_vars.update(
             {

awx/main/tasks/receptor.py

@@ -464,7 +464,6 @@ class AWXReceptorJob:
             event_handler=self.task.runner_callback.event_handler,
             finished_callback=self.task.runner_callback.finished_callback,
             status_handler=self.task.runner_callback.status_handler,
-            artifacts_handler=self.task.runner_callback.artifacts_handler,
             **self.runner_params,
         )
 
@@ -640,11 +639,11 @@ class AWXReceptorJob:
 #
 RECEPTOR_CONFIG_STARTER = (
     {'local-only': None},
-    {'log-level': settings.RECEPTOR_LOG_LEVEL},
+    {'log-level': 'debug'},
     {'node': {'firewallrules': [{'action': 'reject', 'tonode': settings.CLUSTER_HOST_ID, 'toservice': 'control'}]}},
     {'control-service': {'service': 'control', 'filename': '/var/run/receptor/receptor.sock', 'permissions': '0660'}},
     {'work-command': {'worktype': 'local', 'command': 'ansible-runner', 'params': 'worker', 'allowruntimeparams': True}},
-    {'work-signing': {'privatekey': '/etc/receptor/work_private_key.pem', 'tokenexpiration': '1m'}},
+    {'work-signing': {'privatekey': '/etc/receptor/signing/work-private-key.pem', 'tokenexpiration': '1m'}},
     {
         'work-kubernetes': {
             'worktype': 'kubernetes-runtime-auth',
@@ -666,7 +665,7 @@ RECEPTOR_CONFIG_STARTER = (
     {
         'tls-client': {
             'name': 'tlsclient',
-            'rootcas': '/etc/receptor/tls/ca/mesh-CA.crt',
+            'rootcas': '/etc/receptor/tls/ca/receptor-ca.crt',
             'cert': '/etc/receptor/tls/receptor.crt',
             'key': '/etc/receptor/tls/receptor.key',
             'mintls13': False,

awx/main/tasks/system.py

@@ -2,7 +2,6 @@
 from collections import namedtuple
 import functools
 import importlib
-import itertools
 import json
 import logging
 import os
@@ -15,7 +14,7 @@ from datetime import datetime
 
 # Django
 from django.conf import settings
-from django.db import connection, transaction, DatabaseError, IntegrityError
+from django.db import transaction, DatabaseError, IntegrityError
 from django.db.models.fields.related import ForeignKey
 from django.utils.timezone import now, timedelta
 from django.utils.encoding import smart_str
@@ -49,7 +48,6 @@ from awx.main.models import (
     SmartInventoryMembership,
     Job,
     HostMetric,
-    convert_jsonfields,
 )
 from awx.main.constants import ACTIVE_STATES
 from awx.main.dispatch.publish import task
@@ -88,11 +86,6 @@ def dispatch_startup():
     if settings.IS_K8S:
         write_receptor_config()
 
-    try:
-        convert_jsonfields()
-    except Exception:
-        logger.exception("Failed json field conversion, skipping.")
-
     startup_logger.debug("Syncing Schedules")
     for sch in Schedule.objects.all():
         try:
@@ -136,52 +129,6 @@ def inform_cluster_of_shutdown():
         logger.exception('Encountered problem with normal shutdown signal.')
 
 
-@task(queue=get_task_queuename)
-def migrate_jsonfield(table, pkfield, columns):
-    batchsize = 10000
-    with advisory_lock(f'json_migration_{table}', wait=False) as acquired:
-        if not acquired:
-            return
-
-        from django.db.migrations.executor import MigrationExecutor
-
-        # If Django is currently running migrations, wait until it is done.
-        while True:
-            executor = MigrationExecutor(connection)
-            if not executor.migration_plan(executor.loader.graph.leaf_nodes()):
-                break
-            time.sleep(120)
-
-        logger.warning(f"Migrating json fields for {table}: {', '.join(columns)}")
-
-        with connection.cursor() as cursor:
-            for i in itertools.count(0, batchsize):
-                # Are there even any rows in the table beyond this point?
-                cursor.execute(f"select count(1) from {table} where {pkfield} >= %s limit 1;", (i,))
-                if not cursor.fetchone()[0]:
-                    break
-
-                column_expr = ', '.join(f"{colname} = {colname}_old::jsonb" for colname in columns)
-                # If any of the old columns have non-null values, the data needs to be cast and copied over.
-                empty_expr = ' or '.join(f"{colname}_old is not null" for colname in columns)
-                cursor.execute(  # Only clobber the new fields if there is non-null data in the old ones.
-                    f"""
-                    update {table}
-                      set {column_expr}
-                      where {pkfield} >= %s and {pkfield} < %s
-                        and {empty_expr};
-                    """,
-                    (i, i + batchsize),
-                )
-                rows = cursor.rowcount
-                logger.debug(f"Batch {i} to {i + batchsize} copied on {table}, {rows} rows affected.")
-
-            column_expr = ', '.join(f"DROP COLUMN {column}_old" for column in columns)
-            cursor.execute(f"ALTER TABLE {table} {column_expr};")
-
-        logger.warning(f"Migration of {table} to jsonb is finished.")
-
-
 @task(queue=get_task_queuename)
 def apply_cluster_membership_policies():
     from awx.main.signals import disable_activity_stream
@@ -369,8 +316,13 @@ def send_notifications(notification_list, job_id=None):
 @task(queue=get_task_queuename)
 def gather_analytics():
     from awx.conf.models import Setting
+    from rest_framework.fields import DateTimeField
 
-    if is_run_threshold_reached(Setting.objects.filter(key='AUTOMATION_ANALYTICS_LAST_GATHER').first(), settings.AUTOMATION_ANALYTICS_GATHER_INTERVAL):
+    last_gather = Setting.objects.filter(key='AUTOMATION_ANALYTICS_LAST_GATHER').first()
+    last_time = DateTimeField().to_internal_value(last_gather.value) if last_gather and last_gather.value else None
+    gather_time = now()
+
+    if not last_time or ((gather_time - last_time).total_seconds() > settings.AUTOMATION_ANALYTICS_GATHER_INTERVAL):
         analytics.gather()
 
 
@@ -429,25 +381,16 @@ def cleanup_images_and_files():
 
 @task(queue=get_task_queuename)
 def cleanup_host_metrics():
-    """Run cleanup host metrics ~each month"""
-    # TODO: move whole method to host_metrics in follow-up PR
     from awx.conf.models import Setting
-
-    if is_run_threshold_reached(
-        Setting.objects.filter(key='CLEANUP_HOST_METRICS_LAST_TS').first(), getattr(settings, 'CLEANUP_HOST_METRICS_INTERVAL', 30) * 86400
-    ):
-        months_ago = getattr(settings, 'CLEANUP_HOST_METRICS_SOFT_THRESHOLD', 12)
-        logger.info("Executing cleanup_host_metrics")
-        HostMetric.cleanup_task(months_ago)
-        logger.info("Finished cleanup_host_metrics")
-
-
-def is_run_threshold_reached(setting, threshold_seconds):
     from rest_framework.fields import DateTimeField
 
-    last_time = DateTimeField().to_internal_value(setting.value) if setting and setting.value else DateTimeField().to_internal_value('1970-01-01')
+    last_cleanup = Setting.objects.filter(key='CLEANUP_HOST_METRICS_LAST_TS').first()
+    last_time = DateTimeField().to_internal_value(last_cleanup.value) if last_cleanup and last_cleanup.value else None
 
-    return (now() - last_time).total_seconds() > threshold_seconds
+    cleanup_interval_secs = getattr(settings, 'CLEANUP_HOST_METRICS_INTERVAL', 30) * 86400
+    if not last_time or ((now() - last_time).total_seconds() > cleanup_interval_secs):
+        months_ago = getattr(settings, 'CLEANUP_HOST_METRICS_THRESHOLD', 12)
+        HostMetric.cleanup_task(months_ago)
 
 
 @task(queue=get_task_queuename)
@@ -598,7 +541,7 @@ def cluster_node_heartbeat(dispatch_time=None, worker_tasks=None):
             logger.warning(f'Heartbeat skew - interval={(nowtime - last_last_seen).total_seconds():.4f}, expected={settings.CLUSTER_NODE_HEARTBEAT_PERIOD}')
     else:
         if settings.AWX_AUTO_DEPROVISION_INSTANCES:
-            (changed, this_inst) = Instance.objects.register(ip_address=os.environ.get('MY_POD_IP'), node_type='control', node_uuid=settings.SYSTEM_UUID)
+            (changed, this_inst) = Instance.objects.register(ip_address=os.environ.get('MY_POD_IP'), node_type='control', uuid=settings.SYSTEM_UUID)
             if changed:
                 logger.warning(f'Recreated instance record {this_inst.hostname} after unexpected removal')
             this_inst.local_health_check()
@@ -896,7 +839,10 @@ def delete_inventory(inventory_id, user_id, retries=5):
             user = None
     with ignore_inventory_computed_fields(), ignore_inventory_group_removal(), impersonate(user):
         try:
-            Inventory.objects.get(id=inventory_id).delete()
+            i = Inventory.objects.get(id=inventory_id)
+            for host in i.hosts.iterator():
+                host.job_events_as_primary_host.update(host=None)
+            i.delete()
             emit_channel_notification('inventories-status_changed', {'group_name': 'inventories', 'inventory_id': inventory_id, 'status': 'deleted'})
             logger.debug('Deleted inventory {} as user {}.'.format(inventory_id, user_id))
         except Inventory.DoesNotExist:

awx/main/tests/docs/test_swagger_generation.py

@@ -7,7 +7,7 @@ from django.core.serializers.json import DjangoJSONEncoder
 from django.utils.functional import Promise
 from django.utils.encoding import force_str
 
-from drf_yasg.codecs import OpenAPICodecJson
+from openapi_codec.encode import generate_swagger_object
 import pytest
 
 from awx.api.versioning import drf_reverse
@@ -43,12 +43,12 @@ class TestSwaggerGeneration:
     @pytest.fixture(autouse=True, scope='function')
     def _prepare(self, get, admin):
         if not self.__class__.JSON:
-            url = drf_reverse('api:schema-swagger-ui') + '?format=openapi'
+            url = drf_reverse('api:swagger_view') + '?format=openapi'
             response = get(url, user=admin)
-            codec = OpenAPICodecJson([])
-            data = codec.generate_swagger_object(response.data)
+            data = generate_swagger_object(response.data)
             if response.has_header('X-Deprecated-Paths'):
                 data['deprecated_paths'] = json.loads(response['X-Deprecated-Paths'])
+            data.update(response.accepted_renderer.get_customizations() or {})
 
             data['host'] = None
             data['schemes'] = ['https']
@@ -60,21 +60,12 @@ class TestSwaggerGeneration:
                 # change {version} in paths to the actual default API version (e.g., v2)
                 revised_paths[path.replace('{version}', settings.REST_FRAMEWORK['DEFAULT_VERSION'])] = node
                 for method in node:
-                    # Ignore any parameters methods, these cause issues because it can come as an array instead of a dict
-                    # Which causes issues in the last for loop in here
-                    if method == 'parameters':
-                        continue
-
                     if path in deprecated_paths:
                         node[method]['deprecated'] = True
                     if 'description' in node[method]:
                         # Pop off the first line and use that as the summary
                         lines = node[method]['description'].splitlines()
-                        # If there was a description then set the summary as the description, otherwise make something up
-                        if lines:
-                            node[method]['summary'] = lines.pop(0).strip('#:')
-                        else:
-                            node[method]['summary'] = f'No Description for {method} on {path}'
+                        node[method]['summary'] = lines.pop(0).strip('#:')
                         node[method]['description'] = '\n'.join(lines)
 
                     # remove the required `version` parameter
@@ -99,13 +90,13 @@ class TestSwaggerGeneration:
         # The number of API endpoints changes over time, but let's just check
         # for a reasonable number here; if this test starts failing, raise/lower the bounds
         paths = JSON['paths']
-        assert 250 < len(paths) < 375
-        assert set(list(paths['/api/'].keys())) == set(['get', 'parameters'])
-        assert set(list(paths['/api/v2/'].keys())) == set(['get', 'parameters'])
-        assert set(list(sorted(paths['/api/v2/credentials/'].keys()))) == set(['get', 'post', 'parameters'])
-        assert set(list(sorted(paths['/api/v2/credentials/{id}/'].keys()))) == set(['delete', 'get', 'patch', 'put', 'parameters'])
-        assert set(list(paths['/api/v2/settings/'].keys())) == set(['get', 'parameters'])
-        assert set(list(paths['/api/v2/settings/{category_slug}/'].keys())) == set(['get', 'put', 'patch', 'delete', 'parameters'])
+        assert 250 < len(paths) < 350
+        assert list(paths['/api/'].keys()) == ['get']
+        assert list(paths['/api/v2/'].keys()) == ['get']
+        assert list(sorted(paths['/api/v2/credentials/'].keys())) == ['get', 'post']
+        assert list(sorted(paths['/api/v2/credentials/{id}/'].keys())) == ['delete', 'get', 'patch', 'put']
+        assert list(paths['/api/v2/settings/'].keys()) == ['get']
+        assert list(paths['/api/v2/settings/{category_slug}/'].keys()) == ['get', 'put', 'patch', 'delete']
 
     @pytest.mark.parametrize(
         'path',
@@ -171,8 +162,4 @@ class TestSwaggerGeneration:
             data = re.sub(r'[0-9]{4}-[0-9]{2}-[0-9]{2}(T|\s)[0-9]{2}:[0-9]{2}:[0-9]{2}.[0-9]+(Z|\+[0-9]{2}:[0-9]{2})?', r'2018-02-01T08:00:00.000000Z', data)
             data = re.sub(r'''(\s+"client_id": ")([a-zA-Z0-9]{40})("\,\s*)''', r'\1xxxx\3', data)
             data = re.sub(r'"action_node": "[^"]+"', '"action_node": "awx"', data)
-
-            # replace uuids to prevent needless diffs
-            pattern = r'[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}'
-            data = re.sub(pattern, r'00000000-0000-0000-0000-000000000000', data)
             f.write(data)

awx/main/tests/factories/fixtures.py

@@ -1,9 +1,6 @@
 import json
 
 from django.contrib.auth.models import User
-from django.core.exceptions import ValidationError
-
-from unittest import mock
 
 from awx.main.models import (
     Organization,
@@ -23,7 +20,6 @@ from awx.main.models import (
     WorkflowJobNode,
     WorkflowJobTemplateNode,
 )
-from awx.main.models.inventory import HostMetric, HostMetricSummaryMonthly
 
 # mk methods should create only a single object of a single type.
 # they should also have the option of being persisted or not.
@@ -252,42 +248,3 @@ def mk_workflow_job_node(unified_job_template=None, success_nodes=None, failure_
     if persisted:
         workflow_node.save()
     return workflow_node
-
-
-def mk_host_metric(hostname, first_automation, last_automation=None, last_deleted=None, deleted=False, persisted=True):
-    ok, idx = False, 1
-    while not ok:
-        try:
-            with mock.patch("django.utils.timezone.now") as mock_now:
-                mock_now.return_value = first_automation
-                metric = HostMetric(
-                    hostname=hostname or f"host-{first_automation}-{idx}",
-                    first_automation=first_automation,
-                    last_automation=last_automation or first_automation,
-                    last_deleted=last_deleted,
-                    deleted=deleted,
-                )
-                metric.validate_unique()
-                if persisted:
-                    metric.save()
-                ok = True
-        except ValidationError as e:
-            # Repeat create for auto-generated hostname
-            if not hostname and e.message_dict.get('hostname', None):
-                idx += 1
-            else:
-                raise e
-
-
-def mk_host_metric_summary(date, license_consumed=0, license_capacity=0, hosts_added=0, hosts_deleted=0, indirectly_managed_hosts=0, persisted=True):
-    summary = HostMetricSummaryMonthly(
-        date=date,
-        license_consumed=license_consumed,
-        license_capacity=license_capacity,
-        hosts_added=hosts_added,
-        hosts_deleted=hosts_deleted,
-        indirectly_managed_hosts=indirectly_managed_hosts,
-    )
-    if persisted:
-        summary.save()
-    return summary