Feat: enables metrics ingestion to signoz

.gitignore

@@ -36,5 +36,6 @@ frontend/cypress.env.json
 **/build
 **/storage
 **/locust-scripts/__pycache__/
+**/__debug_bin
 
 frontend/*.env

deploy/docker/clickhouse-setup/docker-compose.yaml

@@ -22,9 +22,11 @@ services:
   query-service:
     image: signoz/query-service:0.3.6
     container_name: query-service
-
+    command: ["-config=/root/config/prometheus.yml"]
     ports:
       - "8080:8080"
+    volumes:
+      - ./prometheus.yml:/root/config/prometheus.yml
 
     environment:
       - ClickHouseUrl=tcp://clickhouse:9000

deploy/docker/clickhouse-setup/otel-collector-config.yaml

@@ -7,6 +7,11 @@ receivers:
     protocols:
       grpc:
       thrift_http:
+  hostmetrics:
+    collection_interval: 10s
+    scrapers:
+      load:
+      memory:
 processors:
   batch:
     send_batch_size: 1000
@@ -29,11 +34,19 @@ extensions:
 exporters:
   clickhouse:
     datasource: tcp://clickhouse:9000
+  clickhousemetricswrite:
+    endpoint: tcp://clickhouse:9000/?database=signoz_metrics
+    resource_to_telemetry_conversion:
+      enabled: true
 
 service:
   extensions: [health_check, zpages]
   pipelines:
     traces:
       receivers: [jaeger, otlp]
       processors: [batch]
-      exporters: [clickhouse]
+      exporters: [clickhouse]
+    metrics:
+      receivers: [otlp, hostmetrics]
+      processors: [batch]
+      exporters: [clickhousemetricswrite]

deploy/docker/clickhouse-setup/prometheus.yml

@@ -0,0 +1,25 @@
+# my global config
+global:
+  scrape_interval:     5s # Set the scrape interval to every 15 seconds. Default is every 1 minute.
+  evaluation_interval: 15s # Evaluate rules every 15 seconds. The default is every 1 minute.
+  # scrape_timeout is set to the global default (10s).
+
+# Alertmanager configuration
+alerting:
+  alertmanagers:
+  - static_configs:
+    - targets:
+      # - alertmanager:9093
+
+# Load rules once and periodically evaluate them according to the global 'evaluation_interval'.
+rule_files:
+  # - "first_rules.yml"
+  # - "second_rules.yml"
+
+# A scrape configuration containing exactly one endpoint to scrape:
+# Here it's Prometheus itself.
+scrape_configs:
+
+
+remote_read:
+  - url: tcp://clickhouse:9000/?database=signoz_metrics

pkg/query-service/Dockerfile

@@ -32,6 +32,12 @@ RUN apk update && apk add ca-certificates && rm -rf /var/cache/apk/*
 WORKDIR /root
 # copy the binary from builder
 COPY --from=builder /go/src/github.com/signoz/signoz/pkg/query-service/bin/query-service .
+
+COPY config/prometheus.yml /root/config/prometheus.yml
+
 # run the binary
-CMD ["./query-service"]
+ENTRYPOINT ["./query-service"]
+CMD ["-config", "/root/config/prometheus.yml"]
+# CMD ["./query-service -config /root/config/prometheus.yml"]
+EXPOSE 8080
 

pkg/query-service/app/clickhouseReader/reader.go

@@ -3,13 +3,21 @@ package clickhouseReader
 import (
 	"context"
 	"errors"
+	"flag"
 	"fmt"
 	"os"
 	"strconv"
 	"time"
 
 	_ "github.com/ClickHouse/clickhouse-go"
+	"github.com/go-kit/log"
 	"github.com/jmoiron/sqlx"
+	promModel "github.com/prometheus/common/model"
+	"github.com/prometheus/common/promlog"
+	"github.com/prometheus/prometheus/config"
+	"github.com/prometheus/prometheus/promql"
+	"github.com/prometheus/prometheus/storage/remote"
+	"github.com/prometheus/prometheus/util/stats"
 
 	"go.signoz.io/query-service/model"
 	"go.uber.org/zap"
@@ -36,6 +44,8 @@ type ClickHouseReader struct {
 	operationsTable string
 	indexTable      string
 	spansTable      string
+	queryEngine     *promql.Engine
+	remoteStorage   *remote.Storage
 }
 
 // NewTraceReader returns a TraceReader for the database
@@ -48,11 +58,55 @@ func NewReader() *ClickHouseReader {
 	if err != nil {
 		zap.S().Error(err)
 	}
+
+	logLevel := promlog.AllowedLevel{}
+	logLevel.Set("debug")
+	// allowedFormat := promlog.AllowedFormat{}
+	// allowedFormat.Set("logfmt")
+
+	// promlogConfig := promlog.Config{
+	// 	Level:  &logLevel,
+	// 	Format: &allowedFormat,
+	// }
+
+	logger := promlog.New(logLevel)
+
+	opts := promql.EngineOpts{
+		Logger:        log.With(logger, "component", "query engine"),
+		Reg:           nil,
+		MaxConcurrent: 20,
+		MaxSamples:    50000000,
+		Timeout:       time.Duration(2 * time.Minute),
+	}
+
+	queryEngine := promql.NewEngine(opts)
+
+	startTime := func() (int64, error) {
+		return int64(promModel.Latest), nil
+
+	}
+
+	remoteStorage := remote.NewStorage(log.With(logger, "component", "remote"), startTime, time.Duration(1*time.Minute))
+
+	filename := flag.String("config", "./config/prometheus.yml", "(prometheus config to read metrics)")
+	flag.Parse()
+	conf, err := config.LoadFile(*filename)
+	if err != nil {
+		zap.S().Error("couldn't load configuration (--config.file=%q): %v", filename, err)
+	}
+
+	err = remoteStorage.ApplyConfig(conf)
+	if err != nil {
+		zap.S().Error("Error in remoteStorage.ApplyConfig: ", err)
+	}
+
 	return &ClickHouseReader{
 		db:              db,
 		operationsTable: options.primary.OperationsTable,
 		indexTable:      options.primary.IndexTable,
 		spansTable:      options.primary.SpansTable,
+		queryEngine:     queryEngine,
+		remoteStorage:   remoteStorage,
 	}
 }
 
@@ -74,6 +128,45 @@ func connect(cfg *namespaceConfig) (*sqlx.DB, error) {
 	return cfg.Connector(cfg)
 }
 
+func (r *ClickHouseReader) GetInstantQueryMetricsResult(ctx context.Context, queryParams *model.InstantQueryMetricsParams) (*promql.Result, *stats.QueryStats, *model.ApiError) {
+	qry, err := r.queryEngine.NewInstantQuery(r.remoteStorage, queryParams.Query, queryParams.Time)
+	if err != nil {
+		return nil, nil, &model.ApiError{model.ErrorBadData, err}
+	}
+
+	res := qry.Exec(ctx)
+
+	// Optional stats field in response if parameter "stats" is not empty.
+	var qs *stats.QueryStats
+	if queryParams.Stats != "" {
+		qs = stats.NewQueryStats(qry.Stats())
+	}
+
+	qry.Close()
+	return res, qs, nil
+
+}
+
+func (r *ClickHouseReader) GetQueryRangeResult(ctx context.Context, query *model.QueryRangeParams) (*promql.Result, *stats.QueryStats, *model.ApiError) {
+
+	qry, err := r.queryEngine.NewRangeQuery(r.remoteStorage, query.Query, query.Start, query.End, query.Step)
+
+	if err != nil {
+		return nil, nil, &model.ApiError{model.ErrorBadData, err}
+	}
+
+	res := qry.Exec(ctx)
+
+	// Optional stats field in response if parameter "stats" is not empty.
+	var qs *stats.QueryStats
+	if query.Stats != "" {
+		qs = stats.NewQueryStats(qry.Stats())
+	}
+
+	qry.Close()
+	return res, qs, nil
+}
+
 func (r *ClickHouseReader) GetServices(ctx context.Context, queryParams *model.GetServicesParams) (*[]model.ServiceItem, error) {
 
 	if r.indexTable == "" {

pkg/query-service/app/druidReader/reader.go

@@ -2,8 +2,11 @@ package druidReader
 
 import (
 	"context"
+	"fmt"
 	"os"
 
+	"github.com/prometheus/prometheus/promql"
+	"github.com/prometheus/prometheus/util/stats"
 	"go.signoz.io/query-service/druidQuery"
 	"go.signoz.io/query-service/godruid"
 	"go.signoz.io/query-service/model"
@@ -39,6 +42,16 @@ func initialize() {
 
 }
 
+func (druid *DruidReader) GetQueryRangeResult(ctx context.Context, query *model.QueryRangeParams) (*promql.Result, *stats.QueryStats, *model.ApiError) {
+
+	return nil, nil, &model.ApiError{model.ErrorNotImplemented, fmt.Errorf("Druid does not support metrics")}
+}
+
+func (druid *DruidReader) GetInstantQueryMetricsResult(ctx context.Context, query *model.InstantQueryMetricsParams) (*promql.Result, *stats.QueryStats, *model.ApiError) {
+
+	return nil, nil, &model.ApiError{model.ErrorNotImplemented, fmt.Errorf("Druid does not support metrics")}
+}
+
 func (druid *DruidReader) GetServiceOverview(ctx context.Context, query *model.GetServiceOverviewParams) (*[]model.ServiceOverviewItem, error) {
 	return druidQuery.GetServiceOverview(druid.SqlClient, query)
 }