perf(promproxy): implement query recording

* Add query recording schema
* Add promproxy to ch-bench-read

Author: ernado

.dockerignore

@@ -2,4 +2,7 @@
 /.github
 /docs
 *.md
-*.Dockerfile
+*.Dockerfile
+
+# remore write queries
+*.rwq

cmd/otelproxy/main.go

@@ -30,9 +30,17 @@ type service struct {
 	name      string
 	handler   http.Handler
 	findRoute httpmiddleware.RouteFinder
+	cleanup   func() error
 }
 
 func (s service) Run(ctx context.Context, lg *zap.Logger, m *app.Metrics) error {
+	if s.cleanup != nil {
+		defer func() {
+			if err := s.cleanup(); err != nil {
+				lg.Error("Cleanup failed", zap.Error(err))
+			}
+		}()
+	}
 	httpServer := &http.Server{
 		Addr:              s.addr,
 		Handler:           ServiceMiddleware(s, lg, m),
@@ -108,9 +116,25 @@ func (s *services) Prometheus(m *app.Metrics) error {
 	if err != nil {
 		return errors.Wrap(err, "create client")
 	}
-
+	var (
+		rec     *promproxy.Recorder
+		cleanup func() error
+	)
+	if fName := os.Getenv(prefix + "_RECORD"); fName != "" {
+		f, err := os.Create(fName) // #nosec G304
+		if err != nil {
+			return errors.Wrap(err, "create record file")
+		}
+		rec = promproxy.NewRecorder(f)
+		cleanup = func() error {
+			if err := f.Close(); err != nil {
+				return errors.Wrap(err, "close record file")
+			}
+			return nil
+		}
+	}
 	server, err := promapi.NewServer(
-		promproxy.NewServer(client),
+		promproxy.NewServer(client, rec),
 		promapi.WithTracerProvider(m.TracerProvider()),
 		promapi.WithMeterProvider(m.MeterProvider()),
 	)
@@ -127,6 +151,7 @@ func (s *services) Prometheus(m *app.Metrics) error {
 		name:      strings.ToLower(prefix),
 		handler:   server,
 		findRoute: httpmiddleware.MakeRouteFinder[promapi.Route](server),
+		cleanup:   cleanup,
 	})
 }
 

dev/local/ch-bench-read/.dockerignore

@@ -0,0 +1 @@
+record

dev/local/ch-bench-read/.gitignore

@@ -0,0 +1 @@
+record

dev/local/ch-bench-read/docker-compose.yml

@@ -24,7 +24,7 @@ services:
     environment:
       - OTELDB_STORAGE=ch
       - CH_DSN=clickhouse://clickhouse:9000
-      - OTEL_LOG_LEVEL=debug
+      - OTEL_LOG_LEVEL=info
       - OTEL_METRICS_EXPORTER=none
       - OTEL_LOGS_EXPORTER=none
       - OTEL_TRACES_EXPORTER=none
@@ -40,6 +40,21 @@ services:
     depends_on:
       - clickhouse
 
+  otelproxy:
+    build:
+      context: ../../../
+      dockerfile: otelproxy.Dockerfile
+    volumes:
+      - ./record:/record
+    environment:
+      - PROMETHEUS_URL=http://oteldb:9090
+      - PROMETHEUS_RECORD=/record/queries.jsonl
+      - OTEL_LOG_LEVEL=info
+      - OTEL_METRICS_EXPORTER=none
+      - OTEL_LOGS_EXPORTER=none
+      - OTEL_TRACES_EXPORTER=none
+      - OTEL_RESOURCE_ATTRIBUTES=service.name=go-faster.otelproxy
+
   grafana:
     image: "grafana/grafana:10.0.0"
     environment:

dev/local/ch-bench-read/grafana/dashboards/node-exporter.json

@@ -23636,7 +23636,7 @@
       "type": "row"
     }
   ],
-  "refresh": "5s",
+  "refresh": false,
   "revision": 1,
   "schemaVersion": 38,
   "style": "dark",
