Forward-port PR #16520 to main: Feature: health report api

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

@@ -0,0 +1,18 @@
+## Description
+This package for integration tests of the Health Report API.
+Export `LS_BRANCH` 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 tests
+```shell
+python3 .buildkite/scripts/health-report-tests/main.py
+```
+
+### Troubleshooting
+- If you get `WARNING: pip is configured with locations that require TLS/SSL,...` warning message, make sure you have python >=3.12.4 installed.

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

@@ -0,0 +1,101 @@
+"""
+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_BRANCH
+            Checks out git branch
+        """
+        logstash_branch = os.environ.get("LS_BRANCH")
+        if logstash_branch is None:
+            # version is not specified, use the main branch, no need to git checkout
+            print(f"LS_BRANCH is not specified, using main branch.")
+        else:
+            # LS_BRANCH accepts major latest as a major.x or specific branch as X.Y
+            if logstash_branch.find(".x") == -1:
+                print(f"Using specified branch: {logstash_branch}")
+                util.git_check_out_branch(logstash_branch)
+            else:
+                major_version = logstash_branch.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_BRANCH. Please set it properly (ex: 8.x or 9.0) and "
+                                     f"rerun again")
+
+    def __resolve_latest_stack_version_for(self, major_version: str) -> str:
+        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() + "/.buildkite/scripts/health-report-tests/config/pipelines.yml", 'w') as pipelines_file:
+            yaml.dump(config, pipelines_file)
+
+    def run_logstash(self, full_start_required: bool) -> subprocess.Popen:
+        # --config.reload.automatic is to make instance active
+        # it is helpful when testing crash pipeline cases
+        config_path = os.getcwd() + "/.buildkite/scripts/health-report-tests/config"
+        process = subprocess.Popen(["bin/logstash", "--config.reload.automatic", "--path.settings", config_path,
+                                    "-w 1"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, shell=False)
+        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, ""):
+            logs.append(stdout_line.strip())
+            # we don't wait for Logstash fully start as we also test slow pipeline start scenarios
+            if full_start_required is False and "Starting pipeline" in stdout_line:
+                break
+            if full_start_required is True and "Pipeline started" in stdout_line:
+                break
+            if "Logstash shut down" in stdout_line or "Logstash stopped" 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

.buildkite/scripts/health-report-tests/config/pipelines.yml

@@ -0,0 +1 @@
+# Intentionally left blank

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

@@ -0,0 +1,69 @@
+import yaml
+from typing import Any, List, Dict
+
+
+class ConfigValidator:
+    REQUIRED_KEYS = {
+        "root": ["name", "config", "conditions", "expectation"],
+        "config": ["pipeline.id", "config.string"],
+        "conditions": ["full_start_required"],
+        "expectation": ["status", "symptom", "indicators"],
+        "indicators": ["pipelines"],
+        "pipelines": ["status", "symptom", "indicators"],
+        "DYNAMIC": ["status", "symptom", "diagnosis", "impacts", "details"],
+        "details": ["status"],
+        "status": ["state"]
+    }
+
+    def __init__(self):
+        self.yaml_content = None
+
+    def __has_valid_keys(self, data: any, key_path: str, repeated: bool) -> bool:
+        if isinstance(data, str) or isinstance(data, bool):   # we reached values
+            return True
+
+        # we have two indicators section and for the next repeated ones, we go deeper
+        first_key = next(iter(data))
+        data = data[first_key] if repeated and key_path == "indicators" else data
+
+        if isinstance(data, dict):
+            # pipeline-id is a DYNAMIC
+            required = self.REQUIRED_KEYS.get("DYNAMIC" if repeated and key_path == "indicators" else key_path, [])
+            repeated = not repeated if key_path == "indicators" else repeated
+            for key in required:
+                if key not in data:
+                    print(f"Missing key '{key}' in '{key_path}'")
+                    return False
+                else:
+                    dic_keys_result = self.__has_valid_keys(data[key], key, repeated)
+                    if dic_keys_result is False:
+                        return False
+        elif isinstance(data, list):
+            for item in data:
+                list_keys_result = self.__has_valid_keys(item, key_path, repeated)
+                if list_keys_result is False:
+                    return False
+        return True
+
+    def load(self, file_path: str) -> None:
+        """Load the YAML file content into self.yaml_content."""
+        self.yaml_content: [Dict[str, Any]] = 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, dict):
+            print(f"YAML structure is not as expected, it should start with a Dict.")
+            return False
+
+        result = self.__has_valid_keys(self.yaml_content, "root", False)
+        return True if result is True else False

.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/main.py

