[Packetbeat] Redis: Limit memory used by replication (elastic#12741)

This patch limits the amount of memory that can be used by outstanding
requests by adding two new configuration options to the Redis protocol:

- max_queue_size: Limits the total size of the queued requests (in bytes).
- max_queue_length: Limits the number of requests queued.

These limits apply to individual connections. The defaults are to queue
up to 1MB or 20.000 requests. This is enough to limit the currently
unbounded memory used by replication streams while at the same time
allow for pipelining requests.

Closes elastic#12657

(cherry picked from commit 1275ee8)

CHANGELOG.next.asciidoc

@@ -45,6 +45,8 @@ https://github.com/elastic/beats/compare/v6.8.0...6.8.1[Check the HEAD diff]
 
 *Packetbeat*
 
+- Limit memory usage of Redis replication sessions. {issue}12657[12657]
+
 *Winlogbeat*
 
 *Functionbeat*

packetbeat/_meta/beat.reference.yml

@@ -329,6 +329,15 @@ packetbeat.protocols:
   # incoming responses, but sent to Elasticsearch immediately.
   #transaction_timeout: 10s
 
+  # Max size for per-session message queue. This places a limit on the memory
+  # that can be used to buffer requests and responses for correlation.
+  #queue_max_bytes: 1048576
+
+  # Max number of messages for per-session message queue. This limits the number
+  # of requests or responses that can be buffered for correlation. Set a value
+  # large enough to allow for pipelining.
+  #queue_max_messages: 20000
+
 - type: thrift
   # Enable thrift monitoring. Default: true
   #enabled: true

packetbeat/docs/packetbeat-options.asciidoc

@@ -1273,6 +1273,39 @@ include the raw certificate encoded in PEM format as a `raw` field.
 
 If `send_certificates` is false, this setting is ignored. The default is false.
 
+[[packetbeat-redis-options]]
+=== Capture Redis traffic
+
+++++
+<titleabbrev>Redis</titleabbrev>
+++++
+
+The Redis protocol has several specific configuration options. Here is a
+sample configuration for the `redis` section of the +{beatname_lc}.yml+ config file:
+
+[source,yaml]
+------------------------------------------------------------------------------
+packetbeat.protocols:
+- type: redis
+  ports: [6379]
+  queue_max_bytes: 1048576
+  queue_max_messages: 20000
+------------------------------------------------------------------------------
+
+==== Configuration options
+
+Also see <<common-protocol-options>>.
+
+===== `queue_max_bytes` and `queue_max_messages`
+
+In order for request/response correlation to work, {beatname_uc} needs to
+store requests in memory until a response is received. These settings impose
+a limit on the number of bytes (`queue_max_bytes`) and number of requests
+(`queue_max_messages`) that can be stored. These limits are per-connection.
+The default is to queue up to 1MB or 20.000 requests per connection, which
+allows to use request pipelining while at the same time limiting the amount
+of memory consumed by replication sessions.
+
 [[configuration-processes]]
 == Specify which processes to monitor
 

packetbeat/packetbeat.reference.yml

@@ -329,6 +329,15 @@ packetbeat.protocols:
   # incoming responses, but sent to Elasticsearch immediately.
   #transaction_timeout: 10s
 
+  # Max size for per-session message queue. This places a limit on the memory
+  # that can be used to buffer requests and responses for correlation.
+  #queue_max_bytes: 1048576
+
+  # Max number of messages for per-session message queue. This limits the number
+  # of requests or responses that can be buffered for correlation. Set a value
+  # large enough to allow for pipelining.
+  #queue_max_messages: 20000
+
 - type: thrift
   # Enable thrift monitoring. Default: true
   #enabled: true

packetbeat/protos/redis/config.go

@@ -24,12 +24,17 @@ import (
 
 type redisConfig struct {
 	config.ProtocolCommon `config:",inline"`
+	QueueLimits           MessageQueueConfig `config:",inline"`
 }
 
 var (
 	defaultConfig = redisConfig{
 		ProtocolCommon: config.ProtocolCommon{
 			TransactionTimeout: protos.DefaultTransactionExpiration,
 		},
+		QueueLimits: MessageQueueConfig{
+			MaxBytes:    1024 * 1024,
+			MaxMessages: 20000,
+		},
 	}
 )

packetbeat/protos/redis/messagequeue.go

@@ -0,0 +1,117 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package redis
+
+import (
+	"math"
+)
+
+// Message interface needs to be implemented by types in order to be stored
+// in a MessageQueue.
+type Message interface {
+	// Size returns the size of the current element.
+	Size() int
+}
+
+type listEntry struct {
+	item Message
+	next *listEntry
+}
+
+// MessageQueue defines a queue that automatically evicts messages based on
+// the total size or number of elements contained.
+type MessageQueue struct {
+	head, tail *listEntry
+	bytesAvail int64
+	slotsAvail int32
+}
+
+// MessageQueueConfig represents the configuration for a MessageQueue.
+// Setting any limit to zero disables the limit.
+type MessageQueueConfig struct {
+	// MaxBytes is the maximum number of bytes that can be stored in the queue.
+	MaxBytes int64 `config:"queue_max_bytes"`
+
+	// MaxMessages sets a limit on the number of messages that the queue can hold.
+	MaxMessages int32 `config:"queue_max_messages"`
+}
+
+// NewMessageQueue creates a new MessageQueue with the given configuration.
+func NewMessageQueue(c MessageQueueConfig) (queue MessageQueue) {
+	queue.bytesAvail = c.MaxBytes
+	if queue.bytesAvail <= 0 {
+		queue.bytesAvail = math.MaxInt64
+	}
+	queue.slotsAvail = c.MaxMessages
+	if queue.slotsAvail <= 0 {
+		queue.slotsAvail = math.MaxInt32
+	}
+	return queue
+}
+
+// Append appends a new message into the queue, returning the number of
+// messages evicted to make room, if any.
+func (ml *MessageQueue) Append(msg Message) (evicted int) {
+	size := int64(msg.Size())
+	evicted = ml.adjust(size)
+	ml.slotsAvail--
+	ml.bytesAvail -= size
+	entry := &listEntry{
+		item: msg,
+	}
+	if ml.tail == nil {
+		ml.head = entry
+	} else {
+		ml.tail.next = entry
+	}
+	ml.tail = entry
+	return evicted
+}
+
+// IsEmpty returns if the MessageQueue is empty.
+func (ml *MessageQueue) IsEmpty() bool {
+	return ml.head == nil
+}
+
+// Pop returns the oldest message in the queue, if any.
+func (ml *MessageQueue) Pop() Message {
+	if ml.head == nil {
+		return nil
+	}
+
+	msg := ml.head
+	ml.head = msg.next
+	if ml.head == nil {
+		ml.tail = nil
+	}
+	ml.slotsAvail++
+	ml.bytesAvail += int64(msg.item.Size())
+	return msg.item
+}
+
+func (ml *MessageQueue) adjust(msgSize int64) (evicted int) {
+	if ml.slotsAvail == 0 {
+		ml.Pop()
+		evicted++
+	}
+	for ml.bytesAvail < msgSize && !ml.IsEmpty() {
+		ml.Pop()
+		evicted++
+	}
+	return evicted
+}

packetbeat/protos/redis/messagequeue_test.go

@@ -0,0 +1,128 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package redis
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+type testMessage int
+
+func (t testMessage) Size() int {
+	return int(t)
+}
+
+func TestMessageList_Append(t *testing.T) {
+	for _, test := range []struct {
+		title    string
+		maxBytes int64
+		maxCount int32
+		input    []int
+		expected []int
+	}{
+		{
+			title:    "unbounded queue",
+			maxBytes: 0,
+			maxCount: 0,
+			input:    []int{1, 2, 3, 4, 5},
+			expected: []int{1, 2, 3, 4, 5},
+		},
+		{
+			title:    "count limited",
+			maxBytes: 0,
+			maxCount: 3,
+			input:    []int{1, 2, 3, 4, 5},
+			expected: []int{3, 4, 5},
+		},
+		{
+			title:    "count limit boundary",
+			maxBytes: 0,
+			maxCount: 3,
+			input:    []int{1, 2, 3},
+			expected: []int{1, 2, 3},
+		},
+		{
+			title:    "size limited",
+			maxBytes: 10,
+			maxCount: 0,
+			input:    []int{1, 2, 3, 4, 5},
+			expected: []int{4, 5},
+		},
+		{
+			title:    "size limited boundary",
+			maxBytes: 10,
+			maxCount: 0,
+			input:    []int{1, 2, 3, 4},
+			expected: []int{1, 2, 3, 4},
+		},
+		{
+			title:    "excess size",
+			maxBytes: 10,
+			maxCount: 0,
+			input:    []int{1, 2, 3, 100},
+			expected: []int{100},
+		},
+		{
+			title:    "excess size 2",
+			maxBytes: 10,
+			maxCount: 0,
+			input:    []int{100, 1},
+			expected: []int{1},
+		},
+		{
+			title:    "excess size 3",
+			maxBytes: 10,
+			maxCount: 0,
+			input:    []int{1, 2, 3, 4, 5, 5},
+			expected: []int{5, 5},
+		},
+		{
+			title:    "both",
+			maxBytes: 10,
+			maxCount: 3,
+			input:    []int{3, 4, 2, 1},
+			expected: []int{4, 2, 1},
+		},
+	} {
+		t.Run(test.title, func(t *testing.T) {
+			conf := MessageQueueConfig{
+				MaxBytes:    test.maxBytes,
+				MaxMessages: test.maxCount,
+			}
+			q := NewMessageQueue(conf)
+			for _, elem := range test.input {
+				q.Append(testMessage(elem))
+			}
+			var result []int
+			for !q.IsEmpty() {
+				msg := q.Pop()
+				if !assert.NotNil(t, msg) {
+					t.FailNow()
+				}
+				value, ok := msg.(testMessage)
+				if !assert.True(t, ok) {
+					t.FailNow()
+				}
+				result = append(result, value.Size())
+			}
+			assert.Equal(t, test.expected, result)
+		})
+	}
+}