Merge pull request #2416 from bonitoo-io/feat/servicenow

feat: ServiceNow event handler

alert.go

@@ -28,6 +28,7 @@ import (
 	"github.com/influxdata/kapacitor/services/pagerduty2"
 	"github.com/influxdata/kapacitor/services/pushover"
 	"github.com/influxdata/kapacitor/services/sensu"
+	"github.com/influxdata/kapacitor/services/servicenow"
 	"github.com/influxdata/kapacitor/services/slack"
 	"github.com/influxdata/kapacitor/services/smtp"
 	"github.com/influxdata/kapacitor/services/snmptrap"
@@ -518,6 +519,30 @@ func newAlertNode(et *ExecutingTask, n *pipeline.AlertNode, d NodeDiagnostic) (a
 		n.IsStateChangesOnly = true
 	}
 
+	for _, s := range n.ServiceNowHandlers {
+		c := servicenow.HandlerConfig{
+			URL:        s.URL,
+			Source:     s.Source,
+			Node:       s.Node,
+			Type:       s.Type,
+			Resource:   s.Resource,
+			MetricName: s.MetricName,
+			MessageKey: s.MessageKey,
+		}
+		h := et.tm.ServiceNowService.Handler(c, ctx...)
+		an.handlers = append(an.handlers, h)
+	}
+	if len(n.ServiceNowHandlers) == 0 && (et.tm.ServiceNowService != nil && et.tm.ServiceNowService.Global()) {
+		h := et.tm.ServiceNowService.Handler(servicenow.HandlerConfig{}, ctx...)
+		an.handlers = append(an.handlers, h)
+	}
+	// If servicenow has been configured with state changes only set it.
+	if et.tm.ServiceNowService != nil &&
+		et.tm.ServiceNowService.Global() &&
+		et.tm.ServiceNowService.StateChangesOnly() {
+		n.IsStateChangesOnly = true
+	}
+
 	// Parse level expressions
 	an.levels = make([]stateful.Expression, alert.Critical+1)
 	an.scopePools = make([]stateful.ScopePool, alert.Critical+1)

etc/kapacitor/kapacitor.conf

@@ -509,6 +509,18 @@ default-retention-policy = ""
   # Default origin.
   origin = "kapacitor"
 
+[servicenow]
+  # Configure ServiceNow.
+  enabled = false
+  # The ServiceNow URL for target table. Replace instance with actual hostname.
+  url = "https://instance.service-now.com/api/now/v1/table/em_alert"
+  # Default source identification.
+  source = "Kapacitor"
+  # Username for HTTP BASIC authentication
+  # username = ""
+  # Password for HTTP BASIC authentication
+  # password = ""
+
 [sensu]
   # Configure Sensu.
   enabled = false

integrations/streamer_test.go

@@ -62,6 +62,8 @@ import (
 	"github.com/influxdata/kapacitor/services/pushover/pushovertest"
 	"github.com/influxdata/kapacitor/services/sensu"
 	"github.com/influxdata/kapacitor/services/sensu/sensutest"
+	"github.com/influxdata/kapacitor/services/servicenow"
+	"github.com/influxdata/kapacitor/services/servicenow/servicenowtest"
 	"github.com/influxdata/kapacitor/services/sideload"
 	"github.com/influxdata/kapacitor/services/slack"
 	"github.com/influxdata/kapacitor/services/slack/slacktest"
@@ -10263,6 +10265,81 @@ stream
 	}
 }
 
