fix #1976: Add support for configurable pinning only on untrusted peers

allocate.go

@@ -136,8 +136,10 @@ func (c *Cluster) filterMetrics(ctx context.Context, mSet api.MetricsSet, numMet
 				// discard blacklisted peers
 				continue
 			case c.config.PinOnlyOnTrustedPeers && !c.consensus.IsTrustedPeer(ctx, m.Peer):
-				// discard peer that are not trusted when
+				// discard peers that are not trusted
-				// configured.
+				continue
+			case c.config.PinOnlyOnUntrustedPeers && c.consensus.IsTrustedPeer(ctx, m.Peer):
+				// discard peers that are trusted
 				continue
 			case containsPeer(currentAllocs, m.Peer):
 				curPeersMap[m.Peer] = append(curPeersMap[m.Peer], m)

cluster_config.go

@@ -37,6 +37,7 @@ const (
 	DefaultReplicationFactor       = -1
 	DefaultLeaveOnShutdown         = false
 	DefaultPinOnlyOnTrustedPeers   = false
+	DefaultPinOnlyOnUntrustedPeers = false
 	DefaultDisableRepinning        = true
 	DefaultPeerstoreFile           = "peerstore"
 	DefaultConnMgrHighWater        = 400
@@ -141,6 +142,9 @@ type Config struct {
 	// PinOnlyOnTrustedPeers limits allocations to trusted peers only.
 	PinOnlyOnTrustedPeers bool
 
+	// PinOnlyOnUntrustedPeers limits allocations to untrusted peers only.
+	PinOnlyOnUntrustedPeers bool
+
 	// If true, DisableRepinning, ensures that no repinning happens
 	// when a node goes down.
 	// This is useful when doing certain types of maintenance, or simply
@@ -186,6 +190,7 @@ type configJSON struct {
 	PeerWatchInterval       string             `json:"peer_watch_interval"`
 	MDNSInterval            string             `json:"mdns_interval"`
 	PinOnlyOnTrustedPeers   bool               `json:"pin_only_on_trusted_peers"`
+	PinOnlyOnUntrustedPeers bool               `json:"pin_only_on_untrusted_peers"`
 	DisableRepinning        bool               `json:"disable_repinning"`
 	FollowerMode            bool               `json:"follower_mode,omitempty"`
 	PeerstoreFile           string             `json:"peerstore_file,omitempty"`
@@ -287,6 +292,10 @@ func (cfg *Config) Validate() error {
 		return errors.New("cluster.peer_watch_interval is invalid")
 	}
 
+	if cfg.PinOnlyOnTrustedPeers && cfg.PinOnlyOnUntrustedPeers {
+		return errors.New("cluster.pin_only_on_trusted_peers and pin_only_on_untrusted_peers cannot both be true")
+	}
+
 	rfMax := cfg.ReplicationFactorMax
 	rfMin := cfg.ReplicationFactorMin
 
@@ -379,6 +388,7 @@ func (cfg *Config) setDefaults() {
 	cfg.PeerWatchInterval = DefaultPeerWatchInterval
 	cfg.MDNSInterval = DefaultMDNSInterval
 	cfg.PinOnlyOnTrustedPeers = DefaultPinOnlyOnTrustedPeers
+	cfg.PinOnlyOnUntrustedPeers = DefaultPinOnlyOnUntrustedPeers
 	cfg.DisableRepinning = DefaultDisableRepinning
 	cfg.FollowerMode = DefaultFollowerMode
 	cfg.PeerstoreFile = "" // empty so it gets omitted.
@@ -469,6 +479,7 @@ func (cfg *Config) applyConfigJSON(jcfg *configJSON) error {
 	cfg.PeerAddresses = peerAddrs
 	cfg.LeaveOnShutdown = jcfg.LeaveOnShutdown
 	cfg.PinOnlyOnTrustedPeers = jcfg.PinOnlyOnTrustedPeers
+	cfg.PinOnlyOnUntrustedPeers = jcfg.PinOnlyOnUntrustedPeers
 	cfg.DisableRepinning = jcfg.DisableRepinning
 	cfg.FollowerMode = jcfg.FollowerMode
 
@@ -520,6 +531,7 @@ func (cfg *Config) toConfigJSON() (jcfg *configJSON, err error) {
 	jcfg.PeerWatchInterval = cfg.PeerWatchInterval.String()
 	jcfg.MDNSInterval = cfg.MDNSInterval.String()
 	jcfg.PinOnlyOnTrustedPeers = cfg.PinOnlyOnTrustedPeers
+	jcfg.PinOnlyOnUntrustedPeers = cfg.PinOnlyOnUntrustedPeers
 	jcfg.DisableRepinning = cfg.DisableRepinning
 	jcfg.PeerstoreFile = cfg.PeerstoreFile
 	jcfg.PeerAddresses = []string{}

cluster_config_test.go

@@ -213,6 +213,22 @@ func TestLoadJSON(t *testing.T) {
 			t.Error("default conn manager values not set")
 		}
 	})
+
+	t.Run("expected pin_only_on_untrusted_peers", func(t *testing.T) {
+		cfg, err := loadJSON2(
+			t,
+			func(j *configJSON) {
+				j.PinOnlyOnTrustedPeers = false
+				j.PinOnlyOnUntrustedPeers = true
+			},
+		)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if !cfg.PinOnlyOnUntrustedPeers {
+			t.Error("expected pin_only_on_untrusted_peers to be true")
+		}
+	})
 }
 
 func TestToJSON(t *testing.T) {
@@ -283,4 +299,11 @@ func TestValidate(t *testing.T) {
 	if cfg.Validate() == nil {
 		t.Fatal("expected error validating")
 	}
+
+	cfg.Default()
+	cfg.PinOnlyOnTrustedPeers = true
+	cfg.PinOnlyOnUntrustedPeers = true
+	if cfg.Validate() == nil {
+		t.Fatal("expected error validating")
+	}
 }