Skip to content

perf(manager): preaggregate cached CVE subquery #2438

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
pindo696 wants to merge 1 commit into
base: master
Choose a base branch
from
+84 −48
Open

perf(manager): preaggregate cached CVE subquery #2438

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
132 changes: 84 additions & 48 deletions manager/vulnerabilities_handler.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -113,15 +113,19 @@ def __init__(self, account_data, list_args, uri, args):
UNLEASH.is_enabled(CACHES_FEATURE), is_cacheable_request(args), valid_cache)
if self.is_cached:
count_subquery = self._granular_cached_count_subquery(account_data.id, account_data.cves_without_errata, args)
query = self._full_query(account_data.id, join_type, count_subquery, pre_aggregated=True)
# OperatingSystems is joined inside the subquery, so we apply the filter inside
# '_granular_cached_count_subquery', see the end of the function
cached_filters = [f for f in full_query_filters if f != filter_types.CVE_RHEL_VERSION]
query = apply_filters(query, args, cached_filters, {"count_subquery": count_subquery})
else:
if account_data.cves_without_errata:
count_subquery = self._unpatched_count_subquery(account_data.id, args, subquery_filters)
else:
count_subquery = self._count_subquery(account_data.id, args, subquery_filters,
remediation_filter=DEFAULT_REMEDIATION_FILTER)

query = self._full_query(account_data.id, join_type, count_subquery)
query = apply_filters(query, args, full_query_filters, {"count_subquery": count_subquery})
query = self._full_query(account_data.id, join_type, count_subquery)
query = apply_filters(query, args, full_query_filters, {"count_subquery": count_subquery})
if not account_data.cves_without_errata:
query = query.where(CveMetadata.advisories_list != SQL("'[]'"))
query = query.dicts()
Expand Down Expand Up @@ -176,42 +180,60 @@ def __init__(self, account_data, list_args, uri, args):
filterable_columns, filter_expressions, list_args, args, uri)

@staticmethod
def _full_query(rh_account_id, join_type, count_subquery):
return (CveMetadata
.select(CveMetadata.id.alias("cve_id"),
CveMetadata.cve.alias("cve_name"),
CveMetadata.cvss3_score,
CveMetadata.cvss2_score,
CveMetadata.impact_id,
CveMetadata.public_date,
CveMetadata.description.alias("cve_description"),
CveMetadata.exploit_data,
CveMetadata.advisories_list,
fn.COALESCE(CveAccountData.business_risk_id, 0).alias("business_risk_id"),
CveAccountData.business_risk_text.alias("business_risk_text"),
fn.COALESCE(BusinessRisk.name, DEFAULT_BUSINESS_RISK).alias("business_risk"),
fn.COALESCE(CveAccountData.status_id, 0).alias("status_id"),
CveAccountData.status_text.alias("status_text"),
fn.Sum((fn.COALESCE(count_subquery.c.systems_affected_rpmdnf_, 0) +
fn.COALESCE(count_subquery.c.systems_affected_edge_, 0))).alias("systems_affected"),
fn.Sum(fn.COALESCE(count_subquery.c.systems_status_divergent_, 0)).alias("systems_status_divergent"),
fn.Bool_Or(fn.COALESCE(count_subquery.c.advisory_available_, fn.COALESCE(CveMetadata.advisories_list, '[]') != SQL("'[]'")))
.alias("advisory_available"),
fn.COALESCE(
fn.ARRAY_AGG(
fn.DISTINCT(RHEL_VERSION_COLUMN)
).order_by(RHEL_VERSION_COLUMN).filter(OperatingSystem.id.is_null(False)), "{}").alias("rhel_versions"))
.join(count_subquery, join_type, on=(CveMetadata.id == count_subquery.c.cve_id_))
.join(OperatingSystem, JOIN.LEFT_OUTER, on=(count_subquery.c.operating_system_id_ == OperatingSystem.id))
.join(CveAccountData, JOIN.LEFT_OUTER, on=((CveMetadata.id == CveAccountData.cve_id)
& (CveAccountData.rh_account_id == rh_account_id)))
.join(BusinessRisk, JOIN.LEFT_OUTER, on=(CveAccountData.business_risk_id == BusinessRisk.id))
.group_by(CveMetadata.id,
fn.COALESCE(CveAccountData.business_risk_id, 0).alias("business_risk_id"),
CveAccountData.business_risk_text.alias("business_risk_text"),
fn.COALESCE(BusinessRisk.name, DEFAULT_BUSINESS_RISK).alias("business_risk"),
fn.COALESCE(CveAccountData.status_id, 0).alias("status_id"),
CveAccountData.status_text.alias("status_text")))
def _full_query(rh_account_id, join_type, count_subquery, pre_aggregated=False):

