Optimize ED pipelines

README.md

@@ -107,6 +107,32 @@ From the repository root:
 5. **Downloads results** – Saves to `benchmark_results/<YYYYMMDD_HHMMSS>/`
 6. **Cleans up** – Deletes pipelines and runs `terraform destroy`
 
+### Running a subset
+
+By default `./run.sh` runs every case for every vendor. For faster local
+iteration you can restrict the run with two optional flags:
+
+```bash
+# Only the pass-through and mask cases (all vendors):
+./run.sh --cases pass-through,mask
+
+# Only Edge Delta and Cribl (all cases):
+./run.sh --vendors edgedelta,cribl
+
+# Edge Delta, filter case only:
+./run.sh --vendors edgedelta --cases filter
+```
+
+- `--cases` accepts `pass-through`, `filter`, `mask`, `lookup` (`passthrough` is
+  accepted as an alias for `pass-through`). Values may be comma- or
+  space-separated.
+- `--vendors` accepts `edgedelta`, `bindplane`, `cribl`, `otelcol`.
+- Prerequisite checks (env vars, Bindplane CLI) only run for the selected
+  vendors, so you don't need Cribl credentials to run an Edge Delta–only pass.
+- `otelcol` has no `lookup` case; it is skipped automatically if `lookup` is the
+  only selected case.
+- Run `./run.sh --help` for the full usage.
+
 **Note:** Run from the repository root. The script sources `functions.sh` and expects to execute from that directory.
 
 ## Benchmark Types

pipelines/edgedelta/filter.yaml

@@ -53,7 +53,7 @@ nodes:
   parallel_worker_count: 256
   channel_size: 20000
   read_timeout: 1m0s
-  log_parsing_mode: json
+  log_parsing_mode: basic
 - name: http_input_b410_multiprocessor
   type: sequence
   user_description: Multi Processor
@@ -136,12 +136,11 @@ nodes:
   type: s3_output
   user_description: S3 Destination
   schema: Archive
-  compression: gzip
+  compression: zstd
   region: us-west-2
-  parallel_worker_count: 2
+  parallel_worker_count: 6
   bucket: {S3_PLACEHOLDER}
   encoding: json
-  use_native_compression: false
   disable_metadata_ingestion: false
   path_prefix:
     order:

pipelines/edgedelta/lookup.yaml

@@ -53,7 +53,7 @@ nodes:
   parallel_worker_count: 256
   channel_size: 20000
   read_timeout: 1m0s
-  log_parsing_mode: json
+  log_parsing_mode: basic
 - name: http_input_b410_multiprocessor
   type: sequence
   user_description: Multi Processor
@@ -136,12 +136,11 @@ nodes:
   type: s3_output
   user_description: S3 Destination
   schema: Archive
-  compression: gzip
+  compression: zstd
   region: us-west-2
-  parallel_worker_count: 2
+  parallel_worker_count: 6
   bucket: {S3_PLACEHOLDER}
   encoding: json
-  use_native_compression: false
   disable_metadata_ingestion: false
   path_prefix:
     order:

pipelines/edgedelta/mask.yaml

@@ -53,7 +53,7 @@ nodes:
   parallel_worker_count: 256
   channel_size: 20000
   read_timeout: 1m0s
-  log_parsing_mode: json
+  log_parsing_mode: basic
 - name: http_input_b410_multiprocessor
   type: sequence
   user_description: Multi Processor
@@ -136,12 +136,11 @@ nodes:
   type: s3_output
   user_description: S3 Destination
   schema: Archive
-  compression: gzip
+  compression: zstd
   region: us-west-2
-  parallel_worker_count: 2
+  parallel_worker_count: 6
   bucket: {S3_PLACEHOLDER}
   encoding: json
-  use_native_compression: false
   disable_metadata_ingestion: false
   path_prefix:
     order:

pipelines/edgedelta/pass-through.yaml

@@ -53,7 +53,7 @@ nodes:
   parallel_worker_count: 256
   channel_size: 20000
   read_timeout: 1m0s
-  log_parsing_mode: json
+  log_parsing_mode: basic
 - name: http_input_b410_multiprocessor
   type: sequence
   user_description: Multi Processor
@@ -138,12 +138,11 @@ nodes:
   type: s3_output
   user_description: S3 Destination
   schema: Archive
-  compression: gzip
+  compression: zstd
   region: us-west-2
-  parallel_worker_count: 2
+  parallel_worker_count: 6
   bucket: {S3_PLACEHOLDER}
   encoding: json
-  use_native_compression: false
   disable_metadata_ingestion: false
   path_prefix:
     order:

run.sh

@@ -10,23 +10,164 @@ ed_benchmark=0
 bindplane_benchmark=0
 cribl_benchmark=0
 
