feat: Adds fragment exporter tool | NPG-0000 (#546)

# Description

Please include a summary of the changes and the related issue.
Please also include relevant motivation and context.
List any dependencies that are required for this change.

Fixes # (issue) _if applicable_

## Type of change

Please delete options that are not relevant.

- [ ] Bug fix (non-breaking change which fixes an issue)
- [ ] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing
functionality to not work as expected)
- [ ] This change requires a documentation update

## How Has This Been Tested?

Please describe the tests that you ran to verify your changes. Provide
instructions so we can reproduce.
Please also list any relevant details for your test configuration

- [ ] Test A
- [ ] Test B

**Test Configuration**: _if applicable_

- Firmware version:
- Hardware:
- Toolchain:
- SDK:

**List of new dependencies**: _if applicable_

Provide a list of newly added dependencies in this PR, with the
description of what are they doing and why do we need them.

- Crate A: description
- Crate B: description

## Checklist

- [ ] My code follows the style guidelines of this project
- [ ] I have performed a self-review of my code
- [ ] I have commented my code, particularly in hard-to-understand areas
- [ ] I have made corresponding changes to the documentation
- [ ] My changes generate no new warnings
- [ ] I have added tests that prove my fix is effective or that my
feature works
- [ ] New and existing unit tests pass locally with my changes
- [ ] Any dependent changes have been merged and published in downstream
modules

Author: jmgilman

.github/workflows/ci.yml

@@ -105,7 +105,7 @@ jobs:
         with:
           deployment_repo: input-output-hk/catalyst-world
           # NOTE: For new services being deployed, this list must be updated
-          images: cat-data-service migrations voting-node
+          images: cat-data-service fragment-exporter migrations voting-node
           environment: dev
           tag: ${{ env.TAG }}
           token: ${{ steps.setup.outputs.token }}

nix/automation/devshells.nix

@@ -41,6 +41,7 @@
 
       # Misc tools
       jq
+      poetry
       python310Packages.pylddwrap
     ];
   };

utilities/fragment-exporter/Earthfile

@@ -0,0 +1,63 @@
+# Set the Earthly version to 0.7
+VERSION 0.7
+
+# Use current debian stable with python
+FROM python:3.11-slim-bookworm
+
+poetry:
+    WORKDIR /work
+
+    ENV POETRY_HOME=/tmp/poetry
+    ENV PATH=$POETRY_HOME/bin:$PATH
+
+    RUN apt-get update && \
+        apt-get install -y --no-install-recommends curl
+
+    RUN curl -sSL https://install.python-poetry.org | python3 -
+
+    COPY pyproject.toml .
+    COPY poetry.lock .
+
+    RUN poetry install --only main --no-root
+
+src:
+    FROM +poetry
+
+    COPY --dir fragment_exporter README.md .
+
+check:
+    FROM +src
+
+    COPY --dir tests tests
+
+    RUN poetry install --only dev
+    RUN poetry run black --check .
+    RUN poetry run ruff check .
+    RUN poetry run pytest -v
+
+build:
+    FROM +check
+
+    RUN poetry export --without-hashes -f requirements.txt --output requirements.txt
+    RUN poetry build --no-cache -f wheel
+
+    SAVE ARTIFACT dist
+    SAVE ARTIFACT requirements.txt
+
+docker:
+    ARG tag="latest"
+    ARG registry
+
+    WORKDIR /app
+
+    COPY +build/dist .
+    COPY +build/requirements.txt .
+    COPY entry.sh /app
+
+    RUN chmod +x entry.sh
+    RUN pip3 install --no-cache -r requirements.txt
+    RUN pip3 install --no-cache *.whl
+
+    ENTRYPOINT ["/app/entry.sh"]
+
+    SAVE IMAGE --push ${registry}fragment-exporter:$tag

utilities/fragment-exporter/README.md

@@ -0,0 +1,5 @@
+# Fragment Exporter
+
+This is a small utility for exporting fragment logs from Jormungandr leader
+nodes to `stdout`. This exporter is intended to be run alongside the leaders and
+scraped by Loki in order to trace fragments logs for leader nodes.

utilities/fragment-exporter/entry.sh

