✅ Update benchmark

README.md

@@ -47,9 +47,10 @@ dynamically structured logs, as well as log compression.
 
 Run with (see *Build* section bellow):
 
-```
-$ task build
-$ task bench
+```bash
+export BENCHMARK_ITERATIONS="1000000"
+export BENCHMARK_JOBS="24"  # usually, number of CPUs
+task docker:build test:bench
 ```
 
 System:
@@ -59,9 +60,11 @@ System:
 
 Result:
 
- - **Total requests sent:** 1,000,000
- - **Total time:** 481.65s
- - **Average:** 2076.18 request/s
+```
+Requests sent: 1000000
+Total time:    266.75s
+Rate:          3748.76 req/s
+```
 
 ## :construction: Build
 

internal/data/auth/hash/hash_test.go

@@ -21,3 +21,30 @@ func TestHashPassword(t *testing.T) {
 		t.Fatal("VerifyPassword() = false; want true")
 	}
 }
+
+func BenchmarkHashPassword(b *testing.B) {
+	for n := 0; n < b.N; n++ {
+		_, err := hash.HashPassword("password")
+		if err != nil {
+			b.Fatalf("HashPassword() error = %v", err)
+		}
+	}
+}
+
+func BenchmarkVerifyPassword(b *testing.B) {
+	h, err := hash.HashPassword("password")
+	if err != nil {
+		b.Fatalf("HashPassword() error = %v", err)
+	}
+
+	for n := 0; n < b.N; n++ {
+		ok, err := hash.VerifyPassword("password", h)
+		if err != nil {
+			b.Fatalf("VerifyPassword() error = %v", err)
+		}
+
+		if !ok {
+			b.Fatal("VerifyPassword() = false; want true")
+		}
+	}
+}

scripts/Taskfile.test.yml

@@ -23,7 +23,7 @@ tasks:
   "test:unit:go":
     internal: true
     cmds:
-      - go test ./...
+      - go test -bench=. ./...
 
   "test:unit:rust:filterdsl":
     internal: true

tests/benchmark/.gitignore

@@ -1,2 +1,2 @@
-/data/logs/
-/data/auth/
+/data
+/.venv

tests/benchmark/config/pipelines/test.json

@@ -0,0 +1 @@
+{"nodes":[{"id":"__builtin__source_direct","type":"source","position":{"x":-90,"y":-30},"data":{"type":"direct"}},{"id":"__builtin__source_syslog","type":"source","position":{"x":-90,"y":105},"data":{"type":"syslog"}},{"id":"node-1","type":"transform","position":{"x":255,"y":-30},"data":{"transformer":"syslog"}},{"id":"node-2","type":"switch","position":{"x":600,"y":-30},"data":{"condition":"appname  = \"iam01\""}},{"id":"node-3","type":"switch","position":{"x":600,"y":105},"data":{"condition":"appname = \"iam02\""}},{"id":"node-4","type":"switch","position":{"x":600,"y":240},"data":{"condition":"appname = \"db\""}},{"id":"node-5","type":"transform","position":{"x":915,"y":-30},"data":{"transformer":"logfmt"}},{"id":"node-6","type":"transform","position":{"x":915,"y":105},"data":{"transformer":"json"}},{"id":"node-7","type":"transform","position":{"x":915,"y":240},"data":{"transformer":"logfmt"}},{"id":"node-8","type":"switch","position":{"x":600,"y":375},"data":{"condition":"appname = \"apache2\""}},{"id":"node-9","type":"transform","position":{"x":915,"y":375},"data":{"transformer":"apache"}},{"id":"node-10","type":"router","position":{"x":1275,"y":30},"data":{"stream":"users"}},{"id":"node-11","type":"router","position":{"x":1275,"y":240},"data":{"stream":"database"}},{"id":"node-12","type":"router","position":{"x":1275,"y":375},"data":{"stream":"access"}}],"edges":[{"id":"xy-edge____builtin__source_direct-node-1","source":"__builtin__source_direct","sourceHandle":"","target":"node-1"},{"id":"xy-edge__node-1-node-2","source":"node-1","sourceHandle":"","target":"node-2"},{"id":"xy-edge__node-1-node-3","source":"node-1","sourceHandle":"","target":"node-3"},{"id":"xy-edge__node-1-node-4","source":"node-1","sourceHandle":"","target":"node-4"},{"id":"xy-edge__node-2-node-5","source":"node-2","sourceHandle":"","target":"node-5"},{"id":"xy-edge__node-3-node-6","source":"node-3","sourceHandle":"","target":"node-6"},{"id":"xy-edge__node-4-node-7","source":"node-4","sourceHandle":"","target":"node-7"},{"id":"xy-edge__node-1-node-8","source":"node-1","sourceHandle":"","target":"node-8"},{"id":"xy-edge__node-8-node-9","source":"node-8","sourceHandle":"","target":"node-9"},{"id":"xy-edge__node-5-node-10","source":"node-5","sourceHandle":"","target":"node-10"},{"id":"xy-edge__node-6-node-10","source":"node-6","sourceHandle":"","target":"node-10"},{"id":"xy-edge__node-7-node-11","source":"node-7","sourceHandle":"","target":"node-11"},{"id":"xy-edge__node-9-node-12","source":"node-9","sourceHandle":"","target":"node-12"}]}

tests/benchmark/data/config/transformers/apache.vrl → tests/benchmark/config/transformers/apache.vrl

@@ -1,2 +1 @@
 . = parse_apache_log!(.message, "common")
-.

tests/benchmark/data/config/transformers/json.vrl → tests/benchmark/config/transformers/json.vrl

@@ -1,2 +1 @@
 . = parse_json!(.message)
-.

tests/benchmark/data/config/transformers/logfmt.vrl → tests/benchmark/config/transformers/logfmt.vrl

@@ -1,2 +1 @@
 . = parse_logfmt!(.message)
-.

tests/benchmark/data/config/transformers/syslog.vrl → tests/benchmark/config/transformers/syslog.vrl

@@ -1,2 +1 @@
 . = parse_syslog!(.message)
-.

tests/benchmark/generate-logs.py

@@ -1,15 +1,17 @@
-#!/usr/bin/env python3
-
 from argparse import ArgumentParser
 import tomllib
+import os
 
 from datetime import datetime
-from time import time
+from time import perf_counter_ns
+import multiprocessing
 import random
 
-from urllib.request import Request, urlopen
+import requests
 import json
 
+from tqdm import tqdm
+
 
 def format_syslog(hostname: str, appname: str, message: str) -> str:
     timestamp = datetime.now().strftime("%b %d %H:%M:%S")
@@ -18,63 +20,71 @@ def format_syslog(hostname: str, appname: str, message: str) -> str:
 
 def send_log(token: str, log: str):
     payload = json.dumps({"record": {"message": log}}).encode("utf-8")
-    req = Request(
+    resp = requests.post(
         "http://localhost:5080/api/v1/pipelines/test/logs",
         data=payload,
         headers={
             "Content-Type": "application/json",
             "Authorization": f"Bearer pat:{token}",
         },
     )
+    resp.raise_for_status()
+
 
-    with urlopen(req) as resp:
-        assert resp.status == 200
+def iteration(args: tuple[str, dict]):
+    token, data = args
+
+    hostname = random.choice(data["hosts"])
+    app = random.choice(data["apps"])
+    message = random.choice(app["messages"])
+
+    send_log(token, format_syslog(hostname, app["name"], message))
 
 
 def main():
     parser = ArgumentParser()
     parser.add_argument("--token", required=True)
-    parser.add_argument("--conftest", default="conftest.toml")
-    parser.add_argument("--req-count", type=int, default=1_000_000)
+    parser.add_argument("--conftest", default="testdata.toml")
+
+    parser.add_argument(
+        "--repeat",
+        type=int,
+        default=int(os.getenv("BENCHMARK_ITERATIONS", 1_000_000)),
+    )
+    parser.add_argument(
+        "--jobs",
+        type=int,
+        default=int(os.getenv("BENCHMARK_JOBS", multiprocessing.cpu_count())),
+    )
+
     args = parser.parse_args()
 
     with open(args.conftest, "rb") as f:
         data = tomllib.load(f)
 
-    req_count = args.req_count
-    total_sent = 0
-    total_time = 0
-
-    while req_count > 0:
-        start = time()
-
-        batch_count = 0
-        while time() - start < 1:
-            hostname = random.choice(data["hosts"])
-            app = random.choice(data["apps"])
-            message = random.choice(app["messages"])
-
-            send_log(
-                args.token,
-                format_syslog(hostname, app["name"], message),
-            )
+    start_time = perf_counter_ns()
 
-            batch_count += 1
-            req_count -= 1
+    with multiprocessing.Pool(args.jobs) as pool:
+        params = [(args.token, data) for _ in range(args.repeat)]
+        iterable = tqdm(
+            pool.imap_unordered(iteration, params),
+            total=args.repeat,
+            desc="Sending logs",
+            unit="req",
+        )
 
-            if req_count <= 0:
-                break
+        for _ in iterable:
+            pass
 
-        end = time()
-        elapsed = end - start
-        total_time += elapsed
-        total_sent += batch_count
+    end_time = perf_counter_ns()
 
-        perc = (args.req_count - req_count) / args.req_count * 100
-        print(f"progress={perc:.2f}% batch_count={batch_count} elapsed={elapsed:.2f}s")
+    total_time_ns = end_time - start_time
+    total_time_s = total_time_ns / 1_000_000_000
+    rate = args.repeat / total_time_s
 
-    print(f"total_sent={total_sent} total_time={total_time:.2f}s")
-    print(f"avg_rate=\"{total_sent / total_time:.2f} req/s\"")
+    print(f"Requests sent: {args.repeat}")
+    print(f"Total time:    {total_time_s:.2f}s")
+    print(f"Rate:          {rate:.2f} req/s")
 
 
 if __name__ == "__main__":

tests/benchmark/requirements.txt

@@ -0,0 +1,6 @@
+certifi==2024.8.30
+charset-normalizer==3.3.2
+idna==3.10
+requests==2.32.3
+tqdm==4.66.5
+urllib3==2.2.3

tests/benchmark/run.sh

@@ -4,4 +4,12 @@ set -e
 
 . ../flowg.sh
 
+sudo cp -r config/* data/config/
+
+python -m venv .venv
+. .venv/bin/activate
+pip install -r requirements.txt
+
+echo "--------------------------------------------------------------------------------"
+
 python generate-logs.py --token $FLOWG_ADMIN_TOKEN

tests/e2e/.gitignore

@@ -1,3 +1,2 @@
-/data/logs/
-/data/auth/
+/data
 /reports

tests/e2e/run.sh

@@ -13,13 +13,4 @@ hurl \
   --report-html reports/html \
   --report-junit reports/junit.xml \
   --jobs 1 \
-  --test integration/
-
-if [ -z "$NOBENCHMARK" ]
-then
-  hurl \
-    --variable admin_token=${FLOWG_ADMIN_TOKEN} \
-    --variable guest_token=${FLOWG_GUEST_TOKEN} \
-    --repeat 1000 \
-    --test benchmark/
-fi
+  --test spec/