Merge pull request #1 from sreerajkksd/master

Support imudp worker thread statistics

README.md

@@ -104,3 +104,11 @@ metrics, a "global" dynstats namespace is also published with some additional bo
 
 See the [dyn_stats](https://www.rsyslog.com/doc/master/configuration/dyn_stats.html)
 documentation for more information.
+
+### IMUDP Workerthread stats
+The [imudp](https://www.rsyslog.com/rsyslog-statistic-counter-plugin-imudp/) module can be configured
+to run on multiple worker threads and the following metrics are returned:
+
+* input_called_recvmmsg - Number of recvmmsg called
+* input_called_recvmsg -Number of recvmmsg called
+* input_received - Messages received

exporter.go

@@ -21,6 +21,7 @@ const (
 	rsyslogResource
 	rsyslogDynStat
 	rsyslogDynafileCache
+	rsyslogInputIMDUP
 )
 
 type rsyslogExporter struct {
@@ -69,6 +70,15 @@ func (re *rsyslogExporter) handleStatLine(rawbuf []byte) error {
 			re.set(p)
 		}
 
+	case rsyslogInputIMDUP:
+		u, err := newInputIMUDPFromJSON(buf)
+		if err != nil {
+			return err
+		}
+		for _, p := range u.toPoints() {
+			re.set(p)
+		}
+
 	case rsyslogQueue:
 		q, err := newQueueFromJSON(buf)
 		if err != nil {

input_imudp.go

+package main
+
+import (
+	"encoding/json"
+	"fmt"
+)
+
+type inputIMUDP struct {
+	Name     string `json:"name"`
+	Recvmmsg int64  `json:"called.recvmmsg"`
+	Recvmsg  int64  `json:"called.recvmsg"`
+	Received int64  `json:"msgs.received"`
+}
+
+func newInputIMUDPFromJSON(b []byte) (*inputIMUDP, error) {
+	var pstat inputIMUDP
+	err := json.Unmarshal(b, &pstat)
+	if err != nil {
+		return nil, fmt.Errorf("error decoding input stat `%v`: %v", string(b), err)
+	}
+	return &pstat, nil
+}
+
+func (i *inputIMUDP) toPoints() []*point {
+	points := make([]*point, 3)
+
+	points[0] = &point{
+		Name:        "input_called_recvmmsg",
+		Type:        counter,
+		Value:       i.Recvmmsg,
+		Description: "Number of recvmmsg called",
+		LabelName:   "worker",
+		LabelValue:  i.Name,
+	}
+	points[1] = &point{
+		Name:        "input_called_recvmsg",
+		Type:        counter,
+		Value:       i.Recvmsg,
+		Description: "Number of recvmmsg called",
+		LabelName:   "worker",
+		LabelValue:  i.Name,
+	}
+
+	points[2] = &point{
+		Name:        "input_received",
+		Type:        counter,
+		Value:       i.Received,
+		Description: "messages received",
+		LabelName:   "worker",
+		LabelValue:  i.Name,
+	}
+
+	return points
+}

input_imudp_test.go

+package main
+
+import "testing"
+
+var (
+	inputIMUDPLog = []byte(`{ "name": "test_input_imudp", "origin": "imudp", "called.recvmmsg":1000, "called.recvmsg":2000, "msgs.received":500}`)
+)
+
+func TestgetInputIMUDP(t *testing.T) {
+	logType := getStatType(inputIMUDPLog)
+	if logType != rsyslogInputIMDUP {
+		t.Errorf("detected pstat type should be %d but is %d", rsyslogInputIMDUP, logType)
+	}
+
+	pstat, err := newInputIMUDPFromJSON([]byte(inputLog))
+	if err != nil {
+		t.Fatalf("expected parsing input stat not to fail, got: %v", err)
+	}
+
+	if want, got := "test_input_imudp", pstat.Name; want != got {
+		t.Errorf("want '%s', got '%s'", want, got)
+	}
+
+	if want, got := int64(1000), pstat.Recvmsg; want != got {
+		t.Errorf("want '%d', got '%d'", want, got)
+	}
+
+	if want, got := int64(2000), pstat.Recvmmsg; want != got {
+		t.Errorf("want '%d', got '%d'", want, got)
+	}
+
+	if want, got := int64(500), pstat.Received; want != got {
+		t.Errorf("want '%d', got '%d'", want, got)
+	}
+}
+
+func TestInputIMUDPtoPoints(t *testing.T) {
+	pstat, err := newInputIMUDPFromJSON([]byte(inputIMUDPLog))
+	if err != nil {
+		t.Fatalf("expected parsing input stat not to fail, got: %v", err)
+	}
+
+	points := pstat.toPoints()
+
+	point := points[0]
+	if want, got := "input_called_recvmmsg", point.Name; want != got {
+		t.Errorf("want '%s', got '%s'", want, got)
+	}
+
+	if want, got := int64(1000), point.Value; want != got {
+		t.Errorf("want '%d', got '%d'", want, got)
+	}
+
+	if want, got := "test_input_imudp", point.LabelValue; want != got {
+		t.Errorf("wanted '%s', got '%s'", want, got)
+	}
+
+	point = points[1]
+	if want, got := "input_called_recvmsg", point.Name; want != got {
+		t.Errorf("want '%s', got '%s'", want, got)
+	}
+
+	if want, got := int64(2000), point.Value; want != got {
+		t.Errorf("want '%d', got '%d'", want, got)
+	}
+
+	if want, got := "test_input_imudp", point.LabelValue; want != got {
+		t.Errorf("wanted '%s', got '%s'", want, got)
+	}
+
+	point = points[2]
+	if want, got := "input_received", point.Name; want != got {
+		t.Errorf("want '%s', got '%s'", want, got)
+	}
+
+	if want, got := int64(500), point.Value; want != got {
+		t.Errorf("want '%d', got '%d'", want, got)
+	}
+
+	if want, got := "test_input_imudp", point.LabelValue; want != got {
+		t.Errorf("wanted '%s', got '%s'", want, got)
+	}
+}

utils.go

@@ -8,6 +8,8 @@ func getStatType(buf []byte) rsyslogType {
 		return rsyslogAction
 	} else if strings.Contains(line, "submitted") {
 		return rsyslogInput
+	} else if strings.Contains(line, "called.recvmmsg") {
+		return rsyslogInputIMDUP
 	} else if strings.Contains(line, "enqueued") {
 		return rsyslogQueue
 	} else if strings.Contains(line, "utime") {