Add Amqp notification client (#63)

doc/configuration.md

@@ -22,6 +22,14 @@ watch:
   first_check_notif: false
 
 notif:
+  amqp:
+    enable: false
+    host: localhost
+    port: 5672
+    username: guest
+    password: guest
+    exchange: 
+    queue: queue
   gotify:
     enable: false
     endpoint: http://gotify.foo.com
@@ -107,6 +115,17 @@ providers:
 
 ### notif
 
+* `amqp`
+  * `enable`: Enable AMQP notifications (default: `false`).
+  * `host`: AMQP server host (default: `localhost`). **required**
+  * `port`: AMQP server port (default: `5672`). **required**
+  * `username`: AMQP username. **required**
+  * `username_file`: Use content of secret file as AMQP username if `username` not defined.
+  * `password`: AMQP password. **required**
+  * `password_file`: Use content of secret file as AMQP password if `password` not defined.
+  * `exchange`: Name of the exchange the message will be sent to. (default: `empty`)
+  * `queue`: Name of the queue the message will be sent to. **required**
+
 * `gotify`
   * `enable`: Enable gotify notification (default: `false`).
   * `endpoint`: Gotify base URL (e.g. `http://gotify.foo.com`). **required**

doc/notifications.md

@@ -1,5 +1,5 @@
 # Notifications
-
+* [Amqp](#amqp)
 * [Gotify](#gotify)
 * [Mail](#mail)
 * [Rocket.Chat](#rocketchat)
@@ -7,6 +7,23 @@
 * [Telegram](#telegram)
 * [Webhook](#webhook)
 
+## Amqp
+
+You can send notifications to any amqp compatible server, the body will be a JSON format that looks like:
+
+```json
+{
+  "diun_version": "0.3.0",
+  "status": "new",
+  "provider": "static-0",
+  "image": "docker.io/crazymax/swarm-cronjob:0.2.1",
+  "mime_type": "application/vnd.docker.distribution.manifest.v2+json",
+  "digest": "sha256:5913d4b5e8dc15430c2f47f40e43ab2ca7f2b8df5eee5db4d5c42311e08dfb79",
+  "created": "2019-01-24T10:26:49.152006005Z",
+  "platform": "linux/amd64",
+}
+```
+
 ## Gotify
 
 Notifications can be sent using a [Gotify](https://gotify.net/) instance. [Follow the instructions](https://gotify.net/docs/install) to set up a Gotify server.

go.mod

@@ -19,6 +19,7 @@ require (
 	github.com/robfig/cron/v3 v3.0.1
 	github.com/rs/zerolog v1.18.0
 	github.com/sirupsen/logrus v1.6.0
+  github.com/streadway/amqp v0.0.0-20200108173154-1c71cc93ed71
 	github.com/stretchr/testify v1.5.1
 	github.com/technoweenie/multipartstreamer v1.0.1 // indirect
 	go.etcd.io/bbolt v1.3.4

go.sum

@@ -210,6 +210,8 @@ github.com/sirupsen/logrus v1.6.0 h1:UBcNElsrwanuuMsnGSlYmtmgbb23qDR5dG+6X6Oo89I
 github.com/sirupsen/logrus v1.6.0/go.mod h1:7uNnSEd1DgxDLC74fIahvMZmmYsHGZGEOFrfsX/uA88=
 github.com/ssor/bom v0.0.0-20170718123548-6386211fdfcf h1:pvbZ0lM0XWPBqUKqFU8cmavspvIl9nulOYwdy6IFRRo=
 github.com/ssor/bom v0.0.0-20170718123548-6386211fdfcf/go.mod h1:RJID2RhlZKId02nZ62WenDCkgHFerpIOmW0iT7GKmXM=
+github.com/streadway/amqp v0.0.0-20200108173154-1c71cc93ed71 h1:2MR0pKUzlP3SGgj5NYJe/zRYDwOu9ku6YHy+Iw7l5DM=
+github.com/streadway/amqp v0.0.0-20200108173154-1c71cc93ed71/go.mod h1:AZpEONHx3DKn8O/DFsRAY58/XVQiIPMTMB1SddzLXVw=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=

internal/config/config.go

@@ -48,6 +48,12 @@ func Load(cli model.Cli, version string) (*Config, error) {
 			FirstCheckNotif: false,
 		},
 		Notif: model.Notif{
+			Amqp: model.NotifAmqp{
+				Enable:   false,
+				Host:     "localhost",
+				Port:     5672,
+				Exchange: "",
+			},
 			Gotify: model.NotifGotify{
 				Enable:  false,
 				Timeout: 10,

internal/config/config.test.yml

@@ -7,6 +7,13 @@ watch:
   first_check_notif: false
 
 notif:
+  amqp:
+    enable: false
+    host: localhost
+    port: 5672
+    username: guest
+    password: guest
+    queue: queue
   gotify:
     enable: false
     endpoint: http://gotify.foo.com

internal/config/config_test.go

@@ -52,6 +52,14 @@ func TestLoad(t *testing.T) {
 					Schedule: "*/30 * * * *",
 				},
 				Notif: model.Notif{
+					Amqp: model.NotifAmqp{
+						Enable:   false,
+						Host:     "localhost",
+						Port:     5672,
+						Username: "guest",
+						Password: "guest",
+						Queue:    "queue",
+					},
 					Gotify: model.NotifGotify{
 						Enable:   false,
 						Endpoint: "http://gotify.foo.com",

internal/model/notif.go

@@ -14,6 +14,7 @@ type NotifEntry struct {
 
 // Notif holds data necessary for notification configuration
 type Notif struct {
+	Amqp       NotifAmqp       `yaml:"amqp,omitempty"`
 	Gotify     NotifGotify     `yaml:"gotify,omitempty"`
 	Mail       NotifMail       `yaml:"mail,omitempty"`
 	RocketChat NotifRocketChat `yaml:"rocketchat,omitempty"`
@@ -22,6 +23,19 @@ type Notif struct {
 	Webhook    NotifWebhook    `yaml:"webhook,omitempty"`
 }
 
+// NotifAmqp holds amqp notification configuration details
+type NotifAmqp struct {
+	Enable       bool   `yaml:"enable,omitempty"`
+	Username     string `yaml:"username,omitempty"`
+	UsernameFile string `yaml:"username_file,omitempty"`
+	Password     string `yaml:"password,omitempty"`
+	PasswordFile string `yaml:"password_file,omitempty"`
+	Host         string `yaml:"host,omitempty"`
+	Port         int    `yaml:"port,omitempty"`
+	Queue        string `yaml:"queue,omitempty"`
+	Exchange     string `yaml:"exchange,omitempty"`
+}
+
 // NotifGotify holds gotify notification configuration details
 type NotifGotify struct {
 	Enable   bool   `yaml:"enable,omitempty"`

internal/notif/amqp/client.go

@@ -0,0 +1,115 @@
+package amqp
+
+import (
+	"encoding/json"
+	"fmt"
+	"time"
+
+	"github.com/crazy-max/diun/internal/model"
+	"github.com/crazy-max/diun/internal/notif/notifier"
+	"github.com/crazy-max/diun/pkg/utl"
+	"github.com/opencontainers/go-digest"
+	"github.com/streadway/amqp"
+)
+
+// Client represents an active amqp notification object
+type Client struct {
+	*notifier.Notifier
+	cfg model.NotifAmqp
+	app model.App
+}
+
+// New creates a new amqp notification instance
+func New(config model.NotifAmqp, app model.App) notifier.Notifier {
+	return notifier.Notifier{
+		Handler: &Client{
+			cfg: config,
+			app: app,
+		},
+	}
+}
+
+// Name returns notifier's name
+func (c *Client) Name() string {
+	return "amqp"
+}
+
+// Send creates and sends a amqp notification with an entry
+func (c *Client) Send(entry model.NotifEntry) error {
+
+	username, err := utl.GetSecret(c.cfg.Username, c.cfg.UsernameFile)
+
+	if err != nil {
+		return err
+	}
+
+	password, err := utl.GetSecret(c.cfg.Password, c.cfg.PasswordFile)
+	if err != nil {
+		return err
+	}
+
+	connString := fmt.Sprintf("amqp://%s:%s@%s:%d/", username, password, c.cfg.Host, c.cfg.Port)
+
+	conn, err := amqp.Dial(connString)
+	if err != nil {
+		return err
+	}
+
+	defer conn.Close()
+
+	ch, err := conn.Channel()
+	if err != nil {
+		return err
+	}
+
+	defer ch.Close()
+
+	q, err := ch.QueueDeclare(
+		c.cfg.Queue, // name
+		false,       // durable
+		false,       // delete when unused
+		false,       // exclusive
+		false,       // no-wait
+		nil,         // arguments
+	)
+	if err != nil {
+		return err
+	}
+
+	body, err := buildBody(entry, c.app)
+	if err != nil {
+		return err
+	}
+
+	return ch.Publish(
+		c.cfg.Exchange, // exchange
+		q.Name,         // routing key
+		false,          // mandatory
+		false,          // immediate
+		amqp.Publishing{
+			ContentType: "application/json",
+			Body:        body,
+		})
+}
+
+func buildBody(entry model.NotifEntry, app model.App) ([]byte, error) {
+	return json.Marshal(struct {
+		Version  string        `json:"diun_version"`
+		Status   string        `json:"status"`
+		Provider string        `json:"provider"`
+		Image    string        `json:"image"`
+		MIMEType string        `json:"mime_type"`
+		Digest   digest.Digest `json:"digest"`
+		Created  *time.Time    `json:"created"`
+		Platform string        `json:"platform"`
+	}{
+		Version:  app.Version,
+		Status:   string(entry.Status),
+		Provider: entry.Provider,
+		Image:    entry.Image.String(),
+		MIMEType: entry.Manifest.MIMEType,
+		Digest:   entry.Manifest.Digest,
+		Created:  entry.Manifest.Created,
+		Platform: entry.Manifest.Platform,
+	})
+}

internal/notif/client.go

@@ -2,6 +2,7 @@ package notif
 
 import (
 	"github.com/crazy-max/diun/internal/model"
+	"github.com/crazy-max/diun/internal/notif/amqp"
 	"github.com/crazy-max/diun/internal/notif/gotify"
 	"github.com/crazy-max/diun/internal/notif/mail"
 	"github.com/crazy-max/diun/internal/notif/notifier"
@@ -28,6 +29,9 @@ func New(config model.Notif, app model.App, userAgent string) (*Client, error) {
 	}
 
 	// Add notifiers
+	if config.Amqp.Enable {
+		c.notifiers = append(c.notifiers, amqp.New(config.Amqp, app))
+	}
 	if config.Gotify.Enable {
 		c.notifiers = append(c.notifiers, gotify.New(config.Gotify, app, userAgent))
 	}