Merge pull request #20 from lidofinance/todosv2

feat: Fix quorum

.golangci.yml

@@ -73,7 +73,7 @@ linters:
     - dogsled
     - dupl
     - errcheck
-    - exportloopref
+    - copyloopvar
     - funlen
     - gochecknoinits
     - goconst
@@ -147,3 +147,4 @@ run:
     - internal/renameio # extracted from Go code
     - internal/robustio # extracted from Go code
     - mocks
+    - vendor

Changelog.md

@@ -1,3 +1,18 @@
+## 23.09.2024
+1. Add lru for quorum
+2. Tun docker-compose-file 
+3. Upgrade GO 1.23.1
+4. Increased MaxMsgSize for Nats for 3Mb
+5. Lint project
+
+## 16.09.2024
+1. Added redis
+2. Added quorum powered by redis
+3. Added retry for sending message to Telegram, Discord, OpsGenia
+
+## 14.09.2024
+1. Added feeder
+
 ## 22.06.2024
 1. Added DevOps independent consumer
 2. Updated readme.md

Dockerfile

@@ -1,9 +1,11 @@
 # Build stage
-FROM golang:1.22.3-alpine as builder
+FROM golang:1.23.1-alpine AS builder
 
 WORKDIR /go/src/app
 COPY . .
 
+RUN apk add git=2.45.2-r0
+
 RUN go build -ldflags="-X github.com/lidofinance/finding-forwarder/internal/connectors/metrics.Commit=$(git rev-parse HEAD)" -o ./bin/main ./cmd/service
 RUN go build -ldflags="-X github.com/lidofinance/finding-forwarder/internal/connectors/metrics.Commit=$(git rev-parse HEAD)" -o ./bin/worker ./cmd/worker
 

README.md

@@ -2,7 +2,7 @@
 Service forwards findings from forta-local-node to telegram, opsGenia and discord
 
  ## How to develop
- 1. Install go1.22.3+
+ 1. Install go1.23.1+
  2. cd root repository
  3. make tools
  4. make vendor

cmd/worker/main.go

