chore: Sync data availability and integrity tests with latest Nomos node changes

.github/actions/prune-vm/action.yml

@@ -0,0 +1,27 @@
+# Inspired by https://github.com/AdityaGarg8/remove-unwanted-software
+# to free up disk space. Currently removes Dotnet, Android and Haskell.
+name: Remove unwanted software
+description: Default GitHub runners come with a lot of unnecessary software
+runs:
+  using: "composite"
+  steps:
+    - name: Disk space report before modification
+      shell: bash
+      run: |
+        echo "==> Available space before cleanup"
+        echo
+        df -h
+    - name: Maximize build disk space
+      shell: bash
+      run: |
+          set -euo pipefail
+          sudo rm -rf /usr/share/dotnet
+          sudo rm -rf /usr/local/lib/android
+          sudo rm -rf /opt/ghc 
+          sudo rm -rf /usr/local/.ghcup
+    - name: Disk space report after modification
+      shell: bash
+      run: |
+          echo "==> Available space after cleanup"
+          echo
+          df -h

.github/workflows/nomos_daily.yml

@@ -0,0 +1,11 @@
+name: Nomos E2E Tests Daily
+
+on:
+  schedule:
+    - cron:  '0 4 * * *'
+  workflow_dispatch:
+
+jobs:
+  test-common:
+    uses: ./.github/workflows/test_common.yml
+

.github/workflows/test_common.yml

@@ -0,0 +1,32 @@
+name: E2E Tests Common
+
+on:
+  workflow_call:
+
+env:
+  FORCE_COLOR: "1"
+
+jobs:
+  tests:
+    name: tests
+    runs-on: ubuntu-latest
+    timeout-minutes: 120
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Remove unwanted software
+        uses: ./.github/actions/prune-vm
+
+      - uses: actions/setup-python@v4
+        with:
+          python-version: '3.12'
+          cache: 'pip'
+
+      - run: |
+          pip install -r requirements.txt
+          mkdir -p kzgrs
+          wget https://raw.githubusercontent.com/logos-co/nomos-node/master/tests/kzgrs/kzgrs_test_params -O kzgrs/kzgrs_test_params
+
+      - name: Run tests
+        run: |
+          pytest

cluster_config/cfgsync-5node.yaml → cluster_config/cfgsync-4node.yaml

@@ -1,5 +1,5 @@
 port: 4400
-n_hosts: 5
+n_hosts: 4
 timeout: 30
 
 # ConsensusConfig related parameters

cluster_config/cfgsync.yaml

@@ -1,5 +1,5 @@
 port: 4400
-n_hosts: 5
+n_hosts: 2
 timeout: 30
 
 # ConsensusConfig related parameters

src/node/api_clients/rest.py → src/api_clients/rest.py

@@ -1,7 +1,6 @@
 from src.libs.custom_logger import get_custom_logger
 import json
-from urllib.parse import quote
-from src.node.api_clients.base_client import BaseClient
+from src.api_clients.base_client import BaseClient
 
 logger = get_custom_logger(__name__)
 
@@ -12,12 +11,12 @@ def __init__(self, rest_port):
 
     def rest_call(self, method, endpoint, payload=None):
         url = f"http://127.0.0.1:{self._rest_port}/{endpoint}"
-        headers = {"Content-Type": "application/json"}
+        headers = {"Content-Type": "application/json", "Connection": "close"}
         return self.make_request(method, url, headers=headers, data=payload)
 
     def rest_call_text(self, method, endpoint, payload=None):
         url = f"http://127.0.0.1:{self._rest_port}/{endpoint}"
-        headers = {"accept": "text/plain"}
+        headers = {"accept": "text/plain", "Connection": "close"}
         return self.make_request(method, url, headers=headers, data=payload)
 
     def info(self):

src/cli/cli_vars.py

@@ -0,0 +1,11 @@
+from src.env_vars import NOMOS_IMAGE
+
+nomos_cli = {
+    "reconstruct": {
+        "image": NOMOS_IMAGE,
+        "flags": [{"--app-blobs": [0]}],  # Value [] is a list of indexes into list of values required for the flag
+        "volumes": [],
+        "ports": [],
+        "entrypoint": "",
+    },
+}

src/cli/nomos_cli.py