@@ -0,0 +1,64 @@
+#!/usr/bin/env bash
+
+# Enable strict mode
+set +x
+set -o errexit
+set -o pipefail
+set -o nounset
+set -o functrace
+set -o errtrace
+set -o monitor
+set -o posix
+shopt -s dotglob
+
+check_env_vars() {
+  local env_vars=("$@")
+
+  # Iterate over the array and check if each variable is set
+  for var in "${env_vars[@]}"; do
+    echo "Checking $var"
+    if [ -z "${!var}" ]; then
+      echo ">>> Error: $var is required and not set."
+      exit 1
+    fi
+  done
+}
+
+debug_sleep() {
+  if [ -n "${DEBUG_SLEEP:-}" ]; then
+    echo "DEBUG_SLEEP is set. Sleeping for ${DEBUG_SLEEP} seconds..."
+    sleep "${DEBUG_SLEEP}"
+  fi
+}
+
+echo ">>> Starting entrypoint script..."
+
+# Check if all required environment variables are set
+REQUIRED_ENV=(
+  "LEADER_IP_PREFIX"
+  "LEADER_PORT"
+  "FRAGMENT_EXPORTER_INDEX_PATH"
+)
+echo ">>> Checking required env vars..."
+check_env_vars "${REQUIRED_ENV[@]}"
+
+# Calculate the correct leader IP address (e.g. <prefix>.<index>)
+HOSTNAME=$(hostname)
+export FRAGMENT_EXPORTER_URL="http://${LEADER_IP_PREFIX}.${HOSTNAME##*-}:${LEADER_PORT}"
+
+# Print out the environment variables
+echo ">>> Using FRAGMENT_EXPORTER_URL: ${FRAGMENT_EXPORTER_URL}"
+echo ">>> Using FRAGMENT_EXPORTER_INDEX_PATH: ${FRAGMENT_EXPORTER_INDEX_PATH}"
+
+# Sleep if DEBUG_SLEEP is set
+debug_sleep
+
+# Define the command to be executed
+CMD_TO_RUN="fragment_exporter"
+
+# Add $* to the command so that additional flags can be passed
+ARGS="$*"
+CMD="$CMD_TO_RUN $ARGS"
+
+# Expand the command with arguments and capture the exit code
+eval "$CMD"

utilities/fragment-exporter/fragment_exporter/__init__.py

@@ -0,0 +1,60 @@
+from urllib.parse import urljoin
+
+import requests
+import ijson
+import json
+import hashlib
+from .index import Index
+from .node import Node
+
+
+class Exporter:
+    """Exports fragments from a node to stdout.
+
+    Attributes:
+        url: The URL of the node
+        index: A index for storing fragments
+        node: The node to export from
+    """
+
+    def __init__(self, url: str, index: Index, node: Node):
+        self.url = urljoin(url, "/api/v0/fragment/logs")
+        self.index = index
+        self.node = node
+
+    def run(self):
+        """Runs the exporter."""
+        # Purge the index if the node has been restarted
+        current_block_height = self.node.get_last_block_height()
+        previous_block_height = self.index.get_last_block_height()
+        if current_block_height < previous_block_height:
+            print("Node has been restarted. Purging index.")
+            self.index.purge()
+
+        # Update the last block height
+        self.index.set_last_block_height(current_block_height)
+
+        # Get the latest fragments
+        response = requests.get(self.url, stream=True)
+        response.raise_for_status()
+        fragments = ijson.items(response.raw, "item")
+
+        # Find and print new fragments
+        for fragment in fragments:
+            if self.index.exists(hash(fragment)):
+                continue
+
+            print(json.dumps(fragment))
+            self.index.insert(hash(fragment))
+
+
+def hash(fragment):
+    """Returns the hash of a fragment.
+
+    Args:
+        fragment: The fragment to hash
+
+    Returns:
+        The hash of the fragment
+    """
+    return hashlib.sha256(json.dumps(fragment).encode("utf-8")).hexdigest()

utilities/fragment-exporter/fragment_exporter/exporter.py

