From bfed1784c368b5101d107534a4d7698e134e1e2f Mon Sep 17 00:00:00 2001
From: Joseph Heyburn <34041368+jdheyburn@users.noreply.github.com>
Date: Wed, 18 Sep 2024 20:07:34 +0100
Subject: [PATCH] [feat:filebeat/input/redis/slowlog] Add Redis replication
 role to slowlogs (#40578)

Add Redis replication role to slowlogs

---------

Co-authored-by: Mauri de Souza Meneguzzo <mauri870@gmail.com>
---
 CHANGELOG.next.asciidoc             |  1 +
 filebeat/input/redis/harvester.go   | 18 ++++++++++++++----
 filebeat/tests/system/test_redis.py |  1 +
 3 files changed, 16 insertions(+), 4 deletions(-)

diff --git a/CHANGELOG.next.asciidoc b/CHANGELOG.next.asciidoc
index 23cfb81a113..bd2b258c63f 100644
--- a/CHANGELOG.next.asciidoc
+++ b/CHANGELOG.next.asciidoc
@@ -42,6 +42,7 @@ https://github.com/elastic/beats/compare/v8.8.1\...main[Check the HEAD diff]
 - Journald: removed configuration options `include_matches.or`, `include_matches.and`, `backoff`, `max_backoff`, `cursor_seek_fallback`. {pull}40061[40061]
 - Journald: `include_matches.match` now behaves in the same way as matchers in `journalctl`. Users should carefully update their input configuration. {pull}40061[40061]
 - Journald: `seek` and `since` behaviour have been simplified, if there is a cursor (state) `seek` and `since` are ignored and the cursor is used. {pull}40061[40061]
+- Redis: Added replication role as a field to submitted slowlogs
 - Added `container.image.name` to `journald` Filebeat input's Docker-specific translated fields. {pull}40450[40450]
 
 
diff --git a/filebeat/input/redis/harvester.go b/filebeat/input/redis/harvester.go
index 141c5f15773..67854bc04b1 100644
--- a/filebeat/input/redis/harvester.go
+++ b/filebeat/input/redis/harvester.go
@@ -81,29 +81,38 @@ func (h *Harvester) Run() error {
 		return nil
 	default:
 	}
-	// Writes Slowlog get and slowlog reset both to the buffer so they are executed together
+	// Writes Slowlog get, slowlog reset, and role to the buffer so they are executed together
 	if err := h.conn.Send("SLOWLOG", "GET"); err != nil {
 		return fmt.Errorf("error sending slowlog get: %w", err)
 	}
 	if err := h.conn.Send("SLOWLOG", "RESET"); err != nil {
 		return fmt.Errorf("error sending slowlog reset: %w", err)
 	}
+	if err := h.conn.Send("ROLE"); err != nil {
+		return fmt.Errorf("error sending role: %w", err)
+	}
 
-	// Flush the buffer to execute both commands and receive the reply from SLOWLOG GET
+	// Flush the buffer to execute all commands and receive the replies
 	h.conn.Flush()
 
-	// Receives first reply from redis which is the one from GET
+	// Receives first reply from redis which is the one from SLOWLOG GET
 	logs, err := rd.Values(h.conn.Receive())
 	if err != nil {
 		return fmt.Errorf("error receiving slowlog data: %w", err)
 	}
 
-	// Read reply from RESET
+	// Read reply from SLOWLOG RESET
 	_, err = h.conn.Receive()
 	if err != nil {
 		return fmt.Errorf("error receiving reset data: %w", err)
 	}
 
+	// Read reply from ROLE
+	role, err := h.conn.Receive()
+	if err != nil {
+		return fmt.Errorf("error receiving replication role: %w", err)
+	}
+
 	for _, item := range logs {
 		// Stopping here means some of the slowlog events are lost!
 		select {
@@ -146,6 +155,7 @@ func (h *Harvester) Run() error {
 			"duration": mapstr.M{
 				"us": log.duration,
 			},
+			"role": role,
 		}
 
 		if log.args != nil {
diff --git a/filebeat/tests/system/test_redis.py b/filebeat/tests/system/test_redis.py
index 5609af5f38c..831efff5506 100644
--- a/filebeat/tests/system/test_redis.py
+++ b/filebeat/tests/system/test_redis.py
@@ -47,6 +47,7 @@ def test_input(self):
 
         assert output["input.type"] == "redis"
         assert "redis.slowlog.cmd" in output
+        assert "redis.slowlog.role" in output
 
     def get_host(self):
         return os.getenv('REDIS_HOST', 'localhost')