Replace load-generator with Go implementation

.github/dependabot.yml

@@ -12,7 +12,7 @@ updates:
         patterns:
           - "go.opentelemetry.io/*"
     open-pull-requests-limit: 20
-  - package-ecosystem: "pip"
-    directory: "/tools/load-generator"
-    schedule:
-      interval: "monthly"
+  #- package-ecosystem: "pip"
+   # directory: "/tools/load-generator"
+   # schedule:
+    #  interval: "monthly"

prombench/manifests/prombench/benchmark/6_loadgen.yaml

@@ -48,6 +48,25 @@ data:
         - expr: histogram_quantile(0.99, sum by(path, le) (rate(codelab_api_request_duration_seconds_bucket{method="POST"}[5m])))
         - expr: histogram_quantile(0.99, sum by(path, method, le) (rate(codelab_api_request_duration_seconds_bucket{method="POST"}[5m])))
         - expr: histogram_quantile(0.99, sum by(instance, le) (rate(codelab_api_request_duration_seconds_bucket{method="POST"}[5m])))
+      - name: binary_arithmetic_operators
+        interval: 15s
+        type: instant
+        queries:
+        - expr: sum(node_memory_MemAvailable_bytes) / sum(node_memory_MemTotal_bytes)
+        - expr: rate(node_network_transmit_bytes_total[5m]) * 8
+        - expr: node_filesystem_avail_bytes / node_filesystem_size_bytes
+      - name: logical_operators
+        interval: 15s
+        type: instant
+        queries:
+        - expr: rate(node_cpu_seconds_total{mode="system"}[5m]) and rate(node_cpu_seconds_total{mode="user"}[5m])
+        - expr: node_filesystem_avail_bytes unless node_filesystem_size_bytes
+        - expr: rate(node_network_receive_bytes_total[5m]) or rate(node_network_transmit_bytes_total[5m])
+      - name: topk_example
+        interval: 15s
+        type: instant
+        queries:
+        - expr: topk(5, rate(http_requests_total[5m]))
 ---
 apiVersion: apps/v1
 kind: Deployment

tools/load-generator/Dockerfile

@@ -1,10 +1,14 @@
-FROM python:3
+FROM golang:1.20-alpine
 
-COPY ./requirements.txt /usr/src/app/
-RUN pip install -r /usr/src/app/requirements.txt
+WORKDIR /usr/src/app
 