@@ -4,13 +4,13 @@ import (
 	"context"
 	"errors"
 	"fmt"
-
 	"os"
 	"os/signal"
 	"syscall"
 	"time"
 
 	"github.com/go-chi/chi/v5"
+	"github.com/hashicorp/golang-lru/v2/expirable"
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
 	"github.com/prometheus/client_golang/prometheus"
@@ -28,6 +28,8 @@ import (
 	"github.com/lidofinance/finding-forwarder/internal/utils/registry/teams"
 )
 
+const maxMsgSize = 3 * 1024 * 1024 // 3 Mb
+
 func main() {
 	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM, syscall.SIGINT, syscall.SIGKILL)
 	defer stop()
@@ -43,14 +45,14 @@ func main() {
 
 	rds, err := redis.NewRedisClient(gCtx, cfg.AppConfig.RedisURL, log)
 	if err != nil {
-		log.Error(`create redis client error: `, err.Error())
+		log.Error(fmt.Sprintf(`create redis client error: %v`, err))
 		return
 	}
 	defer rds.Close()
 
 	natsClient, natsErr := nc.New(&cfg.AppConfig, log)
 	if natsErr != nil {
-		fmt.Println("Could not connect to nats error:", natsErr.Error())
+		log.Error(fmt.Sprintf(`Could not connect to nats error: %v`, err))
 		return
 	}
 	defer natsClient.Close()
@@ -87,7 +89,7 @@ func main() {
 		Subjects: []string{
 			protocolNatsSubject,
 		},
-		MaxMsgSize: 1 * 1024 * 1024,
+		MaxMsgSize: maxMsgSize,
 	})
 	if err != nil && !errors.Is(err, nats.ErrStreamNameAlreadyInUse) {
 		fmt.Printf("could not create %s stream error: %v\n", natsStreamName, err)
@@ -103,7 +105,7 @@ func main() {
 			protocolFortaSubject,
 			fallbackFortaSubject,
 		},
-		MaxMsgSize: 1 * 1024 * 1024,
+		MaxMsgSize: maxMsgSize,
 	})
 	if err != nil && !errors.Is(err, nats.ErrStreamNameAlreadyInUse) {
 		fmt.Printf("could not create %s stream error: %v\n", fortaStreamName, err)
@@ -117,11 +119,12 @@ func main() {
 		OpsGenie = `OpsGenie`
 	)
 
+	const LruSize = 15
+	cache := expirable.NewLRU[string, uint](LruSize, nil, time.Minute*2)
 	protocolWorker := worker.NewFindingWorker(
-		protocolNatsSubject,
-		cfg.AppConfig.QuorumSize,
-		rds,
-		natStream, log, metricsStore,
+		log, metricsStore, cache,
+		rds, natStream,
+		protocolNatsSubject, cfg.AppConfig.QuorumSize,
 		worker.WithFindingConsumer(services.OnChainAlertsTelegram, `OnChainAlerts_Telegram_Consumer`, registry.OnChainAlerts, Telegram),
 		worker.WithFindingConsumer(services.OnChainUpdatesTelegram, `OnChainUpdates_Telegram_Consumer`, registry.OnChainUpdates, Telegram),
 		worker.WithFindingConsumer(services.ErrorsTelegram, `Protocol_Errors_Telegram_Consumer`, registry.OnChainErrors, Telegram),
@@ -130,8 +133,9 @@ func main() {
 	)
 
 	protocolFortaWorker := worker.NewAlertWorker(
+		log, metricsStore,
+		fortaStream,
 		protocolFortaSubject,
-		fortaStream, log, metricsStore,
 		worker.WithAlertConsumer(services.OnChainAlertsTelegram, `Forta_OnChainAlerts_Telegram_Consumer`, registry.AlertOnChainAlerts, Telegram),
 		worker.WithAlertConsumer(services.OnChainUpdatesTelegram, `Forta_OnChainUpdates_Telegram_Consumer`, registry.AlertOnChainUpdates, Telegram),
 		worker.WithAlertConsumer(services.ErrorsTelegram, `Forta_Protocol_Errors_Telegram_Consumer`, registry.AlertOnChainErrors, Telegram),
@@ -140,13 +144,11 @@ func main() {
 	)
 
 	fallbackFortaWorker := worker.NewAlertWorker(
-		fallbackFortaSubject,
-		fortaStream, log, metricsStore,
+		log, metricsStore,
+		fortaStream, fallbackFortaSubject,
 		worker.WithAlertConsumer(services.Discord, `FortaFallback_Discord_Consumer`, registry.AlertFallBackAlerts, Discord),
 	)
 
-	feederWrk := feeder.New(log, services.ChainSrv, js, metricsStore, cfg.AppConfig.BlockTopic)
-
 	// Listen findings from Nats
 	if err := protocolWorker.Run(gCtx, g); err != nil {
 		fmt.Println("Could not start protocolWorker error:", err.Error())
@@ -164,6 +166,7 @@ func main() {
 		return
 	}
 
+	feederWrk := feeder.New(log, services.ChainSrv, js, metricsStore, cfg.AppConfig.BlockTopic)
 	feederWrk.Run(gCtx, g)
 
 	app.RunHTTPServer(gCtx, g, cfg.AppConfig.Port, r)

docker-compose.yml

@@ -1,4 +1,3 @@
-version: '3.8'
 x-logging: &default-logging
   options:
     max-size: "50m"
@@ -20,19 +19,20 @@ services:
       - "4222:4222"
       - "8222:8222"
     command: >
-      -js
+      -js -c /etc/nats/nats.conf
     environment:
       - JS_STORAGE_DIR=/data/jetstream
     volumes:
-      - ./nats/data:/data
+      - ./nats/jetstream:/data/jetstream
+      - ./nats/nats.conf:/etc/nats/nats.conf
 
   service-forta-json-rpc:
     container_name: forta-json-rpc
     logging: *default-logging
     image: forta-network/forta-node:latest
     command: /forta-node json-rpc
     ports:
-      - 8545:8545
+      - "8545:8545"
     volumes:
       - ./forta-local-config.yml:/.forta/config.yml
     depends_on:
@@ -48,14 +48,13 @@ services:
     depends_on:
       - service-forta-nats
       - service-forta-json-rpc
-      - service-ethereum-steth
 
-  forwarder-server:
-    container_name: finding-forwarder-server
+  forwarder-worker:
+    container_name: finding-forwarder-worker
     build: ./
     restart: always
     command:
-      - ./main
+      - ./worker
     env_file:
       - .env
     environment:
@@ -73,15 +72,20 @@ services:
       - DISCORD_WEBHOOK_URL=${DISCORD_WEBHOOK_URL}
       - NATS_DEFAULT_URL=${NATS_DEFAULT_URL}
       - BOT_CONTAINERS=${BOT_CONTAINERS}
+      - REDIS_ADDRESS=${REDIS_ADDRESS}
+      - QUORUM_SIZE=${QUORUM_SIZE}
     ports:
-      - '8080:8080'
+      - "8081:8080"
+    depends_on:
+      - redis
+      - service-forta-nats
 
-  forwarder-worker:
-    container_name: finding-forwarder-worker
+  forwarder-server:
+    container_name: finding-forwarder-server
     build: ./
     restart: always
     command:
-      - ./worker
+      - ./main
     env_file:
       - .env
     environment:
@@ -102,42 +106,75 @@ services:
       - REDIS_ADDRESS=${REDIS_ADDRESS}
       - QUORUM_SIZE=${QUORUM_SIZE}
     ports:
-      - '8081:8080'
+      - "8082:8080"
     depends_on:
       - redis
       - service-forta-nats
 
-  service-ethereum-steth:
-    container_name: ethereum-steth
-    logging: *default-logging
-    image: monitoring/steth:latest
-    restart: unless-stopped
-    environment:
-      - APP_NAME=ethereum-steth
-      - INSTANCE=forta-local-host
-      - ETHEREUM_RPC_URL=http://forta-json-rpc:8545
-      - USE_FORTA_RPC_URL=false
-      - NODE_ENV=production
-      - NATS_SERVER_URL=http://forta-nats:4222
-    depends_on:
-      - service-forta-json-rpc
-      - service-forta-nats
-    ports:
-      - '3001:3000'
+  #service-ethereum-steth:
+  #  container_name: ethereum-steth
+  #  logging: *default-logging
+  #  image: monitoring/steth:latest
+  #  restart: unless-stopped
+  #  environment:
+  #    - APP_NAME=ethereum-steth
+  #    - INSTANCE=forta-local-host
+  #    - ETHEREUM_RPC_URL=http://forta-json-rpc:8545
+  #    - USE_FORTA_RPC_URL=false
+  #    - NODE_ENV=production
+  #    - NATS_SERVER_URL=http://forta-nats:4222
+  #  depends_on:
+  #    - service-forta-json-rpc
+  #    - service-forta-nats
+  #  ports:
+  #    - "3001:3000"
 
-  service-l2-bridge-arbitrum:
-    container_name: l2-bridge-arbitrum
-    logging: *default-logging
-    image: monitoring/l2-bridge-arbitrum:latest
-    restart: unless-stopped
-    environment:
-      - APP_NAME=l2-bridge-arbitrum
-      - INSTANCE=forta-local-host
-      - ETHEREUM_RPC_URL=http://forta-json-rpc:8545
-      - ARBITRUM_RPC_URL=https://arbitrum-one.blastapi.io/<you_api_ley>
-      - USE_FORTA_RPC_URL=false
-      - NODE_ENV=production
-    depends_on:
-      - service-forta-json-rpc
-    ports:
-      - '3002:3000'
+  #service-l2-bridge-arbitrum:
+  #  container_name: l2-bridge-arbitrum
+  #  logging: *default-logging
+  #  image: monitoring/l2-bridge-arbitrum:latest
+  #  restart: unless-stopped
+  #  environment:
+  #    - APP_NAME=l2-bridge-arbitrum
+  #    - INSTANCE=forta-local-host
+  #    - ETHEREUM_RPC_URL=http://forta-json-rpc:8545
+  #    - ARBITRUM_RPC_URL=https://arbitrum-one.blastapi.io/<your_api_key>
+  #    - USE_FORTA_RPC_URL=false
+  #    - NODE_ENV=production
+  #  depends_on:
+  #    - service-forta-json-rpc
+  #  ports:
+  #    - "3002:3000"
+
+  #service-ethereum-steth-v2:
+  #  container_name: ethereum-steth-v2
+  #  logging: *default-logging
+  #  image: monitoring/steth-v2:latest
+  #  restart: unless-stopped
+  #  environment:
+  #    - APP_NAME=ethereum-steth-v2
+  #    - INSTANCE=forta-local-host
+  #    - ETHEREUM_RPC_URL=https://eth-mainnet.alchemyapi.io/v2/<your_key>
+  #    - NODE_ENV=production
+  #    - NATS_SERVER_URL=http://127.0.0.1:4222
+  #  depends_on:
+  #    - service-forta-nats
+  #  ports:
+  #    - "3002:3000"
+
+  #service-l2-bridge-arbitrum-v2:
+  #  container_name: l2-bridge-arbitrum-v2
+  #  logging: *default-logging
+  #  image: monitoring/l2-bridge-arbitrum-v2:latest
+  #  restart: unless-stopped
+  #  environment:
+  #    - APP_NAME=l2-bridge-arbitrum-v2
+  #    - INSTANCE=forta-local-host
+  #    - ETHEREUM_RPC_URL=https://eth-mainnet.alchemyapi.io/v2/<your_key>
+  #    - ARBITRUM_RPC_URL=https://arbitrum-one.blastapi.io/<your_key>
+  #    - NODE_ENV=production
+  #    - NATS_DEFAULT_URL=http://127.0.0.1:4222
+  #  depends_on:
+  #    - service-forta-nats
+  #  ports:
+  #    - "3003:3000"