VictoriaMetrics/app/vmalert/notifier/init.go

package notifier

import (
	"context"
	"flag"
	"fmt"
	"net/url"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/datasource"
	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httputil"
	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promrelabel"
	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutil"
)

var (
	configPath                    = flag.String("notifier.config", "", "Path to configuration file for notifiers")
	suppressDuplicateTargetErrors = flag.Bool("notifier.suppressDuplicateTargetErrors", false, "Whether to suppress 'duplicate target' errors during discovery")

	addrs = flagutil.NewArrayString("notifier.url", "Prometheus Alertmanager URL, e.g. http://127.0.0.1:9093. "+
		"List all Alertmanager URLs if it runs in the cluster mode to ensure high availability.")
	showNotifierURL = flag.Bool("notifier.showURL", false, "Whether to avoid stripping sensitive information such as passwords from URL in log messages or UI for -notifier.url. "+
		"It is hidden by default, since it can contain sensitive info such as auth key")
	blackHole = flag.Bool("notifier.blackhole", false, "Whether to blackhole alerting notifications. "+
		"Enable this flag if you want vmalert to evaluate alerting rules without sending any notifications to external receivers (eg. alertmanager). "+
		"-notifier.url, -notifier.config and -notifier.blackhole are mutually exclusive.")

	headers = flagutil.NewArrayString("notifier.headers", "Optional HTTP headers to send with each request to the corresponding -notifier.url. "+
		"For example, -remoteWrite.headers='My-Auth:foobar' would send 'My-Auth: foobar' HTTP header with every request to the corresponding -notifier.url. "+
		"Multiple headers must be delimited by '^^': -notifier.headers='header1:value1^^header2:value2,header3:value3'")
	basicAuthUsername     = flagutil.NewArrayString("notifier.basicAuth.username", "Optional basic auth username for -notifier.url")
	basicAuthUsernameFile = flagutil.NewArrayString("notifier.basicAuth.usernameFile", "Optional path to basic auth username file for -notifier.url")
	basicAuthPassword     = flagutil.NewArrayString("notifier.basicAuth.password", "Optional basic auth password for -notifier.url")
	basicAuthPasswordFile = flagutil.NewArrayString("notifier.basicAuth.passwordFile", "Optional path to basic auth password file for -notifier.url")

	bearerToken     = flagutil.NewArrayString("notifier.bearerToken", "Optional bearer token for -notifier.url")
	bearerTokenFile = flagutil.NewArrayString("notifier.bearerTokenFile", "Optional path to bearer token file for -notifier.url")

	tlsInsecureSkipVerify = flagutil.NewArrayBool("notifier.tlsInsecureSkipVerify", "Whether to skip tls verification when connecting to -notifier.url")
	tlsCertFile           = flagutil.NewArrayString("notifier.tlsCertFile", "Optional path to client-side TLS certificate file to use when connecting to -notifier.url")
	tlsKeyFile            = flagutil.NewArrayString("notifier.tlsKeyFile", "Optional path to client-side TLS certificate key to use when connecting to -notifier.url")
	tlsCAFile             = flagutil.NewArrayString("notifier.tlsCAFile", "Optional path to TLS CA file to use for verifying connections to -notifier.url. "+
		"By default, system CA is used")
	tlsServerName = flagutil.NewArrayString("notifier.tlsServerName", "Optional TLS server name to use for connections to -notifier.url. "+
		"By default, the server name from -notifier.url is used")

	oauth2ClientID = flagutil.NewArrayString("notifier.oauth2.clientID", "Optional OAuth2 clientID to use for -notifier.url. "+
		"If multiple args are set, then they are applied independently for the corresponding -notifier.url")
	oauth2ClientSecret = flagutil.NewArrayString("notifier.oauth2.clientSecret", "Optional OAuth2 clientSecret to use for -notifier.url. "+
		"If multiple args are set, then they are applied independently for the corresponding -notifier.url")
	oauth2ClientSecretFile = flagutil.NewArrayString("notifier.oauth2.clientSecretFile", "Optional OAuth2 clientSecretFile to use for -notifier.url. "+
		"If multiple args are set, then they are applied independently for the corresponding -notifier.url")
	oauth2EndpointParams = flagutil.NewArrayString("notifier.oauth2.endpointParams", "Optional OAuth2 endpoint parameters to use for the corresponding -notifier.url . "+
		`The endpoint parameters must be set in JSON format: {"param1":"value1",...,"paramN":"valueN"}`)
	oauth2TokenURL = flagutil.NewArrayString("notifier.oauth2.tokenUrl", "Optional OAuth2 tokenURL to use for -notifier.url. "+
		"If multiple args are set, then they are applied independently for the corresponding -notifier.url")
	oauth2Scopes = flagutil.NewArrayString("notifier.oauth2.scopes", "Optional OAuth2 scopes to use for -notifier.url. Scopes must be delimited by ';'. "+
		"If multiple args are set, then they are applied independently for the corresponding -notifier.url")
	sendTimeout = flagutil.NewArrayDuration("notifier.sendTimeout", 10*time.Second, "Timeout when sending alerts to the corresponding -notifier.url")
)

// AlertURLGeneratorFn returns a URL to the passed alert object.
// Call InitAlertURLGeneratorFn before using this function.
var AlertURLGeneratorFn AlertURLGenerator

// InitAlertURLGeneratorFn populates AlertURLGeneratorFn
func InitAlertURLGeneratorFn(externalURL *url.URL, externalAlertSource string, validateTemplate bool) error {
	if externalAlertSource == "" {
		AlertURLGeneratorFn = func(a Alert) string {
			gID, aID := strconv.FormatUint(a.GroupID, 10), strconv.FormatUint(a.ID, 10)
			return fmt.Sprintf("%s/vmalert/alert?%s=%s&%s=%s", externalURL, "group_id", gID, "alert_id", aID)
		}
		return nil
	}
	if validateTemplate {
		if err := ValidateTemplates(map[string]string{
			"tpl": externalAlertSource,
		}); err != nil {
			return fmt.Errorf("error validating source template %s: %w", externalAlertSource, err)
		}
	}
	m := map[string]string{
		"tpl": externalAlertSource,
	}
	AlertURLGeneratorFn = func(alert Alert) string {
		qFn := func(_ string) ([]datasource.Metric, error) {
			return nil, fmt.Errorf("`query` template isn't supported for alert source template")
		}
		templated, err := alert.ExecTemplate(qFn, alert.Labels, m)
		if err != nil {
			logger.Errorf("cannot template alert source: %s", err)
		}
		return fmt.Sprintf("%s/%s", externalURL, templated["tpl"])
	}
	return nil
}

var (
	// getActiveNotifiers returns the current list of Notifier objects.
	getActiveNotifiers func() []Notifier
	// globalRelabelCfg stores the parsed alert relabeling config from the config file if there is
	globalRelabelCfg *promrelabel.ParsedConfigs

	// cw holds a configWatcher for configPath configuration file
	// configWatcher provides a list of Notifier objects discovered
	// from static config or via service discovery.
	// cw is not nil only if configPath is provided.
	cw *configWatcher

	// externalLabels is a global variable for holding external labels configured via flags
	// It is supposed to be inited via Init function only.
	externalLabels map[string]string
	// externalURL is a global variable for holding external URL value configured via flag
	// It is supposed to be inited via Init function only.
	externalURL string
)

// Reload checks the changes in configPath configuration file
// and applies changes if any.
func Reload() error {
	if cw == nil {
		return nil
	}
	return cw.reload(*configPath)
}

// Init works in two mods:
//   - configuration via flags (for backward compatibility). Is always static
//     and don't support live reloads.
//   - configuration via file. Supports live reloads and service discovery.
//
// Init returns an error if both mods are used.
func Init(extLabels map[string]string, extURL string) error {
	externalURL = extURL
	externalLabels = extLabels
	_, err := url.Parse(externalURL)
	if err != nil {
		return fmt.Errorf("failed to parse external URL: %w", err)
	}

	if *blackHole {
		if len(*addrs) > 0 || *configPath != "" {
			return fmt.Errorf("only one of -notifier.blackhole, -notifier.url and -notifier.config flags must be specified")
		}
		notifier := newBlackHoleNotifier()
		getActiveNotifiers = func() []Notifier {
			return []Notifier{notifier}
		}
		return nil
	}

	if *configPath == "" && len(*addrs) == 0 {
		return nil
	}
	if *configPath != "" && len(*addrs) > 0 {
		return fmt.Errorf("only one of -notifier.config or -notifier.url flags must be specified")
	}

	if len(*addrs) > 0 {
		notifiers, err := notifiersFromFlags(AlertURLGeneratorFn)
		if err != nil {
			return fmt.Errorf("failed to create notifier from flag values: %w", err)
		}
		getActiveNotifiers = func() []Notifier {
			return notifiers
		}
		return nil
	}

	cfg, err := parseConfig(*configPath)
	if err != nil {
		return err
	}
	if cfg.AlertRelabelConfigs != nil {
		globalRelabelCfg = cfg.parsedAlertRelabelConfigs
	}
	cw, err = newWatcher(cfg, AlertURLGeneratorFn)
	if err != nil {
		return fmt.Errorf("failed to init config watcher: %w", err)
	}
	getActiveNotifiers = cw.notifiers
	return nil
}

// InitSecretFlags must be called after flag.Parse and before any logging
func InitSecretFlags() {
	if !*showNotifierURL {
		flagutil.RegisterSecretFlag("notifier.url")
	}
}

