Unit testing & review comments (distributed-system-analysis#3608)

* Unit testing & review comments

PBENCH-1315

The index-map fix was submitted without unit tests so we can get it deployed
to resolve the immediate PostgreSQL resource problems. This follow-on adds
unit testing (and some late review comments).

This proved an "interesting" journey. I found several returns that don't really
make sense: for example, the old delete/update logic reported a `Sync.update`
failure as `CONFLICT` rather than `INTERNAL_SERVER_ERROR`. The fixture we use
install the Elasticsearch mock response and call the API needs to know whether
to install the mock: it has to be right as either installing the mock and not
calling it or calling it without the mock will result in failure. The current
logic around installation of the mock proved difficult to reconcile with the
need to handle `INTERNAL_SERVER_ERROR` both *before* and *after* the call to
Elasticsearch. As a result, I ended up massively refactoring the `query_api`
fixture code to simplify the logic based on a three-state override (neutral,
force, or suppress) in the new `expect_call` parameter.

Incidentally, after the production server PostgreSQL recovery, I found that my
report generator was tripping over datasets which had no operational status: I
suspect this was due to some hole in intake (although I think that failure to
create the `UPLOAD` operational status should have failed the intake, I'm not
going to debug that today). I did, however, adjust the report generator to
detect and report this case gracefully instead of failing with an f-string
formatting error. I also added a summary line of rows, tables, and size for
the SQL report.

Author: dbutenhof

lib/pbench/cli/server/report.py

@@ -295,13 +295,17 @@ def report_sql():
     """Report the SQL table storage statistics"""
 
     watcher.update("inspecting SQL tables")
+    table_count = 0
+    row_count = 0
+    row_size = 0
     click.echo("SQL storage report:")
     t_w = 20
     r_w = 10
     s_w = 10
     click.echo(f"  {'Table':<{t_w}} {'Rows':<{r_w}} {'Storage':<{s_w}}")
     click.echo(f"  {'':-<{t_w}} {'':-<{r_w}} {'':-<{s_w}}")
     for t in inspect(Database.db_session.get_bind()).get_table_names():
+        table_count += 1
         (rows,) = next(
             Database.db_session.execute(statement=text(f"SELECT COUNT(*) FROM {t}"))
         )
@@ -312,6 +316,11 @@ def report_sql():
             )
         )
         click.echo(f"  {t:<{t_w}} {rows:>{r_w},d} {humanize.naturalsize(size):>{s_w}}")
+        row_count += rows
+        row_size += size
+    click.echo(
+        f"  Total of {row_count:,d} rows in {table_count:,d} tables, consuming {humanize.naturalsize(row_size)}"
+    )
 
     if not detailer:
         return
@@ -357,6 +366,7 @@ def report_states():
     index_pattern: re.Pattern = re.compile(r"^(\d+):(.*)$")
     index_errors = defaultdict(int)
     index_messages = defaultdict(str)
+    ops_anomalies = 0
     operations = defaultdict(lambda: defaultdict(int))
     rows = Database.db_session.execute(
         statement=text(
@@ -366,22 +376,26 @@ def report_states():
     )
     for dataset, operation, state, message in rows:
         watcher.update(f"inspecting {dataset}:{operation}")
-        operations[operation][state] += 1
-        if state == "FAILED":
-            detailer.error(f"{operation} {state} for {dataset}: {message!r}")
-            if operation == "INDEX":
-                match = index_pattern.match(message)
-                if match:
-                    try:
-                        code = int(match.group(1))
-                        message = match.group(2)
-                        index_errors[code] += 1
-                        if code not in index_messages:
-                            index_messages[code] = message
-                    except Exception as e:
-                        detailer.error(
-                            f"{dataset} unexpected 'INDEX' error {message}: {str(e)!r}"
-                        )
+        if operation is None:
+            ops_anomalies += 1
+            detailer.error(f"{dataset} doesn't have operational state")
+        else:
+            operations[operation][state] += 1
+            if state == "FAILED":
+                detailer.error(f"{operation} {state} for {dataset}: {message!r}")
+                if operation == "INDEX":
+                    match = index_pattern.match(message)
+                    if match:
+                        try:
+                            code = int(match.group(1))
+                            message = match.group(2)
+                            index_errors[code] += 1
+                            if code not in index_messages:
+                                index_messages[code] = message
+                        except Exception as e:
+                            detailer.error(
+                                f"{dataset} unexpected 'INDEX' error {message}: {str(e)!r}"
+                            )
     click.echo("Operational states:")
     for name, states in operations.items():
         click.echo(f"  {name} states:")
@@ -392,6 +406,8 @@ def report_states():
                     click.echo(
                         f"           CODE {code:2d}: {count:>6,d}  {index_messages[code]}"
                     )
+    if ops_anomalies:
+        click.echo(f"  {ops_anomalies} datasets are missing operational state")
 
 
 @click.command(name="pbench-report-generator")

lib/pbench/server/api/resources/query_apis/datasets/__init__.py

@@ -9,32 +9,13 @@
     ApiParams,
     ApiSchema,
     ParamType,
-    SchemaError,
 )
 from pbench.server.api.resources.query_apis import ElasticBase
 from pbench.server.database.models.datasets import Dataset, Metadata
 from pbench.server.database.models.index_map import IndexMap
 from pbench.server.database.models.templates import Template
 
 
