Refactor task enqueuing

appengine/gce-backend/instances.py

 # Copyright 2016 The LUCI Authors. All rights reserved.
 # Use of this source code is governed under the Apache License, Version 2.0
 # that can be found in the LICENSE file.
 
 """Utilities for operating on instances."""
 
 import json
 import logging
 
 from google.appengine.ext import ndb
 
 from components import gce
 from components import net
 from components import pubsub
-from components import utils
 
 import instance_group_managers
 import metrics
 import models
 import utilities
 
 
 def get_instance_key(base_name, revision, zone, instance_name):
   """Returns a key for an Instance.
 
@@ skipping 198 matching lines @@
       max_concurrent=max_concurrent,
   )
 
   instance_group_managers.set_instances(key, keys.values())
 
 
 def schedule_fetch():
   """Enqueues tasks to fetch instances."""
   for instance_group_manager in models.InstanceGroupManager.query():
     if instance_group_manager.url:
-      if not utils.enqueue_task(
-          '/internal/queues/fetch-instances',
-          'fetch-instances',
-          params={
-              'key': instance_group_manager.key.urlsafe(),
-          },
-      ):
-        logging.warning(
-            'Failed to enqueue task for InstanceGroupManager: %s',
-            instance_group_manager.key,
-        )
+      utilities.enqueue_task('fetch-instances', instance_group_manager.key)
 
 
 def _delete(instance_template_revision, instance_group_manager, instance):
   """Deletes the given instance.
 
   Args:
     instance_template_revision: models.InstanceTemplateRevision.
     instance_group_manager: models.InstanceGroupManager.
     instance: models.Instance
   """
@@ skipping 60 matching lines @@
     )
     return
 
   _delete(instance_template_revision, instance_group_manager, instance)
 
 
 def schedule_pending_deletion():
   """Enqueues tasks to delete instances."""
   for instance in models.Instance.query():
     if instance.pending_deletion and not instance.deleted:
-      if not utils.enqueue_task(
-          '/internal/queues/delete-instance-pending-deletion',
-          'delete-instance-pending-deletion',
-          params={
-              'key': instance.key.urlsafe(),
-          },
-      ):
-        logging.warning('Failed to enqueue task for Instance: %s', instance.key)
+      utilities.enqueue_task('delete-instance-pending-deletion', instance.key)
 
 
 def delete_drained(key):
   """Deletes the given drained instance.
 
   Args:
     key: ndb.Key for a models.Instance entity.
   """
   instance = key.get()
   if not instance:
@@ skipping 49 matching lines @@
 
 def schedule_drained_deletion():
   """Enqueues tasks to delete drained instances."""
   for instance_group_manager_key in (
       instance_group_managers.get_drained_instance_group_managers()):
     instance_group_manager = instance_group_manager_key.get()
     if instance_group_manager:
       for instance_key in instance_group_manager.instances:
         instance = instance_key.get()
         if instance and not instance.cataloged:
-          if not utils.enqueue_task(
-              '/internal/queues/delete-drained-instance',
-              'delete-drained-instance',
-              params={
-                  'key': instance.key.urlsafe(),
-              },
-          ):
-            logging.warning(
-              'Failed to enqueue task for Instance: %s', instance.key)
+          utilities.enqueue_task('delete-drained-instance', instance.key)