Health Report integration tests bootstrapper and initial slow start s…

…cenario implementation.

.buildkite/scripts/health-report-tests/README.md

@@ -0,0 +1,15 @@
+## Description
+This package for integration tests of the Health Report API.
+Export `LS_VERSION` (major and minor version such as 8.x) to run on a specific branch. By default, it uses the main branch.
+
+## How to run the Health Report Integration test?
+### Prerequisites
+Make sure you have python installed. Install the integration test dependencies with the following command:
+```shell
+python3 -mpip install -r .buildkite/scripts/health-report-tests/requirements.txt
+```
+
+### Run the integration test
+```shell
+python3 .buildkite/scripts/health-report-tests/main.py
+```

.buildkite/scripts/health-report-tests/bootstrap.py

@@ -0,0 +1,99 @@
+"""
+Health Report Integration test bootstrapper with Python script
+    - A script to resolve Logstash version if not provided
+    - Download LS docker image and spin up
+    - When tests finished, teardown the Logstash
+"""
+import os
+import subprocess
+import util
+import yaml
+
+
+class Bootstrap:
+    ELASTIC_STACK_VERSIONS_URL = "https://artifacts-api.elastic.co/v1/versions"
+
+    def __init__(self) -> None:
+        f"""
+        A constructor of the {Bootstrap}.
+        Returns:
+            Resolves Logstash branch considering provided LS_VERSION
+            Checks out git branch
+        """
+        logstash_version = os.environ.get("LS_VERSION")
+        if logstash_version is None:
+            # version is not specified, use the main branch, no need to git checkout
+            print(f"LS_VERSION is not specified, using main branch.")
+        else:
+            # LS_VERSION accepts major latest as a major.x or specific version as X.Y
+            if logstash_version.find(".x") == -1:
+                print(f"Using specified branch: {logstash_version}")
+                util.git_check_out_branch(logstash_version)
+            else:
+                major_version = logstash_version.split(".")[0]
+                if major_version and major_version.isnumeric():
+                    resolved_version = self.__resolve_latest_stack_version_for(major_version)
+                    minor_version = resolved_version.split(".")[1]
+                    branch = major_version + "." + minor_version
+                    print(f"Using resolved branch: {branch}")
+                    util.git_check_out_branch(branch)
+                else:
+                    raise ValueError(f"Invalid value set to LS_VERSION. Please set it properly (ex: 8.x or 9.0) and "
+                                     f"rerun again")
+
+    def __resolve_latest_stack_version_for(self, major_version: str) -> None:
+        resolved_version = ""
+        response = util.call_url_with_retry(self.ELASTIC_STACK_VERSIONS_URL)
+        release_versions = response.json()["versions"]
+        for release_version in reversed(release_versions):
+            if release_version.find("SNAPSHOT") > 0:
+                continue
+            if release_version.split(".")[0] == major_version:
+                print(f"Resolved latest version for {major_version} is {release_version}.")
+                resolved_version = release_version
+                break
+
+        if resolved_version == "":
+            raise ValueError(f"Cannot resolve latest version for {major_version} major")
+        return resolved_version
+
+    def install_plugin(self, plugin_path: str) -> None:
+        util.run_or_raise_error(
+            ["bin/logstash-plugin", "install", plugin_path],
+            f"Failed to install {plugin_path}")
+
+    def build_logstash(self):
+        print(f"Building Logstash.")
+        util.run_or_raise_error(
+            ["./gradlew", "clean", "bootstrap", "assemble", "installDefaultGems"],
+            "Failed to build Logstash")
+        print(f"Logstash has successfully built.")
+
+    def apply_config(self, config: dict) -> None:
+        with open(os.getcwd() + "/config/pipelines.yml", 'w') as pipelines_file:
+            yaml.dump(config, pipelines_file)
+
+    def run_logstash(self) -> subprocess.Popen:
+        process = subprocess.Popen(["bin/logstash"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
+        if process.poll() is not None:
+            print(f"Logstash failed to run, check the the config and logs, then rerun.")
+            return None
+
+        # Read stdout and stderr in real-time
+        logs = []
+        for stdout_line in iter(process.stdout.readline, ""):
+            # print("STDOUT:", stdout_line.strip())
+            logs.append(stdout_line.strip())
+            if "Starting pipeline" in stdout_line:
+                break
+            if "Logstash shut down" in stdout_line:
+                print(f"Logstash couldn't spin up.")
+                print(logs)
+                return None
+
+        print(f"Logstash is running with PID: {process.pid}.")
+        return process
+
+    def stop_logstash(self, process: subprocess.Popen) -> None:
+        process.terminate()
+        print(f"Stopping Logstash...")

.buildkite/scripts/health-report-tests/config_validator.py

@@ -0,0 +1,80 @@
+import yaml
+from typing import Any, List, Dict, Union
+
+
+class ConfigValidator:
+
+    REQUIRED_KEYS: Dict[str, List[str]] = {
+        "config": ["pipeline.id", "config.string"],
+        "expectation": ["status", "symptom", "diagnosis", "impacts", "details"],
+        "diagnosis": ["cause"],
+        "impacts": ["description", "impact_areas"],
+        "details": ["run_state"],
+    }
+
+    def __init__(self):
+        self.yaml_content = None
+
+    def __validate_keys(self, actual_keys: List[str], expected_keys: List[str], section: str) -> bool:
+        """Validate the keys at the current level."""
+        missing_keys = set(expected_keys) - set(actual_keys)
+        if len(missing_keys) == len(expected_keys):
+            print(f"Missing keys in {section}: {missing_keys}")
+            return False
+        return True
+
+    def __validate_config(self, config_list: List[Dict[str, Any]]) -> bool:
+        """Validate the 'config' section."""
+        for config_item in config_list:
+            if not self.__validate_keys(list(config_item.keys()), self.REQUIRED_KEYS["config"], "config"):
+                return False
+        return True
+
+    def __validate_expectation(self, expectation_list: List[Dict[str, Any]]) -> bool:
+        """Validate the 'expectation' section."""
+        for expectation_item in expectation_list:
+            if not self.__validate_keys(list(expectation_item.keys()), self.REQUIRED_KEYS["expectation"], "expectation"):
+                return False
+            if "diagnosis" in expectation_item:
+                for diagnosis in expectation_item["diagnosis"]:
+                    if not self.__validate_keys(list(diagnosis.keys()), self.REQUIRED_KEYS["diagnosis"], "diagnosis"):
+                        return False
+            if "impacts" in expectation_item:
+                for impact in expectation_item["impacts"]:
+                    if not self.__validate_keys(list(impact.keys()), self.REQUIRED_KEYS["impacts"], "impacts"):
+                        return False
+            if "details" in expectation_item:
+                for detail in expectation_item["details"]:
+                    if not self.__validate_keys(list(detail.keys()), self.REQUIRED_KEYS["details"], "details"):
+                        return False
+        return True
+
+    def load(self, file_path: str) -> None:
+        """Load the YAML file content into self.yaml_content."""
+        self.yaml_content: Union[List[Dict[str, Any]], None] = None
+        try:
+            with open(file_path, 'r') as file:
+                self.yaml_content = yaml.safe_load(file)
+        except yaml.YAMLError as exc:
+            print(f"Error in YAML file: {exc}")
+            self.yaml_content = None
+
+    def is_valid(self) -> bool:
+        """Validate the entire YAML structure."""
+        if self.yaml_content is None:
+            print(f"YAML content is empty.")
+            return False
+
+        if not isinstance(self.yaml_content, list):
+            print(f"YAML structure is not as expected, it should start with a list.")
+            return False
+
+        for item in self.yaml_content:
+            if "config" in item and not self.__validate_config(item["config"]):
+                return False
+
+            if "expectation" in item and not self.__validate_expectation(item["expectation"]):
+                return False
+
+        print(f"YAML file validation successful!")
+        return True

.buildkite/scripts/health-report-tests/logstash_health_report.py

@@ -0,0 +1,16 @@
+"""
+A class to provide information about Logstash node stats.
+"""
+
+import util
+
+
+class LogstashHealthReport:
+    LOGSTASH_HEALTH_REPORT_URL = "http://localhost:9600/_health_report"
+
+    def __init__(self):
+        pass
+
+    def get(self):
+        response = util.call_url_with_retry(self.LOGSTASH_HEALTH_REPORT_URL)
+        return response.json()

.buildkite/scripts/health-report-tests/logstash_stats.py

@@ -0,0 +1,16 @@
+"""
+A class to provide information about Logstash node stats.
+"""
+
+import util
+
+
+class LogstashStats:
+    LOGSTASH_STATS_URL = "http://localhost:9600/_node/stats"
+
+    def __init__(self):
+        pass
+
+    def get(self):
+        response = util.call_url_with_retry(self.LOGSTASH_STATS_URL)
+        return response.json()

.buildkite/scripts/health-report-tests/main.py

@@ -0,0 +1,74 @@
+"""
+ Main entry point of the LS health report API integration test suites
+"""
+import glob
+import os
+from bootstrap import Bootstrap
+from scenario_executor import ScenarioExecutor
+from config_validator import ConfigValidator
+import yaml
+import util
+
+
+class BootstrapContextManager:
+
+    def __init__(self):
+        pass
+
+    def __enter__(self):
+        print(f"Starting Logstash Health Report Integration test.")
+        self.bootstrap = Bootstrap()
+        # self.bootstrap.build_logstash()
+
+        plugin_path = os.getcwd() + "/qa/support/logstash-integration-failure_injector/logstash-integration" \
+                                    "-failure_injector-*.gem"
+        matching_files = glob.glob(plugin_path)
+        if len(matching_files) == 0:
+            raise ValueError(f"Could not find logstash-integration-failure_injector plugin.")
+
+        # self.bootstrap.install_plugin(matching_files[0])
+        print(f"logstash-integration-failure_injector successfully installed.")
+        return self.bootstrap
+
+    def apply_config(self, bootstrap: Bootstrap, config: str):
+        bootstrap.apply_config(config)
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        if exc_type is not None:
+            traceback.print_exception(exc_type, exc_value, traceback)
+
+
+def main():
+    with BootstrapContextManager() as bootstrap:
+        scenario_executor = ScenarioExecutor()
+        config_validator = ConfigValidator()
+
+        working_dir = os.getcwd()
+        scenario_files_path = working_dir + "/.buildkite/scripts/health-report-tests/tests/*.yaml"
+        scenario_files = glob.glob(scenario_files_path)
+
+        for scenario_file in scenario_files:
+            print(f"Validating {scenario_file} scenario file.")
+            config_validator.load(scenario_file)
+            if not config_validator.is_valid():
+                print(f"{scenario_file} scenario file is not valid.")
+                return
+
+        for scenario_file in scenario_files:
+            with open(scenario_file, 'r') as file:
+                # scenario_content: Union[List[Dict[str, Any]], None] = None
+                scenario_content = yaml.safe_load(file)
+                scenario_name = util.get_element_of_array(scenario_content, 'name')
+                config = util.get_element_of_array(scenario_content, 'config')
+                if config is not None:
+                    bootstrap.apply_config(config)
+                    expectation = util.get_element_of_array(scenario_content, 'expectation')
+                    process = bootstrap.run_logstash()
+                    if process is not None:
+                        scenario_executor.on(scenario_name, expectation)
+                        process.terminate()
+                    break
+
+
+if __name__ == "__main__":
+    main()

.buildkite/scripts/health-report-tests/main.sh

@@ -0,0 +1,6 @@
+#!/usr/bin/env bash
+set -eo pipefail
+
+# Install prerequisites and run integration tests
+python3 -mpip install -r .buildkite/scripts/health-report-tests/requirements.txt
+python3 .buildkite/scripts/health-report-tests/main.py

.buildkite/scripts/health-report-tests/requirements.txt

@@ -0,0 +1,3 @@
+requests==2.32.3
+deepdiff==8.0.1
+pyyaml==6.0.2

.buildkite/scripts/health-report-tests/scenario_executor.py

@@ -0,0 +1,43 @@
+"""
+A class to execute the given scenario for Logstash Health Report integration test
+"""
+from deepdiff import DeepDiff
+from logstash_health_report import LogstashHealthReport
+
+
+class ScenarioExecutor:
+    logstash_health_report_api = LogstashHealthReport()
+
+    def __init__(self):
+        pass
+
+    def __meets_expectation(self, scenario_content: list) -> None:
+        logstash_health = self.logstash_health_report_api.get()
+        print(f"Logstash health report: {logstash_health}")
+
+        differences = []
+        for index, item in enumerate(scenario_content):
+            if "expectation" in item:
+                key = f"Item {index + 1}"
+                stat_value = logstash_health.get(key, {}).get("expectation")
+
+                if stat_value:
+                    diff = DeepDiff(item["expectation"], stat_value, ignore_order=True).to_dict()
+                    if diff:
+                        differences.append({key: diff})
+                else:
+                    print(f"Stats do not contain an 'expectation' entry for {key}")
+
+        if differences:
+            print("Differences found in 'expectation' section between YAML content and stats:")
+            for diff in differences:
+                print(diff)
+            return False
+        else:
+            print("YAML 'expectation' section matches the stats.")
+            return True
+
+    def on(self, scenario_name: str, scenario_content: list) -> None:
+        print(f"Testing the scenario: {scenario_content}")
+        if self.__meets_expectation(scenario_content) is False:
+            raise Exception(f"{scenario_name} failed.")

.buildkite/scripts/health-report-tests/tests/slow-start.yaml

@@ -0,0 +1,17 @@
+- name: "Slow start pipeline"
+- config:
+  - pipeline.id: slow-start-pp
+    config.string: |
+      input { heartbeat {} }
+      filter { failure_injector { degrade_at => [register] } }
+      output { stdout {} }
+- expectation:
+    - status: yellow
+    - symptom: "The pipeline is degraded or at risk of becoming unhealthy; 1 area is impacted and 1 diagnosis is available."
+    - diagnosis:
+        - cause: "pipeline is loading"
+    - impacts:
+      - description: "pipeline is loading"
+      - impact_areas: "pipeline_execution"
+    - details:
+      - run_state: "LOADING"