-class MissingDatasetNameParameter(SchemaError):
-    """The subclass schema is missing the required "name" parameter required
-    to locate a Dataset.
-
-    NOTE: This is a development error, not a client error, and will be raised
-    when the API is initialized at server startup. Arguably, this could be an
-    assert since it prevents launching the server.
-    """
-
-    def __init__(self, subclass_name: str, message: str):
-        super().__init__()
-        self.subclass_name = subclass_name
-        self.message = message
-
-    def __str__(self) -> str:
-        return f"API {self.subclass_name} is {self.message}"
-
-
 class IndexMapBase(ElasticBase):
     """A base class for query apis that depends on the index map.
 
@@ -133,17 +114,17 @@ def get_index(
         """
 
         archive_only = Metadata.getvalue(dataset, Metadata.SERVER_ARCHIVE)
-        if archive_only and ok_no_index:
-            return ""
-
         if archive_only:
+            if ok_no_index:
+                return ""
             raise APIAbort(HTTPStatus.CONFLICT, "Dataset indexing was disabled")
 
         index_keys = list(IndexMap.indices(dataset, root_index_name))
 
         if not index_keys:
             raise APIAbort(
-                HTTPStatus.NOT_FOUND, f"Dataset has no {root_index_name!r} data"
+                HTTPStatus.NOT_FOUND,
+                f"Dataset has no {root_index_name if root_index_name else 'indexed'!r} data",
             )
 
         indices = ",".join(index_keys)

lib/pbench/server/api/resources/query_apis/datasets/datasets.py

@@ -1,3 +1,4 @@
+from collections import defaultdict
 from http import HTTPStatus
 
 from flask import current_app, jsonify, Response
@@ -16,8 +17,9 @@
     ParamType,
     Schema,
 )
-from pbench.server.api.resources.query_apis import ApiContext, PostprocessError
+from pbench.server.api.resources.query_apis import ApiContext
 from pbench.server.api.resources.query_apis.datasets import IndexMapBase
+import pbench.server.auth.auth as Auth
 from pbench.server.cache_manager import CacheManager
 from pbench.server.database.models.audit import AuditReason, AuditStatus, AuditType
 from pbench.server.database.models.datasets import (
@@ -120,14 +122,13 @@ def assemble(self, params: ApiParams, context: ApiContext) -> JSONOBJECT:
 
         dataset = context["dataset"]
         action = context["attributes"].action
-        get_action = action == "get"
         context["action"] = action
         audit_attributes = {}
         access = None
         owner = None
         elastic_options = {"ignore_unavailable": "true"}
 
-        if not get_action:
+        if action != "get":
             elastic_options["refresh"] = "true"
             operations = Operation.by_state(dataset, OperationState.WORKING)
             if context["attributes"].require_stable and operations:
@@ -147,7 +148,9 @@ def assemble(self, params: ApiParams, context: ApiContext) -> JSONOBJECT:
                     OperationState.WORKING.name,
                     e,
                 )
-                raise APIAbort(HTTPStatus.CONFLICT, "Unable to set operational state")
+                raise APIInternalError(
+                    f"can't set {OperationState.WORKING.name} on {dataset.name}: {str(e)!r} "
+                )
             context["sync"] = sync
             context["auditing"]["attributes"] = audit_attributes
             if action == "update":
