Don't retry cleanup tasks (#4633)

# What this PR does
Disable retry on cleanup tasks. Since these will be scheduled to occur
again on a regular interval minimize the tasks left on queue. These are
not critical tasks.

## Which issue(s) this PR closes

Closes [issue link here]

<!--
*Note*: if you have more than one GitHub issue that this PR closes, be
sure to preface
each issue link with a [closing
keyword](https://docs.github.com/en/get-started/writing-on-github/working-with-advanced-formatting/using-keywords-in-issues-and-pull-requests#linking-a-pull-request-to-an-issue).
This ensures that the issue(s) are auto-closed once the PR has been
merged.
-->

## Checklist

- [ ] Unit, integration, and e2e (if applicable) tests updated
- [ ] Documentation added (or `pr:no public docs` PR label added if not
required)
- [ ] Added the relevant release notes label (see labels prefixed w/
`release:`). These labels dictate how your PR will
    show up in the autogenerated release notes.

engine/apps/grafana_plugin/tasks/sync.py

@@ -89,7 +89,7 @@ def run_organization_sync(organization_pk, force_sync):
     logger.info(f"Finish sync Organization {organization_pk}")
 
 
-@shared_dedicated_queue_retry_task(autoretry_for=(Exception,), max_retries=1)
+@shared_dedicated_queue_retry_task(autoretry_for=(Exception,), max_retries=0)
 def start_cleanup_deleted_organizations():
     sync_threshold = timezone.now() - INACTIVE_PERIOD
 
@@ -113,7 +113,7 @@ def start_cleanup_deleted_organizations():
         cleanup_organization_async.apply_async((organization_pk,), countdown=countdown)
 
 
-@shared_dedicated_queue_retry_task(autoretry_for=(Exception,), max_retries=1)
+@shared_dedicated_queue_retry_task(autoretry_for=(Exception,), max_retries=0)
 def cleanup_organization_async(organization_pk):
     cleanup_organization(organization_pk)
 
@@ -166,9 +166,11 @@ def cleanup_empty_deleted_integrations(organization_pk, dry_run=True):
             integration.hard_delete()
 
 
-@shared_dedicated_queue_retry_task(autoretry_for=(Exception,), max_retries=1)
+@shared_dedicated_queue_retry_task(autoretry_for=(Exception,), max_retries=0)
 def start_cleanup_organizations():
-    organization_pks = Organization.objects.all().values_list("pk", flat=True)
+    cleanup_threshold = timezone.now() - INACTIVE_PERIOD
+    organization_qs = Organization.objects.filter(last_time_synced__lte=cleanup_threshold)
+    organization_pks = organization_qs.values_list("pk", flat=True)
     logger.debug(f"Found {len(organization_pks)} organizations")
     max_countdown = CLEANUP_PERIOD.seconds
     for idx, organization_pk in enumerate(organization_pks):