feat: add sentry integrations

pyproject.toml

@@ -14,6 +14,7 @@ ipython = "^8.24.0"
 sqlalchemy = "^2.0.30"
 requests = "^2.31.0"
 elasticsearch = "7.14"
+sentry-sdk = "^2.3.1"
 
 
 [build-system]

run.py

@@ -1,3 +1,5 @@
+import sentry_sdk
+
 import logging
 import time
 
@@ -46,6 +48,12 @@ def bulk_process():
 
 
 def main():
+    if settings.USE_SENTRY:
+        sentry_sdk.init(
+            dsn=settings.SENTRY_DSN,
+            enable_tracing=True,
+        )
+
     logging.info("Service running on bulk process single thread mode")
     bulk_process()
 

settings.py

@@ -7,6 +7,7 @@
 ES_URL = os.environ.get("ES_URL", "")
 ES_FLOWRUN_INDEX_NAME = os.environ.get("ES_FLOWRUN_INDEX_NAME", "flowruns")
 FLOW_RUN_BATCH_LIMIT = int(os.environ.get("FLOW_RUN_BATCH_LIMIT", 100))
+EMPTY_ORG_SLEEP = float(os.environ.get("EMPTY_ORG_SLEEP", 0.1))
 BATCH_PROCESSING_TIME_LIMIT = int(os.environ.get("BATCH_PROCESSING_TIME_LIMIT", 30))
 START_RUN_OFFSET = int(os.environ.get("START_RUN_OFFSET", 30))
 FLOW_LAST_INDEXED_FIELD = os.environ.get("FLOW_LAST_INDEXED_FIELD", "modified_on")
@@ -16,3 +17,6 @@
 
 PROCESS_TYPE = os.environ.get("PROCESS_TYPE", "BULK")
 CONSUMER_MAIN_DELAY = int(os.environ.get("CONSUMER_MAIN_DELAY", 1000))
+
+USE_SENTRY = bool(int(os.environ.get("USE_SENTRY", "0")))
+SENTRY_DSN = os.environ.get("SENTRY_DSN", None)

shared/processors.py

@@ -1,3 +1,4 @@
+import time
 import settings
 
 from typing import Callable
@@ -30,6 +31,7 @@ def execute(self):
                 modified_on=last_indexed_at, org_id=org_id
             )
             if len(from_obj_list) == 0:  # if there's no objects on the list
+                time.sleep(settings.EMPTY_ORG_SLEEP)
                 continue
 
             transformed_objects = []
@@ -46,5 +48,6 @@ def execute(self):
 
             # [L]oad the treated object list into the new storage
             is_inserted: bool = self.storage_to.bulk_insert(transformed_objects)
+            time.sleep(settings.EMPTY_ORG_SLEEP)
 
         return is_inserted