func newAlertNode(et *ExecutingTask, n *pipeline.AlertNode, l *log.Logger) (an *AlertNode, err error) {
	an = &AlertNode{
		node: node{Node: n, et: et, logger: l},
		a:    n,
	}
	an.node.runF = an.runAlert
	an.topic = n.Topic
	an.anonTopic = fmt.Sprintf("%s:%s:%s", et.tm.ID(), et.Task.ID, an.Name())
	an.bufPool = sync.Pool{
		New: func() interface{} {
			return new(bytes.Buffer)
		},
	}
	an.idTmpl, err = text.New("id").Parse(n.Id)
	if err != nil {
		return nil, err
	}
	an.messageTmpl, err = text.New("message").Parse(n.Message)
	if err != nil {
		return nil, err
	}
	an.detailsTmpl, err = html.New("details").Funcs(html.FuncMap{
		"json": func(v interface{}) html.JS {
			tmpBuffer := an.bufPool.Get().(*bytes.Buffer)
			defer func() {
				tmpBuffer.Reset()
				an.bufPool.Put(tmpBuffer)
			}()
			_ = json.NewEncoder(tmpBuffer).Encode(v)
			return html.JS(tmpBuffer.String())
		},
	}).Parse(n.Details)
	if err != nil {
		return nil, err
	}
	for _, tcp := range n.TcpHandlers {
		c := alertservice.TCPHandlerConfig{
			Address: tcp.Address,
		}
		h := alertservice.NewTCPHandler(c, l)
		an.handlers = append(an.handlers, h)
	}
	for _, email := range n.EmailHandlers {
		c := smtp.HandlerConfig{
			To: email.ToList,
		}
		h := et.tm.SMTPService.Handler(c, l)
		an.handlers = append(an.handlers, h)
	}
	if len(n.EmailHandlers) == 0 && (et.tm.SMTPService != nil && et.tm.SMTPService.Global()) {
		c := smtp.HandlerConfig{}
		h := et.tm.SMTPService.Handler(c, l)
		an.handlers = append(an.handlers, h)
	}
	if et.tm.SMTPService != nil &&
		et.tm.SMTPService.Global() &&
		et.tm.SMTPService.StateChangesOnly() {
		n.IsStateChangesOnly = true
	}
	for _, e := range n.ExecHandlers {
		c := alertservice.ExecHandlerConfig{
			Prog:      e.Command[0],
			Args:      e.Command[1:],
			Commander: et.tm.Commander,
		}
		h := alertservice.NewExecHandler(c, l)
		an.handlers = append(an.handlers, h)
	}
	for _, log := range n.LogHandlers {
		c := alertservice.DefaultLogHandlerConfig()
		c.Path = log.FilePath
		if log.Mode != 0 {
			c.Mode = os.FileMode(log.Mode)
		}
		h, err := alertservice.NewLogHandler(c, l)
		if err != nil {
			return nil, errors.Wrap(err, "failed to create log alert handler")
		}
		an.handlers = append(an.handlers, h)
	}
	for _, vo := range n.VictorOpsHandlers {
		c := victorops.HandlerConfig{
			RoutingKey: vo.RoutingKey,
		}
		h := et.tm.VictorOpsService.Handler(c, l)
		an.handlers = append(an.handlers, h)
	}
	if len(n.VictorOpsHandlers) == 0 && (et.tm.VictorOpsService != nil && et.tm.VictorOpsService.Global()) {
		c := victorops.HandlerConfig{}
		h := et.tm.VictorOpsService.Handler(c, l)
		an.handlers = append(an.handlers, h)
	}
	for _, pd := range n.PagerDutyHandlers {
		c := pagerduty.HandlerConfig{
			ServiceKey: pd.ServiceKey,
		}
		h := et.tm.PagerDutyService.Handler(c, l)
		an.handlers = append(an.handlers, h)
	}
	if len(n.PagerDutyHandlers) == 0 && (et.tm.PagerDutyService != nil && et.tm.PagerDutyService.Global()) {
		c := pagerduty.HandlerConfig{}
		h := et.tm.PagerDutyService.Handler(c, l)
		an.handlers = append(an.handlers, h)
	}
	for _, s := range n.SensuHandlers {
		c := sensu.HandlerConfig{
			Source:   s.Source,
			Handlers: s.HandlersList,
		}
		h, err := et.tm.SensuService.Handler(c, l)
		if err != nil {
			return nil, errors.Wrap(err, "failed to create sensu alert handler")
		}
		an.handlers = append(an.handlers, h)
	}
	for _, s := range n.SlackHandlers {
		c := slack.HandlerConfig{
			Channel:   s.Channel,
			Username:  s.Username,
			IconEmoji: s.IconEmoji,
		}
		h := et.tm.SlackService.Handler(c, l)
		an.handlers = append(an.handlers, h)
	}
	if len(n.SlackHandlers) == 0 && (et.tm.SlackService != nil && et.tm.SlackService.Global()) {
		h := et.tm.SlackService.Handler(slack.HandlerConfig{}, l)
		an.handlers = append(an.handlers, h)
	}
	if et.tm.SlackService != nil &&
		et.tm.SlackService.Global() &&
		et.tm.SlackService.StateChangesOnly() {
		n.IsStateChangesOnly = true
	}
	for _, t := range n.TelegramHandlers {
		c := telegram.HandlerConfig{
			ChatId:                t.ChatId,
			ParseMode:             t.ParseMode,
			DisableWebPagePreview: t.IsDisableWebPagePreview,
			DisableNotification:   t.IsDisableNotification,
		}
		h := et.tm.TelegramService.Handler(c, l)
		an.handlers = append(an.handlers, h)
	}
	for _, s := range n.SNMPTrapHandlers {
		dataList := make([]snmptrap.Data, len(s.DataList))
		for i, d := range s.DataList {
			dataList[i] = snmptrap.Data{
				Oid:   d.Oid,
				Type:  d.Type,
				Value: d.Value,
			}
		}
		c := snmptrap.HandlerConfig{
			TrapOid:  s.TrapOid,
			DataList: dataList,
		}
		h, err := et.tm.SNMPTrapService.Handler(c, l)
		if err != nil {
			return nil, errors.Wrapf(err, "failed to create SNMP handler")
		}
		an.handlers = append(an.handlers, h)
	}
	if len(n.TelegramHandlers) == 0 && (et.tm.TelegramService != nil && et.tm.TelegramService.Global()) {
		c := telegram.HandlerConfig{}
		h := et.tm.TelegramService.Handler(c, l)
		an.handlers = append(an.handlers, h)
	}
	if et.tm.TelegramService != nil &&
		et.tm.TelegramService.Global() &&
		et.tm.TelegramService.StateChangesOnly() {
		n.IsStateChangesOnly = true
	}
	for _, hc := range n.HipChatHandlers {
		c := hipchat.HandlerConfig{
			Room:  hc.Room,
			Token: hc.Token,
		}
		h := et.tm.HipChatService.Handler(c, l)
		an.handlers = append(an.handlers, h)
	}
	if len(n.HipChatHandlers) == 0 && (et.tm.HipChatService != nil && et.tm.HipChatService.Global()) {
		c := hipchat.HandlerConfig{}
		h := et.tm.HipChatService.Handler(c, l)
		an.handlers = append(an.handlers, h)
	}
	if et.tm.HipChatService != nil &&
		et.tm.HipChatService.Global() &&
		et.tm.HipChatService.StateChangesOnly() {
		n.IsStateChangesOnly = true
	}
	for _, a := range n.AlertaHandlers {
		c := et.tm.AlertaService.DefaultHandlerConfig()
		if a.Token != "" {
			c.Token = a.Token
		}
		if a.Resource != "" {
			c.Resource = a.Resource
		}
		if a.Event != "" {
			c.Event = a.Event
		}
		if a.Environment != "" {
			c.Environment = a.Environment
		}
		if a.Group != "" {
			c.Group = a.Group
		}
		if a.Value != "" {
			c.Value = a.Value
		}
		if a.Origin != "" {
			c.Origin = a.Origin
		}
		if len(a.Service) != 0 {
			c.Service = a.Service
		}
		h, err := et.tm.AlertaService.Handler(c, l)
		if err != nil {
			return nil, errors.Wrap(err, "failed to create Alerta handler")
		}
		an.handlers = append(an.handlers, h)
	}
	for _, p := range n.PushoverHandlers {
		c := pushover.HandlerConfig{}
		if p.Device != "" {
			c.Device = p.Device
		}
		if p.Title != "" {
			c.Title = p.Title
		}
		if p.URL != "" {
			c.URL = p.URL
		}
		if p.URLTitle != "" {
			c.URLTitle = p.URLTitle
		}
		if p.Sound != "" {
			c.Sound = p.Sound
		}
		h := et.tm.PushoverService.Handler(c, l)
		an.handlers = append(an.handlers, h)
	}
	for _, p := range n.HTTPPostHandlers {
		c := httppost.HandlerConfig{
			URL:      p.URL,
			Endpoint: p.Endpoint,
			Headers:  p.Headers,
		}
		h := et.tm.HTTPPostService.Handler(c, l)
		an.handlers = append(an.handlers, h)
	}
	for _, og := range n.OpsGenieHandlers {
		c := opsgenie.HandlerConfig{
			TeamsList:      og.TeamsList,
			RecipientsList: og.RecipientsList,
		}
		h := et.tm.OpsGenieService.Handler(c, l)
		an.handlers = append(an.handlers, h)
	}
	if len(n.OpsGenieHandlers) == 0 && (et.tm.OpsGenieService != nil && et.tm.OpsGenieService.Global()) {
		c := opsgenie.HandlerConfig{}
		h := et.tm.OpsGenieService.Handler(c, l)
		an.handlers = append(an.handlers, h)
	}
	for range n.TalkHandlers {
		h := et.tm.TalkService.Handler(l)
		an.handlers = append(an.handlers, h)
	}
	for _, m := range n.MQTTHandlers {
		c := mqtt.HandlerConfig{
			BrokerName: m.BrokerName,
			Topic:      m.Topic,
			QoS:        mqtt.QoSLevel(m.Qos),
			Retained:   m.Retained,
		}
		h := et.tm.MQTTService.Handler(c, l)
		an.handlers = append(an.handlers, h)
	}
	an.levels = make([]stateful.Expression, alert.Critical+1)
	an.scopePools = make([]stateful.ScopePool, alert.Critical+1)
	an.levelResets = make([]stateful.Expression, alert.Critical+1)
	an.lrScopePools = make([]stateful.ScopePool, alert.Critical+1)
	if n.Info != nil {
		statefulExpression, expressionCompileError := stateful.NewExpression(n.Info.Expression)
		if expressionCompileError != nil {
			return nil, fmt.Errorf("Failed to compile stateful expression for info: %s", expressionCompileError)
		}
		an.levels[alert.Info] = statefulExpression
		an.scopePools[alert.Info] = stateful.NewScopePool(ast.FindReferenceVariables(n.Info.Expression))
		if n.InfoReset != nil {
			lstatefulExpression, lexpressionCompileError := stateful.NewExpression(n.InfoReset.Expression)
			if lexpressionCompileError != nil {
				return nil, fmt.Errorf("Failed to compile stateful expression for infoReset: %s", lexpressionCompileError)
			}
			an.levelResets[alert.Info] = lstatefulExpression
			an.lrScopePools[alert.Info] = stateful.NewScopePool(ast.FindReferenceVariables(n.InfoReset.Expression))
		}
	}
	if n.Warn != nil {
		statefulExpression, expressionCompileError := stateful.NewExpression(n.Warn.Expression)
		if expressionCompileError != nil {
			return nil, fmt.Errorf("Failed to compile stateful expression for warn: %s", expressionCompileError)
		}
		an.levels[alert.Warning] = statefulExpression
		an.scopePools[alert.Warning] = stateful.NewScopePool(ast.FindReferenceVariables(n.Warn.Expression))
		if n.WarnReset != nil {
			lstatefulExpression, lexpressionCompileError := stateful.NewExpression(n.WarnReset.Expression)
			if lexpressionCompileError != nil {
				return nil, fmt.Errorf("Failed to compile stateful expression for warnReset: %s", lexpressionCompileError)
			}
			an.levelResets[alert.Warning] = lstatefulExpression
			an.lrScopePools[alert.Warning] = stateful.NewScopePool(ast.FindReferenceVariables(n.WarnReset.Expression))
		}
	}
	if n.Crit != nil {
		statefulExpression, expressionCompileError := stateful.NewExpression(n.Crit.Expression)
		if expressionCompileError != nil {
			return nil, fmt.Errorf("Failed to compile stateful expression for crit: %s", expressionCompileError)
		}
		an.levels[alert.Critical] = statefulExpression
		an.scopePools[alert.Critical] = stateful.NewScopePool(ast.FindReferenceVariables(n.Crit.Expression))
		if n.CritReset != nil {
			lstatefulExpression, lexpressionCompileError := stateful.NewExpression(n.CritReset.Expression)
			if lexpressionCompileError != nil {
				return nil, fmt.Errorf("Failed to compile stateful expression for critReset: %s", lexpressionCompileError)
			}
			an.levelResets[alert.Critical] = lstatefulExpression
			an.lrScopePools[alert.Critical] = stateful.NewScopePool(ast.FindReferenceVariables(n.CritReset.Expression))
		}
	}
	if n.History < 2 {
		n.History = 2
	}
	if n.UseFlapping {
		if n.FlapLow > 1 || n.FlapHigh > 1 {
			return nil, errors.New("alert flap thresholds are percentages and should be between 0 and 1")
		}
	}
	return
}
Create a new  AlertNode which caches the most recent item and exposes it over the HTTP API.
package kapacitor
import (
	"bytes"
	"encoding/json"
	"fmt"
	html "html/template"
	"log"
	"os"
	"sync"
	text "text/template"
	"time"
	"github.com/influxdata/kapacitor/alert"
	"github.com/influxdata/kapacitor/edge"
	"github.com/influxdata/kapacitor/expvar"
	"github.com/influxdata/kapacitor/models"
	"github.com/influxdata/kapacitor/pipeline"
	alertservice "github.com/influxdata/kapacitor/services/alert"
	"github.com/influxdata/kapacitor/services/hipchat"
	"github.com/influxdata/kapacitor/services/httppost"
	"github.com/influxdata/kapacitor/services/mqtt"
	"github.com/influxdata/kapacitor/services/opsgenie"
	"github.com/influxdata/kapacitor/services/pagerduty"
	"github.com/influxdata/kapacitor/services/pushover"
	"github.com/influxdata/kapacitor/services/sensu"
	"github.com/influxdata/kapacitor/services/slack"
	"github.com/influxdata/kapacitor/services/smtp"
	"github.com/influxdata/kapacitor/services/snmptrap"
	"github.com/influxdata/kapacitor/services/telegram"
	"github.com/influxdata/kapacitor/services/victorops"
	"github.com/influxdata/kapacitor/tick/ast"
	"github.com/influxdata/kapacitor/tick/stateful"
	"github.com/pkg/errors"
)
const (
	statsAlertsTriggered = "alerts_triggered"
	statsOKsTriggered    = "oks_triggered"
	statsInfosTriggered  = "infos_triggered"
	statsWarnsTriggered  = "warns_triggered"
	statsCritsTriggered  = "crits_triggered"
	statsEventsDropped   = "events_dropped"
)
const weightDiff = 1.5
const maxWeight = 1.2
type AlertNode struct {
	node
	a           *pipeline.AlertNode
	topic       string
	anonTopic   string
	handlers    []alert.Handler
	levels      []stateful.Expression
	scopePools  []stateful.ScopePool
	idTmpl      *text.Template
	messageTmpl *text.Template
	detailsTmpl *html.Template
	alertsTriggered *expvar.Int
	oksTriggered    *expvar.Int
	infosTriggered  *expvar.Int
	warnsTriggered  *expvar.Int
	critsTriggered  *expvar.Int
	eventsDropped   *expvar.Int
	bufPool sync.Pool
	levelResets  []stateful.Expression
	lrScopePools []stateful.ScopePool
}