Cherry-pick #12741 to 7.2: [Packetbeat] Redis: Limit memory used by replication

CHANGELOG.next.asciidoc

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

packetbeat/_meta/beat.reference.yml

@@ -333,6 +333,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

@@ -1336,6 +1336,39 @@ Valid values are `sha1`, `sha256` and `md5`.
 If `send_certificates` is false, this setting is ignored. The default is to
 output SHA-1 fingerprints.
 
+[[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

@@ -333,6 +333,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)
+		})
+	}
+}