hotfix: catch bucket job errors and increase backoff_factor for docdb (#133)

* hotfix: catch bucket job errors and increase backoff_factor for docdb

* hotfix: lower page size for docdb

* ci: version bump

Author: helen-m-lin

src/aind_data_asset_indexer/__init__.py

@@ -1,3 +1,3 @@
 """Package"""
 
-__version__ = "0.16.0"
+__version__ = "0.16.1"

src/aind_data_asset_indexer/aind_bucket_indexer.py

@@ -81,7 +81,7 @@ def _create_docdb_client(self) -> MetadataDbClient:
         """Create a MetadataDbClient with custom retries."""
         retry = Retry(
             total=3,
-            backoff_factor=1,
+            backoff_factor=10,
             status_forcelist=[429, 500, 502, 503, 504],
             allowed_methods=["GET", "POST", "DELETE"],
         )
@@ -761,7 +761,7 @@ def run_job(self):
             logging.info(f"Starting to scan through DocDb: {filter}")
             docdb_pages = paginate_docdb(
                 docdb_api_client=iterator_docdb_client,
-                page_size=500,
+                page_size=200,
                 filter_query=filter,
             )
             for page in docdb_pages:

src/aind_data_asset_indexer/codeocean_bucket_indexer.py

@@ -55,7 +55,7 @@ def _create_docdb_client(self) -> MetadataDbClient:
         """Create a MetadataDbClient with custom retries."""
         retry = Retry(
             total=3,
-            backoff_factor=1,
+            backoff_factor=10,
             status_forcelist=[429, 500, 502, 503, 504],
             allowed_methods=["GET", "POST", "DELETE"],
         )

src/aind_data_asset_indexer/index_aind_buckets.py

@@ -32,7 +32,10 @@ def run_job(self):
                 s3_bucket=bucket, **base_job_configs
             )
             bucket_job = AindIndexBucketJob(job_settings=bucket_job_settings)
-            bucket_job.run_job()
+            try:
+                bucket_job.run_job()
+            except Exception as e:
+                logging.error(f"Error processing {bucket}. Error: {e}")
             logging.info(f"Finished processing {bucket}")
 
 

tests/test_index_aind_buckets.py

@@ -36,6 +36,34 @@ def test_run_job(self, mock_sub_run_job: MagicMock):
 
         mock_sub_run_job.assert_has_calls([call(), call()])
 
+    @patch(
+        "aind_data_asset_indexer.aind_bucket_indexer.AindIndexBucketJob."
+        "run_job"
+    )
+    def test_run_job_error(self, mock_sub_run_job: MagicMock):
+        """Tests run_job method when there is an error in 1 bucket job."""
+
+        mock_sub_run_job.side_effect = [Exception("Some error"), None]
+        job_settings = AindIndexBucketsJobSettings(
+            s3_buckets=["bucket1", "bucket2"],
+            doc_db_host="some_docdb_host",
+            doc_db_db_name="some_docdb_dbname",
+            doc_db_collection_name="some_docdb_collection_name",
+        )
+        job = IndexAindBucketsJob(job_settings=job_settings)
+        with self.assertLogs(level="DEBUG") as captured:
+            job.run_job()
+        expected_log_messages = [
+            "INFO:root:Processing bucket1",
+            "ERROR:root:Error processing bucket1. Error: Some error",
+            "INFO:root:Finished processing bucket1",
+            "INFO:root:Processing bucket2",
+            "INFO:root:Finished processing bucket2",
+        ]
+        self.assertEqual(expected_log_messages, captured.output)
+
+        mock_sub_run_job.assert_has_calls([call(), call()])
+
 
 if __name__ == "__main__":
     unittest.main()