@sourcery-ai sourcery-ai Bot Aug 4, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

suggestion (bug_risk): Align advisory-availability semantics between pre-aggregated and non-pre-aggregated modes.

In the non‑pre‑aggregated path, advisory_available uses Bool_Or(...) over joined rows, but in the pre‑aggregated path you COALESCE(count_subquery.c.advisory_available_, ...). Since _granular_cached_count_subquery now does Bool_Or(advisory_available_column), this should usually match, but the behavior can diverge if the subquery returns NULL or no rows (e.g., after WHERE filtering). Consider making both paths structurally consistent and explicitly aligning how "no rows" vs "all false" are handled so cached and uncached results stay in sync.

Suggested implementation:

        if pre_aggregated:
            systems_affected = (fn.COALESCE(count_subquery.c.systems_affected_rpmdnf_, 0) +
                                fn.COALESCE(count_subquery.c.systems_affected_edge_, 0)).alias("systems_affected")
            systems_status_divergent = fn.COALESCE(count_subquery.c.systems_status_divergent_, 0).alias("systems_status_divergent")
            advisory_available = fn.Bool_Or(
                fn.COALESCE(
                    count_subquery.c.advisory_available_,
                    fn.COALESCE(CveMetadata.advisories_list, '[]') != SQL("'[]'")
                )
            ).alias("advisory_available")
            rhel_versions = fn.COALESCE(count_subquery.c.rhel_versions_, "{}").alias("rhel_versions")
        else:
            systems_affected = fn.Sum((fn.COALESCE(count_subquery.c.systems_affected_rpmdnf_, 0) +
                                       fn.COALESCE(count_subquery.c.systems_affected_edge_, 0))).alias("systems_affected")
            systems_status_divergent = fn.Sum(fn.COALESCE(count_subquery.c.systems_status_divergent_, 0)).alias("systems_status_divergent")
            advisory_available = fn.Bool_Or(
                fn.COALESCE(
                    count_subquery.c.advisory_available_,
                    fn.COALESCE(CveMetadata.advisories_list, '[]') != SQL("'[]'")
                )
            ).alias("advisory_available")
  1. This change assumes _granular_cached_count_subquery is already aggregating advisory_available_ with Bool_Or so that using Bool_Or again in the pre-aggregated branch is idempotent (single-row aggregation). If that assumption is incorrect and advisory_available_ is no longer aggregated in the cached subquery, you may want to keep the pre-aggregated path as a plain COALESCE without Bool_Or.
  2. If you need stricter handling of the “no rows” case (e.g., explicitly falling back to CveMetadata.advisories_list when the join produces no count-subquery rows), you’ll need to verify the join type (join_type) and possibly adjust the join or add additional COALESCE around the outer Bool_Or to enforce a default value.

if pre_aggregated:
systems_affected = (fn.COALESCE(count_subquery.c.systems_affected_rpmdnf_, 0) +
fn.COALESCE(count_subquery.c.systems_affected_edge_, 0)).alias("systems_affected")
systems_status_divergent = fn.COALESCE(count_subquery.c.systems_status_divergent_, 0).alias("systems_status_divergent")
advisory_available = fn.COALESCE(count_subquery.c.advisory_available_,
fn.COALESCE(CveMetadata.advisories_list, '[]') != SQL("'[]'")).alias("advisory_available")
rhel_versions = fn.COALESCE(count_subquery.c.rhel_versions_, "{}").alias("rhel_versions")
else:
systems_affected = fn.Sum((fn.COALESCE(count_subquery.c.systems_affected_rpmdnf_, 0) +
fn.COALESCE(count_subquery.c.systems_affected_edge_, 0))).alias("systems_affected")
systems_status_divergent = fn.Sum(fn.COALESCE(count_subquery.c.systems_status_divergent_, 0)).alias("systems_status_divergent")
advisory_available = fn.Bool_Or(fn.COALESCE(count_subquery.c.advisory_available_,
fn.COALESCE(CveMetadata.advisories_list, '[]') != SQL("'[]'"))).alias("advisory_available")
rhel_versions = fn.COALESCE(
fn.ARRAY_AGG(
fn.DISTINCT(RHEL_VERSION_COLUMN)
).order_by(RHEL_VERSION_COLUMN).filter(OperatingSystem.id.is_null(False)), "{}").alias("rhel_versions")

