0.2.1

* env tweak for task expiry name

Author: nicholac

.github/workflows/build_dev.yaml

@@ -8,7 +8,7 @@ on:
       - develop
 
 env:
-  VERSION: 0.2.0
+  VERSION: 0.2.1
   REGISTRY: ghcr.io
   IMAGE_NAME: ${{ github.repository }}
 

.github/workflows/build_prod_and_release.yaml

@@ -8,7 +8,7 @@ on:
       - master
 
 env:
-  VERSION: 0.2.0
+  VERSION: 0.2.1
   REGISTRY: ghcr.io
   IMAGE_NAME: ${{ github.repository }}
 

README.md

@@ -159,6 +159,9 @@ export AUTOPKG_POSTGRES_PORT= # Used for API Boundaries only (and test natural_e
 export AUTOPKG_POSTGRES_DB= # Used for API Boundaries only (and test natural_earth_vector processor in Worker)
 export AUTOPKG_CELERY_BROKER= # Used for Worker only
 export AUTOPKG_CELERY_BACKEND= # Used in API and Worker
+export AUTOPKG_CELERY_CONCURRENCY= # Celery worker concurrency - dataproc only
+export AUTOPKG_TASK_LOCK_TIMEOUT=600 # Secs - Duplicate task lock timeout (blocks duplicate processors from executing for this time)
+export AUTOPKG_TASK_EXPIRY_SECS=3600 # Secs before tasks expire on Celery - dataproc only
 export AUTOPKG_STORAGE_BACKEND= # Storage backend to use for final packages (see additional backend-specific flags below for more info).  Used in API and Worker
 export AUTOPKG_LOCALFS_STORAGE_BACKEND_ROOT= # Path to root-directory for packages.  Used in API and Worker
 export AUTOPKG_LOCALFS_PROCESSING_BACKEND_ROOT= # Path to root-directory for local interim processing data.  Used by Worker only

config.py

@@ -8,6 +8,7 @@
 import sqlalchemy as sa
 from celery import Celery
 
+
 def get_db_uri(
     dbname: str,
     username_env="AUTOPKG_POSTGRES_USER",
@@ -25,6 +26,7 @@ def get_db_uri(
         database=dbname,
     )
 
+
 def get_db_uri_ogr(
     dbname: str,
     username_env="AUTOPKG_POSTGRES_USER",
@@ -52,7 +54,7 @@ def get_db_uri_sync(
     password_env="AUTOPKG_POSTGRES_PASSWORD",
     host_env="AUTOPKG_POSTGRES_HOST",
     port_env="AUTOPKG_POSTGRES_PORT",
-    ) -> sa.engine.URL:
+) -> sa.engine.URL:
     """Standard user DBURI - non-async"""
     return sa.engine.URL.create(
         drivername="postgresql+psycopg2",
@@ -63,6 +65,7 @@ def get_db_uri_sync(
         database=dbname,
     )
 
+
 # DATAPROC VARS
 REDIS_HOST = getenv("AUTOPKG_REDIS_HOST", "localhost")
 TASK_LOCK_TIMEOUT = int(
@@ -79,6 +82,7 @@ def get_db_uri_sync(
 # Celery Env
 CELERY_BROKER = getenv("AUTOPKG_CELERY_BROKER", "redis://localhost")
 CELERY_BACKEND = getenv("AUTOPKG_CELERY_BACKEND", "redis://localhost")
+CELERY_CONCURRENCY = int(getenv("AUTOPKG_CELERY_CONCURRENCY", "2"))
 REDIS_HOST = getenv("AUTOPKG_REDIS_HOST", "localhost")
 TASK_LOCK_TIMEOUT = getenv("AUTOPKG_TASK_LOCK_TIMEOUT", "600")
 
@@ -98,9 +102,15 @@ def get_db_uri_sync(
 if getenv("AUTOPKG_DEPLOYMENT_ENV", "prod") == "test":
     # TEST
     # The root-level folder when using localfs storage backend
-    LOCALFS_STORAGE_BACKEND_ROOT = getenv("AUTOPKG_LOCALFS_STORAGE_BACKEND_ROOT_TEST", path.join(path.dirname(path.abspath(__file__)), "tests", "data", "packages"))
+    LOCALFS_STORAGE_BACKEND_ROOT = getenv(
+        "AUTOPKG_LOCALFS_STORAGE_BACKEND_ROOT_TEST",
+        path.join(path.dirname(path.abspath(__file__)), "tests", "data", "packages"),
+    )
     # The root-level folder when using localfs processing backend
-    LOCALFS_PROCESSING_BACKEND_ROOT = getenv("AUTOPKG_LOCALFS_PROCESSING_BACKEND_ROOT_TEST", path.join(path.dirname(path.abspath(__file__)), "tests", "data", "processing"))
+    LOCALFS_PROCESSING_BACKEND_ROOT = getenv(
+        "AUTOPKG_LOCALFS_PROCESSING_BACKEND_ROOT_TEST",
+        path.join(path.dirname(path.abspath(__file__)), "tests", "data", "processing"),
+    )
     # Integration tests which require access to the GRIOSM Postgres instance will be run if this is set-True (1)
     TEST_GRI_OSM = bool(int(getenv("TEST_GRI_OSM", "0")))
 else:
@@ -117,10 +127,11 @@ def get_db_uri_sync(
 DBURI_API = get_db_uri(API_POSTGRES_DB)
 CELERY_APP = Celery(
     "AutoPackage",
-    worker_prefetch_multiplier=1,
+    worker_prefetch_multiplier=1,  # Do not change - long running tasks require this. See: https://docs.celeryq.dev/en/stable/userguide/configuration.html#std-setting-worker_prefetch_multiplier
+    worker_concurrency=CELERY_CONCURRENCY,
     broker_url=CELERY_BROKER,
-    result_backend=CELERY_BACKEND
-)  # see: https://docs.celeryq.dev/en/stable/userguide/configuration.html#std-setting-worker_prefetch_multiplier
+    result_backend=CELERY_BACKEND,
+)
 
 # Seconds before submitted tasks expire
-TASK_EXPIRY_SECS = int(getenv("TASK_EXPIRY_SECS", "3600"))
+TASK_EXPIRY_SECS = int(getenv("AUTOPKG_TASK_EXPIRY_SECS", "3600"))

docker-compose.yaml

@@ -51,7 +51,7 @@ services:
       - ./tests/data/tmp:/usr/src/app/tests/data/tmp
     env_file:
       - envs/.api_and_dataproc.env
-    command: celery --app dataproc.tasks worker --loglevel=debug --concurrency=4
+    command: celery --app dataproc.tasks worker
 
   api:
     image: ghcr.io/nismod/irv-autopkg:0.2.0-dev