add metric to track sink info

Author: YaroShkvorets

cmd/substreams-sink-sql/run.go

@@ -110,7 +110,7 @@ func sinkRunE(cmd *cobra.Command, args []string) error {
 		return fmt.Errorf("new db loader: %w", err)
 	}
 
-	postgresSinker, err := sinker.New(sink, dbLoader, zlog, tracer)
+	postgresSinker, err := sinker.New(sink, dbLoader, zlog, tracer, manifestPath)
 	if err != nil {
 		return fmt.Errorf("unable to setup postgres sinker: %w", err)
 	}

db/db.go

@@ -35,6 +35,7 @@ type Loader struct {
 
 	database     string
 	schema       string
+	host         string
 	entries      *OrderedMap[string, *OrderedMap[string, *Operation]]
 	entriesCount uint64
 	tables       map[string]*TableInfo
@@ -78,6 +79,7 @@ func NewLoader(
 		DB:                       db,
 		database:                 dsn.database,
 		schema:                   dsn.schema,
+		host:                     dsn.host,
 		entries:                  NewOrderedMap[string, *OrderedMap[string, *Operation]](),
 		tables:                   map[string]*TableInfo{},
 		batchBlockFlushInterval:  batchBlockFlushInterval,
@@ -297,6 +299,10 @@ func (l *Loader) GetSchema() string {
 	return l.schema
 }
 
+func (l *Loader) GetDatabaseHost() string {
+	return l.host
+}
+
 func (l *Loader) HasTable(tableName string) bool {
 	if _, found := l.tables[tableName]; found {
 		return true

sinker/metrics.go

@@ -16,3 +16,16 @@ var FlushDuration = metrics.NewCounter("substreams_sink_postgres_store_flush_dur
 
 var HeadBlockNumber = metrics.NewHeadBlockNumber("substreams_sink_postgres")
 var HeadBlockTimeDrift = metrics.NewHeadTimeDrift("substreams_sink_postgres")
+
+var SinkInfo = metrics.NewGaugeVec("substreams_sink_sql_info", []string{
+	"endpoint",           // Substreams endpoint
+	"database",           // Database name
+	"schema",             // Schema name
+	"db_host",            // Database host
+	"manifest",           // Manifest path or URL
+	"output_module",      // Output module name
+	"module_hash",        // Output module hash
+	"block_start",        // Start block
+	"block_end",          // End block
+	"block_restarted_at", // Restarted at block
+}, "Information about the SQL sink configuration")

sinker/sinker.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"strconv"
 	"strings"
 	"time"
 
@@ -29,9 +30,10 @@ type SQLSinker struct {
 
 	stats               *Stats
 	lastAppliedBlockNum *uint64
+	manifestPath        string // Store manifest path for metrics
 }
 
-func New(sink *sink.Sinker, loader *db.Loader, logger *zap.Logger, tracer logging.Tracer) (*SQLSinker, error) {
+func New(sink *sink.Sinker, loader *db.Loader, logger *zap.Logger, tracer logging.Tracer, manifestPath string) (*SQLSinker, error) {
 	return &SQLSinker{
 		Shutter: shutter.New(),
 		Sinker:  sink,
@@ -42,6 +44,7 @@ func New(sink *sink.Sinker, loader *db.Loader, logger *zap.Logger, tracer loggin
 
 		stats:               NewStats(logger),
 		lastAppliedBlockNum: nil,
+		manifestPath:        manifestPath,
 	}, nil
 }
 
@@ -90,6 +93,24 @@ func (s *SQLSinker) Run(ctx context.Context) {
 		zap.String("database", s.loader.GetDatabase()),
 		zap.String("schema", s.loader.GetSchema()),
 	)
+
+	endpoint, _, _ := s.EndpointConfig()
+	endBlockStr := "open"
+	if endBlock := s.BlockRange().EndBlock(); endBlock != nil {
+		endBlockStr = strconv.FormatUint(*endBlock, 10)
+	}
+
+	SinkInfo.SetInt64(1,
+		endpoint,
+		s.loader.GetDatabase(),
+		s.loader.GetSchema(),
+		s.loader.GetDatabaseHost(),
+		s.manifestPath,
+		s.OutputModuleName(),
+		s.OutputModuleHash(),
+		strconv.FormatUint(s.BlockRange().StartBlock(), 10),
+		endBlockStr,
+		strconv.FormatUint(cursor.Block().Num(), 10))
 	s.Sinker.Run(ctx, cursor, s)
 }
 

sinker/sinker_test.go

@@ -216,7 +216,7 @@ func TestInserts(t *testing.T) {
 			)
 			s, err := sink.New(sink.SubstreamsModeDevelopment, false, testPackage, testPackage.Modules.Modules[0], []byte("unused"), testClientConfig, logger, nil)
 			require.NoError(t, err)
-			sinker, _ := New(s, l, logger, nil)
+			sinker, _ := New(s, l, logger, nil, "")
 
 			for _, evt := range test.events {
 				if evt.undoSignal {