query = (CveMetadata
.select(CveMetadata.id.alias("cve_id"),
CveMetadata.cve.alias("cve_name"),
CveMetadata.cvss3_score,
CveMetadata.cvss2_score,
CveMetadata.impact_id,
CveMetadata.public_date,
CveMetadata.description.alias("cve_description"),
CveMetadata.exploit_data,
CveMetadata.advisories_list,
fn.COALESCE(CveAccountData.business_risk_id, 0).alias("business_risk_id"),
CveAccountData.business_risk_text.alias("business_risk_text"),
fn.COALESCE(BusinessRisk.name, DEFAULT_BUSINESS_RISK).alias("business_risk"),
fn.COALESCE(CveAccountData.status_id, 0).alias("status_id"),
CveAccountData.status_text.alias("status_text"),
systems_affected,
systems_status_divergent,
advisory_available,
rhel_versions)
.join(count_subquery, join_type, on=(CveMetadata.id == count_subquery.c.cve_id_))
.join(CveAccountData, JOIN.LEFT_OUTER, on=((CveMetadata.id == CveAccountData.cve_id)
& (CveAccountData.rh_account_id == rh_account_id)))
.join(BusinessRisk, JOIN.LEFT_OUTER, on=(CveAccountData.business_risk_id == BusinessRisk.id)))

if not pre_aggregated:
query = (query
.join(OperatingSystem, JOIN.LEFT_OUTER, on=(count_subquery.c.operating_system_id_ == OperatingSystem.id))
.group_by(CveMetadata.id,
fn.COALESCE(CveAccountData.business_risk_id, 0).alias("business_risk_id"),
CveAccountData.business_risk_text.alias("business_risk_text"),
fn.COALESCE(BusinessRisk.name, DEFAULT_BUSINESS_RISK).alias("business_risk"),
fn.COALESCE(CveAccountData.status_id, 0).alias("status_id"),
CveAccountData.status_text.alias("status_text")))

return query

@staticmethod
def _count_subquery(rh_account_id, args, filters, remediation_filter=None):
Expand Down Expand Up @@ -307,21 +329,35 @@ def _granular_cached_count_subquery(rh_account_id, cves_without_errata, args):
systems_status_divergent = fn.Coalesce(CveAccountGranularCache.systems_status_divergent_unpatched, 0)
advisory_available_column = Value(False)

group_set_subquery = SystemGroupSet.select(SystemGroupSet.id)
group_set_subquery = filter_allowed_groups(group_set_subquery, SystemGroupSet.groups)
group_set_subquery = apply_filters(group_set_subquery, args,
[filter_types.INVENTORY_GROUP_IDS, filter_types.INVENTORY_GROUP_NAMES],
{"column": SystemGroupSet.groups})
# group_set_subquery = group_set_subquery.offset(0)

# edge systems are added in the cache, need to mock the systems_affected_edge_ column
query = (CveAccountGranularCache
.select(CveAccountGranularCache.cve_id.alias("cve_id_"),
CveAccountGranularCache.operating_system_id.alias("operating_system_id_"),
systems_affected_rpmdnf.alias("systems_affected_rpmdnf_"),
systems_affected_edge.alias("systems_affected_edge_"),
systems_status_divergent.alias("systems_status_divergent_"),
advisory_available_column.alias("advisory_available_"))
.join(SystemGroupSet, on=(CveAccountGranularCache.group_set_id == SystemGroupSet.id))
fn.SUM(systems_affected_rpmdnf).alias("systems_affected_rpmdnf_"),
fn.SUM(systems_affected_edge).alias("systems_affected_edge_"),
fn.SUM(systems_status_divergent).alias("systems_status_divergent_"),
fn.Bool_Or(advisory_available_column).alias("advisory_available_"),
fn.COALESCE(
fn.ARRAY_AGG(
fn.DISTINCT(RHEL_VERSION_COLUMN)
).order_by(RHEL_VERSION_COLUMN).filter(OperatingSystem.id.is_null(False)), "{}").alias("rhel_versions_"))
.join(OperatingSystem, JOIN.LEFT_OUTER,
on=(CveAccountGranularCache.operating_system_id == OperatingSystem.id))
.where(CveAccountGranularCache.rh_account_id == rh_account_id)
.where((systems_affected_rpmdnf > 0) | (systems_affected_edge > 0)))
query = filter_allowed_groups(query, SystemGroupSet.groups)
.where((systems_affected_rpmdnf > 0) | (systems_affected_edge > 0))
.where(CveAccountGranularCache.group_set_id.in_(group_set_subquery))
.group_by(CveAccountGranularCache.cve_id))

# cached subquery supports only a subset of subquery filters defined above, so adding it here ad-hoc
# would be best to re-design filters again :(
query = apply_filters(query, args, [filter_types.INVENTORY_GROUP_IDS, filter_types.INVENTORY_GROUP_NAMES], {"column": SystemGroupSet.groups})
query = apply_filters(query, args, [filter_types.CVE_RHEL_VERSION], {})

return query


Expand Down
Loading