Fetch issues by queue state rather than in daily batches

openshift/README.md

@@ -123,7 +123,7 @@ Two CronJobs run the fetcher with different JQL queries:
 
 | CronJob | Schedule | QUERY | ConfigMap |
 |---|---|---|---|
-| `jira-issue-fetcher` | `0 8 * * *` (daily, 8am UTC) | Main CVE batch — processes up to `MAX_ISSUES` issues from the filter in `jira-issue-fetcher-filter-env` | `jira-issue-fetcher-filter-env` |
+| `jira-issue-fetcher` | `*/15 * * * *` | Main CVE batch — tops the triage queue up towards `QUEUE_DEPTH_THRESHOLD` from the filter in `jira-issue-fetcher-filter-env` | `jira-issue-fetcher-filter-env` |
 | `jira-issue-fetcher-todo` | `*/5 * * * *` | `labels = "ymir_todo"` OR consolidation labels (`ymir_consolidate_base`, `ymir_consolidate_next`) | `jira-issue-fetcher-todo-env` |
 
 Both share the common knobs (`MAX_ISSUES`, `LOGLEVEL`, `SKIP_MODULAR`) from `jira-issue-fetcher-env`. `SKIP_MODULAR` controls whether modular issues (Downstream Component matching `module:stream/pkg`) are enqueued for triage (`false`) or silently dropped (`true`; code default). Components excluded from scope are part of the Jira filter itself, maintained (with rationale per component) in the separate [`cve-scope`](https://gitlab.cee.redhat.com/jotnar-project/cve-scope) repo — not in a configmap or env var.
@@ -132,7 +132,7 @@ The `jira-issue-fetcher-todo` runs every 5 minutes and processes:
 - **User-triggered issues**: Any issue tagged with `ymir_todo` by a maintainer (not filtered by component, processes regardless of scope exclusions)
 - **Consolidation requests**: Issue pairs tagged with `ymir_consolidate_base` and `ymir_consolidate_next` to trigger MR consolidation (see `configmap-jira-issue-fetcher-todo-env.yml` for the exact JQL)
 
-Both fetchers also process consolidation labels (the daily fetcher provides a slower fallback path). Each pod mounts the shared configmap plus its per-cron QUERY configmap. To target a different batch, edit the corresponding configmap and re-apply.
+Both fetchers also process consolidation labels (`jira-issue-fetcher` provides a fallback path, always run regardless of triage-queue depth). Each pod mounts the shared configmap plus its per-cron QUERY configmap. To target a different batch, edit the corresponding configmap and re-apply.
 
 Both CronJobs ship with `suspend: false` and run on their schedules out of the box. Pause or resume either one:
 
@@ -152,7 +152,7 @@ make run-jira-issue-fetcher       # trigger a one-off run now (works even when s
 make run-jira-issue-fetcher-todo  # ymir_todo sweep
 ```
 
-If the daily scheduled run is already active, check before triggering manually to avoid pushing duplicate issues to the queue:
+If a scheduled run is already active, check before triggering manually to avoid pushing duplicate issues to the queue:
 
 ```bash
 oc get jobs -l app=jira-issue-fetcher

openshift/configmap-jira-issue-fetcher-env.yml

@@ -1,11 +1,16 @@
 apiVersion: v1
 data:
-  MAX_ISSUES: "100"
+  # Only caps jira-issue-fetcher-todo's per-cycle push now — it ignores
+  # QUEUE_DEPTH_THRESHOLD (set in jira-issue-fetcher-filter-env), so this is
+  # its only throttle. jira-issue-fetcher instead derives its per-cycle cap
+  # from QUEUE_DEPTH_THRESHOLD minus current queue depth.
+  MAX_ISSUES: "20"
   LOGLEVEL: "INFO"
   # Conservative default pending real task duration data from Phoenix traces;
   # see the comment next to stale_label_threshold_hours in
-  # jira_issue_fetcher.py. Shared by both cronjobs (daily + todo) since it's a
-  # property of agent task duration, not of which query triggered the sweep.
+  # jira_issue_fetcher.py. Shared by both jira-issue-fetcher cronjobs since
+  # it's a property of agent task duration, not of which query triggered the
+  # sweep.
   STALE_LABEL_THRESHOLD_HOURS: "24"
   # Controls the fetcher-level filter for modular issues (Downstream Component
   # matching module:stream/pkg, e.g. postgresql:16/postgresql). When "true"

openshift/configmap-jira-issue-fetcher-filter-env.yml

@@ -1,6 +1,7 @@
 apiVersion: v1
 data:
   QUERY: 'filter="Ymir CVE work to do"'
+  QUEUE_DEPTH_THRESHOLD: "15"
 immutable: false
 kind: ConfigMap
 metadata:

openshift/cronjob-jira-issue-fetcher.yml

@@ -6,7 +6,8 @@ metadata:
     app: jira-issue-fetcher
     component: scheduler
 spec:
-  schedule: "0 8 * * *"  # Daily at 8am UTC — automated CVE batch
+  # Every 15 minutes — tops up the triage queue toward QUEUE_DEPTH_THRESHOLD.
+  schedule: "*/15 * * * *"
   concurrencyPolicy: Forbid  # Prevent overlapping runs
   successfulJobsHistoryLimit: 3
   failedJobsHistoryLimit: 5
@@ -18,7 +19,12 @@ spec:
         component: job
     spec:
       backoffLimit: 2
-      activeDeadlineSeconds: 600  # 10 minutes max runtime
+      # 10 minutes max runtime — a runaway-job safety net, not overlap
+      # prevention (concurrencyPolicy: Forbid already handles that). Must
+      # comfortably exceed _make_request_with_retries' worst case (4 attempts
+      # x 90s timeout plus backoff, ~6 minutes) so a genuinely slow Jira call
+      # can finish or fail on its own instead of being killed mid-retry.
+      activeDeadlineSeconds: 600
       template:
         metadata:
           labels:

templates/jira-issue-fetcher.env

@@ -21,7 +21,10 @@ REDIS_URL=redis://localhost:6379
 # maintained in https://gitlab.cee.redhat.com/jotnar-project/cve-scope -- not via env var.
 
 # Optional: Maximum number of issues to fetch
-#MAX_ISSUES=100
+#MAX_ISSUES=20
+
+# Optional: Target combined triage_queue + triage_queue_todo depth to top up towards
+#QUEUE_DEPTH_THRESHOLD=15
 
 # Optional: Enable debug logging
 #LOGLEVEL=DEBUG

ymir/jira_issue_fetcher/jira_issue_fetcher.py

@@ -93,6 +93,12 @@ def __init__(self):
         max_issues_str = os.getenv("MAX_ISSUES", "")
         self.max_issues: int | None = int(max_issues_str) if max_issues_str else None
 
+        # Optional: target combined triage-queue depth to top up towards.
+        # Unset/empty disables the depth-based throttle entirely.
+        self.queue_depth_threshold: int | None = (
+            int(v) if (v := os.getenv("QUEUE_DEPTH_THRESHOLD", "")) else None
+        )
+
         # Use constant page size
         self.max_results_per_page = self.MAX_RESULTS_PER_PAGE
 
@@ -568,7 +574,15 @@ async def _get_locked_issue_keys(self, redis_conn: redis.Redis) -> set[str]:
             locked_keys.add(issue_key)
         return locked_keys
 
-    async def push_issues_to_queue(self, issues: list[dict[str, Any]]) -> int:
+    async def _get_queue_depth(self, redis_conn: redis.Redis) -> int:
+        """Return the combined depth of the two queues this fetcher pushes to."""
+        triage_depth = await fix_await(redis_conn.llen(RedisQueues.TRIAGE_QUEUE.value))
+        todo_depth = await fix_await(redis_conn.llen(RedisQueues.TRIAGE_QUEUE_TODO.value))
+        return triage_depth + todo_depth
+
+    async def push_issues_to_queue(
+        self, issues: list[dict[str, Any]], max_issues_override: int | None = None
+    ) -> int:
         """
         Push each issue to the Redis triage_queue, but only if it doesn't already exist
         """
@@ -712,14 +726,15 @@ async def push_issues_to_queue(self, issues: list[dict[str, Any]]) -> int:
                         remove_issues_for_retry.add(issue_key)
                         retry_needed_keys.add(issue_key)
 
+            effective_max_issues = max_issues_override if max_issues_override is not None else self.max_issues
             pushed_count = 0
             skipped_count = 0
             modular_count = 0
 
             for issue in issues:
                 try:
-                    if self.max_issues is not None and pushed_count >= self.max_issues:
-                        logger.info(f"Reached MAX_ISSUES limit ({self.max_issues})")
+                    if effective_max_issues is not None and pushed_count >= effective_max_issues:
+                        logger.info(f"Reached per-cycle push limit ({effective_max_issues})")
                         break
 
                     issue_key = issue["key"]
@@ -1035,8 +1050,24 @@ async def run(self) -> None:
                 logger.info("No issues found matching the query")
                 return
 
-            pushed_count = await self.push_issues_to_queue(issues)
-            logger.info(f"Completed: {pushed_count} issues added to triage_queue")
+            # Depth is read here, immediately before it's consumed, rather
+            # than before search_issues() — the search can take a while
+            # (pagination, rate limiting) during which ymir_todo can push
+            # concurrently, so checking right before push keeps the reading
+            # as fresh as possible. This doesn't gate search/consolidation
+            # below: those must always run regardless of triage-queue depth.
+            push_cap = None
+            if self.queue_depth_threshold is not None:
+                async with redis_client(self.redis_url) as redis_conn:
+                    depth = await self._get_queue_depth(redis_conn)
+                logger.info(f"Triage queue depth: {depth} (threshold: {self.queue_depth_threshold})")
+                push_cap = max(0, self.queue_depth_threshold - depth)
+
+            if push_cap != 0:
+                pushed_count = await self.push_issues_to_queue(issues, max_issues_override=push_cap)
+                logger.info(f"Completed: {pushed_count} issues added to triage_queue")
+            else:
+                logger.info("Queue depth at/above threshold - skipping triage push this cycle")
 
             consolidation_count = await self._process_consolidation_labels(issues)
             if consolidation_count:

ymir/jira_issue_fetcher/tests/unit/test_jira_issue_fetcher.py

@@ -87,6 +87,15 @@ def test_init(mock_env_vars):
     assert fetcher.max_results_per_page == 500
     assert fetcher.headers["Authorization"].startswith("Basic ")
     assert fetcher.skip_modular is True
+    assert fetcher.queue_depth_threshold is None
+
+
+def test_init_queue_depth_threshold(mock_env_vars, monkeypatch):
+    """QUEUE_DEPTH_THRESHOLD is parsed to an int when set."""
+    monkeypatch.setenv("QUEUE_DEPTH_THRESHOLD", "15")
+    fetcher = JiraIssueFetcher()
+
+    assert fetcher.queue_depth_threshold == 15
 
 
 @pytest.mark.asyncio
@@ -482,6 +491,50 @@ async def test_push_issues_to_queue_skip_ignored_components(fetcher, mock_redis_
     assert result == 2
 
 
+@pytest.mark.asyncio
+async def test_get_queue_depth_sums_both_queues(fetcher, mock_redis_context):
+    """_get_queue_depth returns the combined depth of triage_queue and triage_queue_todo."""
+    mock_redis, _ = mock_redis_context
+
+    mock_redis.should_receive("llen").with_args(RedisQueues.TRIAGE_QUEUE.value).and_return(
+        create_async_mock_return_value(7)
+    )
+    mock_redis.should_receive("llen").with_args(RedisQueues.TRIAGE_QUEUE_TODO.value).and_return(
+        create_async_mock_return_value(3)
+    )
+
+    depth = await fetcher._get_queue_depth(mock_redis)
+
+    assert depth == 10
+
+
+@pytest.mark.asyncio
+async def test_push_issues_to_queue_max_issues_override_takes_precedence(fetcher, mock_redis_context):
+    """max_issues_override, when passed, is used instead of self.max_issues."""
+    mock_redis, _ = mock_redis_context
+
+    fetcher.max_issues = 100
+
+    issues = [
+        {"key": "RHEL-1", "fields": {"labels": []}},
+        {"key": "RHEL-2", "fields": {"labels": []}},
+        {"key": "RHEL-3", "fields": {"labels": []}},
+    ]
+
+    flexmock(fetcher).should_receive("_get_existing_issue_keys").and_return(
+        create_async_mock_return_value(set())
+    )
+
+    task1 = Task.from_issue("RHEL-1")
+    mock_redis.should_receive("lpush").with_args(RedisQueues.TRIAGE_QUEUE.value, task1.to_json()).and_return(
+        create_async_mock_return_value(1)
+    ).once()
+
+    result = await fetcher.push_issues_to_queue(issues, max_issues_override=1)
+
+    assert result == 1
+
+
 @pytest.mark.asyncio
 async def test_push_issues_to_queue_max_issues(fetcher, mock_redis_context):
     """Test that MAX_ISSUES limits the number of enqueued issues."""
@@ -679,9 +732,49 @@ async def test_run_full_workflow(fetcher):
     flexmock(fetcher).should_receive("search_issues").and_return(
         create_async_mock_return_value(mock_issues)
     ).once()
-    flexmock(fetcher).should_receive("push_issues_to_queue").with_args(mock_issues).and_return(
-        create_async_mock_return_value(1)
+    flexmock(fetcher).should_receive("push_issues_to_queue").with_args(
+        mock_issues, max_issues_override=None
+    ).and_return(create_async_mock_return_value(1)).once()
+    flexmock(fetcher).should_receive("_process_consolidation_labels").with_args(mock_issues).and_return(
+        create_async_mock_return_value(0)
     ).once()
+
+    await fetcher.run()
+
+
+@pytest.mark.asyncio
+async def test_run_skips_push_but_still_processes_consolidation_when_at_threshold(
+    fetcher, mock_redis_context
+):
+    """When depth >= threshold, run() still searches and processes consolidation, just skips the push."""
+    fetcher.queue_depth_threshold = 15
+    mock_issues = [{"key": "RHEL-1", "fields": {"labels": []}}]
+
+    flexmock(fetcher).should_receive("search_issues").and_return(
+        create_async_mock_return_value(mock_issues)
+    ).once()
+    flexmock(fetcher).should_receive("_get_queue_depth").and_return(create_async_mock_return_value(15)).once()
+    flexmock(fetcher).should_receive("push_issues_to_queue").never()
+    flexmock(fetcher).should_receive("_process_consolidation_labels").with_args(mock_issues).and_return(
+        create_async_mock_return_value(0)
+    ).once()
+
+    await fetcher.run()
+
+
+@pytest.mark.asyncio
+async def test_run_tops_up_when_below_threshold(fetcher, mock_redis_context):
+    """When depth < threshold, run() pushes up to (threshold - depth) issues."""
+    fetcher.queue_depth_threshold = 15
+    mock_issues = [{"key": "RHEL-1", "fields": {"labels": []}}]
+
+    flexmock(fetcher).should_receive("search_issues").and_return(
+        create_async_mock_return_value(mock_issues)
+    ).once()
+    flexmock(fetcher).should_receive("_get_queue_depth").and_return(create_async_mock_return_value(10)).once()
+    flexmock(fetcher).should_receive("push_issues_to_queue").with_args(
+        mock_issues, max_issues_override=5
+    ).and_return(create_async_mock_return_value(1)).once()
     flexmock(fetcher).should_receive("_process_consolidation_labels").with_args(mock_issues).and_return(
         create_async_mock_return_value(0)
     ).once()