Add semaphore to limit concurrent API calls during Kubernetes observer startup

[tomerqodo]

Benchmark PR from agentic-review-benchmarks#11

[qodo-code-review]

Code Review by Qodo

🐞 Bugs (1) 📘 Rule violations (0) 📎 Requirement gaps (0)

1. Semaphore misses API call 🐞 Bug ⛯ Reliability

Description

• The new startup semaphore is intended to limit concurrent /events/filter calls, but it is only
  held while building the EventFilter, then released before the actual
  orchestration_client.request(...) runs.
• This defeats the throttling goal and can still overwhelm the Prefect API during startup scans with
  many existing pods.
• The current behavior also adds avoidable scheduling overhead (queueing for a semaphore that
  doesn’t protect the expensive operation).

Code

src/integrations/prefect-kubernetes/prefect_kubernetes/observer.py[R135-159]

+        # Use semaphore to limit concurrent API calls during startup to prevent
+        # overwhelming the API server when there are many existing pods/jobs
+        async with _startup_event_semaphore:
+            # Use the Kubernetes event timestamp for the filter to avoid "Query time range is too large" error
+            event_filter = EventFilter(
+                event=EventNameFilter(name=[f"prefect.kubernetes.pod.{phase.lower()}"]),
+                resource=EventResourceFilter(
+                    id=[f"prefect.kubernetes.pod.{uid}"],
+                ),
+                occurred=EventOccurredFilter(
+                    since=(
+                        k8s_created_time
+                        if k8s_created_time
+                        else (datetime.now(timezone.utc) - timedelta(hours=1))
+                    )
+                ),
+            )

        response = await orchestration_client.request(
            "POST",
            "/events/filter",
-            json=dict(filter=event_filter.model_dump(exclude_unset=True, mode="json")),
+            json=dict(
+                filter=event_filter.model_dump(exclude_unset=True, mode="json")
+            ),
        )

Evidence

The code comment states the semaphore is meant to limit concurrent startup API calls, but the
request() that performs the API call is outside the async with _startup_event_semaphore: block.
Therefore, the semaphore cannot limit concurrent API calls as intended.

src/integrations/prefect-kubernetes/prefect_kubernetes/observer.py[129-159]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

### Issue description
The startup semaphore is intended to limit concurrent `/events/filter` calls, but it currently wraps only `EventFilter` construction. The actual `orchestration_client.request(...)` happens after the semaphore is released, so concurrency is not throttled.

### Issue Context
This branch runs when `event_type is None` (startup/backfill-style events), which is precisely when many concurrent dedupe checks can occur.

### Fix Focus Areas
- src/integrations/prefect-kubernetes/prefect_kubernetes/observer.py[129-162]

### Suggested change
Move the `orchestration_client.request(...)` and the `if response.json()["events"]: return` logic inside the `async with _startup_event_semaphore:` block so the semaphore actually limits concurrent requests.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

ⓘ The new review experience is currently in Beta. Learn more

[qodo-code-review]

1. Semaphore misses api call 🐞 Bug ⛯ Reliability

• The new startup semaphore is intended to limit concurrent /events/filter calls, but it is only
  held while building the EventFilter, then released before the actual
  orchestration_client.request(...) runs.
• This defeats the throttling goal and can still overwhelm the Prefect API during startup scans with
  many existing pods.
• The current behavior also adds avoidable scheduling overhead (queueing for a semaphore that
  doesn’t protect the expensive operation).

Agent prompt

### Issue description
The startup semaphore is intended to limit concurrent `/events/filter` calls, but it currently wraps only `EventFilter` construction. The actual `orchestration_client.request(...)` happens after the semaphore is released, so concurrency is not throttled.

### Issue Context
This branch runs when `event_type is None` (startup/backfill-style events), which is precisely when many concurrent dedupe checks can occur.

### Fix Focus Areas
- src/integrations/prefect-kubernetes/prefect_kubernetes/observer.py[129-162]

### Suggested change
Move the `orchestration_client.request(...)` and the `if response.json()["events"]: return` logic inside the `async with _startup_event_semaphore:` block so the semaphore actually limits concurrent requests.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools