mirror of
https://github.com/grafana/grafana.git
synced 2024-11-21 16:38:03 -06:00
Notifications: Optional trace propagation through SMTP (#80481)
* Notifications: Optional trace propagation through SMTP Signed-off-by: Dave Henderson <dave.henderson@grafana.com> * fix failing test Signed-off-by: Dave Henderson <dave.henderson@grafana.com> * Add documentation Signed-off-by: Dave Henderson <dave.henderson@grafana.com> --------- Signed-off-by: Dave Henderson <dave.henderson@grafana.com>
This commit is contained in:
parent
6b8b741b3b
commit
e0402115ea
@ -947,6 +947,7 @@ from_address = admin@grafana.localhost
|
||||
from_name = Grafana
|
||||
ehlo_identity =
|
||||
startTLS_policy =
|
||||
enable_tracing = false
|
||||
|
||||
[smtp.static_headers]
|
||||
# Include custom static headers in all outgoing emails
|
||||
|
@ -890,6 +890,8 @@
|
||||
;ehlo_identity = dashboard.example.com
|
||||
# SMTP startTLS policy (defaults to 'OpportunisticStartTLS')
|
||||
;startTLS_policy = NoStartTLS
|
||||
# Enable trace propagation in e-mail headers, using the 'traceparent', 'tracestate' and (optionally) 'baggage' fields (defaults to false)
|
||||
;enable_tracing = false
|
||||
|
||||
[smtp.static_headers]
|
||||
# Include custom static headers in all outgoing emails
|
||||
|
@ -1281,6 +1281,10 @@ Name to be used as client identity for EHLO in SMTP dialog, default is `<instanc
|
||||
|
||||
Either "OpportunisticStartTLS", "MandatoryStartTLS", "NoStartTLS". Default is `empty`.
|
||||
|
||||
### enable_tracing
|
||||
|
||||
Enable trace propagation in e-mail headers, using the `traceparent`, `tracestate` and (optionally) `baggage` fields. Default is `false`. To enable, you must first configure tracing in one of the `tracing.oentelemetry.*` sections.
|
||||
|
||||
<hr>
|
||||
|
||||
## [smtp.static_headers]
|
||||
|
@ -6,6 +6,7 @@ package notifications
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"fmt"
|
||||
"html/template"
|
||||
"net/mail"
|
||||
@ -34,10 +35,10 @@ func init() {
|
||||
}
|
||||
|
||||
type Mailer interface {
|
||||
Send(messages ...*Message) (int, error)
|
||||
Send(ctx context.Context, messages ...*Message) (int, error)
|
||||
}
|
||||
|
||||
func (ns *NotificationService) Send(msg *Message) (int, error) {
|
||||
func (ns *NotificationService) Send(ctx context.Context, msg *Message) (int, error) {
|
||||
messages := []*Message{}
|
||||
|
||||
if msg.SingleEmail {
|
||||
@ -50,7 +51,7 @@ func (ns *NotificationService) Send(msg *Message) (int, error) {
|
||||
}
|
||||
}
|
||||
|
||||
return ns.mailer.Send(messages...)
|
||||
return ns.mailer.Send(ctx, messages...)
|
||||
}
|
||||
|
||||
func (ns *NotificationService) buildEmailMessage(cmd *SendEmailCommand) (*Message, error) {
|
||||
|
@ -112,7 +112,7 @@ func (ns *NotificationService) Run(ctx context.Context) error {
|
||||
ns.log.Error("Failed to send webrequest ", "error", err)
|
||||
}
|
||||
case msg := <-ns.mailQueue:
|
||||
num, err := ns.Send(msg)
|
||||
num, err := ns.Send(ctx, msg)
|
||||
tos := strings.Join(msg.To, "; ")
|
||||
info := ""
|
||||
if err != nil {
|
||||
@ -203,7 +203,7 @@ func (ns *NotificationService) SendEmailCommandHandlerSync(ctx context.Context,
|
||||
return err
|
||||
}
|
||||
|
||||
_, err = ns.Send(message)
|
||||
_, err = ns.Send(ctx, message)
|
||||
return err
|
||||
}
|
||||
|
||||
|
@ -1,18 +1,29 @@
|
||||
package notifications
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/tls"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"net/textproto"
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
"go.opentelemetry.io/otel"
|
||||
"go.opentelemetry.io/otel/attribute"
|
||||
"go.opentelemetry.io/otel/codes"
|
||||
"go.opentelemetry.io/otel/propagation"
|
||||
"go.opentelemetry.io/otel/trace"
|
||||
gomail "gopkg.in/mail.v2"
|
||||
|
||||
"github.com/grafana/grafana/pkg/setting"
|
||||
)
|
||||
|
||||
var tracer = otel.Tracer("github.com/grafana/grafana/pkg/services/notifications")
|
||||
|
||||
type SmtpClient struct {
|
||||
cfg setting.SmtpSettings
|
||||
}
|
||||
@ -29,7 +40,12 @@ func NewSmtpClient(cfg setting.SmtpSettings) (*SmtpClient, error) {
|
||||
return client, nil
|
||||
}
|
||||
|
||||
func (sc *SmtpClient) Send(messages ...*Message) (int, error) {
|
||||
func (sc *SmtpClient) Send(ctx context.Context, messages ...*Message) (int, error) {
|
||||
ctx, span := tracer.Start(ctx, "notifications.SmtpClient.Send",
|
||||
trace.WithAttributes(attribute.Int("messages", len(messages))),
|
||||
)
|
||||
defer span.End()
|
||||
|
||||
sentEmailsCount := 0
|
||||
dialer, err := sc.createDialer()
|
||||
if err != nil {
|
||||
@ -37,7 +53,12 @@ func (sc *SmtpClient) Send(messages ...*Message) (int, error) {
|
||||
}
|
||||
|
||||
for _, msg := range messages {
|
||||
m := sc.buildEmail(msg)
|
||||
span.SetAttributes(
|
||||
attribute.String("smtp.sender", msg.From),
|
||||
attribute.StringSlice("smtp.recipients", msg.To),
|
||||
)
|
||||
|
||||
m := sc.buildEmail(ctx, msg)
|
||||
|
||||
innerError := dialer.DialAndSend(m)
|
||||
emailsSentTotal.Inc()
|
||||
@ -50,6 +71,9 @@ func (sc *SmtpClient) Send(messages ...*Message) (int, error) {
|
||||
}
|
||||
|
||||
err = fmt.Errorf("failed to send notification to email addresses: %s: %w", strings.Join(msg.To, ";"), innerError)
|
||||
span.RecordError(err)
|
||||
span.SetStatus(codes.Error, err.Error())
|
||||
|
||||
continue
|
||||
}
|
||||
|
||||
@ -60,7 +84,7 @@ func (sc *SmtpClient) Send(messages ...*Message) (int, error) {
|
||||
}
|
||||
|
||||
// buildEmail converts the Message DTO to a gomail message.
|
||||
func (sc *SmtpClient) buildEmail(msg *Message) *gomail.Message {
|
||||
func (sc *SmtpClient) buildEmail(ctx context.Context, msg *Message) *gomail.Message {
|
||||
m := gomail.NewMessage()
|
||||
// add all static headers to the email message
|
||||
for h, val := range sc.cfg.StaticHeaders {
|
||||
@ -69,6 +93,11 @@ func (sc *SmtpClient) buildEmail(msg *Message) *gomail.Message {
|
||||
m.SetHeader("From", msg.From)
|
||||
m.SetHeader("To", msg.To...)
|
||||
m.SetHeader("Subject", msg.Subject)
|
||||
|
||||
if sc.cfg.EnableTracing {
|
||||
otel.GetTextMapPropagator().Inject(ctx, gomailHeaderCarrier{m})
|
||||
}
|
||||
|
||||
sc.setFiles(m, msg)
|
||||
for _, replyTo := range msg.ReplyTo {
|
||||
m.SetAddressHeader("Reply-To", replyTo, "")
|
||||
@ -149,3 +178,36 @@ func getStartTLSPolicy(policy string) gomail.StartTLSPolicy {
|
||||
return 0
|
||||
}
|
||||
}
|
||||
|
||||
type gomailHeaderCarrier struct {
|
||||
*gomail.Message
|
||||
}
|
||||
|
||||
var _ propagation.TextMapCarrier = (*gomailHeaderCarrier)(nil)
|
||||
|
||||
func (c gomailHeaderCarrier) Get(key string) string {
|
||||
if hdr := c.Message.GetHeader(key); len(hdr) > 0 {
|
||||
return hdr[0]
|
||||
}
|
||||
|
||||
return ""
|
||||
}
|
||||
|
||||
func (c gomailHeaderCarrier) Set(key string, value string) {
|
||||
c.Message.SetHeader(key, value)
|
||||
}
|
||||
|
||||
func (c gomailHeaderCarrier) Keys() []string {
|
||||
// there's no way to get all the header keys directly from a gomail.Message,
|
||||
// but we can encode the whole message and re-parse. This is not ideal, but
|
||||
// this function shouldn't be used in the hot path.
|
||||
buf := bytes.Buffer{}
|
||||
_, _ = c.Message.WriteTo(&buf)
|
||||
hdr, _ := textproto.NewReader(bufio.NewReader(&buf)).ReadMIMEHeader()
|
||||
keys := make([]string, 0, len(hdr))
|
||||
for k := range hdr {
|
||||
keys = append(keys, k)
|
||||
}
|
||||
|
||||
return keys
|
||||
}
|
||||
|
@ -2,12 +2,14 @@ package notifications
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/grafana/grafana/pkg/infra/tracing"
|
||||
"github.com/grafana/grafana/pkg/setting"
|
||||
)
|
||||
|
||||
@ -30,8 +32,10 @@ func TestBuildMail(t *testing.T) {
|
||||
ReplyTo: []string{"from@address.com"},
|
||||
}
|
||||
|
||||
ctx := context.Background()
|
||||
|
||||
t.Run("Can successfully build mail", func(t *testing.T) {
|
||||
email := sc.buildEmail(message)
|
||||
email := sc.buildEmail(ctx, message)
|
||||
staticHeader := email.GetHeader("Foo-Header")[0]
|
||||
assert.Equal(t, staticHeader, "foo_value")
|
||||
|
||||
@ -45,9 +49,35 @@ func TestBuildMail(t *testing.T) {
|
||||
assert.Contains(t, buf.String(), "Some plain text body")
|
||||
assert.Less(t, strings.Index(buf.String(), "Some plain text body"), strings.Index(buf.String(), "Some HTML body"))
|
||||
})
|
||||
|
||||
t.Run("Skips trace headers when context has no span", func(t *testing.T) {
|
||||
cfg.Smtp.EnableTracing = true
|
||||
|
||||
sc, err := NewSmtpClient(cfg.Smtp)
|
||||
require.NoError(t, err)
|
||||
|
||||
email := sc.buildEmail(ctx, message)
|
||||
assert.Empty(t, email.GetHeader("traceparent"))
|
||||
})
|
||||
|
||||
t.Run("Adds trace headers when context has span", func(t *testing.T) {
|
||||
cfg.Smtp.EnableTracing = true
|
||||
|
||||
sc, err := NewSmtpClient(cfg.Smtp)
|
||||
require.NoError(t, err)
|
||||
|
||||
tracer := tracing.InitializeTracerForTest()
|
||||
ctx, span := tracer.Start(ctx, "notifications.SmtpClient.SendContext")
|
||||
defer span.End()
|
||||
|
||||
email := sc.buildEmail(ctx, message)
|
||||
assert.NotEmpty(t, email.GetHeader("traceparent"))
|
||||
})
|
||||
}
|
||||
|
||||
func TestSmtpDialer(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
|
||||
t.Run("When SMTP hostname is invalid", func(t *testing.T) {
|
||||
cfg := createSmtpConfig()
|
||||
cfg.Smtp.Host = "invalid%hostname:123:456"
|
||||
@ -63,7 +93,7 @@ func TestSmtpDialer(t *testing.T) {
|
||||
},
|
||||
}
|
||||
|
||||
count, err := client.Send(message)
|
||||
count, err := client.Send(ctx, message)
|
||||
|
||||
require.Equal(t, 0, count)
|
||||
require.EqualError(t, err, "address invalid%hostname:123:456: too many colons in address")
|
||||
@ -84,7 +114,7 @@ func TestSmtpDialer(t *testing.T) {
|
||||
},
|
||||
}
|
||||
|
||||
count, err := client.Send(message)
|
||||
count, err := client.Send(ctx, message)
|
||||
|
||||
require.Equal(t, 0, count)
|
||||
require.EqualError(t, err, "strconv.Atoi: parsing \"123a\": invalid syntax")
|
||||
@ -106,7 +136,7 @@ func TestSmtpDialer(t *testing.T) {
|
||||
},
|
||||
}
|
||||
|
||||
count, err := client.Send(message)
|
||||
count, err := client.Send(ctx, message)
|
||||
|
||||
require.Equal(t, 0, count)
|
||||
require.EqualError(t, err, "could not load cert or key file: open /var/certs/does-not-exist.pem: no such file or directory")
|
||||
|
@ -1,6 +1,9 @@
|
||||
package notifications
|
||||
|
||||
import "fmt"
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
)
|
||||
|
||||
type FakeMailer struct {
|
||||
Sent []*Message
|
||||
@ -12,7 +15,7 @@ func NewFakeMailer() *FakeMailer {
|
||||
}
|
||||
}
|
||||
|
||||
func (fm *FakeMailer) Send(messages ...*Message) (int, error) {
|
||||
func (fm *FakeMailer) Send(ctx context.Context, messages ...*Message) (int, error) {
|
||||
sentEmailsCount := 0
|
||||
for _, msg := range messages {
|
||||
fm.Sent = append(fm.Sent, msg)
|
||||
@ -27,7 +30,7 @@ func NewFakeDisconnectedMailer() *FakeDisconnectedMailer {
|
||||
return &FakeDisconnectedMailer{}
|
||||
}
|
||||
|
||||
func (fdm *FakeDisconnectedMailer) Send(messages ...*Message) (int, error) {
|
||||
func (fdm *FakeDisconnectedMailer) Send(ctx context.Context, messages ...*Message) (int, error) {
|
||||
return 0, fmt.Errorf("connect: connection refused")
|
||||
}
|
||||
|
||||
|
@ -20,6 +20,7 @@ type SmtpSettings struct {
|
||||
StartTLSPolicy string
|
||||
SkipVerify bool
|
||||
StaticHeaders map[string]string
|
||||
EnableTracing bool
|
||||
|
||||
SendWelcomeEmailOnSignUp bool
|
||||
TemplatesPatterns []string
|
||||
@ -53,6 +54,8 @@ func (cfg *Cfg) readSmtpSettings() error {
|
||||
return err
|
||||
}
|
||||
|
||||
cfg.Smtp.EnableTracing = sec.Key("enable_tracing").MustBool(false)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
|
Loading…
Reference in New Issue
Block a user