diff --git a/docs/data_ingestion_and_matching.md b/docs/data_ingestion_and_matching.md index 0ca33095..e4f8d591 100644 --- a/docs/data_ingestion_and_matching.md +++ b/docs/data_ingestion_and_matching.md @@ -14,19 +14,19 @@ This document shows how to fetch channels, run evaluations, ingest CVEs and prod manage fetch_all_channels ``` -When running this command for first time, this would take a couple of minutes. -In the meantime you can watch some cat videos. The output will look like this: ```console -{'channel_branch': 'nixos-25.05', - 'head_sha1_commit': 'ac62194c3917d5f474c1a844b6fd6da2db95077d', - 'release_branch': 'release-25.05', - 'state': NixChannel.ChannelState.END_OF_LIFE} -{'channel_branch': 'nixos-25.05-small', - 'head_sha1_commit': 'ac62194c3917d5f474c1a844b6fd6da2db95077d', - 'release_branch': 'release-25.05', - 'state': NixChannel.ChannelState.END_OF_LIFE} +{'channel': 'nixos-25.11-small', + 'release_branch': 'release-25.11', + 'revision': 'c9bfd86ed684d27e63b0ff9ebb18699f84f27a3b', + 'status': NixChannel.ChannelState.END_OF_LIFE, + 'variant': NixChannel.Variant.SMALL} +{'channel': 'nixpkgs-25.11-darwin', + 'release_branch': 'release-25.11', + 'revision': '0921fdb3e13e40fe25fbc52b89661a9d6d32ac68', + 'status': NixChannel.ChannelState.END_OF_LIFE, + 'variant': NixChannel.Variant.DARWIN} ``` 2. Select a `head_sha1_commit` from the output of `fetch_all_channels` command and run evaluation on that: diff --git a/nix/tests/default.nix b/nix/tests/default.nix index 616f91b2..e3774e54 100644 --- a/nix/tests/default.nix +++ b/nix/tests/default.nix @@ -44,6 +44,8 @@ pkgs.testers.runNixOSTest { git init --initial-branch=master git add -A git -c user.name=test -c user.email=test@test commit -m "test" + git branch release-25.11 + git branch release-25.05 git rev-parse HEAD > REVISION ''; in @@ -138,8 +140,10 @@ pkgs.testers.runNixOSTest { with subtest("Check that channels are fetched and only small ones get enqueued for evaluation"): server.succeed("wst-manage fetch_all_channels") ${in-shell "succeed" '' - from shared.models import NixChannel - assert NixChannel.objects.count() == 6 + from shared.models import NixChannel, NixpkgsBranch + + assert NixpkgsBranch.objects.count() == 3, f"expected 3 branches, got {NixpkgsBranch.objects.count()}" + assert NixChannel.objects.count() == 6, f"expected 6 channels, got {NixChannel.objects.count()}" ''} ${ # Give it some time to queue up the evaluations... diff --git a/src/project/settings.py b/src/project/settings.py index 2cec883a..07b51205 100644 --- a/src/project/settings.py +++ b/src/project/settings.py @@ -108,6 +108,12 @@ class DjangoSettings(BaseModel): "https://monitoring.nixos.org/prometheus/api/v1/query?query=channel_revision" ), ) + NETWORK_REQUEST_TIMEOUT: int = Field( + description=""" + Timeout in seconds for outbound network requests. + """, + default=60, + ) SYNC_GITHUB_STATE_AT_STARTUP: bool = Field( description=""" Connect to GitHub when the service is started and update diff --git a/src/shared/git.py b/src/shared/git.py index 40f3832f..2f129ab1 100644 --- a/src/shared/git.py +++ b/src/shared/git.py @@ -4,6 +4,7 @@ import os.path import pathlib import random +import subprocess import time from collections.abc import AsyncGenerator from contextlib import asynccontextmanager @@ -11,10 +12,25 @@ from typing import IO, Any from django.conf import settings +from pydantic import AnyUrl logger = logging.getLogger(__name__) +def get_head_sha1(url: AnyUrl, branch: str) -> str: + result = subprocess.run( + ["git", "ls-remote", str(url), f"refs/heads/{branch}"], + capture_output=True, + text=True, + check=True, + timeout=settings.NETWORK_REQUEST_TIMEOUT, + ) + line = result.stdout.strip() + if not line: + raise ValueError(f"branch {branch!r} not found at {url!r}") + return line.split()[0] + + @dataclass class Worktree: path: pathlib.Path diff --git a/src/shared/management/commands/fetch_all_channels.py b/src/shared/management/commands/fetch_all_channels.py index 60920f82..948a0198 100644 --- a/src/shared/management/commands/fetch_all_channels.py +++ b/src/shared/management/commands/fetch_all_channels.py @@ -1,40 +1,56 @@ -import asyncio -import sys -from collections.abc import Coroutine -from dataclasses import dataclass +import re +from io import StringIO from pprint import pprint -from typing import Any +from typing import Annotated, Any import requests from django.conf import settings from django.core.management.base import BaseCommand +from django.db import transaction +from pydantic import BaseModel, ConfigDict, Field, model_validator -from shared.git import GitRepo -from shared.models.nix_evaluation import NixChannel, release_branch +from shared.git import get_head_sha1 +from shared.models.nix_evaluation import NixChannel, NixpkgsBranch -@dataclass -class MonitoredChannel: - name: str - revision: str - status: str - variant: str | None +class MonitoredChannel(BaseModel): + model_config = ConfigDict(extra="ignore") + channel: str + release_branch: str + revision: Annotated[str, Field(pattern="[0-9a-f]{40}")] + status: NixChannel.ChannelState + variant: NixChannel.Variant | None = None -def aggregate_by_channels(data: list[dict[str, Any]]) -> dict[str, MonitoredChannel]: - channels = {} - for metric in data: - m = metric["metric"] - channels[m["channel"]] = MonitoredChannel( - name=m["channel"], - revision=m["revision"], - status=m["status"], - variant=m.get("variant"), + @model_validator(mode="before") + @classmethod + def parse_channel_name(cls, data: Any) -> Any: + """ + This is a heuristic for deriving the release branch from a channel's name. + The Technically Correct way to obtain that value would be to + 1. Take the underlying `channels.nix` from `github:NixOS/infra` for jobsets and channel states + 2. Resolve the associated branch names and commits through Hydra + because those are the authoritative sources. + + But that would require hitting Hydra with 3 roundtrips per channel and parsing all the responses. + Therefore we'll only implement that if the flexibility is actually required. + """ + name = data.get("channel", "") + match = re.match( + r"^(?Pnixos|nixpkgs)-(?P\d\d\.\d\d|unstable)(?:-(?Pprimary|small|darwin))?$", + name, ) - return channels + if match is None: + raise ValueError(f"unexpected channel name: {name!r}") + release = match.group("release") + release_branch = "master" if release == "unstable" else f"release-{release}" + return { + **data, + "release_branch": release_branch, + } -def fetch_from_monitoring() -> dict[str, MonitoredChannel]: +def fetch_from_monitoring() -> list[MonitoredChannel]: resp = requests.get( # XXX(@fricklerhandwerk): The sources for this are declared in the `NixOS/infra` repo. [tag:channel-structure] # exporter logic: @@ -46,41 +62,44 @@ def fetch_from_monitoring() -> dict[str, MonitoredChannel]: settings.CHANNEL_MONITORING_URL ) resp.raise_for_status() - return aggregate_by_channels(resp.json()["data"]["result"]) - - -async def wait_for_parallel_fetches( - parallel_fetches: list[Coroutine[Any, Any, bool]], -) -> list[Any]: - return await asyncio.gather(*parallel_fetches, return_exceptions=True) + channels = [] + for metric in resp.json()["data"]["result"]: + channels.append(MonitoredChannel.model_validate(metric["metric"])) + return channels class Command(BaseCommand): - help = "Register Nix channels" + help = "Fetch current channel tips" def handle(self, *args: Any, **kwargs: Any) -> str | None: - fresh_channels = fetch_from_monitoring() - for channel in fresh_channels.values(): - channel_branch = channel.name - branch_info = { - "release_branch": release_branch(channel.name), - "state": NixChannel.ChannelState(channel.status), - "head_sha1_commit": channel.revision, - "variant": channel.variant, - } - pprint(branch_info | {"channel_branch": channel.name}) - NixChannel.objects.update_or_create( - branch_info, channel_branch=channel_branch - ) - - repo = GitRepo( - settings.LOCAL_NIXPKGS_CHECKOUT, - stderr=sys.stderr.fileno(), - ) - parallel_fetches = [] - for channel in NixChannel.objects.iterator(): - parallel_fetches.append(repo.update_from_ref(channel.head_sha1_commit)) + channels = fetch_from_monitoring() + + branch_tips: dict[str, str] = { + release_branch: get_head_sha1(settings.GIT_CLONE_URL, release_branch) + for release_branch in {c.release_branch for c in channels} + } + + with transaction.atomic(): + branches: dict[str, NixpkgsBranch] = {} + for name, head in branch_tips.items(): + branch, _ = NixpkgsBranch.objects.update_or_create( + name=name, + defaults={"head_sha1_commit": head}, + ) + branches[name] = branch + + for monitored in channels: + NixChannel.objects.update_or_create( + channel_branch=monitored.channel, + defaults=dict( + release_branch=branches[monitored.release_branch], + head_sha1_commit=monitored.revision, + state=monitored.status, + variant=monitored.variant, + ), + ) - results = asyncio.run(wait_for_parallel_fetches(parallel_fetches)) - # FIXME(@fricklerhandwerk): Fold that into `branch_info`, so there's only one output. - print("Parallel fetches results", results) + # Can't `pprint()` to `self.stdout` directly... + stream = StringIO() + pprint(monitored.__dict__, stream=stream) + self.stdout.write(stream.getvalue()) diff --git a/src/shared/migrations/0098_nixpkgsbranch_nixchannel_release_branch.py b/src/shared/migrations/0098_nixpkgsbranch_nixchannel_release_branch.py new file mode 100644 index 00000000..0140d507 --- /dev/null +++ b/src/shared/migrations/0098_nixpkgsbranch_nixchannel_release_branch.py @@ -0,0 +1,82 @@ +import django.db.models.deletion +from django.db import migrations, models + + +def populate_release_branch(apps, schema_editor): + NixChannel = apps.get_model("shared", "NixChannel") + NixpkgsBranch = apps.get_model("shared", "NixpkgsBranch") + for channel in NixChannel.objects.all(): + branch, _ = NixpkgsBranch.objects.get_or_create( + name=channel.release_branch_name, + defaults={"head_sha1_commit": channel.head_sha1_commit}, + ) + channel.release_branch = branch + channel.save(update_fields=["release_branch"]) + + +class Migration(migrations.Migration): + + dependencies = [ + ("shared", "0097_nixchannel_variant"), + ] + + operations = [ + migrations.CreateModel( + name="NixpkgsBranch", + fields=[ + ( + "name", + models.CharField(max_length=126, primary_key=True, serialize=False), + ), + ("head_sha1_commit", models.CharField(max_length=40)), + ], + options={ + "constraints": [ + models.CheckConstraint( + check=models.Q(head_sha1_commit__regex="^[0-9a-f]{40}$"), + name="nixpkgsbranch_head_sha1_commit_valid", + ) + ] + }, + ), + migrations.RenameField( + model_name="nixchannel", + old_name="release_branch", + new_name="release_branch_name", + ), + migrations.AddField( + model_name="nixchannel", + name="release_branch", + field=models.ForeignKey( + null=True, + on_delete=django.db.models.deletion.PROTECT, + related_name="channels", + to="shared.nixpkgsbranch", + ), + ), + # The table has triggers registered. + # Altering it within the migration will defer them to after the transaction, thus failing the whole thing. + migrations.RunSQL("ALTER TABLE shared_nixchannel DISABLE TRIGGER USER"), + migrations.RunPython( + code=populate_release_branch, + reverse_code=migrations.RunPython.noop, + ), + migrations.AlterField( + model_name="nixchannel", + name="release_branch", + field=models.ForeignKey( + on_delete=django.db.models.deletion.PROTECT, + related_name="channels", + to="shared.nixpkgsbranch", + ), + ), + migrations.RemoveField( + model_name="nixchannel", + name="release_branch_name", + ), + migrations.RemoveField( + model_name="nixchannel", + name="repository", + ), + migrations.RunSQL("ALTER TABLE shared_nixchannel ENABLE TRIGGER USER"), + ] diff --git a/src/shared/models/nix_evaluation.py b/src/shared/models/nix_evaluation.py index 978400c9..2b9c1969 100644 --- a/src/shared/models/nix_evaluation.py +++ b/src/shared/models/nix_evaluation.py @@ -1,5 +1,3 @@ -import re - from django.conf import settings from django.contrib.postgres import fields from django.contrib.postgres.indexes import BTreeIndex, GinIndex @@ -129,6 +127,26 @@ def __str__(self) -> str: return self.description or "" +class NixpkgsBranch(models.Model): + """ + A Nixpkgs branch that gets evaluated, e.g. `master` or `release-26.05`. + """ + + name = models.CharField(max_length=126, primary_key=True) + head_sha1_commit = models.CharField(max_length=40) + + class Meta: + constraints = [ + models.CheckConstraint( + check=models.Q(head_sha1_commit__regex=r"^[0-9a-f]{40}$"), + name="nixpkgsbranch_head_sha1_commit_valid", + ) + ] + + def __str__(self) -> str: + return self.name + + class NixChannel(TimeStampMixin): """ This represents a "Nixpkgs" (*) channel, e.g. @@ -160,9 +178,12 @@ class Variant(models.TextChoices): ChannelState.UNSTABLE, ) - # A staging branch is the `release-$number` branch or `master` for unstable. - # Not to confuse with the `staging` branch itself. - release_branch = models.CharField(max_length=255) + # The underlying Nixpkgs branch (e.g. `master` or `release-25.05`). + release_branch = models.ForeignKey( + NixpkgsBranch, + on_delete=models.PROTECT, + related_name="channels", + ) # A channel branch is the `nixos-$number` branch of # `nixos-unstable(-small)` for unstable(-small). Not to confuse with the # channel tarballs and scripts from releases.nixos.org. @@ -171,11 +192,6 @@ class Variant(models.TextChoices): head_sha1_commit = models.CharField(max_length=255) state = models.CharField(max_length=126, choices=ChannelState.choices) variant = models.CharField(max_length=126, choices=Variant.choices, null=True) - # Repository can be stored as URLs for now... - # We can always reparse them as proper GitHub URIs if necessary - # It's a bit annoying though - # TODO(raitobezarius): make a proper ForeignKey? - repository = models.CharField(max_length=255) def __str__(self) -> str: return f"{self.release_branch} -> {self.channel_branch}" @@ -330,18 +346,3 @@ def get_major_channel(branch_name: str) -> str | None: if mc in branch_name: return f"nixos-{mc}" return None - - -def get_release(channel_branch: str) -> str: - match = re.match( - r"^(?P[a-z]+)-(?P\d\d\.\d\d|unstable)(?:-(?P[a-z]+))?$", - channel_branch, - ) - if match is None: - raise ValueError(f"unexpected channel branch name: {channel_branch!r}") - return match.group("release") - - -def release_branch(channel_branch: str) -> str: - release = get_release(channel_branch) - return "master" if release == "unstable" else f"release-{release}" diff --git a/src/shared/tests/conftest.py b/src/shared/tests/conftest.py index 4bff6c02..0ae3d638 100644 --- a/src/shared/tests/conftest.py +++ b/src/shared/tests/conftest.py @@ -37,7 +37,7 @@ NixDerivationMeta, NixEvaluation, NixMaintainer, - release_branch, + NixpkgsBranch, ) from shared.models.package import Package, PackageAttrpath from shared.notify_users import create_package_subscription_notifications @@ -134,21 +134,41 @@ def cve(make_container: Callable[..., Container]) -> Container: @pytest.fixture -def make_channel(db: None) -> Callable[..., NixChannel]: +def make_branch(db: None) -> Callable[..., NixpkgsBranch]: + def wrapped(name: str = "master") -> NixpkgsBranch: + branch, _ = NixpkgsBranch.objects.get_or_create( + name=name, + defaults={"head_sha1_commit": secrets.token_hex(20)}, + ) + return branch + + return wrapped + + +@pytest.fixture +def branch(make_branch: Callable[..., NixpkgsBranch]) -> NixpkgsBranch: + return make_branch() + + +@pytest.fixture +def make_channel( + db: None, + branch: NixpkgsBranch, +) -> Callable[..., NixChannel]: # FIXME(@fricklerhandwerk): This will fall apart when we obtain the channel structure dynamically [ref:channel-structure] def wrapped( channel_branch: str = settings.TRACKING_BRANCH, + release_branch: NixpkgsBranch = branch, state: NixChannel.ChannelState = NixChannel.ChannelState.UNSTABLE, variant: NixChannel.Variant | None = None, ) -> NixChannel: channel, _ = NixChannel.objects.get_or_create( channel_branch=channel_branch, defaults=dict( - head_sha1_commit=secrets.token_hex(16), + head_sha1_commit=secrets.token_hex(20), state=state, variant=variant, - release_branch=release_branch(channel_branch), - repository="https://github.com/NixOS/nixpkgs", + release_branch=release_branch, ), ) return channel @@ -177,7 +197,7 @@ def wrapped( channel=channel, commit_sha1=commit_sha1 if commit_sha1 is not None - else secrets.token_hex(16), + else secrets.token_hex(20), state=state, ) diff --git a/src/shared/tests/test_fetch_all_channels.py b/src/shared/tests/test_fetch_all_channels.py new file mode 100644 index 00000000..b99dfa39 --- /dev/null +++ b/src/shared/tests/test_fetch_all_channels.py @@ -0,0 +1,179 @@ +from collections.abc import Callable, Generator +from contextlib import contextmanager +from unittest.mock import Mock, patch + +import pytest +from django.core.management import call_command +from django.db import IntegrityError +from pydantic import AnyUrl + +from shared.git import get_head_sha1 +from shared.management.commands.fetch_all_channels import fetch_from_monitoring +from shared.models.nix_evaluation import NixChannel, NixpkgsBranch + + +def monitoring_response(*channels: dict) -> Mock: + resp = Mock() + resp.json.return_value = {"data": {"result": [{"metric": c} for c in channels]}} + return resp + + +@pytest.mark.parametrize( + "name,name_valid", + [ + ("nixos-25.05", True), + ("nixos-unstable", True), + ("nixos", False), + ("NixOS-25.05", False), + ("--evil", False), + ("", False), + ], +) +@pytest.mark.parametrize( + "revision,revision_valid", + [ + ("a" * 40, True), + ("abc123", False), + ("z" * 40, False), + ("--upload-pack=evil" + "a" * 21, False), + ("", False), + ], +) +@pytest.mark.parametrize( + "status,status_valid", + [ + ("stable", True), + ("rolling", True), + ("STABLE", False), + ("1", False), + ("", False), + ], +) +def test_input_validation( + name: str, + name_valid: bool, + revision: str, + revision_valid: bool, + status: str, + status_valid: bool, +) -> None: + with patch( + "requests.get", + return_value=monitoring_response( + {"channel": name, "revision": revision, "status": status} + ), + ): + if name_valid and revision_valid and status_valid: + [channel] = fetch_from_monitoring() + assert channel.channel == name + assert channel.revision == revision + assert channel.status == status + else: + with pytest.raises(Exception): + fetch_from_monitoring() + + +def test_get_head_sha1_returns_commit() -> None: + with patch( + "subprocess.run", + return_value=Mock(stdout=f"{'a' * 40}\trefs/heads/master\n"), + ): + assert ( + get_head_sha1(AnyUrl("https://github.com/NixOS/nixpkgs"), "master") + == "a" * 40 + ) + + +def test_get_head_sha1_raises_on_missing_branch() -> None: + with patch("subprocess.run", return_value=Mock(stdout="")): + with pytest.raises(ValueError, match="not found"): + get_head_sha1(AnyUrl("https://github.com/NixOS/nixpkgs"), "nonexistent") + + +@pytest.mark.django_db +def test_nixpkgsbranch_rejects_invalid_sha1() -> None: + with pytest.raises(IntegrityError): + NixpkgsBranch.objects.create(name="master", head_sha1_commit="not-a-sha1") + + +@pytest.fixture +def mock_monitoring() -> Callable: + @contextmanager + def factory(*channels: dict) -> Generator[None]: + with patch( + "requests.get", + return_value=monitoring_response(*channels), + ): + yield + + return factory + + +@pytest.fixture +def mock_head_sha1() -> Callable: + @contextmanager + def factory(branch_commits: dict[str, str]) -> Generator[None]: + with patch( + "shared.management.commands.fetch_all_channels.get_head_sha1", + side_effect=lambda _, branch: branch_commits[branch], + ): + yield + + return factory + + +@pytest.mark.django_db +def test_handle_creates_branches( + mock_monitoring: Callable, mock_head_sha1: Callable +) -> None: + with ( + mock_monitoring( + { + "channel": "nixos-unstable", + "revision": "a" * 40, + "status": "rolling", + }, + { + "channel": "nixos-unstable-small", + "revision": "b" * 40, + "status": "rolling", + }, + ), + mock_head_sha1(branch_commits={"master": "c" * 40}), + ): + call_command("fetch_all_channels") + + assert NixpkgsBranch.objects.count() == 1 + assert NixpkgsBranch.objects.filter( + name="master", head_sha1_commit="c" * 40 + ).exists() + unstable = NixChannel.objects.get(channel_branch="nixos-unstable") + assert unstable.head_sha1_commit == "a" * 40 + unstable_small = NixChannel.objects.get(channel_branch="nixos-unstable-small") + assert unstable_small.head_sha1_commit == "b" * 40 + + +@pytest.mark.django_db +def test_handle_updates_branch_head( + mock_monitoring: Callable, mock_head_sha1: Callable +) -> None: + NixpkgsBranch.objects.create(name="master", head_sha1_commit="f" * 40) + + with ( + mock_monitoring( + { + "channel": "nixos-unstable", + "revision": "a" * 40, + "status": "rolling", + }, + { + "channel": "nixos-unstable-small", + "revision": "b" * 40, + "status": "rolling", + }, + ), + mock_head_sha1(branch_commits={"master": "c" * 40}), + ): + call_command("fetch_all_channels") + + assert NixpkgsBranch.objects.get(name="master").head_sha1_commit == "c" * 40 diff --git a/src/shared/tests/test_garbage_collect.py b/src/shared/tests/test_garbage_collect.py index af3e8585..dce1f3ac 100644 --- a/src/shared/tests/test_garbage_collect.py +++ b/src/shared/tests/test_garbage_collect.py @@ -237,8 +237,8 @@ def test_deletes_empty_old_evaluations( """ channel = make_channel( state=channel_state, - # Make a unique channel for each state; the concrete name doesn't matter here. - channel_branch=f"{channel_state.value}-unstable", + # FIXME(@fricklerhandwerk): When leaving this at the default value, the channel is created via `make_drv()` at parse time... + channel_branch="nixos-unstable", ) evaluation = make_evaluation( channel=channel, diff --git a/src/shared/tests/test_suggestion_caching.py b/src/shared/tests/test_suggestion_caching.py index 4ab6a2aa..41341afe 100644 --- a/src/shared/tests/test_suggestion_caching.py +++ b/src/shared/tests/test_suggestion_caching.py @@ -2,7 +2,6 @@ from datetime import timedelta import pytest -from django.conf import settings from shared.cache_suggestions import cache_new_suggestions from shared.models.cached import CachedSuggestions @@ -17,6 +16,7 @@ NixDerivation, NixEvaluation, NixMaintainer, + NixpkgsBranch, ) from shared.models.package import Package from shared.package_clustering import cluster_packages @@ -86,6 +86,7 @@ def test_caching_newest_package( @pytest.mark.parametrize("rolling_has_maintainers", [True, False]) def test_maintainers_come_from_rolling_release_channel( cve: Container, + make_branch: Callable[..., NixpkgsBranch], make_channel: Callable[..., NixChannel], make_evaluation: Callable[..., NixEvaluation], make_drv: Callable[..., NixDerivation], @@ -99,8 +100,11 @@ def test_maintainers_come_from_rolling_release_channel( Its maintainers must appear regardless of evaluation order, and stable-only maintainers must never appear. """ - stable_channel = make_channel(channel_branch="nixos-26.05-small") - rolling_channel = make_channel(channel_branch=settings.TRACKING_BRANCH) + stable_channel = make_channel( + channel_branch="nixos-26.05-small", + release_branch=make_branch(name="release-26.05"), + ) + rolling_channel = make_channel() stable_age = timedelta(hours=1) if stable_is_older else timedelta(0) rolling_age = timedelta(0) if stable_is_older else timedelta(hours=1) eval_stable = make_evaluation(channel=stable_channel, age=stable_age)