[uss_qualifier/mock_uss] Add interaction logging functionality (#186)

* Adding interuss logging in mock_uss

* delete interuss logs

* Removing unwanted var

* Fixing lint issues

* Fixing lint issues

* fix return value

* fix lint issue

* Fixing the PR comments about renaming variables and files

* Renaming variable as per PR comment

* As per PR comments -
1. Switched to hook pattern for logging interactions
2. Removed default value for log dir, and response status code if dir not found
3. Fixed num of files deleted log

* As per PR comment - using json format to write interaction logs

* Fix as per PR comments -
1. Added a container for interaction_logging in docker compose
2. return 500 on Errors in GET /logs
3. interaction_logging specific config moved to its own package

* Suggest updates

* Add filter for incoming interaction logging

* Fixing an error found by CI

---------

Co-authored-by: Benjamin Pelletier <[email protected]>

monitoring/mock_uss/README.md

@@ -21,6 +21,7 @@ The available functionality sets are:
 * [`ridsp`](ridsp): Remote ID Service Provider
 * [`scdsc`](scdsc): ASTM F3548 strategic coordinator
 * [`tracer`](tracer): Interoperability ecosystem tracer logger
+* [`interaction_logging`](interaction_logging): Enables logging of the [interuss](https://github.com/astm-utm/Protocol/blob/master/utm.yaml) interactions between mock_uss and other uss participants
 
 
 ## Local deployment

monitoring/mock_uss/__init__.py

@@ -1,7 +1,6 @@
 import inspect
 import os
 from typing import Any, Optional, Callable
-
 from loguru import logger
 
 from monitoring.mock_uss.server import MockUSS
@@ -13,6 +12,7 @@
 SERVICE_MESSAGESIGNING = "msgsigning"
 SERVICE_ATPROXY_CLIENT = "atproxy_client"
 SERVICE_TRACER = "tracer"
+SERVICE_INTERACTION_LOGGING = "interaction_logging"
 
 webapp = MockUSS(__name__)
 enabled_services = set()
@@ -88,6 +88,11 @@ def require_config_value(config_key: str) -> None:
     from monitoring.mock_uss import msgsigning
     from monitoring.mock_uss.msgsigning import routes as msgsigning_routes
 
+if SERVICE_INTERACTION_LOGGING in webapp.config[config.KEY_SERVICES]:
+    enabled_services.add(SERVICE_INTERACTION_LOGGING)
+    from monitoring.mock_uss.interaction_logging import logger as interactions_logger
+    from monitoring.mock_uss.interaction_logging import routes_interactions_log
+
 if SERVICE_ATPROXY_CLIENT in webapp.config[config.KEY_SERVICES]:
     enabled_services.add(SERVICE_ATPROXY_CLIENT)
     from monitoring.mock_uss import atproxy_client

monitoring/mock_uss/docker-compose.yaml

@@ -18,14 +18,18 @@ services:
       - MOCK_USS_PUBLIC_KEY=/var/test-certs/auth2.pem
       - MOCK_USS_TOKEN_AUDIENCE=scdsc.uss1.localutm,localhost,host.docker.internal
       - MOCK_USS_BASE_URL=http://scdsc.uss1.localutm
-      - MOCK_USS_SERVICES=scdsc
+      # TODO: remove interaction_logging once dedicated mock_uss is involved in tests
+      - MOCK_USS_SERVICES=scdsc,interaction_logging
+      - MOCK_USS_INTERACTIONS_LOG_DIR=output/scdsc_a_interaction_logs
       - MOCK_USS_PORT=80
     expose:
       - 80
     ports:
       - 8074:80
     volumes:
       - ../../build/test-certs:/var/test-certs:ro
+      - ./output/scdsc_a_interaction_logs:/app/monitoring/mock_uss/output/scdsc_a_interaction_logs
+    user: "${UID_GID}"
     restart: always
     networks:
       - interop_ecosystem_network
@@ -212,6 +216,35 @@ services:
     extra_hosts:
       - host.docker.internal:host-gateway
 
+  mock_uss_scdsc_interaction_log:
+    container_name: mock_uss_scdsc_interaction_log
+    hostname: scdsc.log.uss6.localutm
+    image: interuss/monitoring
+    command: mock_uss/start.sh
+    environment:
+      - MOCK_USS_AUTH_SPEC=DummyOAuth(http://oauth.authority.localutm:8085/token,uss6)
+      - MOCK_USS_DSS_URL=http://dss.uss1.localutm
+      - MOCK_USS_PUBLIC_KEY=/var/test-certs/auth2.pem
+      - MOCK_USS_TOKEN_AUDIENCE=scdsc.log.uss6.localutm,localhost,host.docker.internal
+      - MOCK_USS_BASE_URL=http://scdsc.log.uss6.localutm
+      - MOCK_USS_SERVICES=scdsc,interaction_logging
+      - MOCK_USS_INTERACTIONS_LOG_DIR=output/scdsc_interaction_logs
+      - MOCK_USS_PORT=80
+    expose:
+      - 80
+    ports:
+      - 8095:80
+    volumes:
+      - ../../build/test-certs:/var/test-certs:ro
+      - ./output/scdsc_interaction_logs:/app/monitoring/mock_uss/output/scdsc_interaction_logs
+    user: "${UID_GID}"
+    restart: always
+    networks:
+      - interop_ecosystem_network
+    extra_hosts:
+      - host.docker.internal:host-gateway
+
+
   # atproxy is currently disabled to troubleshoot occasional connection issues
 #  atproxy:
 #    container_name: atproxy

monitoring/mock_uss/interaction_logging/__init__.py

@@ -0,0 +1 @@
+

monitoring/mock_uss/interaction_logging/config.py

@@ -0,0 +1,11 @@
+import os.path
+
+from monitoring.mock_uss import import_environment_variable, webapp
+
+KEY_INTERACTIONS_LOG_DIR = "MOCK_USS_INTERACTIONS_LOG_DIR"
+
+import_environment_variable(KEY_INTERACTIONS_LOG_DIR)
+
+_full_path = os.path.abspath(webapp.config[KEY_INTERACTIONS_LOG_DIR])
+if not os.path.exists(_full_path):
+    raise ValueError(f"MOCK_USS_INTERACTIONS_LOG_DIR {_full_path} does not exist")

monitoring/mock_uss/interaction_logging/interactions.py

@@ -0,0 +1,28 @@
+from enum import Enum
+from typing import List
+
+from implicitdict import ImplicitDict
+import yaml
+from yaml.representer import Representer
+
+from monitoring.monitorlib.fetch import Query
+
+
+class QueryDirection(str, Enum):
+    Incoming = "Incoming"
+    """The query originated from a remote client and was handled by the system reporting the interaction."""
+
+    Outgoing = "Outgoing"
+    """The system reporting the interaction initiated the query to a remote server."""
+
+
+class Interaction(ImplicitDict):
+    query: Query
+    direction: QueryDirection
+
+
+yaml.add_representer(Interaction, Representer.represent_dict)
+
+
+class ListLogsResponse(ImplicitDict):
+    interactions: List[Interaction]

monitoring/mock_uss/interaction_logging/logger.py

@@ -0,0 +1,72 @@
+import os
+import datetime
+
+import flask
+import json
+from loguru import logger
+
+from monitoring.mock_uss import webapp, require_config_value
+from monitoring.mock_uss.interaction_logging.config import KEY_INTERACTIONS_LOG_DIR
+from monitoring.mock_uss.interaction_logging.interactions import (
+    Interaction,
+    QueryDirection,
+)
+from monitoring.monitorlib.clients import QueryHook, query_hooks
+from monitoring.monitorlib.fetch import Query, describe_flask_query, QueryType
+
+require_config_value(KEY_INTERACTIONS_LOG_DIR)
+
+
+def log_interaction(direction: QueryDirection, query: Query) -> None:
+    """Logs the REST calls between Mock USS to SUT
+    Args:
+        direction: Whether this interaction was initiated or handled by this system.
+        query: Full description of the interaction to log.
+    """
+    interaction: Interaction = Interaction(query=query, direction=direction)
+    method = query.request.method
+    log_file(f"{direction}_{method}", interaction)
+
+
+def log_file(code: str, content: Interaction) -> None:
+    log_path = webapp.config[KEY_INTERACTIONS_LOG_DIR]
+    n = len(os.listdir(log_path))
+    basename = "{:06d}_{}_{}".format(
+        n, code, datetime.datetime.now().strftime("%H%M%S_%f")
+    )
+    logname = "{}.json".format(basename)
+    fullname = os.path.join(log_path, logname)
+
+    with open(fullname, "w") as f:
+        json.dump(content, f)
+
+
+class InteractionLoggingHook(QueryHook):
+    def on_query(self, query: Query) -> None:
+        # TODO: Make this configurable instead of hardcoding exactly these query types
+        if "query_type" in query and query.query_type in {
+            QueryType.F3548v21USSGetOperationalIntentDetails,
+            QueryType.F3548v21USSNotifyOperationalIntentDetailsChanged,
+        }:
+            log_interaction(QueryDirection.Outgoing, query)
+
+
+query_hooks.append(InteractionLoggingHook())
+
+
+# https://stackoverflow.com/a/67856316
+@webapp.before_request
+def interaction_log_before_request():
+    flask.Flask.custom_profiler = {"start": datetime.datetime.utcnow()}
+
+
+@webapp.after_request
+def interaction_log_after_request(response):
+    elapsed_s = (
+        datetime.datetime.utcnow() - flask.current_app.custom_profiler["start"]
+    ).total_seconds()
+    # TODO: Make this configurable instead of hardcoding exactly these query types
+    if "/uss/v1/" in flask.request.url_rule.rule:
+        query = describe_flask_query(flask.request, response, elapsed_s)
+        log_interaction(QueryDirection.Incoming, query)
+    return response

monitoring/mock_uss/interaction_logging/routes_interactions_log.py

@@ -0,0 +1,84 @@
+from typing import Tuple, List
+import json
+import os
+from implicitdict import ImplicitDict, StringBasedDateTime
+from flask import request, jsonify
+from loguru import logger
+
+from monitoring.mock_uss import webapp
+from monitoring.mock_uss.auth import requires_scope
+from monitoring.monitorlib.scd_automated_testing.scd_injection_api import (
+    SCOPE_SCD_QUALIFIER_INJECT,
+)
+from monitoring.mock_uss.interaction_logging.interactions import (
+    Interaction,
+    ListLogsResponse,
+)
+
+from monitoring.mock_uss.interaction_logging.config import KEY_INTERACTIONS_LOG_DIR
+
+
+@webapp.route("/mock_uss/interuss_logging/logs", methods=["GET"])
+@requires_scope([SCOPE_SCD_QUALIFIER_INJECT])
+def interaction_logs() -> Tuple[str, int]:
+    """
+    Returns all the interaction logs with requests that were
+    received or initiated between 'from_time' and now
+    Eg - http:/.../mock_uss/interuss/log?from_time=2023-08-30T20:48:21.900000Z
+    """
+    from_time_param = request.args.get("from_time", "1900-01-01T00:00:00Z")
+    from_time = StringBasedDateTime(from_time_param)
+    log_path = webapp.config[KEY_INTERACTIONS_LOG_DIR]
+
+    if not os.path.exists(log_path):
+        raise ValueError(f"Configured log path {log_path} does not exist")
+
+    interactions: List[Interaction] = []
+    for fname in os.listdir(log_path):
+        with open(os.path.join(log_path, fname), "r") as f:
+            try:
+                obj = json.load(f)
+                interaction = ImplicitDict.parse(obj, Interaction)
+                if (
+                    ("received_at" in interaction.query.request)
+                    and interaction.query.request.received_at.datetime
+                    >= from_time.datetime
+                ):
+                    interactions.append(interaction)
+                elif (
+                    "initiated_at" in interaction.query.request
+                    and interaction.query.request.initiated_at.datetime
+                    >= from_time.datetime
+                ):
+                    interactions.append(interaction)
+                else:
+                    raise ValueError(
+                        f"There is no received_at or initiated_at field in the request in {fname}"
+                    )
+
+            except (KeyError, ValueError) as e:
+                msg = f"Error occurred in reading interaction from file {fname}: {e}"
+                raise type(e)(msg)
+
+    return jsonify(ListLogsResponse(interactions=interactions)), 200
+
+
+@webapp.route("/mock_uss/interuss_logging/logs", methods=["DELETE"])
+@requires_scope([SCOPE_SCD_QUALIFIER_INJECT])
+def delete_interaction_logs() -> Tuple[str, int]:
+    """Deletes all the files under the logging directory"""
+    log_path = webapp.config[KEY_INTERACTIONS_LOG_DIR]
+
+    if not os.path.exists(log_path):
+        raise ValueError(f"Configured log path {log_path} does not exist")
+
+    logger.debug(f"Number of files in {log_path}: {len(os.listdir(log_path))}")
+
+    num_removed = 0
+    for file in os.listdir(log_path):
+        file_path = os.path.join(log_path, file)
+        os.remove(file_path)
+        logger.debug(f"Removed log file - {file_path}")
+        num_removed = num_removed + 1
+
+    return f"Removed {num_removed} files", 200

monitoring/mock_uss/run_locally.sh

@@ -31,15 +31,24 @@ elif [[ "$DC_COMMAND" == "debug" ]]; then
   export DEBUG_ON=1
 fi
 
-mkdir -p output/tracer
 UID_GID="$(id -u):$(id -g)"
 export UID_GID
 echo "DC_COMMAND is ${DC_COMMAND}"
-if [[ "$DC_COMMAND" == up* ]]; then
-  echo "Cleaning up past tracer logs"
-  # Prevent logs from building up too much by default
-  find "output/tracer" -name "*.yaml" -exec rm {} \;
-fi
+
+declare -a log_folders=( \
+    "output/tracer" \
+    "output/scdsc_a_interaction_logs" \
+    "output/scdsc_interaction_logs" \
+  )
+for log_folder in "${log_folders[@]}"; do
+  mkdir -p "$log_folder"
+  if [[ "$DC_COMMAND" == up* ]]; then
+    echo "Cleaning up past logs in $log_folder"
+    # Prevent logs from building up too much by default
+    find "$log_folder" -name "*.yaml" -exec rm {} \;
+    find "$log_folder" -name "*.json" -exec rm {} \;
+  fi
+done
 
 # shellcheck disable=SC2086
 docker compose -f docker-compose.yaml -p mocks $DC_COMMAND $DC_OPTIONS

monitoring/mock_uss/scdsc/routes_injection.py

@@ -85,11 +85,10 @@ def query_operational_intents(
 
     updated_op_intents = []
     for op_intent_ref in get_details_for:
-        updated_op_intents.append(
-            scd_client.get_operational_intent_details(
-                utm_client, op_intent_ref.uss_base_url, op_intent_ref.id
-            )
+        op_intent, _ = scd_client.get_operational_intent_details(
+            utm_client, op_intent_ref.uss_base_url, op_intent_ref.id
         )
+        updated_op_intents.append(op_intent)
     result.extend(updated_op_intents)
 
     with db as tx:

monitoring/monitorlib/clients/__init__.py

@@ -0,0 +1,22 @@
+from abc import ABC
+from typing import List
+
+from monitoring.monitorlib.fetch import Query
+
+
+class QueryHook(ABC):
+    def on_query(self, query: Query) -> None:
+        """Called whenever a client performs a query and this hook is included in query_hooks.
+
+        Args:
+            query: Query that was performed.
+        """
+        raise NotImplementedError("QueryHook subclass did not implement on_query")
+
+
+query_hooks: List[QueryHook] = []
+
+
+def call_query_hooks(query: Query) -> None:
+    for hook in query_hooks:
+        hook.on_query(query=query)