feat: add prometheus operations metrics by index name

Author: 3n0ugh

connector.go

@@ -3,6 +3,7 @@ package cdc
 import (
 	"context"
 	"fmt"
+
 	"github.com/Trendyol/go-pq-cdc/pq/timescaledb"
 
 	cdc "github.com/Trendyol/go-pq-cdc"
@@ -60,7 +61,9 @@ func NewConnector(ctx context.Context, cfg config.Config, handler Handler, optio
 	esConnector.bulk, err = bulk.NewBulk(
 		esConnector.cfg,
 		esClient,
-		bulk.WithResponseHandler(esConnector.responseHandler))
+		pqCDC,
+		bulk.WithResponseHandler(esConnector.responseHandler),
+	)
 	if err != nil {
 		return nil, errors.Wrap(err, "elasticsearch new bulk")
 	}
@@ -101,6 +104,9 @@ func (c *connector) listener(ctx *replication.ListenerContext) {
 	}
 
 	if !c.processMessage(msg) {
+		if err := ctx.Ack(); err != nil {
+			logger.Error("ack", "error", err)
+		}
 		return
 	}
 

elasticsearch/bulk/bulk.go

@@ -8,6 +8,7 @@ import (
 	"sync"
 	"time"
 
+	cdc "github.com/Trendyol/go-pq-cdc"
 	"github.com/Trendyol/go-pq-cdc/pq/timescaledb"
 	"github.com/elastic/go-elasticsearch/v7"
 
@@ -59,6 +60,7 @@ type Bulk struct {
 	batchByteSize       int
 	concurrentRequest   int
 	flushLock           sync.Mutex
+	pqCDC               *cdc.Connector
 }
 
 type BatchItem struct {
@@ -69,6 +71,7 @@ type BatchItem struct {
 func NewBulk(
 	config *config.Config,
 	esClient *elasticsearch.Client,
+	pqCDC cdc.Connector,
 	options ...Option,
 ) (*Bulk, error) {
 	readers := make([]*bytes.MultiDimensionReader, config.Elasticsearch.ConcurrentRequest)
@@ -88,7 +91,7 @@ func NewBulk(
 		batchByteSizeLimit:  int(batchByteSizeLimit),
 		isClosed:            make(chan bool, 1),
 		esClient:            esClient,
-		metric:              NewMetric(config.CDC.Slot.Name),
+		metric:              NewMetric(pqCDC, config.CDC.Slot.Name),
 		indexMapping:        config.Elasticsearch.TableIndexMapping,
 		config:              config,
 		typeName:            []byte(config.Elasticsearch.TypeName),
@@ -132,6 +135,8 @@ func (b *Bulk) AddActions(
 			b.typeName,
 		)
 
+		b.metric.incrementOp(action.Type, indexName)
+
 		key := getActionKey(actions[i])
 		if batchIndex, ok := b.batchKeys[key]; ok {
 			b.batchByteSize += len(value) - len(b.batch[batchIndex].Bytes)

elasticsearch/bulk/metric.go

@@ -3,6 +3,8 @@ package bulk
 import (
 	"os"
 
+	cdc "github.com/Trendyol/go-pq-cdc"
+	"github.com/Trendyol/go-pq-cdc-elasticsearch/elasticsearch"
 	"github.com/prometheus/client_golang/prometheus"
 )
 
@@ -12,16 +14,26 @@ type Metric interface {
 	SetProcessLatency(latency int64)
 	SetBulkRequestProcessLatency(latency int64)
 	PrometheusCollectors() []prometheus.Collector
+	incrementOp(opType elasticsearch.ActionType, index string)
 }
 
+var hostname, _ = os.Hostname()
+
 type metric struct {
+	slotName string
+	pqCDC    cdc.Connector
+
 	processLatencyMs            prometheus.Gauge
 	bulkRequestProcessLatencyMs prometheus.Gauge
+
+	totalIndex  map[string]prometheus.Counter
+	totalDelete map[string]prometheus.Counter
 }
 
-func NewMetric(slotName string) Metric {
-	hostname, _ := os.Hostname()
+func NewMetric(pqCDC cdc.Connector, slotName string) Metric {
 	return &metric{
+		slotName: slotName,
+		pqCDC:    pqCDC,
 		processLatencyMs: prometheus.NewGauge(prometheus.GaugeOpts{
 			Namespace: namespace,
 			Subsystem: "process_latency",
@@ -42,6 +54,8 @@ func NewMetric(slotName string) Metric {
 				"slot_name": slotName,
 			},
 		}),
+		totalIndex:  make(map[string]prometheus.Counter),
+		totalDelete: make(map[string]prometheus.Counter),
 	}
 }
 
@@ -59,3 +73,42 @@ func (m *metric) SetProcessLatency(latency int64) {
 func (m *metric) SetBulkRequestProcessLatency(latency int64) {
 	m.bulkRequestProcessLatencyMs.Set(float64(latency))
 }
+
+func (m *metric) incrementOp(opType elasticsearch.ActionType, index string) {
+	switch opType {
+	case elasticsearch.Index:
+		if _, exists := m.totalIndex[index]; !exists {
+			m.totalIndex[index] = prometheus.NewCounter(prometheus.CounterOpts{
+				Namespace: namespace,
+				Subsystem: "index",
+				Name:      "total",
+				Help:      "total number of index operation in elasticsearch",
+				ConstLabels: prometheus.Labels{
+					"slot_name":  m.slotName,
+					"index_name": index,
+					"host":       hostname,
+				},
+			})
+			m.pqCDC.SetMetricCollectors(m.totalIndex[index])
+		}
+
+		m.totalIndex[index].Add(1)
+	case elasticsearch.Delete:
+		if _, exists := m.totalDelete[index]; !exists {
+			m.totalDelete[index] = prometheus.NewCounter(prometheus.CounterOpts{
+				Namespace: namespace,
+				Subsystem: "delete",
+				Name:      "total",
+				Help:      "total number of delete operation in elasticsearch",
+				ConstLabels: prometheus.Labels{
+					"slot_name":  m.slotName,
+					"index_name": index,
+					"host":       hostname,
+				},
+			})
+			m.pqCDC.SetMetricCollectors(m.totalDelete[index])
+		}
+
+		m.totalDelete[index].Add(1)
+	}
+}

example/simple/main.go

@@ -4,16 +4,17 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"log/slog"
+	"os"
+	"strconv"
+	"time"
+
 	cdc "github.com/Trendyol/go-pq-cdc-elasticsearch"
 	"github.com/Trendyol/go-pq-cdc-elasticsearch/config"
 	"github.com/Trendyol/go-pq-cdc-elasticsearch/elasticsearch"
 	cdcconfig "github.com/Trendyol/go-pq-cdc/config"
 	"github.com/Trendyol/go-pq-cdc/pq/publication"
 	"github.com/Trendyol/go-pq-cdc/pq/slot"
-	"log/slog"
-	"os"
-	"strconv"
-	"time"
 )
 
 /*
@@ -25,10 +26,21 @@ import (
 	 created_on timestamptz
 	);
 
+	CREATE TABLE books (
+	 id serial PRIMARY KEY,
+	 name text NOT NULL,
+	 created_on timestamptz
+	);
+
 	INSERT INTO users (name)
 	SELECT
 		'Oyleli' || i
 	FROM generate_series(1, 100) AS i;
+
+	INSERT INTO books (name)
+	SELECT
+		'Oyleli' || i
+	FROM generate_series(1, 100) AS i;
 */
 
 func main() {
@@ -50,10 +62,16 @@ func main() {
 					publication.OperationTruncate,
 					publication.OperationUpdate,
 				},
-				Tables: publication.Tables{publication.Table{
-					Name:            "users",
-					ReplicaIdentity: publication.ReplicaIdentityFull,
-				}},
+				Tables: publication.Tables{
+					publication.Table{
+						Name:            "users",
+						ReplicaIdentity: publication.ReplicaIdentityFull,
+					},
+					publication.Table{
+						Name:            "books",
+						ReplicaIdentity: publication.ReplicaIdentityFull,
+					},
+				},
 			},
 			Slot: slot.Config{
 				CreateIfNotExists:           true,
@@ -69,6 +87,7 @@ func main() {
 			BatchTickerDuration: time.Millisecond * 100,
 			TableIndexMapping: map[string]string{
 				"public.users": "users",
+				"public.books": "books",
 			},
 			TypeName: "_doc",
 			URLs:     []string{"http://127.0.0.1:9200"},