From 8fbbb2a16b55e3710d6d51c8849eab794146d9f8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Nikola=20Forr=C3=B3?= Date: Thu, 27 Aug 2026 13:40:59 +0200 Subject: [PATCH] Fetch issues by queue state rather than in daily batches MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Nikola Forró Assisted-by: Claude Sonnet 5 via Claude Code --- openshift/README.md | 6 +- .../configmap-jira-issue-fetcher-env.yml | 11 ++- ...onfigmap-jira-issue-fetcher-filter-env.yml | 1 + openshift/cronjob-jira-issue-fetcher.yml | 10 +- templates/jira-issue-fetcher.env | 5 +- ymir/jira_issue_fetcher/jira_issue_fetcher.py | 41 +++++++- .../tests/unit/test_jira_issue_fetcher.py | 97 ++++++++++++++++++- 7 files changed, 155 insertions(+), 16 deletions(-) diff --git a/openshift/README.md b/openshift/README.md index c05d1054c..46f21cfb6 100644 --- a/openshift/README.md +++ b/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 diff --git a/openshift/configmap-jira-issue-fetcher-env.yml b/openshift/configmap-jira-issue-fetcher-env.yml index 92a3b596d..3c5cc8259 100644 --- a/openshift/configmap-jira-issue-fetcher-env.yml +++ b/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" diff --git a/openshift/configmap-jira-issue-fetcher-filter-env.yml b/openshift/configmap-jira-issue-fetcher-filter-env.yml index cd53966d6..2f3e6ceb7 100644 --- a/openshift/configmap-jira-issue-fetcher-filter-env.yml +++ b/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: diff --git a/openshift/cronjob-jira-issue-fetcher.yml b/openshift/cronjob-jira-issue-fetcher.yml index d8d8bb1b4..5ad336556 100644 --- a/openshift/cronjob-jira-issue-fetcher.yml +++ b/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: diff --git a/templates/jira-issue-fetcher.env b/templates/jira-issue-fetcher.env index b5f93860c..87b4817df 100644 --- a/templates/jira-issue-fetcher.env +++ b/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 diff --git a/ymir/jira_issue_fetcher/jira_issue_fetcher.py b/ymir/jira_issue_fetcher/jira_issue_fetcher.py index d292f7b01..6d10eaf73 100644 --- a/ymir/jira_issue_fetcher/jira_issue_fetcher.py +++ b/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: diff --git a/ymir/jira_issue_fetcher/tests/unit/test_jira_issue_fetcher.py b/ymir/jira_issue_fetcher/tests/unit/test_jira_issue_fetcher.py index a492b2dda..511fef2aa 100644 --- a/ymir/jira_issue_fetcher/tests/unit/test_jira_issue_fetcher.py +++ b/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()