Implement ingested_spans_count in telemetry data.

Signed-off-by: Harkishen-Singh <[email protected]>

Author: harkishen

CHANGELOG.md

@@ -16,6 +16,7 @@ We use the following categories for changes:
 
 ### Added
 - Add Prometheus metrics support for Tracing [#1102]
+- Add ingested spans-count to telemetry [#1155]
 
 ### Fixed
 - Fix spans with end < start. Start and end are swapped in this case. [#1096]

pkg/migrations/migration_files_generated.go

@@ -36,6 +36,7 @@ $$
         )
         UPDATE _ps_catalog.promscale_instance_information SET
             promscale_ingested_samples_total                    = promscale_ingested_samples_total + COALESCE(x.del_promscale_ingested_samples_total, 0),
+            promscale_ingested_spans_total                      = promscale_ingested_spans_total + COALESCE(x.del_promscale_ingested_spans_total, 0),
             promscale_metrics_queries_success_total             = promscale_metrics_queries_success_total + COALESCE(x.del_promscale_metrics_queries_success_total, 0),
             promscale_metrics_queries_timedout_total            = promscale_metrics_queries_timedout_total + COALESCE(x.del_promscale_metrics_queries_timedout_total, 0),
             promscale_metrics_queries_failed_total              = promscale_metrics_queries_failed_total + COALESCE(x.del_promscale_metrics_queries_failed_total, 0),
@@ -46,6 +47,7 @@ $$
         (
             SELECT
                 sum(promscale_ingested_samples_total)  			        as del_promscale_ingested_samples_total,
+                sum(promscale_ingested_spans_total)  			        as del_promscale_ingested_spans_total,
                 sum(promscale_metrics_queries_success_total)  		    as del_promscale_metrics_queries_success_total,
                 sum(promscale_metrics_queries_timedout_total)  		    as del_promscale_metrics_queries_timedout_total,
                 sum(promscale_metrics_queries_failed_total)  		    as del_promscale_metrics_queries_failed_total,

pkg/migrations/sql/idempotent/telemetry.sql

@@ -2,7 +2,8 @@ CREATE TABLE IF NOT EXISTS _ps_catalog.promscale_instance_information (
     uuid                                                UUID NOT NULL PRIMARY KEY,
     last_updated                                        TIMESTAMPTZ NOT NULL,
     promscale_ingested_samples_total                    BIGINT DEFAULT 0 NOT NULL,
-    promscale_metrics_queries_success_total            BIGINT DEFAULT 0 NOT NULL,
+    promscale_ingested_spans_total                      BIGINT DEFAULT 0 NOT NULL,
+    promscale_metrics_queries_success_total             BIGINT DEFAULT 0 NOT NULL,
     promscale_metrics_queries_timedout_total            BIGINT DEFAULT 0 NOT NULL,
     promscale_metrics_queries_failed_total              BIGINT DEFAULT 0 NOT NULL,
     promscale_trace_query_requests_executed_total       BIGINT DEFAULT 0 NOT NULL,

pkg/migrations/sql/preinstall/012-telemetry.sql

@@ -0,0 +1,2 @@
+ALTER TABLE _ps_catalog.promscale_instance_information
+  ADD COLUMN  promscale_ingested_spans_total BIGINT NOT NULL DEFAULT 0 ;

pkg/migrations/sql/versions/dev/0.8.1-dev/2-alter_promscale_instance_information_column.sql renamed to pkg/migrations/sql/versions/dev/0.10.0-dev/1-alter_promscale_instance_information_column.sql

@@ -9,6 +9,7 @@ import (
 	"fmt"
 	"net/url"
 
+	"github.com/google/uuid"
 	"github.com/jackc/pgx/v4"
 	"github.com/jackc/pgx/v4/pgxpool"
 	"github.com/timescale/promscale/pkg/ha"
@@ -22,10 +23,22 @@ import (
 	"github.com/timescale/promscale/pkg/prompb"
 	"github.com/timescale/promscale/pkg/promql"
 	"github.com/timescale/promscale/pkg/query"
+	"github.com/timescale/promscale/pkg/telemetry"
 	"github.com/timescale/promscale/pkg/tenancy"
 	"go.opentelemetry.io/collector/model/pdata"
 )
 
+var PromscaleID uuid.UUID
+
+func init() {
+	// PromscaleID must always be generated on start, so that it remains constant throughout the lifecycle.
+	PromscaleID = uuid.New()
+}
+
+// Post connect validation function, useful for things such as acquiring locks
+// that should live the duration of the connection
+type LockFunc = func(ctx context.Context, conn *pgx.Conn) error
+
 // Client sends Prometheus samples to TimescaleDB
 type Client struct {
 	Connection        pgxconn.PgxConn
@@ -41,12 +54,9 @@ type Client struct {
 	closePool         bool
 	sigClose          chan struct{}
 	haService         *ha.Service
+	TelemetryEngine   telemetry.Engine
 }
 
-// Post connect validation function, useful for things such as acquiring locks
-// that should live the duration of the connection
-type LockFunc = func(ctx context.Context, conn *pgx.Conn) error
-
 // NewClient creates a new PostgreSQL client
 func NewClient(cfg *Config, mt tenancy.Authorizer, schemaLocker LockFunc, readOnly bool) (*Client, error) {
 	pgConfig, numCopiers, err := getPgConfig(cfg)
@@ -140,25 +150,34 @@ func NewClientWithPool(cfg *Config, numCopiers int, connPool *pgxpool.Pool, mt t
 		AsyncAcks:              cfg.AsyncAcks,
 	}
 
+	labelsReader := lreader.NewLabelsReader(dbConn, labelsCache)
+	exemplarKeyPosCache := cache.NewExemplarLabelsPosCache(cfg.CacheConfig)
+
+	dbQuerierConn := pgxconn.NewQueryLoggingPgxConn(connPool)
+	dbQuerier := querier.NewQuerier(dbQuerierConn, metricsCache, labelsReader, exemplarKeyPosCache, mt.ReadAuthorizer())
+	queryable := query.NewQueryable(dbQuerier, labelsReader)
+
 	var (
-		dbIngestor          *ingestor.DBIngestor
-		exemplarKeyPosCache = cache.NewExemplarLabelsPosCache(cfg.CacheConfig)
+		err             error
+		telemetryEngine telemetry.Engine
 	)
+	telemetryEngine, err = telemetry.NewEngine(dbConn, PromscaleID, queryable)
+	if err != nil {
+		log.Debug("msg", "err creating telemetry engine", "err", err.Error())
+		telemetryEngine = telemetry.NewNoopEngine()
+	}
+	telemetryEngine.Start() // We stop the engine at client.Close().
+
+	var dbIngestor *ingestor.DBIngestor
 	if !readOnly {
 		var err error
-		dbIngestor, err = ingestor.NewPgxIngestor(dbConn, metricsCache, seriesCache, exemplarKeyPosCache, &c)
+		dbIngestor, err = ingestor.NewPgxIngestor(dbConn, metricsCache, seriesCache, exemplarKeyPosCache, &c, telemetryEngine)
 		if err != nil {
 			log.Error("msg", "err starting the ingestor", "err", err)
 			return nil, err
 		}
 	}
 
-	labelsReader := lreader.NewLabelsReader(dbConn, labelsCache)
-
-	dbQuerierConn := pgxconn.NewQueryLoggingPgxConn(connPool)
-	dbQuerier := querier.NewQuerier(dbQuerierConn, metricsCache, labelsReader, exemplarKeyPosCache, mt.ReadAuthorizer())
-	queryable := query.NewQueryable(dbQuerier, labelsReader)
-
 	healthChecker := health.NewHealthChecker(dbConn)
 	client := &Client{
 		Connection:        dbConn,
@@ -171,6 +190,7 @@ func NewClientWithPool(cfg *Config, numCopiers int, connPool *pgxpool.Pool, mt t
 		labelsCache:       labelsCache,
 		seriesCache:       seriesCache,
 		sigClose:          sigClose,
+		TelemetryEngine:   telemetryEngine,
 	}
 
 	InitClientMetrics(client)
@@ -180,6 +200,9 @@ func NewClientWithPool(cfg *Config, numCopiers int, connPool *pgxpool.Pool, mt t
 // Close closes the client and performs cleanup
 func (c *Client) Close() {
 	log.Info("msg", "Shutting down Client")
+	if c.TelemetryEngine != nil {
+		c.TelemetryEngine.Stop()
+	}
 	if c.ingestor != nil {
 		c.ingestor.Close()
 	}

pkg/migrations/sql/versions/dev/0.10.0-dev/2-add_spans_total_column_telemetry.sql

@@ -19,6 +19,7 @@ import (
 	"github.com/timescale/promscale/pkg/pgmodel/model"
 	"github.com/timescale/promscale/pkg/pgxconn"
 	"github.com/timescale/promscale/pkg/prompb"
+	"github.com/timescale/promscale/pkg/telemetry"
 	"github.com/timescale/promscale/pkg/tracer"
 )
 
@@ -38,15 +39,15 @@ type DBIngestor struct {
 
 // NewPgxIngestor returns a new Ingestor that uses connection pool and a metrics cache
 // for caching metric table names.
-func NewPgxIngestor(conn pgxconn.PgxConn, cache cache.MetricCache, sCache cache.SeriesCache, eCache cache.PositionCache, cfg *Cfg) (*DBIngestor, error) {
+func NewPgxIngestor(conn pgxconn.PgxConn, cache cache.MetricCache, sCache cache.SeriesCache, eCache cache.PositionCache, cfg *Cfg, t telemetry.Engine) (*DBIngestor, error) {
 	dispatcher, err := newPgxDispatcher(conn, cache, sCache, eCache, cfg)
 	if err != nil {
 		return nil, err
 	}
 	return &DBIngestor{
 		sCache:     sCache,
 		dispatcher: dispatcher,
-		tWriter:    trace.NewWriter(conn),
+		tWriter:    trace.NewWriter(conn, t),
 	}, nil
 }
 
@@ -60,7 +61,7 @@ func NewPgxIngestorForTests(conn pgxconn.PgxConn, cfg *Cfg) (*DBIngestor, error)
 	c := cache.NewMetricCache(cacheConfig)
 	s := cache.NewSeriesCache(cacheConfig, nil)
 	e := cache.NewExemplarLabelsPosCache(cacheConfig)
-	return NewPgxIngestor(conn, c, s, e, cfg)
+	return NewPgxIngestor(conn, c, s, e, cfg, telemetry.NewNoopEngine())
 }
 
 const (

pkg/pgclient/client.go

@@ -17,8 +17,10 @@ import (
 
 	"github.com/jackc/pgtype"
 	"github.com/timescale/promscale/pkg/clockcache"
+	"github.com/timescale/promscale/pkg/log"
 	"github.com/timescale/promscale/pkg/pgmodel/metrics"
 	"github.com/timescale/promscale/pkg/pgxconn"
+	"github.com/timescale/promscale/pkg/telemetry"
 	tput "github.com/timescale/promscale/pkg/util/throughput"
 )
 
@@ -59,7 +61,10 @@ type traceWriterImpl struct {
 	tagCache     *clockcache.Cache
 }
 
-func NewWriter(conn pgxconn.PgxConn) *traceWriterImpl {
+func NewWriter(conn pgxconn.PgxConn, t telemetry.Engine) *traceWriterImpl {
+	if err := t.RegisterMetric("promscale_ingested_spans_total", metrics.IngestorItems.With(prometheus.Labels{"type": "trace", "kind": "span"})); err != nil {
+		log.Debug("msg", "err registering telemetry metric promscale_ingested_spans_total", "err", err.Error())
+	}
 	return &traceWriterImpl{
 		conn:         conn,
 		schemaCache:  newSchemaCache(),

pkg/pgmodel/ingestor/ingestor.go

@@ -1,6 +1,7 @@
 // This file and its contents are licensed under the Apache License 2.0.
 // Please see the included NOTICE for copyright information and
 // LICENSE for a copy of the license.
+
 package runner
 
 import (
@@ -10,7 +11,6 @@ import (
 	"strconv"
 
 	"github.com/jackc/pgx/v4"
-	"github.com/timescale/promscale/pkg/api"
 	"github.com/timescale/promscale/pkg/dataset"
 	"github.com/timescale/promscale/pkg/log"
 	"github.com/timescale/promscale/pkg/pgclient"
@@ -27,7 +27,7 @@ var (
 	migrationLockError = fmt.Errorf("Could not acquire migration lock. Ensure there are no other connectors running and try again.")
 )
 
-func CreateClient(cfg *Config, promMetrics *api.Metrics) (*pgclient.Client, error) {
+func CreateClient(cfg *Config) (*pgclient.Client, error) {
 	// The TimescaleDB migration has to happen before other connections
 	// are open, also it has to happen as the first command on a connection.
 	// Thus we cannot rely on the migration lock here. Instead we assume