Add log in case softWatchLimit is exceeded during a blocking query

Author: Thibault Gilles

agent/config/builder.go

@@ -809,7 +809,7 @@ func (b *Builder) Build() (rt RuntimeConfig, err error) {
 		VerifyOutgoing:                          b.boolVal(c.VerifyOutgoing),
 		VerifyServerHostname:                    b.boolVal(c.VerifyServerHostname),
 		Watches:                                 c.Watches,
-		WatchSoftLimit:                          b.intValWithDefault(c.WatchSoftLimit, consul.DefaultSoftWatchLimit),
+		WatchSoftLimit:                          b.intValWithDefault(c.Performance.WatchSoftLimit, consul.DefaultSoftWatchLimit),
 	}
 
 	if rt.BootstrapExpect == 1 {

agent/config/config.go

@@ -254,7 +254,6 @@ type Config struct {
 	VerifyOutgoing                   *bool                    `json:"verify_outgoing,omitempty" hcl:"verify_outgoing" mapstructure:"verify_outgoing"`
 	VerifyServerHostname             *bool                    `json:"verify_server_hostname,omitempty" hcl:"verify_server_hostname" mapstructure:"verify_server_hostname"`
 	Watches                          []map[string]interface{} `json:"watches,omitempty" hcl:"watches" mapstructure:"watches"`
-	WatchSoftLimit                   *int                     `json:"watch_soft_limit,omitempty" hcl:"watch_soft_limit" mapstructure:"watch_soft_limit"`
 
 	// This isn't used by Consul but we've documented a feature where users
 	// can deploy their snapshot agent configs alongside their Consul configs
@@ -554,6 +553,7 @@ type Performance struct {
 	LeaveDrainTime *string `json:"leave_drain_time,omitempty" hcl:"leave_drain_time" mapstructure:"leave_drain_time"`
 	RaftMultiplier *int    `json:"raft_multiplier,omitempty" hcl:"raft_multiplier" mapstructure:"raft_multiplier"` // todo(fs): validate as uint
 	RPCHoldTimeout *string `json:"rpc_hold_timeout" hcl:"rpc_hold_timeout" mapstructure:"rpc_hold_timeout"`
+	WatchSoftLimit *int    `json:"watch_soft_limit,omitempty" hcl:"watch_soft_limit" mapstructure:"watch_soft_limit"`
 }
 
 type Telemetry struct {

agent/config/runtime_test.go

@@ -3007,7 +3007,8 @@ func TestFullConfig(t *testing.T) {
 			"performance": {
 				"leave_drain_time": "8265s",
 				"raft_multiplier": 5,
-				"rpc_hold_timeout": "15707s"
+				"rpc_hold_timeout": "15707s",
+				"watch_soft_limit": ` + fmt.Sprint(consul.DefaultSoftWatchLimit) + `
 			},
 			"pid_file": "43xN80Km",
 			"ports": {
@@ -3333,8 +3334,7 @@ func TestFullConfig(t *testing.T) {
 					"key": "sl3Dffu7",
 					"args": ["dltjDJ2a", "flEa7C2d"]
 				}
-			],
-			"watch_soft_limit": ` + fmt.Sprint(consul.DefaultSoftWatchLimit) + `
+			]
 		}`,
 		"hcl": `
 			acl_agent_master_token = "furuQD0b"
@@ -3541,6 +3541,7 @@ func TestFullConfig(t *testing.T) {
 				leave_drain_time = "8265s"
 				raft_multiplier = 5
 				rpc_hold_timeout = "15707s"
+				watch_soft_limit = ` + fmt.Sprint(consul.DefaultSoftWatchLimit) + `
 			}
 			pid_file = "43xN80Km"
 			ports {
@@ -3864,8 +3865,7 @@ func TestFullConfig(t *testing.T) {
 				datacenter = "fYrl3F5d"
 				key = "sl3Dffu7"
 				args = ["dltjDJ2a", "flEa7C2d"]
-			}],
-			watch_soft_limit = ` + fmt.Sprint(consul.DefaultSoftWatchLimit) + `
+			}]
 		`}
 
 	tail := map[string][]Source{

agent/connect/ca/provider_consul_test.go

@@ -46,7 +46,7 @@ func (c *consulCAMockDelegate) ApplyCARequest(req *structs.CARequest) error {
 }
 
 func newMockDelegate(t *testing.T, conf *structs.CAConfiguration) *consulCAMockDelegate {
-	s, err := state.NewStateStore(nil, 1024)
+	s, err := state.NewStateStore(nil, 1024, nil)
 	if err != nil {
 		t.Fatalf("err: %s", err)
 	}

agent/consul/fsm/fsm.go

@@ -69,14 +69,16 @@ type FSM struct {
 
 // New is used to construct a new FSM with a blank state.
 func New(gc *state.TombstoneGC, watchLimit int, logOutput io.Writer) (*FSM, error) {
-	stateNew, err := state.NewStateStore(gc, watchLimit)
+	logger := log.New(logOutput, "", log.LstdFlags)
+
+	stateNew, err := state.NewStateStore(gc, watchLimit, logger)
 	if err != nil {
 		return nil, err
 	}
 
 	fsm := &FSM{
 		logOutput:  logOutput,
-		logger:     log.New(logOutput, "", log.LstdFlags),
+		logger:     logger,
 		apply:      make(map[structs.MessageType]command),
 		state:      stateNew,
 		gc:         gc,
@@ -142,7 +144,7 @@ func (c *FSM) Restore(old io.ReadCloser) error {
 	defer old.Close()
 
 	// Create a new state store.
-	stateNew, err := state.NewStateStore(c.gc, c.watchLimit)
+	stateNew, err := state.NewStateStore(c.gc, c.watchLimit, c.logger)
 	if err != nil {
 		return err
 	}

agent/consul/state/catalog.go

@@ -837,6 +837,11 @@ func (s *Store) ServicesByNodeMeta(ws memdb.WatchSet, filters map[string]string)
 			results[service] = append(results[service], tag)
 		}
 	}
+
+	if len(ws) >= s.watchLimit {
+		s.warnSoftLimitReached("service by node meta")
+	}
+
 	return idx, results, nil
 }
 
@@ -905,6 +910,11 @@ func (s *Store) serviceNodes(ws memdb.WatchSet, serviceName string, connect bool
 	if err != nil {
 		return 0, nil, fmt.Errorf("failed parsing service nodes: %s", err)
 	}
+
+	if len(ws) >= s.watchLimit {
+		s.warnSoftLimitReached("service %s", serviceName)
+	}
+
 	return idx, results, nil
 }
 
@@ -1446,7 +1456,16 @@ func (s *Store) ServiceChecksByNodeMeta(ws memdb.WatchSet, serviceName string,
 	}
 	ws.Add(iter.WatchCh())
 
-	return s.parseChecksByNodeMeta(tx, ws, idx, iter, filters)
+	idx, checks, err := s.parseChecksByNodeMeta(tx, ws, idx, iter, filters)
+	if err != nil {
+		return 0, nil, err
+	}
+
+	if len(ws) >= s.watchLimit {
+		s.warnSoftLimitReached("service %s", serviceName)
+	}
+
+	return idx, checks, nil
 }
 
 // ChecksInState is used to query the state store for all checks
@@ -1780,7 +1799,17 @@ func (s *Store) NodeInfo(ws memdb.WatchSet, node string) (uint64, structs.NodeDu
 		return 0, nil, fmt.Errorf("failed node lookup: %s", err)
 	}
 	ws.Add(nodes.WatchCh())
-	return s.parseNodes(tx, ws, idx, nodes)
+
+	idx, nodeDump, err := s.parseNodes(tx, ws, idx, nodes)
+	if err != nil {
+		return 0, nil, err
+	}
+
+	if len(ws) >= s.watchLimit {
+		s.warnSoftLimitReached("node %s", node)
+	}
+
+	return idx, nodeDump, nil
 }
 
 // NodeDump is used to generate a dump of all nodes. This call is expensive
@@ -1863,3 +1892,12 @@ func (s *Store) parseNodes(tx *memdb.Txn, ws memdb.WatchSet, idx uint64,
 	}
 	return idx, results, nil
 }
+
+func (s *Store) warnSoftLimitReached(f string, a ...interface{}) {
+	if s.watchLimitWarnCounter%100000 > 0 {
+		return
+	}
+
+	s.logger.Printf("[WARN] consul: exceeded soft watch limit of %d for %s, falling back to coarse grained watch", s.watchLimit, fmt.Sprintf(f, a...))
+	s.watchLimitWarnCounter++
+}

agent/consul/state/catalog_test.go

@@ -1015,7 +1015,7 @@ func TestStateStore_GetNodes(t *testing.T) {
 }
 
 func BenchmarkGetNodes(b *testing.B) {
-	s, err := NewStateStore(nil, testWatchLimit)
+	s, err := NewStateStore(nil, testWatchLimit, nil)
 	if err != nil {
 		b.Fatalf("err: %s", err)
 	}
@@ -3051,7 +3051,7 @@ func TestStateStore_CheckConnectServiceNodes(t *testing.T) {
 }
 
 func BenchmarkCheckServiceNodes(b *testing.B) {
-	s, err := NewStateStore(nil, testWatchLimit)
+	s, err := NewStateStore(nil, testWatchLimit, nil)
 	if err != nil {
 		b.Fatalf("err: %s", err)
 	}

agent/consul/state/kvs_test.go

@@ -21,7 +21,7 @@ func TestStateStore_GC(t *testing.T) {
 
 	// Enable it and attach it to the state store.
 	gc.SetEnabled(true)
-	s, err := NewStateStore(gc, testWatchLimit)
+	s, err := NewStateStore(gc, testWatchLimit, nil)
 	if err != nil {
 		t.Fatalf("err: %s", err)
 	}

agent/consul/state/state_store.go

@@ -3,6 +3,8 @@ package state
 import (
 	"errors"
 	"fmt"
+	"log"
+	"os"
 
 	"github.com/hashicorp/consul/types"
 	"github.com/hashicorp/go-memdb"
@@ -65,6 +67,10 @@ type Store struct {
 	// Given the current size of aFew == 32 in memdb's watch_few.go,
 	// this will allow for up to ~ watchLimit/32 goroutines per blocking query.
 	watchLimit int
+
+	watchLimitWarnCounter uint
+
+	logger *log.Logger
 }
 
 // Snapshot is used to provide a point-in-time snapshot. It
@@ -99,14 +105,18 @@ type sessionCheck struct {
 }
 
 // NewStateStore creates a new in-memory state storage layer.
-func NewStateStore(gc *TombstoneGC, watchLimit int) (*Store, error) {
+func NewStateStore(gc *TombstoneGC, watchLimit int, logger *log.Logger) (*Store, error) {
 	// Create the in-memory DB.
 	schema := stateStoreSchema()
 	db, err := memdb.NewMemDB(schema)
 	if err != nil {
 		return nil, fmt.Errorf("Failed setting up state store: %s", err)
 	}
 
+	if logger == nil {
+		logger = log.New(os.Stderr, "", log.LstdFlags)
+	}
+
 	// Create and return the state store.
 	s := &Store{
 		schema:       schema,
@@ -115,6 +125,7 @@ func NewStateStore(gc *TombstoneGC, watchLimit int) (*Store, error) {
 		kvsGraveyard: NewGraveyard(gc),
 		lockDelay:    NewDelay(),
 		watchLimit:   watchLimit,
+		logger:       logger,
 	}
 	return s, nil
 }

agent/consul/state/state_store_test.go

@@ -30,7 +30,7 @@ func testUUID() string {
 }
 
 func testStateStore(t *testing.T) *Store {
-	s, err := NewStateStore(nil, testWatchLimit)
+	s, err := NewStateStore(nil, testWatchLimit, nil)
 	if err != nil {
 		t.Fatalf("err: %s", err)
 	}