fix(aws) Add batching to avoid threads hanging on s3 and ecr GetDetails (#2423)

### Type of change
- [x] Bug fix (non-breaking change that fixes an issue)

### Summary
fix(aws) Add batching to avoid threads hanging when getting s3 and ecr
details


### Related issues or links
With the increase in the scale of our infrastructure, we are seeing
threads hanging in `s3` and `ecr` syncs.
The root cause for s3 is that there is no max limit to the connection
pool to AWS API and all bucket details are being requested at once.

For ecr if a repository has a very large volume of images (think > 500k)
querying all details could also deplete and hangout the running threads.

### Breaking changes
No breaking change, but might slowdown aws syncs.


### How was this tested?
- All unit and integration tests pass.
- Running a local sync for each module.


Signed-off-by: Eryx Paredes <eryxp@lyft.com>

Author: heryxpc

cartography/intel/aws/ecr.py

@@ -27,6 +27,9 @@
 }
 
 
+REPO_BATCH_SIZE = 100
+
+
 @timeit
 @aws_handle_regions
 def get_ecr_repositories(
@@ -378,7 +381,9 @@ async def async_get_images(repo: Dict[str, Any]) -> None:
 
     # Sort repositories by name to ensure consistent processing order
     sorted_repos = sorted(repositories, key=lambda x: x["repositoryName"])
-    to_synchronous(*[async_get_images(repo) for repo in sorted_repos])
+    for i in range(0, len(sorted_repos), REPO_BATCH_SIZE):
+        batch = sorted_repos[i : i + REPO_BATCH_SIZE]
+        to_synchronous(*[async_get_images(repo) for repo in batch])
 
     return image_data
 

cartography/intel/aws/s3.py

@@ -12,6 +12,7 @@
 
 import boto3
 import botocore
+import botocore.config
 import neo4j
 from botocore.exceptions import ClientError
 from botocore.exceptions import EndpointConnectionError
@@ -41,6 +42,8 @@
 logger = logging.getLogger(__name__)
 stat_handler = get_stats_client(__name__)
 
+BUCKET_BATCH_SIZE = 50
+
 
 # Sentinel value to indicate a fetch operation failed (vs None for "no configuration")
 # When a fetch returns FETCH_FAILED, we skip loading that property group to preserve existing data.
@@ -66,7 +69,10 @@ def __repr__(self):
 
 @timeit
 def get_s3_bucket_list(boto3_session: boto3.session.Session) -> List[Dict]:
-    client = boto3_session.client("s3")
+    client = boto3_session.client(
+        "s3",
+        config=botocore.config.Config(max_pool_connections=50),
+    )
     # NOTE no paginator available for this operation
     buckets = client.list_buckets()
     for bucket in buckets["Buckets"]:
@@ -135,7 +141,11 @@ async def _get_bucket_detail(bucket: Dict[str, Any]) -> BucketDetail:
         # in us-east-1 region
         client = s3_regional_clients.get(bucket["Region"])
         if not client:
-            client = boto3_session.client("s3", bucket["Region"])
+            client = boto3_session.client(
+                "s3",
+                bucket["Region"],
+                config=botocore.config.Config(max_pool_connections=50),
+            )
             s3_regional_clients[bucket["Region"]] = client
         (
             acl,
@@ -165,10 +175,13 @@ async def _get_bucket_detail(bucket: Dict[str, Any]) -> BucketDetail:
             bucket_logging,
         )
 
-    bucket_details = to_synchronous(
-        *[_get_bucket_detail(bucket) for bucket in bucket_data["Buckets"]],
-    )
-    yield from bucket_details
+    buckets = bucket_data["Buckets"]
+    for i in range(0, len(buckets), BUCKET_BATCH_SIZE):
+        batch = buckets[i : i + BUCKET_BATCH_SIZE]
+        bucket_details = to_synchronous(
+            *[_get_bucket_detail(bucket) for bucket in batch],
+        )
+        yield from bucket_details
 
 
 @timeit
@@ -1249,7 +1262,10 @@ def _sync_s3_notifications(
     Sync S3 bucket notification configurations to Neo4j.
     """
     logger.info("Syncing S3 bucket notifications")
-    s3_client = boto3_session.client("s3")
+    s3_client = boto3_session.client(
+        "s3",
+        config=botocore.config.Config(max_pool_connections=BUCKET_BATCH_SIZE),
+    )
     notifications = []
 
     for bucket in bucket_data["Buckets"]:

cartography/util.py

@@ -1038,7 +1038,11 @@ def wrapper(*args: Any, **kwargs: Any) -> R:
             raise
 
     # don't use @backoff as decorator, to preserve typing
-    wrapped = backoff.on_exception(backoff.expo, CartographyThrottlingException)(
+    wrapped = backoff.on_exception(
+        backoff.expo,
+        CartographyThrottlingException,
+        max_time=300,
+    )(
         wrapper,
     )
     call = partial(wrapped, *args, **kwargs)