[GCP Functions] Retry function creation on failure to improve reliability

Author: JosepSampe

lithops/job/job.py

@@ -177,8 +177,6 @@ def _create_job(
     """
     Creates a new Job
     """
-    global FUNCTION_CACHE
-
     ext_env = {} if extra_env is None else extra_env.copy()
     if ext_env:
         ext_env = utils.convert_bools_to_string(ext_env)

lithops/serverless/backends/gcp_functions/config.py

@@ -29,9 +29,6 @@
 RUNTIME_MEMORY_MAX = 8192  # 8GB
 RUNTIME_MEMORY_OPTIONS = {128, 256, 512, 1024, 2048, 4096, 8192}
 
-RETRIES = 5
-RETRY_SLEEP = 20
-
 AVAILABLE_PY_RUNTIMES = {
     '3.7': 'python37',
     '3.8': 'python38',
@@ -49,7 +46,9 @@
     'max_workers': 1000,
     'worker_processes': 1,
     'invoke_pool_threads': 1000,
-    'trigger': 'pub/sub'
+    'trigger': 'pub/sub',
+    'retries': 5,
+    'retry_sleep': 10,
 }
 
 REQUIREMENTS_FILE = """
@@ -103,8 +102,5 @@ def load_config(config_data=None):
     if config_data['gcp_functions']['runtime_memory'] > RUNTIME_MEMORY_MAX:
         config_data['gcp_functions']['runtime_memory'] = RUNTIME_MEMORY_MAX
 
-    config_data['gcp_functions']['retries'] = RETRIES
-    config_data['gcp_functions']['retry_sleep'] = RETRY_SLEEP
-
     if 'region' not in config_data['gcp']:
         config_data['gcp']['region'] = config_data['gcp_functions']['region']

lithops/serverless/backends/gcp_functions/gcp_functions.py

@@ -244,10 +244,19 @@ def _create_function(self, runtime_name, memory, timeout=60):
                 'failurePolicy': {}
             }
 
-        operation = self._api_resource.projects().locations().functions().create(
-            location=self._default_location,
-            body=cloud_function
-        ).execute(num_retries=self.num_retries)
+        logger.info(f'Deploying function {function_location}')
+        for attempt in range(self.num_retries):
+            try:
+                operation = self._api_resource.projects().locations().functions().create(
+                    location=self._default_location,
+                    body=cloud_function
+                ).execute()
+                break
+            except Exception as e:
+                if attempt < self.num_retries - 1:
+                    time.sleep(self.retry_sleep)
+                else:
+                    raise Exception(f"Failed to create Cloud Function after {self.num_retries} attempts.") from e
 
         # Wait until the function is completely deployed
         logger.info('Waiting for the function to be deployed')

lithops/standalone/worker.py

@@ -146,8 +146,6 @@ def notify_task_done(job_key, call_id):
 
 
 def redis_queue_consumer(pid, work_queue_name, exec_mode, backend):
-    global worker_threads
-
     worker_threads[pid]['status'] = WorkerStatus.IDLE.value
 
     logger.info(f"Redis consumer process {pid} started")
@@ -213,7 +211,6 @@ def redis_queue_consumer(pid, work_queue_name, exec_mode, backend):
 def run_worker():
     global redis_client
     global budget_keeper
-    global worker_threads
 
     os.makedirs(LITHOPS_TEMP_DIR, exist_ok=True)
 

lithops/storage/storage.py

@@ -462,9 +462,6 @@ def get_runtime_meta(self, key):
         :param runtime: name of the runtime
         :return: runtime metadata
         """
-
-        global RUNTIME_META_CACHE
-
         path = [RUNTIMES_PREFIX, key + ".meta.json"]
         filename_local_path = os.path.join(CACHE_DIR, *path)