Support LATENCY LATEST, LATEST HISTORY command parsing

redis/redis.go

@@ -187,3 +187,27 @@ type SlowLog struct {
 	// ClientName is the name set via the CLIENT SETNAME command (4.0 only).
 	ClientName string
 }
+
+// Latency represents a redis LATENCY LATEST
+type Latency struct {
+	// Latency event name
+	Event string
+
+	// Time is the unix timestamp at which the event was processed.
+	Time time.Time
+
+	// LatestExecutationTime is the amount of time needed for the command execution.
+	LatestExecutionTime time.Duration
+
+	// Records the all-time maximum latency for every element
+	MaxExecutionTime time.Duration
+}
+
+// LatencyHistory represents a redis LATENCY HISTORY
+type LatencyHistory struct {
+	// Time is the unix timestamp at which the event was processed.
+	Time time.Time
+
+	// ExecutationTime is the amount of time needed for the command execution.
+	ExecutionTime time.Duration
+}

redis/reply.go

@@ -645,3 +645,88 @@ func SlowLogs(result interface{}, err error) ([]SlowLog, error) {
 	}
 	return logs, nil
 }
+
+// Latency is a helper that parse the LATENCY LATEST command output and
+// return the array of Latency
+func Latencies(result interface{}, err error) ([]Latency, error) {
+	rawLatencies, err := Values(result, err)
+	if err != nil {
+		return nil, err
+	}
+	latencies := make([]Latency, len(rawLatencies))
+	for i, e := range rawLatencies {
+		rawLatency, ok := e.([]interface{})
+		if !ok {
+			return nil, fmt.Errorf("redigo: rawLatency element is not an array, got %T", e)
+		}
+
+		var event Latency
+		if len(rawLatency) != 4 {
+			return nil, fmt.Errorf("redigo: LATENCY LATEST element has %d elements, expected 4", len(rawLatency))
+		}
+
+		event.Event, err = String(rawLatency[0], nil)
+		if err != nil {
+			return nil, fmt.Errorf("redigo: LATENCY LATEST element[0] is not a string: %w", err)
+		}
+
+		timestamp, ok := rawLatency[1].(int64)
+		if !ok {
+			return nil, fmt.Errorf("redigo: LATENCY LATEST element[1] not an int64, got %T", rawLatency[1])
+		}
+
+		event.Time = time.Unix(timestamp, 0)
+
+		latestDuration, ok := rawLatency[2].(int64)
+		if !ok {
+			return nil, fmt.Errorf("redigo: LATENCY LATEST element[2] not an int64, got %T", rawLatency[2])
+		}
+
+		event.LatestExecutionTime = time.Duration(latestDuration) * time.Millisecond
+
+		maxDuration, ok := rawLatency[3].(int64)
+		if !ok {
+			return nil, fmt.Errorf("redigo: LATENCY LATEST element[3] not an int64, got %T", rawLatency[3])
+		}
+
+		event.MaxExecutionTime = time.Duration(maxDuration) * time.Millisecond
+
+		latencies[i] = event
+	}
+	return latencies, nil
+}
+
+// LatencyHistories is a helper that parse the LATENCY HISTORY command output and
+// return the array of LatencyHistory
+func LatencyHistories(result interface{}, err error) ([]LatencyHistory, error) {
+	rawLogs, err := Values(result, err)
+	if err != nil {
+		return nil, err
+	}
+	latencyHistories := make([]LatencyHistory, len(rawLogs))
+	for i, e := range rawLogs {
+		rawLog, ok := e.([]interface{})
+		if !ok {
+			return nil, fmt.Errorf("LATENCY HISTORY: latencyHistory element is not an array, got %T", e)
+		}
+
+		var event LatencyHistory
+
+		timestamp, ok := rawLog[0].(int64)
+		if !ok {
+			return nil, fmt.Errorf("redigo: LATENCY HISTORY element[0] not an int64, got %T", rawLog[0])
+		}
+
+		event.Time = time.Unix(timestamp, 0)
+
+		duration, ok := rawLog[1].(int64)
+		if !ok {
+			return nil, fmt.Errorf("redigo: LATENCY HISTORY element[1] not an int64, got %T", rawLog[1])
+		}
+
+		event.ExecutionTime = time.Duration(duration) * time.Millisecond
+
+		latencyHistories[i] = event
+	}
+	return latencyHistories, nil
+}

redis/reply_test.go

@@ -225,6 +225,130 @@ func TestSlowLog(t *testing.T) {
 	}
 }
 
