add additional labels to prometheus metrics, new metrics (#100)

Signed-off-by: Amir Malka <[email protected]>

adapters/backend/v1/adapter.go

@@ -16,6 +16,7 @@ import (
 	"github.com/kubescape/synchronizer/domain"
 	"github.com/kubescape/synchronizer/messaging"
 	"github.com/kubescape/synchronizer/utils"
+	"github.com/prometheus/client_golang/prometheus"
 )
 
 type Adapter struct {
@@ -165,6 +166,10 @@ func (b *Adapter) Stop(ctx context.Context) error {
 		b.clientsMap.Delete(toDelete.String())
 	}
 	connectedClientsGauge.Dec()
+	clientDisconnectionCounter.With(prometheus.Labels{
+		prometheusClusterLabel: toDelete.Cluster,
+		prometheusAccountLabel: toDelete.Account,
+	}).Inc()
 
 	return nil
 }

adapters/backend/v1/client.go

@@ -237,7 +237,7 @@ func (c *Client) sendDeleteObjectMessage(ctx context.Context, id domain.KindName
 		return fmt.Errorf("marshal delete object message: %w", err)
 	}
 
-	return c.messageProducer.ProduceMessage(ctx, cId, messaging.MsgPropEventValueDeleteObjectMessage, data, id.ToCustomProperties())
+	return c.messageProducer.ProduceMessage(ctx, cId, messaging.MsgPropEventValueDeleteObjectMessage, data, messaging.KindNameToCustomProperties(id))
 }
 
 func (c *Client) sendGetObjectMessage(ctx context.Context, id domain.KindName, baseObject []byte) error {
@@ -267,7 +267,7 @@ func (c *Client) sendGetObjectMessage(ctx context.Context, id domain.KindName, b
 		return fmt.Errorf("marshal get object message: %w", err)
 	}
 
-	return c.messageProducer.ProduceMessage(ctx, cId, messaging.MsgPropEventValueGetObjectMessage, data, id.ToCustomProperties())
+	return c.messageProducer.ProduceMessage(ctx, cId, messaging.MsgPropEventValueGetObjectMessage, data, messaging.KindNameToCustomProperties(id))
 }
 
 func (c *Client) sendPatchObjectMessage(ctx context.Context, id domain.KindName, checksum string, patch []byte) error {
@@ -299,7 +299,7 @@ func (c *Client) sendPatchObjectMessage(ctx context.Context, id domain.KindName,
 		return fmt.Errorf("marshal patch object message: %w", err)
 	}
 
-	return c.messageProducer.ProduceMessage(ctx, cId, messaging.MsgPropEventValuePatchObjectMessage, data, id.ToCustomProperties())
+	return c.messageProducer.ProduceMessage(ctx, cId, messaging.MsgPropEventValuePatchObjectMessage, data, messaging.KindNameToCustomProperties(id))
 }
 
 func (c *Client) sendPutObjectMessage(ctx context.Context, id domain.KindName, object []byte) error {
@@ -338,7 +338,7 @@ func (c *Client) sendPutObjectMessage(ctx context.Context, id domain.KindName, o
 		return fmt.Errorf("marshal put object message: %w", err)
 	}
 
-	return c.messageProducer.ProduceMessage(ctx, cId, messaging.MsgPropEventValuePutObjectMessage, data, id.ToCustomProperties())
+	return c.messageProducer.ProduceMessage(ctx, cId, messaging.MsgPropEventValuePutObjectMessage, data, messaging.KindNameToCustomProperties(id))
 }
 
 func (c *Client) sendVerifyObjectMessage(ctx context.Context, id domain.KindName, checksum string) error {
@@ -368,7 +368,7 @@ func (c *Client) sendVerifyObjectMessage(ctx context.Context, id domain.KindName
 		return fmt.Errorf("marshal verify object message: %w", err)
 	}
 
-	return c.messageProducer.ProduceMessage(ctx, cId, messaging.MsgPropEventValueVerifyObjectMessage, data, id.ToCustomProperties())
+	return c.messageProducer.ProduceMessage(ctx, cId, messaging.MsgPropEventValueVerifyObjectMessage, data, messaging.KindNameToCustomProperties(id))
 }
 
 func (c *Client) SendReconciliationRequestMessage(ctx context.Context) error {

adapters/backend/v1/metrics.go

@@ -6,7 +6,11 @@ import (
 )
 
 const (
-	prometheusStatusLabel = "status"
+	prometheusStatusLabel    = "status"
+	prometheusKindLabel      = "kind"
+	prometheusAccountLabel   = "account"
+	prometheusClusterLabel   = "cluster"
+	prometheusEventTypeLabel = "event_type"
 
 	prometheusStatusLabelValueSuccess = "success"
 	prometheusStatusLabelValueError   = "error"
@@ -17,14 +21,18 @@ var (
 		Name: "synchronizer_connected_clients_count",
 		Help: "The number of connected clients",
 	})
+	clientDisconnectionCounter = promauto.NewCounterVec(prometheus.CounterOpts{
+		Name: "synchronizer_client_disconnection_count",
+		Help: "Counter of client disconnections",
+	}, []string{prometheusAccountLabel, prometheusClusterLabel})
 	pulsarProducerMessagesProducedCounter = promauto.NewCounterVec(prometheus.CounterOpts{
 		Name: "synchronizer_pulsar_producer_messages_produced_count",
 		Help: "The total number of messages produced to pulsar",
-	}, []string{prometheusStatusLabel})
+	}, []string{prometheusStatusLabel, prometheusEventTypeLabel, prometheusKindLabel, prometheusAccountLabel, prometheusClusterLabel})
 	pulsarProducerMessagePayloadBytesProducedCounter = promauto.NewCounterVec(prometheus.CounterOpts{
 		Name: "synchronizer_pulsar_producer_message_payload_bytes_produced_count",
 		Help: "Counter of bytes published to pulsar (message payload) successfully",
-	}, []string{prometheusStatusLabel})
+	}, []string{prometheusStatusLabel, prometheusEventTypeLabel, prometheusKindLabel, prometheusAccountLabel, prometheusClusterLabel})
 	/*
 		messagesReceivedCounter = promauto.NewCounter(prometheus.CounterOpts{
 			Name: "synchronizer_messages_received_count",

adapters/backend/v1/pulsar.go

@@ -353,7 +353,6 @@ func (p *PulsarMessageProducer) ProduceMessageWithoutIdentifier(ctx context.Cont
 }
 
 func logPulsarSyncAsyncErrors(msgID pulsar.MessageID, message *pulsar.ProducerMessage, err error) {
-	var metricLabels prometheus.Labels
 	var msgIdStr string
 	if msgID != nil {
 		msgIdStr = msgID.String()
@@ -366,18 +365,27 @@ func logPulsarSyncAsyncErrors(msgID pulsar.MessageID, message *pulsar.ProducerMe
 		messageProperties = message.Properties
 	}
 
+	var statusLabelValue string
 	if err != nil {
-		metricLabels = prometheus.Labels{prometheusStatusLabel: prometheusStatusLabelValueError}
+		statusLabelValue = prometheusStatusLabelValueError
 		logger.L().Error("failed to send message to pulsar",
 			helpers.Error(err),
 			helpers.String("messageID", msgIdStr),
 			helpers.Int("payloadBytes", messagePayloadBytes),
 			helpers.Interface("messageProperties", messageProperties))
 	} else {
-		metricLabels = prometheus.Labels{prometheusStatusLabel: prometheusStatusLabelValueSuccess}
+		statusLabelValue = prometheusStatusLabelValueSuccess
+
 		logger.L().Debug("successfully sent message to pulsar", helpers.String("messageID", msgIdStr), helpers.Interface("messageProperties", messageProperties))
 	}
 
+	metricLabels := prometheus.Labels{
+		prometheusStatusLabel:    statusLabelValue,
+		prometheusKindLabel:      messageProperties[messaging.MsgPropResourceKindResource],
+		prometheusClusterLabel:   messageProperties[messaging.MsgPropCluster],
+		prometheusAccountLabel:   messageProperties[messaging.MsgPropAccount],
+		prometheusEventTypeLabel: messageProperties[messaging.MsgPropEvent],
+	}
 	pulsarProducerMessagesProducedCounter.With(metricLabels).Inc()
 	pulsarProducerMessagePayloadBytesProducedCounter.With(metricLabels).Add(float64(messagePayloadBytes))
 }

domain/identifiers.go

@@ -17,17 +17,6 @@ type KindName struct {
 	ResourceVersion int
 }
 
-func (c KindName) ToCustomProperties() map[string]string {
-	return map[string]string{
-		"group":           c.Kind.Group,
-		"version":         c.Kind.Version,
-		"resource":        c.Kind.Resource,
-		"name":            c.Name,
-		"namespace":       c.Namespace,
-		"resourceVersion": strconv.Itoa(c.ResourceVersion),
-	}
-}
-
 func (c KindName) String() string {
 	var kind string
 	if c.Kind == nil {

messaging/messages.go

@@ -9,7 +9,20 @@ const (
 	// MsgPropAccount is the property name for the account name
 	MsgPropAccount = "account"
 	// MsgPropEvent is the property name for the event type
-	MsgPropEvent                                  = "event"
+	MsgPropEvent = "event"
+	// MsgPropResourceKindGroup is the property name for the API Group of the resource
+	MsgPropResourceKindGroup = "group"
+	// MsgPropResourceKindVersion is the property name for the API Version of the resource
+	MsgPropResourceKindVersion = "version"
+	// MsgPropResourceKindResource is the property name for the API Resource of the resource
+	MsgPropResourceKindResource = "resource"
+	// MsgPropResourceName is the property name for the name of the resource
+	MsgPropResourceName = "name"
+	// MsgPropResourceNamespace is the property name for the namespace of the resource
+	MsgPropResourceNamespace = "namespace"
+	// MsgPropResourceVersion is the property name for the resource version of the resource
+	MsgPropResourceVersion = "resourceVersion"
+
 	MsgPropEventValueGetObjectMessage             = "GetObject"
 	MsgPropEventValuePatchObjectMessage           = "PatchObject"
 	MsgPropEventValueVerifyObjectMessage          = "VerifyObject"

messaging/utils.go

@@ -0,0 +1,18 @@
+package messaging
+
+import (
+	"strconv"
+
+	"github.com/kubescape/synchronizer/domain"
+)
+
+func KindNameToCustomProperties(kind domain.KindName) map[string]string {
+	return map[string]string{
+		MsgPropResourceKindGroup:    kind.Kind.Group,
+		MsgPropResourceKindVersion:  kind.Kind.Version,
+		MsgPropResourceKindResource: kind.Kind.Resource,
+		MsgPropResourceName:         kind.Name,
+		MsgPropResourceNamespace:    kind.Namespace,
+		MsgPropResourceVersion:      strconv.Itoa(kind.ResourceVersion),
+	}
+}

utils/utils.go

@@ -21,6 +21,7 @@ import (
 	"github.com/kubescape/go-logger/helpers"
 	spdxv1beta1 "github.com/kubescape/storage/pkg/generated/clientset/versioned/typed/softwarecomposition/v1beta1"
 	"github.com/kubescape/synchronizer/domain"
+
 	"github.com/pmezard/go-difflib/difflib"
 	"github.com/stretchr/testify/assert"
 	"golang.org/x/mod/semver"
@@ -197,7 +198,7 @@ func PulsarMessageIDtoString(msgID pulsar.MessageID) string {
 }
 
 func ServePprof() {
-	if logger.L().GetLevel() == helpers.DebugLevel.String() {
+	if _, present := os.LookupEnv("ENABLE_PROFILER"); present || logger.L().GetLevel() == helpers.DebugLevel.String() {
 		logger.L().Info("starting pprof server", helpers.String("port", "6060"))
 		pprofMux := http.NewServeMux()
 		pprofMux.HandleFunc("/debug/pprof/", pprof.Index)