@@ -156,6 +159,14 @@ def assemble(self, params: ApiParams, context: ApiContext) -> JSONOBJECT:
                 if not access and not owner:
                     raise MissingParameters(["access", "owner"])
 
+                if owner and owner != dataset.owner_id:
+                    auth_user = Auth.token_auth.current_user()
+                    if not auth_user.is_admin():
+                        raise APIAbort(
+                            HTTPStatus.FORBIDDEN,
+                            "ADMIN role is required to change dataset ownership",
+                        )
+
                 if access:
                     audit_attributes["access"] = access
                     context["access"] = access
@@ -167,12 +178,16 @@ def assemble(self, params: ApiParams, context: ApiContext) -> JSONOBJECT:
                 else:
                     owner = dataset.owner_id
 
-        # Get the Elasticsearch indices occupied by the dataset. If there are
-        # none, return with an empty query to disable the Elasticsearch call.
+        # Get the Elasticsearch indices occupied by the dataset.
+        #
+        # We postprocess UPDATE and DELETE even without any indexed documents
+        # in order to update the Dataset object, so tell get_index not to fail
+        # in that case, and return an empty query to disable the Elasticsearch
+        # call.
         #
         # It's important that all context fields required for postprocessing
         # of unindexed datasets have been set before this!
-        indices = self.get_index(dataset, ok_no_index=(not get_action))
+        indices = self.get_index(dataset, ok_no_index=(action != "get"))
         context["indices"] = indices
         if not indices:
             return {}
@@ -195,11 +210,12 @@ def assemble(self, params: ApiParams, context: ApiContext) -> JSONOBJECT:
         }
 
         if action == "update":
-            painless = "ctx._source.authorization=params.authorization"
-            script_params = {"authorization": {"access": access, "owner": owner}}
-            script = {"source": painless, "lang": "painless", "params": script_params}
             json["path"] = f"{indices}/_update_by_query"
-            json["kwargs"]["json"]["script"] = script
+            json["kwargs"]["json"]["script"] = {
+                "source": "ctx._source.authorization=params.authorization",
+                "lang": "painless",
+                "params": {"authorization": {"access": access, "owner": owner}},
+            }
         elif action == "get":
             json["path"] = f"{indices}/_search"
         elif action == "delete":
@@ -232,32 +248,31 @@ def postprocess(self, es_json: JSONOBJECT, context: ApiContext) -> Response:
         current_app.logger.info("POSTPROCESS {}: {}", dataset.name, es_json)
         failures = 0
         if action == "get":
+            count = None
+            hits = []
             try:
                 count = es_json["hits"]["total"]["value"]
+                hits = es_json["hits"]["hits"]
                 if int(count) == 0:
-                    current_app.logger.info("No data returned by Elasticsearch")
-                    return jsonify([])
+                    raise APIInternalError(
+                        f"Elasticsearch returned no matches for {dataset.name}"
+                    )
             except KeyError as e:
