Refactor task enqueuing

appengine/gce-backend/catalog.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 interacting with the Machine Provider catalog."""
 
 import json
 import logging
 
 from google.appengine.ext import ndb
 from protorpc.remote import protojson
 
 from components import gce
 from components import machine_provider
 from components import net
-from components import utils
 
 import instances
 import instance_group_managers
 import metrics
 import models
+import utilities
 
 
 def get_policies(key, service_account):
   """Returns Machine Provider policies governing the given instance.
 
   Args:
     key: ndb.Key for a models.Instance entity.
     service_account: Name of the service account the instance will use to
       talk to Machine Provider.
   """
@@ skipping 126 matching lines @@
     if instance_template.active:
       instance_template_revision = instance_template.active.get()
       if instance_template_revision:
         for instance_group_manager_key in instance_template_revision.active:
           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:
                 if not instance.cataloged and not instance.pending_deletion:
-                  if not utils.enqueue_task(
-                      '/internal/queues/catalog-instance',
-                      'catalog-instance',
-                      params={
-                          'key': instance.key.urlsafe(),
-                      },
-                  ):
-                    logging.warning(
-                        'Failed to enqueue task for Instance: %s', instance.key)
+                  utilities.enqueue_task('catalog-instance', instance.key)
 
 
 def remove(key):
   """Removes the given instance from the catalog.
 
   Args:
     key: ndb.Key for a models.Instance entity.
   """
   instance = key.get()
   if not instance:
@@ skipping 18 matching lines @@
 
 def schedule_removal():
   """Enqueues tasks to remove drained instances from the catalog."""
   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 instance.cataloged:
-          if not utils.enqueue_task(
-              '/internal/queues/remove-cataloged-instance',
-              'remove-cataloged-instance',
-              params={
-                  'key': instance.key.urlsafe(),
-              },
-          ):
-            logging.warning(
-                'Failed to enqueue task for Instance: %s', instance.key)
+          utilities.enqueue_task('remove-cataloged-instance', instance.key)
 
 
 def update_cataloged_instance(key):
   """Updates an Instance based on its state in Machine Provider's catalog.
 
   Args:
     key: ndb.Key for a models.Instance entity.
   """
   instance = key.get()
   if not instance:
@@ skipping 15 matching lines @@
       )
       metrics.send_machine_event('METADATA_UPDATE_PROPOSED', instance.hostname)
   except net.NotFoundError:
     instances.mark_for_deletion(key)
 
 
 def schedule_cataloged_instance_update():
   """Enqueues tasks to update information about cataloged instances."""
   for instance in models.Instance.query():
     if instance.cataloged:
-      if not utils.enqueue_task(
-          '/internal/queues/update-cataloged-instance',
-          'update-cataloged-instance',
-          params={
-              'key': instance.key.urlsafe(),
-          },
-      ):
-        logging.warning('Failed to enqueue task for Instance: %s', instance.key)
+      utilities.enqueue_task('update-cataloged-instance', instance.key)