Implement sane limits on Labels.

swim/labels.go

@@ -1,5 +1,141 @@
 package swim
 
+import (
+	"errors"
+	"strings"
+
+	"github.com/uber/ringpop-go/util"
+)
+
+var (
+	// DefaultLabelOptions contain the default values to be used to limit the
+	// amount of data being gossipped over the network. The defaults have been
+	// chosen with the following assumptions.
+	// 1. 1000 node cluster
+	// 2. Every byte available is used
+	// 3. Worst case would have continues fullsync's, meaning the complete
+	//    memberlist being sent over the wire 5 times a second.
+	// When all contiditions are met the Labels would add the following load
+	// (non-compressed) to the network:
+	//    (32+128)*5*1000*5*8 = ~32mbit
+	DefaultLabelOptions = LabelOptions{
+		KeySize:   32,
+		ValueSize: 128,
+		Count:     5,
+	}
+
+	// ErrLabelSizeExceeded indicates that an operation on labels would exceed
+	// the configured size limits on labels. This is to prevent applications
+	// bloating the gossip protocol with too much data. Ringpop can be
+	// configured with the settings to control the amount of data that can be
+	// put in a nodes labels.
+	ErrLabelSizeExceeded = errors.New("label operation exceeds configured label limits")
+
+	labelsPrivateNamespacePrefix = "__"
+)
+
+// LabelOptions controlls the limits on labels. Since labels are gossiped on
+// every ping/ping-req/fullsync we need to limit the amount of data an
+// application stores in their labels. When needed the defaults can be
+// overwritten during the construction of ringpop. This should be done with care
+// to not overwhelm the network with data.
+type LabelOptions struct {
+	// KeySize is the length a key may use at most
+	KeySize int
+
+	// ValueSize is the length a value may use at most
+	ValueSize int
+
+	// Count is the number of maximum allowed (public) labels on a node
+	Count int
+}
+
+func mergeLabelOptions(opts LabelOptions, def LabelOptions) LabelOptions {
+	return LabelOptions{
+		KeySize:   util.SelectInt(opts.KeySize, def.KeySize),
+		ValueSize: util.SelectInt(opts.ValueSize, def.ValueSize),
+		Count:     util.SelectInt(opts.Count, def.Count),
+	}
+}
+
+func (lo LabelOptions) validateLabel(current map[string]string, key, value string) error {
+	if isPrivateLabel(key) {
+		// we ignore private labels in the limits
+		return nil
+	}
+
+	countAfter := 0
+	for key := range current {
+		if isPrivateLabel(key) {
+			// we ignore private labels in the limits
+			continue
+		}
+
+		countAfter++
+	}
+
+	_, has := current[key]
+	if !has {
+		// only add a count to the countAfter if the key we are looking
+		// at is a new key
+		countAfter++
+	}
+
+	if len(key) > lo.KeySize || len(value) > lo.ValueSize {
+		// if the either the key or the value of the label is bigger then
+		// the max allowed size an error is returned
+		return ErrLabelSizeExceeded
+	}
+
+	// all is ok
+	return nil
+}
+
+func (lo LabelOptions) validateLabels(current map[string]string, additional map[string]string) error {
+	countAfter := 0
+	for key := range current {
+		if isPrivateLabel(key) {
+			// we ignore private labels in the limits
+			continue
+		}
+
+		countAfter++
+	}
+
+	for key, value := range additional {
+		if isPrivateLabel(key) {
+			// we ignore private labels in the limits
+			continue
+		}
+
+		_, has := current[key]
+		if !has {
+			// only add a count to the countAfter if the key we are looking
+			// at is a new key
+			countAfter++
+		}
+
+		if len(key) > lo.KeySize || len(value) > lo.ValueSize {
+			// if the either the key or the value of the label is bigger then
+			// the max allowed size an error is returned
+			return ErrLabelSizeExceeded
+		}
+	}
+
+	if countAfter > lo.Count {
+		// if the count after exceeds the ammount of allowed labels an error is
+		// returned
+		return ErrLabelSizeExceeded
+	}
+
+	// all is ok
+	return nil
+}
+
+func isPrivateLabel(key string) bool {
+	return strings.HasPrefix(key, labelsPrivateNamespacePrefix)
+}
+
 // NodeLabels implements the ringpop.Labels interface and proxies the calls to
 // the swim.Node backing the membership protocol.
 type NodeLabels struct {

swim/memberlist.go

@@ -283,8 +283,10 @@ func (m *memberlist) SetLocalStatus(status string) {
 }
 
 func (m *memberlist) SetLocalLabel(key, value string) error {
-	// TODO implement a sane limit for the size of the labels to prevent users
-	// from impacting the performance of the gossip protocol.
+	if err := m.node.labelLimits.validateLabel(m.local.Labels, key, value); err != nil {
+		// the labels operation violates the label limits that has been configured
+		return err
+	}
 
 	// ensure that there is a labels map
 	if m.local.Labels == nil {
@@ -332,7 +334,12 @@ func (m *memberlist) LocalLabelsAsMap() map[string]string {
 // that are set in the map passed to this function and overwrite the value with
 // the value in the map. Keys that are not present in the provided map will
 // remain in the labels of this node.
-func (m *memberlist) SetLocalLabels(labels map[string]string) {
+func (m *memberlist) SetLocalLabels(labels map[string]string) error {
+	if err := m.node.labelLimits.validateLabels(m.local.Labels, labels); err != nil {
+		// the labels operation violates the label limits that has been configured
+		return err
+	}
+
 	// ensure that there is a labels map
 	if m.local.Labels == nil {
 		m.local.Labels = make(map[string]string, len(labels))
@@ -356,6 +363,8 @@ func (m *memberlist) SetLocalLabels(labels map[string]string) {
 	if changes {
 		m.postLocalUpdate()
 	}
+
+	return nil
 }
 
 // Remove a label from the local map of labels. This will trigger a reincarnation

swim/node.go

@@ -66,6 +66,8 @@ type Options struct {
 	PartitionHealPeriod           time.Duration
 	PartitionHealBaseProbabillity float64
 
+	LabelLimits LabelOptions
+
 	Clock clock.Clock
 }
 
@@ -91,6 +93,8 @@ func defaultOptions() *Options {
 		PartitionHealPeriod:           30 * time.Second,
 		PartitionHealBaseProbabillity: 3,
 
+		LabelLimits: DefaultLabelOptions,
+
 		Clock: clock.New(),
 
 		MaxReverseFullSyncJobs: 5,
@@ -107,6 +111,7 @@ func mergeDefaultOptions(opts *Options) *Options {
 	}
 
 	opts.StateTimeouts = mergeStateTimeouts(opts.StateTimeouts, def.StateTimeouts)
+	opts.LabelLimits = mergeLabelOptions(opts.LabelLimits, def.LabelLimits)
 
 	opts.MinProtocolPeriod = util.SelectDuration(opts.MinProtocolPeriod, def.MinProtocolPeriod)
 
@@ -186,6 +191,8 @@ type Node struct {
 
 	logger log.Logger
 
+	labelLimits LabelOptions
+
 	// clock is used to generate incarnation numbers; it is typically the
 	// system clock, wrapped via clock.New()
 	clock clock.Clock
@@ -216,6 +223,8 @@ func NewNode(app, address string, channel shared.SubChannel, opts *Options) *Nod
 		clock:      opts.Clock,
 	}
 
+	node.labelLimits = opts.LabelLimits
+
 	node.memberlist = newMemberlist(node)
 	node.memberiter = newMemberlistIter(node.memberlist)
 	node.stateTransitions = newStateTransitions(node, opts.StateTimeouts)