Operations cleanup (#3589)

contrib/server/operations/pbench-upload-results.py

@@ -34,6 +34,7 @@
 import time
 from typing import Optional
 
+import bs4
 import dateutil.parser
 import requests
 
@@ -246,7 +247,7 @@ def main() -> int:
         print("Uploading target files...")
     success = 0
     failure = 0
-    failures = set()
+    failures = []
     duplicate = 0
     for t in which:
         try:
@@ -268,24 +269,25 @@ def main() -> int:
                         print(t, file=checkwriter, flush=True)
                 else:
                     failure += 1
-                    failures.add(response.status_code)
 
-                    # TODO: can we handle NGINX's ugly 500 "storage error"
-                    # gracefully somehow?
                     if response.headers["content-type"] in (
                         "text/html",
                         "text/xml",
                         "application/xml",
-                        "text/plain",
                     ):
-                        message = re.sub(r"[\n\s]+", " ", response.text)
+                        try:
+                            m = bs4.BeautifulSoup(response.text, features="html.parser")
+                            message = m.title.text
+                        except Exception:
+                            message = re.sub(r"[\n\s]+", " ", response.text)
                     elif response.headers["content-type"] == "application/json":
                         try:
-                            message = response.json()
+                            message = response.json()["message"]
                         except Exception:
                             message = response.text
                     else:
                         message = f"{response.headers['content-type']}({response.text})"
+                    failures.append(f" {response.status_code}\t{t.name}\t{message}")
                     print(
                         f"Upload of {t} failed: {response.status_code} ({message})",
                         file=sys.stderr,
@@ -298,8 +300,10 @@ def main() -> int:
         checkwriter.close()
 
     print(
-        f"Uploaded {success} successfully; {duplicate} duplicates, {failure} failures: {failures}"
+        f"Uploaded {success} successfully; {duplicate} duplicates, {failure} failures:"
     )
+    for f in failures:
+        print(f)
     return 0
 
 

lib/pbench/server/cache_manager.py

@@ -1422,6 +1422,9 @@ def create(self, tarfile_path: Path) -> Tarball:
         controller directory. The controller directory will be created if
         necessary.
 
+        Datasets without an identifiable controller will be assigned to
+        "unknown".
+
         Args:
             tarfile_path: dataset tarball path
 
@@ -1435,20 +1438,24 @@ def create(self, tarfile_path: Path) -> Tarball:
         Returns
             Tarball object
         """
+        if not tarfile_path.is_file():
+            raise BadFilename(tarfile_path)
+        name = Dataset.stem(tarfile_path)
+        controller_name = None
         try:
             metadata = Tarball._get_metadata(tarfile_path)
-            if metadata:
-                controller_name = metadata["run"]["controller"]
-            else:
-                controller_name = "unknown"
+            controller_name = metadata["run"]["controller"]
         except Exception as exc:
-            raise MetadataError(tarfile_path, exc)
+            self.logger.warning(
+                "{} metadata.log is missing run.controller: {!r}", name, exc
+            )
 
         if not controller_name:
-            raise MetadataError(tarfile_path, ValueError("no controller value"))
-        if not tarfile_path.is_file():
-            raise BadFilename(tarfile_path)
-        name = Dataset.stem(tarfile_path)
+            controller_name = "unknown"
+            self.logger.warning(
+                "{} has no controller name, assuming {!r}", name, controller_name
+            )
+
         if name in self.tarballs:
             raise DuplicateTarball(name)
         if controller_name in self.controllers:

lib/pbench/server/indexing_tarballs.py

@@ -95,6 +95,8 @@ class Index:
     was a bash script.
     """
 
+    BATCH_SIZE = 100  # Number of READY datasets to grab at once
+
     error_code = Errors(
         ErrorCode("OK", 0, None, "Successful completion"),
         ErrorCode("OP_ERROR", 1, False, "Operational error while indexing"),
@@ -176,7 +178,7 @@ def collect_tb(self) -> Tuple[int, List[TarballData]]:
         idxctx = self.idxctx
         error_code = self.error_code
         try:
-            for dataset in self.sync.next():
+            for dataset in self.sync.next(count=self.BATCH_SIZE):
                 tb = Metadata.getvalue(dataset, Metadata.TARBALL_PATH)
                 if not tb:
                     self.sync.error(dataset, "Dataset does not have a tarball-path")

lib/pbench/server/sync.py

@@ -41,7 +41,7 @@ def __init__(self, logger: Logger, component: OperationName):
     def __str__(self) -> str:
         return f"<Synchronizer for component {self.component.name!r}>"
 
-    def next(self) -> list[Dataset]:
+    def next(self, count: Optional[int] = None) -> list[Dataset]:
         """
         This is a specialized query to return a list of datasets with the READY
         OperationState for the Sync component.
@@ -59,6 +59,9 @@ def next(self) -> list[Dataset]:
         transporting only the resource IDs across the barrier and fetching
         new proxy objects using the general session.
 
+        Args:
+            count: Limit the number of returned matches
+
         Returns:
             A list of Dataset objects which have an associated
             Operation object in READY state.
@@ -71,6 +74,8 @@ def next(self) -> list[Dataset]:
                     Operation.state == OperationState.READY,
                 )
                 query = query.order_by(Dataset.resource_id)
+                if count:
+                    query = query.limit(count)
                 Database.dump_query(query, self.logger)
                 id_list = [d.resource_id for d in query.all()]
             return [Dataset.query(resource_id=i) for i in id_list]

lib/pbench/test/unit/server/test_cache_manager.py

@@ -24,7 +24,6 @@
     Inventory,
     LockManager,
     LockRef,
-    MetadataError,
     Tarball,
     TarballModeChangeError,
     TarballNotFound,
@@ -167,49 +166,38 @@ def test_clean_empties(self, server_config, make_logger):
         for c in controllers:
             assert not (cm.archive_root / c).exists()
 
+    @pytest.mark.parametrize(
+        "metadata",
+        (
+            {"pbench": {"date": "2002-05-16T00:00:00"}},
+            {"pbench": {"date": "2002-05-16T00:00:00"}, "run": {}},
+            {
+                "pbench": {"date": "2002-05-16T00:00:00"},
+                "run": {"controller": ""},
+            },
+        ),
+    )
     def test_metadata(
-        self, monkeypatch, selinux_disabled, server_config, make_logger, tarball
+        self,
+        monkeypatch,
+        db_session,
+        selinux_disabled,
+        server_config,
+        make_logger,
+        tarball,
+        metadata,
     ):
         """Test behavior with metadata.log access errors."""
 
-        def fake_metadata(_tar_path):
-            return {"pbench": {"date": "2002-05-16T00:00:00"}}
-
-        def fake_metadata_run(_tar_path):
-            return {"pbench": {"date": "2002-05-16T00:00:00"}, "run": {}}
-
-        def fake_metadata_controller(_tar_path):
-            return {
-                "pbench": {"date": "2002-05-16T00:00:00"},
-                "run": {"controller": ""},
-            }
-
         # fetching metadata from metadata.log file and key/value not
-        # being there should result in a MetadataError
+        # being there should result in assuming an "unknown" controller
         source_tarball, source_md5, md5 = tarball
         cm = CacheManager(server_config, make_logger)
 
-        expected_metaerror = f"A problem occurred processing metadata.log from {source_tarball!s}: \"'run'\""
-        with monkeypatch.context() as m:
-            m.setattr(Tarball, "_get_metadata", fake_metadata)
-            with pytest.raises(MetadataError) as exc:
-                cm.create(source_tarball)
-            assert str(exc.value) == expected_metaerror
-
-        expected_metaerror = f"A problem occurred processing metadata.log from {source_tarball!s}: \"'controller'\""
-        with monkeypatch.context() as m:
-            m.setattr(Tarball, "_get_metadata", fake_metadata_run)
-            with pytest.raises(MetadataError) as exc:
-                cm.create(source_tarball)
-            assert str(exc.value) == expected_metaerror
-
-        expected_metaerror = "A problem occurred processing metadata.log "
-        expected_metaerror += f"from {source_tarball!s}: 'no controller value'"
         with monkeypatch.context() as m:
-            m.setattr(Tarball, "_get_metadata", fake_metadata_controller)
-            with pytest.raises(MetadataError) as exc:
-                cm.create(source_tarball)
-            assert str(exc.value) == expected_metaerror
+            m.setattr(Tarball, "_get_metadata", lambda p: metadata)
+            tarball = cm.create(source_tarball)
+            assert tarball.controller_name == "unknown"
 
     def test_with_metadata(
         self,

lib/pbench/test/unit/server/test_indexing_tarballs.py

@@ -248,7 +248,7 @@ def __init__(self, logger: Logger, component: OperationName):
         self.logger = logger
         self.component = component
 
-    def next(self) -> List[Dataset]:
+    def next(self, count: Optional[int] = None) -> List[Dataset]:
         __class__.called.append(f"next-{self.component.name}")
         assert self.component in __class__.tarballs
         return __class__.tarballs[self.component]