+func TestStream_AlertServiceNow(t *testing.T) {
+	ts := servicenowtest.NewServer()
+	defer ts.Close()
+
+	var script = `
+stream
+	|from()
+		.measurement('cpu')
+		.where(lambda: "host" == 'serverA')
+		.groupBy('host', 'type')
+	|window()
+		.period(10s)
+		.every(10s)
+	|mean('value')
+	|alert()
+		.id('kapacitor/{{ .Name }}/{{ index .Tags "host" }}')
+		.info(lambda: "mean" > 15.0)
+		.warn(lambda: "mean" > 50.0)
+		.crit(lambda: "mean" > 90.0)
+		.serviceNow()
+		.serviceNow()
+			.node('{{ index .Tags "host" }}')
+			.type('CPU')
+			.resource('CPU-Total')
+			.metricName('{{ index .Tags "type" }}')
+			.messageKey('Alert: {{ .ID }}')
+`
+	tmInit := func(tm *kapacitor.TaskMaster) {
+
+		c := servicenow.NewConfig()
+		c.Enabled = true
+		c.URL = ts.URL
+		c.Source = "Kapacitor"
+		sl := servicenow.NewService(c, diagService.NewServiceNowHandler())
+		tm.ServiceNowService = sl
+	}
+
+	testStreamerNoOutput(t, "TestStream_Alert", script, 13*time.Second, tmInit)
+
+	exp := []interface{}{
+		servicenowtest.Request{
+			URL: "/",
+			Alert: servicenow.Alert{
+				Source:      "Kapacitor",
+				Node:        "serverA",
+				Type:        "CPU",       // literal since there is no tag for this in the testdata
+				Resource:    "CPU-Total", // literal since there is no tag for this in the testdata
+				MetricName:  "idle",
+				MessageKey:  "Alert: kapacitor/cpu/serverA",
+				Severity:    "1",
+				Description: "kapacitor/cpu/serverA is CRITICAL",
+			},
+		},
+		servicenowtest.Request{
+			URL: "/",
+			Alert: servicenow.Alert{
+				Source:      "Kapacitor",
+				MessageKey:  "kapacitor/cpu/serverA",
+				Severity:    "1",
+				Description: "kapacitor/cpu/serverA is CRITICAL",
+			},
+		},
+	}
+
+	ts.Close()
+	var got []interface{}
+	for _, g := range ts.Requests() {
+		got = append(got, g)
+	}
+
+	if err := compareListIgnoreOrder(got, exp, nil); err != nil {
+		t.Error(err)
+	}
+}
+
 func TestStream_AlertLog(t *testing.T) {
 	tmpDir, err := ioutil.TempDir("", "TestStream_AlertLog")
 	if err != nil {

pipeline/alert.go

@@ -56,7 +56,8 @@ type AlertNode struct{ *AlertNodeData }
 //    * Telegram -- Post alert message to Telegram client.
 //    * MQTT -- Post alert message to MQTT.
 //    * Teams -- Post alert message to Microsoft Teams.
-//	  * Discord -- Post alert message to Discord webhook.
+//    * Discord -- Post alert message to Discord webhook.
+//    * ServiceNow -- Post alert message to ServiceNow.
 //
 // See below for more details on configuring each handler.
 //
@@ -395,6 +396,10 @@ type AlertNodeData struct {
 	// Send alert to Microsoft Teams channel.
 	// tick:ignore
 	TeamsHandlers []*TeamsHandler `tick:"Teams" json:"teams"`
+
+	// Send alert to ServiceNow.
+	// tick:ignore
+	ServiceNowHandlers []*ServiceNowHandler `tick:"ServiceNow" json:"serviceNow"`
 }
 
 func newAlertNode(wants EdgeType) *AlertNode {
@@ -2119,3 +2124,80 @@ type TeamsHandler struct {
 	// If empty uses the URL from the configuration.
 	ChannelURL string `json:"channel_url"`
 }
+
+// Send the alert to ServiceNow.
+//
+// Example:
+//    [serviceNow]
+//      enabled = true
+//      url = "https://instance.service-now.com/api/now/v1/table/em_alert"
+//
+// In order to not post a message every alert interval
+// use AlertNode.StateChangesOnly so that only events
+// where the alert changed state are posted to the room.
+//
+// Example:
+//    stream
+//         |alert()
+//             .serviceNow()
+//
+// If the 'serviceNow' section in the configuration has the option: global = true
+// then all alerts are sent to ServiceNow without the need to explicitly state it
+// in the TICKscript.
+//
+// Example:
+//    [serviceNow]
+//      enabled = true
+//      url = "https://instance.service-now.com/api/now/v1/table/em_alert"
+//      global = true
+//      state-changes-only = true
+//
+// Example:
+//    stream
+//         |alert()
+//
+// Send alert to ServiceNow using default url.
+// tick:property
+func (n *AlertNodeData) ServiceNow() *ServiceNowHandler {
+	serviceNow := &ServiceNowHandler{
+		AlertNodeData: n,
+	}
+	n.ServiceNowHandlers = append(n.ServiceNowHandlers, serviceNow)
+	return serviceNow
+}
+
+// tick:embedded:AlertNode.ServiceNow
+type ServiceNowHandler struct {
+	*AlertNodeData `json:"-"`
+
+	// ServiceNow API URL to post alerts.
+	// If empty uses the URL from the configuration.
+	URL string `json:"url"`
+
+	// Username for BASIC authentication.
+	// If empty uses username from the configuration.
+	Username string `json:"username"`
+
+	// Password for BASIC authentication.
+	// If empty uses password from the configuration.
+	Password string `json:"password"`
+
+	// Event source identification (event monitoring software).
+	// If empty uses the URL from the configuration.
+	Source string `json:"source"`
+
+	// Node name.
+	Node string `json:"node"`
+
+	// Metric type.
+	Type string `json:"type"`
+
+	// Node resource relevant to the event.
+	Resource string `json:"resource"`
+
+	// Metric name for which event has been created..
+	MetricName string `json:"metric_name"`
+
+	// Message key.
+	MessageKey string `json:"messageKey"`
+}

pipeline/alert_test.go

@@ -82,7 +82,8 @@ func TestAlertNode_MarshalJSON(t *testing.T) {
     "mqtt": null,
     "snmpTrap": null,
     "kafka": null,
-    "teams": null
+    "teams": null,
+    "serviceNow": null
 }`,
 		},
 	}

pipeline/json_test.go

@@ -261,7 +261,8 @@ func TestPipeline_MarshalJSON(t *testing.T) {
             "mqtt": null,
             "snmpTrap": null,
             "kafka": null,
-            "teams": null
+            "teams": null,
+            "serviceNow": null
         },
         {
             "typeOf": "httpOut",

pipeline/tick/alert.go

@@ -157,6 +157,16 @@ func (n *AlertNode) Build(a *pipeline.AlertNode) (ast.Node, error) {
 		}
 	}
 
+	for _, h := range a.ServiceNowHandlers {
+		n.Dot("servicenow").
+			Dot("source", h.Source).
+			Dot("node", h.Node).
+			Dot("type", h.Type).
+			Dot("resource", h.Resource).
+			Dot("metricName", h.MetricName).
+			Dot("messageKey", h.MessageKey)
+	}
+
 	for _, h := range a.SlackHandlers {
 		n.Dot("slack").
 			Dot("workspace", h.Workspace).

server/config.go

@@ -44,6 +44,7 @@ import (
 	"github.com/influxdata/kapacitor/services/scraper"
 	"github.com/influxdata/kapacitor/services/sensu"
 	"github.com/influxdata/kapacitor/services/serverset"
+	"github.com/influxdata/kapacitor/services/servicenow"
 	"github.com/influxdata/kapacitor/services/slack"
 	"github.com/influxdata/kapacitor/services/smtp"
 	"github.com/influxdata/kapacitor/services/snmptrap"
@@ -101,6 +102,7 @@ type Config struct {
 	SMTP       smtp.Config       `toml:"smtp" override:"smtp"`
 	SNMPTrap   snmptrap.Config   `toml:"snmptrap" override:"snmptrap"`
 	Sensu      sensu.Config      `toml:"sensu" override:"sensu"`
+	ServiceNow servicenow.Config `toml:"servicenow" override:"servicenow"`
 	Slack      slack.Configs     `toml:"slack" override:"slack,element-key=workspace"`
 	Talk       talk.Config       `toml:"talk" override:"talk"`
 	Teams      teams.Config      `toml:"teams" override:"teams"`
@@ -171,6 +173,7 @@ func NewConfig() *Config {
 	c.HTTPPost = httppost.Configs{httppost.NewConfig()}
 	c.SMTP = smtp.NewConfig()
 	c.Sensu = sensu.NewConfig()
+	c.ServiceNow = servicenow.NewConfig()
 	c.Slack = slack.Configs{slack.NewDefaultConfig()}
 	c.Talk = talk.NewConfig()
 	c.Teams = teams.NewConfig()
@@ -317,6 +320,9 @@ func (c *Config) Validate() error {
 	if err := c.Sensu.Validate(); err != nil {
 		return errors.Wrap(err, "sensu")
 	}
+	if err := c.ServiceNow.Validate(); err != nil {
+		return errors.Wrap(err, "servicenow")
+	}
 	if err := c.Slack.Validate(); err != nil {
 		return errors.Wrap(err, "slack")
 	}

server/server.go

@@ -55,6 +55,7 @@ import (
 	"github.com/influxdata/kapacitor/services/scraper"
 	"github.com/influxdata/kapacitor/services/sensu"
 	"github.com/influxdata/kapacitor/services/serverset"
+	"github.com/influxdata/kapacitor/services/servicenow"
 	"github.com/influxdata/kapacitor/services/servicetest"
 	"github.com/influxdata/kapacitor/services/sideload"
 	"github.com/influxdata/kapacitor/services/slack"
@@ -258,6 +259,7 @@ func New(c *Config, buildInfo BuildInfo, diagService *diagnostic.Service) (*Serv
 	if err := s.appendHTTPPostService(); err != nil {
 		return nil, errors.Wrap(err, "httppost service")
 	}
+	s.appendServiceNowService()
 	s.appendSMTPService()
 	s.appendTeamsService()
 	s.appendTelegramService()
@@ -1001,6 +1003,18 @@ func (s *Server) appendTeamsService() {
 	s.AppendService("teams", srv)
 }
 
+func (s *Server) appendServiceNowService() {
+	c := s.config.ServiceNow
+	d := s.DiagService.NewServiceNowHandler()
+	srv := servicenow.NewService(c, d)
+
+	s.TaskMaster.ServiceNowService = srv
+	s.AlertService.ServiceNowService = srv
+
+	s.SetDynamicService("servicenow", srv)
+	s.AppendService("servicenow", srv)
+}
+
 // Err returns an error channel that multiplexes all out of band errors received from all services.
 func (s *Server) Err() <-chan error { return s.err }