-function check_prerequisites() {
-  # Edge Delta variables
-  check_vars "ED_ORG_ID" "ED_API_TOKEN" 
-  # Cribl variables
-  check_vars "CRIBL_WORKSPACE" "CRIBL_ORG" "CRIBL_WORKER_GROUP" "CRIBL_CLIENT_ID" "CRIBL_CLIENT_SECRET" "CRIBL_LEADER_TOKEN"
-
-  if ! command -v bindplane &> /dev/null; then
-    echo "Bindplane CLI is not installed"
-    install_bindplane_cli
+# All pipeline cases and vendors the suite knows about, in canonical run order.
+ALL_CASES=("pass-through" "filter" "mask" "lookup")
+ALL_VENDORS=("edgedelta" "bindplane" "cribl" "otelcol")
+
+# Selections default to everything (preserves the previous "run all" behaviour).
+SELECTED_CASES=("${ALL_CASES[@]}")
+SELECTED_VENDORS=("${ALL_VENDORS[@]}")
+
+usage() {
+  cat <<EOF
+Usage: ./run.sh [--cases <list>] [--vendors <list>]
+
+Run the pipeline benchmark suite. With no flags, every case runs for every
+vendor (the previous default behaviour).
+
+Options:
+  --cases <list>     Comma- or space-separated pipeline cases to run.
+                     Valid: ${ALL_CASES[*]}
+                     Example: --cases pass-through,mask
+  --vendors <list>   Comma- or space-separated vendors to run.
+                     Valid: ${ALL_VENDORS[*]}
+                     Example: --vendors edgedelta,cribl
+  -h, --help         Show this help and exit.
+
+Notes:
+  * "passthrough" is accepted as an alias for "pass-through".
+  * otelcol has no lookup case; it is skipped automatically if requested.
+  * Prerequisite checks only run for the selected vendors.
+EOF
+}
+
+# contains reports whether the first argument equals any of the remaining ones.
+contains() {
+  local needle="$1"; shift
+  local item
+  for item in "$@"; do
+    [[ "$item" == "$needle" ]] && return 0
+  done
+  return 1
+}
+
+is_vendor_selected() {
+  contains "$1" "${SELECTED_VENDORS[@]}"
+}
+
+# vendor_supported_cases echoes the cases a given vendor can actually run.
+vendor_supported_cases() {
+  case "$1" in
+    otelcol) echo "pass-through filter mask" ;;
+    *)       echo "pass-through filter mask lookup" ;;
+  esac
+}
+
+# cases_for echoes the selected cases that the given vendor supports, in
+# canonical order. Empty output means "nothing to run for this vendor".
+cases_for() {
+  local vendor="$1"
+  local supported
+  supported=$(vendor_supported_cases "$vendor")
+  local c out=()
+  for c in "${ALL_CASES[@]}"; do
+    if contains "$c" "${SELECTED_CASES[@]}" && contains "$c" $supported; then
+      out+=("$c")
+    fi
+  done
+  echo "${out[*]}"
+}
+
+# maybe_run executes the benchmark function only if its vendor is selected and
+# at least one selected case applies to it.
+maybe_run() {
+  local vendor="$1" fn="$2"
+  if ! is_vendor_selected "$vendor"; then
+    echo "Skipping $vendor benchmark (vendor not selected)"
+    return 0
+  fi
+  if [[ -z "$(cases_for "$vendor")" ]]; then
+    echo "Skipping $vendor benchmark (no selected case applies; selected: ${SELECTED_CASES[*]})"
+    return 0
   fi
-  # Bindplane API key check
-  if ! bindplane profile get | grep -q apiKey; then
-    echo "Bindplane CLI is not installed or not configured"
-    echo "Please install the Bindplane CLI and configure it with your API key"
+  "$fn"
+}
+
+# parse_selection validates a comma/space-separated list against the allowed
+# values for the given kind, normalises aliases, and writes the deduplicated
+# result into the PARSED_SELECTION global array. It exits on invalid input.
+PARSED_SELECTION=()
+parse_selection() {
+  local kind="$1" raw_list="$2"
+  local -a allowed=()
+  if [[ "$kind" == "cases" ]]; then
+    allowed=("${ALL_CASES[@]}")
+  else
+    allowed=("${ALL_VENDORS[@]}")
+  fi
+  PARSED_SELECTION=()
+  local raw v
+  for raw in ${raw_list//,/ }; do
+    v="$raw"
+    [[ "$kind" == "cases" && "$v" == "passthrough" ]] && v="pass-through"
+    if ! contains "$v" "${allowed[@]}"; then
+      echo "Invalid $kind: '$raw' (valid: ${allowed[*]})" >&2
+      exit 1
+    fi
+    contains "$v" "${PARSED_SELECTION[@]}" || PARSED_SELECTION+=("$v")
+  done
+  if [[ ${#PARSED_SELECTION[@]} -eq 0 ]]; then
+    echo "No valid $kind selected" >&2
     exit 1
   fi
-  get_bindplane_installation_command >> "$git_root/scripts/install_agent_bindplane.sh"
+}
+
+while [[ $# -gt 0 ]]; do
+  case "$1" in
+    --cases)
+      [[ -n "${2:-}" ]] || { echo "--cases requires a value" >&2; usage; exit 1; }
+      parse_selection cases "$2"; SELECTED_CASES=("${PARSED_SELECTION[@]}"); shift 2 ;;
+    --cases=*)
+      parse_selection cases "${1#*=}"; SELECTED_CASES=("${PARSED_SELECTION[@]}"); shift ;;
+    --vendors)
+      [[ -n "${2:-}" ]] || { echo "--vendors requires a value" >&2; usage; exit 1; }
+      parse_selection vendors "$2"; SELECTED_VENDORS=("${PARSED_SELECTION[@]}"); shift 2 ;;
+    --vendors=*)
+      parse_selection vendors "${1#*=}"; SELECTED_VENDORS=("${PARSED_SELECTION[@]}"); shift ;;
+    -h|--help)
+      usage; exit 0 ;;
+    *)
+      echo "Unknown argument: $1" >&2; usage; exit 1 ;;
+  esac
+done
+
+echo "Selected vendors: ${SELECTED_VENDORS[*]}"
+echo "Selected cases:   ${SELECTED_CASES[*]}"
+
+function check_prerequisites() {
+  if is_vendor_selected "edgedelta"; then
+    # Edge Delta variables
+    check_vars "ED_ORG_ID" "ED_API_TOKEN"
+  fi
+
+  if is_vendor_selected "cribl"; then
+    # Cribl variables
+    check_vars "CRIBL_WORKSPACE" "CRIBL_ORG" "CRIBL_WORKER_GROUP" "CRIBL_CLIENT_ID" "CRIBL_CLIENT_SECRET" "CRIBL_LEADER_TOKEN"
+  fi
+
+  if is_vendor_selected "bindplane"; then
+    if ! command -v bindplane &> /dev/null; then
+      echo "Bindplane CLI is not installed"
+      install_bindplane_cli
+    fi
+    # Bindplane API key check
+    if ! bindplane profile get | grep -q apiKey; then
+      echo "Bindplane CLI is not installed or not configured"
+      echo "Please install the Bindplane CLI and configure it with your API key"
+      exit 1
+    fi
+    get_bindplane_installation_command >> "$git_root/scripts/install_agent_bindplane.sh"
+  fi
 }
 
 function create_benchmark_environment() {
@@ -82,7 +223,7 @@ function run_ed_benchmark() {
   fi
   echo "Installing Edge Delta agent..."
   run_scripts_on_ec2_instance "$git_root/scripts/install_agent_ed.sh"
-  for type in "pass-through" "filter" "mask" "lookup"; do
+  for type in $(cases_for edgedelta); do
     ./edgedelta-api.sh full $type.yaml
     trigger_benchmark "edgedelta" $type
   done
@@ -100,7 +241,7 @@ function run_bindplane_benchmark() {
   upload_file_to_ec2_instance "$git_root/scripts/update_agent_bindplane.sh" /tmp/update_agent_bindplane.sh
   run_command_on_ec2_instance "chmod +x /tmp/update_agent_bindplane.sh"
   run_command_on_ec2_instance "/tmp/update_agent_bindplane.sh"
-  for type in "pass-through" "filter" "mask" "lookup"; do
+  for type in $(cases_for bindplane); do
     bindplane apply -f "$type.yaml"
     bindplane rollout start benchmark
     trigger_benchmark "bindplane" $type
@@ -124,7 +265,7 @@ function run_cribl_benchmark() {
   run_command_on_ec2_instance "chmod +x /tmp/install_agent_cribl.sh"
   run_command_on_ec2_instance "sudo -E /tmp/install_agent_cribl.sh"
 
-  for type in "pass-through" "filter" "mask" "lookup"; do
+  for type in $(cases_for cribl); do
     ./cribl-api.sh update-pipeline $type.json
     trigger_benchmark "cribl" $type
   done
@@ -136,8 +277,9 @@ function run_otelcol_benchmark() {
   echo "Installing OpenTelemetry Collector agent..."
   run_scripts_on_ec2_instance "$git_root/scripts/install_agent_otelcol.sh"
   upload_folder_to_ec2_instance "$git_root/pipelines/otelcol"
-  # No lookup scenario: OTel contrib has no shipped CSV lookup processor.
-  for type in "pass-through" "filter" "mask"; do
+  # No lookup scenario: OTel contrib has no shipped CSV lookup processor
+  # (lookup is excluded automatically by cases_for).
+  for type in $(cases_for otelcol); do
     run_command_on_ec2_instance "sudo cp /home/ubuntu/otelcol/$type.yaml /etc/otelcol-contrib/config.yaml"
     trigger_benchmark "otelcol" $type
   done
@@ -187,9 +329,9 @@ check_prerequisites
 trap cleanup_benchmark_environment EXIT
 create_benchmark_environment
 prepare_for_benchmark
-run_ed_benchmark
-run_bindplane_benchmark
-run_cribl_benchmark
-run_otelcol_benchmark
+maybe_run edgedelta run_ed_benchmark
+maybe_run bindplane run_bindplane_benchmark
+maybe_run cribl run_cribl_benchmark
+maybe_run otelcol run_otelcol_benchmark
 download_benchmark_results
 generate_versions_csv