Add support for tracking Redis client calls for non-Go programs

Makefile

@@ -282,13 +282,18 @@ oats-test-sql-other-langs: oats-prereq
 	mkdir -p test/oats/sql_other_langs/$(TEST_OUTPUT)/run
 	cd test/oats/sql_other_langs && TESTCASE_BASE_PATH=./yaml $(GINKGO) -v -r
 
+.PHONY: oats-test-redis-other-langs
+oats-test-redis-other-langs: oats-prereq
+	mkdir -p test/oats/redis_other_langs/$(TEST_OUTPUT)/run
+	cd test/oats/redis_other_langs && TESTCASE_BASE_PATH=./yaml $(GINKGO) -v -r
+
 .PHONY: oats-test
-oats-test: oats-test-sql oats-test-sql-statement oats-test-sql-other-langs
+oats-test: oats-test-sql oats-test-sql-statement oats-test-sql-other-langs oats-test-redis-other-langs
 	$(MAKE) itest-coverage-data
 
 .PHONY: oats-test-debug
 oats-test-debug: oats-prereq
-	cd test/oats/sql && TESTCASE_BASE_PATH=./yaml TESTCASE_MANUAL_DEBUG=true TESTCASE_TIMEOUT=1h $(GINKGO) -v -r
+	cd test/oats/redis_other_langs && TESTCASE_BASE_PATH=./yaml TESTCASE_MANUAL_DEBUG=true TESTCASE_TIMEOUT=1h $(GINKGO) -v -r
 
 .PHONY: drone
 drone:

bpf/http_sock.h

@@ -552,6 +552,7 @@ static __always_inline void handle_unknown_tcp_connection(pid_connection_info_t
             bpf_dbg_printk("Sending TCP trace %lx, response length %d", existing, existing->resp_len);
 
             bpf_memcpy(trace, existing, sizeof(tcp_req_t));
+            bpf_probe_read(trace->rbuf, K_TCP_RES_LEN, u_buf);
             bpf_ringbuf_submit(trace, get_flags());
         }
         bpf_map_delete_elem(&ongoing_tcp_req, pid_conn);

bpf/http_types.h

@@ -15,6 +15,7 @@
 #define KPROBES_LARGE_RESPONSE_LEN 100000 // 100K and above we try to track the response actual time with kretprobes
 
 #define K_TCP_MAX_LEN 256
+#define K_TCP_RES_LEN 24
 
 #define CONN_INFO_FLAG_TRACE 0x1
 
