From 8c249f672ab58d9d0690137974a1b040863487fb Mon Sep 17 00:00:00 2001 From: Gabor Gyimesi Date: Wed, 24 Jun 2026 13:37:58 +0200 Subject: [PATCH] MINIFICPP-2854 Add benchmark scripts repositories --- benchmarks/repository_benchmark/README.md | 101 +++++ .../repository_benchmark/generate_report.py | 247 +++++++++++ .../repository_benchmark/requirements.txt | 3 + .../resources/generate_config.json | 69 +++ .../resources/get_config.json | 69 +++ .../repository_benchmark/results/.gitignore | 2 + benchmarks/repository_benchmark/run_batch.py | 109 +++++ .../repository_benchmark/run_benchmark.py | 405 ++++++++++++++++++ 8 files changed, 1005 insertions(+) create mode 100644 benchmarks/repository_benchmark/README.md create mode 100644 benchmarks/repository_benchmark/generate_report.py create mode 100644 benchmarks/repository_benchmark/requirements.txt create mode 100644 benchmarks/repository_benchmark/resources/generate_config.json create mode 100644 benchmarks/repository_benchmark/resources/get_config.json create mode 100644 benchmarks/repository_benchmark/results/.gitignore create mode 100644 benchmarks/repository_benchmark/run_batch.py create mode 100644 benchmarks/repository_benchmark/run_benchmark.py diff --git a/benchmarks/repository_benchmark/README.md b/benchmarks/repository_benchmark/README.md new file mode 100644 index 0000000000..91c6231b7a --- /dev/null +++ b/benchmarks/repository_benchmark/README.md @@ -0,0 +1,101 @@ + + +# Repository benchmark + +Tools for comparing MiNiFi C++ FlowFile and content repository implementations +(RocksDB, LMDB, filesystem, volatile) under a controlled workload. Each run +starts a MiNiFi container with a chosen repository combination, samples its +resource usage over time, and records throughput; a report overlays several runs +so the implementations can be compared side by side. + +## Prerequisites + +- Docker (the current user must be able to run containers). +- A MiNiFi C++ Docker image (e.g. built via `make docker`), referenced with + `--image`. +- Python 3.10+ and the dependencies: + + ```bash + python3 -m venv venv && source venv/bin/activate + pip install -r requirements.txt + ``` + +## Single run + +```bash +python3 run_benchmark.py \ + --image apacheminificpp:1.0.0 \ + --flowfile-repository lmdb \ + --content-repository lmdb +``` + +Results are written to `results/___.json` +unless `--output` is given. + +### Input generation modes (`--input-file-generation-type`) + +- `timed_getfile` (default): write input files into a mounted directory at a + fixed interval; a `GetFile` processor ingests them. Runs for `--duration`. +- `timed_generateflowfile`: a `GenerateFlowFile` processor produces flow files + in-process at a fixed interval. Runs for `--duration`. +- `burst`: generate `--input-file-count` files up front, then ingest them all + with `GetFile`. Ends once every file has been processed (bounded by a timeout). + +Other useful flags: `--duration`, `--input-interval`, `--input-file-size` +(accepts `512K`, `1M`, `1G`), `--input-file-count`, `--metrics-interval`. + +## Batch run (compare several combinations) + +`run_batch.py` runs a set of repository combinations against the same workload, +**sequentially** (containers do not compete for CPU/IO, which keeps the numbers +comparable), and can generate the report automatically: + +```bash +python3 run_batch.py \ + --image apacheminificpp:1.0.0 \ + --combo lmdb:lmdb \ + --combo rocksdb:rocksdb \ + --combo rocksdb:filesystem \ + --input-file-generation-type burst \ + --input-file-count 100 \ + --report report.html +``` + +`--combo FLOWFILE:CONTENT` is repeatable. The remaining workload flags are shared +by every combo and match `run_benchmark.py`. If one combo fails the others still +run, and the failure is listed in the final summary. + +## Report + +To build a report from existing result files directly: + +```bash +python3 generate_report.py results/*.json -o report.html +``` + +The report contains a **summary table** for quick comparison plus time-series +charts. Metrics: + +| Metric | Meaning | +| --- | --- | +| Throughput (ff/s) | Flow files processed per second (from container logs). | +| Files processed | Total flow files processed during the run. | +| Peak / mean memory | Container memory usage (`inactive_file` excluded). | +| Mean / p95 CPU | Container CPU usage; 100% = one core. | +| Final FF / content repo | On-disk repository size at the end of the run (`du`). | +| Peak content repo | Largest content repository size observed. | diff --git a/benchmarks/repository_benchmark/generate_report.py b/benchmarks/repository_benchmark/generate_report.py new file mode 100644 index 0000000000..acc4d5d18d --- /dev/null +++ b/benchmarks/repository_benchmark/generate_report.py @@ -0,0 +1,247 @@ +#!/usr/bin/env python3 +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import argparse +import json +import os +import statistics +from collections import Counter + +CHART_JS_CDN = "https://cdn.jsdelivr.net/npm/chart.js@4.4.1/dist/chart.umd.min.js" + +MIB = 1024 * 1024 + +HTML_TEMPLATE = """ + + + +MiNiFi C++ Repository Benchmark Report + + + + +

