Differential D6344 Diff 24438 swh/loader/bzr/loader.py

Changeset View

Standalone View

View Options

swh/loader/bzr/loader.py

This file was added.

# 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 Bazaar or Breezy.

"""

from collections import deque

from datetime import datetime

from typing import Deque, Dict, Iterator, NewType, Optional, Set, TypeVar, Union

from breezy import repository

from breezy.bzr import bzrdir

from breezy.bzr.inventory import Inventory, InventoryEntry

from breezy.revision import NULL_REVISION

from breezy.revision import Revision as BzrRevision

from swh.loader.core.loader import BaseLoader

from swh.loader.core.utils import clean_dangling_folders

from swh.model import identifiers

from swh.model.from_disk import Content, DentryPerms, Directory

from swh.model.model import (

ExtID,

ObjectType,

Origin,

Person,

Release,

Revision,

RevisionType,

Sha1Git,

Snapshot,

SnapshotBranch,

TargetType,

TimestampWithTimezone,

)

from swh.model.model import Content as ModelContent

from swh.storage.interface import StorageInterface

TEMPORARY_DIR_PREFIX_PATTERN = "swh.loader.bzr.from_disk"

EXTID_TYPE = "bzr-nodeid"

EXTID_VERSION: int = 1

vlorentzUnsubmitted

Done

This is going to be confusing, because swh.model.model also has types Content and Directory.

Could you replace from swh.model.from_disk import Content, DentryPerms, Directory with from swh.model import from_disk?

vlorentz: This is going to be confusing, because `swh.model.model` also has types `Content` and…

BzrRevisionId = NewType("BzrRevisionId", bytes)

T = TypeVar("T")

class RepositoryNeedsUpgrade(Exception):

"""The repository we're trying to load is using an old format.

We only support format 2a (the most recent), see `brz help upgrade`"""

def sort_changes(change):

target_path = change[1][1] or ""

source_path = change[1][0]

return source_path if source_path is not None else target_path

class BzrDirectory(Directory):

"""A more practical directory.

- creates missing parent directories

- removes empty directories

"""

def __setitem__(self, path: bytes, value: Union[Content, "BzrDirectory"]) -> None:

if b"/" in path:

head, tail = path.split(b"/", 1)

directory = self.get(head)

if directory is None or isinstance(directory, Content):

directory = BzrDirectory()

self[head] = directory

directory[tail] = value

else:

super().__setitem__(path, value)

def __delitem__(self, path: bytes) -> None:

super().__delitem__(path)

while b"/" in path: # remove empty parent directories

path = path.rsplit(b"/", 1)[0]

if len(self[path]) == 0:

super().__delitem__(path)

else:

break

vlorentzUnsubmitted

Done

Could you add tests for this?

vlorentz: Could you add tests for this?

def get(

self, path: bytes, default: Optional[T] = None

) -> Optional[Union[Content, "BzrDirectory", T]]:

# TODO move to swh.model.from_disk.Directory

try:

return self[path]

except KeyError:

return default

class BazaarLoaderFromDisk(BaseLoader):

"""Loads a Bazaar repository from disk"""

visit_type = "bzr"

def __init__(

self,

storage: StorageInterface,

url: str,

directory: Optional[str] = None,

logging_class: str = "swh.loader.bzr.LoaderFromDisk",

visit_date: Optional[datetime] = None,

temp_directory: str = "/tmp",

clone_timeout_seconds: int = 7200,

max_content_size: Optional[int] = None,

super().__init__(

storage=storage,

logging_class=logging_class,

max_content_size=max_content_size,

)

self._temp_directory = temp_directory

self._clone_timeout = clone_timeout_seconds

self._revision_id_to_sha1git: Dict[BzrRevisionId, Sha1Git] = {}

self._last_root = BzrDirectory()

self._tags: Optional[Dict[bytes, BzrRevisionId]] = None

self._head_revision_id: Optional[bytes] = None

self._seen_revisions: Set[BzrRevisionId] = set()

self._prev_inventory: Optional[Inventory] = None

self.origin_url = url

self.visit_date = visit_date

self.directory = url # TODO directory/url distinction

self.repo: Optional[repository.Repository] = None

def pre_cleanup(self) -> None:

"""As a first step, will try and check for dangling data to cleanup.

