New configurable service.idle_jobs_threshold

config.yml.example

@@ -217,6 +217,10 @@
 #service.log_level: INFO
 #
 #
+##  The threshold (in seconds) before a sync job is considered idle.
+#service.idle_jobs_threshold: 300
+#
+#
 ## ------------------------------- Extraction Service ----------------------------------
 #
 ##  Local extraction service-related configurations.

connectors/config.py

@@ -102,6 +102,7 @@ def _default_config():
             "max_file_download_size": DEFAULT_MAX_FILE_SIZE,
             "job_cleanup_interval": 300,
             "log_level": "INFO",
+            "idle_jobs_threshold": 300,
         },
         "sources": {
             "azure_blob_storage": "connectors.sources.azure_blob_storage:AzureBlobStorageDataSource",

connectors/protocol/connectors.py

@@ -65,7 +65,6 @@
     "ServiceTypeNotConfiguredError",
     "ServiceTypeNotSupportedError",
     "Status",
-    "IDLE_JOBS_THRESHOLD",
     "JOB_NOT_FOUND_ERROR",
     "Connector",
     "Features",
@@ -1095,9 +1094,6 @@ def _extra(self):
         }
 
 
-IDLE_JOBS_THRESHOLD = 60 * 5  # 5 minutes
-
-
 class SyncJobIndex(ESIndex):
     """
     Represents Elasticsearch index for sync jobs
@@ -1203,7 +1199,7 @@ async def orphaned_idle_jobs(self, connector_ids):
         async for job in self.get_all_docs(query=query):
             yield job
 
-    async def idle_jobs(self, connector_ids):
+    async def idle_jobs(self, idle_jobs_threshold, connector_ids):
         query = {
             "bool": {
                 "filter": [
@@ -1216,7 +1212,7 @@ async def idle_jobs(self, connector_ids):
                             ]
                         }
                     },
-                    {"range": {"last_seen": {"lte": f"now-{IDLE_JOBS_THRESHOLD}s"}}},
+                    {"range": {"last_seen": {"lte": f"now-{idle_jobs_threshold}s"}}},
                 ]
             }
         }

connectors/services/job_cleanup.py

@@ -21,6 +21,9 @@ class JobCleanUpService(BaseService):
     def __init__(self, config):
         super().__init__(config, "job_cleanup_service")
         self.idling = int(self.service_config.get("job_cleanup_interval", 60 * 5))
+        self.idle_jobs_threshold = int(
+            self.service_config.get("idle_jobs_threshold", 60 * 5)
+        )
         self.native_service_types = self.config.get("native_service_types", []) or []
         self.connector_ids = list(self.connectors.keys())
 
@@ -90,7 +93,9 @@ async def _process_idle_jobs(self):
             ]
 
             marked_count = total_count = 0
-            async for job in self.sync_job_index.idle_jobs(connector_ids=connector_ids):
+            async for job in self.sync_job_index.idle_jobs(
+                self.idle_jobs_threshold, connector_ids=connector_ids
+            ):
                 job_id = job.id
                 try:
                     connector_id = job.connector_id

tests/protocol/test_connectors.py

@@ -20,7 +20,6 @@
     InvalidFilteringError,
 )
 from connectors.protocol import (
-    IDLE_JOBS_THRESHOLD,
     JOB_NOT_FOUND_ERROR,
     Connector,
     ConnectorIndex,
@@ -2246,6 +2245,7 @@ async def test_idle_jobs(get_all_docs, set_env):
     job = Mock()
     get_all_docs.return_value = AsyncIterator([job])
     config = load_config(CONFIG)
+    idle_jobs_threshold = config["service"]["idle_jobs_threshold"]
     connector_ids = [1, 2]
     expected_query = {
         "bool": {
@@ -2259,13 +2259,18 @@ async def test_idle_jobs(get_all_docs, set_env):
                         ]
                     }
                 },
-                {"range": {"last_seen": {"lte": f"now-{IDLE_JOBS_THRESHOLD}s"}}},
+                {"range": {"last_seen": {"lte": f"now-{idle_jobs_threshold}s"}}},
             ]
         }
     }
 
     sync_job_index = SyncJobIndex(elastic_config=config["elasticsearch"])
-    jobs = [job async for job in sync_job_index.idle_jobs(connector_ids=connector_ids)]
+    jobs = [
+        job
+        async for job in sync_job_index.idle_jobs(
+            idle_jobs_threshold, connector_ids=connector_ids
+        )
+    ]
 
     get_all_docs.assert_called_with(query=expected_query)
     assert len(jobs) == 1

tests/services/test_job_cleanup.py

@@ -22,6 +22,7 @@
         "max_errors": 20,
         "max_errors_span": 600,
         "job_cleanup_interval": 1,
+        "idle_jobs_threshold": 300,
     },
     "native_service_types": ["mongodb"],
 }