Merge branch 'master' into ci-check-monitoring-from-code

Author: maru-ava

RELEASES.md

@@ -1,5 +1,14 @@
 # Release Notes
 
+## [v1.12.3] (pending)
+
+- Extended the network health check by also alerting if a primary network validator has no nodes connected to it. Runs a configurable time after startup or 10 minutes by default.
+
+### Configs
+-  How long after startup the aforementioned health check runs can be configured via:
+`--network-no-ingress-connections-grace-period`
+
+
 ## [v1.12.2](https://github.com/ava-labs/avalanchego/releases/tag/v1.12.2)
 
 This version is backwards compatible to [v1.12.0](https://github.com/ava-labs/avalanchego/releases/tag/v1.12.0). It is optional, but encouraged.

config/config.go

@@ -327,13 +327,14 @@ func getNetworkConfig(
 		},
 
 		HealthConfig: network.HealthConfig{
-			Enabled:                      sybilProtectionEnabled,
-			MaxTimeSinceMsgSent:          v.GetDuration(NetworkHealthMaxTimeSinceMsgSentKey),
-			MaxTimeSinceMsgReceived:      v.GetDuration(NetworkHealthMaxTimeSinceMsgReceivedKey),
-			MaxPortionSendQueueBytesFull: v.GetFloat64(NetworkHealthMaxPortionSendQueueFillKey),
-			MinConnectedPeers:            v.GetUint(NetworkHealthMinPeersKey),
-			MaxSendFailRate:              v.GetFloat64(NetworkHealthMaxSendFailRateKey),
-			SendFailRateHalflife:         halflife,
+			Enabled:                                 sybilProtectionEnabled,
+			MaxTimeSinceMsgSent:                     v.GetDuration(NetworkHealthMaxTimeSinceMsgSentKey),
+			MaxTimeSinceMsgReceived:                 v.GetDuration(NetworkHealthMaxTimeSinceMsgReceivedKey),
+			MaxPortionSendQueueBytesFull:            v.GetFloat64(NetworkHealthMaxPortionSendQueueFillKey),
+			MinConnectedPeers:                       v.GetUint(NetworkHealthMinPeersKey),
+			MaxSendFailRate:                         v.GetFloat64(NetworkHealthMaxSendFailRateKey),
+			SendFailRateHalflife:                    halflife,
+			NoIngressValidatorConnectionGracePeriod: v.GetDuration(NetworkNoIngressValidatorConnectionsGracePeriodKey),
 		},
 
 		ProxyEnabled:           v.GetBool(NetworkTCPProxyEnabledKey),

config/flags.go

@@ -169,7 +169,7 @@ func addNodeFlags(fs *pflag.FlagSet) {
 	fs.Duration(NetworkReadHandshakeTimeoutKey, constants.DefaultNetworkReadHandshakeTimeout, "Timeout value for reading handshake messages")
 	fs.Duration(NetworkPingTimeoutKey, constants.DefaultPingPongTimeout, "Timeout value for Ping-Pong with a peer")
 	fs.Duration(NetworkPingFrequencyKey, constants.DefaultPingFrequency, "Frequency of pinging other peers")
-
+	fs.Duration(NetworkNoIngressValidatorConnectionsGracePeriodKey, constants.DefaultNoIngressValidatorConnectionGracePeriod, "Time after which nodes are expected to be connected to us if we are a primary network validator, otherwise a health check fails")
 	fs.String(NetworkCompressionTypeKey, constants.DefaultNetworkCompressionType.String(), fmt.Sprintf("Compression type for outbound messages. Must be one of [%s, %s]", compression.TypeZstd, compression.TypeNone))
 
 	fs.Duration(NetworkMaxClockDifferenceKey, constants.DefaultNetworkMaxClockDifference, "Max allowed clock difference value between this node and peers")

config/keys.go

@@ -120,6 +120,7 @@ const (
 	NetworkInboundThrottlerMaxConnsPerSecKey           = "network-inbound-connection-throttling-max-conns-per-sec"
 	NetworkOutboundConnectionThrottlingRpsKey          = "network-outbound-connection-throttling-rps"
 	NetworkOutboundConnectionTimeoutKey                = "network-outbound-connection-timeout"
+	NetworkNoIngressValidatorConnectionsGracePeriodKey = "network-no-ingress-connections-grace-period"
 	BenchlistFailThresholdKey                          = "benchlist-fail-threshold"
 	BenchlistDurationKey                               = "benchlist-duration"
 	BenchlistMinFailingDurationKey                     = "benchlist-min-failing-duration"

go.mod

@@ -12,7 +12,7 @@ go 1.23.6
 require (
 	github.com/DataDog/zstd v1.5.2
 	github.com/NYTimes/gziphandler v1.1.1
-	github.com/StephenButtolph/canoto v0.10.0
+	github.com/StephenButtolph/canoto v0.13.3
 	github.com/antithesishq/antithesis-sdk-go v0.3.8
 	github.com/ava-labs/coreth v0.14.1-rc.2
 	github.com/ava-labs/ledger-avalanche/go v0.0.0-20241009183145-e6f90a8a1a60

go.sum

@@ -53,8 +53,8 @@ github.com/Joker/hpp v1.0.0/go.mod h1:8x5n+M1Hp5hC0g8okX3sR3vFQwynaX/UgSOM9MeBKz
 github.com/NYTimes/gziphandler v1.1.1 h1:ZUDjpQae29j0ryrS0u/B8HZfJBtBQHjqw2rQ2cqUQ3I=
 github.com/NYTimes/gziphandler v1.1.1/go.mod h1:n/CVRwUEOgIxrgPvAQhUUr9oeUtvrhMomdKFjzJNB0c=
 github.com/Shopify/goreferrer v0.0.0-20181106222321-ec9c9a553398/go.mod h1:a1uqRtAwp2Xwc6WNPJEufxJ7fx3npB4UV/JOLmbu5I0=
-github.com/StephenButtolph/canoto v0.10.0 h1:KdW85TYQXH+gwR8vOxfOUf28TRpkLU+X06Kycg1IR7s=
-github.com/StephenButtolph/canoto v0.10.0/go.mod h1:MxppdgKRApRBvIg4ZgO2e14m/NSBjFMuydy97OB/gYY=
+github.com/StephenButtolph/canoto v0.13.3 h1:tZ0BWaKK6MQr+s45Nvp9YUVa4U5aozNIKSXzQzubXA0=
+github.com/StephenButtolph/canoto v0.13.3/go.mod h1:IcnAHC6nJUfQFVR9y60ko2ecUqqHHSB6UwI9NnBFZnE=
 github.com/VictoriaMetrics/fastcache v1.12.1 h1:i0mICQuojGDL3KblA7wUNlY5lOK6a4bwt3uRKnkZU40=
 github.com/VictoriaMetrics/fastcache v1.12.1/go.mod h1:tX04vaqcNoQeGLD+ra5pU5sWkuxnzWhEzLwhP9w653o=
 github.com/aead/siphash v1.0.1/go.mod h1:Nywa3cDsYNNK3gaciGTWPwHt0wlpNV15vwmswBAUSII=

main/main.go

@@ -15,7 +15,6 @@ import (
 	"github.com/ava-labs/avalanchego/app"
 	"github.com/ava-labs/avalanchego/config"
 	"github.com/ava-labs/avalanchego/version"
-	"github.com/ava-labs/avalanchego/vms/platformvm/block/executor"
 )
 
 func main() {
@@ -60,8 +59,6 @@ func main() {
 
 	if term.IsTerminal(int(os.Stdout.Fd())) {
 		fmt.Println(app.Header)
-	} else {
-		executor.EtnaActivationWasLogged.Set(true)
 	}
 
 	nodeApp, err := app.New(nodeConfig)

network/config.go

@@ -26,6 +26,10 @@ type HealthConfig struct {
 	// Marks if the health check should be enabled
 	Enabled bool `json:"-"`
 
+	// NoIngressValidatorConnectionGracePeriod denotes the time after which the health check fails
+	// for primary network validators with no ingress connections.
+	NoIngressValidatorConnectionGracePeriod time.Duration
+
 	// MinConnectedPeers is the minimum number of peers that the network should
 	// be connected to be considered healthy.
 	MinConnectedPeers uint `json:"minConnectedPeers"`

network/network.go

@@ -42,10 +42,11 @@ import (
 )
 
 const (
-	ConnectedPeersKey           = "connectedPeers"
-	TimeSinceLastMsgReceivedKey = "timeSinceLastMsgReceived"
-	TimeSinceLastMsgSentKey     = "timeSinceLastMsgSent"
-	SendFailRateKey             = "sendFailRate"
+	PrimaryNetworkValidatorHealthKey = "primary network validator health"
+	ConnectedPeersKey                = "connectedPeers"
+	TimeSinceLastMsgReceivedKey      = "timeSinceLastMsgReceived"
+	TimeSinceLastMsgSentKey          = "timeSinceLastMsgSent"
+	SendFailRateKey                  = "sendFailRate"
 )
 
 var (
@@ -153,6 +154,8 @@ type network struct {
 	connectedPeers  peer.Set
 	closing         bool
 
+	startupTime time.Time
+
 	// router is notified about all peer [Connected] and [Disconnected] events
 	// as well as all non-handshake peer messages.
 	//
@@ -257,11 +260,10 @@ func NewNetwork(
 	}
 
 	peerConfig := &peer.Config{
-		ReadBufferSize:  config.PeerReadBufferSize,
-		WriteBufferSize: config.PeerWriteBufferSize,
-		Metrics:         peerMetrics,
-		MessageCreator:  msgCreator,
-
+		ReadBufferSize:       config.PeerReadBufferSize,
+		WriteBufferSize:      config.PeerWriteBufferSize,
+		Metrics:              peerMetrics,
+		MessageCreator:       msgCreator,
 		Log:                  log,
 		InboundMsgThrottler:  inboundMsgThrottler,
 		Network:              nil, // This is set below.
@@ -284,6 +286,7 @@ func NewNetwork(
 
 	onCloseCtx, cancel := context.WithCancel(context.Background())
 	n := &network{
+		startupTime:          time.Now(),
 		config:               config,
 		peerConfig:           peerConfig,
 		metrics:              metrics,
@@ -401,6 +404,15 @@ func (n *network) HealthCheck(context.Context) (interface{}, error) {
 	details[SendFailRateKey] = sendFailRate
 	n.metrics.sendFailRate.Set(sendFailRate)
 
+	reachablePrimaryNetworkValidator := true
+	// If we're a primary network validator, make sure we have ingress connections
+	if time.Since(n.startupTime) > n.config.NoIngressValidatorConnectionGracePeriod {
+		connectedPrimaryValidatorInfo, isConnectedPrimaryValidatorErr := checkNoIngressConnections(n.config.MyNodeID, n, n.config.Validators)
+		reachablePrimaryNetworkValidator = isConnectedPrimaryValidatorErr == nil
+		details[PrimaryNetworkValidatorHealthKey] = connectedPrimaryValidatorInfo
+	}
+	healthy = healthy && reachablePrimaryNetworkValidator
+
 	// emit metrics about the lifetime of peer connections
 	n.metrics.updatePeerConnectionLifetimeMetrics()
 
@@ -427,9 +439,18 @@ func (n *network) HealthCheck(context.Context) (interface{}, error) {
 	if !isMsgFailRate {
 		errorReasons = append(errorReasons, fmt.Sprintf("messages failure send rate %g > %g", sendFailRate, n.config.HealthConfig.MaxSendFailRate))
 	}
+
+	if !reachablePrimaryNetworkValidator {
+		errorReasons = append(errorReasons, ErrNoIngressConnections.Error())
+	}
+
 	return details, fmt.Errorf("network layer is unhealthy reason: %s", strings.Join(errorReasons, ", "))
 }
 
+func (n *network) IngressConnCount() int {
+	return int(n.peerConfig.IngressConnectionCount.Load())
+}
+
 // Connected is called after the peer finishes the handshake.
 // Will not be called after [Disconnected] is called with this peer.
 func (n *network) Connected(nodeID ids.NodeID) {
@@ -630,7 +651,7 @@ func (n *network) Dispatch() error {
 				zap.Stringer("peerIP", ip),
 			)
 
-			if err := n.upgrade(conn, n.serverUpgrader); err != nil {
+			if err := n.upgrade(conn, n.serverUpgrader, true); err != nil {
 				n.peerConfig.Log.Verbo("failed to upgrade connection",
 					zap.String("direction", "inbound"),
 					zap.Error(err),
@@ -977,7 +998,7 @@ func (n *network) dial(nodeID ids.NodeID, ip *trackedIP) {
 				zap.Stringer("peerIP", ip.ip),
 			)
 
-			err = n.upgrade(conn, n.clientUpgrader)
+			err = n.upgrade(conn, n.clientUpgrader, false)
 			if err != nil {
 				n.peerConfig.Log.Verbo(
 					"failed to upgrade, attempting again",
@@ -1000,7 +1021,7 @@ func (n *network) dial(nodeID ids.NodeID, ip *trackedIP) {
 // If the connection is desired by the node, then the resulting upgraded
 // connection will be used to create a new peer. Otherwise the connection will
 // be immediately closed.
-func (n *network) upgrade(conn net.Conn, upgrader peer.Upgrader) error {
+func (n *network) upgrade(conn net.Conn, upgrader peer.Upgrader, isIngress bool) error {
 	upgradeTimeout := n.peerConfig.Clock.Time().Add(n.config.ReadHandshakeTimeout)
 	if err := conn.SetReadDeadline(upgradeTimeout); err != nil {
 		_ = conn.Close()
@@ -1100,6 +1121,7 @@ func (n *network) upgrade(conn net.Conn, upgrader peer.Upgrader) error {
 			n.peerConfig.Log,
 			n.outboundMsgThrottler,
 		),
+		isIngress,
 	)
 	n.connectingPeers.Add(peer)
 	n.peersLock.Unlock()

network/network_test.go

@@ -282,6 +282,11 @@ func newFullyConnectedTestNetwork(t *testing.T, handlers []router.InboundHandler
 		if i != 0 {
 			config := configs[0]
 			net.ManuallyTrack(config.MyNodeID, config.MyIPPort.Get())
+			// Wait until the node is connected to the first node.
+			// This forces nodes to connect to each other in a deterministic order.
+			require.Eventually(func() bool {
+				return len(net.PeerInfo([]ids.NodeID{config.MyNodeID})) > 0
+			}, 10*time.Second, time.Millisecond)
 		}
 
 		go func(net Network) {
@@ -306,6 +311,50 @@ func TestNewNetwork(t *testing.T) {
 	wg.Wait()
 }
 
+func TestIngressConnCount(t *testing.T) {
+	require := require.New(t)
+
+	emptyHandler := func(context.Context, message.InboundMessage) {}
+
+	_, networks, wg := newFullyConnectedTestNetwork(
+		t, []router.InboundHandler{
+			router.InboundHandlerFunc(emptyHandler),
+			router.InboundHandlerFunc(emptyHandler),
+			router.InboundHandlerFunc(emptyHandler),
+		})
+
+	wg.Done()
+
+	for _, net := range networks {
+		net.config.NoIngressValidatorConnectionGracePeriod = 0
+		net.config.HealthConfig.Enabled = true
+	}
+
+	require.Eventually(func() bool {
+		result := true
+		for _, net := range networks {
+			result = result && len(net.PeerInfo(nil)) == len(networks)-1
+		}
+		return result
+	}, time.Minute, time.Millisecond*10)
+
+	ingressConnections := make([]int, 0, len(networks))
+	healthCheckErrors := make([]error, 0, len(networks))
+
+	for _, net := range networks {
+		ingressConnections = append(ingressConnections, net.IngressConnCount())
+		_, err := net.HealthCheck(context.Background())
+		healthCheckErrors = append(healthCheckErrors, err)
+	}
+
+	// First node has all nodes connected to it.
+	// Second node has only the third node connected to it.
+	// Third node has no node connected to it, as it connects to the first and second node.
+	require.Equal([]int{2, 1, 0}, ingressConnections)
+	require.Equal([]error{nil, nil}, healthCheckErrors[:2])
+	require.ErrorContains(healthCheckErrors[2], ErrNoIngressConnections.Error()) //nolint
+}
+
 func TestSend(t *testing.T) {
 	require := require.New(t)
 

network/no_ingress_conn_alert.go

@@ -0,0 +1,39 @@
+// Copyright (C) 2019-2024, Ava Labs, Inc. All rights reserved.
+// See the file LICENSE for licensing terms.
+
+package network
+
+import (
+	"errors"
+
+	"github.com/ava-labs/avalanchego/ids"
+	"github.com/ava-labs/avalanchego/snow/validators"
+	"github.com/ava-labs/avalanchego/utils/constants"
+)
+
+// ErrNoIngressConnections denotes that no node is connected to this validator.
+var ErrNoIngressConnections = errors.New("primary network validator has no inbound connections")
+
+type ingressConnectionCounter interface {
+	IngressConnCount() int
+}
+
+type validatorRetriever interface {
+	GetValidator(subnetID ids.ID, nodeID ids.NodeID) (*validators.Validator, bool)
+}
+
+func checkNoIngressConnections(selfID ids.NodeID, ingressConnections ingressConnectionCounter, validators validatorRetriever) (interface{}, error) {
+	connCount := ingressConnections.IngressConnCount()
+	_, areWeValidator := validators.GetValidator(constants.PrimaryNetworkID, selfID)
+
+	result := map[string]interface{}{
+		"ingressConnectionCount":  connCount,
+		"primaryNetworkValidator": areWeValidator,
+	}
+
+	if connCount > 0 || !areWeValidator {
+		return result, nil
+	}
+
+	return result, ErrNoIngressConnections
+}

network/no_ingress_conn_alert_test.go

@@ -0,0 +1,65 @@
+// Copyright (C) 2019-2024, Ava Labs, Inc. All rights reserved.
+// See the file LICENSE for licensing terms.
+
+package network
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/ava-labs/avalanchego/ids"
+	"github.com/ava-labs/avalanchego/snow/validators"
+)
+
+type fakeValidatorRetriever struct {
+	result bool
+}
+
+func (m *fakeValidatorRetriever) GetValidator(ids.ID, ids.NodeID) (*validators.Validator, bool) {
+	return nil, m.result
+}
+
+type fakeIngressConnectionCounter struct {
+	res int
+}
+
+func (m *fakeIngressConnectionCounter) IngressConnCount() int {
+	return m.res
+}
+
+func TestNoIngressConnAlertHealthCheck(t *testing.T) {
+	for _, testCase := range []struct {
+		name                   string
+		getValidatorResult     bool
+		ingressConnCountResult int
+		expectedErr            error
+		expectedResult         interface{}
+	}{
+		{
+			name:           "not a validator of a primary network",
+			expectedResult: map[string]interface{}{"ingressConnectionCount": 0, "primaryNetworkValidator": false},
+		},
+		{
+			name:               "a validator of the primary network",
+			getValidatorResult: true,
+			expectedResult: map[string]interface{}{
+				"ingressConnectionCount": 0, "primaryNetworkValidator": true,
+			},
+			expectedErr: ErrNoIngressConnections,
+		},
+		{
+			name:                   "a validator with ingress connections",
+			expectedResult:         map[string]interface{}{"ingressConnectionCount": 42, "primaryNetworkValidator": true},
+			expectedErr:            nil,
+			ingressConnCountResult: 42,
+			getValidatorResult:     true,
+		},
+	} {
+		t.Run(testCase.name, func(t *testing.T) {
+			result, err := checkNoIngressConnections(ids.EmptyNodeID, &fakeIngressConnectionCounter{res: testCase.ingressConnCountResult}, &fakeValidatorRetriever{result: testCase.getValidatorResult})
+			require.Equal(t, testCase.expectedErr, err)
+			require.Equal(t, testCase.expectedResult, result)
+		})
+	}
+}