Merge pull request #50 from lidofinance/devel

Devel

Changelog.md

@@ -1,3 +1,11 @@
+## 05.12.2024
+1. Remove PublishedAlerts metrics
+2. Added NotifyChannels: `forwarder_notification_channel_error_total`
+3. Inc `feeder_blocks_published_total` for each unsuccessful network request
+
+## 28.11.2024
+1. Field "uniqueKey" is required for getting quorum.
+
 ## 16.10.2024
 1. Added dynamic yaml notification config
 

brief/databus/finding.dto.json

@@ -37,7 +37,7 @@
       "type": "string"
     }
   },
-  "required": ["severity", "alertId", "name", "description", "botName", "team"],
+  "required": ["severity", "alertId", "name", "description", "botName", "team", "uniqueKey"],
   "definitions": {
     "Severity": {
       "type": "string",

cmd/forwarder/main.go

@@ -81,6 +81,7 @@ func main() {
 		MaxAge:     10 * time.Minute,
 		Subjects:   env.CollectNatsSubjects(notificationConfig),
 		MaxMsgSize: maxMsgSize,
+		Retention:  jetstream.InterestPolicy,
 	})
 	if err != nil && !errors.Is(err, nats.ErrStreamNameAlreadyInUse) {
 		fmt.Printf("could not create %s stream error: %v\n", natsStreamName, err)

internal/app/feeder/feeder.go

@@ -57,6 +57,7 @@ func (w *Feeder) Run(ctx context.Context, g *errgroup.Group) {
 			case <-ticker.C:
 				block, err := w.chainSrv.GetLatestBlock(ctx)
 				if err != nil {
+					w.metricsStore.PublishedBlocks.With(prometheus.Labels{metrics.Status: metrics.StatusFail}).Inc()
 					w.log.Error(fmt.Sprintf("GetLatestBlock error: %v", err))
 					continue
 				}
@@ -67,6 +68,7 @@ func (w *Feeder) Run(ctx context.Context, g *errgroup.Group) {
 
 				blockReceipts, err := w.chainSrv.GetBlockReceipts(ctx, block.Result.Hash)
 				if err != nil {
+					w.metricsStore.PublishedBlocks.With(prometheus.Labels{metrics.Status: metrics.StatusFail}).Inc()
 					w.log.Error(fmt.Sprintf("GetBlockReceipts error: %v", err))
 					continue
 				}
@@ -110,6 +112,7 @@ func (w *Feeder) Run(ctx context.Context, g *errgroup.Group) {
 
 				payload, marshalErr := json.Marshal(blockDto)
 				if marshalErr != nil {
+					w.metricsStore.PublishedBlocks.With(prometheus.Labels{metrics.Status: metrics.StatusFail}).Inc()
 					w.log.Error(fmt.Sprintf(`Could not marshal blockDto %s`, marshalErr))
 					continue
 				}
@@ -132,7 +135,7 @@ func (w *Feeder) Run(ctx context.Context, g *errgroup.Group) {
 				}
 
 				w.log.Info(fmt.Sprintf(`%d, %s`, blockDto.Number, blockDto.Hash), payloadSize)
-				w.metricsStore.PublishedAlerts.With(prometheus.Labels{metrics.Status: metrics.StatusOk}).Inc()
+				w.metricsStore.PublishedBlocks.With(prometheus.Labels{metrics.Status: metrics.StatusOk}).Inc()
 			}
 		}
 	})

internal/app/forwarder/forwarder.go

@@ -8,8 +8,9 @@ import (
 
 	"golang.org/x/sync/errgroup"
 
-	"github.com/lidofinance/onchain-mon/internal/pkg/consumer"
 	"github.com/nats-io/nats.go/jetstream"
+
+	"github.com/lidofinance/onchain-mon/internal/pkg/consumer"
 )
 
 type worker struct {

internal/connectors/metrics/prometheus.go

@@ -11,11 +11,11 @@ import (
 type Store struct {
 	Prometheus      *prometheus.Registry
 	BuildInfo       prometheus.Counter
-	PublishedAlerts *prometheus.CounterVec
 	PublishedBlocks *prometheus.CounterVec
 	SentAlerts      *prometheus.CounterVec
 	RedisErrors     prometheus.Counter
 	SummaryHandlers *prometheus.HistogramVec
+	NotifyChannels  *prometheus.CounterVec
 }
 
 const Status = `status`
@@ -40,17 +40,13 @@ func New(promRegistry *prometheus.Registry, prefix, appName, env string) *Store
 				"version": runtime.Version(),
 			},
 		}),
