From eec8d4a4313672198b1768c9ee33505bcae06104 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Sat, 12 Sep 2026 20:20:11 -0400 Subject: [PATCH 1/2] Add offline cluster data exporter tool --- tools/cluster-data-exporter/.dockerignore | 9 + tools/cluster-data-exporter/.gitignore | 2 + tools/cluster-data-exporter/Cargo.toml | 23 + tools/cluster-data-exporter/Dockerfile | 39 ++ tools/cluster-data-exporter/README.md | 273 +++++++++++ .../bin/alibaba/sort_and_format.sh | 195 ++++++++ tools/cluster-data-exporter/compose.yaml | 24 + .../docker-compose.yml.j2 | 41 ++ ...alibaba-msresource-2021-docker-compose.yml | 31 ++ ...alibaba-msresource-2022-docker-compose.yml | 32 ++ .../alibaba-node-2021-docker-compose.yml | 32 ++ .../alibaba-node-2022-docker-compose.yml | 32 ++ .../base-docker-compose.yml | 39 ++ .../google-docker-compose.yml | 32 ++ .../installation/install.sh | 12 + tools/cluster-data-exporter/prometheus.yml | 7 + .../scripts/generate_docker_compose.py | 253 ++++++++++ .../scripts/requirements.txt | 2 + .../src/alibaba_metrics.rs | 71 +++ .../src/alibaba_metrics/ms_resource.rs | 251 ++++++++++ .../src/alibaba_metrics/node.rs | 235 +++++++++ .../src/google_metrics.rs | 453 ++++++++++++++++++ tools/cluster-data-exporter/src/main.rs | 274 +++++++++++ tools/cluster-data-exporter/src/utilities.rs | 133 +++++ .../tests/fixtures/alibaba/msresource.csv | 3 + .../tests/fixtures/alibaba/node.csv | 3 + .../fixtures/google/part-00000-of-00500.csv | 2 + tools/cluster-data-exporter/tests/smoke.sh | 103 ++++ 28 files changed, 2606 insertions(+) create mode 100644 tools/cluster-data-exporter/.dockerignore create mode 100644 tools/cluster-data-exporter/.gitignore create mode 100644 tools/cluster-data-exporter/Cargo.toml create mode 100644 tools/cluster-data-exporter/Dockerfile create mode 100644 tools/cluster-data-exporter/README.md create mode 100755 tools/cluster-data-exporter/bin/alibaba/sort_and_format.sh create mode 100644 tools/cluster-data-exporter/compose.yaml create mode 100644 tools/cluster-data-exporter/docker-compose.yml.j2 create mode 100644 tools/cluster-data-exporter/docker_compose_frames/alibaba-msresource-2021-docker-compose.yml create mode 100644 tools/cluster-data-exporter/docker_compose_frames/alibaba-msresource-2022-docker-compose.yml create mode 100644 tools/cluster-data-exporter/docker_compose_frames/alibaba-node-2021-docker-compose.yml create mode 100644 tools/cluster-data-exporter/docker_compose_frames/alibaba-node-2022-docker-compose.yml create mode 100644 tools/cluster-data-exporter/docker_compose_frames/base-docker-compose.yml create mode 100644 tools/cluster-data-exporter/docker_compose_frames/google-docker-compose.yml create mode 100755 tools/cluster-data-exporter/installation/install.sh create mode 100644 tools/cluster-data-exporter/prometheus.yml create mode 100644 tools/cluster-data-exporter/scripts/generate_docker_compose.py create mode 100644 tools/cluster-data-exporter/scripts/requirements.txt create mode 100644 tools/cluster-data-exporter/src/alibaba_metrics.rs create mode 100644 tools/cluster-data-exporter/src/alibaba_metrics/ms_resource.rs create mode 100644 tools/cluster-data-exporter/src/alibaba_metrics/node.rs create mode 100644 tools/cluster-data-exporter/src/google_metrics.rs create mode 100644 tools/cluster-data-exporter/src/main.rs create mode 100644 tools/cluster-data-exporter/src/utilities.rs create mode 100644 tools/cluster-data-exporter/tests/fixtures/alibaba/msresource.csv create mode 100644 tools/cluster-data-exporter/tests/fixtures/alibaba/node.csv create mode 100644 tools/cluster-data-exporter/tests/fixtures/google/part-00000-of-00500.csv create mode 100755 tools/cluster-data-exporter/tests/smoke.sh diff --git a/tools/cluster-data-exporter/.dockerignore b/tools/cluster-data-exporter/.dockerignore new file mode 100644 index 00000000..1634158c --- /dev/null +++ b/tools/cluster-data-exporter/.dockerignore @@ -0,0 +1,9 @@ +target/ +.git/ +.gitignore +README.md +data/ +*.csv +*.gz +docker_compose_frames/ +scripts/ diff --git a/tools/cluster-data-exporter/.gitignore b/tools/cluster-data-exporter/.gitignore new file mode 100644 index 00000000..1e7caa9e --- /dev/null +++ b/tools/cluster-data-exporter/.gitignore @@ -0,0 +1,2 @@ +Cargo.lock +target/ diff --git a/tools/cluster-data-exporter/Cargo.toml b/tools/cluster-data-exporter/Cargo.toml new file mode 100644 index 00000000..f2bb5222 --- /dev/null +++ b/tools/cluster-data-exporter/Cargo.toml @@ -0,0 +1,23 @@ +[package] +name = "cluster_data_exporter" +version = "0.1.0" +edition = "2021" + +[dependencies] +prometheus = "0.14.0" +tokio = { version = "1", features = ["full"] } +hyper = { version = "1", features = ["full"] } +hyper-util = { version = "0.1", features = ["full"] } +lazy_static = "1.5" +csv = "1.3" +serde = { version = "1.0", features = ["derive"] } +concurrent-queue = "2.5.0" +flate2 = "1.1.2" +clap = { version = "4.5.41", features = ["derive"] } +tracing = "0.1" +tracing-subscriber = { version = "0.3", features = ["env-filter"] } +tracing-appender = "0.2" + +# Keep this replay source separate from ASAPQuery-backend's production +# workspace and dependency graph. +[workspace] diff --git a/tools/cluster-data-exporter/Dockerfile b/tools/cluster-data-exporter/Dockerfile new file mode 100644 index 00000000..582d812c --- /dev/null +++ b/tools/cluster-data-exporter/Dockerfile @@ -0,0 +1,39 @@ +# Use the official Rust image as a build environment +FROM rust:latest AS builder + +# Set the working directory inside the container +WORKDIR /usr/src/app + +# Copy the manifest before the source tree. This is intentionally a standalone +# Cargo project, so Cargo resolves its lockfile during the image build. +COPY Cargo.toml ./ + +# Copy the source code +COPY src ./src + +# Build the application in release mode +RUN cargo build --release + +# Use a minimal runtime image +FROM debian:bookworm-slim + +# Install necessary runtime dependencies +RUN apt-get update && apt-get install -y \ + ca-certificates \ + && rm -rf /var/lib/apt/lists/* + +# Create a non-root user +RUN useradd -r -s /bin/false exporter + +# Create the data and output directories that will be mounted as volumes +RUN mkdir -p /data /output && chown exporter:exporter /data /output + +# Copy the binary from the builder stage +COPY --from=builder /usr/src/app/target/release/cluster_data_exporter /usr/local/bin/cluster_data_exporter + +# Change to the non-root user +USER exporter + +# Set the entrypoint to the binary +# All arguments including port and input directory must be provided via docker run or docker-compose +ENTRYPOINT ["cluster_data_exporter"] diff --git a/tools/cluster-data-exporter/README.md b/tools/cluster-data-exporter/README.md new file mode 100644 index 00000000..0d31d179 --- /dev/null +++ b/tools/cluster-data-exporter/README.md @@ -0,0 +1,273 @@ +# CLUSTER DATA EXPORTER + +A Prometheus exporter that exposes cluster resource usage metrics from Google and Alibaba cluster trace datasets. + +This is an evaluation tool. It is deliberately outside the backend Cargo +workspace: it replays offline CSV traces through Prometheus's normal scrape +path, after which Prometheus may Remote Write into ASAPQuery-backend. + +Finite inputs retain the historical immediate shutdown behavior by default. +For a final scrape grace period, pass `--exit-after-eof-ms=`; +the smoke test uses this option so it can inspect the terminal metric values. + +## DESCRIPTION + +This exporter reads CSV data from certain datasets provided by Google or Alibaba and exposes them as Prometheus metrics. The exporter supports both Google task resource usage data from 2011 and Alibaba node and microservice resource data from 2021 and 2022. Instructions for downloading this data are linked in this document. + +## INSTALLATION + +### Prerequisites + +- Rust 1.70+ (edition 2021) +- Access to Google or Alibaba cluster datasets + +### Building + +```bash +cargo build --release +``` + +Run the synthetic provider smoke checks (no trace dataset required): + +```bash +./tests/smoke.sh +``` + +## USAGE + +```bash +cluster_data_exporter -i -p [OPTIONS] +``` + +### Google Provider + +```bash +cluster_data_exporter -i ./google/clusterdata-2011/ -p 8080 google [OPTIONS] +``` + +### Alibaba Provider + +```bash +cluster_data_exporter -i ./alibaba/2021/ -p 8080 alibaba [OPTIONS] +``` + +## DATA SOURCES + +### Google Cluster Data + +Instructions on how to download the Google Cluster 2011 task usage data: +https://github.com/google/cluster-data/blob/master/ClusterData2011_2.md + +The only part of the dataset used by the exporter is the task_usage section, so there's no need to install the whole dataset + +Expected directory structure: +``` +path/to/task/resource/usage/dir/ +├── part-00000-of-00500.csv.gz +├── part-00001-of-00500.csv.gz +└── ... +``` + +### Alibaba Cluster Data + +Instructions on downloading the Alibaba microservice trace datasets: +- 2021: https://github.com/alibaba/clusterdata/blob/master/cluster-trace-microservices-v2021/README.md#introduction-of-trace-data +- 2022: https://github.com/alibaba/clusterdata/tree/master/cluster-trace-microservices-v2022#trace-data-download + +The only parts of the datasets used by the exporter are the Node and MSResource sections, the rest can be discarded. + +Expected directory structure (after preprocessing): + +2021 Data: +``` +path/to/Node/ +├── Node_0.csv.gz +├── Node_1.csv.gz +└── ... + +path/to/MSResource/ +├── MSResource_0.csv.gz +├── MSResource_1.csv.gz +└── ... +``` + +2022 Data: +``` +path/to/NodeMetrics/ +├── NodeMetrics_0.csv.gz +├── NodeMetrics_1.csv.gz +└── ... + +path/to/MSMetrics/ +├── MSMetrics_0.csv.gz +├── MSMetrics_1.csv.gz +└── ... +``` + +## DATA PREPROCESSING FOR ALIBABA + +IMPORTANT: Before running the exporter on Alibaba data, you must run the preprocessing script to sort the data by timestamp and recompress it as a .csv.gz: + +```bash +./bin/alibaba/sort_and_format.sh --year <2021|2022> [-n] [-m] +``` + +This script extracts, sorts by timestamp, and recompresses the Alibaba CSV files in a format the exporter can read (.csv.gz). The sorting is necessary because some datasets (mainly 2022 data) are not sorted by timestamp, which is required for proper metric export timing. + +### Input Directory Structure + +The input directory should contain one or both of the subdirectories with unprocessed files, i.e. the untouched /data/ directory created from running the fetchData.sh scripts from the Alibaba github repos. For example: + +``` +alibaba/2021/data/ +├── Node/ +│ ├── Node_0.tar.gz +│ ├── Node_1.tar.gz +│ └── ... +└── MSResource/ + ├── MSResource_0.tar.gz + ├── MSResource_1.tar.gz + └── ... + +alibaba/2022/data/ +├── NodeMetrics/ +│ ├── NodeMetrics_0.tar.gz +│ ├── NodeMetrics_1.tar.gz +│ └── ... +└── MSMetrics/ + ├── MSMetrics_0.tar.gz + ├── MSMetrics_1.tar.gz + └── ... +``` + +Examples: + +```bash +# Process 2021 Node data +./bin/alibaba/sort_and_format.sh alibaba/2021/data --year 2021 -n + +# Process 2021 MSResource data +./bin/alibaba/sort_and_format.sh alibaba/2021/data --year 2021 -m + +# Process both Node and MSResource data for 2021 +./bin/alibaba/sort_and_format.sh alibaba/2021/data --year 2021 -n -m +``` + +## COMMAND LINE ARGUMENTS + +- -i, --input-directory: Path to the directory containing CSV data files +- -p, --port: Port number for the HTTP server +- --exit-after-eof-ms: Optional grace period after finite input is exhausted + (default: 0, immediate shutdown) + +### Provider-specific Options + +#### Google +- --metrics: Specific metrics to export from task resource usage data +- --all-parts: Process all CSV parts (default behavior) +- --part-index: Process only a specific part index (0-499) + +#### Alibaba +- --data-type: Type of data to export (node or msresource) +- --data-year: Year of the dataset (2021 or 2022) +- --all-parts: Process all CSV parts (default behavior) +- --part-index: Process only a specific part index +- --speedup: Alibaba replay speedup factor (1 = real time) + +## DOCKER USAGE + +### Prerequisites for Docker + +1. Download and preprocess your CSV data as described in the DATA SOURCES section above +2. Place the preprocessed data in a local directory (e.g., `./data/`) + +### Building and Running with Docker + +Build the Docker image: +```bash +docker build -t cluster-data-exporter . +``` + +Run with Docker (example for Google data): +```bash +docker run -v ./data:/data:ro -p 40000:40000 cluster-data-exporter \ + --input-directory /data \ + --port 40000 \ + google \ + --metrics mean_cpu_usage_rate,canonical_memory_usage \ + --all-parts +``` + +Run with Docker (example for Alibaba data): +```bash +docker run -v ./data:/data:ro -p 40000:40000 cluster-data-exporter \ + --input-directory /data \ + --port 40000 \ + alibaba \ + --data-type node \ + --data-year 2021 \ + --all-parts +``` + +### Using Docker Compose + +`compose.yaml` is the smallest end-to-end source example: it runs the exporter +and a Prometheus scraper for Google data. Set `CDE_DATA_DIR` to the preprocessed +task-usage directory, then run: + +```bash +CDE_DATA_DIR=/path/to/task/resource/usage docker compose up --build +``` + +Prometheus is then available at `http://localhost:9090`. To send those samples +to ASAPQuery-backend, configure Remote Write through the backend's existing +profile or demo rather than duplicating backend startup here. + +#### Automated Generation with Python Script + +The `scripts/generate_docker_compose.py` script automatically generates docker-compose.yml files from the frame templates and fill in certain fields. + +**Google Provider Example:** +```bash +python scripts/generate_docker_compose.py google --metrics mean_cpu_usage_rate,max_cpu_usage --port 8080 --input-dir ./data +``` + +**Alibaba Provider Example:** +```bash +python scripts/generate_docker_compose.py alibaba --data-type node --data-year 2021 --port 8080 --input-dir ./data +``` + +The script will: +- Validate your configuration options +- Generate a docker-compose.yml file with correct settings +- Update port mappings and volume mounts automatically + +#### Manual Setup with Frame Files + +Alternatively, the `docker_compose_frames/` directory contains pre-configured docker-compose files for different providers and configurations. These frame files will still require small edits before running docker-compose, see each frame file for more information. + +- **Google Provider**: `google-docker-compose.yml` - Edit list of metrics to export +- **Alibaba Provider**: Provider-specific frames for each data type and year combination: + - `alibaba-node-2021-docker-compose.yml` + - `alibaba-node-2022-docker-compose.yml` + - `alibaba-msresource-2021-docker-compose.yml` + - `alibaba-msresource-2022-docker-compose.yml` + +To use a frame file: +1. Copy the appropriate frame file from `docker_compose_frames/` to your working directory as `docker-compose.yml` +2. Edit the file with any options that still need to be filled in (marked with "CHANGE THIS" comments) +3. Run: `docker-compose up -d` + +### Data Volume Requirements + +- The container expects data to be mounted at `/data` +- Data must be preprocessed according to the instructions in the DATA SOURCES section +- For Alibaba data, ensure you've run the sorting and compression scripts before mounting +- Mount the volume as read-only (`:ro`) + +## METRICS ENDPOINT + +Once running, metrics are available at: +``` +http://localhost:/metrics +``` diff --git a/tools/cluster-data-exporter/bin/alibaba/sort_and_format.sh b/tools/cluster-data-exporter/bin/alibaba/sort_and_format.sh new file mode 100755 index 00000000..1cce61dd --- /dev/null +++ b/tools/cluster-data-exporter/bin/alibaba/sort_and_format.sh @@ -0,0 +1,195 @@ +#!/bin/bash + +# Script to process alibaba Node and MSResource data files with year-specific configurations +# Usage: ./sort_and_format.sh --year [-n] [-m] + +usage() { + echo "Usage: $0 --year [-n] [-m]" + echo " Path to directory containing data subdirectories" + echo " --year Year of data (2021 or 2022) - REQUIRED" + echo " -n Clean Node csv files and recompress as .csv.gz" + echo " -m Clean MSResource/MSMetrics csv files and recompress as .csv.gz" + echo " At least one of -n or -m must be specified" + echo "" + echo "Year-specific configurations:" + echo " 2022: Uses NodeMetrics/ and MSMetrics/ subdirectories with NodeMetrics_*.tar.gz and MSMetrics_*.tar.gz files" + echo " Timestamp in first column for both" + echo " 2021: Uses Node/ and MSResource/ subdirectories with Node_*.tar.gz and MSResource_*.tar.gz files" + echo " Timestamp in second column for Node data, seventh column for MSResource data" + exit 1 +} + +# Function to process files in a given directory with a specific pattern +process_files() { + local subdir="$1" + local pattern="$2" + local timestamp_col="$3" + local full_path="${INPUT_DIR}/${subdir}" + + if [[ ! -d "$full_path" ]]; then + echo "Warning: Directory $full_path does not exist, skipping..." + return + fi + + echo "Processing files in $full_path" + + # Find all files matching the pattern, sorted by index + local files=() + mapfile -t files < <(find "$full_path" -maxdepth 1 -type f -name "${pattern}_*.tar.gz" -print | sort -V) + + if [[ ${#files[@]} -eq 0 ]]; then + echo "No files matching ${pattern}_*.tar.gz found in $full_path" + return + fi + + echo "Found ${#files[@]} files in $subdir:" + + for file in "${files[@]}"; do + echo "Processing: $(basename "$file")" + + # Create temporary directory for processing + local temp_dir + temp_dir=$(mktemp -d) + local base_name + base_name=$(basename "$file" .tar.gz) + + echo " -> Extracting $file to temporary directory..." + if ! tar -xzf "$file" -C "$temp_dir"; then + echo " -> Error: Failed to extract $file" + rm -rf "$temp_dir" + continue + fi + + # Find the extracted CSV file + local csv_file + csv_file=$(find "$temp_dir" -name "*.csv" -type f | head -1) + if [[ -z "$csv_file" ]]; then + echo " -> Error: No CSV file found in extracted archive" + rm -rf "$temp_dir" + continue + fi + + # Check if file is already sorted using sort -c + echo " -> Checking if file is already sorted..." + if tail -n +2 "$csv_file" | sort -t',' -k"${timestamp_col}","${timestamp_col}"n -c 2>/dev/null; then + echo " -> File is already sorted, skipping sort step" + else + echo " -> Sorting CSV file by timestamp (column $timestamp_col)..." + # Use external sort for memory efficiency with large files + # Preserve header line by extracting first line, sorting the rest, then combining + # -t',' specifies comma as field separator + # -k${timestamp_col},${timestamp_col}n sorts by specified field numerically + # -S 1G uses 1GB of memory for sorting (adjust if needed) + # --temporary-directory ensures temp files go to a writable location + local sorted_file="${temp_dir}/sorted.csv" + if ! (head -n 1 "$csv_file"; tail -n +2 "$csv_file" | sort -t',' -k"${timestamp_col}","${timestamp_col}"n -S 1G --temporary-directory="$temp_dir") > "$sorted_file"; then + echo " -> Error: Failed to sort CSV file" + rm -rf "$temp_dir" + continue + fi + mv "$sorted_file" "$csv_file" + fi + + echo " -> Compressing sorted file..." + local output_file="${full_path}/${base_name}.csv.gz" + if ! gzip -c "$csv_file" > "$output_file"; then + echo " -> Error: Failed to compress sorted file" + rm -rf "$temp_dir" + continue + fi + + echo " -> Successfully processed: $(basename "$output_file")" + + # Clean up temporary directory + rm -rf "$temp_dir" + done +} + +# Parse command line arguments +if [[ $# -lt 4 ]]; then + usage +fi + +INPUT_DIR="$1" +shift + +YEAR="" +PROCESS_NODE=false +PROCESS_MS=false + +while [[ $# -gt 0 ]]; do + case $1 in + --year) + YEAR="$2" + if [[ "$YEAR" != "2021" && "$YEAR" != "2022" ]]; then + echo "Error: Year must be either 2021 or 2022" + usage + fi + shift 2 + ;; + -n) + PROCESS_NODE=true + shift + ;; + -m) + PROCESS_MS=true + shift + ;; + *) + echo "Unknown option: $1" + usage + ;; + esac +done + +# Validate required arguments +if [[ -z "$YEAR" ]]; then + echo "Error: --year parameter is required" + usage +fi + +# Validate input directory +if [[ ! -d "$INPUT_DIR" ]]; then + echo "Error: Input directory '$INPUT_DIR' does not exist" + exit 1 +fi + +# Check that at least one flag is specified +if [[ "$PROCESS_NODE" == false && "$PROCESS_MS" == false ]]; then + echo "Error: At least one of -n or -m must be specified" + usage +fi + +echo "Input directory: $INPUT_DIR" +echo "Year: $YEAR" +echo "Process Node data: $PROCESS_NODE" +echo "Process MSResource data: $PROCESS_MS" +echo + +# Configure year-specific settings +if [[ "$YEAR" == "2022" ]]; then + NODE_SUBDIR="NodeMetrics" + MS_SUBDIR="MSMetrics" + NODE_PATTERN="NodeMetrics" + MS_PATTERN="MSMetrics" + NODE_TIMESTAMP_COL=1 + MS_TIMESTAMP_COL=1 +else # 2021 + NODE_SUBDIR="Node" + MS_SUBDIR="MSResource" + NODE_PATTERN="Node" + MS_PATTERN="MSResource" + NODE_TIMESTAMP_COL=2 + MS_TIMESTAMP_COL=7 +fi + +# Process files based on flags +if [[ "$PROCESS_NODE" == true ]]; then + process_files "$NODE_SUBDIR" "$NODE_PATTERN" "$NODE_TIMESTAMP_COL" +fi + +if [[ "$PROCESS_MS" == true ]]; then + process_files "$MS_SUBDIR" "$MS_PATTERN" "$MS_TIMESTAMP_COL" +fi + +echo "Processing complete!" diff --git a/tools/cluster-data-exporter/compose.yaml b/tools/cluster-data-exporter/compose.yaml new file mode 100644 index 00000000..d2e4bb72 --- /dev/null +++ b/tools/cluster-data-exporter/compose.yaml @@ -0,0 +1,24 @@ +services: + cluster-data-exporter: + build: . + volumes: + - ${CDE_DATA_DIR:?set CDE_DATA_DIR to the preprocessed Google task-usage directory}:/data:ro + ports: + - "40000:40000" + command: + - --input-directory + - /data + - --port + - "40000" + - google + - --metrics=${CDE_GOOGLE_METRICS:-mean-cpu-usage-rate} + - --part-index=${CDE_PART_INDEX:-0} + + prometheus: + image: prom/prometheus:v2.55.1 + depends_on: + - cluster-data-exporter + ports: + - "9090:9090" + volumes: + - ./prometheus.yml:/etc/prometheus/prometheus.yml:ro diff --git a/tools/cluster-data-exporter/docker-compose.yml.j2 b/tools/cluster-data-exporter/docker-compose.yml.j2 new file mode 100644 index 00000000..2587b52a --- /dev/null +++ b/tools/cluster-data-exporter/docker-compose.yml.j2 @@ -0,0 +1,41 @@ +# cluster_data_exporter Docker Compose Template +# This template is rendered with Jinja2 to generate the final docker-compose.yml + +{% if x_bake %} +x-bake: + - COMPOSE_BAKE=true + +{% endif %} +services: + cluster-data-exporter: + image: sketchdb-cluster-data-exporter:latest + container_name: {{ container_name | default('sketchdb-cluster-data-exporter') }} + volumes: + - {{ data_directory }}:/data:ro + ports: + - "{{ port }}:{{ port }}" + command: [ + "--input-directory","/data", + "--port","{{ port }}", + "{{ provider }}",{% if provider == "google" %} + "--metrics={{ metrics }}",{% if process_mode == "all-parts" %} + "--all-parts"{% else %} + "--part-index={{ part_index }}"{% endif %}{% elif provider == "alibaba" %} + "--data-type={{ data_type }}", + "--data-year={{ data_year }}",{% if process_mode == "all-parts" %} + "--all-parts"{% else %} + "--part-index={{ part_index }}"{% endif %}{% endif %} + ] + restart: unless-stopped +{% if memory_limit or memory_reservation %} + deploy: + resources: +{% if memory_limit %} + limits: + memory: {{ memory_limit }} +{% endif %} +{% if memory_reservation %} + reservations: + memory: {{ memory_reservation }} +{% endif %} +{% endif %} diff --git a/tools/cluster-data-exporter/docker_compose_frames/alibaba-msresource-2021-docker-compose.yml b/tools/cluster-data-exporter/docker_compose_frames/alibaba-msresource-2021-docker-compose.yml new file mode 100644 index 00000000..683bbfaa --- /dev/null +++ b/tools/cluster-data-exporter/docker_compose_frames/alibaba-msresource-2021-docker-compose.yml @@ -0,0 +1,31 @@ +x-bake: + - COMPOSE_BAKE=true + +services: + cluster-data-exporter: + build: . + container_name: cluster-data-exporter + volumes: + # CHANGE THIS: Replace './data' with the path to your Alibaba MsResource 2021 preprocessed csv.gz files + - ./data:/data:ro + ports: + # Map container port to host port - adjust as needed + - "40000:40000" + command: [ + "--input-directory", + "/data", + "--port", + "40000", + "alibaba", + "--data-type=ms-resource", + "--data-year=2021", + "--all-parts", # or "--part-index=" + ] + restart: unless-stopped + # Optional: set resource limits + deploy: + resources: + limits: + memory: 2G + reservations: + memory: 512M diff --git a/tools/cluster-data-exporter/docker_compose_frames/alibaba-msresource-2022-docker-compose.yml b/tools/cluster-data-exporter/docker_compose_frames/alibaba-msresource-2022-docker-compose.yml new file mode 100644 index 00000000..4bdf90ee --- /dev/null +++ b/tools/cluster-data-exporter/docker_compose_frames/alibaba-msresource-2022-docker-compose.yml @@ -0,0 +1,32 @@ +x-bake: + - COMPOSE_BAKE=true + +services: + cluster-data-exporter: + build: . + container_name: cluster-data-exporter + volumes: + # CHANGE THIS: Replace './data' with the path to your Alibaba MSResource 2022 preprocessed csv.gz files + - ./data:/data:ro + ports: + # Map container port to host port - adjust as needed + - "40000:40000" + command: [ + "--input-directory", + "/data", + "--port", + "40000", + "alibaba", + "--data-type=ms-resource", + "--data-year=2022", + "--all-parts", # or "--part-index=" + + ] + restart: unless-stopped + # Optional: set resource limits + deploy: + resources: + limits: + memory: 2G + reservations: + memory: 512M diff --git a/tools/cluster-data-exporter/docker_compose_frames/alibaba-node-2021-docker-compose.yml b/tools/cluster-data-exporter/docker_compose_frames/alibaba-node-2021-docker-compose.yml new file mode 100644 index 00000000..21349437 --- /dev/null +++ b/tools/cluster-data-exporter/docker_compose_frames/alibaba-node-2021-docker-compose.yml @@ -0,0 +1,32 @@ +x-bake: + - COMPOSE_BAKE=true + +services: + cluster-data-exporter: + build: . + container_name: cluster-data-exporter + volumes: + # CHANGE THIS: Replace './data' with the path to your Alibaba Node 2021 preprocessed csv.gz files + - ./data:/data:ro + ports: + # Map container port to host port - adjust as needed + - "40000:40000" + command: [ + "--input-directory", + "/data", + "--port", + "40000", + "alibaba", + "--data-type=node", + "--data-year=2021", + "--all-parts", # or "--part-index=" + + ] + restart: unless-stopped + # Optional: set resource limits + deploy: + resources: + limits: + memory: 2G + reservations: + memory: 512M diff --git a/tools/cluster-data-exporter/docker_compose_frames/alibaba-node-2022-docker-compose.yml b/tools/cluster-data-exporter/docker_compose_frames/alibaba-node-2022-docker-compose.yml new file mode 100644 index 00000000..bbe7b7b1 --- /dev/null +++ b/tools/cluster-data-exporter/docker_compose_frames/alibaba-node-2022-docker-compose.yml @@ -0,0 +1,32 @@ +x-bake: + - COMPOSE_BAKE=true + +services: + cluster-data-exporter: + build: . + container_name: cluster-data-exporter + volumes: + # CHANGE THIS: Replace './data' with the path to your Alibaba Node 2022 preprocessed csv.gz files + - ./data:/data:ro + ports: + # Map container port to host port - adjust as needed + - "40000:40000" + command: [ + "--input-directory", + "/data", + "--port", + "40000", + "alibaba", + "--data-type=node", + "--data-year=2022", + "--all-parts", # or "--part-index=" + + ] + restart: unless-stopped + # Optional: set resource limits + deploy: + resources: + limits: + memory: 2G + reservations: + memory: 512M diff --git a/tools/cluster-data-exporter/docker_compose_frames/base-docker-compose.yml b/tools/cluster-data-exporter/docker_compose_frames/base-docker-compose.yml new file mode 100644 index 00000000..e013892e --- /dev/null +++ b/tools/cluster-data-exporter/docker_compose_frames/base-docker-compose.yml @@ -0,0 +1,39 @@ +x-bake: + - COMPOSE_BAKE=true + +services: + cluster-data-exporter: + build: . + container_name: cluster-data-exporter + volumes: + # Mount the local data directory to /data in the container + # Replace './data' with the path to your csv.gz files + - ./data:/data:ro + ports: + # Map container port to host port - adjust as needed + - "40000:40000" + command: [ + "--input-directory", + "/data", + "--port", + "40000", + # Add your provider-specific arguments here + # For Google data example: + # "google", + # "--metrics", "mean_cpu_usage_rate,canonical_memory_usage", + # "--all-parts" + # + # For Alibaba data example: + # "alibaba", + # "--data-type", "node", + # "--data-year", "2021", + # "--all-parts" + ] + restart: unless-stopped + # Optional: set resource limits + deploy: + resources: + limits: + memory: 2G + reservations: + memory: 512M diff --git a/tools/cluster-data-exporter/docker_compose_frames/google-docker-compose.yml b/tools/cluster-data-exporter/docker_compose_frames/google-docker-compose.yml new file mode 100644 index 00000000..22ce3847 --- /dev/null +++ b/tools/cluster-data-exporter/docker_compose_frames/google-docker-compose.yml @@ -0,0 +1,32 @@ +x-bake: + - COMPOSE_BAKE=true + +services: + cluster-data-exporter: + build: . + container_name: cluster-data-exporter + volumes: + # CHANGE THIS: Replace './data' with the path to your Google csv.gz files + - ./data:/data:ro + ports: + # Map container port to host port - adjust as needed + - "40000:40000" + command: [ + "--input-directory", + "/data", + "--port", + "40000", + "google", + # CHANGE THIS: Replace with your desired metrics (comma-separated) + "--metrics=mean_cpu_usage_rate,canonical_memory_usage", + "--all-parts", # or "--part-index=" + + ] + restart: unless-stopped + # Optional: set resource limits + deploy: + resources: + limits: + memory: 2G + reservations: + memory: 512M diff --git a/tools/cluster-data-exporter/installation/install.sh b/tools/cluster-data-exporter/installation/install.sh new file mode 100755 index 00000000..873019bc --- /dev/null +++ b/tools/cluster-data-exporter/installation/install.sh @@ -0,0 +1,12 @@ +#!/bin/bash + +set -e + +THIS_DIR=$(dirname "$(readlink -f "${BASH_SOURCE[0]}")") +PARENT_DIR=$(dirname "$THIS_DIR") + +echo "Building Cluster Data Exporter Docker image..." +cd "$PARENT_DIR" +docker build . -f Dockerfile -t sketchdb-cluster-data-exporter:latest + +echo "Cluster Data Exporter Docker image built successfully: sketchdb-cluster-data-exporter:latest" diff --git a/tools/cluster-data-exporter/prometheus.yml b/tools/cluster-data-exporter/prometheus.yml new file mode 100644 index 00000000..263498f4 --- /dev/null +++ b/tools/cluster-data-exporter/prometheus.yml @@ -0,0 +1,7 @@ +global: + scrape_interval: 1s + +scrape_configs: + - job_name: cluster-data-exporter + static_configs: + - targets: ["cluster-data-exporter:40000"] diff --git a/tools/cluster-data-exporter/scripts/generate_docker_compose.py b/tools/cluster-data-exporter/scripts/generate_docker_compose.py new file mode 100644 index 00000000..38c32f9a --- /dev/null +++ b/tools/cluster-data-exporter/scripts/generate_docker_compose.py @@ -0,0 +1,253 @@ +""" +Script to generate docker-compose.yml files from frame templates based on data provider configuration. + +This script takes a data provider (google or alibaba) and provider-specific arguments, +then generates a docker-compose.yml file by copying and modifying the appropriate frame file. +""" + +import argparse +import sys +from pathlib import Path +from typing import Any, Dict, List, Optional + +import yaml + +# Valid values from Rust enums (CLI format with hyphens) +VALID_GOOGLE_METRICS = [ + "mean-cpu-usage-rate", + "canonical-memory-usage", + "assigned-memory-usage", + "unmapped-page-cache-memory-usage", + "total-page-cache-memory-usage", + "max-memory-usage", + "mean-disk-io-time", + "mean-local-disk-space-used", + "max-cpu-usage", + "max-disk-io-time", + "cycles-per-instruction", + "memory-accesses-per-instruction", + "sample-portion", + "sampled-cpu-usage", +] + +VALID_ALIBABA_DATA_TYPES = ["node", "msresource"] +VALID_ALIBABA_DATA_YEARS = [2021, 2022] + + +def validate_google_metrics(metrics: List[str]) -> None: + """Validate that all provided Google metrics are valid.""" + invalid_metrics = [m for m in metrics if m not in VALID_GOOGLE_METRICS] + if invalid_metrics: + print(f"Error: Invalid Google metrics: {', '.join(invalid_metrics)}") + print(f"Valid metrics: {', '.join(VALID_GOOGLE_METRICS)}") + sys.exit(1) + + +def validate_alibaba_args(data_type: str, data_year: int) -> None: + """Validate Alibaba data type and year arguments.""" + if data_type not in VALID_ALIBABA_DATA_TYPES: + print(f"Error: Invalid data type: {data_type}") + print(f"Valid data types: {', '.join(VALID_ALIBABA_DATA_TYPES)}") + sys.exit(1) + + if data_year not in VALID_ALIBABA_DATA_YEARS: + print(f"Error: Invalid data year: {data_year}") + print(f"Valid years: {', '.join(map(str, VALID_ALIBABA_DATA_YEARS))}") + sys.exit(1) + + +def get_frame_file_path( + provider: str, data_type: Optional[str] = None, data_year: Optional[int] = None +) -> Path: + """Get the path to the appropriate frame file based on provider and arguments.""" + frames_dir = Path("docker_compose_frames") + + if provider == "google": + return frames_dir / "google-docker-compose.yml" + elif provider == "alibaba": + return frames_dir / f"alibaba-{data_type}-{data_year}-docker-compose.yml" + else: + raise ValueError(f"Unknown provider: {provider}") + + +def load_yaml_file(file_path: Path) -> Dict[str, Any]: + """Load YAML file and return parsed content.""" + with open(file_path, "r") as f: + return yaml.safe_load(f) + + +def save_yaml_file(file_path: Path, data: Dict[str, Any]) -> None: + """Save data to YAML file.""" + with open(file_path, "w") as f: + yaml.dump(data, f, default_flow_style=False, sort_keys=False) + + +def update_command_arg(command: List[str], arg_name: str, new_value: str) -> List[str]: + """Update a command line argument in the command list.""" + updated_command = [] + i = 0 + while i < len(command): + if command[i] == arg_name: + updated_command.append(command[i]) + if i + 1 < len(command): + updated_command.append(new_value) + i += 2 + else: + updated_command.append(new_value) + i += 1 + elif command[i].startswith(f"{arg_name}="): + updated_command.append(f"{arg_name}={new_value}") + i += 1 + else: + updated_command.append(command[i]) + i += 1 + return updated_command + + +def generate_google_compose( + metrics: List[str], port: Optional[int], input_dir: Optional[str] +) -> None: + """Generate docker-compose.yml for Google provider.""" + frame_file = get_frame_file_path("google") + output_file = Path("docker-compose.yml") + + # Load frame file + compose_data = load_yaml_file(frame_file) + + # Update metrics + metrics_str = ",".join(metrics) + service = compose_data["services"]["cluster-data-exporter"] + command = service["command"] + + # Find and update metrics argument + for i, arg in enumerate(command): + if arg.startswith("--metrics="): + command[i] = f"--metrics={metrics_str}" + break + + # Update optional arguments if provided + if port is not None: + # Update port mapping + service["ports"] = [f"{port}:{port}"] + # Update port in command + command = update_command_arg(command, "--port", str(port)) + service["command"] = command + + if input_dir is not None: + # Update volume mapping + service["volumes"] = [f"{input_dir}:/data:ro"] + + # Save updated compose file + save_yaml_file(output_file, compose_data) + + +def generate_alibaba_compose( + data_type: str, + data_year: int, + port: Optional[int], + input_dir: Optional[str], + speedup: Optional[int], +) -> None: + """Generate docker-compose.yml for Alibaba provider.""" + frame_file = get_frame_file_path("alibaba", data_type, data_year) + output_file = Path("docker-compose.yml") + + # Load frame file + compose_data = load_yaml_file(frame_file) + + service = compose_data["services"]["cluster-data-exporter"] + command = service["command"] + + # Update optional arguments if provided + if port is not None: + # Update port mapping + service["ports"] = [f"{port}:{port}"] + # Update port in command + command = update_command_arg(command, "--port", str(port)) + service["command"] = command + + if input_dir is not None: + # Update volume mapping + service["volumes"] = [f"{input_dir}:/data:ro"] + + # Add speedup if specified + if speedup is not None: + if "--speedup" not in " ".join(command): + command.append(f"--speedup={speedup}") + else: + command = update_command_arg(command, "--speedup", str(speedup)) + service["command"] = command + + # Save updated compose file + save_yaml_file(output_file, compose_data) + + +def main(): + parser = argparse.ArgumentParser( + description="Generate docker-compose.yml from frame files based on data provider configuration", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=""" +Examples: + # Google provider with specific metrics + python scripts/generate_docker_compose.py google --metrics mean_cpu_usage_rate,max_cpu_usage --port 8080 + + # Alibaba provider with node data from 2021 + python scripts/generate_docker_compose.py alibaba --data-type node --data-year 2021 --port 8080 + + # With custom input directory + python scripts/generate_docker_compose.py google --metrics canonical_memory_usage --input-dir /path/to/data + """, + ) + + parser.add_argument("provider", choices=["google", "alibaba"], help="Data provider") + parser.add_argument("--port", type=int, help="Port number for the HTTP server") + parser.add_argument("--input-dir", "--input-directory", help="Input directory path") + + # Google-specific arguments + google_group = parser.add_argument_group("Google provider arguments") + google_group.add_argument( + "--metrics", type=str, help="Comma-separated list of metrics to export" + ) + + # Alibaba-specific arguments + alibaba_group = parser.add_argument_group("Alibaba provider arguments") + alibaba_group.add_argument( + "--data-type", choices=VALID_ALIBABA_DATA_TYPES, help="Type of data to export" + ) + alibaba_group.add_argument( + "--data-year", + type=int, + choices=VALID_ALIBABA_DATA_YEARS, + help="Year of the dataset", + ) + alibaba_group.add_argument( + "--speedup", + type=int, + help="Speedup factor for faster-than-realtime export (1=real-time, 10=10x faster)", + ) + + args = parser.parse_args() + + # Validate provider-specific required arguments + if args.provider == "google": + if not args.metrics: + parser.error("Google provider requires --metrics argument") + metrics_list = [m.strip() for m in args.metrics.split(",")] + validate_google_metrics(metrics_list) + generate_google_compose(metrics_list, args.port, args.input_dir) + + elif args.provider == "alibaba": + if not args.data_type: + parser.error("Alibaba provider requires --data-type argument") + if not args.data_year: + parser.error("Alibaba provider requires --data-year argument") + validate_alibaba_args(args.data_type, args.data_year) + generate_alibaba_compose( + args.data_type, args.data_year, args.port, args.input_dir, args.speedup + ) + + print(f"Generated docker-compose.yml for {args.provider} provider") + + +if __name__ == "__main__": + main() diff --git a/tools/cluster-data-exporter/scripts/requirements.txt b/tools/cluster-data-exporter/scripts/requirements.txt new file mode 100644 index 00000000..5fde2580 --- /dev/null +++ b/tools/cluster-data-exporter/scripts/requirements.txt @@ -0,0 +1,2 @@ +PyYAML==6.0.2 +types-PyYAML==6.0.12.20250516 diff --git a/tools/cluster-data-exporter/src/alibaba_metrics.rs b/tools/cluster-data-exporter/src/alibaba_metrics.rs new file mode 100644 index 00000000..98bc4eca --- /dev/null +++ b/tools/cluster-data-exporter/src/alibaba_metrics.rs @@ -0,0 +1,71 @@ +use clap::ValueEnum; +use std::sync::OnceLock; + +type BoxedErr = Box; + +// Speedup factor for faster-than-realtime export (set via CLI) +pub static SPEEDUP_FACTOR: OnceLock = OnceLock::new(); + +#[derive(Copy, Clone, Debug, ValueEnum)] +pub enum MsDataType { + // BM Node runtime information. + // It records CPU and memory utilization of 1300+ BM nodes in a production cluster. + Node, + // MS runtime information. + // It records CPU and memory utilization of 90000+ containers for 1300+ MSs in the same production cluster. + MsResource, +} + +pub mod ms_resource; +pub mod node; + +// The type of microservice data to export. Should be initialized before any +// reading or exporting begins +pub static EXPORTER_DATA_TYPE: OnceLock = OnceLock::new(); + +/// @brief Calls the export_from_queue() function based on runtime initialized +/// EXPORTER_DATA_TYPE +pub fn export_from_queue() { + match EXPORTER_DATA_TYPE.get().unwrap() { + MsDataType::Node => node::export_from_queue(), + MsDataType::MsResource => ms_resource::export_from_queue(), + } +} + +/// @brief Main routine for the thread that will be reading csv data and +/// exporting. This function just uses a match statement to call the reading +/// and exporting routine required by the specified mode +/// +/// @param[in] input_dir The input directory containing csv files +/// @param[in] all_parts Whether to start from part 0 of csv files and continue +/// until no more files are found. This should be false if +/// part_index is Some(part) +/// @param[in] part_index Which csv file part to use as the data source. +/// This should be None if all_parts is true. +/// @param[in] data_type The type of data out of the different types of trace +/// data in the Alibaba micro-services trace data +/// @param[in] data_year The year of the trace data. Supported values are +/// 2021 and 2022 +/// @param[in] speedup Speedup factor for faster-than-realtime export +/// +/// @return The result returned by the reader thread. +pub fn reader_thread_routine( + input_dir: String, + all_parts: bool, + part_index: Option, + data_type: MsDataType, + data_year: u32, + speedup: u64, +) -> Result<(), BoxedErr> { + use crate::alibaba_metrics::node; + let _ = EXPORTER_DATA_TYPE.set(data_type); + let _ = SPEEDUP_FACTOR.set(speedup); + let result = match EXPORTER_DATA_TYPE.get().unwrap() { + MsDataType::Node => node::read_and_queue(&input_dir, all_parts, part_index, data_year), + MsDataType::MsResource => { + ms_resource::read_and_queue(&input_dir, all_parts, part_index, data_year) + } + }; + + result +} diff --git a/tools/cluster-data-exporter/src/alibaba_metrics/ms_resource.rs b/tools/cluster-data-exporter/src/alibaba_metrics/ms_resource.rs new file mode 100644 index 00000000..b1c81c1a --- /dev/null +++ b/tools/cluster-data-exporter/src/alibaba_metrics/ms_resource.rs @@ -0,0 +1,251 @@ +use crate::utilities; +pub use concurrent_queue::ConcurrentQueue; +use csv::Reader; +use flate2::read::GzDecoder; +use lazy_static::lazy_static; +use prometheus::{register_gauge_vec, GaugeVec}; +use std::fs::File; +use std::io::BufReader; +use std::thread; +use std::time::Duration; +use tracing::info; + +const FILENAME_PARTS_2021: [&str; 2] = ["MSResource_", ".csv.gz"]; +const FILENAME_PARTS_2022: [&str; 2] = ["MSMetrics_", ".csv.gz"]; + +const DATA_QUEUE_CAP: usize = 400_000; +const QUEUE_POLL_INTERVAL_MS: u64 = 250; +const CSV_DELIMITER: u8 = b','; +const LABELS: [&str; 3] = ["ms_name", "ms_instance_id", "node_id"]; + +type CsvGzReader = Reader>>; +type BoxedErr = Box; +/// Struct for holding fields after deserialization +/// for both 2021 and 2022 +#[derive(Debug, serde::Deserialize)] +pub struct MsResourceCsvFields { + #[serde(rename = "", skip)] + _trace: u64, + + #[serde(rename = "timestamp")] + timestamp: u64, + + #[serde(rename = "nodeid")] + node_id: String, + + #[serde(rename = "msname")] + ms_name: String, + + #[serde(rename = "msinstanceid")] + ms_instance_id: String, + + #[serde(alias = "instance_cpu_usage", alias = "cpu_utilization")] + cpu_usage: Option, + + #[serde(alias = "instance_memory_usage", alias = "memory_utilization")] + memory_usage: Option, +} + +lazy_static! { + pub static ref MS_RESOURCE_DATA_QUEUE: ConcurrentQueue = + ConcurrentQueue::bounded(DATA_QUEUE_CAP); + pub static ref CPU_USAGE: GaugeVec = register_gauge_vec!( + "alibaba_microservice_cpu_usage", + "Cpu usages for microservices by alibaba nodes", + &LABELS, + ) + .unwrap(); + pub static ref MEMORY_USAGE: GaugeVec = register_gauge_vec!( + "alibaba_microservice_memory_usage", + "Memory usages for microservices by alibaba nodes", + &LABELS, + ) + .unwrap(); +} + +/// @brief Gets the filename for the MsResource csv data based on the year +/// and the index number +/// +/// @param[in] year The year of the trace data. Supported values are 2021 +/// and 2022 +/// @param[in] index_no The index of the csv file +/// +/// @return A String of the filename based on the data year and index num +fn get_filename(year: u32, index_no: u16) -> String { + let mut filename: String = String::new(); + let prefix: &str; + let suffix: &str; + let index: &str = &format!("{}", index_no); + + match year { + 2021 => { + prefix = FILENAME_PARTS_2021[0]; + suffix = FILENAME_PARTS_2021[1]; + } + 2022 => { + prefix = FILENAME_PARTS_2022[0]; + suffix = FILENAME_PARTS_2022[1]; + } + _ => { + panic!("Invalid year: {}", year); + } + } + filename.push_str(prefix); + filename.push_str(index); + filename.push_str(suffix); + + filename +} + +/// @brief Gets a csv reader for MsResource data +/// +/// @param[in] input_dir The directory containing the csv file +/// @param[in] year Which trace data year to create the reader for. +/// supported years are 2021 and 2022 +/// @param[in] index The index of the csv file +/// +/// @return A Result type containing either the reader or an Error if the file +/// cannot be found +pub fn get_reader(input_dir: &str, year: u32, index: u16) -> Result, BoxedErr> { + use csv::ReaderBuilder; + use std::path::Path; + + let filename: String = get_filename(year, index); + let file_path = Path::new(input_dir).join(&filename); + let fd: File = File::open(file_path)?; + let buf_rdr: BufReader = BufReader::new(fd); + let gz_decoder: GzDecoder> = GzDecoder::new(buf_rdr); + + let csv_rdr: CsvGzReader = ReaderBuilder::new() + .delimiter(CSV_DELIMITER) + .flexible(true) + .has_headers(true) + .from_reader(gz_decoder); + + Ok(csv_rdr) +} + +/// @brief Routine for reading MSResource csv data and enqueuing it +/// +/// @param[in] input_dir The input directory containing the csv file +/// @param[in] all_parts Whether or not to read all csv files in the +/// directory, starting from part 0. Once a file +/// cannot be found, this will return. This should +/// be false if a part_index is given. +/// @param[in] part_index The part index for a single csv file to use as +/// the data source. This should be None if all_parts +/// is true. +/// @param[in] year The year of the trace data. Supported values are +/// 2021 and 2022 +/// +/// @pre All csv files are uncompressed +/// @pre If all_parts is specified, at least part 0 must exist +/// @pre Either all_parts is true and part_index is None, or all_parts is +/// false and part_index is Some(part) +pub fn read_and_queue( + input_dir: &str, + all_parts: bool, + part_index: Option, + year: u32, +) -> Result<(), BoxedErr> { + let mut part: u16 = 0; + if !all_parts { + part = part_index.unwrap(); + } + + while let Ok(mut rdr) = get_reader(input_dir, year, part) { + let csv_iter = rdr.deserialize(); + for csv_line in csv_iter { + while MS_RESOURCE_DATA_QUEUE.is_full() { + thread::sleep(Duration::from_millis(QUEUE_POLL_INTERVAL_MS)); + } + let parsed_line: MsResourceCsvFields = csv_line?; + let _ = MS_RESOURCE_DATA_QUEUE.push(parsed_line); + } + part += 1; + if !all_parts { + break; + } + } // No more files to read, or couldn't find initial file + + if part == 0 { + // Reading always starts at part 0 + panic!( + "Failed to read initial .csv.gz file. Check that all data files + are named in the correct format (2021: '{}{}', 2022: '{}{}), + and that the csv files contian the field headers at the top + ", + FILENAME_PARTS_2021[0], + FILENAME_PARTS_2021[1], + FILENAME_PARTS_2022[0], + FILENAME_PARTS_2022[1] + ); + } else { + MS_RESOURCE_DATA_QUEUE.close(); + Ok(()) + } +} + +/// @brief Takes the timestamp of a trace in milliseconds and +/// returns the normalized time as a Duration +/// +/// @param[in] time_millis The trace timestamp in milliseconds +/// +/// @return The normalized timestamp as a Duration +/// +/// @NOTE: Brief check of data suggests no dilation is necessary +/// +/// @NOTE: MSResource data from 2022 is not sorted by timestamp whatsoever, +/// sometimes the data is listed in order of decreasing timestamp and other +/// times it's listed in order of increasing timestamp, so the timestamps +/// are modified to work with the exporter before being queued +/// +/// @NOTE: SPEEDUP_FACTOR can be set via --speedup CLI argument for faster-than-realtime export +pub fn get_normalized_start_time(time_millis: u64) -> Duration { + let speedup = crate::alibaba_metrics::SPEEDUP_FACTOR.get().unwrap_or(&1); + Duration::from_millis(time_millis / speedup) +} + +/// @brief Exports a single line from the MS_RESOURCE_DATA_QUEUE +/// +/// @param[in] csv_line A parsed line from a MsResource csv file +pub fn export_line(csv_line: MsResourceCsvFields) { + let label_vals: [&str; 3] = [ + csv_line.ms_name.as_str(), + csv_line.ms_instance_id.as_str(), + csv_line.node_id.as_str(), + ]; + + if let Some(cpu_usage) = csv_line.cpu_usage { + CPU_USAGE.with_label_values(&label_vals).set(cpu_usage); + } + + if let Some(memory_usage) = csv_line.memory_usage { + MEMORY_USAGE + .with_label_values(&label_vals) + .set(memory_usage); + } +} + +/// @brief Exports lines from the queue until a line is found with a timestamp +/// later than the current runtime. This function will terminate the +/// the program once the queue has both been closed by the reader thread +/// and the queue is empty +pub fn export_from_queue() { + let elapsed_t: Duration = utilities::get_time_elapsed(); + let check_time = + |line: &MsResourceCsvFields| get_normalized_start_time(line.timestamp) <= elapsed_t; + MS_RESOURCE_DATA_QUEUE + .try_iter() + .take_while(check_time) + .for_each(export_line); + + // No more files to read and empty queue + if MS_RESOURCE_DATA_QUEUE.is_closed() + && MS_RESOURCE_DATA_QUEUE.is_empty() + && utilities::should_exit_after_eof() + { + info!("No more MSResource data to export, shutting down"); + std::process::exit(0); + } +} diff --git a/tools/cluster-data-exporter/src/alibaba_metrics/node.rs b/tools/cluster-data-exporter/src/alibaba_metrics/node.rs new file mode 100644 index 00000000..78ff25f5 --- /dev/null +++ b/tools/cluster-data-exporter/src/alibaba_metrics/node.rs @@ -0,0 +1,235 @@ +use crate::utilities; +use concurrent_queue::ConcurrentQueue; +use csv::{Reader, ReaderBuilder}; +use flate2::read::GzDecoder; +use lazy_static::lazy_static; +use prometheus::{register_gauge_vec, GaugeVec}; +use std::fs::File; +use std::io::BufReader; +use std::path::Path; +use std::thread; +use std::time::Duration; +use tracing::info; + +type BoxedErr = Box; +type CsvGzReader = Reader>>; + +const FILENAME_PARTS_2021: [&str; 2] = ["Node_", ".csv.gz"]; +const FILENAME_PARTS_2022: [&str; 2] = ["NodeMetrics_", ".csv.gz"]; + +const DATA_QUEUE_CAP: usize = 400_000; +const QUEUE_POLL_INTERVAL_MS: u64 = 250; +const CSV_DELIMITER: u8 = b','; + +const LABELS: [&str; 1] = ["node_id"]; + +/// Struct for holding fields after deserialization +#[derive(Debug, serde::Deserialize)] +pub struct NodeCsvFields { + #[serde(rename = "", skip)] + _trace: u64, + + #[serde(rename = "timestamp")] + timestamp: u64, + + #[serde(rename = "nodeid")] + node_id: String, + + #[serde(alias = "node_cpu_usage", alias = "cpu_utilization")] + cpu_usage: Option, + + #[serde(alias = "node_memory_usage", alias = "memory_utilization")] + memory_usage: Option, +} + +lazy_static! { + pub static ref NODE_DATA_QUEUE: ConcurrentQueue = + ConcurrentQueue::bounded(DATA_QUEUE_CAP); + pub static ref CPU_USAGE: GaugeVec = register_gauge_vec!( + "alibaba_node_cpu_usage", + "Cpu usages by alibaba nodes", + &LABELS, + ) + .unwrap(); + pub static ref MEMORY_USAGE: GaugeVec = register_gauge_vec!( + "alibaba_node_memory_usage", + "Memory usages by alibaba nodes", + &LABELS, + ) + .unwrap(); +} + +/// @brief Gets the filename for the Node_.csv.gz data based on the year +/// and the index number +/// +/// @param[in] year The year of the trace data. Supported values are 2021 +/// and 2022 +/// @param[in] index_no The index of the csv file +/// +/// @return A String of the filename based on the data year and index num +fn get_filename(year: u32, index_no: u16) -> String { + let mut filename: String = String::new(); + let prefix: &str; + let suffix: &str; + let index: &str = &format!("{}", index_no); + + match year { + 2021 => { + prefix = FILENAME_PARTS_2021[0]; + suffix = FILENAME_PARTS_2021[1]; + } + 2022 => { + prefix = FILENAME_PARTS_2022[0]; + suffix = FILENAME_PARTS_2022[1]; + } + _ => { + panic!("Invalid year: {}", year); + } + } + filename.push_str(prefix); + filename.push_str(index); + filename.push_str(suffix); + + filename +} + +/// @brief Gets a csv reader for Node data +/// +/// @param[in] input_dir The directory containing the csv file +/// @param[in] year Which trace data year to create the reader for. +/// supported years are 2021 and 2022 +/// +/// @return A reader for the .csv.gz files +/// +/// @pre All files should have been converted to a .csv.gz format from the +/// .tar.gz format that they come as initially. +pub fn get_reader( + input_dir: &str, + year: u32, + index_no: u16, +) -> Result, BoxedErr> { + let filename = get_filename(year, index_no); + let file_path = Path::new(input_dir).join(&filename); + let fd: File = File::open(file_path)?; + let buf_rdr: BufReader = BufReader::new(fd); + let gz_decoder: GzDecoder> = GzDecoder::new(buf_rdr); + + let csv_rdr: CsvGzReader = ReaderBuilder::new() + .delimiter(CSV_DELIMITER) + .flexible(true) + .has_headers(true) + .from_reader(gz_decoder); + + Ok(csv_rdr) +} + +/// @brief Takes the timestamp of a trace in milliseconds and +/// returns the normalized time as a Duration +/// +/// @param[in] time_millis The trace timestamp in milliseconds +/// +/// @return The normalized timestamp as a Duration +/// +/// @NOTE: Brief check of data suggests no dilation is necessary +/// +/// @NOTE: Node data from 2022 is not sorted by timestamp whatsoever, +/// sometimes the data is listed in order of decreasing timestamp and other +/// times it's listed in order of increasing timestamp, so the timestamps +/// are modified to work with the exporter before being queued +/// +/// @NOTE: SPEEDUP_FACTOR can be set via --speedup CLI argument for faster-than-realtime export +pub fn get_normalized_start_time(time_millis: u64) -> Duration { + let speedup = crate::alibaba_metrics::SPEEDUP_FACTOR.get().unwrap_or(&1); + Duration::from_millis(time_millis / speedup) +} + +/// @brief Reads the csv data from .csv.gz files and adds them to the queue. +/// +/// @param[in] input_dir The input directory +/// @param[in] data_year The year of the trace data +/// +/// @pre All csv data should have been sorted by timestamp and compressed with +/// gzip +pub fn read_and_queue( + input_dir: &str, + all_parts: bool, + part_index: Option, + data_year: u32, +) -> Result<(), BoxedErr> { + let mut part: u16 = 0; + if !all_parts { + part = part_index.unwrap(); + } + + while let Ok(mut rdr) = get_reader(input_dir, data_year, part) { + let csv_iter = rdr.deserialize(); + for csv_line in csv_iter { + while NODE_DATA_QUEUE.is_full() { + thread::sleep(Duration::from_millis(QUEUE_POLL_INTERVAL_MS)); + } + let parsed_line: NodeCsvFields = csv_line?; + let _ = NODE_DATA_QUEUE.push(parsed_line); + } // EOF + part += 1; + + if !all_parts { + break; + } + } // No more files to read, or couldn't find initial file + + if part == 0 { + // Reading always starts at part 0 + panic!( + "Failed to read initial .csv.gz file. Check that all data files + are named in the correct format (2021: '{}{}', 2022: '{}{}), + and that the csv files contain the field headers at the top. + ", + FILENAME_PARTS_2021[0], + FILENAME_PARTS_2021[1], + FILENAME_PARTS_2022[0], + FILENAME_PARTS_2022[1] + ); + } else { + NODE_DATA_QUEUE.close(); + Ok(()) + } +} + +/// @brief Exports a single line from the NODE_DATA_QUEUE +/// +/// @param[in] csv_line A parsed line from a Node csv file +pub fn export_line(csv_line: NodeCsvFields) { + let label_vals: [&str; 1] = [csv_line.node_id.as_str()]; + + if let Some(cpu_usage) = csv_line.cpu_usage { + CPU_USAGE.with_label_values(&label_vals).set(cpu_usage); + } + + if let Some(memory_usage) = csv_line.memory_usage { + MEMORY_USAGE + .with_label_values(&label_vals) + .set(memory_usage); + } +} + +/// @brief Exports lines from the queue until a line is found with a timestamp +/// later than the current runtime. This function will terminate the +/// the program once the queue has both been closed by the reader thread +/// and the queue is empty +pub fn export_from_queue() { + let elapsed_t: Duration = utilities::get_time_elapsed(); + let check_time = |line: &NodeCsvFields| get_normalized_start_time(line.timestamp) <= elapsed_t; + NODE_DATA_QUEUE + .try_iter() + .take_while(check_time) + .for_each(export_line); + + // No more files to read and empty queue + if NODE_DATA_QUEUE.is_closed() + && NODE_DATA_QUEUE.is_empty() + && utilities::should_exit_after_eof() + { + info!("No more Node data to export, shutting down"); + std::process::exit(0); + } +} diff --git a/tools/cluster-data-exporter/src/google_metrics.rs b/tools/cluster-data-exporter/src/google_metrics.rs new file mode 100644 index 00000000..afc29c76 --- /dev/null +++ b/tools/cluster-data-exporter/src/google_metrics.rs @@ -0,0 +1,453 @@ +use crate::utilities; +use crate::utilities::*; +use clap::ValueEnum; +use concurrent_queue::ConcurrentQueue; +use csv::Reader; +use flate2::read::GzDecoder; +use lazy_static::lazy_static; +use prometheus::{register_gauge_vec, GaugeVec}; +use std::sync::OnceLock; +use std::thread; +use std::time::Duration; +use std::{fs::File, io::BufReader}; +use tracing::info; + +type CsvGzReader = Reader>>; + +/* Standard labels for google's task resource usage data */ +const TRU_LABELS: [&str; 3] = ["job_id", "task_index", "machine_id"]; +const TRU_CSV_DELIMITER: u8 = b','; +const DATA_QUEUE_CAP: usize = 400_000; // Max lines in the queue +const CSV_MAX_PART_NO: u16 = 500; + +const MICRO_SECONDS_PER_SECOND: u64 = 1_000_000; +const T_OFFSET_SECS: u64 = 600; +const DILATION_FACTOR: u64 = 10; // Factor for scaling time stamps relative to when they are exported + +/// Each line of the csv file is serialized into the following struct. +/// The ordering of the struct fields MUST match the order that fields +/// appear in a line of the csv file. +/// +/// All fields wrapped in Option<> are not considered mandatory by +/// the schema and, therefore, may be missing from a given trace. +/// The rest of the fields should never be missing, so failure to +/// deserialize will result in an error and program termination +#[derive(Debug, serde::Deserialize)] +pub struct TruCsvFields { + pub start_time: u64, + pub _end_time: u64, // unused, only here for parsing + pub job_id: String, // label + pub task_index: String, // label + pub machine_id: String, // label + pub mean_cpu_usage_rate: Option, + pub canonical_memory_usage: Option, + pub assigned_memory_usage: Option, + pub unmapped_page_cache_memory_usage: Option, + pub total_page_cache_memory_usage: Option, + pub max_memory_usage: Option, + pub mean_disk_io_time: Option, + pub mean_local_disk_space_used: Option, + pub max_cpu_usage: Option, + pub max_disk_io_time: Option, + pub cycles_per_instruction: Option, + pub memory_accesses_per_instruction: Option, + pub sample_portion: Option, + pub aggregation_type: Option, // Divides metrics into two + pub sampled_cpu_usage: Option, +} + +/// @brief An enum for matching the metrics to export with their +/// corresponding prometheus gauges +#[derive(Copy, Clone, Debug, ValueEnum)] +pub enum TruMetrics { + MeanCpuUsageRate, + CanonicalMemoryUsage, + AssignedMemoryUsage, + UnmappedPageCacheMemoryUsage, + TotalPageCacheMemoryUsage, + MaxMemoryUsage, + MeanDiskIoTime, + MeanLocalDiskSpaceUsed, + MaxCpuUsage, + MaxDiskIoTime, + CyclesPerInstruction, + MemoryAccessesPerInstruction, + SamplePortion, + SampledCpuUsage, +} + +/// @brief A tuple struct representing two of the same prometheus metrics, +/// but partitioned by their aggregation type. Index number directly +/// corresponds to the aggregation type, i.e. i=0 => aggregation_type=0 +pub struct GaugePair(GaugeVec, GaugeVec); + +impl GaugePair { + /// @brief Create and register both GaugeVecs in the GaugePair to the + /// default registry. + /// + /// @param[in] base_name The string used as the base of both metrics + /// names as seen by prometheus, where aggregation type will be appended + /// + /// @param[in] base_help The string used as the base of both metrics + /// help strings when scraped by prometheus. Aggregation type is + /// appended + fn new(base_name: &str, base_help: &str) -> GaugePair { + let mut name_0 = String::from(base_name); + name_0.push_str("_0"); + let mut help_0 = String::from(base_help); + help_0.push_str(" (aggregation_type=0)"); + let gauge_0 = register_gauge_vec!(name_0.as_str(), help_0.as_str(), &TRU_LABELS).unwrap(); + + let mut name_1 = String::from(base_name); + name_1.push_str("_1"); + let mut help_1 = String::from(base_help); + help_1.push_str(" (aggregation_type=1)"); + let gauge_1 = register_gauge_vec!(name_1.as_str(), help_1.as_str(), &TRU_LABELS).unwrap(); + + GaugePair(gauge_0, gauge_1) + } + + /// @brief Retrieve a static reference to the gauge from the pair for + /// the given aggregation type + /// + /// @param[in] self Statically defined GaugePair + /// @param[in] aggregation_type 0 or 1 (The aggregation type) + fn get(&'static self, aggregation_type: u8) -> &'static GaugeVec { + match aggregation_type { + 0 => &self.0, + 1 => &self.1, + _ => panic!("Invalid index into gauge vec"), + } + } +} + +/// List of metrics to export from the google task resource usage data +pub static GOOGLE_METRICS: OnceLock> = OnceLock::new(); + +lazy_static! { + /// Queue for parsed csv lines + pub static ref GOOGLE_DATA_QUEUE: ConcurrentQueue = ConcurrentQueue::bounded(DATA_QUEUE_CAP); + + /* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * + * ALL METRICS * + * * + * Each static reference is a GaugePair corresponding to a single * + * metric. Each element of the pair corresponds to an aggregation * + * type of 0 or 1. When the aggregation type is missing from a * + * trace the aggregation type defaults to 0 * + * * + * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * */ + pub static ref MEAN_CPU_USAGE_RATE_PAIR: GaugePair = GaugePair::new( + "google_mean_cpu_usage_rate", "Mean cpu usage rate by google machines", + ); + + pub static ref CANONICAL_MEMORY_USAGE_PAIR: GaugePair = GaugePair::new( + "google_canonical_memory_usage", "Canonical memory usage by google cluster machines", + ); + + pub static ref ASSIGNED_MEMORY_USAGE_PAIR: GaugePair = GaugePair::new( + "google_assigned_memory_usage", "Assigned memory usage for google cluster machines", + ); + + pub static ref UNMAPPED_PAGE_CACHE_MEMORY_USAGE_PAIR: GaugePair = GaugePair::new( + "google_unmapped_page_cache_memory_usage", "Unmapped page cache memory usage for google cluster machines", + ); + + pub static ref TOTAL_PAGE_CACHE_MEMORY_USAGE_PAIR: GaugePair = GaugePair::new( + "google_total_page_cache_memory_usage", "Total page cache memory usage for google cluster machines", + ); + + pub static ref MAX_MEMORY_USAGE_PAIR: GaugePair = GaugePair::new( + "google_max_memory_usage", "Maximum memory usage by google cluster machines", + ); + + pub static ref MEAN_DISK_IO_TIME_PAIR: GaugePair = GaugePair::new( + "google_mean_disk_io_time", "Mean disk I/O time for google cluster machines", + ); + + pub static ref MEAN_LOCAL_DISK_SPACE_USED_PAIR: GaugePair = GaugePair::new( + "google_mean_local_disk_space_used", "Mean local disk space used by google cluster machines", + ); + + pub static ref MAX_CPU_USAGE_PAIR: GaugePair = GaugePair::new( + "google_max_cpu_usage", "Maximum cpu usage for google cluster machines", + ); + + pub static ref MAX_DISK_IO_TIME_PAIR: GaugePair = GaugePair::new( + "google_max_disk_io_time", "Maximum disk I/O time for google cluster machines", + ); + + pub static ref CYCLES_PER_INSTRUCTION_PAIR: GaugePair = GaugePair::new( + "google_cycles_per_instruction", "Cycles per instruction for google cluster machines", + ); + + pub static ref MEMORY_ACCESSES_PER_INSTRUCTION_PAIR: GaugePair = GaugePair::new( + "google_memory_accesses_per_instruction", "Memory accesses per instruction for google cluster machines", + ); + + pub static ref SAMPLE_PORTION_PAIR: GaugePair = GaugePair::new( + "google_sample_portion", "Sample portion for google cluster machines", + ); + + pub static ref SAMPLED_CPU_USAGE_PAIR: GaugePair = GaugePair::new( + "google_sampled_cpu_usage", "Sampled cpu usage for google cluster machines", + ); +} + +/// @brief Given the part number, create a String for the filename. +/// +/// @param[in] part The csv part number such that: part ∈ [0, 500] +/// @param[in] gzipped Whether or not .gz should be appended to the filename +/// +/// @return The csv filename as a String, in the form: +/// +pub fn get_csv_filename(part: u16, gzipped: bool) -> String { + const TRU_CSV_PATH_PARTS: [&str; 4] = ["part-", "00000", "-of-00500.csv", ".gz"]; + + let mut filename = String::new(); + let part_name_str = if part < 10 { + format!("0000{}", part) + } else if (10..100).contains(&part) { + format!("000{}", part) + } else if (100..=CSV_MAX_PART_NO).contains(&part) { + format!("00{}", part) + } else { + panic!( + "Invalid part number: {} => part must be between 0 and 500", + part + ) + }; + + filename.push_str(TRU_CSV_PATH_PARTS[0]); + filename.push_str(&part_name_str); + filename.push_str(TRU_CSV_PATH_PARTS[2]); + + if gzipped { + filename.push_str(TRU_CSV_PATH_PARTS[3]); + } + + filename +} + +/// @brief Creates a new csv reader wrapped around a gzip decoder which +/// streams data from the underlying file +/// +/// @param[in] input_dir The directory containing gzipped csv files +/// @param[in] part The part number out of the total number of csv files +/// +/// @return The configured reader +fn get_reader(input_dir: &str, part: u16) -> Result, BoxedErr> { + use csv::ReaderBuilder; + use flate2::read::GzDecoder; + use std::fs::File; + use std::io::BufReader; + use std::path::Path; + + let filename: String = get_csv_filename(part, true); + let file_path = Path::new(input_dir).join(&filename); + let fd: File = File::open(file_path)?; + let buf_rdr = BufReader::new(fd); + let gz_decoder = GzDecoder::new(buf_rdr); + + let csv_rdr: CsvGzReader = ReaderBuilder::new() + .delimiter(TRU_CSV_DELIMITER) + .flexible(true) + .has_headers(false) + .from_reader(gz_decoder); + + Ok(csv_rdr) +} + +/// @brief Main routine of the helper (reader) thread. +/// +/// The purpose of the thread is to handle all of the work involved in +/// reading and enqueuing lines from the csv.gz file for the +/// main thread to then pop and export on scrape +/// +/// @param[in] input_dir The path to the directory containing the csv.gz files +/// @param[in] all_parts Whether or not to run the exporter on all 500 parts of +/// the task resource usage csv data. Running in this mode +/// and not providing all 500 parts will cause the reader +/// thread to panic. If this option is true, part_index +/// should be None +/// @param[in] part_index Specify a single part (out of 500) to read csv data +/// from. The reader thread will stop after reading this +/// single file. If part_index is not None, then all_parts +/// should be false +/// @param[in] metrics The list of metrics, or csv fields, for the exporter +/// to expose to prometheus. At least one must be given +/// +/// @pre All csv files are expected to be of the form: +/// "part-00xxx-of-00500.csv.gz" +pub fn reader_thread_routine( + input_dir: String, + all_parts: bool, + part_index: Option, + metrics: Vec, +) -> Result<(), BoxedErr> { + const QUEUE_POLL_INTERVAL_MS: u64 = 250; + GOOGLE_METRICS.set(metrics).unwrap(); + let mut part: u16 = 0_u16; + + if !all_parts { + part = part_index.unwrap(); + } + + while let Ok(mut rdr) = get_reader(&input_dir, part) { + let csv_iter = rdr.deserialize(); + for csv_line in csv_iter { + while GOOGLE_DATA_QUEUE.is_full() { + thread::sleep(Duration::from_millis(QUEUE_POLL_INTERVAL_MS)); + } + let parsed_line: TruCsvFields = csv_line?; + let _ = GOOGLE_DATA_QUEUE.push(parsed_line); + } + part += 1; + + if !all_parts || part > CSV_MAX_PART_NO { + break; + } + } + + // Never read any parts or all parts was specified and we never read all 500 + // parts of the csv data + if part == 0 || (all_parts && part <= CSV_MAX_PART_NO) { + panic!( + "Failed to read initial .csv.gz file. Check that all data files + are named in the correct format ('part-?????-of-00500.csv.gz'). + If running with --all-parts, ensure all 500 parts exist in the + input directory. + " + ); + } else { + GOOGLE_DATA_QUEUE.close(); + Ok(()) + } +} + +/// @brief: Converts the start time of a job into seconds and normalizes it +/// +/// From pg.2 of the schema doc: +/// "Each record has a timestamp, which is in microseconds since 600 +/// seconds before the beginning of the trace period, and recorded as a +/// 64 bit integer (i.e., an event 20 second after the start of the +/// trace would have a timestamp=620s)." +/// +/// @param[in] time_micros The event start time in microseconds, +/// offset by T_OFFSET_SECS (600s) +/// +/// @return A duration representing the dilated trace start time in seconds +/// after subtracting the offset +pub fn get_normalized_start_time(time_micros: u64) -> Duration { + let time_secs = time_micros / MICRO_SECONDS_PER_SECOND; + Duration::from_secs((time_secs - T_OFFSET_SECS) * DILATION_FACTOR) +} + +/// @brief Given a single parsed line from the csv file, update all gauges +/// corresponding to the metrics in the list +/// +/// @param[in] csv_line A parsed line from the csv file containing label +/// values and metric data to export +pub fn export_line(csv_line: TruCsvFields) { + let metrics = GOOGLE_METRICS.get().unwrap(); + let label_vals: [&str; 3] = [ + csv_line.job_id.as_str(), + csv_line.task_index.as_str(), + csv_line.machine_id.as_str(), + ]; + + let aggregation_type = csv_line.aggregation_type.unwrap_or(0_u8); + + for metric in metrics { + let curr_gauge: &'static GaugeVec; + let wrapped_value: Option; + + (curr_gauge, wrapped_value) = match metric { + TruMetrics::MeanCpuUsageRate => ( + MEAN_CPU_USAGE_RATE_PAIR.get(aggregation_type), + csv_line.mean_cpu_usage_rate, + ), + TruMetrics::CanonicalMemoryUsage => ( + CANONICAL_MEMORY_USAGE_PAIR.get(aggregation_type), + csv_line.canonical_memory_usage, + ), + TruMetrics::AssignedMemoryUsage => ( + ASSIGNED_MEMORY_USAGE_PAIR.get(aggregation_type), + csv_line.assigned_memory_usage, + ), + TruMetrics::UnmappedPageCacheMemoryUsage => ( + UNMAPPED_PAGE_CACHE_MEMORY_USAGE_PAIR.get(aggregation_type), + csv_line.unmapped_page_cache_memory_usage, + ), + TruMetrics::TotalPageCacheMemoryUsage => ( + TOTAL_PAGE_CACHE_MEMORY_USAGE_PAIR.get(aggregation_type), + csv_line.total_page_cache_memory_usage, + ), + TruMetrics::MaxMemoryUsage => ( + MAX_MEMORY_USAGE_PAIR.get(aggregation_type), + csv_line.max_memory_usage, + ), + TruMetrics::MeanDiskIoTime => ( + MEAN_DISK_IO_TIME_PAIR.get(aggregation_type), + csv_line.mean_disk_io_time, + ), + TruMetrics::MeanLocalDiskSpaceUsed => ( + MEAN_LOCAL_DISK_SPACE_USED_PAIR.get(aggregation_type), + csv_line.mean_local_disk_space_used, + ), + TruMetrics::MaxCpuUsage => ( + MAX_CPU_USAGE_PAIR.get(aggregation_type), + csv_line.max_cpu_usage, + ), + TruMetrics::MaxDiskIoTime => ( + MAX_DISK_IO_TIME_PAIR.get(aggregation_type), + csv_line.max_disk_io_time, + ), + TruMetrics::CyclesPerInstruction => ( + CYCLES_PER_INSTRUCTION_PAIR.get(aggregation_type), + csv_line.cycles_per_instruction, + ), + TruMetrics::MemoryAccessesPerInstruction => ( + MEMORY_ACCESSES_PER_INSTRUCTION_PAIR.get(aggregation_type), + csv_line.memory_accesses_per_instruction, + ), + TruMetrics::SamplePortion => ( + SAMPLE_PORTION_PAIR.get(aggregation_type), + csv_line.sample_portion, + ), + TruMetrics::SampledCpuUsage => ( + SAMPLED_CPU_USAGE_PAIR.get(aggregation_type), + csv_line.sampled_cpu_usage, + ), + }; + + if let Some(metric_value) = wrapped_value { + // Set the metric, unless it was missing + curr_gauge.with_label_values(&label_vals).set(metric_value); + } + } +} + +/// @brief Exports all parsed CSV lines from the queue +/// +/// This function will continue popping lines from the queue until it +/// pops one with a start timestamp which should be exported later in time. +/// This line will be saved in FUTURE_LINE and then exported on the next +/// scrape for which the program runtime <= start time +pub fn export_from_queue() { + let elapsed_t: Duration = utilities::get_time_elapsed(); + let check_time = |line: &TruCsvFields| get_normalized_start_time(line.start_time) <= elapsed_t; + + GOOGLE_DATA_QUEUE + .try_iter() + .take_while(check_time) + .for_each(export_line); + + if GOOGLE_DATA_QUEUE.is_closed() + && GOOGLE_DATA_QUEUE.is_empty() + && utilities::should_exit_after_eof() + { + info!("No more task resource usage to export, shutting down"); + std::process::exit(0); + } +} diff --git a/tools/cluster-data-exporter/src/main.rs b/tools/cluster-data-exporter/src/main.rs new file mode 100644 index 00000000..8ff4dad6 --- /dev/null +++ b/tools/cluster-data-exporter/src/main.rs @@ -0,0 +1,274 @@ +/// @NOTE: As new label-value combinations are added to each metric, +/// they will persist unless another metric with the same label-value combo +/// overwipes it. Therefore, user should be wary about the possibility +/// of program memory usage steadily increasing over the course of the runtime +use crate::alibaba_metrics::*; +use crate::google_metrics::*; +use crate::utilities::*; +use clap::Parser; +use hyper::body::Incoming; +use hyper::header::CONTENT_TYPE; +use hyper::server::conn::http1; +use hyper::service::service_fn; +use hyper::Request; +use hyper::Response; +use hyper_util::rt::TokioIo; +use prometheus::{Encoder, TextEncoder}; +use std::net::{Ipv4Addr, SocketAddr}; +use std::sync::OnceLock; +use std::{panic, process, thread}; +use tokio::net::TcpListener; +use tracing::{debug, error, info}; +use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt, EnvFilter}; + +mod alibaba_metrics; +mod google_metrics; +mod utilities; + +type BoxedErr = Box; + +/// Google or Alibaba. Must be initialized before starting export routine +static DATA_PROVIDER: OnceLock = OnceLock::new(); + +/// @brief Async call-back function for servicing http requests, like +/// prometheus scrapes +/// +/// @param[in] _req The incoming http request +/// +/// @return Prometheus metrics on success +/// BoxedErr on failure +async fn serve_req(_req: Request) -> Result, BoxedErr> { + let encoder = TextEncoder::new(); + let provider = DATA_PROVIDER.get().unwrap(); + + match provider { + Provider::Google => google_metrics::export_from_queue(), + Provider::Alibaba => alibaba_metrics::export_from_queue(), + } + + let metric_families = prometheus::gather(); + let body = encoder.encode_to_string(&metric_families)?; + let response = Response::builder() + .status(200) + .header(CONTENT_TYPE, encoder.format_type()) + .body(body)?; + + Ok(response) +} + +/// @brief Starts a thread to read and queue Google cluster data +/// +/// @param[in] input_dir The input directory to Google task resource usage +/// cluster data +/// @param[in] all_parts Whether to run the exporter across all csv parts or +/// not. This should be false if part index is not None +/// @param[in] part_index The part number, out of 500, of the csv file to use +/// when exporting task resource usage data. This should +/// be None if all_parts is true. +/// @param[in] metrics The list of metrics from the task resource usage data +/// to export +/// +/// @post All globals required by the main exporter thread are initialized. +fn start_google_thread( + input_dir: String, + all_parts: bool, + part_index: Option, + metrics: Vec, +) { + debug!("Starting Google reader thread"); + thread::spawn(move || { + // start reader thread + // Drops thread handle => thread is implicitly detached + if let Err(err) = + google_metrics::reader_thread_routine(input_dir, all_parts, part_index, metrics) + { + error!("Error in Google reader thread: {:?}", err); + process::exit(1); + } + }); + // Must be initialized before main thread starts exporting + google_metrics::GOOGLE_METRICS.wait(); + debug!("Google reader thread initialized"); +} + +/// @brief Starts a thread to read and queue Alibaba cluster data +/// +/// @param[in] input_dir The input directory containing the csv files for +/// reading +/// @param[in] all_parts Whether to run the exporter from part 0 until no more +/// csv files are found, or not. This should be false if +/// part index is not None. +/// @param[in] part_index Which csv file part to use as the data source. +/// This should be None if all_parts is true. +/// @param[in] data_type Which type of microservice data the reading thread +/// should be configured to read and queue +/// @param[in] data_year The year from which the source data comes from. Valid +/// options are 2021 and 2022 +/// @param[in] speedup Speedup factor for faster-than-realtime export +/// +/// @post All globals required by the main exporter thread are initialized. +fn start_alibaba_thread( + input_dir: String, + all_parts: bool, + part_index: Option, + data_type: MsDataType, + data_year: u32, + speedup: u64, +) { + debug!("Starting Alibaba reader thread"); + thread::spawn(move || { + if let Err(err) = alibaba_metrics::reader_thread_routine( + input_dir, all_parts, part_index, data_type, data_year, speedup, + ) { + error!("Error in Alibaba reader thread: {:?}", err); + process::exit(1); + } + }); + // Must be initialized before main thread starts exporting + alibaba_metrics::EXPORTER_DATA_TYPE.wait(); + debug!("Alibaba reader thread initialized"); +} + +/// @brief Sets up logging with optional file output +/// +/// @param[in] log_dir Optional directory for log file output +/// @param[in] log_level Log level string (DEBUG, INFO, WARN, ERROR) +/// +/// @return WorkerGuard if file logging is enabled, None otherwise. +/// The guard must be kept alive for the duration of the program. +fn setup_logging( + log_dir: Option<&str>, + log_level: &str, +) -> Result, BoxedErr> { + // Create env filter that respects RUST_LOG, with fallback to command line arg + let env_filter = EnvFilter::try_from_default_env() + .or_else(|_| EnvFilter::try_new(log_level)) + .unwrap_or_else(|_| EnvFilter::new("info")); + + if let Some(dir) = log_dir { + // Log to file AND stdout + std::fs::create_dir_all(dir)?; + let file_appender = tracing_appender::rolling::never(dir, "cluster_data_exporter.log"); + let (non_blocking, guard) = tracing_appender::non_blocking(file_appender); + + tracing_subscriber::registry() + .with(env_filter) + .with( + tracing_subscriber::fmt::layer() + .with_writer(std::io::stdout) + .with_ansi(true), + ) + .with( + tracing_subscriber::fmt::layer() + .with_writer(non_blocking) + .with_ansi(false), + ) + .init(); + + info!( + "Logging initialized with file output: {}/cluster_data_exporter.log", + dir + ); + Ok(Some(guard)) + } else { + // Log to stdout only + tracing_subscriber::registry() + .with(env_filter) + .with(tracing_subscriber::fmt::layer()) + .init(); + + info!("Logging initialized (stdout only)"); + Ok(None) + } +} + +#[tokio::main] +async fn main() -> Result<(), BoxedErr> { + let cli = Cli::parse(); + let _ = utilities::EXIT_AFTER_EOF.set(std::time::Duration::from_millis(cli.exit_after_eof_ms)); + + // Initialize logging (keep guard alive for lifetime of program) + let _log_guard = setup_logging(cli.log_dir.as_deref(), &cli.log_level)?; + + info!("Starting cluster_data_exporter"); + info!("Input directory: {}", cli.input_directory); + info!("Port: {}", cli.port); + info!("Exit-after-EOF grace period: {}ms", cli.exit_after_eof_ms); + + // This code forces the program to exit if a reader thread panics. + // Comment it out if it's preferable for the main thread to remain + let orig_hook = panic::take_hook(); + panic::set_hook(Box::new(move |panic_info| { + // invoke the default handler and then exit the process + orig_hook(panic_info); + process::exit(1); + })); + + let input_directory: String = cli.input_directory.clone(); + let port: u16 = cli.port; + let addr: SocketAddr = (Ipv4Addr::UNSPECIFIED, port).into(); + + let _ = utilities::T_START; // init t_start + + // Spin up reader thread to start queueing csv data + match cli.provider { + ProviderCmd::Google { + metrics, + all_parts, + part_index, + } => { + info!("Provider: Google"); + info!("Metrics: {:?}", metrics); + info!( + "Parts mode: {}", + if all_parts { "all-parts" } else { "part-index" } + ); + if let Some(idx) = part_index { + info!("Part index: {}", idx); + } + let _ = DATA_PROVIDER.set(Provider::Google); + start_google_thread(input_directory, all_parts, part_index, metrics); + } + ProviderCmd::Alibaba { + data_type, + data_year, + all_parts, + part_index, + speedup, + } => { + info!("Provider: Alibaba"); + info!("Data type: {:?}", data_type); + info!("Data year: {}", data_year); + info!( + "Parts mode: {}", + if all_parts { "all-parts" } else { "part-index" } + ); + if let Some(idx) = part_index { + info!("Part index: {}", idx); + } + info!("Speedup factor: {}x", speedup); + let _ = DATA_PROVIDER.set(Provider::Alibaba); + start_alibaba_thread( + input_directory, + all_parts, + part_index, + data_type, + data_year, + speedup, + ); + } + } + + let listener = TcpListener::bind(addr).await?; + info!("Server listening on http://{}", addr); + + loop { + // Main exporter routine + let (stream, _) = listener.accept().await?; + let io = TokioIo::new(stream); + let service = service_fn(serve_req); + if let Err(err) = http1::Builder::new().serve_connection(io, service).await { + error!("Server error: {:?}", err); + }; + } +} diff --git a/tools/cluster-data-exporter/src/utilities.rs b/tools/cluster-data-exporter/src/utilities.rs new file mode 100644 index 00000000..bd9af884 --- /dev/null +++ b/tools/cluster-data-exporter/src/utilities.rs @@ -0,0 +1,133 @@ +use crate::alibaba_metrics::*; +use crate::google_metrics::*; +use clap::{ArgGroup, Parser, Subcommand, ValueEnum}; +use lazy_static::lazy_static; +use std::sync::OnceLock; +use std::time::{Duration, Instant}; + +pub type BoxedErr = Box; + +/// Optional grace period after finite input is exhausted. A zero duration +/// preserves the legacy behavior of exiting during the terminal scrape. +pub static EXIT_AFTER_EOF: OnceLock = OnceLock::new(); +static EOF_REACHED_AT: OnceLock = OnceLock::new(); + +pub fn should_exit_after_eof() -> bool { + let timeout = EXIT_AFTER_EOF.get().copied().unwrap_or_default(); + timeout.is_zero() + || EOF_REACHED_AT + .get_or_init(Instant::now) + .elapsed() + .ge(&timeout) +} + +lazy_static! { + /// An instant in time to roughly represent the start time of the exporter + /// This is used as the reference point for calculating how much time has + /// elapsed, and therefore which traces should be exported during a scrape + /// and which ones should be held onto until later + pub static ref T_START: Instant = Instant::now(); +} + +/// @brief Returns the time since T_START as a Duration +/// +/// @return Duration since the Instant defined by T_START +/// +/// @note Since T_START isn't initialized until it is referenced for the first +/// time, so if this function is called before T_START is ever referenced +/// then T_START will be initialized here with Duration::Zero returned +pub fn get_time_elapsed() -> Duration { + T_START.elapsed() +} + +#[derive(Debug, Clone, ValueEnum)] +pub enum Provider { + Google, + Alibaba, +} + +#[derive(Parser, Debug)] +#[command(name = "cluster_data_exporter", version, about)] +#[command(subcommand_required = true)] +pub struct Cli { + #[arg(short, long, aliases = ["input, in, dir, input_dir"])] + #[arg(required = true)] + pub input_directory: String, + + #[arg(short, long)] + #[arg(required = true)] + pub port: u16, + + /// Grace period after the finite input is exhausted before exiting. + /// Zero preserves the legacy immediate-shutdown behavior. + #[arg(long, default_value_t = 0)] + pub exit_after_eof_ms: u64, + + /// Log level (DEBUG, INFO, WARN, ERROR) + #[arg(long, default_value = "INFO")] + pub log_level: String, + + /// Output directory for log files (optional, defaults to stdout only) + #[arg(long)] + pub log_dir: Option, + + #[command(subcommand)] + pub provider: ProviderCmd, +} + +#[derive(Subcommand, Debug)] +pub enum ProviderCmd { + /// Run the exporter on google task resource usage data + #[command(group(ArgGroup::new("csv-parts") + .args(&["all_parts", "part_index"]) + .required(true)) + )] + Google { + #[arg(long, value_enum, value_delimiter = ',', num_args = 1..)] + #[arg(required = true, require_equals = true)] + metrics: Vec, + + #[arg(long, group = "csv-parts", alias = "all")] + all_parts: bool, + + #[arg(long, group = "csv-parts", aliases = ["part", "index"])] + #[arg(require_equals = true)] + part_index: Option, + }, + + /// Run the exporter on Alibaba microservice data + #[command(group(ArgGroup::new("csv-parts") + .args(&["all_parts", "part_index"]) + .required(true)) + )] + Alibaba { + /// The type of microservice data to use + #[arg(long, value_enum)] + #[arg(required = true, require_equals = true)] + data_type: MsDataType, + + /// Which year the microservice data comes from + #[arg(long)] + #[arg(required = true, require_equals = true)] + #[arg(value_parser = clap::value_parser!(u32).range(2021..=2022))] + data_year: u32, + + /// Whether or not to run the exporter starting on part 0 of the csv + /// files and continue sequentially until no more files are found. + /// This option is mutually exclusive with --part-index + #[arg(long, group = "csv-parts", alias = "all")] + all_parts: bool, + + /// Specify a single csv file to use as trace data. + /// This option is mutually exclusive with --all-parts + #[arg(long, group = "csv-parts", aliases = ["part", "index"])] + #[arg(require_equals = true)] + part_index: Option, + + /// Speedup factor for faster-than-realtime export + /// 1 = real-time, 10 = 10x faster, 100 = 100x faster + #[arg(long, require_equals = true)] + #[arg(value_parser = clap::value_parser!(u64).range(1..))] + speedup: u64, + }, +} diff --git a/tools/cluster-data-exporter/tests/fixtures/alibaba/msresource.csv b/tools/cluster-data-exporter/tests/fixtures/alibaba/msresource.csv new file mode 100644 index 00000000..66d91d24 --- /dev/null +++ b/tools/cluster-data-exporter/tests/fixtures/alibaba/msresource.csv @@ -0,0 +1,3 @@ +timestamp,nodeid,msname,msinstanceid,instance_cpu_usage,instance_memory_usage +0,node-a,frontend,frontend-1,0.5,0.25 +60000,node-b,worker,worker-1,0.75,0.5 diff --git a/tools/cluster-data-exporter/tests/fixtures/alibaba/node.csv b/tools/cluster-data-exporter/tests/fixtures/alibaba/node.csv new file mode 100644 index 00000000..0e1fe83c --- /dev/null +++ b/tools/cluster-data-exporter/tests/fixtures/alibaba/node.csv @@ -0,0 +1,3 @@ +timestamp,nodeid,node_cpu_usage,node_memory_usage +0,node-a,0.5,0.25 +60000,node-b,0.75,0.5 diff --git a/tools/cluster-data-exporter/tests/fixtures/google/part-00000-of-00500.csv b/tools/cluster-data-exporter/tests/fixtures/google/part-00000-of-00500.csv new file mode 100644 index 00000000..dd061f1f --- /dev/null +++ b/tools/cluster-data-exporter/tests/fixtures/google/part-00000-of-00500.csv @@ -0,0 +1,2 @@ +600000000,600000001,job-a,0,machine-a,0.5,,,,,,,,,,,,,, +601000000,601000001,job-b,1,machine-b,0.75,,,,,,,,,,,,,, diff --git a/tools/cluster-data-exporter/tests/smoke.sh b/tools/cluster-data-exporter/tests/smoke.sh new file mode 100755 index 00000000..c7898aee --- /dev/null +++ b/tools/cluster-data-exporter/tests/smoke.sh @@ -0,0 +1,103 @@ +#!/usr/bin/env bash +set -Eeuo pipefail + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +TOOL_DIR="$(cd "${SCRIPT_DIR}/.." && pwd)" +FIXTURE_DIR="${SCRIPT_DIR}/fixtures" +WORK_DIR="$(mktemp -d)" +PORT_BASE="${CDE_TEST_PORT_BASE:-19400}" +CURRENT_PID="" + +cleanup() { + if [[ -n "${CURRENT_PID}" ]]; then + kill "${CURRENT_PID}" 2>/dev/null || true + wait "${CURRENT_PID}" 2>/dev/null || true + fi + rm -rf "${WORK_DIR}" +} +trap cleanup EXIT + +require() { + command -v "$1" >/dev/null || { + echo "missing required command: $1" >&2 + exit 1 + } +} + +for command in cargo curl gzip; do + require "${command}" +done + +cargo build --manifest-path "${TOOL_DIR}/Cargo.toml" +EXPORTER="${TOOL_DIR}/target/debug/cluster_data_exporter" + +start_exporter() { + local port="$1" + local input_directory="$2" + shift 2 + "${EXPORTER}" -i "${input_directory}" -p "${port}" --exit-after-eof-ms=2000 "$@" & + CURRENT_PID="$!" + for _ in $(seq 1 80); do + if curl -fsS "http://127.0.0.1:${port}/metrics" >/dev/null 2>&1; then + return + fi + sleep 0.05 + done + echo "exporter did not become ready on port ${port}" >&2 + exit 1 +} + +assert_metric() { + local port="$1" + local metric="$2" + local label_fragment="$3" + curl -fsS "http://127.0.0.1:${port}/metrics" | + grep -F "${metric}" | + grep -Fq "${label_fragment}" +} + +stop_exporter() { + kill "${CURRENT_PID}" + wait "${CURRENT_PID}" 2>/dev/null || true + CURRENT_PID="" +} + +google_dir="${WORK_DIR}/google" +mkdir -p "${google_dir}" +# Verifies Google task-usage CSV replay and metric labels. +gzip -c "${FIXTURE_DIR}/google/part-00000-of-00500.csv" > "${google_dir}/part-00000-of-00500.csv.gz" +start_exporter "${PORT_BASE}" "${google_dir}" google --metrics=mean-cpu-usage-rate --part-index=0 +assert_metric "${PORT_BASE}" "google_mean_cpu_usage_rate_0" 'job_id="job-a"' +stop_exporter + +for year in 2021 2022; do + node_dir="${WORK_DIR}/node-${year}" + mkdir -p "${node_dir}" + # Verifies both Alibaba Node filename conventions and metric output. + node_name="Node_0.csv.gz" + if [[ "${year}" == 2022 ]]; then + node_name="NodeMetrics_0.csv.gz" + fi + gzip -c "${FIXTURE_DIR}/alibaba/node.csv" > "${node_dir}/${node_name}" + port=$((PORT_BASE + year - 2020)) + start_exporter "${port}" "${node_dir}" alibaba --data-type=node --data-year="${year}" --part-index=0 --speedup=1 + assert_metric "${port}" "alibaba_node_cpu_usage" 'node_id="node-a"' + stop_exporter +done + +for year in 2021 2022; do + ms_dir="${WORK_DIR}/msresource-${year}" + mkdir -p "${ms_dir}" + # Verifies both Alibaba MSResource filename conventions and metric output. + ms_name="MSResource_0.csv.gz" + if [[ "${year}" == 2022 ]]; then + ms_name="MSMetrics_0.csv.gz" + fi + gzip -c "${FIXTURE_DIR}/alibaba/msresource.csv" > "${ms_dir}/${ms_name}" + port=$((PORT_BASE + year - 2018)) + start_exporter "${port}" "${ms_dir}" alibaba --data-type=ms-resource --data-year="${year}" --part-index=0 --speedup=1 + assert_metric "${port}" "alibaba_microservice_cpu_usage" 'ms_name="frontend"' + stop_exporter +done + +echo "cluster-data-exporter smoke checks passed" From cac8c4fb428eb7bcdac551f0ba8b3b1247f5991d Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Sat, 12 Sep 2026 20:55:35 -0400 Subject: [PATCH 2/2] Fix unused enumerate index in compatibility test --- data_plane/tests/asapquery_compatibility_process_e2e.rs | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/data_plane/tests/asapquery_compatibility_process_e2e.rs b/data_plane/tests/asapquery_compatibility_process_e2e.rs index 45397fcd..6d491f7f 100644 --- a/data_plane/tests/asapquery_compatibility_process_e2e.rs +++ b/data_plane/tests/asapquery_compatibility_process_e2e.rs @@ -48,8 +48,7 @@ fn quote_snapshot_for_frontend_test( let quotes = workload_cost::with_exact_alternative(request) .unwrap() .into_iter() - .enumerate() - .filter_map(|(_index, candidate)| { + .filter_map(|candidate| { let plan = if metricsql { PhysicalCompiler.compile_metricsql(candidate.clone(), environment.clone()) } else {