Source code for swh.loader.metadata.journal_client

# Copyright (C) 2023  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 dataclasses
import datetime
from typing import Any, Dict, Iterable, List, Type, TypeVar, cast
import uuid

from swh.core.api.classes import stream_results
from swh.core.statsd import Statsd
from swh.loader.core.metadata_fetchers import CredentialsType, get_fetchers_for_lister
from swh.loader.metadata.base import BaseMetadataFetcher
from swh.model.model import (
from swh.scheduler.interface import ListedOrigin, Lister, SchedulerInterface
from import StorageInterface

def _now() -> datetime.datetime:
    # Used by tests for mocking

_TItem = TypeVar("_TItem")

[docs] @dataclasses.dataclass class JournalClient: scheduler: SchedulerInterface storage: StorageInterface metadata_fetcher_credentials: CredentialsType reload_after_days: int def __post_init__(self): self._listers = {} self._added_fetchers = set() self._added_authorities = set() self.statsd = Statsd(namespace="swh_loader_metadata_journal_client")
[docs] def statsd_timed(self, name: str, tags: Dict[str, Any] = {}): """ Wrapper for :meth:`swh.core.statsd.Statsd.timed`, which uses the standard metric name and tag. """ return self.statsd.timed( "operation_duration_seconds", tags={"operation": name, **tags} )
[docs] def statsd_timing(self, name: str, value: float, tags: Dict[str, Any] = {}) -> None: """ Wrapper for :meth:`swh.core.statsd.Statsd.timing`, which uses the standard metric name and tags for loaders. """ self.statsd.timing( "operation_duration_seconds", value, tags={"operation": name, **tags} )
def _get_lister(self, lister_id: uuid.UUID) -> Lister: if lister_id not in self._listers: with self.statsd_timed("get_listers_by_id"): (lister,) = self.scheduler.get_listers_by_id([str(lister_id)]) self._listers[] = lister return self._listers[lister_id] def _add_metadata_fetchers(self, fetchers: Iterable[MetadataFetcher]) -> None: for fetcher in fetchers: if fetcher not in self._added_fetchers:[fetcher]) self._added_fetchers.add(fetcher) def _add_metadata_authorities( self, authorities: Iterable[MetadataAuthority] ) -> None: for authority in authorities: if authority not in self._added_authorities:[authority]) self._added_authorities.add(authority)
[docs] def process_journal_objects(self, messages: Dict[str, List[Dict]]) -> None: """Loads metadata for origins not recently loaded: 1. reads messages from the origin journal topic 2. queries the scheduler for a list of listers that produced this origin (to guess what type of forge it is) 3. if it is a forge we can get extrinsic metadata from, check if we got any recently, using the storage 4. if not, trigger a metadata load """ assert set(messages) == {"origin"}, f"Unexpected message types: {set(messages)}" for origin in messages["origin"]: for listed_origin in stream_results( self.statsd_timed("get_listed_origins")( self.scheduler.get_listed_origins ), url=origin["url"], ): self._process_listed_origin(listed_origin) with self.statsd_timed("flush_storage"):
def _process_listed_origin( self, listed_origin: ListedOrigin, ) -> List[RawExtrinsicMetadata]: origin = Origin(url=listed_origin.url) lister = self._get_lister(listed_origin.lister_id) tags = { "lister":, "lister_instance": lister.instance_name, } fetcher_classes = cast( List[Type[BaseMetadataFetcher]], get_fetchers_for_lister( ) self.statsd.histogram("metadata_fetchers", len(fetcher_classes), tags=tags) now = _now() metadata: List[RawExtrinsicMetadata] = [] for cls in fetcher_classes: tags["fetcher"] = cls.FETCHER_NAME metadata_fetcher = cls( origin=origin,, lister_instance_name=lister.instance_name, credentials=self.metadata_fetcher_credentials, ) with self.statsd_timed("raw_extrinsic_metadata_get"): last_metadata = target=origin.swhid(), authority=metadata_fetcher.metadata_authority(), after=now - datetime.timedelta(days=self.reload_after_days), limit=1, ) if last_metadata.results: # We already have recent metadata; don't load it again. self.statsd.increment( "metadata_items_fetched_total", len(last_metadata.results), tags=tags, ) continue with self.statsd_timed("get_origin_metadata", tags=tags): metadata = list(metadata_fetcher.get_origin_metadata()) self.statsd.increment( "metadata_items_added_total", len(metadata), tags=tags ) with self.statsd_timed("metadata_fetcher_add"): self._add_metadata_fetchers({m.fetcher for m in metadata}) with self.statsd_timed("metadata_authority_add"): self._add_metadata_authorities({m.authority for m in metadata}) with self.statsd_timed("raw_extrinsic_metadata_add"): return metadata