[dbnode] Add aggregator client maxBatchSize config for bigger buffer …

…for data sent to aggregator (#2166)

src/aggregator/client/config.go

@@ -45,6 +45,7 @@ type Configuration struct {
 	ShardCutoffLingerDuration  *time.Duration                 `yaml:"shardCutoffLingerDuration"`
 	Encoder                    EncoderConfiguration           `yaml:"encoder"`
 	FlushSize                  int                            `yaml:"flushSize"`
+	MaxBatchSize               int                            `yaml:"maxBatchSize"`
 	MaxTimerBatchSize          int                            `yaml:"maxTimerBatchSize"`
 	QueueSize                  int                            `yaml:"queueSize"`
 	QueueDropType              *DropType                      `yaml:"queueDropType"`
@@ -127,6 +128,9 @@ func (c *Configuration) newClientOptions(
 	if c.FlushSize != 0 {
 		opts = opts.SetFlushSize(c.FlushSize)
 	}
+	if c.MaxBatchSize != 0 {
+		opts = opts.SetMaxBatchSize(c.MaxBatchSize)
+	}
 	if c.MaxTimerBatchSize != 0 {
 		opts = opts.SetMaxTimerBatchSize(c.MaxTimerBatchSize)
 	}

src/aggregator/client/options.go

@@ -37,7 +37,9 @@ const (
 	// By default there is no limit on the timer batch size.
 	defaultMaxTimerBatchSize = 0
 
-	defaultInstanceQueueSize = 4096
+	// defaultInstanceQueueSize determines how many metrics can be buffered
+	// before it must wait for an existing batch to be flushed to an instance.
+	defaultInstanceQueueSize = 2 << 15 // ~65k
 
 	// By default traffic is cut over to shards 10 minutes before the designated
 	// cutover time in case there are issues with the instances owning the shards.
@@ -51,10 +53,10 @@ const (
 	// By default the oldest metrics in the queue are dropped when it is full.
 	defaultDropType = DropOldest
 
-	// By default set maximum batch size to 32k
-	defaultMaxBatchSize = 2 << 14
+	// By default set maximum batch size to 8mb.
+	defaultMaxBatchSize = 2 << 22
 
-	// By default write at least every 100ms
+	// By default write at least every 100ms.
 	defaultBatchFlushDeadline = 100 * time.Millisecond
 )