@@ -23648,8 +23648,8 @@
       {
         "current": {
           "selected": false,
-          "text": "Prometheus",
-          "value": "Prometheus"
+          "text": "PromQL",
+          "value": "PromQL"
         },
         "hide": 0,
         "includeAll": false,
@@ -23667,8 +23667,8 @@
       {
         "current": {
           "selected": false,
-          "text": "node-exporter",
-          "value": "node-exporter"
+          "text": "node_exporter",
+          "value": "node_exporter"
         },
         "datasource": {
           "type": "prometheus",
@@ -23685,7 +23685,7 @@
           "query": "label_values(node_uname_info, job)",
           "refId": "Prometheus-job-Variable-Query"
         },
-        "refresh": 1,
+        "refresh": 2,
         "regex": "",
         "skipUrlSync": false,
         "sort": 1,
@@ -23697,8 +23697,8 @@
       {
         "current": {
           "selected": false,
-          "text": "node-exporter:9100",
-          "value": "node-exporter:9100"
+          "text": "host-0",
+          "value": "host-0"
         },
         "datasource": {
           "type": "prometheus",
@@ -23715,7 +23715,7 @@
           "query": "label_values(node_uname_info{job=\"$job\"}, instance)",
           "refId": "Prometheus-node-Variable-Query"
         },
-        "refresh": 1,
+        "refresh": 2,
         "regex": "",
         "skipUrlSync": false,
         "sort": 1,
@@ -23748,8 +23748,8 @@
     ]
   },
   "time": {
-    "from": "now-5m",
-    "to": "now"
+    "from": "2024-01-06T14:05:29.481Z",
+    "to": "2024-01-06T15:01:09.313Z"
   },
   "timepicker": {
     "refresh_intervals": [
@@ -23770,6 +23770,6 @@
   "timezone": "browser",
   "title": "Node Exporter Full",
   "uid": "node-exporter",
-  "version": 2,
+  "version": 1,
   "weekStart": ""
-}
+}

dev/local/ch-bench-read/grafana/datasources.yml

@@ -16,7 +16,7 @@ datasources:
     type: prometheus
     access: proxy
     orgId: 1
-    url: http://oteldb:9090
+    url: http://otelproxy:9090
     uid: promoteldb
     isDefault: true
 

internal/promproxy/generate.go

@@ -0,0 +1,3 @@
+package promproxy
+
+//go:generate go run github.com/ogen-go/ogen/cmd/jschemagen -typename Record --target record_gen.go --package promproxy schema.yml

internal/promproxy/promproxy.go

@@ -15,15 +15,17 @@ import (
 var _ promapi.Handler = &Server{}
 
 // NewServer initializes new proxy Server from openapi client.
-func NewServer(api *promapi.Client) *Server {
+func NewServer(api *promapi.Client, rec *Recorder) *Server {
 	return &Server{
 		api: api,
+		rec: rec,
 	}
 }
 
 // Server implement proxy server.
 type Server struct {
 	api *promapi.Client
+	rec *Recorder
 }
 
 // GetLabelValues implements getLabelValues operation.
@@ -70,6 +72,9 @@ func (s *Server) GetQueryExemplars(ctx context.Context, params promapi.GetQueryE
 //
 // GET /api/v1/query_range
 func (s *Server) GetQueryRange(ctx context.Context, params promapi.GetQueryRangeParams) (*promapi.QueryResponse, error) {
+	if err := s.rec.RecordGetQueryRange(params); err != nil {
+		return nil, errors.Wrap(err, "record")
+	}
 	return s.api.GetQueryRange(ctx, params)
 }
 
@@ -85,6 +90,9 @@ func (s *Server) GetRules(ctx context.Context, params promapi.GetRulesParams) (*
 //
 // GET /api/v1/series
 func (s *Server) GetSeries(ctx context.Context, params promapi.GetSeriesParams) (*promapi.SeriesResponse, error) {
+	if err := s.rec.RecordSeriesQuery(params); err != nil {
+		return nil, errors.Wrap(err, "record")
+	}
 	return s.api.GetSeries(ctx, params)
 }
 

internal/promproxy/record.go

@@ -0,0 +1,93 @@
+package promproxy
+
+import (
+	"encoding/json"
+	"io"
+	"strconv"
+	"time"
+
+	"github.com/go-faster/errors"
+
+	"github.com/go-faster/oteldb/internal/promapi"
+)
+
+// Recorder of prometheus queries.
+type Recorder struct {
+	encoder *json.Encoder
+}
+
+// NewRecorder returns new Recorder.
+func NewRecorder(writer io.Writer) *Recorder {
+	e := json.NewEncoder(writer)
+	return &Recorder{encoder: e}
+}
+
+func (r *Recorder) encode(v any) error {
+	if r == nil {
+		return nil
+	}
+	var q Query
+	switch t := v.(type) {
+	case RangeQuery:
+		q.SetRangeQuery(t)
+	case InstantQuery:
+		q.SetInstantQuery(t)
+	case SeriesQuery:
+		q.SetSeriesQuery(t)
+	default:
+		return errors.Errorf("unknown type %T", t)
+	}
+	if err := r.encoder.Encode(q); err != nil {
+		return errors.Wrap(err, "encode")
+	}
+	return nil
+}
+
+func ts(v promapi.PrometheusTimestamp) time.Time {
+	seconds, err := strconv.ParseInt(string(v), 10, 64)
+	if err != nil {
+		return time.Time{}
+	}
+	return time.Unix(seconds, 0)
+}
+
+func step(v string) int {
+	seconds, err := strconv.Atoi(v)
+	if err != nil {
+		return 0
+	}
+	return seconds
+}
+
+func optTS(v promapi.OptPrometheusTimestamp) OptDateTime {
+	var dt OptDateTime
+	if t, ok := v.Get(); ok {
+		dt.SetTo(ts(t))
+	}
+	return dt
+}
+
+func (r *Recorder) RecordSeriesQuery(v promapi.GetSeriesParams) error {
+	return r.encode(SeriesQuery{
+		Start:    optTS(v.Start),
+		End:      optTS(v.End),
+		Matchers: v.Match,
+	})
+}
+
+func (r *Recorder) RecordGetQuery(v promapi.GetQueryParams) error {
+	return r.encode(InstantQuery{
+		Query: v.Query,
+		Time:  optTS(v.Time),
+	})
+}
+
+// RecordGetQueryRange records range query.
+func (r *Recorder) RecordGetQueryRange(v promapi.GetQueryRangeParams) error {
+	return r.encode(RangeQuery{
+		Query: v.Query,
+		Start: ts(v.Start),
+		End:   ts(v.End),
+		Step:  step(v.Step),
+	})
+}