objstorage_checker: Use check method from object storage interface

Instead of reinventing the wheel, prefer to use the check method from the object storage interface for verifying content presence and integrity.

Depends on !65 (merged) and swh-objstorage!178 (merged).

Related to #4694 (closed).

swh/scrubber/objstorage_checker.py

@@ -5,17 +5,19 @@
 
 import json
 import logging
-from typing import Iterable, Optional
+from typing import Any, Dict, List, Optional, Protocol
 
-from swh.journal.serializers import value_to_kafka
+from swh.core.statsd import Statsd
+from swh.journal.client import get_journal_client
+from swh.journal.serializers import kafka_to_value, value_to_kafka
 from swh.model.model import Content
 from swh.model.swhids import ObjectType
-from swh.objstorage.exc import ObjNotFoundError
+from swh.objstorage.exc import ObjCorruptedError, ObjNotFoundError
 from swh.objstorage.interface import ObjStorageInterface, objid_from_dict
 from swh.storage.interface import StorageInterface
 
-from .base_checker import BasePartitionChecker
-from .db import Datastore, ScrubberDb
+from .base_checker import BaseChecker, BasePartitionChecker
+from .db import ConfigEntry, Datastore, ScrubberDb
 
 logger = logging.getLogger(__name__)
 
@@ -29,12 +31,56 @@ def get_objstorage_datastore(objstorage_config):
     )
 
 