func notifiersFromFlags(gen AlertURLGenerator) ([]Notifier, error) {
	var notifiers []Notifier
	for i, addr := range *addrs {
		endpointParamsJSON := oauth2EndpointParams.GetOptionalArg(i)
		endpointParams, err := flagutil.ParseJSONMap(endpointParamsJSON)
		if err != nil {
			return nil, fmt.Errorf("cannot parse JSON for -notifier.oauth2.endpointParams=%s: %w", endpointParamsJSON, err)
		}
		authCfg := promauth.HTTPClientConfig{
			TLSConfig: &promauth.TLSConfig{
				CAFile:             tlsCAFile.GetOptionalArg(i),
				CertFile:           tlsCertFile.GetOptionalArg(i),
				KeyFile:            tlsKeyFile.GetOptionalArg(i),
				ServerName:         tlsServerName.GetOptionalArg(i),
				InsecureSkipVerify: tlsInsecureSkipVerify.GetOptionalArg(i),
			},
			BasicAuth: &promauth.BasicAuthConfig{
				Username:     basicAuthUsername.GetOptionalArg(i),
				UsernameFile: basicAuthUsernameFile.GetOptionalArg(i),
				Password:     promauth.NewSecret(basicAuthPassword.GetOptionalArg(i)),
				PasswordFile: basicAuthPasswordFile.GetOptionalArg(i),
			},
			BearerToken:     promauth.NewSecret(bearerToken.GetOptionalArg(i)),
			BearerTokenFile: bearerTokenFile.GetOptionalArg(i),
			OAuth2: &promauth.OAuth2Config{
				ClientID:         oauth2ClientID.GetOptionalArg(i),
				ClientSecret:     promauth.NewSecret(oauth2ClientSecret.GetOptionalArg(i)),
				ClientSecretFile: oauth2ClientSecretFile.GetOptionalArg(i),
				EndpointParams:   endpointParams,
				Scopes:           strings.Split(oauth2Scopes.GetOptionalArg(i), ";"),
				TokenURL:         oauth2TokenURL.GetOptionalArg(i),
			},
			Headers: []string{headers.GetOptionalArg(i)},
		}

		if err := httputil.CheckURL(addr); err != nil {
			return nil, fmt.Errorf("invalid notifier.url %q: %w", addr, err)
		}
		addr = strings.TrimSuffix(addr, "/")
		am, err := NewAlertManager(addr+alertManagerPath, gen, authCfg, nil, sendTimeout.GetOptionalArg(i))
		if err != nil {
			return nil, err
		}
		notifiers = append(notifiers, am)
	}
	return notifiers, nil
}

// Target represents a Notifier and optional
// list of labels added during discovery.
type Target struct {
	Notifier
	Labels *promutil.Labels
}

// TargetType defines how the Target was discovered
type TargetType string

const (
	// TargetStatic is for targets configured statically
	TargetStatic TargetType = "static"
	// TargetConsul is for targets discovered via Consul
	TargetConsul TargetType = "consulSD"
	// TargetDNS is for targets discovered via DNS
	TargetDNS TargetType = "DNSSD"
)

// GetTargets returns list of static or discovered targets
// via notifier configuration.
//
// Must be called after Init.
func GetTargets() map[TargetType][]Target {
	if getActiveNotifiers == nil {
		return nil
	}
	targets := make(map[TargetType][]Target)
	// use cached targets from configWatcher instead of getActiveNotifiers for the extra target labels
	if cw != nil {
		cw.targetsMu.RLock()
		for key, ns := range cw.targets {
			targets[key] = append(targets[key], ns...)
		}
		cw.targetsMu.RUnlock()
		return targets
	}

	// static notifiers don't have labels
	for _, ns := range getActiveNotifiers() {
		targets[TargetStatic] = append(targets[TargetStatic], Target{
			Notifier: ns,
		})
	}
	return targets
}

// Send sends alerts to all active notifiers
func Send(ctx context.Context, alerts []Alert, notifierHeaders map[string]string) chan error {
	alertsToSend := make([]Alert, 0, len(alerts))
	lblss := make([][]prompb.Label, 0, len(alerts))
	// apply global relabel config first without modifying original alerts in alerts
	for _, a := range alerts {
		lbls := a.applyRelabelingIfNeeded(globalRelabelCfg)
		if len(lbls) == 0 {
			continue
		}
		alertsToSend = append(alertsToSend, a)
		lblss = append(lblss, lbls)
	}

	wg := sync.WaitGroup{}
	activeNotifiers := getActiveNotifiers()
	errCh := make(chan error, len(activeNotifiers))
	defer close(errCh)
	for i := range activeNotifiers {
		nt := activeNotifiers[i]
		wg.Go(func() {
			if err := nt.Send(ctx, alertsToSend, lblss, notifierHeaders); err != nil {
				errCh <- fmt.Errorf("failed to send alerts to addr %q: %w", nt.Addr(), err)
			}
		})
	}
	wg.Wait()
	return errCh
}