diff --git a/swh/loader/git/dumb.py b/swh/loader/git/dumb.py index 35826e9..7197363 100644 --- a/swh/loader/git/dumb.py +++ b/swh/loader/git/dumb.py @@ -1,204 +1,204 @@ # Copyright (C) 2021 The Software Heritage developers # See the AUTHORS file at the top-level directory of this distribution # License: GNU General Public License version 3, or any later version # See top-level LICENSE file for more information from __future__ import annotations from collections import defaultdict import logging import stat import struct from tempfile import SpooledTemporaryFile from typing import TYPE_CHECKING, Callable, Dict, Iterable, List, Set, cast import urllib.parse from dulwich.errors import NotGitRepository from dulwich.objects import S_IFGITLINK, Commit, ShaFile, Tree from dulwich.pack import Pack, PackData, PackIndex, load_pack_index_file import requests from swh.loader.git.utils import HexBytes if TYPE_CHECKING: from .loader import RepoRepresentation logger = logging.getLogger(__name__) HEADERS = {"User-Agent": "Software Heritage dumb Git loader"} def check_protocol(repo_url: str) -> bool: """Checks if a git repository can be cloned using the dumb protocol. Args: repo_url: Base URL of a git repository Returns: Whether the dumb protocol is supported. """ if not repo_url.startswith("http"): return False url = urllib.parse.urljoin( repo_url.rstrip("/") + "/", "info/refs?service=git-upload-pack/" ) logger.debug("Fetching %s", url) response = requests.get(url, headers=HEADERS) content_type = response.headers.get("Content-Type") return ( response.status_code in ( 200, 304, ) # header is not mandatory in protocol specification and (content_type is None or not content_type.startswith("application/x-git-")) ) class GitObjectsFetcher: """Git objects fetcher using dumb HTTP protocol. Fetches a set of git objects for a repository according to its archival state by Software Heritage and provides iterators on them. Args: repo_url: Base URL of a git repository base_repo: State of repository archived by Software Heritage """ def __init__(self, repo_url: str, base_repo: RepoRepresentation): self._session = requests.Session() self.repo_url = repo_url self.base_repo = base_repo self.objects: Dict[bytes, Set[bytes]] = defaultdict(set) self.refs = self._get_refs() self.head = self._get_head() if self.refs else {} self.packs = self._get_packs() def fetch_object_ids(self) -> None: """Fetches identifiers of git objects to load into the archive.""" wants = self.base_repo.determine_wants(self.refs) # process refs commit_objects = [] for ref in wants: ref_object = self._get_git_object(ref) if ref_object.type_num == Commit.type_num: commit_objects.append(cast(Commit, ref_object)) self.objects[b"commit"].add(ref) else: self.objects[b"tag"].add(ref) # perform DFS on commits graph while commit_objects: commit = commit_objects.pop() # fetch tree and blob ids recursively self._fetch_tree_objects(commit.tree) for parent in commit.parents: if ( # commit not already seen in the current load parent not in self.objects[b"commit"] # commit not already archived by a previous load - and parent not in self.base_repo.heads + and parent not in self.base_repo.local_heads ): commit_objects.append(cast(Commit, self._get_git_object(parent))) self.objects[b"commit"].add(parent) def iter_objects(self, object_type: bytes) -> Iterable[ShaFile]: """Returns a generator on fetched git objects per type. Args: object_type: Git object type, either b"blob", b"commit", b"tag" or b"tree" Returns: A generator fetching git objects on the fly. """ return map(self._get_git_object, self.objects[object_type]) def _http_get(self, path: str) -> SpooledTemporaryFile: url = urllib.parse.urljoin(self.repo_url.rstrip("/") + "/", path) logger.debug("Fetching %s", url) response = self._session.get(url, headers=HEADERS) buffer = SpooledTemporaryFile(max_size=100 * 1024 * 1024) for chunk in response.iter_content(chunk_size=10 * 1024 * 1024): buffer.write(chunk) buffer.flush() buffer.seek(0) return buffer def _get_refs(self) -> Dict[bytes, HexBytes]: refs = {} refs_resp_bytes = self._http_get("info/refs") for ref_line in refs_resp_bytes.readlines(): ref_target, ref_name = ref_line.replace(b"\n", b"").split(b"\t") refs[ref_name] = ref_target return refs def _get_head(self) -> Dict[bytes, HexBytes]: head_resp_bytes = self._http_get("HEAD") _, head_target = head_resp_bytes.readline().replace(b"\n", b"").split(b" ") return {b"HEAD": head_target} def _get_pack_data(self, pack_name: str) -> Callable[[], PackData]: def _pack_data() -> PackData: pack_data_bytes = self._http_get(f"objects/pack/{pack_name}") return PackData(pack_name, file=pack_data_bytes) return _pack_data def _get_pack_idx(self, pack_idx_name: str) -> Callable[[], PackIndex]: def _pack_idx() -> PackIndex: pack_idx_bytes = self._http_get(f"objects/pack/{pack_idx_name}") return load_pack_index_file(pack_idx_name, pack_idx_bytes) return _pack_idx def _get_packs(self) -> List[Pack]: packs = [] packs_info_bytes = self._http_get("objects/info/packs") packs_info = packs_info_bytes.read().decode() for pack_info in packs_info.split("\n"): if pack_info: pack_name = pack_info.split(" ")[1] pack_idx_name = pack_name.replace(".pack", ".idx") # pack index and data file will be lazily fetched when required packs.append( Pack.from_lazy_objects( self._get_pack_data(pack_name), self._get_pack_idx(pack_idx_name), ) ) return packs def _get_git_object(self, sha: bytes) -> ShaFile: # try to get the object from a pack file first to avoid flooding # git server with numerous HTTP requests for pack in list(self.packs): try: if sha in pack: return pack[sha] except (NotGitRepository, struct.error): # missing (dulwich http client raises NotGitRepository on 404) # or invalid pack index/content, remove it from global packs list logger.debug("A pack file is missing or its content is invalid") self.packs.remove(pack) # fetch it from objects/ directory otherwise sha_hex = sha.decode() object_path = f"objects/{sha_hex[:2]}/{sha_hex[2:]}" return ShaFile.from_file(self._http_get(object_path)) def _fetch_tree_objects(self, sha: bytes) -> None: if sha not in self.objects[b"tree"]: tree = cast(Tree, self._get_git_object(sha)) self.objects[b"tree"].add(sha) for item in tree.items(): if item.mode == S_IFGITLINK: # skip submodules as objects are not stored in repository continue if item.mode & stat.S_IFDIR: self._fetch_tree_objects(item.sha) else: self.objects[b"blob"].add(item.sha) diff --git a/swh/loader/git/loader.py b/swh/loader/git/loader.py index 86b4c6b..eebd19b 100644 --- a/swh/loader/git/loader.py +++ b/swh/loader/git/loader.py @@ -1,570 +1,562 @@ # Copyright (C) 2016-2022 The Software Heritage developers # See the AUTHORS file at the top-level directory of this distribution # License: GNU General Public License version 3, or any later version # See top-level LICENSE file for more information from dataclasses import dataclass import datetime import logging import os import pickle import sys from tempfile import SpooledTemporaryFile from typing import Any, Callable, Dict, Iterable, Iterator, List, Optional, Set, Type import dulwich.client from dulwich.errors import GitProtocolError, NotGitRepository from dulwich.object_store import ObjectStoreGraphWalker from dulwich.objects import ShaFile from dulwich.pack import PackData, PackInflater from swh.core.statsd import Statsd from swh.loader.exception import NotFound from swh.model import hashutil from swh.model.model import ( BaseContent, Directory, Release, Revision, Snapshot, SnapshotBranch, TargetType, ) from swh.storage.algos.snapshot import snapshot_get_latest from swh.storage.interface import StorageInterface from . import converters, dumb, utils from .base import BaseGitLoader from .utils import HexBytes logger = logging.getLogger(__name__) heads_logger = logger.getChild("refs") class RepoRepresentation: """Repository representation for a Software Heritage origin.""" def __init__( self, storage, base_snapshots: List[Snapshot] = None, incremental: bool = True, statsd: Statsd = None, ): self.storage = storage self.incremental = incremental self.statsd = statsd if base_snapshots and incremental: self.base_snapshots: List[Snapshot] = base_snapshots else: self.base_snapshots = [] - self.heads: Set[HexBytes] = set() - - def get_parents(self, commit: bytes) -> List[bytes]: - """This method should return the list of known parents""" - return [] + # Cache existing heads + self.local_heads: Set[HexBytes] = set() + heads_logger.debug("Heads known in the archive:") + for base_snapshot in self.base_snapshots: + for branch_name, branch in base_snapshot.branches.items(): + if not branch or branch.target_type == TargetType.ALIAS: + continue + heads_logger.debug(" %r: %s", branch_name, branch.target.hex()) + self.local_heads.add(HexBytes(hashutil.hash_to_bytehex(branch.target))) def graph_walker(self) -> ObjectStoreGraphWalker: - return ObjectStoreGraphWalker(self.heads, self.get_parents) + return ObjectStoreGraphWalker(self.local_heads, get_parents=lambda commit: []) def determine_wants(self, refs: Dict[bytes, HexBytes]) -> List[HexBytes]: """Get the list of bytehex sha1s that the git loader should fetch. This compares the remote refs sent by the server with the base snapshot provided by the loader. """ if not refs: return [] if heads_logger.isEnabledFor(logging.DEBUG): heads_logger.debug("Heads returned by the git remote:") for name, value in refs.items(): heads_logger.debug(" %r: %s", name, value.decode()) - heads_logger.debug("Heads known in the archive:") - # Cache existing heads - local_heads: Set[HexBytes] = set() - for base_snapshot in self.base_snapshots: - for branch_name, branch in base_snapshot.branches.items(): - if not branch or branch.target_type == TargetType.ALIAS: - continue - heads_logger.debug(" %r: %s", branch_name, branch.target.hex()) - local_heads.add(HexBytes(hashutil.hash_to_bytehex(branch.target))) - - self.heads = local_heads - # Get the remote heads that we want to fetch remote_heads: Set[HexBytes] = set() for ref_name, ref_target in refs.items(): if utils.ignore_branch_name(ref_name): continue remote_heads.add(ref_target) - logger.debug("local_heads_count=%s", len(local_heads)) + logger.debug("local_heads_count=%s", len(self.local_heads)) logger.debug("remote_heads_count=%s", len(remote_heads)) - wanted_refs = list(remote_heads - local_heads) + wanted_refs = list(remote_heads - self.local_heads) logger.debug("wanted_refs_count=%s", len(wanted_refs)) if self.statsd is not None: self.statsd.histogram( "git_ignored_refs_percent", len(remote_heads - set(refs.values())) / len(refs), tags={}, ) self.statsd.histogram( "git_known_refs_percent", - len(local_heads & remote_heads) / len(remote_heads), + len(self.local_heads & remote_heads) / len(remote_heads), tags={}, ) return wanted_refs @dataclass class FetchPackReturn: remote_refs: Dict[bytes, HexBytes] symbolic_refs: Dict[bytes, HexBytes] pack_buffer: SpooledTemporaryFile pack_size: int class GitLoader(BaseGitLoader): """A bulk loader for a git repository Emits the following statsd stats: * increments ``swh_loader_git`` * histogram ``swh_loader_git_ignored_refs_percent`` is the ratio of refs ignored over all refs of the remote repository * histogram ``swh_loader_git_known_refs_percent`` is the ratio of (non-ignored) remote heads that are already local over all non-ignored remote heads All three are tagged with ``{{"incremental": ""}}`` where ``incremental_mode`` is one of: * ``from_same_origin`` when the origin was already loaded * ``from_parent_origin`` when the origin was not already loaded, but it was detected as a forge-fork of an origin that was already loaded * ``no_previous_snapshot`` when the origin was not already loaded, and it was detected as a forge-fork of origins that were not already loaded either * ``no_parent_origin`` when the origin was no already loaded, and it was not detected as a forge-fork of any other origin * ``disabled`` when incremental loading is disabled by configuration """ visit_type = "git" def __init__( self, storage: StorageInterface, url: str, incremental: bool = True, repo_representation: Type[RepoRepresentation] = RepoRepresentation, pack_size_bytes: int = 4 * 1024 * 1024 * 1024, temp_file_cutoff: int = 100 * 1024 * 1024, **kwargs: Any, ): """Initialize the bulk updater. Args: repo_representation: swh's repository representation which is in charge of filtering between known and remote data. ... incremental: If True, the default, this starts from the last known snapshot (if any) references. Otherwise, this loads the full repository. """ super().__init__(storage=storage, origin_url=url, **kwargs) self.incremental = incremental self.repo_representation = repo_representation self.pack_size_bytes = pack_size_bytes self.temp_file_cutoff = temp_file_cutoff # state initialized in fetch_data self.remote_refs: Dict[bytes, HexBytes] = {} self.symbolic_refs: Dict[bytes, HexBytes] = {} self.ref_object_types: Dict[bytes, Optional[TargetType]] = {} def fetch_pack_from_origin( self, origin_url: str, base_repo: RepoRepresentation, do_activity: Callable[[bytes], None], ) -> FetchPackReturn: """Fetch a pack from the origin""" pack_buffer = SpooledTemporaryFile(max_size=self.temp_file_cutoff) transport_url = origin_url logger.debug("Transport url to communicate with server: %s", transport_url) client, path = dulwich.client.get_transport_and_path( transport_url, thin_packs=False ) logger.debug("Client %s to fetch pack at %s", client, path) size_limit = self.pack_size_bytes def do_pack(data: bytes) -> None: cur_size = pack_buffer.tell() would_write = len(data) if cur_size + would_write > size_limit: raise IOError( f"Pack file too big for repository {origin_url}, " f"limit is {size_limit} bytes, current size is {cur_size}, " f"would write {would_write}" ) pack_buffer.write(data) pack_result = client.fetch_pack( path, base_repo.determine_wants, base_repo.graph_walker(), do_pack, progress=do_activity, ) remote_refs = pack_result.refs or {} symbolic_refs = pack_result.symrefs or {} pack_buffer.flush() pack_size = pack_buffer.tell() pack_buffer.seek(0) logger.debug("fetched_pack_size=%s", pack_size) # check if repository only supports git dumb transfer protocol, # fetched pack file will be empty in that case as dulwich do # not support it and do not fetch any refs self.dumb = transport_url.startswith("http") and getattr(client, "dumb", False) return FetchPackReturn( remote_refs=utils.filter_refs(remote_refs), symbolic_refs=utils.filter_refs(symbolic_refs), pack_buffer=pack_buffer, pack_size=pack_size, ) def get_full_snapshot(self, origin_url) -> Optional[Snapshot]: return snapshot_get_latest(self.storage, origin_url) def prepare(self) -> None: assert self.origin is not None self.prev_snapshot = Snapshot(branches={}) """Last snapshot of this origin if any; empty snapshot otherwise""" self.base_snapshots = [] """Last snapshot of this origin and all its parents, if any.""" self.statsd.constant_tags["incremental_enabled"] = self.incremental self.statsd.constant_tags["has_parent_origins"] = bool(self.parent_origins) # May be set to True later self.statsd.constant_tags["has_parent_snapshot"] = False if self.incremental: prev_snapshot = self.get_full_snapshot(self.origin.url) self.statsd.constant_tags["has_previous_snapshot"] = bool(prev_snapshot) if prev_snapshot: self.prev_snapshot = prev_snapshot self.base_snapshots.append(prev_snapshot) if self.parent_origins is not None: # If this origin is a forge fork, load incrementally from the # origins it was forked from for parent_origin in self.parent_origins: parent_snapshot = self.get_full_snapshot(parent_origin.url) if parent_snapshot is not None: self.statsd.constant_tags["has_parent_snapshot"] = True self.base_snapshots.append(parent_snapshot) # Increments a metric with full name 'swh_loader_git'; which is useful to # count how many runs of the loader are with each incremental mode self.statsd.increment("git_total", tags={}) def fetch_data(self) -> bool: assert self.origin is not None base_repo = self.repo_representation( storage=self.storage, base_snapshots=self.base_snapshots, incremental=self.incremental, statsd=self.statsd, ) def do_progress(msg: bytes) -> None: sys.stderr.buffer.write(msg) sys.stderr.flush() try: fetch_info = self.fetch_pack_from_origin( self.origin.url, base_repo, do_progress ) except (dulwich.client.HTTPUnauthorized, NotGitRepository) as e: raise NotFound(e) except GitProtocolError as e: # unfortunately, that kind of error is not specific to a not found # scenario... It depends on the value of message within the exception. for msg in [ "Repository unavailable", # e.g DMCA takedown "Repository not found", "unexpected http resp 401", ]: if msg in e.args[0]: raise NotFound(e) # otherwise transmit the error raise except (AttributeError, NotImplementedError, ValueError): # with old dulwich versions, those exceptions types can be raised # by the fetch_pack operation when encountering a repository with # dumb transfer protocol so we check if the repository supports it # here to continue the loading if it is the case self.dumb = dumb.check_protocol(self.origin.url) if not self.dumb: raise logger.debug( "Protocol used for communication: %s", "dumb" if self.dumb else "smart" ) if self.dumb: self.dumb_fetcher = dumb.GitObjectsFetcher(self.origin.url, base_repo) self.dumb_fetcher.fetch_object_ids() self.remote_refs = utils.filter_refs(self.dumb_fetcher.refs) # type: ignore self.symbolic_refs = self.dumb_fetcher.head else: self.pack_buffer = fetch_info.pack_buffer self.pack_size = fetch_info.pack_size self.remote_refs = fetch_info.remote_refs self.symbolic_refs = fetch_info.symbolic_refs self.ref_object_types = {sha1: None for sha1 in self.remote_refs.values()} logger.info( "Listed %d refs for repo %s", len(self.remote_refs), self.origin.url, extra={ "swh_type": "git_repo_list_refs", "swh_repo": self.origin.url, "swh_num_refs": len(self.remote_refs), }, ) # No more data to fetch return False def save_data(self) -> None: """Store a pack for archival""" assert isinstance(self.visit_date, datetime.datetime) write_size = 8192 pack_dir = self.get_save_data_path() pack_name = "%s.pack" % self.visit_date.isoformat() refs_name = "%s.refs" % self.visit_date.isoformat() with open(os.path.join(pack_dir, pack_name), "xb") as f: self.pack_buffer.seek(0) while True: r = self.pack_buffer.read(write_size) if not r: break f.write(r) self.pack_buffer.seek(0) with open(os.path.join(pack_dir, refs_name), "xb") as f: pickle.dump(self.remote_refs, f) def iter_objects(self, object_type: bytes) -> Iterator[ShaFile]: """Read all the objects of type `object_type` from the packfile""" if self.dumb: yield from self.dumb_fetcher.iter_objects(object_type) else: self.pack_buffer.seek(0) count = 0 for obj in PackInflater.for_pack_data( PackData.from_file(self.pack_buffer, self.pack_size) ): if obj.type_name != object_type: continue yield obj count += 1 logger.debug("packfile_read_count_%s=%s", object_type.decode(), count) def get_contents(self) -> Iterable[BaseContent]: """Format the blobs from the git repository as swh contents""" for raw_obj in self.iter_objects(b"blob"): if raw_obj.id in self.ref_object_types: self.ref_object_types[raw_obj.id] = TargetType.CONTENT yield converters.dulwich_blob_to_content( raw_obj, max_content_size=self.max_content_size ) def get_directories(self) -> Iterable[Directory]: """Format the trees as swh directories""" for raw_obj in self.iter_objects(b"tree"): if raw_obj.id in self.ref_object_types: self.ref_object_types[raw_obj.id] = TargetType.DIRECTORY yield converters.dulwich_tree_to_directory(raw_obj) def get_revisions(self) -> Iterable[Revision]: """Format commits as swh revisions""" for raw_obj in self.iter_objects(b"commit"): if raw_obj.id in self.ref_object_types: self.ref_object_types[raw_obj.id] = TargetType.REVISION yield converters.dulwich_commit_to_revision(raw_obj) def get_releases(self) -> Iterable[Release]: """Retrieve all the release objects from the git repository""" for raw_obj in self.iter_objects(b"tag"): if raw_obj.id in self.ref_object_types: self.ref_object_types[raw_obj.id] = TargetType.RELEASE yield converters.dulwich_tag_to_release(raw_obj) def get_snapshot(self) -> Snapshot: """Get the snapshot for the current visit. The main complexity of this function is mapping target objects to their types, as the `refs` dictionaries returned by the git server only give us the identifiers for the target objects, and not their types. The loader itself only knows the types of the objects that it has fetched from the server (as it has parsed them while loading them to the archive). As we only fetched an increment between the previous snapshot and the current state of the server, we are missing the type information for the objects that would already have been referenced by the previous snapshot, and that the git server didn't send us. We infer the type of these objects from the previous snapshot. """ branches: Dict[bytes, Optional[SnapshotBranch]] = {} unfetched_refs: Dict[bytes, bytes] = {} # Retrieve types from the objects loaded by the current loader for ref_name, ref_object in self.remote_refs.items(): if ref_name in self.symbolic_refs: continue target = hashutil.hash_to_bytes(ref_object.decode()) target_type = self.ref_object_types.get(ref_object) if target_type: branches[ref_name] = SnapshotBranch( target=target, target_type=target_type ) else: # The object pointed at by this ref was not fetched, supposedly # because it existed in the base snapshot. We record it here, # and we can get it from the base snapshot later. unfetched_refs[ref_name] = target dangling_branches = {} # Handle symbolic references as alias branches for ref_name, target in self.symbolic_refs.items(): branches[ref_name] = SnapshotBranch( target_type=TargetType.ALIAS, target=target, ) if target not in branches and target not in unfetched_refs: # This handles the case where the pointer is "dangling". # There's a chance that a further symbolic reference # override this default value, which is totally fine. dangling_branches[target] = ref_name branches[target] = None if unfetched_refs: # Handle inference of object types from the contents of the # previous snapshot unknown_objects = {} base_snapshot_reverse_branches = { branch.target: branch for base_snapshot in reversed(self.base_snapshots) for branch in base_snapshot.branches.values() if branch and branch.target_type != TargetType.ALIAS } assert all( base_snapshot_reverse_branches[branch.target] == branch for branch in self.prev_snapshot.branches.values() if branch and branch.target_type != TargetType.ALIAS ), "base_snapshot_reverse_branches is not a superset of prev_snapshot" for ref_name, target in unfetched_refs.items(): branch = base_snapshot_reverse_branches.get(target) branches[ref_name] = branch if not branch: unknown_objects[ref_name] = target if unknown_objects: # This object was referenced by the server; We did not fetch # it, and we do not know it from the previous snapshot. This is # likely a bug in the loader. raise RuntimeError( "Unknown objects referenced by remote refs: %s" % ( ", ".join( f"{name.decode()}: {hashutil.hash_to_hex(obj)}" for name, obj in unknown_objects.items() ) ) ) utils.warn_dangling_branches( branches, dangling_branches, logger, self.origin.url ) self.snapshot = Snapshot(branches=branches) return self.snapshot def load_status(self) -> Dict[str, Any]: """The load was eventful if the current snapshot is different to the one we retrieved at the beginning of the run""" eventful = False if self.prev_snapshot and self.snapshot: eventful = self.snapshot.id != self.prev_snapshot.id elif self.snapshot: eventful = bool(self.snapshot.branches) return {"status": ("eventful" if eventful else "uneventful")} if __name__ == "__main__": import click logging.basicConfig( level=logging.DEBUG, format="%(asctime)s %(process)d %(message)s" ) from deprecated import deprecated @deprecated(version="1.1", reason="Use `swh loader run git --help` instead") @click.command() @click.option("--origin-url", help="Origin url", required=True) @click.option("--base-url", default=None, help="Optional Base url") @click.option( "--ignore-history/--no-ignore-history", help="Ignore the repository history", default=False, ) def main(origin_url: str, incremental: bool) -> Dict[str, Any]: from swh.storage import get_storage storage = get_storage(cls="memory") loader = GitLoader( storage, origin_url, incremental=incremental, ) return loader.load() main() diff --git a/swh/loader/git/tests/test_loader.py b/swh/loader/git/tests/test_loader.py index 62f7131..45ee3e9 100644 --- a/swh/loader/git/tests/test_loader.py +++ b/swh/loader/git/tests/test_loader.py @@ -1,756 +1,813 @@ # Copyright (C) 2018-2022 The Software Heritage developers # See the AUTHORS file at the top-level directory of this distribution # License: GNU General Public License version 3, or any later version # See top-level LICENSE file for more information import datetime from functools import partial from http.server import HTTPServer, SimpleHTTPRequestHandler import os import subprocess import sys from tempfile import SpooledTemporaryFile from threading import Thread from unittest.mock import MagicMock, call from dulwich.errors import GitProtocolError, NotGitRepository, ObjectFormatException from dulwich.porcelain import push import dulwich.repo import pytest from swh.loader.git import converters, dumb from swh.loader.git.loader import GitLoader from swh.loader.git.tests.test_from_disk import SNAPSHOT1, FullGitLoaderTests from swh.loader.tests import ( assert_last_visit_matches, get_stats, prepare_repository_from_archive, ) -from swh.model.model import Origin, OriginVisit, OriginVisitStatus, Snapshot +from swh.model.model import ( + Origin, + OriginVisit, + OriginVisitStatus, + Snapshot, + SnapshotBranch, + TargetType, +) class CommonGitLoaderNotFound: @pytest.fixture(autouse=True) def __inject_fixtures(self, mocker): """Inject required fixtures in unittest.TestCase class""" self.mocker = mocker @pytest.mark.parametrize( "failure_exception", [ GitProtocolError("Repository unavailable"), # e.g DMCA takedown GitProtocolError("Repository not found"), GitProtocolError("unexpected http resp 401"), NotGitRepository("not a git repo"), ], ) def test_load_visit_not_found(self, failure_exception): """Ingesting an unknown url result in a visit with not_found status""" # simulate an initial communication error (e.g no repository found, ...) mock = self.mocker.patch( "swh.loader.git.loader.GitLoader.fetch_pack_from_origin" ) mock.side_effect = failure_exception res = self.loader.load() assert res == {"status": "uneventful"} assert_last_visit_matches( self.loader.storage, self.repo_url, status="not_found", type="git", snapshot=None, ) @pytest.mark.parametrize( "failure_exception", [ IOError, ObjectFormatException, OSError, ValueError, GitProtocolError, ], ) def test_load_visit_failure(self, failure_exception): """Failing during the fetch pack step result in failing visit""" # simulate a fetch communication error after the initial connection # server error (e.g IOError, ObjectFormatException, ...) mock = self.mocker.patch( "swh.loader.git.loader.GitLoader.fetch_pack_from_origin" ) mock.side_effect = failure_exception("failure") res = self.loader.load() assert res == {"status": "failed"} assert_last_visit_matches( self.loader.storage, self.repo_url, status="failed", type="git", snapshot=None, ) class TestGitLoader(FullGitLoaderTests, CommonGitLoaderNotFound): """Prepare a git directory repository to be loaded through a GitLoader. This tests all git loader scenario. """ @pytest.fixture(autouse=True) def init(self, swh_storage, datadir, tmp_path): archive_name = "testrepo" archive_path = os.path.join(datadir, f"{archive_name}.tgz") tmp_path = str(tmp_path) self.repo_url = prepare_repository_from_archive( archive_path, archive_name, tmp_path=tmp_path ) self.destination_path = os.path.join(tmp_path, archive_name) self.loader = GitLoader(swh_storage, self.repo_url) self.repo = dulwich.repo.Repo(self.destination_path) def test_metrics(self, mocker): statsd_report = mocker.patch.object(self.loader.statsd, "_report") res = self.loader.load() assert res == {"status": "eventful"} # TODO: assert "incremental" is added to constant tags before these # metrics are sent statsd_calls = statsd_report.mock_calls assert [c for c in statsd_calls if c[1][0].startswith("git_")] == [ call("git_total", "c", 1, {}, 1), call("git_ignored_refs_percent", "h", 0.0, {}, 1), call("git_known_refs_percent", "h", 0.0, {}, 1), ] total_sum_name = "filtered_objects_total_sum" total_count_name = "filtered_objects_total_count" percent_name = "filtered_objects_percent" assert [c for c in statsd_calls if c[1][0].startswith("filtered_")] == [ call(percent_name, "h", 0.0, {"object_type": "content"}, 1), call(total_sum_name, "c", 0, {"object_type": "content"}, 1), call(total_count_name, "c", 4, {"object_type": "content"}, 1), call(percent_name, "h", 0.0, {"object_type": "directory"}, 1), call(total_sum_name, "c", 0, {"object_type": "directory"}, 1), call(total_count_name, "c", 7, {"object_type": "directory"}, 1), call(percent_name, "h", 0.0, {"object_type": "revision"}, 1), call(total_sum_name, "c", 0, {"object_type": "revision"}, 1), call(total_count_name, "c", 7, {"object_type": "revision"}, 1), call(percent_name, "h", 0.0, {"object_type": "snapshot"}, 1), call(total_sum_name, "c", 0, {"object_type": "snapshot"}, 1), call(total_count_name, "c", 1, {"object_type": "snapshot"}, 1), ] assert self.loader.statsd.constant_tags == { "visit_type": "git", "incremental_enabled": True, "has_parent_snapshot": False, "has_previous_snapshot": False, "has_parent_origins": False, } def test_metrics_filtered(self, mocker): """Tests that presence of some objects in the storage (but not referenced by a snapshot) is reported""" known_revs = [ converters.dulwich_commit_to_revision(self.repo[sha1]) for sha1 in [ b"b6f40292c4e94a8f7e7b4aff50e6c7429ab98e2a", b"1135e94ccf73b5f9bd6ef07b3fa2c5cc60bba69b", ] ] known_dirs = [ converters.dulwich_tree_to_directory(self.repo[sha1]) for sha1 in [ b"fbf70528223d263661b5ad4b80f26caf3860eb8e", b"9ca0c7d6ffa3f9f0de59fd7912e08f11308a1338", b"5df34ec74d6f69072d9a0a6677d8efbed9b12e60", ] ] known_cnts = [ converters.dulwich_blob_to_content(self.repo[sha1]) for sha1 in [ b"534d61ecee4f6da4d6ca6ddd8abf258208d2d1bc", ] ] self.loader.storage.revision_add(known_revs) self.loader.storage.directory_add(known_dirs) self.loader.storage.content_add(known_cnts) self.loader.storage.flush() statsd_report = mocker.patch.object(self.loader.statsd, "_report") res = self.loader.load() assert res == {"status": "eventful"} # TODO: assert "incremental" is added to constant tags before these # metrics are sent statsd_calls = statsd_report.mock_calls assert [c for c in statsd_calls if c[1][0].startswith("git_")] == [ call("git_total", "c", 1, {}, 1), call("git_ignored_refs_percent", "h", 0.0, {}, 1), call("git_known_refs_percent", "h", 0.0, {}, 1), ] total_sum_name = "filtered_objects_total_sum" total_count_name = "filtered_objects_total_count" percent_name = "filtered_objects_percent" assert [c for c in statsd_calls if c[1][0].startswith("filtered_")] == [ call(percent_name, "h", 1 / 4, {"object_type": "content"}, 1), call(total_sum_name, "c", 1, {"object_type": "content"}, 1), call(total_count_name, "c", 4, {"object_type": "content"}, 1), call(percent_name, "h", 3 / 7, {"object_type": "directory"}, 1), call(total_sum_name, "c", 3, {"object_type": "directory"}, 1), call(total_count_name, "c", 7, {"object_type": "directory"}, 1), call(percent_name, "h", 2 / 7, {"object_type": "revision"}, 1), call(total_sum_name, "c", 2, {"object_type": "revision"}, 1), call(total_count_name, "c", 7, {"object_type": "revision"}, 1), call(percent_name, "h", 0.0, {"object_type": "snapshot"}, 1), call(total_sum_name, "c", 0, {"object_type": "snapshot"}, 1), call(total_count_name, "c", 1, {"object_type": "snapshot"}, 1), ] assert self.loader.statsd.constant_tags == { "visit_type": "git", "incremental_enabled": True, "has_parent_snapshot": False, "has_previous_snapshot": False, "has_parent_origins": False, } class TestGitLoader2(FullGitLoaderTests, CommonGitLoaderNotFound): """Mostly the same loading scenario but with a ``parent_origin`` different from the ``origin``; as if the ``origin`` was a forge-fork of ``parent_origin``, detected by the metadata loader. To walk slightly different paths, the end result should stay the same. """ @pytest.fixture(autouse=True) def init(self, swh_storage, datadir, tmp_path, mocker): archive_name = "testrepo" archive_path = os.path.join(datadir, f"{archive_name}.tgz") tmp_path = str(tmp_path) self.repo_url = prepare_repository_from_archive( archive_path, archive_name, tmp_path=tmp_path ) self.destination_path = os.path.join(tmp_path, archive_name) self.fetcher = MagicMock() self.fetcher.get_origin_metadata.return_value = [] self.fetcher.get_parent_origins.return_value = [ Origin(url=f"base://{self.repo_url}") ] self.fetcher_cls = MagicMock(return_value=self.fetcher) self.fetcher_cls.SUPPORTED_LISTERS = ["fake-lister"] mocker.patch( "swh.loader.core.metadata_fetchers._fetchers", return_value=[self.fetcher_cls], ) self.loader = GitLoader( MagicMock(wraps=swh_storage), self.repo_url, lister_name="fake-lister", lister_instance_name="", ) self.repo = dulwich.repo.Repo(self.destination_path) def test_no_previous_snapshot(self, mocker): statsd_report = mocker.patch.object(self.loader.statsd, "_report") res = self.loader.load() assert res == {"status": "eventful"} self.fetcher_cls.assert_called_once_with( credentials={}, lister_name="fake-lister", lister_instance_name="", origin=Origin(url=self.repo_url), ) self.fetcher.get_parent_origins.assert_called_once_with() # First tries the same origin assert self.loader.storage.origin_visit_get_latest.mock_calls == [ call( self.repo_url, allowed_statuses=None, require_snapshot=True, type=None, ), # As it does not already have a snapshot, fall back to the parent origin call( f"base://{self.repo_url}", allowed_statuses=None, require_snapshot=True, type=None, ), ] # TODO: assert "incremental" is added to constant tags before these # metrics are sent assert [c for c in statsd_report.mock_calls if c[1][0].startswith("git_")] == [ call("git_total", "c", 1, {}, 1), call("git_ignored_refs_percent", "h", 0.0, {}, 1), call("git_known_refs_percent", "h", 0.0, {}, 1), ] assert self.loader.statsd.constant_tags == { "visit_type": "git", "incremental_enabled": True, "has_parent_snapshot": False, "has_previous_snapshot": False, "has_parent_origins": True, } def test_load_incremental(self, mocker): statsd_report = mocker.patch.object(self.loader.statsd, "_report") snapshot_id = b"\x01" * 20 now = datetime.datetime.now(tz=datetime.timezone.utc) def ovgl(origin_url, allowed_statuses, require_snapshot, type): if origin_url == f"base://{self.repo_url}": return OriginVisit(origin=origin_url, visit=42, date=now, type="git") else: return None self.loader.storage.origin_visit_get_latest.side_effect = ovgl self.loader.storage.origin_visit_status_get_latest.return_value = ( OriginVisitStatus( origin=f"base://{self.repo_url}", visit=42, snapshot=snapshot_id, date=now, status="full", ) ) self.loader.storage.snapshot_get_branches.return_value = { "id": snapshot_id, "branches": { b"refs/heads/master": SNAPSHOT1.branches[b"refs/heads/master"] }, "next_branch": None, } res = self.loader.load() assert res == {"status": "eventful"} self.fetcher_cls.assert_called_once_with( credentials={}, lister_name="fake-lister", lister_instance_name="", origin=Origin(url=self.repo_url), ) self.fetcher.get_parent_origins.assert_called_once_with() # First tries the same origin assert self.loader.storage.origin_visit_get_latest.mock_calls == [ call( self.repo_url, allowed_statuses=None, require_snapshot=True, type=None, ), # As it does not already have a snapshot, fall back to the parent origin call( f"base://{self.repo_url}", allowed_statuses=None, require_snapshot=True, type=None, ), ] # TODO: assert "incremental*" is added to constant tags before these # metrics are sent assert [c for c in statsd_report.mock_calls if c[1][0].startswith("git_")] == [ call("git_total", "c", 1, {}, 1), call("git_ignored_refs_percent", "h", 0.0, {}, 1), call("git_known_refs_percent", "h", 0.25, {}, 1), ] assert self.loader.statsd.constant_tags == { "visit_type": "git", "incremental_enabled": True, "has_parent_snapshot": True, "has_previous_snapshot": False, "has_parent_origins": True, } self.fetcher.reset_mock() self.fetcher_cls.reset_mock() if sys.version_info >= (3, 9, 0): self.loader.storage.reset_mock(return_value=True, side_effect=True) else: # Reimplement https://github.com/python/cpython/commit/aef7dc89879d099dc704bd8037b8a7686fb72838 # noqa # for old Python versions: def reset_mock(m): m.reset_mock(return_value=True, side_effect=True) for child in m._mock_children.values(): reset_mock(child) reset_mock(self.loader.storage) statsd_report.reset_mock() # Load again res = self.loader.load() assert res == {"status": "uneventful"} self.fetcher_cls.assert_called_once_with( credentials={}, lister_name="fake-lister", lister_instance_name="", origin=Origin(url=self.repo_url), ) self.fetcher.get_parent_origins.assert_not_called() assert self.loader.storage.origin_visit_get_latest.mock_calls == [ # Tries the same origin, and finds a snapshot call( self.repo_url, type=None, allowed_statuses=None, require_snapshot=True, ), # also fetches the parent, in case the origin was rebased on the parent # since the last visit call( f"base://{self.repo_url}", type=None, allowed_statuses=None, require_snapshot=True, ), ] # TODO: assert "incremental*" is added to constant tags before these # metrics are sent assert [c for c in statsd_report.mock_calls if c[1][0].startswith("git_")] == [ call("git_total", "c", 1, {}, 1), call("git_ignored_refs_percent", "h", 0.0, {}, 1), call("git_known_refs_percent", "h", 1.0, {}, 1), ] assert self.loader.statsd.constant_tags == { "visit_type": "git", "incremental_enabled": True, "has_parent_snapshot": False, # Because we reset the mock since last time "has_previous_snapshot": True, "has_parent_origins": True, } @pytest.mark.parametrize( "parent_snapshot,previous_snapshot,expected_git_known_refs_percent", [ pytest.param( Snapshot( branches={ b"refs/heads/master": SNAPSHOT1.branches[b"refs/heads/master"] } ), Snapshot(branches={}), 0.25, id="partial-parent-and-empty-previous", ), pytest.param( SNAPSHOT1, Snapshot( branches={ b"refs/heads/master": SNAPSHOT1.branches[b"refs/heads/master"] } ), 1.0, id="full-parent-and-partial-previous", ), ], ) def test_load_incremental_from( self, parent_snapshot, previous_snapshot, expected_git_known_refs_percent, mocker, ): """Snapshot of parent origin has all branches, but previous snapshot was empty.""" statsd_report = mocker.patch.object(self.loader.statsd, "_report") now = datetime.datetime.now(tz=datetime.timezone.utc) self.loader.storage.snapshot_add([parent_snapshot, previous_snapshot]) self.loader.storage.origin_add( [Origin(url=f"base://{self.repo_url}"), Origin(url=self.repo_url)] ) self.loader.storage.origin_visit_add( [ OriginVisit( origin=f"base://{self.repo_url}", visit=42, date=now - datetime.timedelta(seconds=-1), type="git", ), OriginVisit( origin=self.repo_url, visit=42, date=now - datetime.timedelta(seconds=-1), type="git", ), ] ) self.loader.storage.origin_visit_status_add( [ OriginVisitStatus( origin=f"base://{self.repo_url}", visit=42, type="git", snapshot=parent_snapshot.id, date=now, status="full", ), OriginVisitStatus( origin=self.repo_url, visit=42, type="git", snapshot=previous_snapshot.id, date=now, status="full", ), ] ) self.loader.storage.flush() res = self.loader.load() assert res == {"status": "eventful"} self.fetcher_cls.assert_called_once_with( credentials={}, lister_name="fake-lister", lister_instance_name="", origin=Origin(url=self.repo_url), ) self.fetcher.get_parent_origins.assert_called_once_with() # First tries the same origin assert self.loader.storage.origin_visit_get_latest.mock_calls == [ call( self.repo_url, allowed_statuses=None, require_snapshot=True, type=None, ), # As it does not already have a snapshot, fall back to the parent origin call( f"base://{self.repo_url}", allowed_statuses=None, require_snapshot=True, type=None, ), ] assert self.loader.statsd.constant_tags == { "visit_type": "git", "incremental_enabled": True, "has_parent_snapshot": True, "has_previous_snapshot": True, "has_parent_origins": True, } assert [c for c in statsd_report.mock_calls if c[1][0].startswith("git_")] == [ call("git_total", "c", 1, {}, 1), call("git_ignored_refs_percent", "h", 0.0, {}, 1), call("git_known_refs_percent", "h", expected_git_known_refs_percent, {}, 1), ] + def test_load_incremental_negotiation(self): + """Check that the packfile negotiated when running an incremental load only + contains the "new" commits, and not all objects.""" + + snapshot_id = b"\x01" * 20 + now = datetime.datetime.now(tz=datetime.timezone.utc) + + def ovgl(origin_url, allowed_statuses, require_snapshot, type): + if origin_url == f"base://{self.repo_url}": + return OriginVisit(origin=origin_url, visit=42, date=now, type="git") + else: + return None + + self.loader.storage.origin_visit_get_latest.side_effect = ovgl + self.loader.storage.origin_visit_status_get_latest.return_value = ( + OriginVisitStatus( + origin=f"base://{self.repo_url}", + visit=42, + snapshot=snapshot_id, + date=now, + status="full", + ) + ) + self.loader.storage.snapshot_get_branches.return_value = { + "id": snapshot_id, + "branches": { + b"refs/heads/master": SnapshotBranch( + # id of the initial commit in the git repository fixture + target=bytes.fromhex("b6f40292c4e94a8f7e7b4aff50e6c7429ab98e2a"), + target_type=TargetType.REVISION, + ), + }, + "next_branch": None, + } + + res = self.loader.load() + assert res == {"status": "eventful"} + + stats = get_stats(self.loader.storage) + assert stats == { + "content": 3, # instead of 4 for the full repository + "directory": 6, # instead of 7 + "origin": 1, + "origin_visit": 1, + "release": 0, + "revision": 6, # instead of 7 + "skipped_content": 0, + "snapshot": 1, + } + class DumbGitLoaderTestBase(FullGitLoaderTests): """Prepare a git repository to be loaded using the HTTP dumb transfer protocol.""" @pytest.fixture(autouse=True) def init(self, swh_storage, datadir, tmp_path): # remove any proxy settings in order to successfully spawn a local HTTP server http_proxy = os.environ.get("http_proxy") https_proxy = os.environ.get("https_proxy") if http_proxy: del os.environ["http_proxy"] if http_proxy: del os.environ["https_proxy"] # prepare test base repository using smart transfer protocol archive_name = "testrepo" archive_path = os.path.join(datadir, f"{archive_name}.tgz") tmp_path = str(tmp_path) base_repo_url = prepare_repository_from_archive( archive_path, archive_name, tmp_path=tmp_path ) destination_path = os.path.join(tmp_path, archive_name) self.destination_path = destination_path with_pack_files = self.with_pack_files if with_pack_files: # create a bare clone of that repository in another folder, # all objects will be contained in one or two pack files in that case http_root_dir = tmp_path repo_name = archive_name + "_bare" bare_repo_path = os.path.join(http_root_dir, repo_name) subprocess.run( ["git", "clone", "--bare", base_repo_url, bare_repo_path], check=True, ) else: # otherwise serve objects from the bare repository located in # the .git folder of the base repository http_root_dir = destination_path repo_name = ".git" bare_repo_path = os.path.join(http_root_dir, repo_name) # spawn local HTTP server that will serve the bare repository files hostname = "localhost" handler = partial(SimpleHTTPRequestHandler, directory=http_root_dir) httpd = HTTPServer((hostname, 0), handler, bind_and_activate=True) def serve_forever(httpd): with httpd: httpd.serve_forever() thread = Thread(target=serve_forever, args=(httpd,)) thread.start() repo = dulwich.repo.Repo(self.destination_path) class DumbGitLoaderTest(GitLoader): def load(self): """ Override load method to ensure the bare repository will be synchronized with the base one as tests can modify its content. """ if with_pack_files: # ensure HEAD ref will be the same for both repositories with open(os.path.join(bare_repo_path, "HEAD"), "wb") as fw: with open( os.path.join(destination_path, ".git/HEAD"), "rb" ) as fr: head_ref = fr.read() fw.write(head_ref) # push possibly modified refs in the base repository to the bare one for ref in repo.refs.allkeys(): if ref != b"HEAD" or head_ref in repo.refs: push( repo, remote_location=f"file://{bare_repo_path}", refspecs=ref, ) # generate or update the info/refs file used in dumb protocol subprocess.run( ["git", "-C", bare_repo_path, "update-server-info"], check=True, ) return super().load() # bare repository with dumb protocol only URL self.repo_url = f"http://{httpd.server_name}:{httpd.server_port}/{repo_name}" self.loader = DumbGitLoaderTest(swh_storage, self.repo_url) self.repo = repo yield # shutdown HTTP server httpd.shutdown() thread.join() # restore HTTP proxy settings if any if http_proxy: os.environ["http_proxy"] = http_proxy if https_proxy: os.environ["https_proxy"] = https_proxy @pytest.mark.parametrize( "failure_exception", [AttributeError, NotImplementedError, ValueError] ) def test_load_despite_dulwich_exception(self, mocker, failure_exception): """Checks repository can still be loaded when dulwich raises exception when encountering a repository with dumb transfer protocol. """ fetch_pack_from_origin = mocker.patch( "swh.loader.git.loader.GitLoader.fetch_pack_from_origin" ) fetch_pack_from_origin.side_effect = failure_exception("failure") res = self.loader.load() assert res == {"status": "eventful"} stats = get_stats(self.loader.storage) assert stats == { "content": 4, "directory": 7, "origin": 1, "origin_visit": 1, "release": 0, "revision": 7, "skipped_content": 0, "snapshot": 1, } def test_load_empty_repository(self, mocker): class GitObjectsFetcherNoRefs(dumb.GitObjectsFetcher): def _get_refs(self): return {} mocker.patch.object(dumb, "GitObjectsFetcher", GitObjectsFetcherNoRefs) res = self.loader.load() assert res == {"status": "uneventful"} stats = get_stats(self.loader.storage) assert stats == { "content": 0, "directory": 0, "origin": 1, "origin_visit": 1, "release": 0, "revision": 0, "skipped_content": 0, "snapshot": 1, } class TestDumbGitLoaderWithPack(DumbGitLoaderTestBase): @classmethod def setup_class(cls): cls.with_pack_files = True def test_load_with_missing_pack(self, mocker): """Some dumb git servers might reference a no longer existing pack file while it is possible to load a repository without it. """ class GitObjectsFetcherMissingPack(dumb.GitObjectsFetcher): def _http_get(self, path: str) -> SpooledTemporaryFile: buffer = super()._http_get(path) if path == "objects/info/packs": # prepend a non existing pack to the returned packs list packs = buffer.read().decode("utf-8") buffer.seek(0) buffer.write( ( "P pack-a70762ba1a901af3a0e76de02fc3a99226842745.pack\n" + packs ).encode() ) buffer.flush() buffer.seek(0) return buffer mocker.patch.object(dumb, "GitObjectsFetcher", GitObjectsFetcherMissingPack) res = self.loader.load() assert res == {"status": "eventful"} class TestDumbGitLoaderWithoutPack(DumbGitLoaderTestBase): @classmethod def setup_class(cls): cls.with_pack_files = False