@@ -0,0 +1,110 @@
+import sqlite3
+import os
+
+
+class Index:
+    """Represents a local index of fragments.
+
+    Attributes:
+        db_path: The path to the index database
+    """
+
+    def __init__(self, path: str):
+        self.db_path = os.path.join(path, "index.db")
+        self._init_db()
+
+    def _init_db(self):
+        """Initializes the index database."""
+        with sqlite3.connect(self.db_path) as conn:
+            conn.execute(
+                """
+            CREATE TABLE IF NOT EXISTS fragments (
+                fragment_id TEXT PRIMARY KEY
+            )
+            """
+            )
+            conn.execute(
+                """
+            CREATE TABLE IF NOT EXISTS node_stats (
+                key TEXT PRIMARY KEY,
+                value INTEGER
+            )
+            """
+            )
+            conn.commit()
+
+    def insert(self, fragment_id: str):
+        """Inserts a fragment into the index.
+
+        Args:
+            fragment_id: The ID of the fragment to insert
+        """
+        with sqlite3.connect(self.db_path) as conn:
+            conn.execute(
+                """
+            INSERT OR IGNORE INTO fragments (fragment_id) VALUES (?)
+            """,
+                (fragment_id,),
+            )
+            conn.commit()
+
+    def exists(self, fragment_id: str) -> bool:
+        """Checks if a fragment exists in the index.
+
+        Args:
+            fragment_id: The ID of the fragment to check
+
+        Returns:
+            True if the fragment exists, False otherwise
+        """
+        with sqlite3.connect(self.db_path) as conn:
+            cur = conn.execute(
+                """
+            SELECT 1 FROM fragments WHERE fragment_id = ?
+            """,
+                (fragment_id,),
+            )
+            return cur.fetchone() is not None
+
+    def set_last_block_height(self, height: int):
+        """Sets the last block height in the index.
+
+        Args:
+            height: The last block height
+        """
+        with sqlite3.connect(self.db_path) as conn:
+            conn.execute(
+                """
+            INSERT OR REPLACE INTO node_stats
+            (key, value)
+            VALUES
+            ('last_block_height', ?)
+            """,
+                (height,),
+            )
+            conn.commit()
+
+    def get_last_block_height(self) -> int:
+        """Gets the last block height from the index.
+
+        Returns:
+            The last block height
+        """
+        with sqlite3.connect(self.db_path) as conn:
+            cur = conn.execute(
+                """
+            SELECT value FROM node_stats WHERE key = 'last_block_height'
+            """
+            )
+            result = cur.fetchone()
+            return result[0] if result else 0
+
+    def purge(self):
+        """Purges the index."""
+        with sqlite3.connect(self.db_path) as conn:
+            conn.execute(
+                """
+            DELETE FROM fragments
+            """
+            )
+            conn.commit()

utilities/fragment-exporter/fragment_exporter/index.py

@@ -0,0 +1,61 @@
+import argparse
+import os
+import sys
+import time
+
+from loguru import logger
+
+from .exporter import Exporter
+from .index import Index
+from .node import Node
+
+
+def main():
+    logger.remove()
+    logger.add(sys.stderr, serialize=True)
+    parser = argparse.ArgumentParser(description="Exports node fragment logs to stdout")
+    parser.add_argument(
+        "--url",
+        type=str,
+        default=os.environ.get("FRAGMENT_EXPORTER_URL"),
+        help="URL of the node to export from",
+        required=not bool(os.environ.get("FRAGMENT_EXPORTER_URL")),
+    )
+    parser.add_argument(
+        "--index-path",
+        type=str,
+        default=os.environ.get("FRAGMENT_EXPORTER_INDEX_PATH"),
+        help="A local path to a directory where the index file will be stored",
+        required=not bool(os.environ.get("FRAGMENT_EXPORTER_INDEX_PATH")),
+    )
+    parser.add_argument(
+        "--interval",
+        type=int,
+        default=int(os.environ.get("FRAGMENT_EXPORTER_INTERVAL", 3600)),
+        help="Interval in seconds between consecutive export runs",
+    )
+
+    args = parser.parse_args()
+    if not os.path.exists(args.index_path):
+        logger.error(f"Path {args.index_path} does not exist")
+        sys.exit(1)
+
+    node = Node(args.url)
+    index = Index(args.index_path)
+    exporter = Exporter(args.url, index, node)
+
+    while True:
+        logger.info("Entering main control loop")
+
+        try:
+            logger.info("Running exporter")
+            exporter.run()
+        except Exception as e:
+            logger.error(f"Error encountered during export: {e}")
+        finally:
+            logger.info(f"Sleeping for {args.interval} seconds")
+            time.sleep(args.interval)
+
+
+if __name__ == "__main__":
+    main()