First version of the program

README.md

@@ -1,2 +1,66 @@
 # prometheus-xmpp-alerting
+
 Basic XMPP Alertmanager Webhook Receiver for Prometheus
+
+## Purpose
+
+This repository has been made to receive Prometheus alerts on my Phone without relying on a third party provider.  
+To do so I have installed on my Raspberry PI:
+
+ - [Prometheus](https://prometheus.io/)
+ - [Alertmanager](https://prometheus.io/docs/alerting/alertmanager/)
+ - [Prosody](https://prosody.im/), an XMPP server
+
+On my phone, I have just installed an XMPP client.
+
+## Having a working Golang environment:
+
+```bash
+go get github.com/trazfr/prometheus-xmpp-alerting
+go install github.com/trazfr/prometheus-xmpp-alerting
+```
+
+## Use
+
+This program is configured through a JSON file.
+
+To run, just `prometheus-xmpp-alerting config.json`
+
+This example of configuration file shows:
+
+ - the webhook listening on `127.0.0.1:9091`
+ - when the instance is starting, it sends to everyone `Prometheus Monitoring Started`
+ - that it sends a different message depending on a `severity` label
+ - the program uses the XMPP user `monitoring@example.com` with a password
+ - when it is working, it has the status `Monitoring Prometheus...`
+ - it doesn't use a TLS socket due to the `no_tls` flag. Actually it will use STARTTLS due to the server configuration
+ - it doesn't check the TLS certificates thanks to `tls_insecure` (for some reason, it doesn't work on my Prosody install, but as I'm connecting to localhost, it doesn't matter)
+ - each time it receives an alert, it sends a notification to 2 XMPP accounts `on-duty-1@example.com` and `on-duty-2@example.com`.
+
+```json
+{
+    "listen": "127.0.0.1:9091",
+    "startup_message": "Prometheus Monitoring Started",
+    "firing": "{{ if eq .Labels.severity \"error\" }}🔥{{ else if eq .Labels.severity \"warning\" }}💣{{ else }}💡{{ end }} Firing {{ .Labels.alertname }}\n{{ .Annotations.description }} since {{ .StartsAt }}\n{{ .GeneratorURL }}",
+    "xmpp": {
+        "user": "monitoring@example.com",
+        "password": "MyXmppPassword",
+        "status": "Monitoring Prometheus...",
+        "no_tls": true,
+        "tls_insecure": true,
+        "send_notif": [
+            "on-duty-1@example.com",
+            "on-duty-2@example.com"
+        ]
+    }
+}
+```
+
+## Features
+
+This program uses HTTP with 3 different paths:
+
+ - `/alert` is used by Prometheus' Alertmanager to send alerts
+ - `/send` is mainly used for debugging or if one just want to send simple message from another program. To send a message:  
+   `curl -H 'Content-Type: text/plain' -X POST <my_ip:port>/send -d 'my message'`
+ - `/metrics` to be scrapped by Prometheus. It exposes some basic metrics

alert_handler.go

@@ -0,0 +1,73 @@
+package main
+
+import (
+	"bytes"
+	"encoding/json"
+	"log"
+	"net/http"
+	"text/template"
+
+	promTemplate "github.com/prometheus/alertmanager/template"
+)
+
+type alertHandler struct {
+	sender           Sender
+	firingTemplate   *template.Template
+	resolvedTemplate *template.Template
+}
+
+// NewAlertHandler create an HTTP handler to receive prometheus webhook alerts
+func NewAlertHandler(config *Config, sender Sender) http.Handler {
+	return &alertHandler{
+		sender:           sender,
+		firingTemplate:   config.Firing,
+		resolvedTemplate: config.Resolved,
+	}
+}
+
+func (a *alertHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	defer r.Body.Close()
+
+	// https://godoc.org/github.com/prometheus/alertmanager/template#Data
+	promAlert := promTemplate.Data{}
+	if err := json.NewDecoder(r.Body).Decode(&promAlert); err != nil {
+		a.handleError(w, http.StatusBadRequest, err, "Cannot decode payload")
+		return
+	}
+	promAlertTriggeredMetric.Inc()
+
+	a.instantiateTemplate(a.firingTemplate, promAlert.Alerts.Firing())
+	a.instantiateTemplate(a.resolvedTemplate, promAlert.Alerts.Resolved())
+}
+
+func (a *alertHandler) handleError(w http.ResponseWriter, statusCode int, err error, message string) {
+	w.WriteHeader(statusCode)
+	w.Write([]byte("Error: "))
+	w.Write([]byte(err.Error()))
+	if message != "" {
+		w.Write([]byte("\n"))
+		w.Write([]byte(message))
+	}
+}
+
+func (a *alertHandler) instantiateTemplate(tmpl *template.Template, alerts []promTemplate.Alert) {
+	if tmpl == nil {
+		return
+	}
+
+	for alertIdx := range alerts {
+		if message := a.generateString(tmpl, &alerts[alertIdx]); message != "" {
+			promAlertsProcessedMetric.Inc()
+			a.sender.Send(message)
+		}
+	}
+}
+
+func (a *alertHandler) generateString(tmpl *template.Template, alert *promTemplate.Alert) string {
+	buf := bytes.Buffer{}
+	if err := tmpl.Execute(&buf, alert); err != nil {
+		log.Printf("Could not instantiate template :%s\n", err)
+		return ""
+	}
+	return buf.String()
+}

config.go

@@ -0,0 +1,100 @@
+package main
+
+import (
+	"encoding/json"
+	"log"
+	"os"
+	"sort"
+	"text/template"
+)
+
+// Config is the internal configuration type
+type Config struct {
+	Listen         string
+	Debug          bool
+	StartupMessage string
+	Firing         *template.Template
+	Resolved       *template.Template
+	XMPP           ConfigXMPP
+}
+
+// ConfigXMPP is the configuration for XMPP connection
+type ConfigXMPP struct {
+	User        string
+	Password    string
+	SendNotif   []string
+	Status      string
+	NoTLS       bool
+	TLSInsecure bool
+}
+
+type internalConfig struct {
+	Debug          bool               `json:"debug"`
+	Listen         string             `json:"listen"`
+	StartupMessage string             `json:"startup_message"`
+	Firing         string             `json:"firing"`
+	Resolved       string             `json:"resolved"`
+	XMPP           internalConfigXMPP `json:"xmpp"`
+}
+
+type internalConfigXMPP struct {
+	User        string   `json:"user"`
+	Password    string   `json:"password"`
+	SendNotif   []string `json:"send_notif"`
+	Status      string   `json:"status"`
+	NoTLS       bool     `json:"no_tls"`
+	TLSInsecure bool     `json:"tls_insecure"`
+}
+
+// NewConfig reads the JSON file filename and generates a configuration
+func NewConfig(filename string) *Config {
+	fd, err := os.Open(filename)
+	if err != nil {
+		log.Fatalln(err)
+	}
+	defer fd.Close()
+
+	internalConfig := &internalConfig{
+		Listen: ":9091",
+		XMPP: internalConfigXMPP{
+			Status: "Monitoring",
+		},
+	}
+	if err := json.NewDecoder(fd).Decode(internalConfig); err != nil {
+		log.Fatalln(err)
+	}
+
+	return internalConfig.parse()
+}
+
+func (i *internalConfig) parse() *Config {
+	return &Config{
+		Debug:          i.Debug,
+		Listen:         i.Listen,
+		StartupMessage: i.StartupMessage,
+		Firing:         parseTemplate(i.Firing),
+		Resolved:       parseTemplate(i.Resolved),
+		XMPP:           i.XMPP.parse(),
+	}
+}
+
+func (i *internalConfigXMPP) parse() ConfigXMPP {
+	result := ConfigXMPP{
+		User:        i.User,
+		Password:    i.Password,
+		SendNotif:   i.SendNotif,
+		Status:      i.Status,
+		NoTLS:       i.NoTLS,
+		TLSInsecure: i.TLSInsecure,
+	}
+	sort.Strings(result.SendNotif)
+	return result
+}
+
+func parseTemplate(tmpl string) *template.Template {
+	if tmpl == "" {
+		return nil
+	}
+
+	return template.Must(template.New("").Parse(tmpl))
+}

interfaces.go

@@ -0,0 +1,12 @@
+package main
+
+// SendCloser is the interface to send messages
+type SendCloser interface {
+	Sender
+	Close() error
+}
+
+// Sender is the interface to send messages
+type Sender interface {
+	Send(message string) error
+}

main.go

@@ -0,0 +1,25 @@
+package main
+
+import (
+	"fmt"
+	"net/http"
+	"os"
+
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+)
+
+func main() {
+	if len(os.Args) != 2 {
+		fmt.Fprintln(os.Stderr, "Usage", os.Args[0], "<config_file>")
+		os.Exit(1)
+	}
+
+	config := NewConfig(os.Args[1])
+	xmpp := NewXMPP(config)
+	defer xmpp.Close()
+
+	http.Handle("/alert", NewAlertHandler(config, xmpp))
+	http.Handle("/send", NewSendHandler(xmpp))
+	http.Handle("/metrics", promhttp.Handler())
+	fmt.Println(http.ListenAndServe(config.Listen, nil))
+}

prometheus.go

@@ -0,0 +1,45 @@
+package main
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+const (
+	promNamespace = "xmpp"
+)
+
+var (
+	promAlertTriggeredMetric = prometheus.NewCounter(prometheus.CounterOpts{
+		Namespace: promNamespace,
+		Name:      "alert_trigger_total",
+		Help:      "Number of successful calls to the /alert webhook.",
+	})
+	promAlertsProcessedMetric = prometheus.NewCounter(prometheus.CounterOpts{
+		Namespace: promNamespace,
+		Name:      "alert_processed_total",
+		Help:      "Number of alerts processed in the /alert webhook.",
+	})
+	promSendTriggeredMetric = prometheus.NewCounter(prometheus.CounterOpts{
+		Namespace: promNamespace,
+		Name:      "send_trigger_total",
+		Help:      "Number of successful calls to the /send webhook.",
+	})
+	promMessagesSentMetric = prometheus.NewCounterVec(prometheus.CounterOpts{
+		Namespace: promNamespace,
+		Name:      "messages_sent_total",
+		Help:      "Number of messages sent.",
+	}, []string{"recipient"})
+	promMessagesReceivedMetric = prometheus.NewCounterVec(prometheus.CounterOpts{
+		Namespace: promNamespace,
+		Name:      "messages_received_total",
+		Help:      "Number of messages received.",
+	}, []string{"recipient"})
+)
+
+func init() {
+	prometheus.MustRegister(promAlertTriggeredMetric)
+	prometheus.MustRegister(promAlertsProcessedMetric)
+	prometheus.MustRegister(promSendTriggeredMetric)
+	prometheus.MustRegister(promMessagesSentMetric)
+	prometheus.MustRegister(promMessagesReceivedMetric)
+}

send_handler.go

@@ -0,0 +1,26 @@
+package main
+
+import (
+	"io/ioutil"
+	"log"
+	"net/http"
+)
+
+type sendHandler struct {
+	sender Sender
+}
+
+// NewSendHandler create an HTTP handler which copies the body into the sender
+func NewSendHandler(sender Sender) http.Handler {
+	return &sendHandler{sender}
+}
+
+func (s *sendHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	defer r.Body.Close()
+	if data, err := ioutil.ReadAll(r.Body); err == nil {
+		promSendTriggeredMetric.Inc()
+		s.sender.Send(string(data))
+	} else {
+		log.Printf("sendHandler: could not read the body: %s\n", err)
+	}
+}

xmpp.go

@@ -0,0 +1,182 @@
+package main
+
+import (
+	"crypto/tls"
+	"fmt"
+	"io"
+	"log"
+	"sort"
+	"strings"
+
+	libxmpp "github.com/mattn/go-xmpp"
+)
+
+const (
+	xmppStatusChat  = "chat"
+	xmppHelpMessage = `Help:
+ - help
+ - quit`
+)
+
+type xmpp struct {
+	client    *libxmpp.Client
+	debugMode bool
+	channel   chan xmppMessage
+	status    string
+	sendNotif []string
+}
+
+type xmppMessage struct {
+	message string
+}
+
+// NewXMPP create an XMPP connection. Use Close() to end it
+func NewXMPP(config *Config) SendCloser {
+	log.Printf("Connect to XMPP account %s\n", config.XMPP.User)
+	options := libxmpp.Options{
+		User:          config.XMPP.User,
+		Password:      config.XMPP.Password,
+		Debug:         config.Debug,
+		NoTLS:         config.XMPP.NoTLS,
+		Status:        xmppStatusChat,
+		StatusMessage: config.XMPP.Status,
+	}
+	if config.XMPP.TLSInsecure {
+		options.TLSConfig = &tls.Config{
+			InsecureSkipVerify: true,
+		}
+	}
+
+	client, err := options.NewClient()
+	if err != nil {
+		log.Fatalf("Could not connect to XMPP server: %s", err)
+	}
+
+	result := &xmpp{
+		client:    client,
+		debugMode: config.Debug,
+		channel:   make(chan xmppMessage),
+		status:    config.XMPP.Status,
+		sendNotif: config.XMPP.SendNotif,
+	}
+	go result.runSender()
+	go result.runReceiver()
+	if config.StartupMessage != "" {
+		result.Send(config.StartupMessage)
+	}
+	return result
+}
+
+func (x *xmpp) Send(message string) error {
+	if message != "" {
+		x.channel <- xmppMessage{
+			message: message,
+		}
+	}
+	return nil
+}
+
+func (x *xmpp) Close() error {
+	close(x.channel)
+	x.client.SendPresence(libxmpp.Presence{
+		Show:   "unavailable",
+		Status: "No monitoring",
+	})
+	return x.client.Close()
+}
+
+func (x *xmpp) runSender() {
+	for payload := range x.channel {
+		for _, sendNotif := range x.sendNotif {
+			if err := x.sendTo(sendNotif, payload.message); err != nil {
+				log.Printf("ERROR %s\n", err)
+			}
+		}
+	}
+}
+
+func (x *xmpp) runReceiver() {
+	for {
+		stanza, err := x.client.Recv()
+		if err != nil {
+			if err == io.EOF {
+				x.Close()
+				return
+			}
+			log.Fatal(err)
+		}
+		x.debug("Stanza: %v\n", stanza)
+		switch v := stanza.(type) {
+		case libxmpp.Chat:
+			x.handleChat(&v)
+		case libxmpp.Presence:
+			x.handlePresence(&v)
+		}
+	}
+}
+
+func (x *xmpp) isKnown(person string) bool {
+	idx := sort.SearchStrings(x.sendNotif, person)
+	return idx < len(x.sendNotif) && x.sendNotif[idx] == person
+}
+
+func (x *xmpp) handleChat(chat *libxmpp.Chat) {
+	if chat.Text != "" {
+		remoteUser := strings.Split(chat.Remote, "/")
+		if len(remoteUser) == 2 && x.isKnown(remoteUser[0]) {
+			x.debug("CHAT type=%s, remote=%s, text=%s\n", chat.Type, chat.Remote, chat.Text)
+			x.handleCommand(remoteUser[0], chat.Text)
+		} else {
+			x.debug("Unknown user: %v\n", chat)
+		}
+	}
+}
+
+func (x *xmpp) handlePresence(presence *libxmpp.Presence) {
+	switch presence.Type {
+	case "":
+	case "unavailable":
+		// something puts us as unavailable
+		if presence.From == x.client.JID() {
+			x.client.SendOrg(fmt.Sprintf("<presence xml:lang='en'><show>%s</show><status>%s</status></presence>", xmppStatusChat, x.status))
+		}
+	case "subscribe":
+		if x.isKnown(presence.From) {
+			x.client.ApproveSubscription(presence.From)
+			x.debug("Approved subscription to %s\n", presence.From)
+		} else {
+			x.client.RevokeSubscription(presence.From)
+			x.debug("Revoked subscription to %s\n", presence.From)
+		}
+	default:
+		x.debug("Unhandled presence: %v\n", presence)
+	}
+}
+
+func (x *xmpp) handleCommand(from, command string) {
+	promMessagesReceivedMetric.WithLabelValues(from).Inc()
+	switch strings.ToLower(command) {
+	case "quit":
+		x.Close()
+	case "help":
+		x.Send(xmppHelpMessage)
+	default:
+		x.sendTo(from, fmt.Sprintf("Unknown command: %s", command))
+	}
+}
+
+func (x *xmpp) sendTo(to, message string) error {
+	promMessagesSentMetric.WithLabelValues(to).Inc()
+	_, err := x.client.Send(libxmpp.Chat{
+		Remote: to,
+		Type:   "chat",
+		Text:   message,
+	})
+	return err
+}
+
+func (x *xmpp) debug(fmt string, v ...interface{}) {
+	if x.debugMode {
+		log.Printf(fmt, v...)
+	}
+}