This should do its best to avoid raising issues.

"""

clean_dangling_folders(

self._temp_directory,

pattern_check=TEMPORARY_DIR_PREFIX_PATTERN,

log=self.log,

)

def prepare_origin_visit(self) -> None:

"""First step executed by the loader to prepare origin and visit

references. Set/update self.origin, and

optionally self.origin_url, self.visit_date.

"""

self.origin = Origin(url=self.origin_url)

def prepare(self) -> None:

"""Second step executed by the loader to prepare some state needed by

the loader.

"""

res = bzrdir.BzrDir.open_containing_tree_branch_or_repository(self.directory)

(_tree, _branch, repo, _relpath) = res

repository_format = repo._format.as_string() # lies about being a string

if not repository_format.startswith(b"Bazaar repository format 2a"):

raise RepositoryNeedsUpgrade()

self.repo = repo

self.repo.lock_read()

def cleanup(self) -> None:

if self.repo is not None:

self.repo.unlock()

def fetch_data(self) -> bool:

"""Fetch the data from the source the loader is currently loading

Returns:

a value that is interpreted as a boolean. If True, fetch_data needs

to be called again to complete loading.

"""

self.head_revision_id # set the property

self.tags # set the property

return False

def get_bzr_revs_to_load(self) -> Iterator[BzrRevision]:

assert self.repo is not None

repo: repository.Repository = self.repo

self.log.debug("Getting fully sorted revision tree")

head_revision = repo.get_revision(self.head_revision_id)

# bazaar's model doesn't allow it to iterate on its graph from

# the bottom lazily, but basically all DAGs (especially bzr ones)

# are small enough to fit in RAM.

yield from self.iterate_ancestors(head_revision)

def iterate_ancestors(self, rev: BzrRevision) -> Iterator[BzrRevisionId]:

"""Return an iterator of this revision's ancestors"""

assert self.repo is not None

parents = rev.parent_ids

seen = self._seen_revisions

vlorentzUnsubmitted

Not Done

this should probably match some sort of delimiter after the format version (eg. if a future version uses "Bazaar repository format 2abc")

Also, it looks like this would fail with "needs upgrade" in the presence of repositories newer than this.

