diff --git a/swh/loader/mercurial/bundle20_reader.py b/swh/loader/mercurial/bundle20_reader.py index c6952ec..10b96a3 100644 --- a/swh/loader/mercurial/bundle20_reader.py +++ b/swh/loader/mercurial/bundle20_reader.py @@ -1,624 +1,624 @@ # Copyright (C) 2017 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 """This document contains code for extracting all of the data from Mercurial version 2 bundle file. It is referenced by bundle20_loader.py """ # ============================================================================= # ============================================================================= # BACKGROUND # ============================================================================= # ============================================================================= # # https://www.mercurial-scm.org/wiki/BundleFormat says: # "The new bundle format design is described on the BundleFormat2 page." # # https://www.mercurial-scm.org/wiki/BundleFormat2#Format_of_the_Bundle2_Container says: # noqa # "The latest description of the binary format can be found as comment in the # Mercurial source code." # # https://www.mercurial-scm.org/repo/hg/file/default/mercurial/help/internals/bundles.txt says: # noqa # "The 'HG20' format is not yet documented here. See the inline comments in # 'mercurial/exchange.py' for now." # # ----------------------------------------------------------------------------- # Avi says: # ----------------------------------------------------------------------------- # # All of the above official(?) statements seem to be quite wrong. # # The mercurial-scm wiki is a cluster#@*& of missing pages, bad links, wrong # information, obsolete information, undecipherable names, and half-started # leavings that only sort of look like content. I don't understand who or what # it's there for. I think that means it's not there for me? # # https://www.mercurial-scm.org/wiki/BundleFormat2#New_header is wrong and # bizarre, and there isn't any other information on the page. # # # https://www.mercurial-scm.org/repo/hg/file/de86a6872d06/mercurial/help/internals/changegroups.txt # noqa # (`hg help internals.changegroups`) is very close to what we need. # It is accurate, current, and thorough. # It describes much of the internal structure, which is super helpful if you # know in advance which info can be trusted, but it doesn't describe any of the # file-level details, including the file headers and that the entire bundle # is broken into overlaid 4KB chunks starting from just after the bundle # header, nor does it describe what any of the component elements are used for, # nor does it explain the meta-message segment in the blob deltas, nor does it # explain the file flags occasionally appended to manifest file hashes. Also it # says: "The [delta data] format is described more fully in 'hg help # internals.bdiff'", which is also wrong. As far as I can tell, that # file has never existed. # # It does however have one potentially extremely useful note buried in the # middle that, in hindsight, could have significant implications for complexity # and performance in future Mercurial loading work. # # It says: "In version 1, the delta is always applied against the previous node # from the changegroup or the first parent if this is the first entry in the # changegroup." # # If the next version of HG support for SWH can reliably get version 1 data, # then it could be implemented entirely without worrying about ballooning # memory utilization, which would shrink the code significantly and probably be # faster too. So maybe HG10 bundles instead of HG20 bundles are superior for # this task? But then I read that servers can optionally disable serving # version 1 content, and I like to think that this loader could eventually # be applied directly to a network stream without an intermediate phase for # cloning and local bundling, so...It seemed like a good idea at the time? # # ----------------------------------------------------------------------------- # Other notes and thoughts: # ----------------------------------------------------------------------------- # 1) # This is a relatively minor detail, but # Mercurial nodes are not content-addressable like Git's are. # # https://www.mercurial-scm.org/wiki/Nodeid explains: "If you modify a file, # commit the change, and then modify it to restore the original contents, the # contents are the same but the history is different, so the file will get a # new nodeid. This history-sensitivity is obtained by calculating the nodeid # from the concatenation of the parent nodeids with the file's contents..." # # The result is that we always have to collect and hash everything at least # once in order to know if we've seen something like it before, because nothing # tells us that the node we're looking at is unique. We can use node ids for # linking disparate elements together (e.g. commit to manifest) but not for # determining whether two elements in the same group are identical in all but # descendency. So there's no way to save time on duplicate hashing. Well... # there is the copied_file blob metadata, but, lol. # # 2) # Most of the code complexity is due to dealing with 'version 2' changegroups, # for which we need to keep track of the entire history of all updates made # to a given file or working directory tree structure, because a revision # delta could be applied over any of the prior revisions all the way back to # rev 0, according to whenever files were branched/merged/uncommitted/etc. For # very large repositories with a lot of churn, this can quickly expand to # require multiple gigabytes of space, possibly exceeding RAM availability if # one desires to keep as much data resident in memory as possible to boost # performance. mozilla-unified, for instance, produces some 2 million+ blobs # (1.6 million+ unique). Nested umpteen subdirectory levels deep, those blobs # balloon into a quantity of directory subtrees that rapidly exceeds an 8GB RAM # laptop's ability to keep them all active without a good amount of care and # pruning. The code here tries to strike a balance between memory utilization # and performance. # # This problem is also referenced in the last paragraph of the previous # section, where potentially this problem wouldn't exist for 'version 1' data # if we can reliably get it. Can we? Either that or not use bundles at all, # which has other costs. # # 3) # If the list of changed files stored by the changesets had indicated which # of those changed files were added or modified and which ones were removed, # this code could be much faster. Right now we have to perform a huge number of # substring replacements (see the apply_revdata method) to produce a complete # file manifest for each commit (as a string!!!) in order to know how to get # the set of removed files from the next delta. We can intuit from every # manifest delta which files were modified or added, but I believe there's no # way to intuit which files were removed without actually having the complete # prior state and without the list of removals being explicitly given. If you # have an explicit list of all the files that were removed for any given commit # changegroup, and you combine that with the delta updates in the manifest # changegroups which detail the set of files that have been added or modified, # then you wouldn't even have to apply any of the string deltas to get a # complete understanding of the set of differences between one manifest and the # next. Not having this effective speed boost is rather unfortunate; it would # require only one extra stored byte per commit to differentiate removals and # would make extracting bundles lightning fast. # ============================================================================ ## import itertools import struct from binascii import unhexlify from collections import OrderedDict from datetime import datetime from .chunked_reader import ChunkedFileReader from .objects import SelectiveCache def unpack(fmt_str, source): """Utility function for fetching the right number of bytes from a stream to satisfy a struct.unpack pattern. args: fmt_str: a struct.unpack string pattern (e.g. '>I' for 4 bytes big-endian) source: any IO object that has a read() method which returns an appropriate sequence of bytes """ ret = struct.unpack(fmt_str, source.read(struct.calcsize(fmt_str))) if len(ret) == 1: return ret[0] return ret class Bundle20Reader(object): """Parser for extracting data from Mercurial Bundle20 files. NOTE: Currently only works on uncompressed HG20 bundles, but checking for COMPRESSION=<2chars> and loading the appropriate stream decompressor at that point would be trivial to add if necessary. args: bundlefile (str): name of the binary repository bundle file cache_filename (str): path to the disk cache used (transited to the SelectiveCache instance) cache_size (int): tuning parameter for the upper RAM limit used by historical data caches. The default is defined in the SelectiveCache class. """ NAUGHT_NODE = b'\x00' * 20 def __init__(self, bundlefile, cache_filename, cache_size=None): self.bundlefile = bundlefile self.cache_filename = cache_filename bfile = open(bundlefile, 'rb', buffering=200*1024*1024) btype = bfile.read(4) # 'HG20' if btype != b'HG20': raise Exception(bundlefile, b'Not an HG20 bundle. First 4 bytes:' + btype) bfile.read(4) # '\x00\x00\x00\x00' self.params = self.read_bundle_header(bfile) # print('PARAMETERS', self.params) self.num_commits = self.params[b'nbchanges'] self.filereader = ChunkedFileReader(bfile) self.cache_size = cache_size self.blobs_offset = None self.changes_offset = self.filereader.tell() self.changes_next_offset = None self.manifests_offset = None self.manifests_next_offset = None self.id_to_info = {} def read_bundle_header(self, bfile): """Parse the file header which describes the format and parameters. See the structure diagram at the top of the file for more insight. args: bfile: bundle file handle with the cursor at the start offset of the content header (the 9th byte in the file) returns: dict of decoded bundle parameters """ unpack('>I', bfile) # header length chg_len = unpack('>B', bfile) # len('CHANGEGROUP') == 11 bfile.read(chg_len) # should say 'CHANGEGROUP' unpack('>I', bfile) # probably \x00\x00\x00\x00 n_mandatory, n_advisory = unpack('>BB', bfile) # parameter counts mandatory_params = [ (key_len, val_len) for key_len, val_len in [unpack('>BB', bfile) for i in range(n_mandatory)] ] advisory_params = [ (key_len, val_len) for key_len, val_len in [unpack('>BB', bfile) for i in range(n_advisory)] ] params = {} for key_len, val_len in mandatory_params+advisory_params: key = unpack('>%ds' % key_len, bfile) val = int(unpack('>%ds' % val_len, bfile)) params[key] = val return params def revdata_iterator(self, bytes_to_read): """A chunk's revdata section is a series of start/end/length/data_delta content updates called RevDiffs that indicate components of a text diff applied to the node's basenode. The sum length of all diffs is the length indicated at the beginning of the chunk at the start of the header. See the structure diagram at the top of the file for more insight. args: bytes_to_read: int total number of bytes in the chunk's revdata yields: (int, int, read iterator) representing a single text diff component """ while bytes_to_read > 0: start_offset = unpack('>I', self.filereader) end_offset = unpack('>I', self.filereader) blocklen = unpack('>I', self.filereader) delta_it = self.filereader.read_iterator(blocklen) bytes_to_read -= (12 + blocklen) yield (start_offset, end_offset, delta_it) # RevDiff def read_chunk_header(self): """The header of a RevChunk describes the id ('node') for the current change, the commit id ('linknode') associated with this change, the parental heritage ('p1' and 'p2'), and the node to which the revdata updates will apply ('basenode'). 'linknode' is the same as 'node' when reading the commit log because any commit is already itself. 'basenode' for a changeset will be NAUGHT_NODE, because changeset chunks include complete information and not diffs. See the structure diagram at the top of the file for more insight. returns: dict of the next delta header """ header = self.filereader.read(100) header = { 'node': header[0:20], 'p1': header[20:40], 'p2': header[40:60], 'basenode': header[60:80], 'linknode': header[80:100] } return header def read_revchunk(self): """Fetch a complete RevChunk. A RevChunk contains the collection of line changes made in a particular update. header['node'] identifies which update. Commits, manifests, and files all have these. Each chunk contains an indicator of the whole chunk size, an update header, and then the body of the update as a series of text diff components. See the structure diagram at the top of the file for more insight. returns: tuple(dict, iterator) of (header, chunk data) if there is another chunk in the group, else None """ size = unpack('>I', self.filereader) - 104 if size >= 0: header = self.read_chunk_header() return (header, self.revdata_iterator(size)) else: return None # NullChunk def extract_commit_metadata(self, data): """Converts the binary commit metadata format into a dict. args: data: bytestring of encoded commit information returns: dict of decoded commit information """ parts, message = data.split(b'\n\n', 1) parts = parts.split(b'\n') commit = {} commit['message'] = message commit['manifest'] = unhexlify(parts[0]) commit['user'] = parts[1] tstamp, tz, *extra = parts[2].split(b' ') commit['time'] = datetime.fromtimestamp(float(tstamp)) commit['time_offset_seconds'] = int(tz) if extra: - commit['extra'] = extra[0] + commit['extra'] = b' '.join(extra) commit['changed_files'] = parts[3:] return commit def skip_sections(self, num_sections=1): """Skip past sections quickly. args: num_sections: int number of sections to skip """ for i in range(num_sections): size = unpack('>I', self.filereader) while size >= 104: self.filereader.seek(size - 4, from_current=True) size = unpack('>I', self.filereader) def apply_revdata(self, revdata_it, prev_state): """Compose the complete text body for a change from component deltas. args: revdata_it: output from the revdata_iterator method prev_state: bytestring the base complete text on which the new deltas will be applied returns: (bytestring, list, list) the new complete string and lists of added and removed components (used in manifest processing) """ state = [] added = [] removed = [] next_start = 0 for delta_start, delta_end, rev_diff_it in revdata_it: removed.append(prev_state[delta_start:delta_end]) added.append(b''.join(rev_diff_it)) state.append(prev_state[next_start:delta_start]) state.append(added[-1]) next_start = delta_end state.append(prev_state[next_start:]) state = b''.join(state) return (state, added, removed) def skim_headers(self): """Get all header data from a change group but bypass processing of the contained delta components. yields: output of read_chunk_header method for all chunks in the group """ size = unpack('>I', self.filereader) - 104 while size >= 0: header = self.read_chunk_header() self.filereader.seek(size, from_current=True) yield header size = unpack('>I', self.filereader) - 104 def group_iterator(self): """Bundle sections are called groups. These are composed of one or more revision chunks of delta components. Iterate over all the chunks in a group and hand each one back. yields: see output of read_revchunk method """ revchunk = self.read_revchunk() while revchunk: # A group is terminated by a NullChunk yield revchunk # (header, revdata_iterator) revchunk = self.read_revchunk() def yield_group_objects(self, cache_hints=None, group_offset=None): """Bundles are sectioned into groups: the log of all commits, the log of all manifest changes, and a series of logs of blob changes (one for each file). All groups are structured the same way, as a series of revisions each with a series of delta components. Iterate over the current group and return the completed object data for the current update by applying all of the internal delta components to each prior revision. args: cache_hints: see build_cache_hints (this will be built automatically if not pre-built and passed in) group_offset: int file position of the start of the desired group yields: (dict, bytestring, list, list) the output from read_chunk_header followed by the output from apply_revdata """ if group_offset is not None: self.filereader.seek(group_offset) if cache_hints is None: cache_hints = self.build_cache_hints() data_cache = SelectiveCache(max_size=self.cache_size, cache_hints=cache_hints, filename=self.cache_filename) # Loop over all revisions in the group data = b'' for header, revdata_it in self.group_iterator(): node = header['node'] basenode = header['basenode'] data = data_cache.fetch(basenode) or b'' data, added, removed = self.apply_revdata(revdata_it, data) data_cache.store(node, data) yield (header, data, added, removed) # each RevChunk def extract_meta_from_blob(self, data): """File revision data sometimes begins with a metadata section of dubious value. Strip it off and maybe decode it. It seems to be mostly useless. Why indicate that a file node is a copy of another node? You can already get that information from the delta header. args: data: bytestring of one revision of a file, possibly with metadata embedded at the start returns: (bytestring, dict) of (the blob data, the meta information) """ meta = {} if data.startswith(b'\x01\n'): empty, metainfo, data = data.split(b'\x01\n', 2) metainfo = b'\x01\n' + metainfo + b'\x01\n' if metainfo.startswith(b'copy:'): # direct file copy (?) copyinfo = metainfo.split(b'\n') meta['copied_file'] = copyinfo[0][6:] meta['copied_rev'] = copyinfo[1][9:] elif metainfo.startswith(b'censored:'): # censored revision deltas must be full-replacements (?) meta['censored'] = metainfo else: # no idea meta['text'] = metainfo return data, meta def seek_changelog(self): """Seek to the beginning of the change logs section. """ self.filereader.seek(self.changes_offset) def seek_manifests(self): """Seek to the beginning of the manifests section. """ if self.manifests_offset is None: self.seek_changelog() self.skip_sections(1) # skip past commits self.manifests_offset = self.filereader.tell() else: self.filereader.seek(self.manifests_offset) def seek_filelist(self): """Seek to the beginning of the file changes section. """ if self.blobs_offset is None: self.seek_manifests() self.skip_sections(1) # skip past manifests self.blobs_offset = self.filereader.tell() else: self.filereader.seek(self.blobs_offset) def yield_all_blobs(self): """Gets blob data from the bundle. yields: (bytestring, (bytestring, int, dict)) of (blob data, (file name, start offset of the file within the bundle, node header)) """ self.seek_filelist() # Loop through all files that have commits size = unpack('>I', self.filereader) while size > 0: file_name = self.filereader.read(size-4) file_start_offset = self.filereader.tell() # get all of the blobs for each file for header, data, *_ in self.yield_group_objects(): blob, meta = self.extract_meta_from_blob(data) yield blob, (file_name, file_start_offset, header) size = unpack('>I', self.filereader) def yield_all_changesets(self): """Gets commit data from the bundle. yields: (dict, dict) of (read_chunk_header output, extract_commit_metadata output) """ self.seek_changelog() for header, data, *_ in self.yield_group_objects(): changeset = self.extract_commit_metadata(data) yield (header, changeset) def yield_all_manifest_deltas(self, cache_hints=None): """Gets manifest data from the bundle. In order to process the manifests in a reasonable amount of time, we want to use only the deltas and not the entire manifest at each change, because if we're processing them in sequential order (we are) then we already have the previous state so we only need the changes. args: cache_hints: see build_cache_hints method yields: (dict, dict, dict) of (read_chunk_header output, extract_manifest_elements output on added/modified files, extract_manifest_elements on removed files) """ self.seek_manifests() for header, data, added, removed in self.yield_group_objects( cache_hints=cache_hints ): added = self.extract_manifest_elements(added) removed = self.extract_manifest_elements(removed) yield (header, added, removed) def build_manifest_hints(self): """Just a minor abstraction shortcut for the build_cache_hints method. returns: see build_cache_hints method """ self.seek_manifests() return self.build_cache_hints() def build_cache_hints(self): """The SelectiveCache class that we use in building nodes can accept a set of key counters that makes its memory usage much more efficient. returns: dict of key=a node id, value=the number of times we will need data from that node when building subsequent nodes """ cur_pos = self.filereader.tell() hints = OrderedDict() prev_node = None for header in self.skim_headers(): basenode = header['basenode'] if (basenode != self.NAUGHT_NODE) and (basenode != prev_node): # If the base isn't immediately prior, then cache it once more. hints[basenode] = hints.get(basenode, 0) + 1 prev_node = header['node'] self.filereader.seek(cur_pos) return hints def extract_manifest_elements(self, data): """Parses data that looks like a manifest. In practice we only pass in the bits extracted from the application of a manifest delta describing which files were added/modified or which ones were removed. args: data: either a string or a list of strings that, when joined, embodies the composition of a manifest. This takes the form of repetitions of (without the brackets):: b'\x00[flag]\\n' ...repeat... where ``[flag]`` may or may not be there depending on whether the file is specially flagged as executable or something returns: dict: ``{file_path: (file_node, permissions), ...}`` where permissions is given according to the flag that optionally exists in the data """ elements = {} if isinstance(data, str): data = data.split(b'\n') else: data = itertools.chain.from_iterable( [chunk.split(b'\n') for chunk in data] ) for line in data: if line != b'': f = line.split(b'\x00') node = f[1] flag_bytes = node[40:] elements[f[0]] = ( unhexlify(node[:40]), b'l' in flag_bytes, b'755' if (b'x' in flag_bytes) else b'644' ) return elements diff --git a/swh/loader/mercurial/loader.py b/swh/loader/mercurial/loader.py index ee796b2..835c3ef 100644 --- a/swh/loader/mercurial/loader.py +++ b/swh/loader/mercurial/loader.py @@ -1,531 +1,535 @@ # Copyright (C) 2017-2018 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 """This document contains a SWH loader for ingesting repository data from Mercurial version 2 bundle files. """ # NOTE: The code here does expensive work twice in places because of the # intermediate need to check for what is missing before sending to the database # and the desire to not juggle very large amounts of data. # TODO: Decide whether to also serialize to disk and read back more quickly # from there. Maybe only for very large repos and fast drives. # - Avi import datetime import hglib import os import random import re from dateutil import parser from shutil import rmtree from tempfile import mkdtemp from swh.model import identifiers from swh.model.hashutil import ( MultiHash, hash_to_hex, hash_to_bytes, DEFAULT_ALGORITHMS ) from swh.loader.core.loader import UnbufferedLoader from swh.loader.core.converters import content_for_storage from swh.loader.core.utils import clean_dangling_folders from . import converters from .archive_extract import tmp_extract from .bundle20_reader import Bundle20Reader from .converters import PRIMARY_ALGO as ALGO from .objects import SelectiveCache, SimpleTree TAG_PATTERN = re.compile('[0-9A-Fa-f]{40}') TEMPORARY_DIR_PREFIX_PATTERN = 'swh.loader.mercurial.' HEAD_POINTER_NAME = b'tip' class HgBundle20Loader(UnbufferedLoader): """Mercurial loader able to deal with remote or local repository. """ CONFIG_BASE_FILENAME = 'loader/mercurial' ADDITIONAL_CONFIG = { 'bundle_filename': ('str', 'HG20_none_bundle'), 'reduce_effort': ('bool', False), 'temp_directory': ('str', '/tmp'), 'cache1_size': ('int', 800*1024*1024), 'cache2_size': ('int', 800*1024*1024), } def __init__(self, logging_class='swh.loader.mercurial.Bundle20Loader'): super().__init__(logging_class=logging_class) self.content_max_size_limit = self.config['content_size_limit'] self.bundle_filename = self.config['bundle_filename'] self.reduce_effort_flag = self.config['reduce_effort'] self.empty_repository = None self.temp_directory = self.config['temp_directory'] self.cache1_size = self.config['cache1_size'] self.cache2_size = self.config['cache2_size'] self.working_directory = None self.bundle_path = None def pre_cleanup(self): """Cleanup potential dangling files from prior runs (e.g. OOM killed tasks) """ clean_dangling_folders(self.temp_directory, pattern_check=TEMPORARY_DIR_PREFIX_PATTERN, log=self.log) def cleanup(self): """Clean temporary working directory """ if self.bundle_path and os.path.exists(self.bundle_path): self.log.debug('Cleanup up working bundle %s' % self.bundle_path) os.unlink(self.bundle_path) if self.working_directory and os.path.exists(self.working_directory): self.log.debug('Cleanup up working directory %s' % ( self.working_directory, )) rmtree(self.working_directory) def get_heads(self, repo): """Read the closed branches heads (branch, bookmarks) and returns a dict with key the branch_name (bytes) and values the tuple (pointer nature (bytes), mercurial's node id (bytes)). Those needs conversion to swh-ids. This is taken care of in get_revisions. """ b = {} for _, node_hash_id, pointer_nature, branch_name, *_ in repo.heads(): b[branch_name] = ( pointer_nature, hash_to_bytes(node_hash_id.decode())) bookmarks = repo.bookmarks() if bookmarks and bookmarks[0]: for bookmark_name, _, target_short in bookmarks[0]: target = repo[target_short].node() b[bookmark_name] = (None, hash_to_bytes(target.decode())) return b def prepare_origin_visit(self, *, origin_url, visit_date, **kwargs): self.origin_url = origin_url self.origin = {'url': self.origin_url, 'type': 'hg'} if isinstance(visit_date, str): # visit_date can be string or datetime visit_date = parser.parse(visit_date) self.visit_date = visit_date def prepare(self, *, origin_url, visit_date, directory=None): """Prepare the necessary steps to load an actual remote or local repository. To load a local repository, pass the optional directory parameter as filled with a path to a real local folder. To load a remote repository, pass the optional directory parameter as None. Args: origin_url (str): Origin url to load visit_date (str/datetime): Date of the visit directory (str/None): The local directory to load """ self.branches = {} self.tags = [] self.releases = {} self.node_2_rev = {} if not directory: # remote repository self.working_directory = mkdtemp( prefix=TEMPORARY_DIR_PREFIX_PATTERN, suffix='-%s' % os.getpid(), dir=self.temp_directory) os.makedirs(self.working_directory, exist_ok=True) self.hgdir = self.working_directory self.log.debug('Cloning %s to %s' % ( self.origin['url'], self.hgdir)) hglib.clone(source=self.origin['url'], dest=self.hgdir) else: # local repository self.working_directory = None self.hgdir = directory self.bundle_path = os.path.join(self.hgdir, self.bundle_filename) self.log.debug('Bundling at %s' % self.bundle_path) with hglib.open(self.hgdir) as repo: self.heads = self.get_heads(repo) repo.bundle(bytes(self.bundle_path, 'utf-8'), all=True, type=b'none-v2') self.cache_filename1 = os.path.join( self.hgdir, 'swh-cache-1-%s' % ( hex(random.randint(0, 0xffffff))[2:], )) self.cache_filename2 = os.path.join( self.hgdir, 'swh-cache-2-%s' % ( hex(random.randint(0, 0xffffff))[2:], )) try: self.br = Bundle20Reader(bundlefile=self.bundle_path, cache_filename=self.cache_filename1, cache_size=self.cache1_size) except FileNotFoundError: # Empty repository! Still a successful visit targeting an # empty snapshot self.log.warn('%s is an empty repository!' % self.hgdir) self.empty_repository = True else: self.reduce_effort = set() if self.reduce_effort_flag: now = datetime.datetime.now(tz=datetime.timezone.utc) if (now - self.visit_date).days > 1: # Assuming that self.visit_date would be today for # a new visit, treat older visit dates as # indication of wanting to skip some processing # effort. for header, commit in self.br.yield_all_changesets(): ts = commit['time'].timestamp() if ts < self.visit_date.timestamp(): self.reduce_effort.add(header['node']) def has_contents(self): return not self.empty_repository def has_directories(self): return not self.empty_repository def has_revisions(self): return not self.empty_repository def has_releases(self): return not self.empty_repository def fetch_data(self): """Fetch the data from the data source.""" pass def get_contents(self): """Get the contents that need to be loaded.""" # NOTE: This method generates blobs twice to reduce memory usage # without generating disk writes. self.file_node_to_hash = {} hash_to_info = {} self.num_contents = 0 contents = {} missing_contents = set() for blob, node_info in self.br.yield_all_blobs(): self.num_contents += 1 file_name = node_info[0] header = node_info[2] length = len(blob) if header['linknode'] in self.reduce_effort: algorithms = [ALGO] else: algorithms = DEFAULT_ALGORITHMS h = MultiHash.from_data(blob, hash_names=algorithms) content = h.digest() content['length'] = length blob_hash = content[ALGO] self.file_node_to_hash[header['node']] = blob_hash if header['linknode'] in self.reduce_effort: continue hash_to_info[blob_hash] = node_info contents[blob_hash] = content missing_contents.add(blob_hash) if file_name == b'.hgtags': # https://www.mercurial-scm.org/wiki/GitConcepts#Tag_model # overwrite until the last one self.tags = (t for t in blob.split(b'\n') if t != b'') if contents: missing_contents = set( self.storage.content_missing( list(contents.values()), key_hash=ALGO ) ) # Clusters needed blobs by file offset and then only fetches the # groups at the needed offsets. focs = {} # "file/offset/contents" for blob_hash in missing_contents: _, file_offset, header = hash_to_info[blob_hash] focs.setdefault(file_offset, {}) focs[file_offset][header['node']] = blob_hash hash_to_info = None for offset, node_hashes in sorted(focs.items()): for header, data, *_ in self.br.yield_group_objects( group_offset=offset ): node = header['node'] if node in node_hashes: blob, meta = self.br.extract_meta_from_blob(data) content = contents.pop(node_hashes[node], None) if content: content['data'] = blob yield content_for_storage( content, log=self.log, max_content_size=self.content_max_size_limit, origin_id=self.origin_id ) def load_directories(self): """This is where the work is done to convert manifest deltas from the repository bundle into SWH directories. """ self.mnode_to_tree_id = {} cache_hints = self.br.build_manifest_hints() def tree_size(t): return t.size() self.trees = SelectiveCache(cache_hints=cache_hints, size_function=tree_size, filename=self.cache_filename2, max_size=self.cache2_size) tree = SimpleTree() for header, added, removed in self.br.yield_all_manifest_deltas( cache_hints ): node = header['node'] basenode = header['basenode'] tree = self.trees.fetch(basenode) or tree # working tree for path in removed.keys(): tree = tree.remove_tree_node_for_path(path) for path, info in added.items(): file_node, is_symlink, perms_code = info tree = tree.add_blob( path, self.file_node_to_hash[file_node], is_symlink, perms_code ) if header['linknode'] in self.reduce_effort: self.trees.store(node, tree) else: new_dirs = [] self.mnode_to_tree_id[node] = tree.hash_changed(new_dirs) self.trees.store(node, tree) yield header, tree, new_dirs def get_directories(self): """Compute directories to load """ dirs = {} self.num_directories = 0 for _, _, new_dirs in self.load_directories(): for d in new_dirs: self.num_directories += 1 dirs[d['id']] = d missing_dirs = list(dirs.keys()) if missing_dirs: missing_dirs = self.storage.directory_missing(missing_dirs) for _id in missing_dirs: yield dirs[_id] dirs = {} def get_revisions(self): """Compute revisions to load """ revisions = {} self.num_revisions = 0 for header, commit in self.br.yield_all_changesets(): if header['node'] in self.reduce_effort: continue self.num_revisions += 1 date_dict = identifiers.normalize_timestamp( int(commit['time'].timestamp()) ) author_dict = converters.parse_author(commit['user']) if commit['manifest'] == Bundle20Reader.NAUGHT_NODE: directory_id = SimpleTree().hash_changed() else: directory_id = self.mnode_to_tree_id[commit['manifest']] extra_meta = [] extra = commit.get('extra') if extra: for e in extra.split(b'\x00'): k, v = e.split(b':', 1) k = k.decode('utf-8') + # transplant_source stores binary reference to a changeset + # prefer to dump hexadecimal one in the revision metadata + if k == 'transplant_source': + v = hash_to_hex(v) extra_meta.append([k, v]) revision = { 'author': author_dict, 'date': date_dict, 'committer': author_dict, 'committer_date': date_dict, 'type': 'hg', 'directory': directory_id, 'message': commit['message'], 'metadata': { 'node': hash_to_hex(header['node']), 'extra_headers': [ ['time_offset_seconds', str(commit['time_offset_seconds']).encode('utf-8')], ] + extra_meta }, 'synthetic': False, 'parents': [] } p1 = self.node_2_rev.get(header['p1']) p2 = self.node_2_rev.get(header['p2']) if p1: revision['parents'].append(p1) if p2: revision['parents'].append(p2) revision['id'] = hash_to_bytes( identifiers.revision_identifier(revision) ) self.node_2_rev[header['node']] = revision['id'] revisions[revision['id']] = revision # Converts heads to use swh ids self.heads = { branch_name: (pointer_nature, self.node_2_rev[node_id]) for branch_name, (pointer_nature, node_id) in self.heads.items() } missing_revs = revisions.keys() if missing_revs: missing_revs = set( self.storage.revision_missing(list(missing_revs)) ) for r in missing_revs: yield revisions[r] self.mnode_to_tree_id = None def _read_tag(self, tag, split_byte=b' '): node, *name = tag.split(split_byte) name = split_byte.join(name) return node, name def get_releases(self): """Get the releases that need to be loaded.""" self.num_releases = 0 releases = {} missing_releases = [] for t in self.tags: self.num_releases += 1 node, name = self._read_tag(t) node = node.decode() node_bytes = hash_to_bytes(node) if not TAG_PATTERN.match(node): self.log.warn('Wrong pattern (%s) found in tags. Skipping' % ( node, )) continue if node_bytes not in self.node_2_rev: self.log.warn('No matching revision for tag %s ' '(hg changeset: %s). Skipping' % (name.decode(), node)) continue tgt_rev = self.node_2_rev[node_bytes] release = { 'name': name, 'target': tgt_rev, 'target_type': 'revision', 'message': None, 'metadata': None, 'synthetic': False, 'author': {'name': None, 'email': None, 'fullname': b''}, 'date': None } id_hash = hash_to_bytes( identifiers.release_identifier(release)) release['id'] = id_hash missing_releases.append(id_hash) releases[id_hash] = release self.releases[name] = id_hash if missing_releases: missing_releases = set( self.storage.release_missing(missing_releases)) for _id in missing_releases: yield releases[_id] def get_snapshot(self): """Get the snapshot that need to be loaded.""" branches = {} for name, (pointer_nature, target) in self.heads.items(): branches[name] = {'target': target, 'target_type': 'revision'} if pointer_nature == HEAD_POINTER_NAME: branches[b'HEAD'] = {'target': name, 'target_type': 'alias'} for name, target in self.releases.items(): branches[name] = {'target': target, 'target_type': 'release'} snap = { 'id': None, 'branches': branches, } snap['id'] = identifiers.identifier_to_bytes( identifiers.snapshot_identifier(snap)) return snap def get_fetch_history_result(self): """Return the data to store in fetch_history.""" return { 'contents': self.num_contents, 'directories': self.num_directories, 'revisions': self.num_revisions, 'releases': self.num_releases, } class HgArchiveBundle20Loader(HgBundle20Loader): """Mercurial loader for repository wrapped within archives. """ def __init__(self): super().__init__( logging_class='swh.loader.mercurial.HgArchiveBundle20Loader') self.temp_dir = None def prepare(self, *, origin_url, archive_path, visit_date): self.temp_dir = tmp_extract(archive=archive_path, dir=self.temp_directory, prefix=TEMPORARY_DIR_PREFIX_PATTERN, suffix='.dump-%s' % os.getpid(), log=self.log, source=origin_url) repo_name = os.listdir(self.temp_dir)[0] directory = os.path.join(self.temp_dir, repo_name) super().prepare(origin_url=origin_url, visit_date=visit_date, directory=directory) def cleanup(self): if self.temp_dir and os.path.exists(self.temp_dir): rmtree(self.temp_dir) super().cleanup() diff --git a/swh/loader/mercurial/tests/resources/transplant.tgz b/swh/loader/mercurial/tests/resources/transplant.tgz new file mode 100644 index 0000000..99be373 Binary files /dev/null and b/swh/loader/mercurial/tests/resources/transplant.tgz differ diff --git a/swh/loader/mercurial/tests/test_loader.py b/swh/loader/mercurial/tests/test_loader.py index 0210ad1..c153fc8 100644 --- a/swh/loader/mercurial/tests/test_loader.py +++ b/swh/loader/mercurial/tests/test_loader.py @@ -1,249 +1,289 @@ # Copyright (C) 2018 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 os from unittest.mock import patch from swh.loader.core.tests import BaseLoaderTest from .common import HgLoaderMemoryStorage, HgArchiveLoaderMemoryStorage class BaseHgLoaderTest(BaseLoaderTest): """Mixin base loader test to prepare the mercurial repository to uncompress, load and test the results. This sets up """ def setUp(self, loader=HgLoaderMemoryStorage, archive_name='the-sandbox.tgz', filename='the-sandbox', uncompress_archive=True): super().setUp(archive_name=archive_name, filename=filename, prefix_tmp_folder_name='swh.loader.mercurial.', start_path=os.path.dirname(__file__), uncompress_archive=uncompress_archive) self.loader = loader() self.storage = self.loader.storage -class LoaderTestT1(BaseHgLoaderTest): +class WithoutReleaseLoaderTest(BaseHgLoaderTest): """Load a mercurial repository without release """ def test_load(self): """Load a repository with multiple branches results in 1 snapshot """ # when self.loader.load( origin_url=self.repo_url, visit_date='2016-05-03 15:16:32+00', directory=self.destination_path) # then self.assertCountContents(2) self.assertCountDirectories(3) self.assertCountReleases(0) self.assertCountRevisions(58) tip_revision_develop = 'a9c4534552df370f43f0ef97146f393ef2f2a08c' tip_revision_default = '70e750bb046101fdced06f428e73fee471509c56' # same from rev 3 onward directory_hash = '180bd57623a7c2c47a8c43514a5f4d903503d0aa' # cf. test_loader.org for explaining from where those hashes # come from expected_revisions = { # revision hash | directory hash # noqa 'aafb69fd7496ca617f741d38c40808ff2382aabe': 'e2e117569b086ceabeeedee4acd95f35298d4553', # noqa 'b6932cb7f59e746899e4804f3d496126d1343615': '9cd8160c67ac4b0bc97e2e2cd918a580425167d3', # noqa tip_revision_default: directory_hash, '18012a93d5aadc331c468dac84b524430f4abc19': directory_hash, 'bec4c0a31b0b2502f44f34aeb9827cd090cca621': directory_hash, '5f4eba626c3f826820c4475d2d81410759ec911b': directory_hash, 'dcba06661c607fe55ec67b1712d153b69f65e38c': directory_hash, 'c77e776d22548d47a8d96463a3556172776cd59b': directory_hash, '61d762d65afb3150e2653d6735068241779c1fcf': directory_hash, '40def747398c76ceec1bd248e3a6cb2a52e22dc5': directory_hash, '6910964416438ca8d1698f6295871d727c4d4851': directory_hash, 'be44d5e6cc66580f59c108f8bff5911ee91a22e4': directory_hash, 'c4a95d5097519dedac437fddf0ef775136081241': directory_hash, '32eb0354a660128e205bf7c3a84b46040ef70d92': directory_hash, 'dafa445964230e808148db043c126063ea1dc9b6': directory_hash, 'a41e2a548ba51ee47f22baad8e88994853d3e2f5': directory_hash, 'dc3e3ab7fe257d04769528e5e17ad9f1acb44659': directory_hash, 'd2164061453ecb03d4347a05a77db83f706b8e15': directory_hash, '34192ceef239b8b72141efcc58b1d7f1676a18c9': directory_hash, '2652147529269778757d96e09aaf081695548218': directory_hash, '4d640e8064fe69b4c851dfd43915c431e80c7497': directory_hash, 'c313df50bfcaa773dcbe038d00f8bd770ba997f8': directory_hash, '769db00b34b9e085dc699c8f1550c95793d0e904': directory_hash, '2973e5dc9568ac491b198f6b7f10c44ddc04e0a3': directory_hash, 'be34b8c7857a6c04e41cc06b26338d8e59cb2601': directory_hash, '24f45e41637240b7f9e16d2791b5eacb4a406d0f': directory_hash, '62ff4741eac1821190f6c2cdab7c8a9d7db64ad0': directory_hash, 'c346f6ff7f42f2a8ff867f92ab83a6721057d86c': directory_hash, 'f2afbb94b319ef5d60823859875284afb95dcc18': directory_hash, '4e2dc6d6073f0b6d348f84ded52f9143b10344b9': directory_hash, '31cd7c5f669868651c57e3a2ba25ac45f76fa5cf': directory_hash, '25f5b27dfa5ed15d336188ef46bef743d88327d4': directory_hash, '88b80615ed8561be74a700b92883ec0374ddacb0': directory_hash, '5ee9ea92ed8cc1737b7670e39dab6081c64f2598': directory_hash, 'dcddcc32740d2de0e1403e21a5c4ed837b352992': directory_hash, '74335db9f45a5d1c8133ff7a7db5ed7a8d4a197b': directory_hash, 'cb36b894129ca7910bb81c457c72d69d5ff111bc': directory_hash, 'caef0cb155eb6c55215aa59aabe04a9c702bbe6a': directory_hash, '5017ce0b285351da09a2029ea2cf544f79b593c7': directory_hash, '17a62618eb6e91a1d5d8e1246ccedae020d3b222': directory_hash, 'a1f000fb8216838aa2a120738cc6c7fef2d1b4d8': directory_hash, '9f82d95bd3edfb7f18b1a21d6171170395ea44ce': directory_hash, 'a701d39a17a9f48c61a06eee08bd9ac0b8e3838b': directory_hash, '4ef794980f820d44be94b2f0d53eb34d4241638c': directory_hash, 'ddecbc16f4c916c39eacfcb2302e15a9e70a231e': directory_hash, '3565e7d385af0745ec208d719e469c2f58be8e94': directory_hash, 'c875bad563a73a25c5f3379828b161b1441a7c5d': directory_hash, '94be9abcf9558213ff301af0ecd8223451ce991d': directory_hash, '1ee770fd10ea2d8c4f6e68a1dbe79378a86611e0': directory_hash, '553b09724bd30d9691b290e157b27a73e2d3e537': directory_hash, '9e912851eb64e3a1e08fbb587de7a4c897ce5a0a': directory_hash, '9c9e0ff08f215a5a5845ce3dbfc5b48c8050bdaf': directory_hash, 'db9e625ba90056304897a94c92e5d27bc60f112d': directory_hash, '2d4a801c9a9645fcd3a9f4c06418d8393206b1f3': directory_hash, 'e874cd5967efb1f45282e9f5ce87cc68a898a6d0': directory_hash, 'e326a7bbb5bc00f1d8cacd6108869dedef15569c': directory_hash, '3ed4b85d30401fe32ae3b1d650f215a588293a9e': directory_hash, tip_revision_develop: directory_hash, } self.assertRevisionsContain(expected_revisions) self.assertCountSnapshots(1) expected_snapshot = { 'id': '3b8fe58e467deb7597b12a5fd3b2c096b8c02028', 'branches': { 'develop': { 'target': tip_revision_develop, 'target_type': 'revision' }, 'default': { 'target': tip_revision_default, 'target_type': 'revision' }, 'HEAD': { 'target': 'develop', 'target_type': 'alias', } } } self.assertSnapshotEqual(expected_snapshot) self.assertEqual(self.loader.load_status(), {'status': 'eventful'}) self.assertEqual(self.loader.visit_status(), 'full') class CommonHgLoaderData: def assert_data_ok(self): # then self.assertCountContents(3) self.assertCountDirectories(3) self.assertCountReleases(1) self.assertCountRevisions(3) tip_release = '515c4d72e089404356d0f4b39d60f948b8999140' self.assertReleasesContain([tip_release]) tip_revision_default = 'c3dbe4fbeaaa98dd961834e4007edb3efb0e2a27' # cf. test_loader.org for explaining from where those hashes # come from expected_revisions = { # revision hash | directory hash # noqa '93b48d515580522a05f389bec93227fc8e43d940': '43d727f2f3f2f7cb3b098ddad1d7038464a4cee2', # noqa '8dd3db5d5519e4947f035d141581d304565372d2': 'b3f85f210ff86d334575f64cb01c5bf49895b63e', # noqa tip_revision_default: '8f2be433c945384c85920a8e60f2a68d2c0f20fb', } self.assertRevisionsContain(expected_revisions) self.assertCountSnapshots(1) expected_snapshot = { 'id': 'd35668e02e2ba4321dc951cd308cf883786f918a', 'branches': { 'default': { 'target': tip_revision_default, 'target_type': 'revision' }, '0.1': { 'target': tip_release, 'target_type': 'release' }, 'HEAD': { 'target': 'default', 'target_type': 'alias', } } } self.assertSnapshotEqual(expected_snapshot) self.assertEqual(self.loader.load_status(), {'status': 'eventful'}) self.assertEqual(self.loader.visit_status(), 'full') -class LoaderTest2(BaseHgLoaderTest, CommonHgLoaderData): +class WithReleaseLoaderTest(BaseHgLoaderTest, CommonHgLoaderData): """Load a mercurial repository with release """ def setUp(self): super().setUp(archive_name='hello.tgz', filename='hello') def test_load(self): """Load a repository with tags results in 1 snapshot """ # when self.loader.load( origin_url=self.repo_url, visit_date='2016-05-03 15:16:32+00', directory=self.destination_path) self.assert_data_ok() class ArchiveLoaderTest(BaseHgLoaderTest, CommonHgLoaderData): """Load a mercurial repository archive with release """ def setUp(self): super().setUp(loader=HgArchiveLoaderMemoryStorage, archive_name='hello.tgz', filename='hello', uncompress_archive=False) def test_load(self): """Load a mercurial repository archive with tags results in 1 snapshot """ # when self.loader.load( origin_url=self.repo_url, visit_date='2016-05-03 15:16:32+00', archive_path=self.destination_path) self.assert_data_ok() @patch('swh.loader.mercurial.archive_extract.patoolib') def test_load_with_failure(self, mock_patoo): mock_patoo.side_effect = ValueError # when r = self.loader.load( origin_url=self.repo_url, visit_date='2016-05-03 15:16:32+00', archive_path=self.destination_path) self.assertEqual(r, {'status': 'failed'}) self.assertCountContents(0) self.assertCountDirectories(0) self.assertCountRevisions(0) self.assertCountReleases(0) self.assertCountSnapshots(0) + + +class WithTransplantLoaderTest(BaseHgLoaderTest): + """Load a mercurial repository where transplant operations + have been used. + + """ + def setUp(self): + super().setUp(archive_name='transplant.tgz', filename='transplant') + + def test_load(self): + # load hg repository + self.loader.load( + origin_url=self.repo_url, + visit_date='2019-05-23 12:06:00+00', + directory=self.destination_path) + + # collect swh revisions + origin_id = self.storage.origin_get([ + {'type': 'hg', 'url': self.repo_url}])[0]['id'] + snapshot = self.storage.snapshot_get_latest(origin_id) + revisions = [] + for branch in snapshot['branches'].values(): + if branch['target_type'] != 'revision': + continue + revisions.append(branch['target']) + + # extract original changesets info and the transplant sources + hg_changesets = set() + transplant_sources = set() + for rev in self.storage.revision_log(revisions): + hg_changesets.add(rev['metadata']['node']) + for k, v in rev['metadata']['extra_headers']: + if k == 'transplant_source': + transplant_sources.add(v) + + # check extracted data are valid + self.assertTrue(len(hg_changesets) > 0) + self.assertTrue(len(transplant_sources) > 0) + self.assertTrue(transplant_sources.issubset(hg_changesets))