Implement Sentry Mux (#11793)

This implements a pass through multiplexer which forward sentry grpc requests to multiple sentry instances. This change also rationalizes the use of direct vs grpc sentry interfaces. All sentry consumers now work with the base grpc interface rather than the direct implementation. --------- Co-authored-by: alex.sharov <[email protected]> Co-authored-by: Mark Holt <[email protected]>
erigontech · Sep 2, 2024 · d7bd2c5 · d7bd2c5
1 parent a8e795d
commit d7bd2c5
Show file tree

Hide file tree

Showing 35 changed files with 1,866 additions and 567 deletions.
diff --git a/cmd/txpool/main.go b/cmd/txpool/main.go
@@ -144,7 +144,7 @@ func doTxpool(ctx context.Context, logger log.Logger) error {
 
 	log.Info("TxPool started", "db", filepath.Join(datadirCli, "txpool"))
 
-	sentryClients := make([]direct.SentryClient, len(sentryAddr))
+	sentryClients := make([]proto_sentry.SentryClient, len(sentryAddr))
 	for i := range sentryAddr {
 		creds, err := grpcutil.TLS(TLSCACert, TLSCertfile, TLSKeyFile)
 		if err != nil {

diff --git a/consensus/clique/clique.go b/consensus/clique/clique.go
@@ -43,7 +43,6 @@ import (
 	"github.com/erigontech/erigon-lib/kv/dbutils"
 	"github.com/erigontech/erigon-lib/log/v3"
 
-	"github.com/erigontech/erigon/common"
 	"github.com/erigontech/erigon/common/debug"
 	"github.com/erigontech/erigon/consensus"
 	"github.com/erigontech/erigon/core/state"
@@ -453,7 +452,7 @@ func (c *Clique) Seal(chain consensus.ChainHeaderReader, blockWithReceipts *type
 		wiggle := time.Duration(len(snap.Signers)/2+1) * wiggleTime
 		delay += time.Duration(rand.Int63n(int64(wiggle))) // nolint: gosec
 
-		c.logger.Trace("Out-of-turn signing requested", "wiggle", common.PrettyDuration(wiggle))
+		c.logger.Trace("Out-of-turn signing requested", "wiggle", libcommon.PrettyDuration(wiggle))
 	}
 	// Sign all the things!
 	sighash, err := signFn(signer, accounts.MimetypeClique, CliqueRLP(header))
@@ -462,7 +461,7 @@ func (c *Clique) Seal(chain consensus.ChainHeaderReader, blockWithReceipts *type
 	}
 	copy(header.Extra[len(header.Extra)-ExtraSeal:], sighash)
 	// Wait until sealing is terminated or delay timeout.
-	c.logger.Trace("Waiting for slot to sign and propagate", "delay", common.PrettyDuration(delay))
+	c.logger.Trace("Waiting for slot to sign and propagate", "delay", libcommon.PrettyDuration(delay))
 	go func() {
 		defer debug.LogPanic()
 		select {

diff --git a/consensus/clique/snapshot.go b/consensus/clique/snapshot.go
@@ -40,7 +40,6 @@ import (
 
 	"github.com/erigontech/erigon-lib/log/v3"
 
-	"github.com/erigontech/erigon/common"
 	"github.com/erigontech/erigon/core/types"
 )
 
@@ -314,12 +313,12 @@ func (s *Snapshot) apply(sigcache *lru.ARCCache[libcommon.Hash, libcommon.Addres
 		}
 		// If we're taking too much time (ecrecover), notify the user once a while
 		if time.Since(logged) > 8*time.Second {
-			logger.Info("Reconstructing voting history", "processed", i, "total", len(headers), "elapsed", common.PrettyDuration(time.Since(start)))
+			logger.Info("Reconstructing voting history", "processed", i, "total", len(headers), "elapsed", libcommon.PrettyDuration(time.Since(start)))
 			logged = time.Now()
 		}
 	}
 	if time.Since(start) > 8*time.Second {
-		logger.Info("Reconstructed voting history", "processed", len(headers), "elapsed", common.PrettyDuration(time.Since(start)))
+		logger.Info("Reconstructed voting history", "processed", len(headers), "elapsed", libcommon.PrettyDuration(time.Since(start)))
 	}
 	snap.Number += uint64(len(headers))
 	snap.Hash = headers[len(headers)-1].Hash()

diff --git a/consensus/ethash/algorithm.go b/consensus/ethash/algorithm.go
@@ -30,15 +30,14 @@ import (
 	"time"
 	"unsafe"
 
-	common2 "github.com/erigontech/erigon-lib/common"
+	libcommon "github.com/erigontech/erigon-lib/common"
 
 	"golang.org/x/crypto/sha3"
 
 	"github.com/erigontech/erigon-lib/common/length"
 
 	"github.com/erigontech/erigon-lib/log/v3"
 
-	"github.com/erigontech/erigon/common"
 	"github.com/erigontech/erigon/common/bitutil"
 	"github.com/erigontech/erigon/common/debug"
 	"github.com/erigontech/erigon/crypto"
@@ -141,7 +140,7 @@ func seedHash(block uint64) []byte {
 		return seed
 	}
 
-	h := common2.NewHasher()
+	h := libcommon.NewHasher()
 
 	for i := 0; i < int(block/epochLength); i++ {
 		h.Sha.Reset()
@@ -157,7 +156,7 @@ func seedHash(block uint64) []byte {
 		}
 	}
 
-	common2.ReturnHasherToPool(h)
+	libcommon.ReturnHasherToPool(h)
 
 	return seed
 }
@@ -192,7 +191,7 @@ func generateCache(dest []uint32, epoch uint64, seed []byte) {
 		if elapsed > 5*time.Second {
 			logFn = logger.Info
 		}
-		logFn("Generated ethash verification cache", "elapsed", common.PrettyDuration(elapsed))
+		logFn("Generated ethash verification cache", "elapsed", libcommon.PrettyDuration(elapsed))
 	}()
 	// Convert our destination slice to a byte buffer
 	var cache []byte
@@ -218,7 +217,7 @@ func generateCache(dest []uint32, epoch uint64, seed []byte) {
 			case <-done:
 				return
 			case <-time.After(3 * time.Second):
-				logger.Info("Generating ethash verification cache", "percentage", atomic.LoadUint32(&progress)*100/uint32(rows)/4, "elapsed", common.PrettyDuration(time.Since(start)))
+				logger.Info("Generating ethash verification cache", "percentage", atomic.LoadUint32(&progress)*100/uint32(rows)/4, "elapsed", libcommon.PrettyDuration(time.Since(start)))
 			}
 		}
 	}()
@@ -348,7 +347,7 @@ func generateDataset(dest []uint32, epoch uint64, cache []uint32) {
 		if elapsed > 3*time.Second {
 			logFn = logger.Info
 		}
-		logFn("Generated ethash verification cache", "elapsed", common.PrettyDuration(elapsed))
+		logFn("Generated ethash verification cache", "elapsed", libcommon.PrettyDuration(elapsed))
 	}()
 
 	// Figure out whether the bytes need to be swapped for the machine
@@ -396,7 +395,7 @@ func generateDataset(dest []uint32, epoch uint64, cache []uint32) {
 				})
 
 				if status := atomic.AddUint64(&progress, 1); status%percent == 0 {
-					logger.Info("Generating DAG in progress", "percentage", (status*100)/(size/hashBytes), "elapsed", common.PrettyDuration(time.Since(start)))
+					logger.Info("Generating DAG in progress", "percentage", (status*100)/(size/hashBytes), "elapsed", libcommon.PrettyDuration(time.Since(start)))
 				}
 			}
 		}(i)

diff --git a/consensus/ethash/sealer.go b/consensus/ethash/sealer.go
@@ -36,7 +36,6 @@ import (
 	libcommon "github.com/erigontech/erigon-lib/common"
 	"github.com/erigontech/erigon-lib/common/hexutil"
 
-	"github.com/erigontech/erigon/common"
 	"github.com/erigontech/erigon/consensus"
 	"github.com/erigontech/erigon/core/types"
 )
@@ -307,7 +306,7 @@ func (s *remoteSealer) submitWork(nonce types.BlockNonce, mixDigest libcommon.Ha
 	start := time.Now()
 	if !s.noverify {
 		if err := s.ethash.verifySeal(header, true); err != nil {
-			s.ethash.config.Log.Warn("Invalid proof-of-work submitted", "sealhash", sealhash, "elapsed", common.PrettyDuration(time.Since(start)), "err", err)
+			s.ethash.config.Log.Warn("Invalid proof-of-work submitted", "sealhash", sealhash, "elapsed", libcommon.PrettyDuration(time.Since(start)), "err", err)
 			return false
 		}
 	}
@@ -316,7 +315,7 @@ func (s *remoteSealer) submitWork(nonce types.BlockNonce, mixDigest libcommon.Ha
 		s.ethash.config.Log.Warn("Ethash result channel is empty, submitted mining result is rejected")
 		return false
 	}
-	s.ethash.config.Log.Trace("Verified correct proof-of-work", "sealhash", sealhash, "elapsed", common.PrettyDuration(time.Since(start)))
+	s.ethash.config.Log.Trace("Verified correct proof-of-work", "sealhash", sealhash, "elapsed", libcommon.PrettyDuration(time.Since(start)))
 
 	// Solutions seems to be valid, return to the miner and notify acceptance.
 	solution := block.Block.WithSeal(header)

diff --git a/common/format.go → erigon-lib/common/format.go b/common/format.go → erigon-lib/common/format.go