Repository: influxdata/kapacitor
Pull request #2650 (2021-11-01 11:47:34)

feat(alerts): templating for address in email alert
This adds templating email addresses for email alerts.


Patch:
diff --git a/alert.go b/alert.go
index 868b8f88..b481ac37 100644
--- a/alert.go
+++ b/alert.go
@@ -167,7 +167,8 @@ func newAlertNode(et *ExecutingTask, n *pipeline.AlertNode, d NodeDiagnostic) (a
 
 	for _, email := range n.EmailHandlers {
 		c := smtp.HandlerConfig{
-			To: email.ToList,
+			To:          email.ToList,
+			ToTemplates: email.ToTemplatesList,
 		}
 		h := et.tm.SMTPService.Handler(c, ctx...)
 		an.handlers = append(an.handlers, h)
diff --git a/go.mod b/go.mod
index 28cea472..2dffd905 100644
--- a/go.mod
+++ b/go.mod
@@ -50,7 +50,7 @@ require (
 	golang.org/x/crypto v0.0.0-20201221181555-eec23a3978ad
 	golang.org/x/tools v0.1.0
 	google.golang.org/protobuf v1.27.1
-	gopkg.in/gomail.v2 v2.0.0-20150902115704-41f357289737
+	gopkg.in/gomail.v2 v2.0.0-20160411212932-81ebce5c23df
 )
 
 require (
diff --git a/go.sum b/go.sum
index c4b38c98..2c48ec0d 100644
--- a/go.sum
+++ b/go.sum
@@ -1705,8 +1705,8 @@ gopkg.in/errgo.v2 v2.1.0/go.mod h1:hNsd1EY+bozCKY1Ytp96fpM3vjJbqLJn88ws8XvfDNI=
 gopkg.in/fsnotify/fsnotify.v1 v1.4.7 h1:XNNYLJHt73EyYiCZi6+xjupS9CpvmiDgjPTAjrBlQbo=
 gopkg.in/fsnotify/fsnotify.v1 v1.4.7/go.mod h1:Fyux9zXlo4rWoMSIzpn9fDAYjalPqJ/K1qJ27s+7ltE=
 gopkg.in/gcfg.v1 v1.2.3/go.mod h1:yesOnuUOFQAhST5vPY4nbZsb/huCgGGXlipJsBn0b3o=
-gopkg.in/gomail.v2 v2.0.0-20150902115704-41f357289737 h1:NvePS/smRcFQ4bMtTddFtknbGCtoBkJxGmpSpVRafCc=
-gopkg.in/gomail.v2 v2.0.0-20150902115704-41f357289737/go.mod h1:LRQQ+SO6ZHR7tOkpBDuZnXENFzX8qRjMDMyPD6BRkCw=
+gopkg.in/gomail.v2 v2.0.0-20160411212932-81ebce5c23df h1:n7WqCuqOuCbNr617RXOY0AWRXxgwEyPp2z+p0+hgMuE=
+gopkg.in/gomail.v2 v2.0.0-20160411212932-81ebce5c23df/go.mod h1:LRQQ+SO6ZHR7tOkpBDuZnXENFzX8qRjMDMyPD6BRkCw=
 gopkg.in/inf.v0 v0.9.1 h1:73M5CoZyi3ZLMOyDlQh031Cx6N9NDJ2Vvfl76EDAgDc=
 gopkg.in/inf.v0 v0.9.1/go.mod h1:cWUDdTG/fYaXco+Dcufb5Vnc6Gp2YChqWtbxRZE0mXw=
 gopkg.in/ini.v1 v1.42.0/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k=
diff --git a/pipeline/alert.go b/pipeline/alert.go
index 9f5daf04..01849e69 100644
--- a/pipeline/alert.go
+++ b/pipeline/alert.go
@@ -796,6 +796,10 @@ type EmailHandler struct {
 	// List of email recipients.
 	// tick:ignore
 	ToList []string `tick:"To" json:"to"`
+
+	// ToTemplatesList is the Field or Value from which to grab email addresses
+	// tick:ignore
+	ToTemplatesList []string `tick:"ToTemplates" json:"to-templates"`
 }
 
 // Define the To addresses for the email alert.
@@ -826,6 +830,33 @@ func (h *EmailHandler) To(to ...string) *EmailHandler {
 	return h
 }
 
+// Define the To addresses for the email alert.
+// Multiple calls append to the existing list of addresses.
+// If empty uses the addresses from the configuration.
+//
+// Example:
+//    |alert()
+//       .id('{{ .Name }}')
+//       // Email subject
+//       .message('{{ .ID }}:{{ .Level }}')
+//       //Email body as HTML
+//       .details('''
+//<h1>{{ .ID }}</h1>
+//<b>{{ .Message }}</b>
+//Value: {{ index .Fields "value" }}
+//''')
+//       .email('admin@example.com')
+//         .toTemplates('oncall@example.com')
+//
+// All three email addresses will receive the alert message.
+//
+// Passing addresses to the `email` property directly or using the `email.to` property is the same.
+// tick:property
+func (h *EmailHandler) ToTemplates(to ...string) *EmailHandler {
+	h.ToTemplatesList = append(h.ToTemplatesList, to...)
+	return h
+}
+
 // Execute a command whenever an alert is triggered and pass the alert data over STDIN in JSON format.
 // tick:property
 func (n *AlertNodeData) Exec(executable string, args ...string) *ExecHandler {
diff --git a/pipeline/tick/alert.go b/pipeline/tick/alert.go
index ffc2d11f..9f0f827a 100644
--- a/pipeline/tick/alert.go
+++ b/pipeline/tick/alert.go
@@ -91,6 +91,10 @@ func (n *AlertNode) Build(a *pipeline.AlertNode) (ast.Node, error) {
 		for _, to := range h.ToList {
 			n.Dot("to", to)
 		}
+
+		if len(h.ToTemplatesList) != 0 {
+			n.Dot("toTemplates", h.ToTemplatesList)
+		}
 	}
 
 	for _, h := range a.ExecHandlers {
diff --git a/services/smtp/config.go b/services/smtp/config.go
index 7541098c..90cea90e 100644
--- a/services/smtp/config.go
+++ b/services/smtp/config.go
@@ -3,7 +3,7 @@ package smtp
 import (
 	"errors"
 	"fmt"
-	"strings"
+	"net/mail"
 	"time"
 
 	"github.com/influxdata/influxdb/toml"
@@ -26,6 +26,8 @@ type Config struct {
 	From string `toml:"from" override:"from"`
 	// Default To addresses
 	To []string `toml:"to" override:"to"`
+	//ToTemplates is the field or value to grab which address to send an email to
+	ToTemplates []string `toml:"toTemplates" override:"to-templates"`
 	// Close connection to SMTP server after idle timeout has elapsed
 	IdleTimeout toml.Duration `toml:"idle-timeout" override:"idle-timeout"`
 }
@@ -51,14 +53,15 @@ func (c Config) Validate() error {
 	if c.Enabled && c.From == "" {
 		return errors.New("must provide a 'from' address")
 	}
-	// Poor mans email validation, but since emails have a very large domain this is probably good enough
-	// to catch user error.
-	if c.From != "" && !strings.ContainsRune(c.From, '@') {
-		return fmt.Errorf("invalid from email address: %q", c.From)
+
+	if c.From != "" {
+		if _, err := mail.ParseAddress(c.From); err != nil {
+			return fmt.Errorf("invalid from email address: %q, %s", c.From, err.Error())
+		}
 	}
 	for _, t := range c.To {
-		if !strings.ContainsRune(t, '@') {
-			return fmt.Errorf("invalid to email address: %q", t)
+		if _, err := mail.ParseAddress(t); err != nil {
+			return fmt.Errorf("invalid to email address: %q, %s", t, err.Error())
 		}
 	}
 	return nil
diff --git a/services/smtp/service.go b/services/smtp/service.go
index 8cb01592..61c2acfa 100644
--- a/services/smtp/service.go
+++ b/services/smtp/service.go
@@ -1,16 +1,19 @@
 package smtp
 
 import (
+	"bytes"
 	"crypto/tls"
 	"errors"
 	"fmt"
+	"strconv"
 	"sync"
 	"sync/atomic"
+	text "text/template"
 	"time"
 
 	"github.com/influxdata/kapacitor/alert"
 	"github.com/influxdata/kapacitor/keyvalue"
-	"gopkg.in/gomail.v2"
+	gomail "gopkg.in/gomail.v2"
 )
 
 var ErrNoRecipients = errors.New("not sending email, no recipients defined")
@@ -176,7 +179,7 @@ func (s *Service) runMailer() {
 }
 
 func (s *Service) SendMail(to []string, subject, body string) error {
-	m, err := s.prepareMessge(to, subject, body)
+	m, err := s.prepareMessage(to, subject, body)
 	if err != nil {
 		return err
 	}
@@ -184,7 +187,7 @@ func (s *Service) SendMail(to []string, subject, body string) error {
 	return nil
 }
 
-func (s *Service) prepareMessge(to []string, subject, body string) (*gomail.Message, error) {
+func (s *Service) prepareMessage(to []string, subject, body string) (*gomail.Message, error) {
 	c := s.config()
 	if !c.Enabled {
 		return nil, errors.New("service is not enabled")
@@ -233,25 +236,49 @@ func (s *Service) Test(options interface{}) error {
 type HandlerConfig struct {
 	// List of email recipients.
 	To []string `mapstructure:"to"`
+
+	// ToTemplate allows you to template out email addresses
+	ToTemplates []string `mapstructure:"to-field"`
 }
 
 type handler struct {
-	s    *Service
-	c    HandlerConfig
-	diag Diagnostic
+	s           *Service
+	c           HandlerConfig
+	diag        Diagnostic
+	toTemplates []*text.Template
 }
 
 func (s *Service) Handler(c HandlerConfig, ctx ...keyvalue.T) alert.Handler {
-	return &handler{
+	h := &handler{
 		s:    s,
 		c:    c,
 		diag: s.diag.WithContext(ctx...),
 	}
+	for i := range c.ToTemplates {
+		tmpl, err := text.New(strconv.Itoa(i)).Parse(c.ToTemplates[i])
+		if err != nil {
+			h.diag.Error("bad template in email address template", err)
+			return h
+		}
+		h.toTemplates = append(h.toTemplates, tmpl)
+	}
+	return h
 }
 
 func (h *handler) Handle(event alert.Event) {
+	to := append([]string(nil), h.c.To...)
+	buf := &bytes.Buffer{}
+	for i := range h.toTemplates {
+		if err := h.toTemplates[i].ExecuteTemplate(buf, strconv.Itoa(i), event.Data); err != nil {
+			h.diag.Error("error in email template", err)
+		}
+		if buf.Len() != 0 {
+			to = append(to, buf.String())
+		}
+		buf.Reset()
+	}
 	if err := h.s.SendMail(
-		h.c.To,
+		to,
 		event.State.Message,
 		event.State.Details,
 	); err != nil {


Test patch:
diff --git a/integrations/streamer_test.go b/integrations/streamer_test.go
index 9e9b60cb..cc28cf92 100644
--- a/integrations/streamer_test.go
+++ b/integrations/streamer_test.go
@@ -10775,108 +10775,231 @@ stream
 }
 
 func TestStream_AlertEmail(t *testing.T) {
-	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" }}')
-		.details('''
+	var cases = []struct {
+		name     string
+		script   string
+		expected []*smtptest.Message
+	}{
+		{
+			name: "emails directly",
+			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" }}')
+				.details('''
 <b>{{.Message}}</b>
 
 Value: {{ index .Fields "count" }}
 <a href="http://graphs.example.com/host/{{index .Tags "host"}}">Details</a>
 ''')
-		.info(lambda: "count" > 6.0)
-		.warn(lambda: "count" > 7.0)
-		.crit(lambda: "count" > 8.0)
-		.email('user1@example.com', 'user2@example.com')
-		.email()
-			.to('user1@example.com', 'user2@example.com')
-`
+						.info(lambda: "count" > 6.0)
+						.warn(lambda: "count" > 7.0)
+						.crit(lambda: "count" > 8.0)
+						.email('user1@example.com', 'user2@example.com')
+						.email()
+							.to('user1@example.com', 'user2@example.com')
+				`,
+			expected: []*smtptest.Message{
+				{
+					Header: mail.Header{
+						"Mime-Version":              []string{"1.0"},
+						"Content-Type":              []string{"text/html; charset=UTF-8"},
+						"Content-Transfer-Encoding": []string{"quoted-printable"},
+						"To":                        []string{"user1@example.com, user2@example.com"},
+						"From":                      []string{"test@example.com"},
+						"Subject":                   []string{"kapacitor.cpu.serverA is CRITICAL"},
+					},
+					Body: `
+<b>kapacitor.cpu.serverA is CRITICAL</b>
+
+Value: 10
+<a href=3D"http://graphs.example.com/host/serverA">Details</a>
+`,
+				},
+				{
+					Header: mail.Header{
+						"Mime-Version":              []string{"1.0"},
+						"Content-Type":              []string{"text/html; charset=UTF-8"},
+						"Content-Transfer-Encoding": []string{"quoted-printable"},
+						"To":                        []string{"user1@example.com, user2@example.com"},
+						"From":                      []string{"test@example.com"},
+						"Subject":                   []string{"kapacitor.cpu.serverA is CRITICAL"},
+					},
+					Body: `
+<b>kapacitor.cpu.serverA is CRITICAL</b>
 
-	expMail := []*smtptest.Message{
+Value: 10
+<a href=3D"http://graphs.example.com/host/serverA">Details</a>
+`,
+				}},
+		},
 		{
-			Header: mail.Header{
-				"Mime-Version":              []string{"1.0"},
-				"Content-Type":              []string{"text/html; charset=UTF-8"},
-				"Content-Transfer-Encoding": []string{"quoted-printable"},
-				"To":                        []string{"user1@example.com, user2@example.com"},
-				"From":                      []string{"test@example.com"},
-				"Subject":                   []string{"kapacitor.cpu.serverA is CRITICAL"},
-			},
-			Body: `
+			name: "emails directly and in fields",
+			script: `stream
+					|from()
+						.measurement('cpu')
+						.where(lambda: "host" == 'serverA')
+						.groupBy('host')
+					|window()
+						.period(10s)
+						.every(10s)
+					|count('value')
+					|default()
+						.field('extraemail','bob@example.com')
+						.tag('tagemail','bob2@example.com')
+					|alert()
+						.id('kapacitor.{{ .Name }}.{{ index .Tags "host" }}')
+						.details('''
+<b>{{.Message}}</b>
+
+Value: {{ index .Fields "count" }}
+<a href="http://graphs.example.com/host/{{index .Tags "host"}}">Details</a>
+''')
+						.info(lambda: "count" > 6.0)
+						.warn(lambda: "count" > 7.0)
+						.crit(lambda: "count" > 8.0)
+						.email()
+							.to('user1@example.com', 'user2@example.com')
+							.toTemplates('{{ index .Fields "extraemail" }}')
+				`,
+			expected: []*smtptest.Message{
+				{
+					Header: mail.Header{
+						"Mime-Version":              []string{"1.0"},
+						"Content-Type":              []string{"text/html; charset=UTF-8"},
+						"Content-Transfer-Encoding": []string{"quoted-printable"},
+						"To":                        []string{"user1@example.com, user2@example.com, bob@example.com"},
+						"From":                      []string{"test@example.com"},
+						"Subject":                   []string{"kapacitor.cpu.serverA is CRITICAL"},
+					},
+					Body: `
 <b>kapacitor.cpu.serverA is CRITICAL</b>
 
 Value: 10
 <a href=3D"http://graphs.example.com/host/serverA">Details</a>
 `,
+				}},
 		},
 		{
-			Header: mail.Header{
-				"Mime-Version":              []string{"1.0"},
-				"Content-Type":              []string{"text/html; charset=UTF-8"},
-				"Content-Transfer-Encoding": []string{"quoted-printable"},
-				"To":                        []string{"user1@example.com, user2@example.com"},
-				"From":                      []string{"test@example.com"},
-				"Subject":                   []string{"kapacitor.cpu.serverA is CRITICAL"},
-			},
-			Body: `
+			name: "emails directly and in tag",
+			script: `stream
+							|from()
+								.measurement('cpu')
+								.where(lambda: "host" == 'serverA')
+								.groupBy('host')
+							|window()
+								.period(10s)
+								.every(10s)
+							|count('value')
+							|default()
+								.field('extraemail','bob@example.com')
+								.tag('tagemail','bob2@example.com')
+							|alert()
+								.id('kapacitor.{{ .Name }}.{{ index .Tags "host" }}')
+								.details('''
+<b>{{.Message}}</b>
+
+Value: {{ index .Fields "count" }}
+<a href="http://graphs.example.com/host/{{index .Tags "host"}}">Details</a>
+''')
+								.info(lambda: "count" > 6.0)
+								.warn(lambda: "count" > 7.0)
+								.crit(lambda: "count" > 8.0)
+								.email()
+									.to('user1@example.com', 'user2@example.com')
+									.toTemplates('{{ index .Tags "tagemail" }}')
+						`,
+			expected: []*smtptest.Message{
+				{
+					Header: mail.Header{
+						"Mime-Version":              []string{"1.0"},
+						"Content-Type":              []string{"text/html; charset=UTF-8"},
+						"Content-Transfer-Encoding": []string{"quoted-printable"},
+						"To":                        []string{"user1@example.com, user2@example.com, bob2@example.com"},
+						"From":                      []string{"test@example.com"},
+						"Subject":                   []string{"kapacitor.cpu.serverA is CRITICAL"},
+					},
+					Body: `
 <b>kapacitor.cpu.serverA is CRITICAL</b>
 
 Value: 10
 <a href=3D"http://graphs.example.com/host/serverA">Details</a>
 `,
+				},
+			},
 		},
 	}
+	for i := range cases {
+		t.Run(cases[i].name, func(t *testing.T) {
+			script := cases[i].script
+			expMail := cases[i].expected
+			smtpServer, err := smtptest.NewServer()
+			if err != nil {
+				t.Fatal(err)
+			}
+			defer func() {
+				err := smtpServer.Close()
+				if err != nil {
+					t.Fatalf("error in closing smtpService %v", err)
+				}
+			}()
 
-	smtpServer, err := smtptest.NewServer()
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer smtpServer.Close()
-	sc := smtp.Config{
-		Enabled: true,
-		Host:    smtpServer.Host,
-		Port:    smtpServer.Port,
-		From:    "test@example.com",
-	}
-	smtpService := smtp.NewService(sc, diagService.NewSMTPHandler())
-	if err := smtpService.Open(); err != nil {
-		t.Fatal(err)
-	}
-	defer smtpService.Close()
+			sc := smtp.Config{
+				Enabled: true,
+				Host:    smtpServer.Host,
+				Port:    smtpServer.Port,
+				From:    "test@example.com",
+			}
+			smtpService := smtp.NewService(sc, diagService.NewSMTPHandler())
+			if err := smtpService.Open(); err != nil {
+				t.Fatal(err)
+			}
 
-	tmInit := func(tm *kapacitor.TaskMaster) {
-		tm.SMTPService = smtpService
-	}
+			// make sure its closed
+			defer func() {
+				err := smtpService.Close()
+				if err != nil {
+					t.Fatalf("error in closing smtpService %v", err)
+				}
+			}()
 
-	testStreamerNoOutput(t, "TestStream_Alert", script, 13*time.Second, tmInit)
+			tmInit := func(tm *kapacitor.TaskMaster) {
+				tm.SMTPService = smtpService
+			}
 
-	// Close both client and server to ensure all message are processed
-	smtpService.Close()
-	smtpServer.Close()
+			testStreamerNoOutput(t, "TestStream_Alert", script, 13*time.Second, tmInit)
 
-	errors := smtpServer.Errors()
-	if got, exp := len(errors), 0; got != exp {
-		t.Errorf("unexpected smtp server errors: %v", errors)
-	}
+			// Close both client and server to ensure all message are processed
+			if err := smtpService.Close(); err != nil {
+				t.Fatalf("error in closing smtpService %v", err)
+			}
+			if err := smtpServer.Close(); err != nil {
+				t.Fatalf("error in closing smtpServer %v", err)
+			}
 
-	msgs := smtpServer.SentMessages()
-	if got, exp := len(msgs), len(expMail); got != exp {
-		t.Errorf("unexpected number of messages sent: got %d exp %d", got, exp)
-	}
-	for i, exp := range expMail {
-		got := msgs[i]
-		if err := exp.Compare(got); err != nil {
-			t.Errorf("%d %s", i, err)
-		}
+			errors := smtpServer.Errors()
+			if got, exp := len(errors), 0; got != exp {
+				t.Errorf("unexpected smtp server errors: %v", errors)
+			}
+
+			msgs := smtpServer.SentMessages()
+			if got, exp := len(msgs), len(expMail); got != exp {
+				t.Errorf("unexpected number of messages sent: got %d exp %d", got, exp)
+			}
+			for i, exp := range expMail {
+				got := msgs[i]
+				if err := exp.Compare(got); err != nil {
+					t.Errorf("%d %s", i, err)
+				}
+			}
+		})
 	}
 }
 
@@ -13672,6 +13795,7 @@ func testStreamerNoOutput(
 	duration time.Duration,
 	tmInit func(tm *kapacitor.TaskMaster),
 ) {
+	t.Helper()
 	clock, et, replayErr, tm := testStreamer(t, name, script, tmInit)
 	defer tm.Close()
 	err := fastForwardTask(clock, et, replayErr, tm, duration)
diff --git a/server/server_test.go b/server/server_test.go
index 72ec6fe3..c911f12c 100644
--- a/server/server_test.go
+++ b/server/server_test.go
@@ -7643,6 +7643,7 @@ func TestServer_UpdateConfig(t *testing.T) {
 						"port":               float64(25),
 						"state-changes-only": false,
 						"to":                 nil,
+						"to-templates":       nil,
 						"username":           "",
 					},
 					Redacted: []string{
@@ -7663,6 +7664,7 @@ func TestServer_UpdateConfig(t *testing.T) {
 					"port":               float64(25),
 					"state-changes-only": false,
 					"to":                 nil,
+					"to-templates":       nil,
 					"username":           "",
 				},
 				Redacted: []string{
@@ -7692,6 +7694,7 @@ func TestServer_UpdateConfig(t *testing.T) {
 								"password":           true,
 								"port":               float64(25),
 								"state-changes-only": false,
+								"to-templates":       nil,
 								"to":                 nil,
 								"username":           "",
 							},
@@ -7712,6 +7715,7 @@ func TestServer_UpdateConfig(t *testing.T) {
 							"password":           true,
 							"port":               float64(25),
 							"state-changes-only": false,
+							"to-templates":       nil,
 							"to":                 nil,
 							"username":           "",
 						},