-COPY ./*.py /usr/src/app/
+COPY go.mod go.sum ./
+RUN go mod download
 
-WORKDIR /usr/src/app
+COPY . .
+
+RUN go build -o load-generator .
+
+EXPOSE 8080
 
-ENTRYPOINT ["python", "-u", "./main.py"]
+ENTRYPOINT ["./load-generator"]

tools/load-generator/main.go

@@ -0,0 +1,230 @@
+package main
+
+import (
+	"fmt"
+	"io"
+	"log"
+	"net/http"
+	"os"
+	"sync"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+	"gopkg.in/yaml.v2"
+)
+
+// Global variables and Prometheus metrics
+var (
+	namespace    string
+	max404Errors = 30
+	domainName   = os.Getenv("DOMAIN_NAME")
+
+	queryDuration = prometheus.NewHistogramVec(
+		prometheus.HistogramOpts{
+			Namespace: "loadgen",
+			Name:      "query_duration_seconds",
+			Help:      "Query duration",
+			Buckets:   prometheus.LinearBuckets(0.05, 0.1, 20),
+		},
+		[]string{"prometheus", "group", "expr", "type"},
+	)
+	queryCount = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Namespace: "loadgen",
+			Name:      "queries_total",
+			Help:      "Total amount of queries",
+		},
+		[]string{"prometheus", "group", "expr", "type"},
+	)
+	queryFailCount = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Namespace: "loadgen",
+			Name:      "failed_queries_total",
+			Help:      "Amount of failed queries",
+		},
+		[]string{"prometheus", "group", "expr", "type"},
+	)
+)
+
+// Querier struct and methods
+type Querier struct {
+	target         string
+	name           string
+	groupID        int
+	numberOfErrors int
+
+	interval time.Duration
+	queries  []Query
+	qtype    string
+	start    time.Duration
+	end      time.Duration
+	step     string
+	url      string
+}
+
+type Query struct {
+	Expr string `yaml:"expr"`
+}
+
+type QueryGroup struct {
+	Name     string  `yaml:"name"`
+	Interval string  `yaml:"interval"`
+	Queries  []Query `yaml:"queries"`
+	Type     string  `yaml:"type,omitempty"`
+	Start    string  `yaml:"start,omitempty"`
+	End      string  `yaml:"end,omitempty"`
+	Step     string  `yaml:"step,omitempty"`
+}
+
+func NewQuerier(groupID int, target, prNumber string, qg QueryGroup) *Querier {
+	qtype := qg.Type
+	if qtype == "" {
+		qtype = "instant"
+	}
+
+	start := durationSeconds(qg.Start)
+	end := durationSeconds(qg.End)
+
+	url := fmt.Sprintf("http://%s/%s/prometheus-%s/api/v1/query", domainName, prNumber, target)
+	if qtype == "range" {
+		url = fmt.Sprintf("http://%s/%s/prometheus-%s/api/v1/query_range", domainName, prNumber, target)
+	}
+
+	return &Querier{
+		target:   target,
+		name:     qg.Name,
+		groupID:  groupID,
+		interval: durationSeconds(qg.Interval),
+		queries:  qg.Queries,
+		qtype:    qtype,
+		start:    start,
+		end:      end,
+		step:     qg.Step,
+		url:      url,
+	}
+}
+
+func (q *Querier) run(wg *sync.WaitGroup) {
+	defer wg.Done()
+	fmt.Printf("Running querier %s %s for %s\n", q.target, q.name, q.url)
+	time.Sleep(20 * time.Second)
+
+	for {
+		start := time.Now()
+
+		for _, query := range q.queries {
+			q.query(query.Expr)
+		}
+
+		wait := q.interval - time.Since(start)
+		if wait > 0 {
+			time.Sleep(wait)
+		}
+	}
+}
+
+func (q *Querier) query(expr string) {
+	queryCount.WithLabelValues(q.target, q.name, expr, q.qtype).Inc()
+	start := time.Now()
+
+	req, err := http.NewRequest("GET", q.url, nil)
+	if err != nil {
+		log.Printf("Error creating request: %v", err)
+		queryFailCount.WithLabelValues(q.target, q.name, expr, q.qtype).Inc()
+		return
+	}
+
+	qParams := req.URL.Query()
+	qParams.Set("query", expr)
+	if q.qtype == "range" {
+		qParams.Set("start", fmt.Sprintf("%d", int64(time.Now().Add(-q.start).Unix())))
+		qParams.Set("end", fmt.Sprintf("%d", int64(time.Now().Add(-q.end).Unix())))
+		qParams.Set("step", q.step)
+	}
+	req.URL.RawQuery = qParams.Encode()
+
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		log.Printf("Error querying Prometheus: %v", err)
+		queryFailCount.WithLabelValues(q.target, q.name, expr, q.qtype).Inc()
+		return
+	}
+	defer resp.Body.Close()
+
+	duration := time.Since(start)
+	queryDuration.WithLabelValues(q.target, q.name, expr, q.qtype).Observe(duration.Seconds())
+
+	if resp.StatusCode == http.StatusNotFound {
+		log.Printf("WARNING: GroupID#%d: Querier returned 404 for Prometheus instance %s.", q.groupID, q.url)
+		q.numberOfErrors++
+		if q.numberOfErrors >= max404Errors {
+			log.Fatalf("ERROR: GroupID#%d: Querier returned 404 for Prometheus instance %s %d times.", q.groupID, q.url, max404Errors)
+		}
+	} else if resp.StatusCode != http.StatusOK {
+		log.Printf("WARNING: GroupID#%d: Querier returned %d for Prometheus instance %s.", q.groupID, resp.StatusCode, q.url)
+	} else {
+		body, _ := io.ReadAll(resp.Body)
+		log.Printf("GroupID#%d: query %s %s, status=%d, size=%d, duration=%.3f", q.groupID, q.target, expr, resp.StatusCode, len(body), duration.Seconds())
+	}
+}
+
+func durationSeconds(s string) time.Duration {
+	num := s[:len(s)-1]
+	value, err := time.ParseDuration(num + string(s[len(s)-1]))
+	if err != nil {
+		log.Fatalf("Invalid duration: %s", s)
+	}
+	return value
+}
+
+func main() {
+	if len(os.Args) < 3 {
+		fmt.Println("unexpected arguments")
+		fmt.Println("usage: <load_generator> <namespace> <pr_number>")
+		os.Exit(2)
+	}
+
+	namespace = os.Args[1]
+	prNumber := os.Args[2]
+
+	configFile, err := os.ReadFile("/etc/loadgen/config.yaml")
+	if err != nil {
+		log.Fatalf("Failed to load config: %v", err)
+	}
+	if err != nil {
+		log.Fatalf("Failed to load config: %v", err)
+	}
+
+	var config struct {
+		Querier struct {
+			Groups []QueryGroup `yaml:"groups"`
+		} `yaml:"querier"`
+	}
+	if err := yaml.Unmarshal(configFile, &config); err != nil {
+		log.Fatalf("Failed to parse config: %v", err)
+	}
+
+	fmt.Println("Loaded configuration")
+
+	var wg sync.WaitGroup
+
+	for i, group := range config.Querier.Groups {
+		wg.Add(1)
+		go NewQuerier(i, "pr", prNumber, group).run(&wg)
+	}
+
+	for i, group := range config.Querier.Groups {
+		wg.Add(1)
+		go NewQuerier(i, "release", prNumber, group).run(&wg)
+	}
+
+	prometheus.MustRegister(queryDuration, queryCount, queryFailCount)
+	http.Handle("/metrics", promhttp.Handler())
+	go func() {
+		log.Println("Starting HTTP server on :8080")
+		log.Fatal(http.ListenAndServe(":8080", nil))
+	}()
+
+	wg.Wait()
+}