-		PublishedAlerts: promauto.NewCounterVec(prometheus.CounterOpts{
-			Name: fmt.Sprintf("%s_finding_published_total", prefix),
-			Help: "The total number of published findings",
-		}, []string{Status}),
 		PublishedBlocks: promauto.NewCounterVec(prometheus.CounterOpts{
 			Name: fmt.Sprintf("%s_blocks_published_total", prefix),
 			Help: "The total number of published blocks",
 		}, []string{Status}),
 		SentAlerts: promauto.NewCounterVec(prometheus.CounterOpts{
 			Name: fmt.Sprintf("%s_finding_sent_total", prefix),
-			Help: "The total number of set findings",
+			Help: "The total number of published findings",
 		}, []string{ConsumerName, Status}),
 		RedisErrors: promauto.NewCounter(prometheus.CounterOpts{
 			Name: fmt.Sprintf("%s_redis_error_total", prefix),
@@ -61,11 +57,14 @@ func New(promRegistry *prometheus.Registry, prefix, appName, env string) *Store
 			Help:    "Time spent processing request to notification channel",
 			Buckets: prometheus.DefBuckets,
 		}, []string{Channel}),
+		NotifyChannels: promauto.NewCounterVec(prometheus.CounterOpts{
+			Name: fmt.Sprintf("%s_notification_channel_error_total", prefix),
+			Help: "The total number of network errors of telegram, discord, opsgenie channels",
+		}, []string{Channel, Status}),
 	}
 
 	store.Prometheus.MustRegister(
 		store.BuildInfo,
-		store.PublishedAlerts,
 		store.PublishedBlocks,
 		store.SentAlerts,
 		store.RedisErrors,

internal/env/notify_config.go

@@ -59,7 +59,7 @@ type NotificationConfig struct {
 	Consumers        []*Consumer       `mapstructure:"consumers"`
 }
 
-func ReadNotificationConfig(env string, configPath string) (*NotificationConfig, error) {
+func ReadNotificationConfig(env, configPath string) (*NotificationConfig, error) {
 	v := viper.New()
 
 	if env != `local` {
@@ -179,7 +179,7 @@ func CollectNatsSubjects(cfg *NotificationConfig) []string {
 	}
 
 	out := make([]string, 0, len(natsSubjectsMap))
-	for subject, _ := range natsSubjectsMap {
+	for subject := range natsSubjectsMap {
 		out = append(out, subject)
 	}
 

internal/pkg/consumer/consumer.go

@@ -1,10 +1,7 @@
 package consumer
 
 import (
-	"bytes"
 	"context"
-	"crypto/sha256"
-	"encoding/hex"
 	"encoding/json"
 	"fmt"
 	"log/slog"
@@ -13,10 +10,10 @@ import (
 
 	"github.com/go-redis/redis/v8"
 	"github.com/hashicorp/golang-lru/v2/expirable"
-	"github.com/lidofinance/onchain-mon/generated/databus"
 	"github.com/nats-io/nats.go/jetstream"
 	"github.com/prometheus/client_golang/prometheus"
 
+	"github.com/lidofinance/onchain-mon/generated/databus"
 	"github.com/lidofinance/onchain-mon/internal/connectors/metrics"
 	"github.com/lidofinance/onchain-mon/internal/env"
 	"github.com/lidofinance/onchain-mon/internal/pkg/notifiler"
@@ -114,7 +111,15 @@ func NewConsumers(log *slog.Logger, metrics *metrics.Store, redisClient *redis.C
 			const LruSize = 125
 			cache := expirable.NewLRU[string, uint](LruSize, nil, time.Minute*10)
 
-			consumer := New(log, metrics, cache, redisClient, repo, consumerName, subject, consumerCfg.SeveritySet, consumerCfg.ByQuorum, quorumSize, notificationChannel)
+			consumer := New(
+				log, metrics, cache, redisClient, repo,
+				consumerName,
+				subject,
+				consumerCfg.SeveritySet,
+				consumerCfg.ByQuorum,
+				quorumSize,
+				notificationChannel,
+			)
 
 			consumers = append(consumers, consumer)
 		}
@@ -208,7 +213,7 @@ func (c *Consumer) GetConsumeHandler(ctx context.Context) func(msg jetstream.Msg
 			return
 		}
 
-		key := findingToUniqueHash(finding)
+		key := finding.UniqueKey
 		countKey := fmt.Sprintf(countTemplate, c.name, key)
 		statusKey := fmt.Sprintf(statusTemplate, c.name, key)
 
@@ -224,6 +229,7 @@ func (c *Consumer) GetConsumeHandler(ctx context.Context) func(msg jetstream.Msg
 			if err != nil {
 				c.log.Error(fmt.Sprintf(`Could not increase key value: %v`, err))
 				c.metrics.RedisErrors.Inc()
+				c.metrics.SentAlerts.With(prometheus.Labels{metrics.ConsumerName: c.name, metrics.Status: metrics.StatusFail}).Inc()
 				c.nackMessage(msg)
 				return
 			}
@@ -232,9 +238,11 @@ func (c *Consumer) GetConsumeHandler(ctx context.Context) func(msg jetstream.Msg
 				if err := c.redisClient.Expire(ctx, countKey, TTLMins10).Err(); err != nil {
 					c.log.Error(fmt.Sprintf(`Could not set expire time: %v`, err))
 					c.metrics.RedisErrors.Inc()
+					c.metrics.SentAlerts.With(prometheus.Labels{metrics.ConsumerName: c.name, metrics.Status: metrics.StatusFail}).Inc()
 
 					if _, err := c.redisClient.Decr(ctx, countKey).Result(); err != nil {
 						c.metrics.RedisErrors.Inc()
+						c.metrics.SentAlerts.With(prometheus.Labels{metrics.ConsumerName: c.name, metrics.Status: metrics.StatusFail}).Inc()
 						c.log.Error(fmt.Sprintf(`Could not decrease count key %s: %v`, countKey, err))
 					}
 
@@ -250,6 +258,7 @@ func (c *Consumer) GetConsumeHandler(ctx context.Context) func(msg jetstream.Msg
 			if err != nil {
 				c.log.Error(fmt.Sprintf(`Could not get key value: %v`, err))
 				c.metrics.RedisErrors.Inc()
+				c.metrics.SentAlerts.With(prometheus.Labels{metrics.ConsumerName: c.name, metrics.Status: metrics.StatusFail}).Inc()
 				c.nackMessage(msg)
 				return
 			}
@@ -290,13 +299,12 @@ func (c *Consumer) GetConsumeHandler(ctx context.Context) func(msg jetstream.Msg
 			if err != nil {
 				c.log.Error(fmt.Sprintf(`Could not get notification status: %v`, err))
 				c.metrics.RedisErrors.Inc()
+				c.metrics.SentAlerts.With(prometheus.Labels{metrics.ConsumerName: c.name, metrics.Status: metrics.StatusFail}).Inc()
 				c.nackMessage(msg)
 				return
 			}
 
 			if status == StatusSending {
-				c.metrics.SentAlerts.With(prometheus.Labels{metrics.ConsumerName: c.name, metrics.Status: metrics.StatusOk}).Inc()
-
 				c.log.Info(fmt.Sprintf("Another instance is sending finding: %s", finding.AlertId))
 				return
 			}
@@ -405,26 +413,3 @@ func (c *Consumer) ackMessage(msg jetstream.Msg) {
 		c.log.Error(fmt.Sprintf(`Could not ack msg: %v`, ackErr))
 	}
 }
-
-func computeSHA256Hash(data []byte) string {
-	hash := sha256.Sum256(data)
-	return hex.EncodeToString(hash[:])
-}
-
-func findingToUniqueHash(f *databus.FindingDtoJson) string {
-	var buffer bytes.Buffer
-
-	if f.UniqueKey != nil {
-		buffer.WriteString(f.Team)
-		buffer.WriteString(f.BotName)
-		buffer.WriteString(*f.UniqueKey)
-	} else {
-		buffer.WriteString(f.Team)
-		buffer.WriteString(f.BotName)
-		buffer.WriteString(f.AlertId)
-		buffer.WriteString(f.Name)
-		buffer.WriteString(string(f.Severity))
-	}
-
-	return computeSHA256Hash(buffer.Bytes())
-}

internal/pkg/notifiler/discord.go

@@ -25,6 +25,8 @@ type MessagePayload struct {
 	Content string `json:"content"`
 }
 
+const DiscordLabel = `discord`
+
 func NewDiscord(webhookURL string, httpClient *http.Client, metricsStore *metrics.Store, source string) *Discord {
 	return &Discord{
 		webhookURL: webhookURL,
@@ -72,13 +74,15 @@ func (d *Discord) send(ctx context.Context, message string) error {
 	defer func() {
 		resp.Body.Close()
 		duration := time.Since(start).Seconds()
-		d.metrics.SummaryHandlers.With(prometheus.Labels{metrics.Channel: `discord`}).Observe(duration)
+		d.metrics.SummaryHandlers.With(prometheus.Labels{metrics.Channel: DiscordLabel}).Observe(duration)
 	}()
 
 	if resp.StatusCode != http.StatusNoContent {
+		d.metrics.NotifyChannels.With(prometheus.Labels{metrics.Channel: DiscordLabel, metrics.Status: metrics.StatusFail}).Inc()
 		return fmt.Errorf("received from Discord non-204 response code: %v", resp.Status)
 	}
 
+	d.metrics.NotifyChannels.With(prometheus.Labels{metrics.Channel: DiscordLabel, metrics.Status: metrics.StatusOk}).Inc()
 	return nil
 }
 

internal/pkg/notifiler/opsgenie.go

@@ -37,7 +37,9 @@ func NewOpsgenie(opsGenieKey string, httpClient *http.Client, metricsStore *metr
 	}
 }
 
-func (u *OpsGenie) SendFinding(ctx context.Context, alert *databus.FindingDtoJson) error {
+const OpsGenieLabel = `opsgenie`
+
+func (o *OpsGenie) SendFinding(ctx context.Context, alert *databus.FindingDtoJson) error {
 	opsGeniePriority := ""
 	switch alert.Severity {
 	case databus.SeverityCritical:
@@ -51,7 +53,7 @@ func (u *OpsGenie) SendFinding(ctx context.Context, alert *databus.FindingDtoJso
 		return nil
 	}
 
-	message := FormatAlert(alert, u.source)
+	message := FormatAlert(alert, o.source)
 
 	payload := AlertPayload{
 		Message:     alert.Name,
@@ -60,10 +62,10 @@ func (u *OpsGenie) SendFinding(ctx context.Context, alert *databus.FindingDtoJso
 		Priority:    opsGeniePriority,
 	}
 
-	return u.send(ctx, payload)
+	return o.send(ctx, payload)
 }
 
-func (u *OpsGenie) send(ctx context.Context, payload AlertPayload) error {
+func (o *OpsGenie) send(ctx context.Context, payload AlertPayload) error {
 	payloadBytes, err := json.Marshal(payload)
 	if err != nil {
 		return fmt.Errorf("could not marshal OpsGenie payload: %w", err)
@@ -78,26 +80,28 @@ func (u *OpsGenie) send(ctx context.Context, payload AlertPayload) error {
 	}
 
 	req.Header.Set("Content-Type", "application/json")
-	req.Header.Set("Authorization", "GenieKey "+u.opsGenieKey)
+	req.Header.Set("Authorization", "GenieKey "+o.opsGenieKey)
 
 	start := time.Now()
-	resp, err := u.httpClient.Do(req)
+	resp, err := o.httpClient.Do(req)
 	if err != nil {
 		return fmt.Errorf("could not send OpsGenie request: %w", err)
 	}
 	defer func() {
 		resp.Body.Close()
 		duration := time.Since(start).Seconds()
-		u.metrics.SummaryHandlers.With(prometheus.Labels{metrics.Channel: `opsgenie`}).Observe(duration)
+		o.metrics.SummaryHandlers.With(prometheus.Labels{metrics.Channel: OpsGenieLabel}).Observe(duration)
 	}()
 
 	if resp.StatusCode != http.StatusAccepted {
+		o.metrics.NotifyChannels.With(prometheus.Labels{metrics.Channel: OpsGenieLabel, metrics.Status: metrics.StatusFail}).Inc()
 		return fmt.Errorf("received from OpsGenie non-202 response code: %v", resp.Status)
 	}
 
+	o.metrics.NotifyChannels.With(prometheus.Labels{metrics.Channel: OpsGenieLabel, metrics.Status: metrics.StatusOk}).Inc()
 	return nil
 }
 
-func (d *OpsGenie) GetType() string {
+func (o *OpsGenie) GetType() string {
 	return "OpsGenie"
 }