@@ -0,0 +1,111 @@
+import json
+import os
+import re
+
+from src.data_storage import DS
+from src.libs.common import generate_log_prefix
+from src.libs.custom_logger import get_custom_logger
+from tenacity import retry, stop_after_delay, wait_fixed
+
+from src.cli.cli_vars import nomos_cli
+from src.docker_manager import DockerManager, stop, kill
+from src.env_vars import DOCKER_LOG_DIR, NOMOS_CLI
+from src.steps.da import remove_padding
+
+logger = get_custom_logger(__name__)
+
+
+class NomosCli:
+    def __init__(self, **kwargs):
+        if "command" not in kwargs:
+            raise ValueError("The command parameter is required")
+
+        command = kwargs["command"]
+        if command not in nomos_cli:
+            raise ValueError("Unknown command provided")
+
+        logger.debug(f"Cli is going to be initialized with this config {nomos_cli[command]}")
+        self._command = command
+        self._image_name = nomos_cli[command]["image"]
+        self._internal_ports = nomos_cli[command]["ports"]
+        self._volumes = nomos_cli[command]["volumes"]
+        self._entrypoint = nomos_cli[command]["entrypoint"]
+
+        container_name = "nomos-cli-" + generate_log_prefix()
+        self._log_path = os.path.join(DOCKER_LOG_DIR, f"{container_name}__{self._image_name.replace('/', '_')}.log")
+        self._docker_manager = DockerManager(self._image_name)
+        self._container_name = container_name
+        self._container = None
+
+        cwd = os.getcwd()
+        self._volumes = [cwd + "/" + volume for volume in self._volumes]
+
+    def run(self, input_values=None, **kwargs):
+        logger.debug(f"NomosCli starting with log path {self._log_path}")
+
+        self._port_map = {}
+
+        cmd = [NOMOS_CLI, self._command]
+        for flag in nomos_cli[self._command]["flags"]:
+            for f, indexes in flag.items():
+                cmd.append(f)
+                for j in indexes:
+                    cmd.append(input_values[j])
+
+        logger.debug(f"NomosCli command to run {cmd}")
+
+        self._container = self._docker_manager.start_container(
+            self._docker_manager.image,
+            port_bindings=self._port_map,
+            args=None,
+            log_path=self._log_path,
+            volumes=self._volumes,
+            entrypoint=self._entrypoint,
+            remove_container=True,
+            name=self._container_name,
+            command=cmd,
+        )
+
+        DS.nomos_nodes.append(self)
+
+        match self._command:
+            case "reconstruct":
+                decode_only = kwargs.get("decode_only", False)
+                return self.reconstruct(input_values=input_values, decode_only=decode_only)
+            case _:
+                return
+
+    def reconstruct(self, input_values=None, decode_only=False):
+        keywords = ["Reconstructed data"]
+
+        log_stream = self._container.logs(stream=True)
+
+        matches = self._docker_manager.search_log_for_keywords(self._log_path, keywords, False, log_stream)
+        assert len(matches) > 0, f"Reconstructed data not found {matches}"
+
+        # Use regular expression that captures the byte list after "Reconstructed data"
+        result = re.sub(r".*Reconstructed data\s*(\[[^\]]+\]).*", r"\1", matches[keywords[0]][0])
+
+        result_bytes = []
+        try:
+            result_bytes = json.loads(result)
+        except Exception as ex:
+            logger.debug(f"Conversion to bytes failed with exception {ex}")
+
+        if decode_only:
+            result_bytes = result_bytes[:-31]
+
+        result_bytes = remove_padding(result_bytes)
+        result = bytes(result_bytes).decode("utf-8")
+
+        DS.nomos_nodes.remove(self)
+
+        return result
+
+    @retry(stop=stop_after_delay(5), wait=wait_fixed(0.1), reraise=True)
+    def stop(self):
+        self._container = stop(self._container)
+
+    @retry(stop=stop_after_delay(5), wait=wait_fixed(0.1), reraise=True)
+    def kill(self):
+        self._container = kill(self._container)

src/node/docker_mananger.py → src/docker_manager.py

@@ -35,30 +35,41 @@ def create_network(self, network_name=NETWORK_NAME):
         logger.debug(f"Network {network_name} created")
         return network
 
-    def start_container(self, image_name, port_bindings, args, log_path, volumes, entrypoint, remove_container=True, name=None):
+    def start_container(self, image_name, port_bindings, args, log_path, volumes, entrypoint, **kwargs):
+        remove_container = kwargs.get("remove_container", True)
+        name = kwargs.get("name")
+        command = kwargs.get("command")
+
         cli_args = []
-        for key, value in args.items():
-            if isinstance(value, list):  # Check if value is a list
-                cli_args.extend([f"--{key}={item}" for item in value])  # Add a command for each item in the list
-            elif value is None:
-                cli_args.append(f"{key}")  # Add simple command as it is passed in the key
-            else:
-                cli_args.append(f"--{key}={value}")  # Add a single command
+        if command is None:
+            for key, value in args.items():
+                if isinstance(value, list):  # Check if value is a list
+                    cli_args.extend([f"--{key}={item}" for item in value])  # Add a command for each item in the list
+                elif value is None:
+                    cli_args.append(f"{key}")  # Add simple command as it is passed in the key
+                else:
+                    cli_args.append(f"--{key}={value}")  # Add a single command
+        else:
+            cli_args = command
 
         cli_args_str_for_log = " ".join(cli_args)
