Implement sweeper data versioning for repairkit

src/pds/registrysweepers/ancestry/__init__.py

@@ -18,7 +18,6 @@
 from pds.registrysweepers.utils import parse_args
 from pds.registrysweepers.utils.db import write_updated_docs
 from pds.registrysweepers.utils.db.client import get_opensearch_client
-from pds.registrysweepers.utils.db.host import Host
 from pds.registrysweepers.utils.db.update import Update
 
 log = logging.getLogger(__name__)

src/pds/registrysweepers/ancestry/generation.py

@@ -12,7 +12,6 @@
 from pds.registrysweepers.ancestry.queries import get_collection_ancestry_records_bundles_query
 from pds.registrysweepers.ancestry.queries import get_collection_ancestry_records_collections_query
 from pds.registrysweepers.ancestry.queries import get_nonaggregate_ancestry_records_query
-from pds.registrysweepers.utils.db.host import Host
 from pds.registrysweepers.utils.misc import coerce_list_type
 from pds.registrysweepers.utils.productidentifiers.factory import PdsProductIdentifierFactory
 from pds.registrysweepers.utils.productidentifiers.pdslid import PdsLid

src/pds/registrysweepers/ancestry/queries.py

@@ -8,7 +8,6 @@
 
 from opensearchpy import OpenSearch
 from pds.registrysweepers.utils.db import query_registry_db_or_mock
-from pds.registrysweepers.utils.db.host import Host
 
 log = logging.getLogger(__name__)
 

src/pds/registrysweepers/provenance.py → src/pds/registrysweepers/provenance/__init__.py

@@ -53,7 +53,6 @@
 from pds.registrysweepers.utils.db import get_extant_lidvids
 from pds.registrysweepers.utils.db import write_updated_docs
 from pds.registrysweepers.utils.db.client import get_opensearch_client
-from pds.registrysweepers.utils.db.host import Host
 from pds.registrysweepers.utils.db.update import Update
 
 log = logging.getLogger(__name__)

src/pds/registrysweepers/repairkit/__init__.py

@@ -12,12 +12,14 @@
 from typing import Union
 
 from opensearchpy import OpenSearch
+from pds.registrysweepers.repairkit.versioning import SWEEPERS_REPAIRKIT_VERSION
 from pds.registrysweepers.utils import configure_logging
 from pds.registrysweepers.utils import parse_args
 from pds.registrysweepers.utils import query_registry_db
 from pds.registrysweepers.utils.db.client import get_opensearch_client
-from pds.registrysweepers.utils.db.host import Host
+from pds.registrysweepers.utils.db.indexing import ensure_index_mapping
 from pds.registrysweepers.utils.db.update import Update
+from pds.registrysweepers.utils.misc import get_sweeper_version_metadata_key
 
 from . import allarrays
 from ..utils.db import write_updated_docs
@@ -51,12 +53,14 @@ def function_name (document:{}, fieldname:str)->{}
 log = logging.getLogger(__name__)
 
 
-def generate_updates(docs: Iterable[Dict]) -> Iterable[Update]:
+def generate_updates(
+    docs: Iterable[Dict], repairkit_version_metadata_key: str, repairkit_version: int
+) -> Iterable[Update]:
     """Lazily generate necessary Update objects for a collection of db documents"""
     for document in docs:
         id = document["_id"]
         src = document["_source"]
-        repairs = {}
+        repairs = {repairkit_version_metadata_key: int(repairkit_version)}
         log.debug(f"applying repairkit sweeper to document: {id}")
         for fieldname, data in src.items():
             for regex, funcs in REPAIR_TOOLS.items():
@@ -65,7 +69,7 @@ def generate_updates(docs: Iterable[Dict]) -> Iterable[Update]:
                         repairs.update(func(src, fieldname))
 
         if repairs:
-            log.info(f"Writing repairs to document: {id}")
+            log.debug(f"Writing repairs to document: {id}")
             yield Update(id=id, content=repairs)
 
 
@@ -75,10 +79,19 @@ def run(
     log_level: int = logging.INFO,
 ):
     configure_logging(filepath=log_filepath, log_level=log_level)
-    log.info("Starting repairkit sweeper processing...")
+    log.info(f"Starting repairkit v{SWEEPERS_REPAIRKIT_VERSION} sweeper processing...")
 
