grafana metrics for eh

flow/connectors/eventhub/eventhub.go

@@ -14,6 +14,7 @@ import (
 	"github.com/Azure/azure-sdk-for-go/sdk/azidentity"
 	"github.com/Azure/azure-sdk-for-go/sdk/resourcemanager/eventhub/armeventhub"
 	"github.com/PeerDB-io/peer-flow/connectors/utils"
+	"github.com/PeerDB-io/peer-flow/connectors/utils/metrics"
 	"github.com/PeerDB-io/peer-flow/generated/protos"
 	"github.com/PeerDB-io/peer-flow/model"
 	log "github.com/sirupsen/logrus"
@@ -125,6 +126,7 @@ func (c *EventHubConnector) SyncRecords(req *model.SyncRecordsRequest) (*model.S
 	}
 
 	batchPerTopic := make(map[string][]*eventhub.Event)
+	startTime := time.Now()
 	for i, record := range batch.Records {
 		json, err := record.GetItems().ToJSON()
 		if err != nil {
@@ -166,17 +168,20 @@ func (c *EventHubConnector) SyncRecords(req *model.SyncRecordsRequest) (*model.S
 			return nil, err
 		}
 	}
-
+	rowsSynced := len(batch.Records)
 	log.WithFields(log.Fields{
 		"flowName": req.FlowJobName,
-	}).Infof("[total] successfully sent %d records to event hub", len(batch.Records))
+	}).Infof("[total] successfully sent %d records to event hub", rowsSynced)
 
 	err := c.UpdateLastOffset(req.FlowJobName, batch.LastCheckPointID)
 	if err != nil {
 		log.Errorf("failed to update last offset: %v", err)
 		return nil, err
 	}
 
+	metrics.LogSyncMetrics(c.ctx, req.FlowJobName, int64(rowsSynced), time.Since(startTime))
+	metrics.LogNormalizeMetrics(c.ctx, req.FlowJobName, int64(rowsSynced),
+		time.Since(startTime), int64(rowsSynced))
 	return &model.SyncResponse{
 		FirstSyncedCheckPointID: batch.FirstCheckPointID,
 		LastSyncedCheckPointID:  batch.LastCheckPointID,