-class ObjectStorageChecker(BasePartitionChecker):
-    """A checker to detect missing and corrupted contents in an object storage.
+class ObjectStorageCheckerProtocol(Protocol):
+    db: ScrubberDb
+    objstorage: ObjStorageInterface
 
-    It iterates on content objects referenced in a storage instance, check they
-    are available in a given object storage instance then retrieve their bytes
-    from it in order to recompute checksums and detect corruptions."""
+    @property
+    def config(self) -> ConfigEntry:
+        ...
+
+    @property
+    def statsd(self) -> Statsd:
+        ...
+
+
+class ContentCheckerMixin(ObjectStorageCheckerProtocol):
+    """Mixin class implementing content checks used by object storage checkers."""
+
+    def check_content(self, content: Content) -> None:
+        """Checks if a content exists in an object storage (if ``check_references`` is set to
+        :const:`True` in checker config) or if a content is corrupted in an object storage (if
+        ``check_hashes`` is set to :const:`True` in checker config).
+        """
+
+        content_hashes = objid_from_dict(content.hashes())
+        try:
+            self.objstorage.check(content_hashes)
+        except ObjNotFoundError:
+            if self.config.check_references:
+                self.statsd.increment("missing_object_total")
+                self.db.missing_object_add(
+                    id=content.swhid(), reference_ids={}, config=self.config
+                )
+        except ObjCorruptedError:
+            if self.config.check_hashes:
+                self.statsd.increment("hash_mismatch_total")
+                self.db.corrupt_object_add(
+                    id=content.swhid(),
+                    config=self.config,
+                    serialized_object=value_to_kafka(content.to_dict()),
+                )
+
+
+class ObjectStorageCheckerFromStoragePartition(
+    BasePartitionChecker, ContentCheckerMixin
+):
+    """A partition based checker to detect missing and corrupted contents in an object storage.
+
+    It iterates on content objects referenced in a storage instance, check they are available
+    in a given object storage instance (if ``check_references`` is set to :const:`True` in
+    checker config) then retrieve their bytes from it in order to recompute checksums and detect
+    corruptions (if ``check_hashes`` is set to :const:`True` in checker config)."""
 
     def __init__(
         self,
@@ -49,15 +95,21 @@ class ObjectStorageChecker(BasePartitionChecker):
         self.objstorage = (
             objstorage if objstorage is not None else getattr(storage, "objstorage")
         )
-        self.statsd_constant_tags["datastore_instance"] = self.datastore.instance
 
-    def check_partition(self, object_type: ObjectType, partition_id: int) -> None:
-        if object_type != ObjectType.CONTENT:
+        object_type = self.object_type.name.lower()
+
+        if object_type != "content":
+            raise ValueError(
+                "ObjectStorageCheckerFromStoragePartition can only check objects of type "
+                f"content, checking objects of type {object_type} is not supported."
+            )
+
+        if self.objstorage is None:
             raise ValueError(
-                "ObjectStorageChecker can only check objects of type content,"
-                f"checking objects of type {object_type.name.lower()} is not supported."
+                "An object storage must be provided to ObjectStorageCheckerFromStoragePartition."  # noqa
             )
 
+    def check_partition(self, object_type: ObjectType, partition_id: int) -> None:
         page_token = None
         while True:
             page = self.storage.content_get_partition(
@@ -71,32 +123,56 @@ class ObjectStorageChecker(BasePartitionChecker):
                 "batch_duration_seconds", tags={"operation": "check_hashes"}
             ):
                 logger.debug("Checking %s content object hashes", len(contents))
-                self.check_contents(contents)
+                for content in contents:
+                    self.check_content(content)
 
             page_token = page.next_page_token
             if page_token is None:
                 break
 
-    def check_contents(self, contents: Iterable[Content]) -> None:
-        for content in contents:
-            content_hashes = objid_from_dict(content.hashes())
-            try:
-                content_bytes = self.objstorage.get(content_hashes)
-            except ObjNotFoundError:
-                if self.check_references:
-                    self.statsd.increment("missing_object_total")
-                    self.db.missing_object_add(
-                        id=content.swhid(), reference_ids={}, config=self.config
-                    )
-            else:
-                if self.check_hashes:
-                    recomputed_hashes = objid_from_dict(
-                        Content.from_data(content_bytes).hashes()
-                    )
-                    if content_hashes != recomputed_hashes:
-                        self.statsd.increment("hash_mismatch_total")
-                        self.db.corrupt_object_add(
-                            id=content.swhid(),
-                            config=self.config,
-                            serialized_object=value_to_kafka(content.to_dict()),
-                        )
+
+class ObjectStorageCheckerFromJournal(BaseChecker, ContentCheckerMixin):
+    """A journal based checker to detect missing and corrupted contents in an object storage.
+
+    It iterates on content objects referenced in a kafka topic, check they are available
+    in a given object storage instance then retrieve their bytes from it in order to
+    recompute checksums and detect corruptions."""
+
+    def __init__(
+        self,
+        db: ScrubberDb,
+        config_id: int,
+        journal_client_config: Dict[str, Any],
+        objstorage: ObjStorageInterface,
+    ):
+        super().__init__(db=db, config_id=config_id)
+        self.objstorage = objstorage
+
+        object_type = self.object_type.name.lower()
+
+        if object_type != "content":
+            raise ValueError(
+                "ObjectStorageCheckerFromJournal can only check objects of type content,"
+                f"checking objects of type {object_type} is not supported."
+            )
+
+        self.journal_client_config = journal_client_config.copy()
+        if "object_types" in self.journal_client_config:
+            raise ValueError(
+                "The journal_client configuration entry should not define the "
+                "object_types field; this is handled by the scrubber configuration entry"
+            )
+        self.journal_client_config["object_types"] = [object_type]
+        self.journal_client = get_journal_client(
+            **self.journal_client_config,
+            # Remove default deserializer; so process_kafka_values() gets the message
+            # verbatim so it can archive it with as few modifications a possible.
+            value_deserializer=lambda obj_type, msg: msg,
+        )
+
+    def run(self) -> None:
+        self.journal_client.process(self.process_kafka_messages)
+
+    def process_kafka_messages(self, all_messages: Dict[str, List[bytes]]):
+        for message in all_messages["content"]:
+            self.check_content(Content.from_dict(kafka_to_value(message)))