vlorentz: this should probably match some sort of delimiter after the format version (eg. if a future…

AlphareAuthorUnsubmitted

Done

this should probably match some sort of delimiter after the format version (eg. if a future version uses "Bazaar repository format 2abc")

Sure, good point

Also, it looks like this would fail with "needs upgrade" in the presence of repositories newer than this.

This is on purpose since it's very unlikely that Breezy will get a new repository format. In the case that it does, we will need to update this code anyway. Should the exception carry more detail in its docstring to encourage checking for a newer format if brz upgrade does work?

Alphare: > this should probably match some sort of delimiter after the format version (eg. if a future…

vlorentzUnsubmitted

Done

I would expect something like this:

if format is older:
    raise RepositoryNeedsUpgrade()
elif format is newer:
    assert False, "oh no, we need to update this loader for format XXX"

vlorentz: I would expect something like this: ``` if format is older: raise RepositoryNeedsUpgrade()…

visit = list(parents)

graph = [(rev.revision_id, rev.parent_ids)]

while visit:

rev_id = visit.pop()

length = len(seen)

seen.add(rev_id)

if length != len(seen):

revision = self.repo.get_revision(rev_id)

visit.extend(revision.parent_ids)

graph.append((rev_id, revision.parent_ids))

from breezy import tsort

sorter = tsort.TopoSorter(graph)

return sorter.sorted()

def store_data(self):

"""Store fetched data in the database."""

# Insert revisions using a topological sorting

revs = self.get_bzr_revs_to_load()

for rev in revs:

self.store_revision(self.repo.get_revision(rev))

snapshot_branches: Dict[bytes, SnapshotBranch] = {}

for tag_name, target in self.tags.items():

label = b"tags/%s" % tag_name

vlorentzUnsubmitted

Not Done

We would probably want to store a dangling branch for this.

vlorentz: We would probably want to store a dangling branch for this.

AlphareAuthorUnsubmitted

Done

I'm not 100% sure how to do that. Is the target just the empty string as git_objects.py explains?

Alphare: I'm not 100% sure how to do that. Is the target just the empty string as `git_objects.py`…

vlorentzUnsubmitted

Not Done

er, I think so

vlorentz: er, I think so

AlphareAuthorUnsubmitted

Done

Looking at the code in swh-model/swh/model.git_objects.py, dangling branches are not exposed as a TargetType variant but hardcoded bytes, so there appears to be no easy way for me to do it from the outside. Is there?

Alphare: Looking at the code in `swh-model/swh/model.git_objects.py`, dangling branches are not exposed…

vlorentzUnsubmitted

Not Done

SnapshotBranch(target=b"", target_type=TargetType.REVISION)

vlorentz: `SnapshotBranch(target=b"", target_type=TargetType.REVISION)`

AlphareAuthorUnsubmitted

Done

I had tried this, and it gives the following error: `swh.storage.exc.StorageArgumentException: value for domain sha1_git violates check constraint "sha1_git_check"
CONTEXT: COPY tmp_release, line 1, column target: "\x"`

Alphare: I had tried this, and it gives the following error: `swh.storage.exc.StorageArgumentException…

vlorentzUnsubmitted

Not Done

heh, alright, it's not worth the trouble then

vlorentz: heh, alright, it's not worth the trouble then

olasdUnsubmitted

Not Done

A dangling branch is stored in the Snapshot as a None instead of the SnapshotBranch object.

olasd: A dangling branch is stored in the Snapshot as a `None` instead of the `SnapshotBranch` object.

vlorentzUnsubmitted

Done

Aaah, that's what I was missing, thanks

vlorentz: Aaah, that's what I was missing, thanks

target = self.get_revision_id_from_bzr_id(target)

snapshot_branches[label] = SnapshotBranch(

target=self.store_release(tag_name, target),

target_type=TargetType.RELEASE,

)

head_revision_git_hash = self.get_revision_id_from_bzr_id(self.head_revision_id)

snapshot_branches[b"trunk"] = SnapshotBranch(

target=head_revision_git_hash, target_type=TargetType.REVISION

)

snapshot_branches[b"HEAD"] = SnapshotBranch(

target=b"trunk", target_type=TargetType.ALIAS,

vlorentzUnsubmitted

Done

ditto

vlorentz: ditto

)

snapshot = Snapshot(branches=snapshot_branches)

self.storage.snapshot_add([snapshot])

self.flush()

self.loaded_snapshot_id = snapshot.id

def store_release(self, name: bytes, target: Sha1Git) -> Sha1Git:

"""Store a release given its name and its target.

vlorentzUnsubmitted

Not Done

what is a branch nick?

vlorentz: what is a branch nick?

AlphareAuthorUnsubmitted

Done

The name associated with the branch in the clone source. It's usually the name of the project, but not always. It's not really meaningful in our case since we're keeping metadata about the project including its name anyway and we only have one branch per repository.

Alphare: The name associated with the branch in the clone source. It's usually the name of the project…

Args:

name: name of the release.

target: sha1_git of the target revision.

Returns:

the sha1_git of the stored release.

"""

release = Release(

name=name,

target=target,

target_type=ObjectType.REVISION,

message=None,

metadata=None,

synthetic=False,

author=Person(name=None, email=None, fullname=b""),

date=None,

vlorentzUnsubmitted

Not Done

self.log.debug(f"Storing revision {bzr_rev.revision_id}")

- date = TimestampWithTimezone.from_dict(int(bzr_rev.timestamp))

+ date = TimestampWithTimezone.from_numeric_offset(int(bzr_rev.timestamp), math.round(int(bzr_rev.timezone)/60))

extra_headers = [

brand-new method, added last week :)

vlorentz: brand-new method, added last week :)

AlphareAuthorUnsubmitted

Done

This does not seems to be valid, and I'm not sure exactly how to use the new method to be honest.

Alphare: This does not seems to be valid, and I'm not sure exactly how to use the new method to be…

vlorentzUnsubmitted

Not Done

why not?

vlorentz: why not?

)

vlorentzUnsubmitted

Not Done

We need to discuss this format. Could you open a task and ping olasd and I?

We are working on changes to TimestampWithTimezone, which makes it more extensible.

vlorentz: We need to discuss this format. Could you open a task and ping olasd and I? We are working on…

vlorentzUnsubmitted

Not Done

actually, I just did: T3886

vlorentz: actually, I just did: T3886

self.storage.release_add([release])

return release.id

vlorentzUnsubmitted

Done

another task for this. We could settle on GitHub's ad-hoc Co-authored-by format: https://docs.github.com/en/pull-requests/committing-changes-to-your-project/creating-and-editing-commits/creating-a-commit-with-multiple-authors but this needs a discussion

vlorentz: another task for this. We could settle on GitHub's ad-hoc `Co-authored-by` format: https://docs.

AlphareAuthorUnsubmitted

Done

Done in T3887 and T3888

Alphare: Done in T3887 and T3888

def store_content(

self, bzr_rev: BzrRevision, file_path: bytes, entry: InventoryEntry

) -> Content:

repo: repository.Repository = self.repo

if entry.executable:

perms = DentryPerms.executable_content

elif entry.kind == "directory":

perms = DentryPerms.directory

elif entry.kind == "symlink":

perms = DentryPerms.symlink

elif entry.kind == "file":

perms = DentryPerms.content

else:

raise RuntimeError("Hit unreachable condition")

data = b""

if entry.has_text():

rev_tree = repo.revision_tree(bzr_rev.revision_id)

data = rev_tree.get_file(file_path).read()

assert len(data) == entry.text_size

content = ModelContent.from_data(data)

self.storage.content_add([content])

return Content({"sha1_git": content.sha1_git, "perms": perms})

def store_directories(self, bzr_rev: BzrRevision) -> Sha1Git:

repo: repository.Repository = self.repo

vlorentzUnsubmitted

Done

# Save the mapping from SWHID to bzr id

- revision_swhid = identifiers.CoreSWHID(

- object_type=identifiers.ObjectType.REVISION, object_id=revision.id,

- )

self.storage.extid_add(

[

ExtID(

extid_type=EXTID_TYPE,

extid_version=EXTID_VERSION,

extid=bzr_rev.revision_id,

- target=revision_swhid,

+ target=revision.swhid(),

)

]

)

def store_directories(self, bzr_rev: BzrRevision) -> Sha1Git:

vlorentz:

inventory: Inventory = repo.get_inventory(bzr_rev.revision_id)

if self._prev_inventory is None:

for path, entry in inventory.iter_entries():

if path == "":

# root repo is created by default

continue

content = self.store_content(bzr_rev, path, entry)

self._last_root[path.encode()] = content

else:

changes_from_last_revision = sorted(

inventory.iter_changes(self._prev_inventory),

key=sort_changes,

reverse=True,

)

for change in changes_from_last_revision:

file_id = change[0]

(path_in_source, path_in_target) = change[1]

changed_content = change[2]

kind = change[6]

renamed = False

if kind[0] == "directory" or kind[1] == "directory":

# Should we somehow store empty directories?

continue

if path_in_target == "":

# root repo is created by default

continue

if kind[0] is None:

# added

pass

elif kind[1] is None:

# removed

try:

del self._last_root[path_in_source.encode()]

except KeyError:

corresponding_add = (

# target path == path in source

c[1][1] == path_in_source

for c in changes_from_last_revision

)

if not any(corresponding_add):

# A file can sometimes change representation in the tree

# which only changes its fileid. This can result in

# a removal *and* an add of the exact same content.

# If we don't find a corresponding add, it's an actual

# error.

vlorentzUnsubmitted

Not Done

what does it mean when entry.has_text() is falsy?

vlorentz: what does it mean when `entry.has_text()` is falsy?

AlphareAuthorUnsubmitted

Done

It means it's from an old tree that did not have the "weave index" for all inventory entries (something to help with merges IIUC), basically this is for backwards compat and the text is functionally empty.

Alphare: It means it's from an old tree that did not have the "weave index" for all inventory entries…

raise

continue

elif path_in_source != path_in_target:

# renamed, so remove the source

del self._last_root[path_in_source.encode()]

renamed = True

pass

if not changed_content and not renamed:

continue

content = self.store_content(

bzr_rev, path_in_target, inventory.get_entry(file_id)

)

self._last_root[path_in_target.encode()] = content

directories: Deque[Directory] = deque([self._last_root])

while directories:

directory = directories.pop()

self.storage.directory_add([directory.to_model()])

directories.extend(

[item for item in directory.values() if isinstance(item, Directory)]

)

self._prev_inventory = inventory

return self._last_root.hash

def store_revision(self, bzr_rev: BzrRevision):

self.log.debug(f"Storing revision {bzr_rev.revision_id}")

date = TimestampWithTimezone.from_dict(int(bzr_rev.timestamp))

extra_headers = [

(b"time_offset_seconds", str(bzr_rev.timezone).encode(),),

]

directory = self.store_directories(bzr_rev)

# TODO (how) should we store multiple authors, associated bugs?

revision = Revision(

author=Person.from_fullname(bzr_rev.get_apparent_authors()[0].encode()),

date=date,

committer=Person.from_fullname(bzr_rev.committer.encode()),

committer_date=date,

type=RevisionType.BAZAAR,

directory=directory,

message=bzr_rev.message.encode(),

extra_headers=extra_headers,

synthetic=False,

parents=self.get_revision_parents(bzr_rev),

)

vlorentzUnsubmitted

Done

should be faster with a regular list, since you're only using it as a stack

vlorentz: should be faster with a regular list, since you're only using it as a stack

self._revision_id_to_sha1git[bzr_rev.revision_id] = revision.id

self.storage.revision_add([revision])

# Save the mapping from SWHID to bzr id

revision_swhid = identifiers.CoreSWHID(

object_type=identifiers.ObjectType.REVISION, object_id=revision.id,

)

self.storage.extid_add(

[

ExtID(

extid_type=EXTID_TYPE,

extid_version=EXTID_VERSION,

extid=bzr_rev.revision_id,

target=revision_swhid,

)

]

)

def get_revision_parents(self, bzr_rev: BzrRevision):

parents = []

for parent_id in bzr_rev.parent_ids:

if parent_id == NULL_REVISION:

continue

revision_id = self.get_revision_id_from_bzr_id(parent_id)

parents.append(revision_id)

return tuple(parents)

def get_revision_id_from_bzr_id(self, bzr_id: BzrRevisionId) -> Sha1Git:

"""Return the git sha1 of a revision given its bazaar revision id.

"""

from_cache = self._revision_id_to_sha1git.get(bzr_id)

if from_cache is not None:

return from_cache

# The parent was not loaded in this run, get it from storage

from_storage = [

extid

for extid in self.storage.extid_get_from_extid(EXTID_TYPE, ids=[bzr_id])

if extid.extid_version == EXTID_VERSION

]

msg = "Expected 1 match from storage for bzr node %r, got %d"

assert len(from_storage) == 1, msg % (bzr_id, len(from_storage))

return from_storage[0].target.object_id

@property

def head_revision_id(self) -> bytes:

assert self.repo is not None

if self._head_revision_id is None:

self._head_revision_id = next(

self.repo.find_branches(using=True)

).last_revision()

return self._head_revision_id

@property

def tags(self) -> Optional[Dict[bytes, BzrRevisionId]]:

assert self.repo is not None

if self._tags is None:

for branch in self.repo.find_branches(using=True):

vlorentzUnsubmitted

Not Done

dead code?

vlorentz: dead code?

AlphareAuthorUnsubmitted

Done

Ah this will only be useful with the incremental loader which (most likely) will arrive in the next patch. I should probably remove the whole storage fallback for now.

Alphare: Ah this will only be useful with the incremental loader which (most likely) will arrive in the…

if branch.supports_tags():

self._tags = {

n.encode(): r for n, r in branch.tags.get_tag_dict().items()

vlorentzUnsubmitted

Done

could you add a docstring?

vlorentz: could you add a docstring?

}

break # TODO branches should be multiple origins

return self._tags