@@ -96,6 +97,7 @@ typedef struct tcp_req {
     u64 start_monotime_ns;
     u64 end_monotime_ns;
     unsigned char buf[K_TCP_MAX_LEN] __attribute__ ((aligned (8))); // ringbuffer memcpy complains unless this is 8 byte aligned
+    unsigned char rbuf[K_TCP_RES_LEN] __attribute__ ((aligned (8))); // ringbuffer memcpy complains unless this is 8 byte aligned
     u32 len;
     u32 resp_len;
     u8  ssl;

docs/sources/metrics.md

@@ -24,6 +24,7 @@ The following table describes the exported metrics in both OpenTelemetry and Pro
 | Application | `rpc.client.duration`           | `rpc_client_duration_seconds`          | Histogram | seconds | Duration of GRPC service calls from the client side                              |
 | Application | `rpc.server.duration`           | `rpc_server_duration_seconds`          | Histogram | seconds | Duration of RPC service calls from the server side                               |
 | Application | `sql.client.duration`           | `sql_client_duration_seconds`          | Histogram | seconds | Duration of SQL client operations (Experimental)                                 |
+| Application | `redis.client.duration`         | `redis_client_duration_seconds`        | Histogram | seconds | Duration of Redis client operations (Experimental)                               |
 | Network     | `beyla.network.flow.bytes`      | `beyla_network_flow_bytes`             | Counter   | bytes   | Bytes submitted from a source network endpoint to a destination network endpoint |
 
 Beyla can also export [Span metrics](/docs/tempo/latest/metrics-generator/span_metrics/) and
@@ -64,7 +65,9 @@ default, check the `attributes`->`select` section in the [configuration document
 | Application (server)       | `client.address`            | hidden                                        |
 | `beyla.network.flow.bytes` | `beyla.ip`                  | hidden                                        |
 | `sql.client.duration`      | `db.operation`              | shown                                         |
-| `sql.client.duration`      | `db.statement`              | shown                                         |
+| `sql.client.duration`      | `db.statement`              | hidden                                        |
+| `redis.client.duration`    | `db.operation`              | shown                                         |
+| `redis.client.duration`    | `db.statement`              | hidden                                        |
 | `beyla.network.flow.bytes` | `direction`                 | hidden                                        |
 | `beyla.network.flow.bytes` | `dst.address`               | hidden                                        |
 | `beyla.network.flow.bytes` | `dst.cidr`                  | shown if the `cidrs` configuration is defined |

pkg/internal/ebpf/common/redis_detect_transform.go

@@ -0,0 +1,149 @@
+package ebpfcommon
+
+import (
+	"bytes"
+	"strings"
+
+	trace2 "go.opentelemetry.io/otel/trace"
+
+	"github.com/grafana/beyla/pkg/internal/request"
+)
+
+const minRedisFrameLen = 3
+
+func isRedis(buf []uint8) bool {
+	if len(buf) < minRedisFrameLen {
+		return false
+	}
+
+	return isRedisOp(buf)
+}
+
+// nolint:cyclop
+func isRedisOp(buf []uint8) bool {
+	if len(buf) == 0 {
+		return false
+	}
+	c := buf[0]
+
+	switch c {
+	case '+':
+		return crlfTerminatedMatch(buf[1:], func(c uint8) bool {
+			return (c >= 'A' && c <= 'Z') || (c >= 'a' && c <= 'z') || c == '.' || c == ' ' || c == '-' || c == '_'
+		})
+	case '-':
+		return isRedisError(buf[1:])
+	case ':', '$', '*':
+		return crlfTerminatedMatch(buf[1:], func(c uint8) bool {
+			return (c >= '0' && c <= '9')
+		})
+	}
+
+	return false
+}
+
+func isRedisError(buf []uint8) bool {
+	return bytes.HasPrefix(buf, []byte("ERR ")) ||
+		bytes.HasPrefix(buf, []byte("WRONGTYPE ")) ||
+		bytes.HasPrefix(buf, []byte("MOVED ")) ||
+		bytes.HasPrefix(buf, []byte("ASK ")) ||
+		bytes.HasPrefix(buf, []byte("BUSY ")) ||
+		bytes.HasPrefix(buf, []byte("NOSCRIPT ")) ||
+		bytes.HasPrefix(buf, []byte("CLUSTERDOWN "))
+}
+
+func crlfTerminatedMatch(buf []uint8, matches func(c uint8) bool) bool {
+	cr := false
+	i := 0
+	for ; i < len(buf); i++ {
+		c := buf[i]
+		if matches(c) {
+			continue
+		}
+		if c == '\r' {
+			cr = true
+			break
+		}
+
+		return false
+	}
+
+	if !cr || i >= len(buf)-1 {
+		return false
+	}
+
+	return buf[i+1] == '\n'
+}
+
+func parseRedisRequest(buf string) (string, string, bool) {
+	lines := strings.Split(buf, "\r\n")
+
+	if len(lines) < 2 {
+		return "", "", false
+	}
+
+	// It's not a command, something else?
+	if lines[0][0] != '*' {
+		return "", "", true
+	}
+
+	op := ""
+	text := ""
+
+	read := false
+	// Skip the first line
+	for _, l := range lines[1:] {
+		if len(l) == 0 {
+			continue
+		}
+		if !read {
+			if isRedisOp([]uint8(l + "\r\n")) {
+				read = true
+			} else {
+				break
+			}
+		} else {
+			if op == "" {
+				op = l
+			}
+			text += l + " "
+			read = false
+		}
+	}
+
+	return op, text, true
+}
+
+func TCPToRedisToSpan(trace *TCPRequestInfo, op, text string, status int) request.Span {
+	peer := ""
+	hostname := ""
+	hostPort := 0
+
+	if trace.ConnInfo.S_port != 0 || trace.ConnInfo.D_port != 0 {
+		peer, hostname = trace.reqHostInfo()
+		hostPort = int(trace.ConnInfo.D_port)
+	}
+
+	return request.Span{
+		Type:          request.EventTypeRedisClient,
+		Method:        op,
+		Path:          text,
+		Peer:          peer,
+		Host:          hostname,
+		HostPort:      hostPort,
+		ContentLength: 0,
+		RequestStart:  int64(trace.StartMonotimeNs),
+		Start:         int64(trace.StartMonotimeNs),
+		End:           int64(trace.EndMonotimeNs),
+		Status:        status,
+		TraceID:       trace2.TraceID(trace.Tp.TraceId),
+		SpanID:        trace2.SpanID(trace.Tp.SpanId),
+		ParentSpanID:  trace2.SpanID(trace.Tp.ParentId),
+		Flags:         trace.Tp.Flags,
+		Pid: request.PidInfo{
+			HostPID:   trace.Pid.HostPid,
+			UserPID:   trace.Pid.UserPid,
+			Namespace: trace.Pid.Ns,
+		},
+	}
+}

pkg/internal/ebpf/common/redis_detect_transform_test.go

@@ -0,0 +1,29 @@
+package ebpfcommon
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+type crlfTest struct {
+	testStr string
+	result  bool
+}
+
+func TestCRLFMatching(t *testing.T) {
+	for _, ts := range []crlfTest{
+		{testStr: "Not a sql or any known protocol", result: false},
+		{testStr: "Not a sql or any known protocol\r\n", result: true},
+		{testStr: "123\r\n", result: false},
+		{testStr: "\r\n", result: true},
+		{testStr: "\n", result: false},
+		{testStr: "\r", result: false},
+		{testStr: "", result: false},
+	} {
+		res := crlfTerminatedMatch([]uint8(ts.testStr), func(c uint8) bool {
+			return (c >= 'A' && c <= 'Z') || (c >= 'a' && c <= 'z') || c == '.' || c == ' ' || c == '-' || c == '_'
+		})
+		assert.Equal(t, res, ts.result)
+	}
+}

pkg/internal/ebpf/common/tcp_detect_transform.go

@@ -33,10 +33,22 @@ func ReadTCPRequestIntoSpan(record *ringbuf.Record) (request.Span, bool, error)
 
 	// Check if we have a SQL statement
 	sqlIndex := isSQL(buf)
-	if sqlIndex >= 0 {
+	switch {
+	case sqlIndex >= 0:
 		return TCPToSQLToSpan(&event, buf[sqlIndex:]), false, nil
-	} else if isHTTP2(b, &event) {
+	case isHTTP2(b, &event):
 		MisclassifiedEvents <- MisclassifiedEvent{EventType: EventTypeKHTTP2, TCPInfo: &event}
+	case isRedis(event.Buf[:l]) && isRedis(event.Rbuf[:]):
+		op, text, ok := parseRedisRequest(buf)
+
+		if ok {
+			status := 0
+			if isErr := isRedisError(event.Rbuf[:]); isErr {
+				status = 1
+			}
+
+			return TCPToRedisToSpan(&event, op, text, status), false, nil
+		}
 	}
 
 	return request.Span{}, true, nil // ignore if we couldn't parse it

pkg/internal/ebpf/common/tcp_detect_transform_test.go

@@ -89,6 +89,64 @@ func TestReadTCPRequestIntoSpan_Overflow(t *testing.T) {
 	assert.Equal(t, "foo", span.Path)
 }
 
+func TestRedisDetection(t *testing.T) {
+	for _, s := range []string{
+		`*2|$3|GET|$5|beyla|`,
+		`*2|$7|HGETALL|$16|users_sessions`,
+		`*8|$4|name|$4|John|`,
+		`+OK|`,
+		"-ERR ",
+		":123|",
+		"-WRONGTYPE ",
+		"-MOVED ",
+	} {
+		lines := strings.Split(s, "|")
+		test := strings.Join(lines, "\r\n")
+		assert.True(t, isRedis([]uint8(test)))
+		assert.True(t, isRedisOp([]uint8(test)))
+	}
+
+	for _, s := range []string{
+		"",
+		`*2`,
+		`*$7`,
+		`+OK`,
+		"-ERR",
+		"-WRONGTYPE",
+	} {
+		lines := strings.Split(s, "|")
+		test := strings.Join(lines, "\r\n")
+		assert.False(t, isRedis([]uint8(test)))
+		assert.False(t, isRedisOp([]uint8(test)))
+	}
+}
+
+func TestRedisParsing(t *testing.T) {
+	proper := fmt.Sprintf("*2\r\n$3\r\nGET\r\n$5\r\n%s", "beyla")
+
+	op, text, ok := parseRedisRequest(proper)
+	assert.True(t, ok)
+	assert.Equal(t, "GET", op)
+	assert.Equal(t, "GET beyla ", text)
+
+	weird := fmt.Sprintf("*2\r\nGET\r\n%s", "beyla")
+	op, text, ok = parseRedisRequest(weird)
+	assert.True(t, ok)
+	assert.Equal(t, "", op)
+	assert.Equal(t, "", text)
+
+	unknown := fmt.Sprintf("2\r\nGET\r\n%s", "beyla")
+	op, text, ok = parseRedisRequest(unknown)
+	assert.True(t, ok)
+	assert.Equal(t, "", op)
+	assert.Equal(t, "", text)
+
+	op, text, ok = parseRedisRequest("2")
+	assert.False(t, ok)
+	assert.Equal(t, "", op)
+	assert.Equal(t, "", text)
+}
+
 const charset = "\\0\\1\\2abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
 
 func randomString(length int) string {