+func TestLatency(t *testing.T) {
+	c, err := dial()
+	if err != nil {
+		t.Errorf("TestLatency failed during dial with error " + err.Error())
+		return
+	}
+	defer c.Close()
+
+	resultStr, err := redis.Strings(c.Do("CONFIG", "GET", "latency-monitor-threshold"))
+	if err != nil {
+		t.Errorf("TestLatency failed during CONFIG GET latency-monitor-threshold with error " + err.Error())
+		return
+	}
+	// in case of older verion < 2.8.13 where LATENCY command is not supported
+	// don't run the test
+	if len(resultStr) == 0 {
+		return
+	}
+	latencyMonitorThresholdOldCfg, err := strconv.Atoi(resultStr[1])
+	if err != nil {
+		t.Errorf("TestLatency failed during strconv.Atoi with error " + err.Error())
+		return
+	}
+	// Enable latency monitoring for events that take 1ms or longer
+	result, err := c.Do("CONFIG", "SET", "latency-monitor-threshold", "1")
+	if err != nil && result != "OK" {
+		t.Errorf("TestLatency failed during CONFIG SET with error " + err.Error())
+		return
+	}
+	// Sleep for 1ms to register a slow event
+	_, err = c.Do("DEBUG", "SLEEP", 0.001)
+	if err != nil {
+		t.Errorf("TestLatency failed during DEBUG SLEEP 0.001 with error " + err.Error())
+		return
+	}
+
+	result, err = c.Do("LATENCY", "LATEST")
+	if err != nil {
+		t.Errorf("TestLatency failed during LATENCY LATEST with error " + err.Error())
+		return
+	}
+	latestLatencies, err := redis.Latencies(result, err)
+	if err != nil {
+		t.Errorf("TestLatency failed during redis.Latencies with error " + err.Error())
+		return
+	}
+	latencyEvent := latestLatencies[0]
+	if len(latestLatencies) != 1 ||
+		latencyEvent.Event != "command" ||
+		latencyEvent.LatestExecutionTime != time.Millisecond ||
+		latencyEvent.MaxExecutionTime != time.Millisecond {
+		t.Errorf("%s=%+v, want %+v", "TestLatency test failed : ",
+			fmt.Sprint(len(latestLatencies))+" "+fmt.Sprint(latencyEvent.Event)+" "+fmt.Sprint(latencyEvent.LatestExecutionTime)+" "+fmt.Sprint(latencyEvent.MaxExecutionTime), "1 command 1ms 1ms")
+	}
+
+	// reset the old configuration after test
+	result, err = c.Do("CONFIG", "SET", "latency-monitor-threshold", latencyMonitorThresholdOldCfg)
+	if err != nil && result != "OK" {
+		t.Errorf("TestLatency failed during CONFIG SET with error " + err.Error())
+		return
+	}
+}
+
+func TestLatencyHistories(t *testing.T) {
+	c, err := dial()
+	if err != nil {
+		t.Errorf("TestLatencyHistories failed during dial with error " + err.Error())
+		return
+	}
+	defer c.Close()
+
+	resultStr, err := redis.Strings(c.Do("CONFIG", "GET", "latency-monitor-threshold"))
+	if err != nil {
+		t.Errorf("TestLatencyHistories failed during CONFIG GET latency-monitor-threshold with error " + err.Error())
+		return
+	}
+	// in case of older verion < 2.8.13 where LATENCY command is not supported
+	// don't run the test
+	if len(resultStr) == 0 {
+		return
+	}
+	latencyMonitorThresholdOldCfg, err := strconv.Atoi(resultStr[1])
+	if err != nil {
+		t.Errorf("TestLatencyHistories failed during strconv.Atoi with error " + err.Error())
+		return
+	}
+	// Enable latency monitoring for events that take 1ms or longer
+	result, err := c.Do("CONFIG", "SET", "latency-monitor-threshold", "1")
+	if err != nil && result != "OK" {
+		t.Errorf("TestLatencyHistories failed during CONFIG SET with error " + err.Error())
+		return
+	}
+	// Sleep for 1ms to register a slow event
+	_, err = c.Do("DEBUG", "SLEEP", 0.001)
+	if err != nil {
+		t.Errorf("TestLatencyHistories failed during DEBUG SLEEP 0.001 with error " + err.Error())
+		return
+	}
+
+	result, err = c.Do("LATENCY", "HISTORY", "command")
+	if err != nil {
+		t.Errorf("TestLatencyHistories failed during LATENCY HISTORY command with error " + err.Error())
+		return
+	}
+	latencyHistory, err := redis.LatencyHistories(result, err)
+	if err != nil {
+		t.Errorf("TestLatencyHistories failed during redis.LatencyHistories with error " + err.Error())
+		return
+	}
+	latencyEvent := latencyHistory[0]
+	if len(latencyHistory) != 1 ||
+		latencyEvent.ExecutionTime != time.Millisecond {
+		t.Errorf("%s=%+v, want %+v", "TestLatencyHistories test failed : ",
+			fmt.Sprint(len(latencyHistory))+" "+fmt.Sprint(latencyEvent.ExecutionTime), "1 1ms")
+	}
+
+	// reset the old configuration after test
+	result, err = c.Do("CONFIG", "SET", "latency-monitor-threshold", latencyMonitorThresholdOldCfg)
+	if err != nil && result != "OK" {
+		t.Errorf("TestLatencyHistories failed during CONFIG SET with error " + err.Error())
+		return
+	}
+}
+
 // dial wraps DialDefaultServer() with a more suitable function name for examples.
 func dial() (redis.Conn, error) {
 	return redis.DialDefaultServer()