-                raise PostprocessError(
-                    HTTPStatus.BAD_REQUEST,
-                    f"Can't find Elasticsearch match data {e} in {es_json!r}",
+                raise APIInternalError(
+                    f"Can't find Elasticsearch match data for {dataset.name} ({e}) in {es_json!r}",
                 )
             except ValueError as e:
-                raise PostprocessError(
-                    HTTPStatus.BAD_REQUEST,
-                    f"Elasticsearch hit count {count!r} value: {e}",
+                raise APIInternalError(
+                    f"Elasticsearch bad hit count {count!r} for {dataset.name}: {e}",
                 )
-            results = []
-            for hit in es_json["hits"]["hits"]:
-                s = hit["_source"]
-                s["id"] = hit["_id"]
-                results.append(s)
-
+            results = defaultdict(int)
+            for hit in hits:
+                results[hit["_index"]] += 1
             return jsonify(results)
         else:
             if es_json:
                 fields = ("deleted", "updated", "total", "version_conflicts")
-                results = {f: es_json[f] if f in es_json else None for f in fields}
+                results = {f: es_json.get(f, 0) for f in fields}
                 failures = len(es_json["failures"]) if "failures" in es_json else 0
                 results["failures"] = failures
             else:
@@ -282,7 +297,7 @@ def postprocess(self, es_json: JSONOBJECT, context: ApiContext) -> Response:
                         dataset.update()
                     except Exception as e:
                         raise APIInternalError(
-                            f"unable to update dataset {dataset.name}: {str(e)!r}"
+                            f"Unable to update dataset {dataset.name}: {str(e)!r}"
                         ) from e
                 elif action == "delete":
                     try:
@@ -295,7 +310,7 @@ def postprocess(self, es_json: JSONOBJECT, context: ApiContext) -> Response:
                         del context["sync"]
                     except Exception as e:
                         raise APIInternalError(
-                            f"unable to delete dataset {dataset.name}: {str(e)!r}"
+                            f"Unable to delete dataset {dataset.name}: {str(e)!r}"
                         ) from e
 
             # The DELETE API removes the "sync" context on success to signal
@@ -312,7 +327,9 @@ def postprocess(self, es_json: JSONOBJECT, context: ApiContext) -> Response:
                     auditing["attributes"] = {"message": str(e)}
                     auditing["status"] = AuditStatus.WARNING
                     auditing["reason"] = AuditReason.INTERNAL
-                    raise APIInternalError(f"Unexpected sync unlock error '{e}'") from e
+                    raise APIInternalError(
+                        f"Unexpected sync error {dataset.name} {str(e)!r}"
+                    ) from e
 
             # Return the summary document as the success response, or abort with an
             # internal error if we tried to operate on Elasticsearch documents but
@@ -323,9 +340,9 @@ def postprocess(self, es_json: JSONOBJECT, context: ApiContext) -> Response:
                 auditing["reason"] = AuditReason.INTERNAL
                 auditing["attributes"][
                     "message"
-                ] = f"Unable to {context['attributes'].action} some indexed documents"
+                ] = f"Unable to {action} some indexed documents"
                 raise APIInternalError(
-                    f"Failed to {context['attributes'].action} any of {results['total']} "
+                    f"Failed to {action} any of {results['total']} "
                     f"Elasticsearch documents: {es_json}"
                 )
             elif sync:

lib/pbench/server/api/resources/query_apis/datasets/datasets_detail.py

@@ -6,21 +6,17 @@
 from pbench.server.api.resources import (
     APIAbort,
     ApiAuthorizationType,
+    APIInternalError,
     ApiMethod,
     ApiParams,
     ApiSchema,
     Parameter,
     ParamType,
     Schema,
 )
-from pbench.server.api.resources.query_apis import ApiContext, PostprocessError
+from pbench.server.api.resources.query_apis import ApiContext
 from pbench.server.api.resources.query_apis.datasets import IndexMapBase
-from pbench.server.database.models.datasets import (
-    Dataset,
-    DatasetNotFound,
-    Metadata,
-    MetadataError,
-)
+from pbench.server.database.models.datasets import Metadata, MetadataError
 
 
 class DatasetsDetail(IndexMapBase):
@@ -122,18 +118,17 @@ def postprocess(self, es_json: JSONOBJECT, context: ApiContext) -> Response:
             }
         ]
         """
+
+        dataset = context["dataset"]
         hits = es_json["hits"]["hits"]
 
         # NOTE: we're expecting just one. We're matching by just the
-        # dataset name, which ought to be unique.
+        # dataset resource ID, which ought to be unique.
         if len(hits) == 0:
-            raise PostprocessError(
-                HTTPStatus.BAD_REQUEST, "The specified dataset has gone missing"
-            )
+            raise APIInternalError(f"Dataset {dataset.name} run document is missing")
         elif len(hits) > 1:
-            raise PostprocessError(
-                HTTPStatus.BAD_REQUEST, "Too many hits for a unique query"
-            )
+            raise APIInternalError(f"Dataset {dataset.name} has multiple run documents")
+
         src = hits[0]["_source"]
 
         # We're merging the "run" and "@metadata" sub-documents into
@@ -147,12 +142,7 @@ def postprocess(self, es_json: JSONOBJECT, context: ApiContext) -> Response:
         }
 
         try:
-            dataset = Dataset.query(resource_id=(src["run"]["id"]))
             m = self._get_dataset_metadata(dataset, context["metadata"])
-        except DatasetNotFound:
-            raise APIAbort(
-                HTTPStatus.BAD_REQUEST, f"Dataset {src['run']['id']} not found"
-            )
         except MetadataError as e:
             raise APIAbort(HTTPStatus.BAD_REQUEST, str(e))