[tests] Support aggregator's leader/follower mode in in-process clust…

…er setup (#3925)

This commit enables the aggregators to run with RF=2 in the in-process cluster setup.

src/cluster/generated/proto/placementpb/placement.proto

@@ -96,5 +96,6 @@ message PlacementSnapshots {
 
 message Options {
   google.protobuf.BoolValue is_sharded = 1;
+  google.protobuf.BoolValue skip_port_mirroring = 2;
   // TODO: cover all the fields in src/cluster/placement/config.go
 }

src/cluster/placement/config.go

@@ -38,6 +38,7 @@ type Configuration struct {
 	IsSharded           *bool           `yaml:"isSharded"`
 	ShardStateMode      *ShardStateMode `yaml:"shardStateMode"`
 	IsMirrored          *bool           `yaml:"isMirrored"`
+	SkipPortMirroring   *bool           `yaml:"skipPortMirroring"`
 	IsStaged            *bool           `yaml:"isStaged"`
 	ValidZone           *string         `yaml:"validZone"`
 }
@@ -63,6 +64,9 @@ func (c *Configuration) NewOptions() Options {
 	if value := c.IsMirrored; value != nil {
 		opts = opts.SetIsMirrored(*value)
 	}
+	if value := c.SkipPortMirroring; value != nil {
+		opts = opts.SetSkipPortMirroring(*value)
+	}
 	if value := c.IsStaged; value != nil {
 		opts = opts.SetIsStaged(*value)
 	}
@@ -94,6 +98,10 @@ func (c Configuration) ApplyOverride(opts *placementpb.Options) Configuration {
 		isShardedValueCopy := opts.IsSharded.Value
 		c.IsSharded = &isShardedValueCopy
 	}
+	if opts.SkipPortMirroring != nil {
+		skipPortMirroringCopy := opts.SkipPortMirroring.Value
+		c.SkipPortMirroring = &skipPortMirroringCopy
+	}
 	return c
 }
 

src/cluster/placement/options.go

@@ -76,6 +76,7 @@ type options struct {
 	dryrun              bool
 	isSharded           bool
 	isMirrored          bool
+	skipPortMirroring   bool
 	isStaged            bool
 	compress            bool
 	instanceSelector    InstanceSelector
@@ -153,6 +154,15 @@ func (o options) SetIsMirrored(v bool) Options {
 	return o
 }
 
+func (o options) SkipPortMirroring() bool {
+	return o.skipPortMirroring
+}
+
+func (o options) SetSkipPortMirroring(v bool) Options {
+	o.skipPortMirroring = v
+	return o
+}
+
 func (o options) IsStaged() bool {
 	return o.isStaged
 }