buildkite serverless test (#15150)

This commit adds a Buildkite pipeline to test against serverless endpoint daily

Tests cover 
- es-output
- es-input
- es-filter
- central pipeline management
- legacy monitoring
- dlq
- integration-filter
- kibana API
- metricbeat stack monitoring

Co-authored-by: João Duarte <[email protected]>
Co-authored-by: João Duarte <[email protected]>

Author: kaisecheng

.buildkite/scripts/setup_java.sh

@@ -0,0 +1,10 @@
+#!/bin/bash
+
+set -e
+
+install_java() {
+  # TODO: let's think about regularly creating a custom image for Logstash which may align on version.yml definitions
+  sudo apt update && sudo apt install -y openjdk-17-jdk && sudo apt install -y openjdk-17-jre
+}
+
+install_java

.buildkite/serverless_integration_pipeline.yml

@@ -1,3 +1,24 @@
+agents:
+  provider: "gcp"
+  machineType: "n1-standard-4"
+  image: family/core-ubuntu-2204
+
 steps:
-  - label: "Elasticsearch output test"
-    command: ./ci/serverless/serverless_core_rspec_tests.sh
+  - label: "DLQ rspec integration test"
+    command: ./.buildkite/scripts/setup_java.sh && ./ci/serverless/dlq_rspec_tests.sh
+  - label: "es-output test"
+    command: ./.buildkite/scripts/setup_java.sh && ./ci/serverless/es_output_tests.sh
+  - label: "es-input test"
+    command: ./.buildkite/scripts/setup_java.sh && ./ci/serverless/es_input_tests.sh
+  - label: "es-filter test"
+    command: ./.buildkite/scripts/setup_java.sh && ./ci/serverless/es_filter_tests.sh
+  - label: "elastic_integration filter test"
+    command: ./.buildkite/scripts/setup_java.sh && ./ci/serverless/elastic_integration_filter_tests.sh
+  - label: "central pipeline management test"
+    command: ./.buildkite/scripts/setup_java.sh && ./ci/serverless/cpm_tests.sh
+  - label: "Logstash legacy monitoring test"
+    command: ./.buildkite/scripts/setup_java.sh && ./ci/serverless/monitoring_tests.sh
+  - label: "Kibana API test"
+    command: ./.buildkite/scripts/setup_java.sh && ./ci/serverless/kibana_api_tests.sh
+  - label: "metricbeat stack monitoring test"
+    command: ./.buildkite/scripts/setup_java.sh && ./ci/serverless/metricbeat_monitoring_tests.sh

.gitignore

@@ -62,4 +62,5 @@ lib/pluginmanager/plugin_aliases.yml
 logstash-core/src/main/resources/org/logstash/plugins/plugin_aliases.yml
 spec/unit/plugin_manager/plugin_aliases.yml
 logstash-core/src/test/resources/org/logstash/plugins/plugin_aliases.yml
-qa/integration/fixtures/logs_rollover/log4j2.properties
+qa/integration/fixtures/logs_rollover/log4j2.properties
+ci/serverless/config/*.log

ci/serverless/common.sh

@@ -0,0 +1,151 @@
+#!/usr/bin/env bash
+set -ex
+
+export CURRENT_DIR="$(dirname "$0")"
+export INDEX_NAME="serverless_it_${BUILDKITE_BUILD_NUMBER:-`date +%s`}"
+# store all error messages
+export ERR_MSGS=()
+# numeric values representing the results of the checks. 0: pass, >0: fail
+export CHECKS=()
+
+setup_vault() {
+  vault_path=secret/ci/elastic-logstash/serverless-test
+  set +x
+  export ES_ENDPOINT=$(vault read -field=es_host "${vault_path}")
+  export ES_USER=$(vault read -field=es_user "${vault_path}")
+  export ES_PW=$(vault read -field=es_user_pw "${vault_path}")
+  export KB_ENDPOINT=$(vault read -field=kb_host "${vault_path}")
+  set -x
+}
+
+build_logstash() {
+  ./gradlew clean bootstrap assemble installDefaultGems
+}
+
+index_test_data() {
+  curl -X POST -u "$ES_USER:$ES_PW" "$ES_ENDPOINT/$INDEX_NAME/_bulk" -H 'Content-Type: application/json' --data-binary @"$CURRENT_DIR/test_data/book.json"
+}
+
+# $1: check function
+run_cpm_logstash() {
+  # copy log4j
+  cp  "$CURRENT_DIR/../../config/log4j2.properties" "$CURRENT_DIR/config/log4j2.properties"
+
+  # run logstash
+  $CURRENT_DIR/../../bin/logstash --path.settings "$CURRENT_DIR/config" 2>/dev/null &
+  export LS_PID=$!
+
+  check_logstash_readiness
+
+  $1 # check function
+
+  kill "$LS_PID" || true
+}
+
+# $1: pipeline file
+# $2: check function
+# run_logstash 001_es-output.conf check_es_output
+run_logstash() {
+  $CURRENT_DIR/../../bin/logstash -f "$1" 2>/dev/null &
+  export LS_PID=$!
+
+  check_logstash_readiness
+
+  $2 # check function
+
+  kill "$LS_PID" || true
+}
+
+
+# $1: number of try
+# $n: check function with args - return non empty string as pass
+count_down_check() {
+    count=$1
+    while ! [[ $("${@:2}") ]] && [[ $count -gt 0 ]]; do
+        count=$(( count - 1 ))
+        sleep 1
+    done
+
+    [[ $count -eq 0 ]] && echo "1" && return
+
+    echo "Passed check!"
+    echo "0"
+}
+
+
+check_logstash_readiness() {
+  curl_logstash() {
+    [[ $(curl --silent localhost:9600) ]] && echo "0"
+  }
+  check_readiness() {
+    count_down_check 120 curl_logstash
+  }
+  add_check check_readiness "Failed readiness check."
+
+  [[ "${CHECKS[-1]}" -eq "1" ]] && exit 1
+
+  echo "Logstash is Up !"
+  return 0
+}
+
+# $1: jq filter
+# $2: expected value
+# check_logstash_api '.pipelines.main.plugins.outputs[0].documents.successes' '1'
+check_logstash_api() {
+  curl_node_stats() {
+    [[ $(curl --silent localhost:9600/_node/stats | jq "$1") -ge "$2" ]] && echo "0"
+  }
+
+  count_down_check 30 curl_node_stats "$1" "$2"
+}
+
+# add check result to CHECKS
+# $1: check function - expected the last char of result to be 0 or positive number
+# $2: err msg
+add_check() {
+  FEATURE_CHECK=$($1)
+  FEATURE_CHECK="${FEATURE_CHECK: -1}"
+
+  ERR_MSGS+=("$2")
+  CHECKS+=("$FEATURE_CHECK")
+}
+
+# check log if the line contains [ERROR] or [FATAL] and does not relate to "unreachable"
+check_err_log() {
+  LOG_FILE="$CURRENT_DIR/../../logs/logstash-plain.log"
+  LOG_CHECK=$(grep -E "\[ERROR\]|\[FATAL\]" "$LOG_FILE" | grep -cvE "unreachable|Connection refused") || true
+
+  ERR_MSGS+=("Found error in log")
+  CHECKS+=("$LOG_CHECK")
+}
+
+# if CHECKS[i] is 1, print ERR_MSGS[i]
+print_result() {
+  for i in "${!CHECKS[@]}"; do
+    [[ "${CHECKS[$i]}" -gt 0 ]] && echo "${ERR_MSGS[$i]}" || true
+  done
+}
+
+# exit 1 if one of the checks fails
+exit_with_code() {
+  for c in "${CHECKS[@]}"; do
+      [[ $c -gt 0 ]] && exit 1
+  done
+
+  exit 0
+}
+
+clean_up_and_get_result() {
+  [[ -n "$LS_PID" ]] && kill "$LS_PID" || true
+
+  check_err_log
+  print_result
+  exit_with_code
+}
+
+# common setup
+setup() {
+  setup_vault
+  build_logstash
+  trap clean_up_and_get_result INT TERM EXIT
+}

ci/serverless/config/logstash.yml

@@ -0,0 +1,10 @@
+xpack.management.enabled: true
+xpack.management.pipeline.id: ["gen_es"]
+xpack.management.elasticsearch.username: ${ES_USER}
+xpack.management.elasticsearch.password: ${ES_PW}
+xpack.management.elasticsearch.hosts: ["${ES_ENDPOINT}"]
+
+xpack.monitoring.enabled: true
+xpack.monitoring.elasticsearch.username: ${ES_USER}
+xpack.monitoring.elasticsearch.password: ${ES_PW}
+xpack.monitoring.elasticsearch.hosts: ["${ES_ENDPOINT}"]

ci/serverless/cpm_tests.sh

@@ -0,0 +1,48 @@
+#!/usr/bin/env bash
+set -ex
+
+source ./$(dirname "$0")/common.sh
+
+export PIPELINE_NAME='gen_es'
+
+# update pipeline and check response code
+index_pipeline() {
+  RESP_CODE=$(curl -s -w "%{http_code}" -X PUT -u "$ES_USER:$ES_PW" "$ES_ENDPOINT/_logstash/pipeline/$1"  -H 'Content-Type: application/json' -d "$2")
+  if [[ $RESP_CODE -ge '400' ]]; then
+    echo "failed to update pipeline for Central Pipeline Management. Got $RESP_CODE from Elasticsearch"
+    exit 1
+  fi
+}
+
+# index pipeline to serverless ES
+index_cpm_pipelines() {
+  index_pipeline "$PIPELINE_NAME" '{
+    "pipeline": "input { generator { count => 100 } } output { elasticsearch { hosts => \"${ES_ENDPOINT}\" user => \"${ES_USER}\" password => \"${ES_PW}\" index=> \"${INDEX_NAME}\" } }",
+    "last_modified": "2023-07-04T22:22:22.222Z",
+    "pipeline_metadata": { "version": "1"},
+    "username": "log.stash",
+    "pipeline_settings": {"pipeline.batch.delay": "50"}
+  }'
+}
+
+check_es_output() {
+  check_logstash_api '.pipelines.gen_es.plugins.outputs[0].documents.successes' '100'
+}
+
+check_plugin() {
+  add_check check_es_output "Failed central pipeline management check."
+}
+
+delete_pipeline() {
+  curl -u "$ES_USER:$ES_PW" -X DELETE "$ES_ENDPOINT/_logstash/pipeline/$PIPELINE_NAME"  -H 'Content-Type: application/json';
+}
+
+cpm_clean_up_and_get_result() {
+  delete_pipeline
+  clean_up_and_get_result
+}
+
+setup
+trap cpm_clean_up_and_get_result INT TERM EXIT
+index_cpm_pipelines
+run_cpm_logstash check_plugin

ci/serverless/dlq_rspec_tests.sh

@@ -0,0 +1,22 @@
+#!/usr/bin/env bash
+set -ex
+
+vault_path=secret/ci/elastic-logstash/serverless-test
+
+export JRUBY_OPTS="-J-Xmx1g"
+export SERVERLESS=true
+set +x
+export ES_ENDPOINT=$(vault read -field=es_host "${vault_path}")
+export ES_USER=$(vault read -field=es_user "${vault_path}")
+export ES_PW=$(vault read -field=es_user_pw "${vault_path}")
+set -x
+
+./gradlew clean bootstrap assemble installDefaultGems unpackTarDistribution
+./gradlew :logstash-core:copyGemjar
+
+export GEM_PATH=vendor/bundle/jruby/3.1.0
+export GEM_HOME=vendor/bundle/jruby/3.1.0
+
+vendor/jruby/bin/jruby -S bundle install --with development
+
+vendor/jruby/bin/jruby -S bundle exec rspec -fd qa/integration/specs/dlq_spec.rb -e "using pipelines.yml"

ci/serverless/elastic_integration_filter_tests.sh

@@ -0,0 +1,54 @@
+#!/usr/bin/env bash
+set -ex
+
+source ./$(dirname "$0")/common.sh
+
+deploy_ingest_pipeline() {
+  PIPELINE_RESP_CODE=$(curl -s -w "%{http_code}" -o /dev/null -X PUT -u "$ES_USER:$ES_PW" "$ES_ENDPOINT/_ingest/pipeline/integration-logstash_test.events-default" \
+    -H 'Content-Type: application/json' \
+    --data-binary @"$CURRENT_DIR/test_data/ingest_pipeline.json")
+
+  TEMPLATE_RESP_CODE=$(curl -s -w "%{http_code}" -o /dev/null -X PUT -u "$ES_USER:$ES_PW" "$ES_ENDPOINT/_index_template/logs-serverless-default-template" \
+    -H 'Content-Type: application/json' \
+    --data-binary @"$CURRENT_DIR/test_data/index_template.json")
+
+  # ingest pipeline is likely be there from the last run
+  # failing to update pipeline does not stop the test
+  if [[ $PIPELINE_RESP_CODE -ge '400' ]]; then
+    ERR_MSGS+=("Failed to update ingest pipeline. Got $PIPELINE_RESP_CODE")
+  fi
+
+  if [[ $TEMPLATE_RESP_CODE -ge '400' ]]; then
+    ERR_MSGS+=("Failed to update index template. Got $TEMPLATE_RESP_CODE")
+  fi
+}
+
+# processor should append 'serverless' to message
+check_integration_filter() {
+  check_logstash_api '.pipelines.main.plugins.filters[] | select(.id == "mutate1") | .events.out' '1'
+}
+
+get_doc_msg_length() {
+  curl -s -u "$ES_USER:$ES_PW" "$ES_ENDPOINT/logs-$INDEX_NAME.004-default/_search?size=1" | jq '.hits.hits[0]._source.message | length'
+}
+
+# ensure no double run of ingest pipeline
+# message = ['ok', 'serverless*']
+validate_ds_doc() {
+   [[ $(get_doc_msg_length) -eq "2" ]] && echo "0"
+}
+
+check_doc_no_duplication() {
+  count_down_check 20 validate_ds_doc
+}
+
+check_plugin() {
+  add_check check_integration_filter "Failed ingest pipeline processor check."
+  add_check check_doc_no_duplication "Failed ingest pipeline duplication check."
+}
+
+setup
+# install plugin
+"$CURRENT_DIR/../../bin/logstash-plugin" install logstash-filter-elastic_integration
+deploy_ingest_pipeline
+run_logstash "$CURRENT_DIR/pipeline/004_integration-filter.conf" check_plugin

ci/serverless/es_filter_tests.sh

@@ -0,0 +1,16 @@
+#!/usr/bin/env bash
+set -ex
+
+source ./$(dirname "$0")/common.sh
+
+check_es_filter() {
+  check_logstash_api '.pipelines.main.plugins.filters[] | select(.id == "ok") | .events.out' '1'
+}
+
+check_plugin() {
+  add_check check_es_filter "Failed es-filter check."
+}
+
+setup
+index_test_data
+run_logstash "$CURRENT_DIR/pipeline/002_es-filter.conf" check_plugin

ci/serverless/es_input_tests.sh

@@ -0,0 +1,16 @@
+#!/usr/bin/env bash
+set -ex
+
+source ./$(dirname "$0")/common.sh
+
+check_es_input() {
+  check_logstash_api '.pipelines.main.plugins.inputs[0].events.out' '1'
+}
+
+check_plugin() {
+  add_check check_es_input "Failed es-input check."
+}
+
+setup
+index_test_data
+run_logstash "$CURRENT_DIR/pipeline/003_es-input.conf" check_plugin