Add tests

alert.go

@@ -243,22 +243,22 @@ func newAlertNode(et *ExecutingTask, n *pipeline.AlertNode, l *log.Logger) (an *
 
 	for _, snmpTrap := range n.SnmpTrapHandlers {
 		// Validate snmp templates
-//		var tmpDataList [][3]string
-//		for _, data := range snmpTrap.DataList {
-	//		var rowData [3]string
-	//		for _, attr := range data {
-  //              data = 4
-//				tpattrtmpl, err := text.New("trap").Parse(attr.(string))
-//				if err != nil {
-//					return nil, err
-//				}
-//				rowData = append(rowData, tpattrtmpl)
-//			}
-//			tmpDataList = append(tmpDataList, data)
-//		}
+		//		var tmpDataList [][3]string
+		//		for _, data := range snmpTrap.DataList {
+		//		var rowData [3]string
+		//		for _, attr := range data {
+		//              data = 4
+		//				tpattrtmpl, err := text.New("trap").Parse(attr.(string))
+		//				if err != nil {
+		//					return nil, err
+		//				}
+		//				rowData = append(rowData, tpattrtmpl)
+		//			}
+		//			tmpDataList = append(tmpDataList, data)
+		//		}
 
 		snmpTrap := snmpTrap
-//		snmpTrap.DataList = tmpDataList
+		//		snmpTrap.DataList = tmpDataList
 		an.handlers = append(an.handlers, func(ad *AlertData) { an.handleSnmpTrap(snmpTrap, ad) })
 	}
 
@@ -1085,22 +1085,22 @@ func (a *AlertNode) handleSnmpTrap(snmpTrap *pipeline.SnmpTrapHandler, ad *Alert
 	}
 
 	// Template
-    /*
-	var buf bytes.Buffer
-	var tmpDataList [][]interface{}
-	for _, data := range snmpTrap.DataList {
-		var rowData []interface{}
-		for _, attr := range data {
-			err := attr.(*text.Template).Execute(&buf, ad.info)
-			if err != nil {
-				a.logger.Printf("E! failed to evaluate SNMP Trap attribute template %s", attr)
-				return
+	/*
+		var buf bytes.Buffer
+		var tmpDataList [][]interface{}
+		for _, data := range snmpTrap.DataList {
+			var rowData []interface{}
+			for _, attr := range data {
+				err := attr.(*text.Template).Execute(&buf, ad.info)
+				if err != nil {
+					a.logger.Printf("E! failed to evaluate SNMP Trap attribute template %s", attr)
+					return
+				}
+				rowData = append(rowData, buf.String())
+				buf.Reset()
 			}
-			rowData = append(rowData, buf.String())
-			buf.Reset()
-		}
-		tmpDataList = append(tmpDataList, rowData)
-	}*/
+			tmpDataList = append(tmpDataList, rowData)
+		}*/
 
 	err := a.et.tm.SnmpTrapService.Alert(
 		snmpTrap.TrapOid,

integrations/streamer_test.go

@@ -33,12 +33,14 @@ import (
 	"github.com/influxdata/kapacitor/services/pagerduty"
 	"github.com/influxdata/kapacitor/services/sensu"
 	"github.com/influxdata/kapacitor/services/slack"
+	"github.com/influxdata/kapacitor/services/snmptrap"
 	"github.com/influxdata/kapacitor/services/talk"
 	"github.com/influxdata/kapacitor/services/telegram"
 	"github.com/influxdata/kapacitor/services/victorops"
 	"github.com/influxdata/kapacitor/udf"
 	"github.com/influxdata/kapacitor/udf/test"
 	"github.com/influxdata/wlog"
+	"github.com/k-sone/snmpgo"
 )
 
 var httpService *httpd.Service
@@ -8245,3 +8247,79 @@ func testStreamerWithOutput(
 		}
 	}
 }
+
+// SNMPTrap
+
+type TrapListener struct {
+	data string
+}
+
+func (l *TrapListener) OnTRAP(trap *snmpgo.TrapRequest) {
+	l.data = trap.Pdu.String()
+}
+
+func NewTrapListener(data *string) *TrapListener {
+	return &TrapListener{*data}
+}
+
+func TestStream_AlertSnmpTrap(t *testing.T) {
+	trapData := ""
+	server, err := snmpgo.NewTrapServer(snmpgo.ServerArguments{
+		LocalAddr: "127.0.0.1:9162",
+	})
+	if err != nil {
+		log.Fatal(err)
+	}
+	// V2c
+	err = server.AddSecurity(&snmpgo.SecurityEntry{
+		Version:   snmpgo.V2c,
+		Community: "public",
+	})
+	if err != nil {
+		log.Fatal(err)
+	}
+	go server.Serve(NewTrapListener(&trapData))
+
+	var script = `
+stream
+	|from()
+		.measurement('cpu')
+		.where(lambda: "host" == 'serverA')
+		.groupBy('host')
+	|window()
+		.period(10s)
+		.every(10s)
+	|count('value')
+	|alert()
+		.id('kapacitor/{{ .Name }}/{{ index .Tags "host" }}')
+		.info(lambda: "count" > 6.0)
+		.warn(lambda: "count" > 7.0)
+		.crit(lambda: "count" > 8.0)
+		.snmptrap('1.1.1')
+        .data('1.1.1.2', 's', 'SNMP ALERT')
+`
+
+	clock, et, replayErr, tm := testStreamer(t, "TestStream_Alert", script, nil)
+	defer tm.Close()
+
+	c := snmptrap.NewConfig()
+	c.Enabled = true
+	c.TargetIp = "127.0.0.1"
+	c.TargetPort = 9162
+	c.Community = "public"
+	c.Version = "2c"
+	st := snmptrap.NewService(c, logService.NewLogger("[test_snmptrap] ", log.LstdFlags))
+	tm.SnmpTrapService = st
+
+	err = fastForwardTask(clock, et, replayErr, tm, 13*time.Second)
+	if err != nil {
+		t.Error(err)
+	}
+
+	// Stop server
+	server.Close()
+
+	if trapData == "" {
+		t.Errorf("No SNMP Trap data received")
+	}
+}

pipeline/alert.go

@@ -1335,7 +1335,7 @@ type SnmpTrapHandler struct {
 // tick:property
 func (h *SnmpTrapHandler) Data(oid, rawType, value string) *SnmpTrapHandler {
 	// TODO check element validity
-    data := [3]string{oid, rawType, value}
+	data := [3]string{oid, rawType, value}
 	h.DataList = append(h.DataList, data)
 	return h
 }