-    all_docs = query_registry_db(client, {"query": {"match_all": {}}}, {})
-    updates = generate_updates(all_docs)
+    repairkit_version_metadata_key = get_sweeper_version_metadata_key("repairkit")
+
+    unprocessed_docs_query = {
+        "query": {
+            "bool": {"must_not": [{"range": {repairkit_version_metadata_key: {"gte": SWEEPERS_REPAIRKIT_VERSION}}}]}
+        }
+    }
+
+    all_docs = query_registry_db(client, unprocessed_docs_query, {})
+    updates = generate_updates(all_docs, repairkit_version_metadata_key, SWEEPERS_REPAIRKIT_VERSION)
+    ensure_index_mapping(client, "registry", repairkit_version_metadata_key, "integer")
     write_updated_docs(client, updates)
 
     log.info("Repairkit sweeper processing complete!")

src/pds/registrysweepers/repairkit/versioning.py

@@ -0,0 +1,5 @@
+# Defines constants used for versioning updated documents with the in-use version of sweepers
+# SWEEPERS_VERSION must be incremented any time sweepers is changed in a way which requires reprocessing of
+# previously-processed data
+SWEEPERS_REPAIRKIT_VERSION = 1
+SWEEPERS_REPAIRKIT_VERSION_METADATA_KEY = "ops:Provenance/ops:registry_sweepers_repairkit_version"

src/pds/registrysweepers/utils/__init__.py

@@ -5,7 +5,6 @@
 from typing import Union
 
 from pds.registrysweepers.utils.db import query_registry_db
-from pds.registrysweepers.utils.db.host import Host
 
 log = logging.getLogger(__name__)
 

src/pds/registrysweepers/utils/db/__init__.py

@@ -1,22 +1,16 @@
-import functools
 import json
 import logging
 import sys
-import urllib.parse
-from datetime import timedelta
 from typing import Callable
 from typing import Dict
 from typing import Iterable
 from typing import List
 from typing import Mapping
 from typing import Optional
 
-import requests
 from opensearchpy import OpenSearch
-from pds.registrysweepers.utils.db.host import Host
 from pds.registrysweepers.utils.db.update import Update
 from pds.registrysweepers.utils.misc import get_random_hex_id
-from requests import HTTPError
 from retry import retry
 from retry.api import retry_call
 