@@ -0,0 +1,87 @@
+"""
+ Main entry point of the LS health report API integration test suites
+"""
+import glob
+import os
+import time
+import traceback
+import yaml
+from bootstrap import Bootstrap
+from scenario_executor import ScenarioExecutor
+from config_validator import ConfigValidator
+
+
+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 __exit__(self, exc_type, exc_value, exc_traceback):
+        if exc_type is not None:
+            print(traceback.format_exception(exc_type, exc_value, exc_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 config_validator.is_valid() is False:
+                print(f"{scenario_file} scenario file is not valid.")
+                return
+            else:
+                print(f"Validation succeeded.")
+
+        has_failed_scenario = False
+        for scenario_file in scenario_files:
+            with open(scenario_file, 'r') as file:
+                # scenario_content: Dict[str, Any] = None
+                scenario_content = yaml.safe_load(file)
+                print(f"Testing `{scenario_content.get('name')}` scenario.")
+                scenario_name = scenario_content['name']
+
+                is_full_start_required = next(sub.get('full_start_required') for sub in
+                                              scenario_content.get('conditions') if 'full_start_required' in sub)
+                config = scenario_content['config']
+                if config is not None:
+                    bootstrap.apply_config(config)
+                    expectations = scenario_content.get("expectation")
+                    process = bootstrap.run_logstash(is_full_start_required)
+                    if process is not None:
+                        try:
+                            scenario_executor.on(scenario_name, expectations)
+                        except Exception as e:
+                            print(e)
+                            has_failed_scenario = True
+                        process.terminate()
+                        time.sleep(5)   # leave some window to terminate the process
+
+        if has_failed_scenario:
+            # intentionally fail due to visibility
+            raise Exception("Some of scenarios failed, check the log for details.")
+
+
+if __name__ == "__main__":
+    main()

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

@@ -1,9 +1,5 @@
 #!/usr/bin/env bash
 set -eo pipefail
 
-# TODO:
-#  if branch is specified with X.Y, pull branches from ACTIVE_BRANCHES_URL="https://raw.githubusercontent.com/elastic/logstash/main/ci/branches.json", parse and use
-#  build Logstash from specificed (ex: 8.x -> translates to 8.latest, 8.16) branch, defaults to main
-#  install requirements of the python package and run main.py
-
-
+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,2 @@
+requests==2.32.3
+pyyaml==6.0.2

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

@@ -0,0 +1,65 @@
+"""
+A class to execute the given scenario for Logstash Health Report integration test
+"""
+import time
+from logstash_health_report import LogstashHealthReport
+
+
+class ScenarioExecutor:
+    logstash_health_report_api = LogstashHealthReport()
+
+    def __init__(self):
+        pass
+
+    def __has_intersection(self, expects, results):
+        # we expect expects to be existing in results
+        for expect in expects:
+            for result in results:
+                if result.get('help_url') and "health-report-pipeline-status.html#" not in result.get('help_url'):
+                    return False
+                if not all(key in result and result[key] == value for key, value in expect.items()):
+                    return False
+        return True
+
+    def __get_difference(self, differences: list, expectations: dict, reports: dict) -> dict:
+        for key in expectations.keys():
+
+            if type(expectations.get(key)) != type(reports.get(key)):
+                differences.append(f"Scenario expectation and Health API report structure differs for {key}.")
+                return differences
+
+            if isinstance(expectations.get(key), str):
+                if expectations.get(key) != reports.get(key):
+                    differences.append({key: {"expected": expectations.get(key), "got": reports.get(key)}})
+                continue
+            elif isinstance(expectations.get(key), dict):
+                self.__get_difference(differences, expectations.get(key), reports.get(key))
+            elif isinstance(expectations.get(key), list):
+                if not self.__has_intersection(expectations.get(key), reports.get(key)):
+                    differences.append({key: {"expected": expectations.get(key), "got": reports.get(key)}})
+        return differences
+
+    def __is_expected(self, expectations: dict) -> None:
+        reports = self.logstash_health_report_api.get()
+        differences = self.__get_difference([], expectations, reports)
+        if differences:
+            print("Differences found in 'expectation' section between YAML content and stats:")
+            for diff in differences:
+                print(f"Difference: {diff}")
+            return False
+        else:
+            return True
+
+    def on(self, scenario_name: str, expectations: dict) -> None:
+        # retriable check the expectations
+        attempts = 5
+        while self.__is_expected(expectations) is False:
+            attempts = attempts - 1
+            if attempts == 0:
+                break
+            time.sleep(1)
+
+        if attempts == 0:
+            raise Exception(f"{scenario_name} failed.")
+        else:
+            print(f"Scenario `{scenario_name}` expectaion meets the health report stats.")