refactor: slow response and big response

Author: c1982

README.md

@@ -21,8 +21,8 @@ Features:
 
 - [x] Count commands and arguments  
 - [ ] Count network traffic of command
-- [ ] Monitor Slow commands
-- [ ] Monitor big response
+- [x] Monitor Slow commands
+- [x] Monitor big response
 
 ## Installation
 
@@ -51,6 +51,8 @@ password | **** | Prometheus metrics password
 s | empty | Separator of keys (for split). If it empty does not split keys.
 r | empty | Regex pattern of keys (for clean)
 max | 150 | Maximum lookup size of key. If value -1 unlimited lookup. 
+slow-response-threshold | 500 | threshold for recording slow response. Millisecond
+big-response-threshold | 1500 | threshold for recording slow response. Bytes
 
 ### Grafana Dashboard
 

durations.go

@@ -20,6 +20,10 @@ type Durations struct {
 	list map[uint32]DurationItem
 }
 
+func NewDurations() *Durations {
+	return &Durations{list: make(map[uint32]DurationItem)}
+}
+
 func (d *Durations) Set(k uint32, command, args string) {
 	d.m.Lock()
 	defer d.m.Unlock()

main.go

@@ -19,8 +19,8 @@ func main() {
 		keyseparator         = flag.String("s", "", "Separator of keys (for split). If it empty does not split keys.")
 		keycleanerregex      = flag.String("r", "", "Regex pattern for cleaner in keys")
 		maxkeysizenumber     = flag.Int("max", 120, "Key size to be lookup")
-		slowresponsethresold = flag.Duration("slow-response-threshold", time.Millisecond*500, "")
-		bigresponsethreshold = flag.Int("big-response-threshold", 50000, "")
+		slowresponsethresold = flag.Duration("slow-response-threshold", time.Millisecond*500, "threshold for recording slow response. Millisecond")
+		bigresponsethreshold = flag.Int("big-response-threshold", 1500, "threshold for recording slow response. Bytes")
 	)
 
 	flag.Parse()

monitor_packet.go

@@ -11,8 +11,11 @@ import (
 
 //StartMonitor monitor redis packet destination or source port
 func StartMonitor(devicename string, redisport uint16) error {
-	bpffilter := fmt.Sprintf("port %d and tcp[((tcp[12:1] & 0xf0) >> 2):1] = 0x2A", redisport)
-	handle, err := pcap.OpenLive(devicename, 1600, false, -1*time.Second)
+	// + -> 0x2b, $ -> 0x24, * -> 0x2A
+	bpffilter := fmt.Sprintf(`port %d and tcp[((tcp[12:1] & 0xf0) >> 2):1] = 0x2A 
+	|| tcp[((tcp[12:1] & 0xf0) >> 2):1] = 0x24 
+	|| tcp[((tcp[12:1] & 0xf0) >> 2):1] = 0x2b`, redisport)
+	handle, err := pcap.OpenLive(devicename, 1800, false, -1*time.Second)
 	if err != nil {
 		return fmt.Errorf("error opening device %s: %v", devicename, err)
 	}

monitor_resp.go

@@ -27,6 +27,7 @@ func monitorRespPackets(redisport uint, sep, cleaner string, maxkeysize int, slo
 		cleanerxp = regexp.MustCompile(cleaner)
 	}
 
+	durations = NewDurations()
 	tcpchan = make(chan *layers.TCP, 100)
 
 	for {
@@ -35,23 +36,35 @@ func monitorRespPackets(redisport uint, sep, cleaner string, maxkeysize int, slo
 			if packet.SrcPort == layers.TCPPort(redisport) { //response
 				ditem, ok := durations.Get(packet.Seq)
 				if !ok {
-					break
+					continue
 				}
 				if l := ditem.ToLatency(); l > slowresponsethresold {
-					slowCommands.WithLabelValues(ditem.Command, ditem.Args).Observe(float64(l))
+					if ditem.Args != "" {
+						slowCommands.WithLabelValues(ditem.Command, ditem.Args).Observe(float64(l))
+					}
 				}
 				if size := len(packet.Payload); size > bigresponsethreshold {
-					bigCommands.WithLabelValues(ditem.Command, ditem.Args).Observe(float64(size))
+					if ditem.Args != "" {
+						bigCommands.WithLabelValues(ditem.Command, ditem.Args).Observe(float64(size))
+					}
 				}
+
+				log.Debug().Str("command", ditem.Command).Str("args", ditem.Args).Int("size", len(packet.Payload)).Msg("response")
 			} else if packet.DstPort == layers.TCPPort(redisport) { //request
 				rsp, err := parseRespPacket(packet.Payload, separator, cleanerxp, maxkeysize)
 				if err != nil {
 					log.Debug().Caller().Hex("payload", packet.Payload).Err(err).Msg("request parse error")
-					break
+					continue
+				}
+				if rsp.Args() != "" {
+					durations.Set(packet.Ack, rsp.Command(), rsp.Args())
+					commandCountDetail.WithLabelValues(rsp.Command(), rsp.Args()).Inc()
+				}
+				if rsp.Command() != "" {
+					commandCount.WithLabelValues(rsp.Command()).Inc()
 				}
-				durations.Set(packet.Ack, rsp.Command(), rsp.Args())
-				commandCount.WithLabelValues(rsp.Command()).Inc()
-				commandCountDetail.WithLabelValues(rsp.Command(), rsp.Args()).Inc()
+
+				log.Debug().Str("command", rsp.Command()).Str("args", rsp.Args()).Float64("size", rsp.Size()).Msg("request")
 			}
 		}
 	}
@@ -62,12 +75,5 @@ func parseRespPacket(payload []byte, sep []byte, cleaner *regexp.Regexp, maxkeys
 	if err != nil {
 		return rsp, err
 	}
-
-	log.Debug().Hex("payload", payload).Msg("payload")
-	log.Debug().Str("command", rsp.Command()).
-		Str("args", rsp.Args()).
-		Float64("size", rsp.Size()).
-		Msg("received")
-
 	return rsp, err
 }

resp_parser.go

@@ -63,7 +63,7 @@ func (c *RespReader) parse(sep []byte, cleaner *regexp.Regexp, maxsize int) erro
 			}
 		}
 
-		if len(argsindex) >= 1 {
+		if len(argsindex) >= 2 {
 			first := pp[argsindex[1]]
 			if len(first) > maxsize {
 				if maxsize > 0 {