diff --git a/test/e2e/01-install.sh b/test/e2e/01-install.sh index 83207a32d..a9e7357c5 100755 --- a/test/e2e/01-install.sh +++ b/test/e2e/01-install.sh @@ -82,7 +82,7 @@ echo "Fetching access tokens..." mkdir -p "${SA_TOKEN_PATH}" service_accounts=(all-namespaces-read-access single-namespace-read-access all-namespaces-admin-access all-namespaces-impersonate-access) for service_account in "${service_accounts[@]}"; do - kubectl create token "$service_account" > "${SA_TOKEN_PATH}"/"$service_account" + kubectl create token "$service_account" > "${SA_TOKEN_PATH}"/"$service_account" --duration=24h echo "Created ${SA_TOKEN_PATH}/$service_account" done diff --git a/test/performance/README.md b/test/performance/README.md new file mode 100644 index 000000000..1df07d980 --- /dev/null +++ b/test/performance/README.md @@ -0,0 +1,154 @@ +# Tekton Results — Performance Benchmark Framework + +A repeatable harness for measuring **store** (write/ingest) and **query** +(read/list) performance of Tekton Results — separately and in parallel — always +starting from an identical, versioned seed dataset. It drives the **real deployed +API** only (gRPC for writes, REST/gRPC for reads); it never touches the database +directly. + +Use it to prove improvement and catch regressions across schema changes (metadata +columns, label tables, indexes) and new code paths. + +``` +generator/ deterministic PipelineRun/TaskRun generator + golden DatasetDefinition +metrics/ latency percentiles, error-by-code counters, throughput (importable) +report/ JSON report schema + metadata capture (importable) +harness/ cobra "bench" CLI + store/query/mixed/load drivers (package main) +datasets/ dataset specs (see seed-small.md) +environments/ kind cluster + install automation (local & external DB) +baselines/ committed baseline reports (follow-up) +``` + +`generator`, `metrics`, and `report` are importable packages — the compare tool +(Story 02) and compliance suite (Story 03) reuse the generator, datasets, and +report schema. + +## Quick start (tier-1 kind, local DB) + +```bash +# 1. Cluster + install with fixed Postgres tuning +./test/performance/environments/kind/00-kind-up.sh +./test/performance/environments/kind/01-install-localdb.sh + +# The install prints the cert/token paths; export them for convenience: +export SSL_CERT_PATH=/tmp/tekton-results/ssl +export SA_TOKEN_PATH=/tmp/tekton-results/tokens +export API_SERVER_ADDR=https://localhost:8080 + +CERT="${SSL_CERT_PATH}/tekton-results-cert.pem" +TOKEN="${SA_TOKEN_PATH}/all-namespaces-admin-access" + +# 2. Load + verify the seed dataset through the API +go run ./test/performance/harness load --verify --cert "$CERT" --token "$TOKEN" + +# 3. Benchmark +go run ./test/performance/harness store --cert "$CERT" --token "$TOKEN" --output store.json +go run ./test/performance/harness query --cert "$CERT" --token "$TOKEN" --output query.json --transport both +go run ./test/performance/harness mixed --cert "$CERT" --token "$TOKEN" --output mixed.json --duration 60 +``` + +## Subcommands + +| Command | What it measures | +| --------- | ---------------------------------------------------------------------- | +| `load` | Loads the seed dataset via the API; `--verify` checks row counts against the golden definition. | +| `store` | Write/ingest — replays the watcher lifecycle (CreateResult → CreateRecord(pending) → 1–3 UpdateRecord → UpdateResult → 2–15 child records) over the **live** UID range. | +| `query` | Read/list — a fixed CEL query mix (by status, recent window, by type, by label) against the **seed** range, paginated. `--transport grpc\|rest\|both`. | +| `mixed` | Writers (live range) and readers (seed range) in parallel; sized by `--read-ratio`/`--write-ratio`. | +| `dataset` | Emits the golden `DatasetDefinition` JSON — no cluster required. | + +## Common flags + +Defaults come from the same environment variables the e2e suite uses. + +| Flag | Default | Env | +| ----------------- | ------------------------------------ | ----------------- | +| `--server-addr` | `https://localhost:8080` | `API_SERVER_ADDR` | +| `--server-name` | `tekton-results-api-service…svc…` | `API_SERVER_NAME` | +| `--cert` | — | `SSL_CERT_PATH` | +| `--token` | — | `SA_TOKEN_PATH` | +| `--concurrency` | `8` | | +| `--count` | `1000` (count-bounded runs) | | +| `--duration` | `0` → 30s default for time-bounded | | +| `--seed` | `42` | | +| `--namespaces` | `50` | | +| `--child-min/max` | `2` / `15` | | +| `--db-backend` | `local` | `BENCH_DB_BACKEND`| +| `--output` | stdout | | + +## Dataset determinism + +The generator is seeded (`math/rand/v2` PCG per index) and uses a fixed absolute +time window — never `time.Now()` — so a given `(seed, count, namespaces)` always +produces byte-identical objects and a stable `content_hash`. Seed data and live +(store/mixed) data draw UIDs from **disjoint** ranges so writes never collide with +loaded reads. See [`datasets/seed-small.md`](datasets/seed-small.md) for the +tier-1 spec and golden fingerprint. + +Templates are anonymized real PipelineRun/TaskRun manifests dropped into +[`generator/templates/`](generator/templates/) per the contract documented there; +a sample ships so the framework builds and tests run before real manifests arrive. + +## Report schema + +Every run emits a JSON `Report` (`report.SchemaVersion`) — the comparison input +for Story 02: + +```jsonc +{ + "schema_version": "1", + "meta": { + "git_commit": "…", "git_dirty": false, + "dataset_version": "seed-small-v1", "dataset_hash": "…", + "tier": "tier1", "mode": "store", + "started_at": "…", "duration_ms": 1234, + "hostname": "…", "api_server_addr": "…", + "db_backend": "local", "server_image": "…" + }, + "config": { "count": 1000, "concurrency": 8, "transport": "grpc", "seed": 42, … }, + "metrics": { + "throughput_per_sec": 812.3, + "total_ops": 5000, "total_errors": 0, + "by_op": { + "create_record": { "count": 1000, "errors": 0, "error_codes": {}, + "p50_ms": 3.1, "p90_ms": 7.4, "p99_ms": 19.0, + "min_ms": 1.2, "max_ms": 41.0, "mean_ms": 4.0 } + } + } +} +``` + +Percentiles are exact (nearest-rank over sorted samples); errors are classified by +gRPC status code. Map keys serialize in stable order. + +## External database + +To benchmark against a managed/external Postgres instead of the in-cluster one: + +```bash +export BENCH_DB_URL="postgres://user:pass@host:5432/results?sslmode=require" +./test/performance/environments/kind/02-install-externaldb.sh +``` + +This rewires only configuration (`DB_*` config + credentials secret) — no code or +image change — and sets `db_backend=external` in reports. The harness is unchanged +because it only talks to the API. + +## Comparing runs + +Reports are self-describing (git commit, dataset hash, tier, DB backend). To +compare two runs, diff `metrics.by_op[*].p99_ms` and `throughput_per_sec` for the +same `mode` and `dataset_hash`. Guideline variance for a pass/fail gate is ±10% on +p99. Committed reference runs live in [`baselines/`](baselines/) (populated as a +follow-up). + +## Tests + +```bash +go test ./test/performance/... +``` + +Covers generator determinism and distributions, UID non-overlap, metrics +percentile math, report round-trip, and harness workload wiring. The CI smoke run +stands up local-db kind, loads `seed-small`, then runs `store|query|mixed` capped +at 1k records and asserts each emits valid JSON. diff --git a/test/performance/baselines/.gitkeep b/test/performance/baselines/.gitkeep new file mode 100644 index 000000000..de6ccbfc4 --- /dev/null +++ b/test/performance/baselines/.gitkeep @@ -0,0 +1,3 @@ +# Committed baseline reports land here (one JSON per mode/tier/dataset version). +# Populated as a follow-up: run store|query|mixed against main and commit the +# reports as tier1---baseline.json. diff --git a/test/performance/datasets/seed-small.md b/test/performance/datasets/seed-small.md new file mode 100644 index 000000000..766f5942c --- /dev/null +++ b/test/performance/datasets/seed-small.md @@ -0,0 +1,82 @@ +# Dataset: `seed-small` (tier-1) + +The tier-1 seed dataset used for CI smoke runs and the committed tier-1 baseline. +It is **deterministic**: the same generator seed always produces byte-identical +objects, so the content hash below is a versioned fingerprint. If any value in +this file changes, treat it as a dataset-version bump and refresh the committed +baseline. + +## Parameters + +| Parameter | Value | Flag | +| ---------------- | ------ | -------------- | +| PipelineRuns | 1000 | `--count 1000` | +| RNG seed | 42 | `--seed 42` | +| Namespaces | 50 | `--namespaces 50` | +| Child TaskRuns | 2–15 | `--child-min 2 --child-max 15` | +| Dataset version | `seed-small` | `--dataset-version seed-small` | + +Reproduce the golden definition with no cluster: + +```bash +go run ./test/performance/harness dataset \ + --count 1000 --seed 42 --namespaces 50 > /tmp/seed-small.json +``` + +## Golden fingerprint + +| Field | Expected value | +| -------------------- | ----------------------------------------------------------------- | +| `content_hash` | `4ab4cbcfbdbfe4e89e455eaf13971edef720d6fa1e24a8fbedf7e53d3dc60fa5` | +| PipelineRun records | 1000 | +| TaskRun records | 3028 | +| Total records | 4028 | + +The content hash covers only the generated objects (UIDs, namespaces, labels, +outcomes, timestamps), never wall-clock metadata, so it is stable across runs and +machines. `bench load --verify` recomputes it and asserts the loaded record +counts match this definition. + +## Distributions + +Outcomes (target 85% / 10% / 5%): + +| Outcome | Count | Share | +| ---------- | ----- | ------ | +| succeeded | 845 | 84.5% | +| failed | 101 | 10.1% | +| cancelled | 54 | 5.4% | + +Namespaces: 1000 PipelineRuns spread across `ns-00`..`ns-49` (~20 each). + +Labels — a small set of hot values plus a long tail: + +`appstudio.openshift.io/component` + +| Value | Count | +| -------- | ----- | +| backend | 413 | +| frontend | 401 | +| db | 33 | +| api | 24 | +| auth | 23 | +| cache | 23 | +| ingest | 22 | +| worker | 22 | +| reporting| 20 | +| gateway | 19 | + +`pipelinesascode.tekton.dev/event-type` + +| Value | Count | +| ------------ | ----- | +| push | 780 | +| retest | 85 | +| pull_request | 71 | +| incoming | 64 | + +## UID ranges + +Seed and live (store/mixed) streams draw UIDs from **disjoint** UUIDv5 ranges, so +a store or mixed run never collides with the loaded seed data. The ranges are +recorded in the definition's `uid_range` and in every report's metadata. diff --git a/test/performance/environments/kind/00-kind-up.sh b/test/performance/environments/kind/00-kind-up.sh new file mode 100755 index 000000000..5b6a3c1d9 --- /dev/null +++ b/test/performance/environments/kind/00-kind-up.sh @@ -0,0 +1,24 @@ +#!/usr/bin/env bash +# Copyright 2026 The Tekton Authors +# +# Licensed 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. +# +# Stand up the tier-1 kind cluster. Delegates to the e2e cluster setup so the +# benchmark runs against the exact node topology the conformance suite uses. + +set -euo pipefail + +ROOT="$(git rev-parse --show-toplevel)" + +echo "Creating kind cluster for performance benchmarks..." +"${ROOT}/test/e2e/00-setup.sh" diff --git a/test/performance/environments/kind/01-install-localdb.sh b/test/performance/environments/kind/01-install-localdb.sh new file mode 100755 index 000000000..6ce55c116 --- /dev/null +++ b/test/performance/environments/kind/01-install-localdb.sh @@ -0,0 +1,64 @@ +#!/usr/bin/env bash +# Copyright 2026 The Tekton Authors +# +# Licensed 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. +# +# Install Tekton Results with the bundled local Postgres, then apply the fixed +# performance tuning (pinned resources + postgresql.conf) for reproducible runs. +# +# This delegates the full app install (Pipelines, certs, tokens, ko deploy) to +# the e2e installer, then layers Postgres tuning on top imperatively so no ko +# rebuild is needed for the database. + +set -euo pipefail + +ROOT="$(git rev-parse --show-toplevel)" +HERE="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +NAMESPACE="tekton-pipelines" + +# Expose the API on localhost so the harness can reach it without port-forward +# gymnastics; the e2e installer honours this when generating the TLS cert. +export SSL_INCLUDE_LOCALHOST=${SSL_INCLUDE_LOCALHOST:-"true"} + +echo "Installing Tekton Results (standard e2e deployment)..." +"${ROOT}/test/e2e/01-install.sh" + +echo "Applying Postgres performance tuning..." +kubectl create configmap postgres-tuning \ + --namespace="${NAMESPACE}" \ + --from-file=postgresql.conf="${HERE}/postgresql.conf" \ + --dry-run=client -o yaml | kubectl apply -f - + +kubectl patch statefulset tekton-results-postgres \ + --namespace="${NAMESPACE}" \ + --type=strategic \ + --patch-file "${HERE}/postgres-patch.yaml" + +echo "Waiting for Postgres to roll out with tuning applied..." +kubectl rollout status statefulset/tekton-results-postgres --namespace="${NAMESPACE}" --timeout=180s +kubectl wait deployment tekton-results-api --namespace="${NAMESPACE}" --for=condition=available --timeout=120s + +cat <&2 + exit 1 +fi + +# Parse BENCH_DB_URL into the discrete DB_* settings the API expects. +# Format: ://:@:/[?sslmode=] +proto_stripped="${BENCH_DB_URL#*://}" +creds="${proto_stripped%%@*}" +hostpart="${proto_stripped#*@}" +DB_USER="${creds%%:*}" +DB_PASSWORD="${creds#*:}" +hostport="${hostpart%%/*}" +DB_HOST="${hostport%%:*}" +DB_PORT="${hostport##*:}" +[ "${DB_PORT}" = "${DB_HOST}" ] && DB_PORT="5432" # no explicit port +pathpart="${hostpart#*/}" +DB_NAME="${pathpart%%\?*}" +if [[ "${BENCH_DB_URL}" == *"sslmode="* ]]; then + DB_SSLMODE="${BENCH_DB_URL##*sslmode=}" + DB_SSLMODE="${DB_SSLMODE%%&*}" +else + DB_SSLMODE="require" +fi + +if [ -z "${DB_HOST}" ] || [ -z "${DB_NAME}" ] || [ -z "${DB_USER}" ]; then + echo "could not parse BENCH_DB_URL (need host, db name and user)" >&2 + exit 1 +fi + +echo "External DB target: ${DB_USER}@${DB_HOST}:${DB_PORT}/${DB_NAME} (sslmode=${DB_SSLMODE})" + +echo "Installing Tekton Pipelines..." +TEKTON_PIPELINE_CONFIG=${TEKTON_PIPELINE_CONFIG:-"https://infra.tekton.dev/tekton-releases/pipeline/latest/release.yaml"} +kubectl apply --filename "${TEKTON_PIPELINE_CONFIG}" + +echo "Creating DB credentials secret..." +kubectl create secret generic tekton-results-postgres \ + --namespace="${NAMESPACE}" \ + --from-literal=POSTGRES_USER="${DB_USER}" \ + --from-literal=POSTGRES_PASSWORD="${DB_PASSWORD}" \ + --dry-run=client -o yaml | kubectl apply -f - + +echo "Generating TLS key pair..." +mkdir -p "${SSL_CERT_PATH}" +altNames="DNS:tekton-results-api-service.${NAMESPACE}.svc.cluster.local,DNS:localhost" +openssl req -x509 \ + -newkey rsa:4096 \ + -keyout "${SSL_CERT_PATH}/tekton-results-key.pem" \ + -out "${SSL_CERT_PATH}/tekton-results-cert.pem" \ + -days 365 -nodes \ + -subj "/CN=tekton-results-api-service.${NAMESPACE}.svc.cluster.local" \ + -addext "subjectAltName = ${altNames}" +kubectl create secret tls tekton-results-tls \ + --namespace="${NAMESPACE}" \ + --cert="${SSL_CERT_PATH}/tekton-results-cert.pem" \ + --key="${SSL_CERT_PATH}/tekton-results-key.pem" \ + --dry-run=client -o yaml | kubectl apply -f - + +echo "Deploying Tekton Results (base-only, no in-cluster DB)..." +extra_ko_params="linux/$(go env GOARCH)" +kubectl kustomize "${ROOT}/config/overlays/base-only" | ko apply --platform="${extra_ko_params}" -f - + +echo "Applying benchmark RBAC (service accounts + access) ..." +kubectl apply -f "${ROOT}/test/e2e/kustomize/rbac.yaml" + +echo "Exposing the API on NodePort 30080..." +kubectl patch service tekton-results-api-service \ + --namespace="${NAMESPACE}" \ + --type=json \ + --patch "$(cat "${ROOT}/test/e2e/kustomize/api-service.yaml")" + +echo "Rewriting api-config with external DB settings..." +config_tmp="$(mktemp)" +trap 'rm -f "${config_tmp}"' EXIT +sed -e "s|^DB_HOST=.*|DB_HOST=${DB_HOST}|" \ + -e "s|^DB_PORT=.*|DB_PORT=${DB_PORT}|" \ + -e "s|^DB_NAME=.*|DB_NAME=${DB_NAME}|" \ + -e "s|^DB_SSLMODE=.*|DB_SSLMODE=${DB_SSLMODE}|" \ + "${ROOT}/config/base/env/config" > "${config_tmp}" +kubectl create configmap api-config \ + --namespace="${NAMESPACE}" \ + --from-file=config="${config_tmp}" \ + --dry-run=client -o yaml | kubectl apply -f - + +echo "Restarting the API to pick up external DB settings..." +kubectl rollout restart deployment/tekton-results-api --namespace="${NAMESPACE}" + +echo "Fetching access tokens..." +mkdir -p "${SA_TOKEN_PATH}" +for sa in all-namespaces-read-access single-namespace-read-access all-namespaces-admin-access all-namespaces-impersonate-access; do + kubectl create token "${sa}" > "${SA_TOKEN_PATH}/${sa}" +done + +echo "Waiting for the API to be ready..." +kubectl rollout status deployment/tekton-results-api --namespace="${NAMESPACE}" --timeout=180s + +cat <&2 + exit 1 + fi + echo "ok: ${file}" +} + +echo "== load --verify ==" +go run "${HARNESS}" load --verify "${common[@]}" --output "${OUT}/load.json" +assert_json "${OUT}/load.json" + +echo "== store ==" +go run "${HARNESS}" store "${common[@]}" --output "${OUT}/store.json" +assert_json "${OUT}/store.json" + +echo "== query ==" +go run "${HARNESS}" query "${common[@]}" --transport both --duration "${DURATION}" --output "${OUT}/query.json" +assert_json "${OUT}/query.json" + +echo "== mixed ==" +go run "${HARNESS}" mixed "${common[@]}" --duration "${DURATION}" --output "${OUT}/mixed.json" +assert_json "${OUT}/mixed.json" + +echo "smoke run passed" diff --git a/test/performance/environments/kind/postgres-patch.yaml b/test/performance/environments/kind/postgres-patch.yaml new file mode 100644 index 000000000..0722539a3 --- /dev/null +++ b/test/performance/environments/kind/postgres-patch.yaml @@ -0,0 +1,43 @@ +# Copyright 2026 The Tekton Authors +# +# Licensed 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. +# +# Strategic-merge patch: pin Postgres resources for reproducibility and mount the +# fixed postgresql.conf tuning into bitnami's conf.d. +apiVersion: apps/v1 +kind: StatefulSet +metadata: + name: tekton-results-postgres +spec: + template: + spec: + containers: + - name: postgres + resources: + requests: + cpu: "1" + memory: 1Gi + limits: + cpu: "2" + memory: 2Gi + volumeMounts: + - name: postgres-tuning + mountPath: /opt/bitnami/postgresql/conf/conf.d/override.conf + subPath: override.conf + volumes: + - name: postgres-tuning + configMap: + name: postgres-tuning + items: + - key: postgresql.conf + path: override.conf diff --git a/test/performance/environments/kind/postgresql.conf b/test/performance/environments/kind/postgresql.conf new file mode 100644 index 000000000..610eaea64 --- /dev/null +++ b/test/performance/environments/kind/postgresql.conf @@ -0,0 +1,33 @@ +# Fixed PostgreSQL tuning for reproducible tier-1 benchmarks. +# +# This file is the single source of truth for database configuration during a +# benchmark run. It is loaded via conf.d/override.conf (bitnami's postgresql.conf +# ends with `include_dir 'conf.d'`), so these settings override the image +# defaults without clobbering bitnami's generated pg_hba.conf / postgresql.conf. +# +# Keep these values version-controlled and stable: changing them changes results, +# so a change should be treated like a dataset-version bump and noted in the +# report/baseline metadata. + +# Memory +shared_buffers = 512MB +work_mem = 16MB +maintenance_work_mem = 128MB +effective_cache_size = 1536MB + +# Write-ahead log — sized for sustained ingest without excessive checkpointing. +wal_buffers = 16MB +max_wal_size = 2GB +min_wal_size = 512MB +checkpoint_completion_target = 0.9 + +# Planner — assume SSD-backed storage on CI/kind nodes. +random_page_cost = 1.1 +effective_io_concurrency = 200 + +# Connections — match the harness concurrency ceiling with headroom. +max_connections = 100 + +# Deterministic autovacuum so background work does not skew back-to-back runs. +autovacuum = on +autovacuum_naptime = 10s diff --git a/test/performance/generator/definition.go b/test/performance/generator/definition.go new file mode 100644 index 000000000..69cc280be --- /dev/null +++ b/test/performance/generator/definition.go @@ -0,0 +1,211 @@ +/* +Copyright 2026 The Tekton Authors + +Licensed 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. +*/ + +package generator + +import ( + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "io" + "sort" + "time" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +// DatasetDefinition is the machine-readable "golden" description of a generated +// dataset: what should exist after loading it through the API. It is the +// contract the compare tool (Story 02) and compliance suite (Story 03) build on +// — golden answers such as "namespace ns-07 has exactly 412 PipelineRuns, 98 +// with component=frontend" are read directly from these aggregates. +type DatasetDefinition struct { + Version string `json:"version"` + Seed int64 `json:"seed"` + Count int `json:"count"` + ContentHash string `json:"content_hash"` + GeneratedAt string `json:"generated_at"` + UIDRange UIDPartition `json:"uid_range"` + + Outcomes map[string]int `json:"outcomes"` + PerNamespace map[string]NamespaceCounts `json:"per_namespace"` + PerLabel map[string]map[string]int `json:"per_label"` + Instances []InstanceRecord `json:"instances"` +} + +// NamespaceCounts aggregates the records expected under a single namespace. +type NamespaceCounts struct { + PipelineRuns int `json:"pipelineruns"` + TaskRuns int `json:"taskruns"` + ByLabel map[string]map[string]int `json:"by_label"` +} + +// InstanceRecord is the expected final state of one generated PipelineRun and +// its children, keyed by UID so it joins 1:1 with the store driver's send log. +type InstanceRecord struct { + UID string `json:"uid"` + Namespace string `json:"namespace"` + TemplateID string `json:"template_id"` + ResultName string `json:"result_name"` + RecordName string `json:"record_name"` + ChildRecords []string `json:"child_records"` + Outcome string `json:"outcome"` + Labels map[string]string `json:"labels"` + StartTime string `json:"start_time"` + EndTime string `json:"end_time"` +} + +// Definition materializes the full dataset and returns its golden definition, +// including a content hash over the canonical serialization of every instance. +func (g *Generator) Definition() (*DatasetDefinition, error) { + d := &DatasetDefinition{ + Version: g.cfg.Version, + Seed: g.cfg.Seed, + Count: g.cfg.Count, + GeneratedAt: time.Now().UTC().Format(time.RFC3339), + UIDRange: g.cfg.UIDs, + Outcomes: map[string]int{}, + PerNamespace: map[string]NamespaceCounts{}, + PerLabel: map[string]map[string]int{}, + Instances: make([]InstanceRecord, 0, g.cfg.Count), + } + + hasher := sha256.New() + for i, inst := range g.Stream() { + if err := hashInstance(hasher, inst); err != nil { + return nil, fmt.Errorf("hashing instance %d: %w", i, err) + } + d.addInstance(inst) + } + d.ContentHash = hex.EncodeToString(hasher.Sum(nil)) + return d, nil +} + +// addInstance folds one instance into the definition aggregates. +func (d *DatasetDefinition) addInstance(inst *Instance) { + d.Outcomes[string(inst.Outcome)]++ + + nc := d.PerNamespace[inst.Namespace] + if nc.ByLabel == nil { + nc.ByLabel = map[string]map[string]int{} + } + nc.PipelineRuns++ + nc.TaskRuns += len(inst.TaskRuns) + for k, v := range inst.Labels { + if nc.ByLabel[k] == nil { + nc.ByLabel[k] = map[string]int{} + } + nc.ByLabel[k][v]++ + if d.PerLabel[k] == nil { + d.PerLabel[k] = map[string]int{} + } + d.PerLabel[k][v]++ + } + d.PerNamespace[inst.Namespace] = nc + + d.Instances = append(d.Instances, InstanceRecord{ + UID: inst.UID, + Namespace: inst.Namespace, + TemplateID: inst.TemplateID, + ResultName: inst.ResultName, + RecordName: inst.RecordName, + ChildRecords: inst.ChildRecords, + Outcome: string(inst.Outcome), + Labels: inst.Labels, + StartTime: formatTime(inst.PipelineRun.Status.StartTime), + EndTime: formatTime(inst.PipelineRun.Status.CompletionTime), + }) +} + +// hashInstance writes a canonical representation of an instance into h. The +// marshaled PipelineRun/TaskRun bytes are included so any change to generated +// object content changes the hash (and therefore requires a version bump). +func hashInstance(h io.Writer, inst *Instance) error { + canonical := struct { + UID string `json:"uid"` + Namespace string `json:"namespace"` + Template string `json:"template"` + Outcome string `json:"outcome"` + ResultName string `json:"result_name"` + RecordName string `json:"record_name"` + ChildRecords []string `json:"child_records"` + Labels []string `json:"labels"` + }{ + UID: inst.UID, + Namespace: inst.Namespace, + Template: inst.TemplateID, + Outcome: string(inst.Outcome), + ResultName: inst.ResultName, + RecordName: inst.RecordName, + ChildRecords: inst.ChildRecords, + Labels: sortedLabelPairs(inst.Labels), + } + enc := json.NewEncoder(h) + if err := enc.Encode(canonical); err != nil { + return err + } + prBytes, err := json.Marshal(inst.PipelineRun) + if err != nil { + return err + } + if _, err := h.Write(prBytes); err != nil { + return err + } + for _, tr := range inst.TaskRuns { + trBytes, err := json.Marshal(tr) + if err != nil { + return err + } + if _, err := h.Write(trBytes); err != nil { + return err + } + } + return nil +} + +// sortedLabelPairs renders labels as a stable "k=v" slice. +func sortedLabelPairs(labels map[string]string) []string { + pairs := make([]string, 0, len(labels)) + for k, v := range labels { + pairs = append(pairs, k+"="+v) + } + sort.Strings(pairs) + return pairs +} + +func formatTime(t *metav1.Time) string { + if t == nil { + return "" + } + return t.Time.UTC().Format(time.RFC3339) +} + +// WriteDefinition writes d as indented JSON. +func WriteDefinition(w io.Writer, d *DatasetDefinition) error { + enc := json.NewEncoder(w) + enc.SetIndent("", " ") + return enc.Encode(d) +} + +// LoadDefinition reads a DatasetDefinition from JSON. +func LoadDefinition(r io.Reader) (*DatasetDefinition, error) { + d := &DatasetDefinition{} + if err := json.NewDecoder(r).Decode(d); err != nil { + return nil, err + } + return d, nil +} diff --git a/test/performance/generator/distribution.go b/test/performance/generator/distribution.go new file mode 100644 index 000000000..01a8b18b3 --- /dev/null +++ b/test/performance/generator/distribution.go @@ -0,0 +1,157 @@ +/* +Copyright 2026 The Tekton Authors + +Licensed 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. +*/ + +package generator + +import ( + "fmt" + "math/rand/v2" + "sort" + "time" + + "github.com/google/uuid" +) + +// hotSelectionProbability is the chance that a label value is drawn from the +// "hot" head of the pool rather than the long tail, producing realistic label +// cardinality (a few dominant values, a long tail of rare ones). +const hotSelectionProbability = 0.8 + +// Outcome is the terminal state of a generated run. +type Outcome string + +// Terminal outcomes for generated runs. +const ( + OutcomeSucceeded Outcome = "succeeded" + OutcomeFailed Outcome = "failed" + OutcomeCancelled Outcome = "cancelled" +) + +// OutcomeRatios configures the distribution of terminal outcomes. The three +// fields should sum to 1.0. +type OutcomeRatios struct { + Succeeded float64 `json:"succeeded"` + Failed float64 `json:"failed"` + Cancelled float64 `json:"cancelled"` +} + +// DefaultOutcomeRatios returns the ~85/10/5 split described in the story. +func DefaultOutcomeRatios() OutcomeRatios { + return OutcomeRatios{Succeeded: 0.85, Failed: 0.10, Cancelled: 0.05} +} + +// pick returns a deterministic Outcome using rng. +func (r OutcomeRatios) pick(rng *rand.Rand) Outcome { + x := rng.Float64() + switch { + case x < r.Succeeded: + return OutcomeSucceeded + case x < r.Succeeded+r.Failed: + return OutcomeFailed + default: + return OutcomeCancelled + } +} + +// LabelValues describes the value pool for a single label key. Values[:HotCount] +// are the frequently-recurring "hot" values; the remainder is the long tail. +type LabelValues struct { + Values []string `json:"values"` + HotCount int `json:"hotCount"` +} + +// LabelPool maps label keys to their value distributions. The generator draws a +// value per key for each instance to create realistic label cardinality. +type LabelPool struct { + Keys map[string]LabelValues `json:"keys"` +} + +// sortedKeys returns the label keys in stable order so selection is +// reproducible regardless of Go's map iteration order. +func (p LabelPool) sortedKeys() []string { + keys := make([]string, 0, len(p.Keys)) + for k := range p.Keys { + keys = append(keys, k) + } + sort.Strings(keys) + return keys +} + +// draw returns a deterministic {key: value} selection for one instance. +func (p LabelPool) draw(rng *rand.Rand) map[string]string { + out := make(map[string]string, len(p.Keys)) + for _, k := range p.sortedKeys() { + lv := p.Keys[k] + if len(lv.Values) == 0 { + continue + } + out[k] = lv.pick(rng) + } + return out +} + +// pick selects a value biased towards the hot head of the pool. +func (lv LabelValues) pick(rng *rand.Rand) string { + hot := lv.HotCount + if hot <= 0 || hot > len(lv.Values) { + hot = len(lv.Values) + } + tailStart := hot + if tailStart >= len(lv.Values) || rng.Float64() < hotSelectionProbability { + return lv.Values[rng.IntN(hot)] + } + return lv.Values[tailStart+rng.IntN(len(lv.Values)-tailStart)] +} + +// TimeWindow is the fixed, absolute span over which run timestamps are spread. +// It must be set from configuration (never time.Now) so datasets are +// byte-reproducible. +type TimeWindow struct { + Start time.Time `json:"start"` + End time.Time `json:"end"` +} + +// at returns a deterministic instant within the window for the given fraction +// in [0,1). +func (w TimeWindow) at(fraction float64) time.Time { + span := w.End.Sub(w.Start) + if span <= 0 { + return w.Start.UTC() + } + offset := time.Duration(fraction * float64(span)) + return w.Start.Add(offset).UTC() +} + +// UIDPartition deterministically derives UUIDs from a disjoint counter range so +// the seed dataset and the live write stream never collide. +type UIDPartition struct { + // Space is the UUIDv5 namespace the derived UIDs are hashed under. + Space uuid.UUID `json:"space"` + // Offset is the first counter value in this partition. + Offset uint64 `json:"offset"` +} + +// uid derives the lowercase-hex UUID for element i of the partition. The result +// always satisfies the Results name regex ([a-z0-9_-]{1,63}). +func (p UIDPartition) uid(i int) string { + return uuid.NewSHA1(p.Space, fmt.Appendf(nil, "%d", p.Offset+uint64(i))).String() //nolint:gosec // i is a non-negative element index +} + +// childUID derives the UID for child j of element i, kept disjoint from parent +// UIDs by qualifying the counter with the child index. +func (p UIDPartition) childUID(i, j int) string { + return uuid.NewSHA1(p.Space, fmt.Appendf(nil, "%d-child-%d", p.Offset+uint64(i), j)).String() //nolint:gosec // i is a non-negative element index +} diff --git a/test/performance/generator/generator.go b/test/performance/generator/generator.go new file mode 100644 index 000000000..9c322b0aa --- /dev/null +++ b/test/performance/generator/generator.go @@ -0,0 +1,178 @@ +/* +Copyright 2026 The Tekton Authors + +Licensed 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. +*/ + +package generator + +import ( + "fmt" + "iter" + "math/rand/v2" + + tektonv1 "github.com/tektoncd/pipeline/pkg/apis/pipeline/v1" + record "github.com/tektoncd/results/pkg/api/server/v1alpha2/record" + result "github.com/tektoncd/results/pkg/api/server/v1alpha2/result" +) + +// Config controls deterministic dataset generation. The same Config with the +// same Seed produces byte-identical data on every run; any change that affects +// output must bump Version so results stay comparable only within a version. +type Config struct { + // Seed drives every random draw; fixed for reproducibility. + Seed int64 + // Version identifies the dataset content; bump on any generator/template change. + Version string + // Count is the number of top-level PipelineRuns to generate. + Count int + // Namespaces is the fixed pool instances are spread across. Each must match + // the Results name regex ([a-z0-9_-]{1,63}). + Namespaces []string + // LabelPool provides realistic label cardinality (hot values + long tail). + LabelPool LabelPool + // Outcomes is the terminal-state distribution (~85/10/5). + Outcomes OutcomeRatios + // TimeWindow is the fixed span run timestamps are spread across. + TimeWindow TimeWindow + // UIDs partitions the UUID space so seed and live streams never collide. + UIDs UIDPartition + // ChildTaskRuns clamps the per-PipelineRun child count (template may narrow it). + ChildTaskRuns IntRange +} + +// Generator materializes deterministic Instances from a Config and TemplateSet. +type Generator struct { + cfg Config + ts *TemplateSet +} + +// Instance is one fully-materialized top-level PipelineRun plus its child +// TaskRuns and the expected final state recorded for golden-answer computation. +type Instance struct { + Index int + UID string + Namespace string + TemplateID string + Outcome Outcome + PipelineRun *tektonv1.PipelineRun + TaskRuns []*tektonv1.TaskRun + ResultName string + RecordName string + ChildRecords []string + Labels map[string]string +} + +// New returns a Generator. If ts is nil the built-in template set is used. +func New(cfg Config, ts *TemplateSet) (*Generator, error) { + if cfg.Count < 0 { + return nil, fmt.Errorf("count must be non-negative, got %d", cfg.Count) + } + if len(cfg.Namespaces) == 0 { + return nil, fmt.Errorf("at least one namespace is required") + } + if ts == nil { + var err error + ts, err = DefaultTemplates() + if err != nil { + return nil, err + } + } + fmt.Println("Creating generator...") + return &Generator{cfg: cfg, ts: ts}, nil +} + +// Count returns the number of instances the generator produces. +func (g *Generator) Count() int { return g.cfg.Count } + +// Config returns a copy of the generator configuration. +func (g *Generator) Config() Config { return g.cfg } + +// rngFor returns an independent RNG for the given index so At(i) is reproducible +// regardless of iteration order. +func (g *Generator) rngFor(index int) *rand.Rand { + return rand.New(rand.NewPCG(uint64(g.cfg.Seed), uint64(index))) //nolint:gosec // deterministic by design +} + +// At returns the instance at the given index deterministically. +func (g *Generator) At(index int) (*Instance, error) { + if index < 0 || index >= g.cfg.Count { + return nil, fmt.Errorf("index %d out of range [0,%d)", index, g.cfg.Count) + } + rng := g.rngFor(index) + + tmpl := g.ts.pick(rng) + ns := g.cfg.Namespaces[rng.IntN(len(g.cfg.Namespaces))] + labels := g.cfg.LabelPool.draw(rng) + outcome := g.cfg.Outcomes.pick(rng) + start := g.cfg.TimeWindow.at(rng.Float64()) + + childRange := clampRange(tmpl.ChildTaskRuns, g.cfg.ChildTaskRuns) + childCount := childRange.pick(rng) + + uid := g.cfg.UIDs.uid(index) + resultName := result.FormatName(ns, uid) + recordName := record.FormatName(resultName, uid) + + inst := &Instance{ + Index: index, + UID: uid, + Namespace: ns, + TemplateID: tmpl.ID, + Outcome: outcome, + ResultName: resultName, + RecordName: recordName, + Labels: labels, + } + + inst.PipelineRun = g.buildPipelineRun(tmpl, inst, start) + inst.TaskRuns = g.buildTaskRuns(tmpl, inst, index, childCount, start) + for _, tr := range inst.TaskRuns { + childName := record.FormatName(resultName, string(tr.UID)) + inst.ChildRecords = append(inst.ChildRecords, childName) + } + + return inst, nil +} + +// Stream yields every instance deterministically in index order. +func (g *Generator) Stream() iter.Seq2[int, *Instance] { + return func(yield func(int, *Instance) bool) { + for i := 0; i < g.cfg.Count; i++ { + inst, err := g.At(i) + if err != nil { + return + } + if !yield(i, inst) { + return + } + } + } +} + +// clampRange narrows the template's child range to the global configured bounds. +func clampRange(tmpl, global IntRange) IntRange { + out := tmpl + if global.Max > 0 { + if out.Max == 0 || out.Max > global.Max { + out.Max = global.Max + } + if out.Min < global.Min { + out.Min = global.Min + } + } + if out.Max < out.Min { + out.Max = out.Min + } + return out +} diff --git a/test/performance/generator/generator_test.go b/test/performance/generator/generator_test.go new file mode 100644 index 000000000..9b95219c3 --- /dev/null +++ b/test/performance/generator/generator_test.go @@ -0,0 +1,235 @@ +/* +Copyright 2026 The Tekton Authors + +Licensed 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. +*/ + +package generator + +import ( + "encoding/json" + "fmt" + "testing" + "time" + + "github.com/google/uuid" + recordutil "github.com/tektoncd/results/pkg/api/server/v1alpha2/record" + resultutil "github.com/tektoncd/results/pkg/api/server/v1alpha2/result" +) + +var ( + seedSpace = uuid.MustParse("11111111-1111-1111-1111-111111111111") + liveSpace = uuid.MustParse("22222222-2222-2222-2222-222222222222") +) + +func testConfig(t *testing.T, count int) Config { + t.Helper() + namespaces := make([]string, 50) + for i := range namespaces { + namespaces[i] = fmt.Sprintf("ns-%02d", i) + } + return Config{ + Seed: 42, + Version: "test-v1", + Count: count, + Namespaces: namespaces, + LabelPool: LabelPool{Keys: map[string]LabelValues{ + "appstudio.openshift.io/component": { + Values: []string{"frontend", "backend", "api", "db", "cache", "worker", "ingest", "reporting"}, + HotCount: 2, + }, + "pipelinesascode.tekton.dev/event-type": { + Values: []string{"push", "pull_request"}, + HotCount: 1, + }, + }}, + Outcomes: DefaultOutcomeRatios(), + TimeWindow: TimeWindow{ + Start: time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC), + End: time.Date(2026, 1, 31, 0, 0, 0, 0, time.UTC), + }, + UIDs: UIDPartition{Space: seedSpace, Offset: 0}, + ChildTaskRuns: IntRange{Min: 2, Max: 15}, + } +} + +func mustGenerator(t *testing.T, cfg Config) *Generator { + t.Helper() + g, err := New(cfg, nil) + if err != nil { + t.Fatalf("New() = %v", err) + } + return g +} + +func marshalInstance(t *testing.T, inst *Instance) []byte { + t.Helper() + b, err := json.Marshal(struct { + PR interface{} + TR interface{} + }{inst.PipelineRun, inst.TaskRuns}) + if err != nil { + t.Fatalf("marshal: %v", err) + } + return b +} + +func TestGeneratorDeterministic(t *testing.T) { + cfg := testConfig(t, 200) + g1 := mustGenerator(t, cfg) + g2 := mustGenerator(t, cfg) + + for i := 0; i < cfg.Count; i++ { + a, err := g1.At(i) + if err != nil { + t.Fatalf("g1.At(%d) = %v", i, err) + } + b, err := g2.At(i) + if err != nil { + t.Fatalf("g2.At(%d) = %v", i, err) + } + if got, want := string(marshalInstance(t, a)), string(marshalInstance(t, b)); got != want { + t.Fatalf("instance %d not byte-identical across runs", i) + } + } + + d1, err := g1.Definition() + if err != nil { + t.Fatalf("Definition() = %v", err) + } + d2, err := g2.Definition() + if err != nil { + t.Fatalf("Definition() = %v", err) + } + if d1.ContentHash != d2.ContentHash { + t.Errorf("ContentHash mismatch: %s != %s", d1.ContentHash, d2.ContentHash) + } + if d1.ContentHash == "" { + t.Error("ContentHash is empty") + } +} + +func TestAtMatchesStream(t *testing.T) { + cfg := testConfig(t, 100) + g := mustGenerator(t, cfg) + for i, inst := range g.Stream() { + at, err := g.At(i) + if err != nil { + t.Fatalf("At(%d) = %v", i, err) + } + if got, want := string(marshalInstance(t, at)), string(marshalInstance(t, inst)); got != want { + t.Fatalf("At(%d) != Stream()[%d]", i, i) + } + } +} + +func TestNameSafety(t *testing.T) { + cfg := testConfig(t, 500) + g := mustGenerator(t, cfg) + for _, inst := range g.Stream() { + if _, _, err := resultutil.ParseName(inst.ResultName); err != nil { + t.Fatalf("invalid result name %q: %v", inst.ResultName, err) + } + if _, _, _, err := recordutil.ParseName(inst.RecordName); err != nil { + t.Fatalf("invalid record name %q: %v", inst.RecordName, err) + } + for _, cr := range inst.ChildRecords { + if _, _, _, err := recordutil.ParseName(cr); err != nil { + t.Fatalf("invalid child record name %q: %v", cr, err) + } + } + } +} + +func TestDistribution(t *testing.T) { + const n = 6000 + cfg := testConfig(t, n) + g := mustGenerator(t, cfg) + d, err := g.Definition() + if err != nil { + t.Fatalf("Definition() = %v", err) + } + + // Outcome ratios within tolerance of the configured 85/10/5 split. + assertRatio(t, "succeeded", d.Outcomes["succeeded"], n, 0.85, 0.03) + assertRatio(t, "failed", d.Outcomes["failed"], n, 0.10, 0.03) + assertRatio(t, "cancelled", d.Outcomes["cancelled"], n, 0.05, 0.03) + + // Every namespace in the pool is exercised. + if len(d.PerNamespace) != len(cfg.Namespaces) { + t.Errorf("namespaces used = %d, want %d", len(d.PerNamespace), len(cfg.Namespaces)) + } + + // Hot label values dominate the long tail. + comp := d.PerLabel["appstudio.openshift.io/component"] + hot := comp["frontend"] + comp["backend"] + tail := 0 + for v, c := range comp { + if v != "frontend" && v != "backend" { + tail += c + } + } + if hot <= tail { + t.Errorf("expected hot component values to dominate: hot=%d tail=%d", hot, tail) + } +} + +func TestUIDPartitionsDisjoint(t *testing.T) { + const n = 1000 + seedCfg := testConfig(t, n) + liveCfg := testConfig(t, n) + liveCfg.UIDs = UIDPartition{Space: liveSpace, Offset: 1 << 32} + + seen := map[string]bool{} + for _, inst := range mustGenerator(t, seedCfg).Stream() { + seen[inst.UID] = true + for _, cr := range inst.ChildRecords { + seen[cr] = true + } + } + for _, inst := range mustGenerator(t, liveCfg).Stream() { + if seen[inst.UID] { + t.Fatalf("live UID %q collides with seed range", inst.UID) + } + } +} + +func TestDefinitionRoundTrip(t *testing.T) { + cfg := testConfig(t, 50) + g := mustGenerator(t, cfg) + d, err := g.Definition() + if err != nil { + t.Fatalf("Definition() = %v", err) + } + + var buf []byte + buf, err = json.Marshal(d) + if err != nil { + t.Fatalf("marshal definition: %v", err) + } + got := &DatasetDefinition{} + if err := json.Unmarshal(buf, got); err != nil { + t.Fatalf("unmarshal definition: %v", err) + } + if got.ContentHash != d.ContentHash || got.Count != d.Count { + t.Errorf("round-trip mismatch: got %+v", got) + } +} + +func assertRatio(t *testing.T, name string, got, total int, want, tol float64) { + t.Helper() + ratio := float64(got) / float64(total) + if ratio < want-tol || ratio > want+tol { + t.Errorf("%s ratio = %.3f, want %.3f ± %.2f", name, ratio, want, tol) + } +} diff --git a/test/performance/generator/pipelinerun.go b/test/performance/generator/pipelinerun.go new file mode 100644 index 000000000..c99b03925 --- /dev/null +++ b/test/performance/generator/pipelinerun.go @@ -0,0 +1,146 @@ +/* +Copyright 2026 The Tekton Authors + +Licensed 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. +*/ + +package generator + +import ( + "fmt" + "time" + + tektonv1 "github.com/tektoncd/pipeline/pkg/apis/pipeline/v1" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + "knative.dev/pkg/apis" + duckv1 "knative.dev/pkg/apis/duck/v1" +) + +const ( + pipelineRunDuration = 12 * time.Minute + taskRunDuration = 8 * time.Minute +) + +// buildPipelineRun deep-copies the template PipelineRun and overwrites the +// identity, label, timestamp and outcome fields for this instance. +func (g *Generator) buildPipelineRun(tmpl *Template, inst *Instance, start time.Time) *tektonv1.PipelineRun { + pr := tmpl.pipelineRun.DeepCopy() + pr.TypeMeta = metav1.TypeMeta{APIVersion: "tekton.dev/v1", Kind: "PipelineRun"} + pr.Name = fmt.Sprintf("pr-%s", inst.UID) + pr.Namespace = inst.Namespace + pr.UID = types.UID(inst.UID) + pr.CreationTimestamp = metav1.NewTime(start) + applyLabels(&pr.ObjectMeta, inst.Labels) + + end := start.Add(pipelineRunDuration) + pr.Status.StartTime = &metav1.Time{Time: start} + pr.Status.CompletionTime = &metav1.Time{Time: end} + pr.Status.Conditions = terminalCondition(inst.Outcome, end) + return pr +} + +// buildTaskRuns synthesizes the child TaskRun records for an instance and +// rewrites the PipelineRun's childReferences to match them. +func (g *Generator) buildTaskRuns(tmpl *Template, inst *Instance, index, count int, start time.Time) []*tektonv1.TaskRun { + skel := tmpl.taskRun + if skel == nil { + skel = defaultTaskRunSkeleton() + } + + trs := make([]*tektonv1.TaskRun, 0, count) + refs := make([]tektonv1.ChildStatusReference, 0, count) + for j := 0; j < count; j++ { + childUID := g.cfg.UIDs.childUID(index, j) + tr := skel.DeepCopy() + tr.TypeMeta = metav1.TypeMeta{APIVersion: "tekton.dev/v1", Kind: "TaskRun"} + name := fmt.Sprintf("tr-%s", childUID) + tr.Name = name + tr.Namespace = inst.Namespace + tr.UID = types.UID(childUID) + tr.CreationTimestamp = metav1.NewTime(start) + applyLabels(&tr.ObjectMeta, inst.Labels) + tr.OwnerReferences = []metav1.OwnerReference{{ + APIVersion: "tekton.dev/v1", + Kind: "PipelineRun", + Name: inst.PipelineRun.Name, + UID: types.UID(inst.UID), + }} + + trStart := start.Add(time.Duration(j) * time.Minute) + trEnd := trStart.Add(taskRunDuration) + tr.Status.StartTime = &metav1.Time{Time: trStart} + tr.Status.CompletionTime = &metav1.Time{Time: trEnd} + // Only the last child inherits a non-successful outcome, mirroring how a + // failed/cancelled PipelineRun typically has one culprit TaskRun. + childOutcome := OutcomeSucceeded + if j == count-1 { + childOutcome = inst.Outcome + } + tr.Status.Conditions = terminalCondition(childOutcome, trEnd) + + trs = append(trs, tr) + refs = append(refs, tektonv1.ChildStatusReference{ + TypeMeta: runtime.TypeMeta{APIVersion: "tekton.dev/v1", Kind: "TaskRun"}, + Name: name, + PipelineTaskName: fmt.Sprintf("task-%d", j), + }) + } + inst.PipelineRun.Status.ChildReferences = refs + return trs +} + +// applyLabels overlays the drawn label subset onto the object's existing labels. +func applyLabels(meta *metav1.ObjectMeta, labels map[string]string) { + if meta.Labels == nil { + meta.Labels = map[string]string{} + } + for k, v := range labels { + meta.Labels[k] = v + } +} + +// terminalCondition returns the Succeeded condition for a terminal outcome. +func terminalCondition(o Outcome, at time.Time) duckv1.Conditions { + c := apis.Condition{ + Type: apis.ConditionSucceeded, + LastTransitionTime: apis.VolatileTime{Inner: metav1.NewTime(at)}, + } + switch o { + case OutcomeSucceeded: + c.Status = corev1.ConditionTrue + c.Reason = "Succeeded" + c.Message = "Tasks Completed successfully" + case OutcomeFailed: + c.Status = corev1.ConditionFalse + c.Reason = "Failed" + c.Message = "Tasks Completed with failures" + case OutcomeCancelled: + c.Status = corev1.ConditionFalse + c.Reason = "Cancelled" + c.Message = "PipelineRun was cancelled" + } + return duckv1.Conditions{c} +} + +// defaultTaskRunSkeleton returns a minimal TaskRun used when a template omits +// taskrun.yaml. +func defaultTaskRunSkeleton() *tektonv1.TaskRun { + return &tektonv1.TaskRun{ + Spec: tektonv1.TaskRunSpec{ + TaskRef: &tektonv1.TaskRef{Name: "generic-task"}, + }, + } +} diff --git a/test/performance/generator/template.go b/test/performance/generator/template.go new file mode 100644 index 000000000..c958e7933 --- /dev/null +++ b/test/performance/generator/template.go @@ -0,0 +1,201 @@ +/* +Copyright 2026 The Tekton Authors + +Licensed 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. +*/ + +// Package generator produces deterministic, realistic PipelineRun and TaskRun +// objects for the Tekton Results performance benchmark framework. The same +// generator builds both the versioned seed dataset and the live write stream, +// so benchmark runs are comparable across milestones. +package generator + +import ( + "embed" + "fmt" + "io/fs" + "math/rand/v2" + "path" + "sort" + + tektonv1 "github.com/tektoncd/pipeline/pkg/apis/pipeline/v1" + "sigs.k8s.io/yaml" +) + +// embeddedTemplates carries the built-in template set so the generator produces +// deterministic data without depending on files on disk. Real anonymized +// manifests are dropped into the templates/ directory following the contract +// documented in templates/README.md. +// +//go:embed all:templates +var embeddedTemplates embed.FS + +// Template is a parameterized PipelineRun+TaskRun blueprint. Identity and +// distribution fields (name, namespace, UID, a subset of labels, timestamps and +// outcome) are overwritten per instance by the Generator; the rest of the +// object is kept verbatim so the JSONB size distribution stays realistic. +type Template struct { + // ID is a stable identifier that participates in the dataset version hash. + ID string + // Weight is the relative probability of selecting this template. + Weight float64 + // ChildTaskRuns bounds how many child TaskRun records are generated per + // PipelineRun instantiated from this template. + ChildTaskRuns IntRange + + pipelineRun *tektonv1.PipelineRun + taskRun *tektonv1.TaskRun +} + +// IntRange is an inclusive integer range. +type IntRange struct { + Min int `json:"min"` + Max int `json:"max"` +} + +// pick returns a deterministic value in [Min, Max] using rng. +func (r IntRange) pick(rng *rand.Rand) int { + if r.Max <= r.Min { + return r.Min + } + return r.Min + rng.IntN(r.Max-r.Min+1) +} + +// templateMeta is the optional template.yaml sidecar describing a template. +type templateMeta struct { + ID string `json:"id"` + Weight float64 `json:"weight"` + ChildTaskRuns IntRange `json:"childTaskRuns"` + Description string `json:"description"` +} + +// TemplateSet is a loaded, weight-normalized collection of templates. +type TemplateSet struct { + Templates []*Template + totalWeight float64 +} + +// IDs returns the template identifiers in stable (sorted) order. +func (s *TemplateSet) IDs() []string { + ids := make([]string, 0, len(s.Templates)) + for _, t := range s.Templates { + ids = append(ids, t.ID) + } + sort.Strings(ids) + return ids +} + +// pick selects a template deterministically by weight using rng. +func (s *TemplateSet) pick(rng *rand.Rand) *Template { + if len(s.Templates) == 1 { + return s.Templates[0] + } + target := rng.Float64() * s.totalWeight + var cum float64 + for _, t := range s.Templates { + cum += t.Weight + if target < cum { + return t + } + } + return s.Templates[len(s.Templates)-1] +} + +// DefaultTemplates loads the template set bundled into the binary. +func DefaultTemplates() (*TemplateSet, error) { + return LoadTemplates(embeddedTemplates, "templates") +} + +// LoadTemplates reads every template directory under root in fsys. Each template +// lives in its own subdirectory containing at minimum a pipelinerun.yaml; an +// optional taskrun.yaml provides the child TaskRun skeleton and an optional +// template.yaml carries metadata (id, weight, child count range). +func LoadTemplates(fsys fs.FS, root string) (*TemplateSet, error) { + entries, err := fs.ReadDir(fsys, root) + if err != nil { + return nil, fmt.Errorf("reading template root %q: %w", root, err) + } + + set := &TemplateSet{} + for _, e := range entries { + if !e.IsDir() { + continue + } + dir := path.Join(root, e.Name()) + fmt.Println(fmt.Sprintf("Loading template %q from %s...", e.Name(), dir)) + t, err := loadTemplate(fsys, dir, e.Name()) + if err != nil { + return nil, err + } + if t == nil { + continue + } + set.Templates = append(set.Templates, t) + set.totalWeight += t.Weight + } + + if len(set.Templates) == 0 { + return nil, fmt.Errorf("no templates found under %q (each template needs its own directory with a pipelinerun.yaml)", root) + } + // Stable ordering keeps template selection reproducible across runs. + sort.Slice(set.Templates, func(i, j int) bool { return set.Templates[i].ID < set.Templates[j].ID }) + return set, nil +} + +// loadTemplate loads a single template directory. It returns (nil, nil) when the +// directory does not contain a pipelinerun.yaml so unrelated directories are +// skipped silently. +func loadTemplate(fsys fs.FS, dir, name string) (*Template, error) { + prBytes, err := fs.ReadFile(fsys, path.Join(dir, "pipelinerun.yaml")) + if err != nil { + return nil, nil //nolint:nilerr // directory without a PipelineRun is not a template + } + + pr := &tektonv1.PipelineRun{} + if err := yaml.UnmarshalStrict(prBytes, pr); err != nil { + return nil, fmt.Errorf("parsing %s/pipelinerun.yaml: %w", dir, err) + } + + t := &Template{ + ID: name, + Weight: 1, + ChildTaskRuns: IntRange{Min: 2, Max: 15}, + pipelineRun: pr, + } + + if trBytes, err := fs.ReadFile(fsys, path.Join(dir, "taskrun.yaml")); err == nil { + tr := &tektonv1.TaskRun{} + if err := yaml.UnmarshalStrict(trBytes, tr); err != nil { + return nil, fmt.Errorf("parsing %s/taskrun.yaml: %w", dir, err) + } + t.taskRun = tr + } + + if metaBytes, err := fs.ReadFile(fsys, path.Join(dir, "template.yaml")); err == nil { + meta := templateMeta{} + if err := yaml.UnmarshalStrict(metaBytes, &meta); err != nil { + return nil, fmt.Errorf("parsing %s/template.yaml: %w", dir, err) + } + if meta.ID != "" { + t.ID = meta.ID + } + if meta.Weight > 0 { + t.Weight = meta.Weight + } + if meta.ChildTaskRuns.Max > 0 { + t.ChildTaskRuns = meta.ChildTaskRuns + } + } + + return t, nil +} diff --git a/test/performance/generator/templates/README.md b/test/performance/generator/templates/README.md new file mode 100644 index 000000000..da6ec562f --- /dev/null +++ b/test/performance/generator/templates/README.md @@ -0,0 +1,57 @@ +# Generator templates + +The benchmark generator instantiates realistic PipelineRun/TaskRun objects from +the templates in this directory. Templates are **anonymized real manifests** +captured from an actual CI deployment — strip secrets, credentials, tokens and +internal hostnames, but keep the structural shape: full `status`, conditions, +results, params, and the real label/annotation sets +(`tekton.dev/pipeline`, `pipelinesascode.tekton.dev/*`, `appstudio.openshift.io/*`, +etc.). + +## Contract + +Each template is a **directory** under `templates/`: + +``` +templates/ + / + pipelinerun.yaml # required: a tektonv1.PipelineRun manifest + taskrun.yaml # optional: a tektonv1.TaskRun skeleton for child records + template.yaml # optional: metadata (see below) +``` + +- `pipelinerun.yaml` — a valid `tekton.dev/v1` PipelineRun. The generator + **overwrites** the identity/distribution fields on every instance: + `metadata.name`, `metadata.namespace`, `metadata.uid`, + `metadata.creationTimestamp`, a configured subset of `metadata.labels`, + `status.startTime`, `status.completionTime`, `status.conditions[Succeeded]`, + and `status.childReferences`. Everything else (spec, params, results, the + remaining labels/annotations) is kept verbatim, so the object's serialized + size is determined by the template — pick templates that cover the real size + distribution (~5 KB small runs up to 50+ KB large runs, median 15–20 KB). +- `taskrun.yaml` — a TaskRun skeleton used to synthesize the child records. The + generator overwrites the same identity fields plus the owner reference back to + the parent PipelineRun. If omitted, a minimal built-in TaskRun is used. +- `template.yaml` — optional metadata: + + ```yaml + id: konflux-build # defaults to the directory name + weight: 3.0 # relative selection probability (default 1.0) + childTaskRuns: # per-template child count range (default 2..15) + min: 8 + max: 15 + description: "Konflux-style container build with 8-15 tasks" + ``` + +## Determinism and versioning + +Template content is part of the dataset version. **Changing any template changes +the generated data**, so bump `Config.Version` (and regenerate the golden +`datasets/*.md` hash) whenever templates are added or edited. Results are only +comparable within the same dataset version. + +## Provided sample + +`sample/` is a structurally-representative placeholder so the framework builds +and tests run before the real anonymized manifests land. Replace it with 5–10 +real templates before capturing a baseline. diff --git a/test/performance/generator/templates/sample/pipelinerun.yaml b/test/performance/generator/templates/sample/pipelinerun.yaml new file mode 100644 index 000000000..779dc3b55 --- /dev/null +++ b/test/performance/generator/templates/sample/pipelinerun.yaml @@ -0,0 +1,92 @@ +apiVersion: tekton.dev/v1 +kind: PipelineRun +metadata: + name: sample-build + namespace: default + labels: + tekton.dev/pipeline: docker-build + pipelinesascode.tekton.dev/event-type: push + pipelinesascode.tekton.dev/branch: main + appstudio.openshift.io/application: sample-app + appstudio.openshift.io/component: sample-component + annotations: + pipelinesascode.tekton.dev/repo-url: https://example.test/org/repo + results.tekton.dev/resultAnnotations: '{"repo":"sample"}' + results.tekton.dev/recordSummaryAnnotations: '{"component":"sample-component"}' +spec: + params: + - name: git-url + value: https://example.test/org/repo + - name: revision + value: 0000000000000000000000000000000000000000 + - name: output-image + value: quay.example.test/org/sample-component:latest + pipelineRef: + resolver: bundles + params: + - name: name + value: docker-build + - name: bundle + value: quay.example.test/konflux-ci/tekton-catalog/pipeline-docker-build:latest + - name: kind + value: pipeline + taskRunTemplate: + serviceAccountName: appstudio-pipeline + timeouts: + pipeline: 1h0m0s +status: + startTime: "2026-01-01T00:00:00Z" + completionTime: "2026-01-01T00:12:00Z" + conditions: + - type: Succeeded + status: "True" + reason: Succeeded + message: "Tasks Completed: 4 (Failed: 0, Cancelled 0), Skipped: 0" + lastTransitionTime: "2026-01-01T00:12:00Z" + pipelineSpec: + params: + - name: git-url + type: string + - name: revision + type: string + - name: output-image + type: string + tasks: + - name: clone-repository + taskRef: + resolver: bundles + - name: build-container + taskRef: + resolver: bundles + - name: run-tests + taskRef: + resolver: bundles + - name: push-image + taskRef: + resolver: bundles + results: + - name: IMAGE_URL + value: quay.example.test/org/sample-component:latest + - name: IMAGE_DIGEST + value: sha256:0000000000000000000000000000000000000000000000000000000000000000 + - name: CHAINS-GIT_URL + value: https://example.test/org/repo + - name: CHAINS-GIT_COMMIT + value: "0000000000000000000000000000000000000000" + childReferences: + - apiVersion: tekton.dev/v1 + kind: TaskRun + name: sample-build-clone-repository + pipelineTaskName: clone-repository + - apiVersion: tekton.dev/v1 + kind: TaskRun + name: sample-build-build-container + pipelineTaskName: build-container + - apiVersion: tekton.dev/v1 + kind: TaskRun + name: sample-build-run-tests + pipelineTaskName: run-tests + - apiVersion: tekton.dev/v1 + kind: TaskRun + name: sample-build-push-image + pipelineTaskName: push-image diff --git a/test/performance/generator/templates/sample/taskrun.yaml b/test/performance/generator/templates/sample/taskrun.yaml new file mode 100644 index 000000000..310501b2a --- /dev/null +++ b/test/performance/generator/templates/sample/taskrun.yaml @@ -0,0 +1,47 @@ +apiVersion: tekton.dev/v1 +kind: TaskRun +metadata: + name: sample-build-build-container + namespace: default + labels: + tekton.dev/pipeline: docker-build + tekton.dev/pipelineTask: build-container + appstudio.openshift.io/application: sample-app + appstudio.openshift.io/component: sample-component +spec: + params: + - name: IMAGE + value: quay.example.test/org/sample-component:latest + taskRef: + resolver: bundles + params: + - name: name + value: buildah + - name: bundle + value: quay.example.test/konflux-ci/tekton-catalog/task-buildah:latest + - name: kind + value: task + serviceAccountName: appstudio-pipeline + timeout: 1h0m0s +status: + startTime: "2026-01-01T00:02:00Z" + completionTime: "2026-01-01T00:10:00Z" + conditions: + - type: Succeeded + status: "True" + reason: Succeeded + message: All Steps have completed executing + lastTransitionTime: "2026-01-01T00:10:00Z" + podName: sample-build-build-container-pod + steps: + - name: build + terminated: + exitCode: 0 + reason: Completed + startedAt: "2026-01-01T00:02:10Z" + finishedAt: "2026-01-01T00:09:50Z" + results: + - name: IMAGE_DIGEST + value: sha256:0000000000000000000000000000000000000000000000000000000000000000 + - name: IMAGE_URL + value: quay.example.test/org/sample-component:latest diff --git a/test/performance/generator/templates/sample/template.yaml b/test/performance/generator/templates/sample/template.yaml new file mode 100644 index 000000000..dd87d256b --- /dev/null +++ b/test/performance/generator/templates/sample/template.yaml @@ -0,0 +1,9 @@ +id: sample +weight: 1.0 +childTaskRuns: + min: 2 + max: 4 +description: >- + Structurally-representative placeholder modeled on a Konflux-style + docker-build PipelineRun. Replace with anonymized real manifests before + capturing a baseline. diff --git a/test/performance/harness/apiclient.go b/test/performance/harness/apiclient.go new file mode 100644 index 000000000..caef18d48 --- /dev/null +++ b/test/performance/harness/apiclient.go @@ -0,0 +1,89 @@ +/* +Copyright 2026 The Tekton Authors + +Licensed 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. +*/ + +package main + +import ( + "crypto/tls" + "fmt" + + "google.golang.org/grpc" + "google.golang.org/grpc/credentials" + "k8s.io/client-go/transport" + + "github.com/tektoncd/results/test/e2e/client" +) + +// APIClients bundles the transport clients the harness drives. The store driver +// uses gRPC; the query driver can use either transport. +type APIClients struct { + GRPC client.GRPCClient + REST client.RESTClient +} + +// ClientConfig locates and authenticates against the deployed API server. The +// TLS + bearer-token construction mirrors test/e2e (resultsClient) so the +// harness talks to the real API surface the same way conformance tests do. +type ClientConfig struct { + ServerAddress string // e.g. https://localhost:8080 + ServerName string // TLS server name (cert CN/SAN) + CertFile string // CA cert file; empty means skip verification + TokenFile string // bearer token file (service-account token) +} + +// NewAPIClients constructs the gRPC and REST clients from cfg. When CertFile is +// empty, TLS verification is skipped (convenient for a local kind cluster with a +// self-signed cert). +func NewAPIClients(cfg ClientConfig) (*APIClients, error) { + var ( + tlsConfig transport.TLSConfig + creds credentials.TransportCredentials + ) + if cfg.CertFile != "" { + tc, err := credentials.NewClientTLSFromFile(cfg.CertFile, cfg.ServerName) + if err != nil { + return nil, fmt.Errorf("loading TLS cert %q: %w", cfg.CertFile, err) + } + creds = tc + tlsConfig = transport.TLSConfig{CAFile: cfg.CertFile, ServerName: cfg.ServerName} + } else { + creds = credentials.NewTLS(&tls.Config{InsecureSkipVerify: true}) //nolint:gosec // opt-in for local clusters + tlsConfig = transport.TLSConfig{Insecure: true} + } + + callOptions := []grpc.CallOption{ + grpc.PerRPCCredentials(&client.CustomCredentials{ + TokenSource: transport.NewCachedFileTokenSource(cfg.TokenFile), + ImpersonationConfig: &transport.ImpersonationConfig{}, + }), + } + grpcOptions := []grpc.DialOption{ + grpc.WithDefaultCallOptions(callOptions...), + grpc.WithTransportCredentials(creds), + } + gc, err := client.NewGRPCClient(cfg.ServerAddress, grpcOptions...) + if err != nil { + return nil, fmt.Errorf("creating gRPC client: %w", err) + } + + restConfig := &transport.Config{TLS: tlsConfig, BearerTokenFile: cfg.TokenFile} + rc, err := client.NewRESTClient(cfg.ServerAddress, client.WithConfig(restConfig)) + if err != nil { + return nil, fmt.Errorf("creating REST client: %w", err) + } + + return &APIClients{GRPC: gc, REST: rc}, nil +} diff --git a/test/performance/harness/config.go b/test/performance/harness/config.go new file mode 100644 index 000000000..3fda9efa9 --- /dev/null +++ b/test/performance/harness/config.go @@ -0,0 +1,97 @@ +/* +Copyright 2026 The Tekton Authors + +Licensed 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. +*/ + +package main + +import ( + "os" + + "github.com/spf13/cobra" +) + +// RunConfig holds the flags shared by every benchmark subcommand. Endpoint and +// auth flags default from the same environment variables the e2e suite uses so a +// harness run drops into an existing kind setup with no extra configuration. +type RunConfig struct { + // Connection / auth. + ServerAddress string + ServerName string + CertFile string + TokenFile string + + // Workload shape. + Concurrency int + Count int + DurationSec int + + // Dataset (shared by generator, loader, and query golden answers). + Seed int64 + DatasetVersion string + Dataset string + Namespaces int + ChildMin int + ChildMax int + + // Reporting. + Tier string + OutputPath string + DBBackend string + ServerImage string +} + +// bindCommon registers the shared flags on cmd and wires their defaults from the +// environment. Subcommands add their own flags (transport, ratios) on top. +func bindCommon(cmd *cobra.Command, cfg *RunConfig) { + f := cmd.Flags() + f.StringVar(&cfg.ServerAddress, "server-addr", envOr("API_SERVER_ADDR", "https://localhost:8080"), "API server address (env API_SERVER_ADDR)") + f.StringVar(&cfg.ServerName, "server-name", envOr("API_SERVER_NAME", "tekton-results-api-service.tekton-pipelines.svc.cluster.local"), "TLS server name (env API_SERVER_NAME)") + f.StringVar(&cfg.CertFile, "cert", os.Getenv("SSL_CERT_PATH"), "CA certificate file; empty skips verification (env SSL_CERT_PATH)") + f.StringVar(&cfg.TokenFile, "token", os.Getenv("SA_TOKEN_PATH"), "bearer token file (env SA_TOKEN_PATH)") + + f.IntVar(&cfg.Concurrency, "concurrency", 8, "number of concurrent workers") + f.IntVar(&cfg.Count, "count", 1000, "number of top-level PipelineRuns for count-bounded runs") + f.IntVar(&cfg.DurationSec, "duration", 0, "run duration in seconds for time-bounded runs (0 = count-bounded)") + + f.Int64Var(&cfg.Seed, "seed", 42, "deterministic generator seed") + f.StringVar(&cfg.DatasetVersion, "dataset-version", "seed-small-v1", "dataset content version tag") + f.StringVar(&cfg.Dataset, "dataset", "seed-small", "named dataset tier") + f.IntVar(&cfg.Namespaces, "namespaces", 50, "number of namespaces to spread runs across") + f.IntVar(&cfg.ChildMin, "child-min", 2, "minimum child TaskRuns per PipelineRun") + f.IntVar(&cfg.ChildMax, "child-max", 15, "maximum child TaskRuns per PipelineRun") + + f.StringVar(&cfg.Tier, "tier", "tier1", "environment tier label for the report") + f.StringVar(&cfg.OutputPath, "output", "", "report output path; empty writes to stdout") + f.StringVar(&cfg.DBBackend, "db-backend", envOr("BENCH_DB_BACKEND", "local"), "database backend label: local|external (env BENCH_DB_BACKEND)") + f.StringVar(&cfg.ServerImage, "server-image", os.Getenv("BENCH_SERVER_IMAGE"), "server image tag for the report (env BENCH_SERVER_IMAGE)") +} + +// clientConfig extracts the connection settings. +func (c *RunConfig) clientConfig() ClientConfig { + return ClientConfig{ + ServerAddress: c.ServerAddress, + ServerName: c.ServerName, + CertFile: c.CertFile, + TokenFile: c.TokenFile, + } +} + +// envOr returns the environment variable value or a fallback. +func envOr(key, fallback string) string { + if v := os.Getenv(key); v != "" { + return v + } + return fallback +} diff --git a/test/performance/harness/dataset.go b/test/performance/harness/dataset.go new file mode 100644 index 000000000..a9511ec8f --- /dev/null +++ b/test/performance/harness/dataset.go @@ -0,0 +1,94 @@ +/* +Copyright 2026 The Tekton Authors + +Licensed 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. +*/ + +package main + +import ( + "fmt" + "time" + + "github.com/google/uuid" + + "github.com/tektoncd/results/test/performance/generator" +) + +// The seed and live streams draw UIDs from disjoint UUIDv5 spaces and disjoint +// counter offsets so a mixed run's writes can never collide with the seed data +// the readers query. See generator.UIDPartition. +const liveUIDOffset = 1 << 32 + +var ( + seedUIDSpace = uuid.MustParse("11111111-1111-1111-1111-111111111111") + liveUIDSpace = uuid.MustParse("22222222-2222-2222-2222-222222222222") + + // datasetWindow is a fixed absolute span; timestamps never use time.Now so + // generated content stays byte-stable across runs and machines. + datasetWindow = generator.TimeWindow{ + Start: time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC), + End: time.Date(2026, 1, 31, 0, 0, 0, 0, time.UTC), + } +) + +// defaultLabelPool models realistic Konflux label cardinality: a couple of hot +// values plus a long tail per key. Kept in one place so seed, live, and query +// golden answers agree. +func defaultLabelPool() generator.LabelPool { + return generator.LabelPool{Keys: map[string]generator.LabelValues{ + "appstudio.openshift.io/component": { + Values: []string{"frontend", "backend", "api", "db", "cache", "worker", "ingest", "reporting", "gateway", "auth"}, + HotCount: 2, + }, + "pipelinesascode.tekton.dev/event-type": { + Values: []string{"push", "pull_request", "retest", "incoming"}, + HotCount: 1, + }, + }} +} + +// buildNamespaces returns the fixed namespace pool ns-00..ns-NN. +func buildNamespaces(n int) []string { + ns := make([]string, n) + for i := range ns { + ns[i] = fmt.Sprintf("ns-%02d", i) + } + return ns +} + +// datasetConfig builds the canonical generator config. When live is true the +// UIDs come from the live partition (used by mixed writers); otherwise from the +// seed partition (used by the loader and read golden answers). +func datasetConfig(cfg *RunConfig, live bool) generator.Config { + uids := generator.UIDPartition{Space: seedUIDSpace, Offset: 0} + if live { + uids = generator.UIDPartition{Space: liveUIDSpace, Offset: liveUIDOffset} + } + return generator.Config{ + Seed: cfg.Seed, + Version: cfg.DatasetVersion, + Count: cfg.Count, + Namespaces: buildNamespaces(cfg.Namespaces), + LabelPool: defaultLabelPool(), + Outcomes: generator.DefaultOutcomeRatios(), + TimeWindow: datasetWindow, + UIDs: uids, + ChildTaskRuns: generator.IntRange{Min: cfg.ChildMin, Max: cfg.ChildMax}, + } +} + +// newGenerator builds a generator for the seed or live stream. +func newGenerator(cfg *RunConfig, live bool) (*generator.Generator, error) { + return generator.New(datasetConfig(cfg, live), nil) +} diff --git a/test/performance/harness/harness_test.go b/test/performance/harness/harness_test.go new file mode 100644 index 000000000..b01eda589 --- /dev/null +++ b/test/performance/harness/harness_test.go @@ -0,0 +1,146 @@ +/* +Copyright 2026 The Tekton Authors + +Licensed 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. +*/ + +package main + +import ( + "testing" + + pb "github.com/tektoncd/results/proto/v1alpha2/results_go_proto" + "github.com/tektoncd/results/test/performance/generator" +) + +func testRunConfig() *RunConfig { + return &RunConfig{ + Seed: 42, + DatasetVersion: "test-v1", + Count: 500, + Namespaces: 50, + ChildMin: 2, + ChildMax: 15, + } +} + +func TestStagedPipelineRuns(t *testing.T) { + g, err := newGenerator(testRunConfig(), false) + if err != nil { + t.Fatalf("newGenerator: %v", err) + } + inst, err := g.At(0) + if err != nil { + t.Fatalf("At(0): %v", err) + } + + for _, updates := range []int{1, 2, 3} { + stages := stagedPipelineRuns(inst.PipelineRun, updates) + if got, want := len(stages), updates+1; got != want { + t.Errorf("updates=%d: got %d stages, want %d", updates, got, want) + } + pending := stages[0] + if pending.Status.CompletionTime != nil { + t.Errorf("updates=%d: pending stage has completion time set", updates) + } + if len(pending.Status.Conditions) == 0 || pending.Status.Conditions[0].Reason != "Running" { + t.Errorf("updates=%d: pending stage is not Running", updates) + } + terminal := stages[len(stages)-1] + if terminal.Status.CompletionTime == nil { + t.Errorf("updates=%d: terminal stage missing completion time", updates) + } + } +} + +func TestUpdateCountForRange(t *testing.T) { + for i := 0; i < 1000; i++ { + if n := updateCountFor(i); n < 1 || n > 3 { + t.Fatalf("updateCountFor(%d) = %d, out of [1,3]", i, n) + } + } +} + +func TestSummaryStatus(t *testing.T) { + tests := []struct { + outcome generator.Outcome + want pb.RecordSummary_Status + }{ + {generator.OutcomeSucceeded, pb.RecordSummary_SUCCESS}, + {generator.OutcomeFailed, pb.RecordSummary_FAILURE}, + {generator.OutcomeCancelled, pb.RecordSummary_CANCELLED}, + {generator.Outcome("other"), pb.RecordSummary_UNKNOWN}, + } + for _, tt := range tests { + if got := summaryStatus(tt.outcome); got != tt.want { + t.Errorf("summaryStatus(%q) = %v, want %v", tt.outcome, got, tt.want) + } + } +} + +func TestSplitWorkers(t *testing.T) { + tests := []struct { + total, read, write int + wantWriters, wantRds int + }{ + {1, 3, 1, 1, 1}, // too few → one each + {8, 3, 1, 2, 6}, // 1/4 writers + {8, 1, 1, 4, 4}, // even + {4, 0, 0, 2, 2}, // no ratio → even split + {10, 9, 1, 1, 9}, // writers floored to at least 1 + {10, 1, 100, 9, 1}, // writers capped at total-1 + } + for _, tt := range tests { + w, r := splitWorkers(tt.total, tt.read, tt.write) + if w != tt.wantWriters || r != tt.wantRds { + t.Errorf("splitWorkers(%d,%d,%d) = (%d,%d), want (%d,%d)", tt.total, tt.read, tt.write, w, r, tt.wantWriters, tt.wantRds) + } + if w+r != max(tt.total, 2) && tt.total >= 2 { + t.Errorf("splitWorkers(%d,...) sum = %d, want %d", tt.total, w+r, tt.total) + } + } +} + +func TestListerPicker(t *testing.T) { + clients := &APIClients{} + if _, err := listerPicker("bogus", clients); err == nil { + t.Error("listerPicker(bogus) = nil error, want error") + } + for _, transport := range []string{"grpc", "rest", "both"} { + if _, err := listerPicker(transport, clients); err != nil { + t.Errorf("listerPicker(%q) = %v", transport, err) + } + } +} + +func TestSeedLiveUIDsDisjoint(t *testing.T) { + cfg := testRunConfig() + seed, err := newGenerator(cfg, false) + if err != nil { + t.Fatalf("seed generator: %v", err) + } + live, err := newGenerator(cfg, true) + if err != nil { + t.Fatalf("live generator: %v", err) + } + + seen := map[string]bool{} + for _, inst := range seed.Stream() { + seen[inst.UID] = true + } + for _, inst := range live.Stream() { + if seen[inst.UID] { + t.Fatalf("live UID %q collides with seed range", inst.UID) + } + } +} diff --git a/test/performance/harness/loader.go b/test/performance/harness/loader.go new file mode 100644 index 000000000..03a21cf4e --- /dev/null +++ b/test/performance/harness/loader.go @@ -0,0 +1,107 @@ +/* +Copyright 2026 The Tekton Authors + +Licensed 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. +*/ + +package main + +import ( + "context" + "fmt" + + pb "github.com/tektoncd/results/proto/v1alpha2/results_go_proto" + "github.com/tektoncd/results/test/performance/generator" + "github.com/tektoncd/results/test/performance/metrics" +) + +// loadSeed writes the full seed dataset through the API and returns the metrics +// and the golden definition it should match. Loading reuses the store write path +// (create Result, create+finalize the top-level Record, create child Records) so +// the seed is produced through exactly the surface the benchmark measures. +func loadSeed(ctx context.Context, gc pb.ResultsClient, cfg *RunConfig) (*metrics.MetricSet, *generator.DatasetDefinition, error) { + g, err := newGenerator(cfg, false) + if err != nil { + return nil, nil, err + } + def, err := g.Definition() + if err != nil { + return nil, nil, fmt.Errorf("computing dataset definition: %w", err) + } + fmt.Println("Running workers...") + m := runIndexed(ctx, cfg.Concurrency, 0, cfg.Count, func(ctx context.Context, index int, m *metrics.MetricSet) { + inst, err := g.At(index) + if err != nil { + m.Observe("generate", 0, err) + return + } + // One update finalizes the pending record to its terminal state. + storeInstance(ctx, gc, inst, 1, m) + }) + return m, def, nil +} + +// verifySeed counts the PipelineRun and TaskRun records actually present and +// compares them to the golden definition. It returns an error describing any +// mismatch so the loader can fail loudly rather than benchmark against a partial +// dataset. +func verifySeed(ctx context.Context, l lister, def *generator.DatasetDefinition) error { + wantPR := def.Count + wantTR := 0 + for _, nc := range def.PerNamespace { + wantTR += nc.TaskRuns + } + + gotPR, err := countRecords(ctx, l, "-/results/-", "data_type == PIPELINE_RUN") + if err != nil { + return fmt.Errorf("counting PipelineRun records: %w", err) + } + gotTR, err := countRecords(ctx, l, "-/results/-", "data_type == TASK_RUN") + if err != nil { + return fmt.Errorf("counting TaskRun records: %w", err) + } + + if gotPR != wantPR || gotTR != wantTR { + return fmt.Errorf("row-count mismatch: PipelineRuns got %d want %d, TaskRuns got %d want %d", gotPR, wantPR, gotTR, wantTR) + } + return nil +} + +// countRecords walks every page of a ListRecords query and returns the total +// number of records matched. +func countRecords(ctx context.Context, l lister, parent, filter string) (int, error) { + const ( + pageSize = int32(1000) + maxPages = 100000 + ) + total := 0 + token := "" + for page := 0; page < maxPages; page++ { + resp, err := l.ListRecords(ctx, &pb.ListRecordsRequest{ + Parent: parent, + Filter: filter, + PageSize: pageSize, + PageToken: token, + }) + if err != nil { + return 0, err + } + total += len(resp.GetRecords()) + next := resp.GetNextPageToken() + if next == "" || next == token { + return total, nil + } + token = next + } + return total, fmt.Errorf("pagination did not terminate after %d pages", maxPages) +} diff --git a/test/performance/harness/main.go b/test/performance/harness/main.go new file mode 100644 index 000000000..a5cd2a962 --- /dev/null +++ b/test/performance/harness/main.go @@ -0,0 +1,327 @@ +/* +Copyright 2026 The Tekton Authors + +Licensed 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. +*/ + +// Command bench is the Tekton Results performance benchmark harness. It drives +// the real deployed API surface (gRPC for store, REST/gRPC for query) against a +// deterministic, versioned dataset and emits machine-readable JSON reports. +// +// Subcommands: +// +// bench load --verify load the seed dataset through the API and verify counts +// bench store replay the watcher write lifecycle over the live range +// bench query run the predefined read mix against the seed range +// bench mixed run writers (live) and readers (seed) in parallel +// bench dataset emit the golden DatasetDefinition (no cluster needed) +package main + +import ( + "context" + "fmt" + "os" + "os/signal" + "path/filepath" + "strings" + "syscall" + "time" + + "github.com/spf13/cobra" + + "github.com/tektoncd/results/test/performance/generator" + "github.com/tektoncd/results/test/performance/metrics" + "github.com/tektoncd/results/test/performance/report" +) + +const defaultReadDuration = 30 * time.Second + +func main() { + if err := run(); err != nil { + fmt.Fprintln(os.Stderr, "error:", err) + os.Exit(1) + } +} + +func run() error { + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + return newRootCmd().ExecuteContext(ctx) +} + +func newRootCmd() *cobra.Command { + root := &cobra.Command{ + Use: "bench", + Short: "Tekton Results performance benchmark harness", + SilenceUsage: true, + SilenceErrors: true, + } + root.AddCommand(newLoadCmd(), newStoreCmd(), newQueryCmd(), newMixedCmd(), newDatasetCmd()) + return root +} + +func newStoreCmd() *cobra.Command { + cfg := &RunConfig{} + cmd := &cobra.Command{ + Use: "store", + Short: "Benchmark write/ingest by replaying the watcher lifecycle over gRPC", + RunE: func(cmd *cobra.Command, _ []string) error { + ctx := cmd.Context() + clients, err := NewAPIClients(cfg.clientConfig()) + if err != nil { + return err + } + g, err := newGenerator(cfg, true) // store writes the live range + if err != nil { + return err + } + log := &sendLog{} + m := runIndexed(ctx, cfg.Concurrency, 0, cfg.Count, func(ctx context.Context, index int, m *metrics.MetricSet) { + inst, err := g.At(index) + if err != nil { + m.Observe("generate", 0, err) + return + } + log.add(storeInstance(ctx, clients.GRPC, inst, updateCountFor(index), m)) + }) + if err := writeSendLog(cfg.OutputPath, log); err != nil { + return err + } + return emitReport(ctx, cfg, "store", "grpc", reportExtra{}, m.Snapshot()) + }, + } + bindCommon(cmd, cfg) + return cmd +} + +func newQueryCmd() *cobra.Command { + cfg := &RunConfig{} + var ( + transport string + pageSize int32 + ) + cmd := &cobra.Command{ + Use: "query", + Short: "Benchmark read/list performance against the seed range", + RunE: func(cmd *cobra.Command, _ []string) error { + ctx := cmd.Context() + clients, err := NewAPIClients(cfg.clientConfig()) + if err != nil { + return err + } + pick, err := listerPicker(transport, clients) + if err != nil { + return err + } + g, err := newGenerator(cfg, false) + if err != nil { + return err + } + namespaces := g.Config().Namespaces + queries := defaultQueries() + + m := runTimed(ctx, cfg.Concurrency, readDuration(cfg), func(ctx context.Context, workerID, iter int, m *metrics.MetricSet) { + q := queries[iter%len(queries)] + ns := namespaces[(workerID+iter)%len(namespaces)] + runQuery(ctx, pick(iter), q, ns, pageSize, m) + }) + return emitReport(ctx, cfg, "query", transport, reportExtra{}, m.Snapshot()) + }, + } + bindCommon(cmd, cfg) + cmd.Flags().StringVar(&transport, "transport", "grpc", "query transport: grpc|rest|both") + cmd.Flags().Int32Var(&pageSize, "page-size", 1000, "list page size") + return cmd +} + +func newMixedCmd() *cobra.Command { + cfg := &RunConfig{} + var ( + readRatio int + writeRatio int + pageSize int32 + ) + cmd := &cobra.Command{ + Use: "mixed", + Short: "Benchmark parallel store (live range) and query (seed range)", + RunE: func(cmd *cobra.Command, _ []string) error { + ctx := cmd.Context() + clients, err := NewAPIClients(cfg.clientConfig()) + if err != nil { + return err + } + live, err := newGenerator(cfg, true) + if err != nil { + return err + } + seed, err := newGenerator(cfg, false) + if err != nil { + return err + } + m := runMixed(ctx, clients.GRPC, live, seed, mixedParams{ + Concurrency: cfg.Concurrency, + Duration: readDuration(cfg), + ReadRatio: readRatio, + WriteRatio: writeRatio, + PageSize: pageSize, + }) + return emitReport(ctx, cfg, "mixed", "grpc", reportExtra{readRatio: readRatio, writeRatio: writeRatio}, m.Snapshot()) + }, + } + bindCommon(cmd, cfg) + cmd.Flags().IntVar(&readRatio, "read-ratio", 3, "reader share of the worker budget") + cmd.Flags().IntVar(&writeRatio, "write-ratio", 1, "writer share of the worker budget") + cmd.Flags().Int32Var(&pageSize, "page-size", 1000, "list page size") + return cmd +} + +func newLoadCmd() *cobra.Command { + cfg := &RunConfig{} + var verify bool + cmd := &cobra.Command{ + Use: "load", + Short: "Load the seed dataset through the API and optionally verify row counts", + RunE: func(cmd *cobra.Command, _ []string) error { + ctx := cmd.Context() + clients, err := NewAPIClients(cfg.clientConfig()) + if err != nil { + return err + } + fmt.Println("Loading seed...") + m, def, err := loadSeed(ctx, clients.GRPC, cfg) + if err != nil { + return err + } + if verify { + fmt.Println("Verifying seed...") + if err := verifySeed(ctx, grpcLister{c: clients.GRPC}, def); err != nil { + return fmt.Errorf("seed verification failed: %w", err) + } + fmt.Fprintf(os.Stderr, "seed verified: %d PipelineRuns, dataset hash %s\n", def.Count, def.ContentHash) + } + return emitReport(ctx, cfg, "load", "grpc", reportExtra{datasetHash: def.ContentHash}, m.Snapshot()) + }, + } + bindCommon(cmd, cfg) + cmd.Flags().BoolVar(&verify, "verify", false, "verify record counts against the golden definition after loading") + return cmd +} + +func newDatasetCmd() *cobra.Command { + cfg := &RunConfig{} + cmd := &cobra.Command{ + Use: "dataset", + Short: "Emit the golden DatasetDefinition (no cluster required)", + RunE: func(_ *cobra.Command, _ []string) error { + g, err := newGenerator(cfg, false) + if err != nil { + return err + } + def, err := g.Definition() + if err != nil { + return err + } + return withOutput(cfg.OutputPath, func(w *os.File) error { + return generator.WriteDefinition(w, def) + }) + }, + } + bindCommon(cmd, cfg) + return cmd +} + +// reportExtra carries mode-specific report fields. +type reportExtra struct { + datasetHash string + readRatio int + writeRatio int +} + +// emitReport assembles and writes the JSON report for a completed run. +func emitReport(ctx context.Context, cfg *RunConfig, mode, transport string, extra reportExtra, snap metrics.Snapshot) error { + meta := report.CaptureMeta(ctx) + meta.DatasetVersion = cfg.DatasetVersion + meta.DatasetHash = extra.datasetHash + meta.Tier = cfg.Tier + meta.Mode = mode + meta.APIServerAddr = cfg.ServerAddress + meta.DBBackend = cfg.DBBackend + meta.ServerImage = cfg.ServerImage + + rc := report.Config{ + Count: cfg.Count, + Concurrency: cfg.Concurrency, + DurationSec: cfg.DurationSec, + Transport: transport, + ReadRatio: extra.readRatio, + WriteRatio: extra.writeRatio, + Dataset: cfg.Dataset, + Seed: cfg.Seed, + } + rep := report.New(meta, rc, snap) + return withOutput(cfg.OutputPath, func(w *os.File) error { + return report.Write(w, rep) + }) +} + +// listerPicker returns a function selecting the transport for a given iteration. +func listerPicker(transport string, clients *APIClients) (func(iter int) lister, error) { + g := grpcLister{c: clients.GRPC} + switch transport { + case "grpc": + return func(int) lister { return g }, nil + case "rest": + return func(int) lister { return clients.REST }, nil + case "both": + return func(iter int) lister { + if iter%2 == 0 { + return g + } + return clients.REST + }, nil + default: + return nil, fmt.Errorf("unknown transport %q (want grpc|rest|both)", transport) + } +} + +// readDuration returns the configured duration, defaulting when unset. +func readDuration(cfg *RunConfig) time.Duration { + if cfg.DurationSec <= 0 { + return defaultReadDuration + } + return time.Duration(cfg.DurationSec) * time.Second +} + +// writeSendLog writes the store driver's actual send log alongside the report so +// it can be joined against the golden DatasetDefinition on UID. It is skipped when +// the report goes to stdout (no path to derive a sidecar from). +func writeSendLog(reportPath string, log *sendLog) error { + if reportPath == "" { + return nil + } + path := strings.TrimSuffix(reportPath, filepath.Ext(reportPath)) + ".sendlog.json" + return withOutput(path, func(w *os.File) error { return log.write(w) }) +} + +// withOutput invokes fn with the report sink: a file when path is set, else stdout. +func withOutput(path string, fn func(*os.File) error) error { + if path == "" { + return fn(os.Stdout) + } + f, err := os.Create(path) //nolint:gosec // operator-provided report output path + if err != nil { + return err + } + defer func() { _ = f.Close() }() + return fn(f) +} diff --git a/test/performance/harness/mixed.go b/test/performance/harness/mixed.go new file mode 100644 index 000000000..35f69f8b6 --- /dev/null +++ b/test/performance/harness/mixed.go @@ -0,0 +1,132 @@ +/* +Copyright 2026 The Tekton Authors + +Licensed 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. +*/ + +package main + +import ( + "context" + "sync" + "sync/atomic" + "time" + + pb "github.com/tektoncd/results/proto/v1alpha2/results_go_proto" + "github.com/tektoncd/results/test/performance/generator" + "github.com/tektoncd/results/test/performance/metrics" +) + +// mixedParams tunes the read/write split of a mixed run. +type mixedParams struct { + Concurrency int + Duration time.Duration + ReadRatio int + WriteRatio int + PageSize int32 +} + +// runMixed drives writers over the live UID range and readers over the seed range +// in parallel. The ranges are disjoint by construction (see dataset.go), so reads +// return stable golden answers uncontaminated by concurrent writes, and writers +// never contend on the same Result. Writers pull distinct live indices via an +// atomic counter (each index, and thus each Result, handled once). +func runMixed(ctx context.Context, gc pb.ResultsClient, live, seed *generator.Generator, p mixedParams) *metrics.MetricSet { + writers, readers := splitWorkers(p.Concurrency, p.ReadRatio, p.WriteRatio) + + runCtx, cancel := context.WithTimeout(ctx, p.Duration) + defer cancel() + + total := metrics.NewMetricSet() + total.Start() + + sets := make([]*metrics.MetricSet, 0, writers+readers) + var mu sync.Mutex + addSet := func(s *metrics.MetricSet) { + mu.Lock() + sets = append(sets, s) + mu.Unlock() + } + + var wg sync.WaitGroup + + // Writers: dispatch distinct live indices until exhausted or time is up. + var nextIndex int64 = -1 + liveCount := live.Count() + for w := 0; w < writers; w++ { + wg.Add(1) + go func() { + defer wg.Done() + ws := metrics.NewMetricSet() + addSet(ws) + for runCtx.Err() == nil { + i := int(atomic.AddInt64(&nextIndex, 1)) + if i >= liveCount { + return + } + inst, err := live.At(i) + if err != nil { + ws.Observe("generate", 0, err) + continue + } + storeInstance(runCtx, gc, inst, updateCountFor(i), ws) + } + }() + } + + // Readers: loop the query mix over the seed namespaces until time is up. + queries := defaultQueries() + namespaces := seed.Config().Namespaces + l := grpcLister{c: gc} + for r := 0; r < readers; r++ { + wg.Add(1) + go func(readerID int) { + defer wg.Done() + rs := metrics.NewMetricSet() + addSet(rs) + for iter := 0; runCtx.Err() == nil; iter++ { + q := queries[iter%len(queries)] + ns := namespaces[(readerID+iter)%len(namespaces)] + runQuery(runCtx, l, q, ns, p.PageSize, rs) + } + }(r) + } + + wg.Wait() + total.Stop() + for _, s := range sets { + total.Merge(s) + } + return total +} + +// splitWorkers divides a worker budget between readers and writers by ratio, +// guaranteeing at least one of each when the budget allows. +func splitWorkers(total, readRatio, writeRatio int) (writers, readers int) { + if total < 2 { + return 1, 1 + } + sum := readRatio + writeRatio + if sum <= 0 { + writers = total / 2 + return writers, total - writers + } + writers = total * writeRatio / sum + if writers < 1 { + writers = 1 + } + if writers >= total { + writers = total - 1 + } + return writers, total - writers +} diff --git a/test/performance/harness/query.go b/test/performance/harness/query.go new file mode 100644 index 000000000..27f56320e --- /dev/null +++ b/test/performance/harness/query.go @@ -0,0 +1,125 @@ +/* +Copyright 2026 The Tekton Authors + +Licensed 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. +*/ + +package main + +import ( + "context" + "fmt" + + pb "github.com/tektoncd/results/proto/v1alpha2/results_go_proto" + "github.com/tektoncd/results/test/performance/metrics" +) + +// queryKind distinguishes the two list surfaces. +type queryKind int + +const ( + kindResults queryKind = iota + kindRecords +) + +// query is one predefined read against the seed range. Parent is formatted with +// a namespace so a run exercises the whole namespace pool. Filters use only the +// CEL fields the server exposes (see pkg/api/server/cel/env.go). +type query struct { + Name string + Kind queryKind + Op string // metric op name + Parent string // fmt template taking the namespace + Filter string +} + +// defaultQueries is the standard read mix: list-all, by-status, recent-window +// (Results) and by-type, by-label (Records). +func defaultQueries() []query { + return []query{ + {Name: "results-all", Kind: kindResults, Op: "list_results_all", Parent: "%s", Filter: ""}, + {Name: "results-success", Kind: kindResults, Op: "list_results_status", Parent: "%s", Filter: "summary.status == SUCCESS"}, + {Name: "results-recent", Kind: kindResults, Op: "list_results_recent", Parent: "%s", Filter: `summary.end_time > timestamp("2026-01-20T00:00:00Z")`}, + {Name: "records-pipelinerun", Kind: kindRecords, Op: "list_records_type", Parent: "%s/results/-", Filter: "data_type == PIPELINE_RUN"}, + {Name: "records-by-label", Kind: kindRecords, Op: "list_records_label", Parent: "%s/results/-", Filter: `data.metadata.labels["appstudio.openshift.io/component"] == "frontend"`}, + } +} + +// lister is the subset of the API used by the query driver, satisfied by the +// REST client directly and by the gRPC client via grpcLister. +type lister interface { + ListResults(ctx context.Context, in *pb.ListResultsRequest) (*pb.ListResultsResponse, error) + ListRecords(ctx context.Context, in *pb.ListRecordsRequest) (*pb.ListRecordsResponse, error) +} + +// grpcLister adapts the variadic gRPC client to the lister interface. +type grpcLister struct{ c pb.ResultsClient } + +func (g grpcLister) ListResults(ctx context.Context, in *pb.ListResultsRequest) (*pb.ListResultsResponse, error) { + return g.c.ListResults(ctx, in) +} + +func (g grpcLister) ListRecords(ctx context.Context, in *pb.ListRecordsRequest) (*pb.ListRecordsResponse, error) { + return g.c.ListRecords(ctx, in) +} + +// runQuery executes one query with full pagination, recording per-page latency +// and errors. maxPages guards against a server returning a non-terminating token. +func runQuery(ctx context.Context, l lister, q query, namespace string, pageSize int32, m *metrics.MetricSet) { + const maxPages = 100000 + parent := fmt.Sprintf(q.Parent, namespace) + token := "" + + for page := 0; page < maxPages; page++ { + var next string + err := observe(m, q.Op, func() error { + var e error + next, e = listPage(ctx, l, q, parent, pageSize, token) + return e + }) + if err != nil || next == "" || next == token { + return + } + token = next + } +} + +// listPage issues a single page request and returns the next page token. +func listPage(ctx context.Context, l lister, q query, parent string, pageSize int32, token string) (string, error) { + switch q.Kind { + case kindResults: + resp, err := l.ListResults(ctx, &pb.ListResultsRequest{ + Parent: parent, + Filter: q.Filter, + PageSize: pageSize, + PageToken: token, + }) + if err != nil { + return "", err + } + return resp.GetNextPageToken(), nil + case kindRecords: + resp, err := l.ListRecords(ctx, &pb.ListRecordsRequest{ + Parent: parent, + Filter: q.Filter, + PageSize: pageSize, + PageToken: token, + }) + if err != nil { + return "", err + } + return resp.GetNextPageToken(), nil + default: + return "", fmt.Errorf("unknown query kind %d", q.Kind) + } +} diff --git a/test/performance/harness/store.go b/test/performance/harness/store.go new file mode 100644 index 000000000..3b1137e49 --- /dev/null +++ b/test/performance/harness/store.go @@ -0,0 +1,265 @@ +/* +Copyright 2026 The Tekton Authors + +Licensed 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. +*/ + +package main + +import ( + "context" + "encoding/json" + "io" + "slices" + "sync" + + corev1 "k8s.io/api/core/v1" + "knative.dev/pkg/apis" + duckv1 "knative.dev/pkg/apis/duck/v1" + + tektonv1 "github.com/tektoncd/pipeline/pkg/apis/pipeline/v1" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + "google.golang.org/protobuf/types/known/timestamppb" + + record "github.com/tektoncd/results/pkg/api/server/v1alpha2/record" + "github.com/tektoncd/results/pkg/watcher/convert" + pb "github.com/tektoncd/results/proto/v1alpha2/results_go_proto" + "github.com/tektoncd/results/test/performance/generator" + "github.com/tektoncd/results/test/performance/metrics" +) + +// storeOps are the per-RPC metric names the store driver records. +const ( + opCreateResult = "create_result" + opCreateRecord = "create_record" + opUpdateRecord = "update_record" + opUpdateResult = "update_result" + opCreateChild = "create_child_record" +) + +// sendLogEntry is the actual record of what the store driver wrote for one +// instance. It joins 1:1 with generator.DatasetDefinition on UID so callers can +// verify the store faithfully persisted the golden dataset. +type sendLogEntry struct { + Index int `json:"index"` + UID string `json:"uid"` + ResultName string `json:"result_name"` + RecordName string `json:"record_name"` + ChildRecords []string `json:"child_records"` + UpdateCount int `json:"update_count"` + Codes []string `json:"codes"` +} + +// sendLog is a concurrency-safe collector of send-log entries. +type sendLog struct { + mu sync.Mutex + entries []sendLogEntry +} + +func (l *sendLog) add(e sendLogEntry) { + l.mu.Lock() + l.entries = append(l.entries, e) + l.mu.Unlock() +} + +// write emits the send log as indented JSON, ordered by instance index so the +// output is stable across runs regardless of worker scheduling. +func (l *sendLog) write(w io.Writer) error { + l.mu.Lock() + entries := make([]sendLogEntry, len(l.entries)) + copy(entries, l.entries) + l.mu.Unlock() + + slices.SortFunc(entries, func(a, b sendLogEntry) int { return a.Index - b.Index }) + enc := json.NewEncoder(w) + enc.SetIndent("", " ") + return enc.Encode(entries) +} + +// storeInstance replays the watcher write lifecycle for one instance over gRPC: +// ensure the parent Result, create the top-level Record in a pending state, apply +// 1-3 terminal-progression updates (etag-chained), update the Result summary, and +// create the child TaskRun records. Every RPC is timed into m by op name. +func storeInstance(ctx context.Context, gc pb.ResultsClient, inst *generator.Instance, updates int, m *metrics.MetricSet) sendLogEntry { + entry := sendLogEntry{ + Index: inst.Index, + UID: inst.UID, + ResultName: inst.ResultName, + RecordName: inst.RecordName, + ChildRecords: inst.ChildRecords, + UpdateCount: updates, + } + record := func(err error) { entry.Codes = append(entry.Codes, metrics.Classify(err)) } + + // 1. Ensure the parent Result exists (tolerate reruns via AlreadyExists). + err := observe(m, opCreateResult, func() error { + _, e := gc.CreateResult(ctx, &pb.CreateResultRequest{ + Parent: inst.Namespace, + Result: &pb.Result{ + Name: inst.ResultName, + Summary: recordSummary(inst, pb.RecordSummary_UNKNOWN), + }, + }) + return e + }) + if status.Code(err) == codes.AlreadyExists { + err = nil + } + record(err) + + // 2. Create the top-level Record in a pending state, then apply the terminal + // progression as etag-chained updates. + stages := stagedPipelineRuns(inst.PipelineRun, updates) + pending, convErr := convert.ToProto(stages[0]) + if convErr != nil { + record(convErr) + return entry + } + var current *pb.Record + err = observe(m, opCreateRecord, func() error { + rec, e := gc.CreateRecord(ctx, &pb.CreateRecordRequest{ + Parent: inst.ResultName, + Record: &pb.Record{Name: inst.RecordName, Data: pending}, + }) + current = rec + return e + }) + record(err) + + for _, stage := range stages[1:] { + data, e := convert.ToProto(stage) + if e != nil { + record(e) + continue + } + if current == nil { + // The create failed; nothing to update against. + record(status.Error(codes.FailedPrecondition, "no record to update")) + continue + } + current.Data = data + err = observe(m, opUpdateRecord, func() error { + updated, e := gc.UpdateRecord(ctx, &pb.UpdateRecordRequest{Record: current, Etag: current.GetEtag()}) + if e == nil { + current = updated + } + return e + }) + record(err) + } + + // 3. Update the Result summary to the terminal status. + err = observe(m, opUpdateResult, func() error { + _, e := gc.UpdateResult(ctx, &pb.UpdateResultRequest{ + Name: inst.ResultName, + Result: &pb.Result{ + Name: inst.ResultName, + Summary: recordSummary(inst, summaryStatus(inst.Outcome)), + }, + }) + return e + }) + record(err) + + // 4. Create the child TaskRun records under the same Result. + for _, tr := range inst.TaskRuns { + data, e := convert.ToProto(tr) + if e != nil { + record(e) + continue + } + name := childRecordName(inst.ResultName, tr) + err = observe(m, opCreateChild, func() error { + _, e := gc.CreateRecord(ctx, &pb.CreateRecordRequest{ + Parent: inst.ResultName, + Record: &pb.Record{Name: name, Data: data}, + }) + return e + }) + if status.Code(err) == codes.AlreadyExists { + err = nil + } + record(err) + } + + return entry +} + +// childRecordName mirrors the generator's child record naming. +func childRecordName(resultName string, tr *tektonv1.TaskRun) string { + return record.FormatName(resultName, string(tr.UID)) +} + +// recordSummary builds the RecordSummary for the top-level record. +func recordSummary(inst *generator.Instance, s pb.RecordSummary_Status) *pb.RecordSummary { + summary := &pb.RecordSummary{ + Record: inst.RecordName, + Type: convert.TypeName(inst.PipelineRun), + Status: s, + } + if t := inst.PipelineRun.Status.StartTime; t != nil { + summary.StartTime = timestamppb.New(t.Time) + } + if t := inst.PipelineRun.Status.CompletionTime; t != nil { + summary.EndTime = timestamppb.New(t.Time) + } + return summary +} + +// summaryStatus maps a generated outcome to the API record-summary status enum. +func summaryStatus(o generator.Outcome) pb.RecordSummary_Status { + switch o { + case generator.OutcomeSucceeded: + return pb.RecordSummary_SUCCESS + case generator.OutcomeFailed: + return pb.RecordSummary_FAILURE + case generator.OutcomeCancelled: + return pb.RecordSummary_CANCELLED + default: + return pb.RecordSummary_UNKNOWN + } +} + +// stagedPipelineRuns returns the lifecycle stages to persist: index 0 is the +// pending state, the final element is the terminal object, and any in between are +// running snapshots. There are updates+1 stages, i.e. one create plus `updates` +// UpdateRecord calls. +func stagedPipelineRuns(terminal *tektonv1.PipelineRun, updates int) []*tektonv1.PipelineRun { + if updates < 1 { + updates = 1 + } + running := terminal.DeepCopy() + running.Status.CompletionTime = nil + running.Status.Conditions = duckv1.Conditions{{ + Type: apis.ConditionSucceeded, + Status: corev1.ConditionUnknown, + Reason: "Running", + Message: "Not all Tasks have completed executing", + }} + + stages := make([]*tektonv1.PipelineRun, 0, updates+1) + stages = append(stages, running) + for i := 1; i < updates; i++ { + stages = append(stages, running.DeepCopy()) + } + stages = append(stages, terminal) + return stages +} + +// updateCountFor derives a deterministic number of record updates (1-3) for an +// instance. It does not affect golden answers (only the final state persists) but +// keeps the write volume reproducible across runs. +func updateCountFor(index int) int { + return 1 + index%3 +} diff --git a/test/performance/harness/worker.go b/test/performance/harness/worker.go new file mode 100644 index 000000000..27c69b6d7 --- /dev/null +++ b/test/performance/harness/worker.go @@ -0,0 +1,115 @@ +/* +Copyright 2026 The Tekton Authors + +Licensed 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. +*/ + +package main + +import ( + "context" + "sync" + "time" + + "github.com/tektoncd/results/test/performance/metrics" +) + +// indexFunc processes a single dataset index, recording its own metrics. +type indexFunc func(ctx context.Context, index int, m *metrics.MetricSet) + +// tickFunc runs one unit of work in a time-bounded loop, recording its own +// metrics. iter increments per call within a worker. +type tickFunc func(ctx context.Context, workerID, iter int, m *metrics.MetricSet) + +// runIndexed spreads indices [start, start+count) across concurrency workers by +// stride, so a given index (and therefore a given Result) is only ever touched +// by one worker — eliminating etag contention on the store path. Each worker +// records into its own MetricSet; all are merged into the returned set. +func runIndexed(ctx context.Context, concurrency, start, count int, fn indexFunc) *metrics.MetricSet { + if concurrency < 1 { + concurrency = 1 + } + total := metrics.NewMetricSet() + sets := make([]*metrics.MetricSet, concurrency) + for i := range sets { + sets[i] = metrics.NewMetricSet() + } + + total.Start() + var wg sync.WaitGroup + for w := 0; w < concurrency; w++ { + wg.Add(1) + go func(w int) { + defer wg.Done() + m := sets[w] + for i := start + w; i < start+count; i += concurrency { + if ctx.Err() != nil { + return + } + fn(ctx, i, m) + } + }(w) + } + wg.Wait() + total.Stop() + + for _, s := range sets { + total.Merge(s) + } + return total +} + +// runTimed runs fn repeatedly on concurrency workers until the duration elapses +// or ctx is cancelled. Used for read-heavy workloads with no natural index bound. +func runTimed(ctx context.Context, concurrency int, duration time.Duration, fn tickFunc) *metrics.MetricSet { + if concurrency < 1 { + concurrency = 1 + } + runCtx, cancel := context.WithTimeout(ctx, duration) + defer cancel() + + total := metrics.NewMetricSet() + sets := make([]*metrics.MetricSet, concurrency) + for i := range sets { + sets[i] = metrics.NewMetricSet() + } + + total.Start() + var wg sync.WaitGroup + for w := 0; w < concurrency; w++ { + wg.Add(1) + go func(w int) { + defer wg.Done() + m := sets[w] + for iter := 0; runCtx.Err() == nil; iter++ { + fn(runCtx, w, iter, m) + } + }(w) + } + wg.Wait() + total.Stop() + + for _, s := range sets { + total.Merge(s) + } + return total +} + +// observe times fn, records the latency and result under op, and returns the +// error so callers can chain (e.g. etag updates). +func observe(m *metrics.MetricSet, op string, fn func() error) error { + start := time.Now() + err := fn() + m.Observe(op, time.Since(start), err) + return err +} diff --git a/test/performance/metrics/metrics.go b/test/performance/metrics/metrics.go new file mode 100644 index 000000000..91c69a14c --- /dev/null +++ b/test/performance/metrics/metrics.go @@ -0,0 +1,307 @@ +/* +Copyright 2026 The Tekton Authors + +Licensed 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. +*/ + +// Package metrics collects per-operation latency and error statistics for the +// performance benchmark harness. Percentiles are computed exactly from the +// recorded samples (no external histogram dependency); per-worker recorders can +// be merged to avoid lock contention on the hot path. +package metrics + +import ( + "maps" + "slices" + "sort" + "sync" + "time" + + "google.golang.org/grpc/status" +) + +// LatencyRecorder accumulates operation latencies and computes exact +// percentiles on demand. +type LatencyRecorder struct { + mu sync.Mutex + samples []time.Duration +} + +// NewLatencyRecorder returns an empty recorder. +func NewLatencyRecorder() *LatencyRecorder { return &LatencyRecorder{} } + +// Observe records a single latency sample. +func (r *LatencyRecorder) Observe(d time.Duration) { + r.mu.Lock() + r.samples = append(r.samples, d) + r.mu.Unlock() +} + +// Merge folds another recorder's samples into r. +func (r *LatencyRecorder) Merge(o *LatencyRecorder) { + o.mu.Lock() + defer o.mu.Unlock() + r.mu.Lock() + defer r.mu.Unlock() + r.samples = append(r.samples, o.samples...) +} + +// LatencyStats is a computed summary of a set of latency samples. +type LatencyStats struct { + Count int64 + Min time.Duration + Max time.Duration + Mean time.Duration + P50 time.Duration + P90 time.Duration + P99 time.Duration +} + +// Snapshot sorts the samples once and computes the summary statistics. +func (r *LatencyRecorder) Snapshot() LatencyStats { + r.mu.Lock() + sorted := make([]time.Duration, len(r.samples)) + copy(sorted, r.samples) + r.mu.Unlock() + + if len(sorted) == 0 { + return LatencyStats{} + } + slices.Sort(sorted) + + var sum time.Duration + for _, d := range sorted { + sum += d + } + return LatencyStats{ + Count: int64(len(sorted)), + Min: sorted[0], + Max: sorted[len(sorted)-1], + Mean: sum / time.Duration(len(sorted)), + P50: percentile(sorted, 0.50), + P90: percentile(sorted, 0.90), + P99: percentile(sorted, 0.99), + } +} + +// percentile returns the p-quantile (p in [0,1]) using nearest-rank on a sorted +// slice. sorted must be non-empty. +func percentile(sorted []time.Duration, p float64) time.Duration { + if len(sorted) == 1 { + return sorted[0] + } + rank := int(p * float64(len(sorted))) + if rank >= len(sorted) { + rank = len(sorted) - 1 + } + return sorted[rank] +} + +// ErrorCounter tallies operation results by gRPC status code (OK for success). +type ErrorCounter struct { + mu sync.Mutex + byCode map[string]int64 +} + +// NewErrorCounter returns an empty counter. +func NewErrorCounter() *ErrorCounter { return &ErrorCounter{byCode: map[string]int64{}} } + +// Record classifies err by gRPC status code and increments the tally. A nil err +// is counted as "OK". +func (e *ErrorCounter) Record(err error) { e.RecordCode(Classify(err)) } + +// RecordCode increments the tally for an explicit code string. +func (e *ErrorCounter) RecordCode(code string) { + e.mu.Lock() + e.byCode[code]++ + e.mu.Unlock() +} + +// Merge folds another counter into e. +func (e *ErrorCounter) Merge(o *ErrorCounter) { + o.mu.Lock() + defer o.mu.Unlock() + e.mu.Lock() + defer e.mu.Unlock() + for k, v := range o.byCode { + e.byCode[k] += v + } +} + +// Snapshot returns a copy of the per-code tallies. +func (e *ErrorCounter) Snapshot() map[string]int64 { + e.mu.Lock() + defer e.mu.Unlock() + out := make(map[string]int64, len(e.byCode)) + maps.Copy(out, e.byCode) + return out +} + +// Failures returns the number of non-OK results. +func (e *ErrorCounter) Failures() int64 { + e.mu.Lock() + defer e.mu.Unlock() + var total int64 + for k, v := range e.byCode { + if k != "OK" { + total += v + } + } + return total +} + +// Classify maps an error to a gRPC status code string. Nil maps to "OK"; +// non-status errors map to "Unknown". +func Classify(err error) string { + return status.Code(err).String() +} + +// OpMetrics bundles the latency recorder and error counter for one operation. +type OpMetrics struct { + Latency *LatencyRecorder + Errors *ErrorCounter +} + +// NewOpMetrics returns an initialized OpMetrics. +func NewOpMetrics() *OpMetrics { + return &OpMetrics{Latency: NewLatencyRecorder(), Errors: NewErrorCounter()} +} + +// Merge folds another OpMetrics into m. +func (m *OpMetrics) Merge(o *OpMetrics) { + m.Latency.Merge(o.Latency) + m.Errors.Merge(o.Errors) +} + +// MetricSet holds per-operation metrics for a benchmark run. +type MetricSet struct { + mu sync.Mutex + ops map[string]*OpMetrics + start time.Time + end time.Time +} + +// NewMetricSet returns an initialized MetricSet. +func NewMetricSet() *MetricSet { return &MetricSet{ops: map[string]*OpMetrics{}} } + +// Start marks the beginning of the measured window. +func (m *MetricSet) Start() { + m.mu.Lock() + m.start = time.Now() + m.mu.Unlock() +} + +// Stop marks the end of the measured window. +func (m *MetricSet) Stop() { + m.mu.Lock() + m.end = time.Now() + m.mu.Unlock() +} + +// Op returns the OpMetrics for name, creating it on first use. +func (m *MetricSet) Op(name string) *OpMetrics { + m.mu.Lock() + defer m.mu.Unlock() + op, ok := m.ops[name] + if !ok { + op = NewOpMetrics() + m.ops[name] = op + } + return op +} + +// Observe is a convenience that records a latency and result for an operation. +func (m *MetricSet) Observe(name string, d time.Duration, err error) { + op := m.Op(name) + op.Latency.Observe(d) + op.Errors.Record(err) +} + +// Merge folds another MetricSet's operations into m, preserving m's window. +func (m *MetricSet) Merge(o *MetricSet) { + o.mu.Lock() + names := make([]string, 0, len(o.ops)) + src := make(map[string]*OpMetrics, len(o.ops)) + for k, v := range o.ops { + names = append(names, k) + src[k] = v + } + o.mu.Unlock() + for _, name := range names { + m.Op(name).Merge(src[name]) + } +} + +// Duration returns the measured wall-clock window. +func (m *MetricSet) Duration() time.Duration { + m.mu.Lock() + defer m.mu.Unlock() + if m.end.IsZero() { + return time.Since(m.start) + } + return m.end.Sub(m.start) +} + +// StartTime returns the window start. +func (m *MetricSet) StartTime() time.Time { + m.mu.Lock() + defer m.mu.Unlock() + return m.start +} + +// OpSnapshot is the computed summary for one operation. +type OpSnapshot struct { + Name string + Latency LatencyStats + Errors map[string]int64 + ErrorTotal int64 + Total int64 +} + +// Snapshot is the computed summary for a whole run. +type Snapshot struct { + Ops []OpSnapshot + Duration time.Duration + Start time.Time + TotalOps int64 + TotalError int64 +} + +// Snapshot computes summaries for every recorded operation, sorted by name. +func (m *MetricSet) Snapshot() Snapshot { + m.mu.Lock() + names := make([]string, 0, len(m.ops)) + for k := range m.ops { + names = append(names, k) + } + ops := m.ops + m.mu.Unlock() + sort.Strings(names) + + snap := Snapshot{Duration: m.Duration(), Start: m.StartTime()} + for _, name := range names { + op := ops[name] + ls := op.Latency.Snapshot() + failures := op.Errors.Failures() + snap.Ops = append(snap.Ops, OpSnapshot{ + Name: name, + Latency: ls, + Errors: op.Errors.Snapshot(), + ErrorTotal: failures, + Total: ls.Count, + }) + snap.TotalOps += ls.Count + snap.TotalError += failures + } + return snap +} diff --git a/test/performance/metrics/metrics_test.go b/test/performance/metrics/metrics_test.go new file mode 100644 index 000000000..7a7811fc2 --- /dev/null +++ b/test/performance/metrics/metrics_test.go @@ -0,0 +1,164 @@ +/* +Copyright 2026 The Tekton Authors + +Licensed 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. +*/ + +package metrics + +import ( + "errors" + "testing" + "time" + + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +func TestLatencyStatsExact(t *testing.T) { + r := NewLatencyRecorder() + // 100 samples: 1ms..100ms, inserted out of order to exercise sorting. + for i := 100; i >= 1; i-- { + r.Observe(time.Duration(i) * time.Millisecond) + } + got := r.Snapshot() + + want := LatencyStats{ + Count: 100, + Min: 1 * time.Millisecond, + Max: 100 * time.Millisecond, + Mean: 50500 * time.Microsecond, // (1+..+100)ms / 100 = 50.5ms + P50: 51 * time.Millisecond, // nearest-rank: index 50 + P90: 91 * time.Millisecond, + P99: 100 * time.Millisecond, + } + if got != want { + t.Errorf("Snapshot() = %+v, want %+v", got, want) + } +} + +func TestLatencyStatsEmpty(t *testing.T) { + if got := NewLatencyRecorder().Snapshot(); got != (LatencyStats{}) { + t.Errorf("empty Snapshot() = %+v, want zero value", got) + } +} + +func TestLatencyStatsSingle(t *testing.T) { + r := NewLatencyRecorder() + r.Observe(7 * time.Millisecond) + got := r.Snapshot() + want := LatencyStats{Count: 1, Min: 7 * time.Millisecond, Max: 7 * time.Millisecond, Mean: 7 * time.Millisecond, P50: 7 * time.Millisecond, P90: 7 * time.Millisecond, P99: 7 * time.Millisecond} + if got != want { + t.Errorf("single Snapshot() = %+v, want %+v", got, want) + } +} + +func TestLatencyMergeEqualsSingle(t *testing.T) { + single := NewLatencyRecorder() + a := NewLatencyRecorder() + b := NewLatencyRecorder() + for i := 1; i <= 200; i++ { + d := time.Duration(i) * time.Microsecond + single.Observe(d) + if i%2 == 0 { + a.Observe(d) + } else { + b.Observe(d) + } + } + a.Merge(b) + if got, want := a.Snapshot(), single.Snapshot(); got != want { + t.Errorf("merged Snapshot() = %+v, want %+v", got, want) + } +} + +func TestClassify(t *testing.T) { + tests := []struct { + name string + err error + want string + }{ + {"nil", nil, "OK"}, + {"grpc not found", status.Error(codes.NotFound, "missing"), "NotFound"}, + {"grpc already exists", status.Error(codes.AlreadyExists, "dup"), "AlreadyExists"}, + {"plain error", errors.New("boom"), "Unknown"}, + } + for _, tt := range tests { + if got := Classify(tt.err); got != tt.want { + t.Errorf("Classify(%s) = %q, want %q", tt.name, got, tt.want) + } + } +} + +func TestErrorCounter(t *testing.T) { + e := NewErrorCounter() + e.Record(nil) + e.Record(nil) + e.Record(status.Error(codes.NotFound, "x")) + e.Record(status.Error(codes.NotFound, "y")) + e.Record(status.Error(codes.AlreadyExists, "z")) + + snap := e.Snapshot() + if snap["OK"] != 2 || snap["NotFound"] != 2 || snap["AlreadyExists"] != 1 { + t.Errorf("Snapshot() = %v", snap) + } + if got := e.Failures(); got != 3 { + t.Errorf("Failures() = %d, want 3", got) + } +} + +func TestMetricSetSnapshot(t *testing.T) { + m := NewMetricSet() + m.Start() + m.Observe("create_record", 10*time.Millisecond, nil) + m.Observe("create_record", 20*time.Millisecond, status.Error(codes.Unavailable, "down")) + m.Observe("update_record", 5*time.Millisecond, nil) + m.Stop() + + snap := m.Snapshot() + if len(snap.Ops) != 2 { + t.Fatalf("ops = %d, want 2", len(snap.Ops)) + } + // Sorted by name: create_record before update_record. + if snap.Ops[0].Name != "create_record" || snap.Ops[1].Name != "update_record" { + t.Errorf("ops not sorted: %s, %s", snap.Ops[0].Name, snap.Ops[1].Name) + } + if snap.TotalOps != 3 { + t.Errorf("TotalOps = %d, want 3", snap.TotalOps) + } + if snap.TotalError != 1 { + t.Errorf("TotalError = %d, want 1", snap.TotalError) + } + if snap.Ops[0].ErrorTotal != 1 { + t.Errorf("create_record ErrorTotal = %d, want 1", snap.Ops[0].ErrorTotal) + } +} + +func TestMetricSetMerge(t *testing.T) { + combined := NewMetricSet() + a := NewMetricSet() + b := NewMetricSet() + for i := 1; i <= 50; i++ { + d := time.Duration(i) * time.Millisecond + combined.Observe("op", d, nil) + if i%2 == 0 { + a.Observe("op", d, nil) + } else { + b.Observe("op", d, nil) + } + } + a.Merge(b) + if got, want := a.Snapshot().Ops[0].Latency, combined.Snapshot().Ops[0].Latency; got != want { + t.Errorf("merged latency = %+v, want %+v", got, want) + } +} diff --git a/test/performance/report/meta.go b/test/performance/report/meta.go new file mode 100644 index 000000000..22f83ab8c --- /dev/null +++ b/test/performance/report/meta.go @@ -0,0 +1,53 @@ +/* +Copyright 2026 The Tekton Authors + +Licensed 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. +*/ + +package report + +import ( + "context" + "os" + "os/exec" + "strings" +) + +// CaptureMeta fills in the environment-derived provenance fields (git commit, +// dirty state, hostname). Callers supply the run-specific fields (tier, mode, +// dataset, backend, server image) on the returned struct. Failures to read git +// or the hostname are non-fatal — the corresponding field is left empty. +func CaptureMeta(ctx context.Context) Meta { + meta := Meta{} + if commit, err := gitOutput(ctx, "rev-parse", "HEAD"); err == nil { + meta.GitCommit = commit + // Only meaningful inside a git repo: a non-zero exit from + // `git diff --quiet` means the working tree has uncommitted changes. + if _, err := gitOutput(ctx, "diff", "--quiet"); err != nil { + meta.GitDirty = true + } + } + if host, err := os.Hostname(); err == nil { + meta.Hostname = host + } + return meta +} + +// gitOutput runs a git command and returns its trimmed stdout. +func gitOutput(ctx context.Context, args ...string) (string, error) { + out, err := exec.CommandContext(ctx, "git", args...).Output() //nolint:gosec // fixed "git" binary, caller-controlled static args + if err != nil { + return "", err + } + return strings.TrimSpace(string(out)), nil +} diff --git a/test/performance/report/report.go b/test/performance/report/report.go new file mode 100644 index 000000000..cf8e12b26 --- /dev/null +++ b/test/performance/report/report.go @@ -0,0 +1,148 @@ +/* +Copyright 2026 The Tekton Authors + +Licensed 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. +*/ + +// Package report defines the machine-readable benchmark report schema emitted by +// the harness and consumed by the compare tool (Story 02). It converts a +// metrics.Snapshot into a stable, versioned JSON document annotated with run +// metadata (git commit, dataset version, tier, backend). +package report + +import ( + "encoding/json" + "io" + "time" + + "github.com/tektoncd/results/test/performance/metrics" +) + +// SchemaVersion identifies the report format; bump on any breaking field change. +const SchemaVersion = "1" + +// Report is the top-level benchmark artifact. +type Report struct { + SchemaVersion string `json:"schema_version"` + Meta Meta `json:"meta"` + Config Config `json:"config"` + Metrics Metrics `json:"metrics"` +} + +// Meta captures the provenance needed to compare two runs meaningfully. +type Meta struct { + GitCommit string `json:"git_commit"` + GitDirty bool `json:"git_dirty"` + DatasetVersion string `json:"dataset_version"` + DatasetHash string `json:"dataset_hash"` + Tier string `json:"tier"` + Mode string `json:"mode"` + StartedAt string `json:"started_at"` + DurationMS int64 `json:"duration_ms"` + Hostname string `json:"hostname"` + APIServerAddr string `json:"api_server_addr"` + DBBackend string `json:"db_backend"` + ServerImage string `json:"server_image"` +} + +// Config records the knobs that shaped the run so results are reproducible. +type Config struct { + Count int `json:"count"` + Concurrency int `json:"concurrency"` + DurationSec int `json:"duration_sec"` + Transport string `json:"transport"` + ReadRatio int `json:"read_ratio"` + WriteRatio int `json:"write_ratio"` + Dataset string `json:"dataset"` + Seed int64 `json:"seed"` +} + +// Metrics is the aggregate performance result. +type Metrics struct { + ThroughputPerSec float64 `json:"throughput_per_sec"` + TotalOps int64 `json:"total_ops"` + TotalErrors int64 `json:"total_errors"` + ByOp map[string]OpReport `json:"by_op"` +} + +// OpReport is the per-operation summary in report units (milliseconds). +type OpReport struct { + Count int64 `json:"count"` + Errors int64 `json:"errors"` + ErrorCodes map[string]int64 `json:"error_codes"` + P50MS float64 `json:"p50_ms"` + P90MS float64 `json:"p90_ms"` + P99MS float64 `json:"p99_ms"` + MinMS float64 `json:"min_ms"` + MaxMS float64 `json:"max_ms"` + MeanMS float64 `json:"mean_ms"` +} + +// New assembles a Report from run metadata, config, and a metrics snapshot. +func New(meta Meta, cfg Config, snap metrics.Snapshot) *Report { + meta.DurationMS = snap.Duration.Milliseconds() + if meta.StartedAt == "" && !snap.Start.IsZero() { + meta.StartedAt = snap.Start.UTC().Format(time.RFC3339) + } + return &Report{ + SchemaVersion: SchemaVersion, + Meta: meta, + Config: cfg, + Metrics: buildMetrics(snap), + } +} + +// buildMetrics converts the metrics snapshot into report units. +func buildMetrics(snap metrics.Snapshot) Metrics { + m := Metrics{ + TotalOps: snap.TotalOps, + TotalErrors: snap.TotalError, + ByOp: make(map[string]OpReport, len(snap.Ops)), + } + if secs := snap.Duration.Seconds(); secs > 0 { + m.ThroughputPerSec = float64(snap.TotalOps) / secs + } + for _, op := range snap.Ops { + m.ByOp[op.Name] = OpReport{ + Count: op.Total, + Errors: op.ErrorTotal, + ErrorCodes: op.Errors, + P50MS: ms(op.Latency.P50), + P90MS: ms(op.Latency.P90), + P99MS: ms(op.Latency.P99), + MinMS: ms(op.Latency.Min), + MaxMS: ms(op.Latency.Max), + MeanMS: ms(op.Latency.Mean), + } + } + return m +} + +// ms converts a duration to fractional milliseconds. +func ms(d time.Duration) float64 { return float64(d) / float64(time.Millisecond) } + +// Write emits the report as indented JSON with a trailing newline. +func Write(w io.Writer, r *Report) error { + enc := json.NewEncoder(w) + enc.SetIndent("", " ") + return enc.Encode(r) +} + +// Read decodes a report from JSON. +func Read(r io.Reader) (*Report, error) { + rep := &Report{} + if err := json.NewDecoder(r).Decode(rep); err != nil { + return nil, err + } + return rep, nil +} diff --git a/test/performance/report/report_test.go b/test/performance/report/report_test.go new file mode 100644 index 000000000..44d946130 --- /dev/null +++ b/test/performance/report/report_test.go @@ -0,0 +1,116 @@ +/* +Copyright 2026 The Tekton Authors + +Licensed 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. +*/ + +package report + +import ( + "bytes" + "encoding/json" + "testing" + "time" + + "github.com/google/go-cmp/cmp" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + + "github.com/tektoncd/results/test/performance/metrics" +) + +func sampleSnapshot() metrics.Snapshot { + m := metrics.NewMetricSet() + m.Start() + for i := 1; i <= 100; i++ { + m.Observe("create_record", time.Duration(i)*time.Millisecond, nil) + } + m.Observe("create_record", 5*time.Millisecond, status.Error(codes.Unavailable, "down")) + m.Stop() + snap := m.Snapshot() + // Pin a deterministic duration so throughput is stable in assertions. + snap.Duration = 2 * time.Second + return snap +} + +func TestNewReportShape(t *testing.T) { + snap := sampleSnapshot() + meta := Meta{GitCommit: "abc123", DatasetVersion: "seed-small-v1", Tier: "tier1", Mode: "store", DBBackend: "local"} + cfg := Config{Count: 100, Concurrency: 8, Transport: "grpc", Seed: 42} + + r := New(meta, cfg, snap) + + if r.SchemaVersion != SchemaVersion { + t.Errorf("SchemaVersion = %q, want %q", r.SchemaVersion, SchemaVersion) + } + if r.Meta.DurationMS != 2000 { + t.Errorf("DurationMS = %d, want 2000", r.Meta.DurationMS) + } + op, ok := r.Metrics.ByOp["create_record"] + if !ok { + t.Fatal("missing create_record op report") + } + if op.Count != 101 { + t.Errorf("Count = %d, want 101", op.Count) + } + if op.Errors != 1 || op.ErrorCodes["Unavailable"] != 1 { + t.Errorf("errors = %d, codes = %v", op.Errors, op.ErrorCodes) + } + // 101 ops over 2s. + if want := 101.0 / 2.0; r.Metrics.ThroughputPerSec != want { + t.Errorf("ThroughputPerSec = %f, want %f", r.Metrics.ThroughputPerSec, want) + } +} + +func TestReportRoundTrip(t *testing.T) { + r := New( + Meta{GitCommit: "deadbeef", Tier: "tier1", Mode: "mixed"}, + Config{Count: 10, Concurrency: 2}, + sampleSnapshot(), + ) + + var buf bytes.Buffer + if err := Write(&buf, r); err != nil { + t.Fatalf("Write() = %v", err) + } + got, err := Read(&buf) + if err != nil { + t.Fatalf("Read() = %v", err) + } + if diff := cmp.Diff(r, got); diff != "" { + t.Errorf("round-trip mismatch (-want +got):\n%s", diff) + } +} + +func TestReportRequiredKeys(t *testing.T) { + r := New(Meta{Mode: "query"}, Config{}, sampleSnapshot()) + var buf bytes.Buffer + if err := Write(&buf, r); err != nil { + t.Fatalf("Write() = %v", err) + } + var generic map[string]any + if err := json.Unmarshal(buf.Bytes(), &generic); err != nil { + t.Fatalf("unmarshal: %v", err) + } + for _, key := range []string{"schema_version", "meta", "config", "metrics"} { + if _, ok := generic[key]; !ok { + t.Errorf("report missing top-level key %q", key) + } + } + metricsObj := generic["metrics"].(map[string]any) + for _, key := range []string{"throughput_per_sec", "total_ops", "total_errors", "by_op"} { + if _, ok := metricsObj[key]; !ok { + t.Errorf("metrics missing key %q", key) + } + } +}