chore(controller): add celery retry mechanism

Author: duanhongyi

rootfs/api/tasks.py

@@ -13,89 +13,106 @@
 logger = logging.getLogger(__name__)
 
 
-@shared_task
-def retrieve_resource(resource):
+@shared_task(bind=True)
+def retrieve_resource(self, resource):
     task_id = uuid.uuid4().hex
     signals.request_started.send(sender=task_id)
     try:
         if not resource.retrieve():
             t = time.time() - resource.created.timestamp()
             if t < 3600:
-                retrieve_resource.apply_async(
-                    args=(resource, ),
-                    eta=now() + timedelta(seconds=30))
+                raise self.retry(exc=None, countdown=30)
             elif t < 3600 * 12:
-                retrieve_resource.apply_async(
-                    args=(resource, ),
-                    eta=now() + timedelta(seconds=1800))
+                raise self.retry(exc=None, countdown=1800)
             else:
                 resource.detach_resource()
     except Resource.DoesNotExist:
-        logger.exception("retrieve task not found resource: {}".format(resource.id))  # noqa
+        logger.exception(
+            "retrieve task not found resource: {}".format(resource.id))
     finally:
         signals.request_finished.send(sender=task_id)
 
 
-@shared_task
+@shared_task(
+    autoretry_for=(Exception, ),
+    retry_backoff=8,
+    retry_jitter=True,
+    retry_backoff_max=3600,
+    retry_kwargs={'max_retries': None}
+)
 def measure_config(config: List[Dict[str, str]]):
     task_id = uuid.uuid4().hex
     signals.request_started.send(sender=task_id)
     try:
         measurement = manager.Measurement()
         measurement.post_config(config)
-    except Exception as e:
-        logger.exception("write influxdb point fail: {}".format(e))
     finally:
         signals.request_finished.send(sender=task_id)
 
 
-@shared_task
+@shared_task(
+    autoretry_for=(Exception, ),
+    retry_backoff=8,
+    retry_jitter=True,
+    retry_backoff_max=3600,
+    retry_kwargs={'max_retries': None}
+)
 def measure_volumes(volumes: List[Dict[str, str]]):
     task_id = uuid.uuid4().hex
     signals.request_started.send(sender=task_id)
     try:
         measurement = manager.Measurement()
         measurement.post_volumes(volumes)
-    except Exception as e:
-        logger.exception("write influxdb point fail: {}".format(e))
     finally:
         signals.request_finished.send(sender=task_id)
 
 
-@shared_task
+@shared_task(
+    autoretry_for=(Exception, ),
+    retry_backoff=8,
+    retry_jitter=True,
+    retry_backoff_max=3600,
+    retry_kwargs={'max_retries': None}
+)
 def measure_networks(networks: List[Dict[str, str]]):
     task_id = uuid.uuid4().hex
     signals.request_started.send(sender=task_id)
     try:
         measurement = manager.Measurement()
         measurement.post_networks(networks)
-    except Exception as e:
-        logger.exception("write influxdb point fail: {}".format(e))
     finally:
         signals.request_finished.send(sender=task_id)
 
 
-@shared_task
+@shared_task(
+    autoretry_for=(Exception, ),
+    retry_backoff=8,
+    retry_jitter=True,
+    retry_backoff_max=3600,
+    retry_kwargs={'max_retries': None}
+)
 def measure_instances(instances: List[Dict[str, str]]):
     task_id = uuid.uuid4().hex
     signals.request_started.send(sender=task_id)
     try:
         measurement = manager.Measurement()
         measurement.post_instances(instances)
-    except Exception as e:
-        logger.exception("write influxdb point fail: {}".format(e))
     finally:
         signals.request_finished.send(sender=task_id)
 
 
-@shared_task
+@shared_task(
+    autoretry_for=(Exception, ),
+    retry_backoff=8,
+    retry_jitter=True,
+    retry_backoff_max=3600,
+    retry_kwargs={'max_retries': None}
+)
 def measure_resources(resources: List[Dict[str, str]]):
     task_id = uuid.uuid4().hex
     signals.request_started.send(sender=task_id)
     try:
         measurement = manager.Measurement()
         measurement.post_resources(resources)
-    except Exception as e:
-        logger.exception("write influxdb point fail: {}".format(e))
     finally:
         signals.request_finished.send(sender=task_id)