@@ -39,6 +33,7 @@ def query_registry_db(
     """
 
     scroll_keepalive = f"{scroll_keepalive_minutes}m"
+    request_timeout = 20
     query_id = get_random_hex_id()  # This is just used to differentiate queries during logging
     log.info(f"Initiating query with id {query_id}: {query}")
 
@@ -57,6 +52,7 @@ def fetch_func():
                     index=index_name,
                     body=query,
                     scroll=scroll_keepalive,
+                    request_timeout=request_timeout,
                     size=page_size,
                     _source_includes=_source.get("includes", []),  # TODO: Break out from the enclosing _source object
                     _source_excludes=_source.get("excludes", []),  # TODO: Break out from the enclosing _source object
@@ -65,7 +61,7 @@ def fetch_func():
         else:
 
             def fetch_func(_scroll_id: str = scroll_id):
-                return client.scroll(scroll_id=_scroll_id, scroll=scroll_keepalive)
+                return client.scroll(scroll_id=_scroll_id, scroll=scroll_keepalive, request_timeout=request_timeout)
 
         results = retry_call(
             fetch_func,
@@ -97,7 +93,7 @@ def fetch_func(_scroll_id: str = scroll_id):
         # expected total hits value.
         # TODO: Remove this upon implementation of https://github.com/NASA-PDS/registry-sweepers/issues/42
         hits_data_present_in_response = len(response_hits) > 0
-        if not hits_data_present_in_response:
+        if not hits_data_present_in_response and served_hits < total_hits:
             log.error(
                 f"Response for query {query_id} contained no hits when hits were expected.  Returned data is incomplete (got {served_hits} of {total_hits} total hits).  Response was: {results}"
             )
@@ -180,7 +176,8 @@ def update_as_statements(update: Update) -> Iterable[str]:
 def _write_bulk_updates_chunk(client: OpenSearch, index_name: str, bulk_updates: Iterable[str]):
     bulk_data = "\n".join(bulk_updates) + "\n"
 
-    response_content = client.bulk(index=index_name, body=bulk_data)
+    request_timeout = 60
+    response_content = client.bulk(index=index_name, body=bulk_data, request_timeout=request_timeout)
 
     if response_content.get("errors"):
         warn_types = {"document_missing_exception"}  # these types represent bad data, not bad sweepers behaviour

src/pds/registrysweepers/utils/majorminorversion.py

@@ -0,0 +1,48 @@
+from __future__ import annotations
+
+import functools
+
+
+@functools.total_ordering
+class MajorMinorVersion:
+    major_version_minimum = 0
+    minor_version_minimum = 0
+
+    def __init__(self, major_version: int, minor_version: int):
+        if major_version < self.major_version_minimum:
+            raise ValueError(f"major_version must be 0 or higher (got {major_version})")
+
+        if minor_version < self.minor_version_minimum:
+            raise ValueError(f"minor_version must be 0 or higher (got {minor_version})")
+
+        self.major_version = major_version
+        self.minor_version = minor_version
+
+    @classmethod
+    def from_string(cls, version_str: str):
+        major_version_chunk, minor_version_chunk = version_str.split(".")
+
+        major_version = int(major_version_chunk)
+        minor_version = int(minor_version_chunk)
+
+        return cls(major_version, minor_version)
+
+    def __str__(self):
+        return f"{self.major_version}.{self.minor_version}"
+
+    def __hash__(self):
+        return hash(str(self))
+
+    def __repr__(self):
+        return f"{self.__class__.__name__}({str(self)})"
+
+    def __eq__(self, other):
+        return self.major_version == other.major_version and self.minor_version == other.minor_version
+
+    def __lt__(self, other):
+        if self.major_version != other.major_version:
+            return self.major_version < other.major_version
+        elif self.minor_version != other.minor_version:
+            return self.minor_version < other.minor_version
+        else:
+            return False

src/pds/registrysweepers/utils/misc.py

@@ -47,3 +47,7 @@ def wrapped_f(*args, **kwargs):
         return resp
 
     return wrapped_f
+
+
+def get_sweeper_version_metadata_key(sweeper_name: str) -> str:
+    return f"ops:Provenance/ops:registry_sweepers_{sweeper_name}_version"

src/pds/registrysweepers/utils/productidentifiers/pdsvid.py

@@ -1,46 +1,6 @@
-from __future__ import annotations
+from pds.registrysweepers.utils.majorminorversion import MajorMinorVersion
 
-import functools
-import logging
 
-
-@functools.total_ordering
-class PdsVid:
-    def __init__(self, major_version: int, minor_version: int):
-        if major_version < 0:
-            raise ValueError(f"major_version must be 0 or higher (got {major_version})")
-
-        if minor_version < 0:
-            raise ValueError(f"minor_version must be 0 or higher (got {minor_version})")
-
-        self.major_version = major_version
-        self.minor_version = minor_version
-
-    @staticmethod
-    def from_string(vid_string: str) -> PdsVid:
-        major_version_chunk, minor_version_chunk = vid_string.split(".")
-
-        major_version = int(major_version_chunk)
-        minor_version = int(minor_version_chunk)
-
-        return PdsVid(major_version, minor_version)
-
-    def __str__(self):
-        return f"{self.major_version}.{self.minor_version}"
-
-    def __hash__(self):
-        return hash(str(self))
-
-    def __repr__(self):
-        return f"PdsVid({str(self)})"
-
-    def __eq__(self, other):
-        return self.major_version == other.major_version and self.minor_version == other.minor_version
-
-    def __lt__(self, other: PdsVid):
-        if self.major_version != other.major_version:
-            return self.major_version < other.major_version
-        elif self.minor_version != other.minor_version:
-            return self.minor_version < other.minor_version
-        else:
-            return False
+class PdsVid(MajorMinorVersion):
+    major_version_minimum = 0
+    minor_version_minimum = 0

tests/pds/registrysweepers/test_ancestry.py

@@ -8,7 +8,6 @@
 from pds.registrysweepers import ancestry
 from pds.registrysweepers.ancestry import AncestryRecord
 from pds.registrysweepers.ancestry import get_collection_ancestry_records
-from pds.registrysweepers.utils.db.host import Host
 from pds.registrysweepers.utils.productidentifiers.pdslidvid import PdsLidVid
 
 from tests.mocks.registryquerymock import RegistryQueryMock
@@ -282,9 +281,8 @@ class AncestryLegacyTypesTestCase(unittest.TestCase):
     registry_query_mock = RegistryQueryMock(input_file_path)
 
     def test_collection_refs_parsing(self):
-        host_stub = Host(None, None, None, None)
         query_mock_f = self.registry_query_mock.get_mocked_query
-        collection_ancestry_records = list(get_collection_ancestry_records(host_stub, query_mock_f))
+        collection_ancestry_records = list(get_collection_ancestry_records(None, query_mock_f))
 
         self.assertEqual(1, len(collection_ancestry_records))