MiNiFi C++ Repository Benchmark Report

+

Summary

+{summary_table} +

Runs

+{config_table} +
+
+
+
+
+ + + +""" + + +def build_config_table(runs: list[dict]) -> str: + columns = [ + ("Label", lambda r: r["label"]), + ("FlowFile repo", lambda r: r["config"].get("flowfile_repository", "")), + ("Content repo", lambda r: r["config"].get("content_repository", "")), + ("File size (B)", lambda r: r["config"].get("input_file_size_bytes", "")), + ("Input interval (s)", lambda r: r["config"].get("input_interval_s", "")), + ("Metrics interval (s)", lambda r: r["config"].get("metrics_interval_s", "")), + ("Duration (s)", lambda r: r["config"].get("duration_s", "")), + ("Samples", lambda r: len(r["samples"])), + ("Input generation type", lambda r: r["config"].get("input_file_generation_type", "")), + ("Input file count", lambda r: r["config"].get("input_file_count", "")), + ] + header = "".join(f"{name}" for name, _ in columns) + rows = "" + for run in runs: + cells = "".join(f"{getter(run)}" for _, getter in columns) + rows += f"{cells}" + return f"{header}{rows}
" + + +def load_run(path: str) -> dict: + with open(path) as result_file: + data = json.load(result_file) + config = data.get("config", {}) + combo = "{}/{}".format( + config.get("flowfile_repository", "?"), + config.get("content_repository", "?"), + ) + return { + "combo": combo, + "path": path, + "config": config, + "samples": data.get("samples", []), + "throughput": data.get("throughput", 0), + "flow_files_processed": data.get("flow_files_processed"), + } + + +def assign_labels(runs: list[dict]) -> None: + # Use the repo combination as the label, only disambiguating with the file + # name when the same combination appears more than once. + combo_counts = Counter(run["combo"] for run in runs) + for run in runs: + if combo_counts[run["combo"]] > 1: + run["label"] = f"{run['combo']} ({os.path.basename(run['path'])})" + else: + run["label"] = run["combo"] + + +def compute_summary(run: dict) -> dict: + samples = run["samples"] + memory = [s.get("memory_bytes", 0) for s in samples] + cpu = [s.get("cpu_percent", 0) for s in samples] + flowfile_sizes = [s.get("flowfile_repo_bytes", 0) for s in samples] + content_sizes = [s.get("content_repo_bytes", 0) for s in samples] + + def peak(values: list) -> float: + return max(values) if values else 0 + + def mean(values: list) -> float: + return statistics.fmean(values) if values else 0 + + def p95(values: list) -> float: + if not values: + return 0 + ordered = sorted(values) + index = min(len(ordered) - 1, int(round(0.95 * (len(ordered) - 1)))) + return ordered[index] + + processed = run.get("flow_files_processed") + final_content = content_sizes[-1] if content_sizes else 0 + + return { + "peak_memory_mib": peak(memory) / MIB, + "mean_memory_mib": mean(memory) / MIB, + "mean_cpu": mean(cpu), + "p95_cpu": p95(cpu), + "final_flowfile_mib": (flowfile_sizes[-1] if flowfile_sizes else 0) / MIB, + "final_content_mib": final_content / MIB, + "peak_content_mib": peak(content_sizes) / MIB, + "throughput": run["throughput"], + "flow_files_processed": processed if processed is not None else "n/a", + } + + +def build_summary_table(runs: list[dict]) -> str: + columns = [ + ("Run", lambda r, s: r["label"]), + ("Throughput (ff/s)", lambda r, s: f"{s['throughput']:.2f}"), + ("Files processed", lambda r, s: s["flow_files_processed"]), + ("Peak mem (MiB)", lambda r, s: f"{s['peak_memory_mib']:.1f}"), + ("Mean mem (MiB)", lambda r, s: f"{s['mean_memory_mib']:.1f}"), + ("Mean CPU (%)", lambda r, s: f"{s['mean_cpu']:.1f}"), + ("p95 CPU (%)", lambda r, s: f"{s['p95_cpu']:.1f}"), + ("Final FF repo (MiB)", lambda r, s: f"{s['final_flowfile_mib']:.1f}"), + ("Final content repo (MiB)", lambda r, s: f"{s['final_content_mib']:.1f}"), + ("Peak content repo (MiB)", lambda r, s: f"{s['peak_content_mib']:.1f}"), + ] + header = "".join(f"{name}" for name, _ in columns) + rows = "" + for run in runs: + summary = compute_summary(run) + cells = "".join(f"{getter(run, summary)}" for _, getter in columns) + rows += f"{cells}" + return f"{header}{rows}
" + + +def write_report(result_paths: list[str], output_path: str) -> None: + runs = [load_run(path) for path in result_paths] + assign_labels(runs) + + html = HTML_TEMPLATE.format( + chart_js_cdn=CHART_JS_CDN, + summary_table=build_summary_table(runs), + config_table=build_config_table(runs), + runs_json=json.dumps(runs), + ) + + with open(output_path, "w") as output_file: + output_file.write(html) + + print(f"Report with {len(runs)} run(s) written to {output_path}") + + +def main() -> None: + parser = argparse.ArgumentParser(description="Generate an HTML report from benchmark results.") + parser.add_argument("results", nargs="+", help="Benchmark result JSON files.") + parser.add_argument("-o", "--output", default="report.html", help="Output HTML file path.") + args = parser.parse_args() + + write_report(args.results, args.output) + + +if __name__ == "__main__": + main() diff --git a/benchmarks/repository_benchmark/requirements.txt b/benchmarks/repository_benchmark/requirements.txt new file mode 100644 index 0000000000..7ac092201c --- /dev/null +++ b/benchmarks/repository_benchmark/requirements.txt @@ -0,0 +1,3 @@ +docker==7.1.0 +humanfriendly==10.0 +Jinja2==3.1.6 diff --git a/benchmarks/repository_benchmark/resources/generate_config.json b/benchmarks/repository_benchmark/resources/generate_config.json new file mode 100644 index 0000000000..2b39093fd8 --- /dev/null +++ b/benchmarks/repository_benchmark/resources/generate_config.json @@ -0,0 +1,69 @@ +{ + "rootGroup": { + "name": "MiNiFi Flow", + "processors": [ + { + "name": "Generate flow files", + "identifier": "2f2a3b47-f5ba-49f6-82b5-bc1c86b96e27", + "type": "org.apache.nifi.processors.standard.GenerateFlowFile", + "schedulingStrategy": "TIMER_DRIVEN", + "schedulingPeriod": "{{ generate_file_interval }} sec", + "properties": { + "File Size": "{{ generate_file_size }}" + }, + "autoTerminatedRelationships": [] + }, + { + "name": "Update flow file attributes", + "identifier": "69ea76e6-e872-460f-8d24-ef8fc80b51b6", + "type": "org.apache.nifi.processors.standard.UpdateAttribute", + "schedulingStrategy": "EVENT_DRIVEN", + "properties": { + "flow_file_count": "${nextInt()}" + }, + "autoTerminatedRelationships": [] + }, + { + "name": "LogAttribute", + "identifier": "e143601d-de4f-44ba-a6ec-d1f97d77ec94", + "type": "org.apache.nifi.processors.standard.LogAttribute", + "schedulingStrategy": "EVENT_DRIVEN", + "properties": { + }, + "autoTerminatedRelationships": [ + "success" + ] + } + ], + "connections": [ + { + "identifier": "098a56ba-f4bf-4323-a3f3-6f8a5e3586bf", + "name": "GenerateFlowFile/success/UpdateAttribute", + "source": { + "id": "2f2a3b47-f5ba-49f6-82b5-bc1c86b96e27" + }, + "destination": { + "id": "69ea76e6-e872-460f-8d24-ef8fc80b51b6" + }, + "selectedRelationships": [ + "success" + ] + }, + { + "identifier": "198a56ba-f4bf-4323-a3f3-6f8a5e3586bf", + "name": "UpdateAttribute/success/LogAttribute", + "source": { + "id": "69ea76e6-e872-460f-8d24-ef8fc80b51b6" + }, + "destination": { + "id": "e143601d-de4f-44ba-a6ec-d1f97d77ec94" + }, + "selectedRelationships": [ + "success" + ] + } + ], + "remoteProcessGroups": [], + "controllerServices": [] + } +} diff --git a/benchmarks/repository_benchmark/resources/get_config.json b/benchmarks/repository_benchmark/resources/get_config.json new file mode 100644 index 0000000000..16d69f9712 --- /dev/null +++ b/benchmarks/repository_benchmark/resources/get_config.json @@ -0,0 +1,69 @@ +{ + "rootGroup": { + "name": "MiNiFi Flow", + "processors": [ + { + "name": "Get files from /tmp/input", + "identifier": "2f2a3b47-f5ba-49f6-82b5-bc1c86b96e27", + "type": "org.apache.nifi.processors.standard.GetFile", + "schedulingStrategy": "TIMER_DRIVEN", + "schedulingPeriod": "{{ get_file_interval }} sec", + "properties": { + "Input Directory": "/tmp/input" + }, + "autoTerminatedRelationships": [] + }, + { + "name": "Update flow file attributes", + "identifier": "69ea76e6-e872-460f-8d24-ef8fc80b51b6", + "type": "org.apache.nifi.processors.standard.UpdateAttribute", + "schedulingStrategy": "EVENT_DRIVEN", + "properties": { + "flow_file_count": "${nextInt()}" + }, + "autoTerminatedRelationships": [] + }, + { + "name": "LogAttribute", + "identifier": "e143601d-de4f-44ba-a6ec-d1f97d77ec94", + "type": "org.apache.nifi.processors.standard.LogAttribute", + "schedulingStrategy": "EVENT_DRIVEN", + "properties": { + }, + "autoTerminatedRelationships": [ + "success" + ] + } + ], + "connections": [ + { + "identifier": "098a56ba-f4bf-4323-a3f3-6f8a5e3586bf", + "name": "GetFile/success/UpdateAttribute", + "source": { + "id": "2f2a3b47-f5ba-49f6-82b5-bc1c86b96e27" + }, + "destination": { + "id": "69ea76e6-e872-460f-8d24-ef8fc80b51b6" + }, + "selectedRelationships": [ + "success" + ] + }, + { + "identifier": "198a56ba-f4bf-4323-a3f3-6f8a5e3586bf", + "name": "UpdateAttribute/success/LogAttribute", + "source": { + "id": "69ea76e6-e872-460f-8d24-ef8fc80b51b6" + }, + "destination": { + "id": "e143601d-de4f-44ba-a6ec-d1f97d77ec94" + }, + "selectedRelationships": [ + "success" + ] + } + ], + "remoteProcessGroups": [], + "controllerServices": [] + } +} diff --git a/benchmarks/repository_benchmark/results/.gitignore b/benchmarks/repository_benchmark/results/.gitignore new file mode 100644 index 0000000000..d6b7ef32c8 --- /dev/null +++ b/benchmarks/repository_benchmark/results/.gitignore @@ -0,0 +1,2 @@ +* +!.gitignore diff --git a/benchmarks/repository_benchmark/run_batch.py b/benchmarks/repository_benchmark/run_batch.py new file mode 100644 index 0000000000..a985a6544e --- /dev/null +++ b/benchmarks/repository_benchmark/run_batch.py @@ -0,0 +1,109 @@ +#!/usr/bin/env python3 +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import argparse +import os +import humanfriendly +import run_benchmark +from argparse import Namespace +from datetime import datetime, timezone +from run_benchmark import CONTENT_REPOSITORY_CLASSES, FLOWFILE_REPOSITORY_CLASSES, InputGenerationType + + +def parse_combo(value: str) -> tuple[str, str]: + parts = value.split(":") + if len(parts) != 2: + raise argparse.ArgumentTypeError(f"Combo must be 'flowfile:content', got '{value}'.") + flowfile, content = parts + if flowfile not in FLOWFILE_REPOSITORY_CLASSES: + raise argparse.ArgumentTypeError( + f"Unknown flowfile repository '{flowfile}', choose from {sorted(FLOWFILE_REPOSITORY_CLASSES)}.") + if content not in CONTENT_REPOSITORY_CLASSES: + raise argparse.ArgumentTypeError( + f"Unknown content repository '{content}', choose from {sorted(CONTENT_REPOSITORY_CLASSES)}.") + return flowfile, content + + +def run_combo(args: argparse.Namespace, flowfile: str, content: str, output_dir: str) -> str: + timestamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + name = f"{timestamp}_{flowfile}_{content}_{args.input_file_generation_type}.json" + output_path = os.path.join(output_dir, name) + combo_args = Namespace( + image=args.image, + flowfile_repository=flowfile, + content_repository=content, + input_file_generation_type=args.input_file_generation_type, + input_file_count=args.input_file_count, + duration=args.duration, + input_interval=args.input_interval, + input_file_size=args.input_file_size, + metrics_interval=args.metrics_interval, + output=output_path, + ) + return run_benchmark.run(combo_args) + + +def main() -> None: + parser = argparse.ArgumentParser( + description="Run the MiNiFi C++ repository benchmark across several repository combinations sequentially and (optionally) generate a single comparison report. " + "Runs are sequential by design so containers do not compete for CPU/IO.") + parser.add_argument("--image", required=True, help="Docker image to use for the benchmark.") + parser.add_argument("--combo", required=True, action="append", type=parse_combo, dest="combos", + metavar="FLOWFILE:CONTENT", + help="Repository combination to benchmark, e.g. --combo lmdb:lmdb. Repeatable.") + parser.add_argument("--input-file-generation-type", default=InputGenerationType.TIMED_GETFILE.value, + choices=sorted([e.value for e in InputGenerationType]), + help="Input file generation type shared by all combos (see run_benchmark.py).") + parser.add_argument("--input-file-count", type=int, default=100, + help="Number of input files for burst input generation type (default: 100).") + parser.add_argument("--duration", type=int, default=120, + help="Total benchmark session length in seconds (default: 120).") + parser.add_argument("--input-interval", type=float, default=1.0, + help="Seconds between input file generation cycles (default: 1).") + parser.add_argument("--input-file-size", type=humanfriendly.parse_size, default=humanfriendly.parse_size("1M"), + help="Size of each generated input file, e.g. 512K, 1M, 1G (default: 1M).") + parser.add_argument("--metrics-interval", type=float, default=5.0, + help="Seconds between metric samples (default: 5).") + parser.add_argument("--output-dir", default=run_benchmark.RESULTS_DIR, + help="Directory for result JSON files (default: results/).") + parser.add_argument("--report", default=None, + help="If set, generate an HTML report at this path from all successful runs.") + args = parser.parse_args() + + os.makedirs(args.output_dir, exist_ok=True) + + result_paths: list[str] = [] + failures: list[tuple[str, str, str]] = [] + for index, (flowfile, content) in enumerate(args.combos, start=1): + print(f"\n=== [{index}/{len(args.combos)}] Benchmarking {flowfile}/{content} ===") + try: + result_paths.append(run_combo(args, flowfile, content, args.output_dir)) + except Exception as error: + print(f"Error: combo {flowfile}/{content} failed with: {error}") + failures.append((flowfile, content, str(error))) + + print("\n=== Batch summary ===") + print(f"Succeeded: {len(result_paths)}/{len(args.combos)}") + for flowfile, content, error in failures: + print(f" FAILED {flowfile}/{content}: {error}") + + if args.report and result_paths: + import generate_report + generate_report.write_report(result_paths, args.report) + + +if __name__ == "__main__": + main() diff --git a/benchmarks/repository_benchmark/run_benchmark.py b/benchmarks/repository_benchmark/run_benchmark.py new file mode 100644 index 0000000000..a76c5385d5 --- /dev/null +++ b/benchmarks/repository_benchmark/run_benchmark.py @@ -0,0 +1,405 @@ +#!/usr/bin/env python3 +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import argparse +import re +import docker +import humanfriendly +import json +import os +import shutil +import tempfile +import threading +import time +import jinja2 +from datetime import datetime, timezone +from enum import Enum + +SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) +RESOURCES_DIR = os.path.join(SCRIPT_DIR, "resources") +GET_CONFIG_FILE_TEMPLATE = "get_config.json" +GENERATE_CONFIG_FILE_TEMPLATE = "generate_config.json" +RESULTS_DIR = os.path.join(SCRIPT_DIR, "results") + +MINIFI_HOME = "/opt/minifi/minifi-current" +FLOWFILE_REPO_DIR = f"{MINIFI_HOME}/flowfile_repository" +CONTENT_REPO_DIR = f"{MINIFI_HOME}/content_repository" +INPUT_DIR = "/tmp/input" + +FLOWFILE_REPOSITORY_CLASSES = { + "rocksdb": "FlowFileRepository", + "lmdb": "LmdbFlowFileRepository", + "volatile": "VolatileFlowFileRepository", +} + +CONTENT_REPOSITORY_CLASSES = { + "rocksdb": "DatabaseContentRepository", + "lmdb": "LmdbContentRepository", + "filesystem": "FileSystemRepository", + "volatile": "VolatileContentRepository", +} + +LOG_TIMESTAMP_RE = re.compile(r'\[(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.\d+)\]') + + +class InputGenerationType(str, Enum): + TIMED_GETFILE = "timed_getfile" + TIMED_GENERATEFLOWFILE = "timed_generateflowfile" + BURST = "burst" + + +def build_properties(flowfile_repository: str, content_repository: str) -> dict[str, str]: + properties = { + "nifi.flow.configuration.file": f"{MINIFI_HOME}/conf/config.yml", + "nifi.extension.path": "../extensions/*", + "nifi.administrative.yield.duration": "1 sec", + "nifi.bored.yield.duration": "100 millis", + "nifi.openssl.fips.support.enable": "false", + "nifi.provenance.repository.class.name": "NoOpRepository", + "nifi.flowfile.repository.directory.default": FLOWFILE_REPO_DIR, + "nifi.database.content.repository.directory.default": CONTENT_REPO_DIR, + "nifi.flowfile.repository.class.name": FLOWFILE_REPOSITORY_CLASSES[flowfile_repository], + "nifi.content.repository.class.name": CONTENT_REPOSITORY_CLASSES[content_repository], + } + return properties + + +def write_properties_file(properties: dict[str, str], path: str) -> None: + with open(path, "w") as properties_file: + for key, value in properties.items(): + properties_file.write(f"{key}={value}\n") + + +def repo_size(container, path: str) -> int: + exit_code, output = container.exec_run(["du", "-sk", path]) + if exit_code != 0: + return 0 + try: + return int(output.decode().split()[0]) * 1024 + except (ValueError, IndexError): + return 0 + + +def read_container_stats(container) -> tuple[int, int, int, int]: + stats = container.stats(stream=False, one_shot=True) + memory_stats = stats.get("memory_stats", {}) + usage = memory_stats.get("usage") + if usage is None: + mem = 0 + else: + inactive_file = memory_stats.get("stats", {}).get("inactive_file", 0) + mem = max(usage - inactive_file, 0) + + cpu_stats = stats.get("cpu_stats", {}) + cpu_total = cpu_stats.get("cpu_usage", {}).get("total_usage", 0) + system_cpu = cpu_stats.get("system_cpu_usage", 0) + num_cpus = cpu_stats.get("online_cpus") or len(cpu_stats.get("cpu_usage", {}).get("percpu_usage") or [1]) + return mem, cpu_total, system_cpu, num_cpus + + +def generate_single_input(input_dir: str, file_size: int, index: int) -> None: + data = os.urandom(file_size) + # Write to a temp name then rename so GetFile never reads a partial file. + tmp_path = os.path.join(input_dir, f".{index}.tmp") + final_path = os.path.join(input_dir, f"input_{index}.bin") + with open(tmp_path, "wb") as input_file: + input_file.write(data) + os.rename(tmp_path, final_path) + + +def generate_input(input_dir: str, input_count: int, file_size: int) -> None: + for i in range(1, input_count + 1): + generate_single_input(input_dir, file_size, i) + + +def input_generator_loop(stop_event: threading.Event, input_dir: str, interval: float, file_size: int) -> None: + counter = 0 + while not stop_event.is_set(): + counter += 1 + generate_single_input(input_dir, file_size, counter) + stop_event.wait(interval) + + +def metrics_collector_loop(stop_event: threading.Event, container, samples: list, interval: float, start: float) -> None: + next_sample = time.monotonic() + prev_cpu_total = None + prev_system_cpu = None + while not stop_event.is_set(): + memory_bytes, cpu_total, system_cpu, num_cpus = read_container_stats(container) + if prev_cpu_total is not None and system_cpu > prev_system_cpu: + cpu_percent = (cpu_total - prev_cpu_total) / (system_cpu - prev_system_cpu) * num_cpus * 100.0 + else: + cpu_percent = 0.0 + prev_cpu_total, prev_system_cpu = cpu_total, system_cpu + sample = { + "elapsed_s": round(time.monotonic() - start, 3), + "flowfile_repo_bytes": repo_size(container, FLOWFILE_REPO_DIR), + "content_repo_bytes": repo_size(container, CONTENT_REPO_DIR), + "memory_bytes": memory_bytes, + "cpu_percent": round(cpu_percent, 2), + } + samples.append(sample) + next_sample += interval + stop_event.wait(max(0.0, next_sample - time.monotonic())) + + +def wait_for_minifi_to_start(container, timeout: float = 30.0) -> None: + deadline = time.monotonic() + timeout + since = datetime.fromisoformat(container.attrs["Created"]) + while time.monotonic() < deadline: + container.reload() + if container.status != "running": + time.sleep(0.5) + continue + now = datetime.now(timezone.utc) + logs = container.logs(since=since).decode(errors="replace") + since = now + if "MiNiFi started" in logs: + return + time.sleep(0.5) + raise RuntimeError(f"Container did not reach running state (status: {container.status})") + + +def write_config_yml(args: argparse.Namespace, work_dir: str) -> None: + jinja_env = jinja2.Environment(loader=jinja2.FileSystemLoader(RESOURCES_DIR)) + if args.input_file_generation_type != InputGenerationType.TIMED_GENERATEFLOWFILE: + flow_config_template = jinja_env.get_template(GET_CONFIG_FILE_TEMPLATE) + flow_config = flow_config_template.render(get_file_interval=args.input_interval) + with open(os.path.join(work_dir, "config.yml"), "w") as config_file: + config_file.write(flow_config) + else: + flow_config_template = jinja_env.get_template(GENERATE_CONFIG_FILE_TEMPLATE) + flow_config = flow_config_template.render(generate_file_interval=args.input_interval, + generate_file_size=args.input_file_size) + with open(os.path.join(work_dir, "config.yml"), "w") as config_file: + config_file.write(flow_config) + + +def create_minifi_container(args: argparse.Namespace, work_dir: str, input_dir: str) -> docker.models.containers.Container: + properties_path = os.path.join(work_dir, "minifi.properties") + properties = build_properties(args.flowfile_repository, args.content_repository) + write_properties_file(properties, properties_path) + + client = docker.from_env() + + container = client.containers.run( + args.image, + detach=True, + volumes={ + properties_path: {"bind": f"{MINIFI_HOME}/conf/minifi.properties", "mode": "ro"}, + os.path.join(work_dir, "config.yml"): {"bind": f"{MINIFI_HOME}/conf/config.yml", "mode": "ro"}, + input_dir: {"bind": INPUT_DIR, "mode": "rw"}, + }, + ) + return container + + +def wait_for_flow_files_to_be_processed(container, expected_count: int, timeout: float = 300.0) -> None: + deadline = time.monotonic() + timeout + since = datetime.fromisoformat(container.attrs["Created"]) + while True: + now = datetime.now(timezone.utc) + logs = container.logs(since=since).decode(errors="replace") + since = now + if f"key:flow_file_count value:{expected_count - 1}" in logs: + break + container.reload() + if container.status != "running": + raise RuntimeError(f"Container exited before processing {expected_count} flow files (status: {container.status})") + if time.monotonic() > deadline: + raise RuntimeError(f"Timed out after {timeout}s waiting for {expected_count} flow files to be processed") + time.sleep(0.1) + + # Wait a bit more to see how the repositories behave after all flow files have been processed. + time.sleep(2) + + +def run_threads(samples: list[dict], args: argparse.Namespace) -> tuple[float, int]: + stop_event = threading.Event() + container = None + work_dir = tempfile.mkdtemp(prefix="repo_benchmark_") + input_dir = os.path.join(work_dir, "input") + os.makedirs(input_dir) + write_config_yml(args, work_dir) + try: + if args.input_file_generation_type == InputGenerationType.TIMED_GETFILE: + start = time.monotonic() + container = create_minifi_container(args, work_dir, input_dir) + wait_for_minifi_to_start(container) + threads = [ + threading.Thread( + target=input_generator_loop, + args=(stop_event, input_dir, args.input_interval, args.input_file_size), + daemon=True, + ), + threading.Thread( + target=metrics_collector_loop, + args=(stop_event, container, samples, args.metrics_interval, start), + daemon=True, + ), + ] + elif args.input_file_generation_type == InputGenerationType.BURST: + generate_input(input_dir, args.input_file_count, args.input_file_size) + start = time.monotonic() + container = create_minifi_container(args, work_dir, input_dir) + threads = [ + threading.Thread( + target=metrics_collector_loop, + args=(stop_event, container, samples, args.metrics_interval, start), + daemon=True, + ), + ] + else: + start = time.monotonic() + container = create_minifi_container(args, work_dir, input_dir) + wait_for_minifi_to_start(container) + threads = [ + threading.Thread( + target=metrics_collector_loop, + args=(stop_event, container, samples, args.metrics_interval, start), + daemon=True, + ), + ] + + for thread in threads: + thread.start() + + if args.input_file_generation_type != InputGenerationType.BURST: + time.sleep(args.duration) + stop_event.set() + else: + print(f"Waiting for {args.input_file_count} flow files to be processed...") + wait_for_flow_files_to_be_processed(container, args.input_file_count) + stop_event.set() + + for thread in threads: + thread.join(timeout=10) + + return calculate_throughput(container) + finally: + stop_event.set() + try: + if container is not None: + container.stop() + finally: + if container is not None: + container.remove() + shutil.rmtree(work_dir, ignore_errors=True) + + +def parse_timestamp(log_line: str) -> datetime | None: + match = LOG_TIMESTAMP_RE.search(log_line) + if match: + return datetime.strptime(match.group(1), "%Y-%m-%d %H:%M:%S.%f") + return None + + +def calculate_throughput(container) -> tuple[float, int]: + logs = container.logs().decode(errors="replace") + count = 0 + first_timestamp = None + last_timestamp = None + for line in logs.splitlines(): + if first_timestamp is None and "MiNiFi started" in line: + timestamp = parse_timestamp(line) + if timestamp is not None: + first_timestamp = timestamp + continue + if "Logging for flow file" in line: + timestamp = parse_timestamp(line) + if timestamp is not None: + last_timestamp = timestamp + count += 1 + + if first_timestamp is None or last_timestamp is None: + return 0.0, count + elapsed = (last_timestamp - first_timestamp).total_seconds() + if elapsed <= 0: + return 0.0, count + return count / elapsed, count + + +def write_results(samples: list[dict], throughput: float, flow_files_processed: int, args: argparse.Namespace) -> str: + result = { + "config": { + "image": args.image, + "flowfile_repository": args.flowfile_repository, + "content_repository": args.content_repository, + "duration_s": args.duration, + "input_interval_s": args.input_interval, + "input_file_size_bytes": args.input_file_size, + "metrics_interval_s": args.metrics_interval, + "input_file_generation_type": args.input_file_generation_type, + "input_file_count": args.input_file_count, + }, + "samples": samples, + "throughput": throughput, + "flow_files_processed": flow_files_processed, + } + + output_path = args.output + if output_path is None: + os.makedirs(RESULTS_DIR, exist_ok=True) + timestamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + name = f"{timestamp}_{args.flowfile_repository}_{args.content_repository}_{args.input_file_generation_type}.json" + output_path = os.path.join(RESULTS_DIR, name) + else: + os.makedirs(os.path.dirname(os.path.abspath(output_path)), exist_ok=True) + + with open(output_path, "w") as output_file: + json.dump(result, output_file, indent=2) + + print(f"Collected {len(samples)} samples; throughput: {throughput:.2f} flow files/sec; results written to {output_path}") + return output_path + + +def run(args) -> str: + samples: list[dict] = [] + throughput, flow_files_processed = run_threads(samples, args) + return write_results(samples, throughput, flow_files_processed, args) + + +def main() -> None: + parser = argparse.ArgumentParser(description="Run the MiNiFi C++ repository benchmark.") + parser.add_argument("--image", required=True, help="Docker image to use for the benchmark.") + parser.add_argument("--flowfile-repository", required=True, + choices=sorted(FLOWFILE_REPOSITORY_CLASSES), help="Flowfile repository type.") + parser.add_argument("--content-repository", required=True, + choices=sorted(CONTENT_REPOSITORY_CLASSES), help="Content repository type.") + parser.add_argument("--input-file-generation-type", default=InputGenerationType.TIMED_GETFILE.value, + choices=sorted(list([e.value for e in InputGenerationType])), + help="Input file generation type. timed_getfile: Generate input files at a fixed interval and use GetFile processor to ingest them. " + "timed_generateflowfile: Use GenerateFlowFile processor to generate flowfiles at a fixed interval. " + "burst: Generate a burst of input files at the start of the benchmark and use GetFile processor to ingest them.") + parser.add_argument("--input-file-count", type=int, default=100, + help="Number of input files to generate for burst input generation type (default: 100).") + parser.add_argument("--duration", type=int, default=120, + help="Total benchmark session length in seconds (default: 120).") + parser.add_argument("--input-interval", type=float, default=1.0, + help="Seconds between input file generation cycles (default: 1).") + parser.add_argument("--input-file-size", type=humanfriendly.parse_size, default=humanfriendly.parse_size("1M"), + help="Size of each generated input file, e.g. 512K, 1M, 1G (default: 1M).") + parser.add_argument("--metrics-interval", type=float, default=1.0, + help="Seconds between metric samples (default: 1).") + parser.add_argument("--output", default=None, + help="Output JSON path (default: results/__.json).") + args = parser.parse_args() + + run(args) + + +if __name__ == "__main__": + main()