-        logger.debug(f"docker run -i -t {port_bindings} {image_name} {cli_args_str_for_log}")
-        container = self._client.containers.run(
-            image_name,
-            command=cli_args,
-            ports=port_bindings,
-            detach=True,
-            remove=remove_container,
-            auto_remove=remove_container,
-            volumes=volumes,
-            entrypoint=entrypoint,
-            name=name,
-            network=NETWORK_NAME,
-        )
+        logger.debug(f"docker run -i -t --entrypoint {entrypoint} {port_bindings} {image_name} {cli_args_str_for_log}")
+
+        try:
+            container = self._client.containers.run(
+                image_name,
+                command=cli_args,
+                ports=port_bindings,
+                detach=True,
+                remove=remove_container,
+                auto_remove=remove_container,
+                volumes=volumes,
+                entrypoint=entrypoint,
+                name=name,
+                network=NETWORK_NAME,
+            )
+        except Exception as ex:
+            logger.debug(f"Docker container run failed with exception {ex}")
 
         logger.debug(f"Container started with ID {container.short_id}. Setting up logs at {log_path}")
         log_thread = threading.Thread(target=self._log_container_output, args=(container, log_path))
@@ -125,19 +136,32 @@ def is_container_running(self, container):
     def image(self):
         return self._image
 
-    def search_log_for_keywords(self, log_path, keywords, use_regex=False):
+    def find_keywords_in_line(self, keywords, line, use_regex=False):
         matches = {keyword: [] for keyword in keywords}
 
-        # Open the log file and search line by line
-        with open(log_path, "r") as log_file:
-            for line in log_file:
-                for keyword in keywords:
-                    if use_regex:
-                        if re.search(keyword, line, re.IGNORECASE):
-                            matches[keyword].append(line.strip())
-                    else:
-                        if keyword.lower() in line.lower():
-                            matches[keyword].append(line.strip())
+        for keyword in keywords:
+            if use_regex:
+                if re.search(keyword, line, re.IGNORECASE):
+                    matches[keyword].append(line.strip())
+            else:
+                if keyword.lower() in line.lower():
+                    matches[keyword].append(line.strip())
+
+        return matches
+
+    def search_log_for_keywords(self, log_path, keywords, use_regex=False, log_stream=None):
+        matches = {}
+
+        # Read from stream
+        if log_stream is not None:
+            for line in log_stream:
+                matches = self.find_keywords_in_line(keywords, line.decode("utf-8"), use_regex=use_regex)
+
+        else:
+            # Open the log file and search line by line
+            with open(log_path, "r") as log_file:
+                for line in log_file:
+                    matches = self.find_keywords_in_line(keywords, line, use_regex=use_regex)
 
         # Check if there were any matches
         if any(matches[keyword] for keyword in keywords):
@@ -146,5 +170,31 @@ def search_log_for_keywords(self, log_path, keywords, use_regex=False):
                     logger.debug(f"Found matches for keyword '{keyword}': {lines}")
             return matches
         else:
-            logger.debug("No errors found in the nomos logs.")
+            logger.debug("No keywords found in the nomos logs.")
             return None
+
+
+def stop(container):
+    if container:
+        logger.debug(f"Stopping container with id {container.short_id}")
+        container.stop()
+        try:
+            container.remove()
+        except:
+            pass
+        logger.debug("Container stopped.")
+
+        return None
+
+
+def kill(container):
+    if container:
+        logger.debug(f"Killing container with id {container.short_id}")
+        container.kill()
+        try:
+            container.remove()
+        except:
+            pass
+        logger.debug("Container killed.")
+
+        return None

src/env_vars.py

@@ -19,10 +19,16 @@ def get_env_var(var_name, default=None):
 NOMOS_EXECUTOR = "nomos_executor"
 CFGSYNC = "cfgsync"
 
+DEFAULT_IMAGE = "ghcr.io/logos-co/nomos-node:latest"
+
 NODE_1 = get_env_var("NODE_1", NOMOS)
 NODE_2 = get_env_var("NODE_2", NOMOS_EXECUTOR)
 NODE_3 = get_env_var("NODE_3", CFGSYNC)
 
+NOMOS_IMAGE = get_env_var("NOMOS_IMAGE", DEFAULT_IMAGE)
+
+NOMOS_CLI = "/usr/bin/nomos-cli"
+
 ADDITIONAL_NODES = get_env_var("ADDITIONAL_NODES", f"{NOMOS},{NOMOS}")
 # more nodes need to follow the NODE_X pattern
 DOCKER_LOG_DIR = get_env_var("DOCKER_LOG_DIR", "./log/docker")