Added MQTT integration

Author: virtualzone

.github/workflows/release.yml

@@ -43,6 +43,13 @@ jobs:
           virtualzone/compose-updater:latest
         cache-from: type=local,src=/tmp/.buildx-cache
         cache-to: type=local,dest=/tmp/.buildx-cache-new
+    - name: Docker Hub Description
+      uses: peter-evans/dockerhub-description@v2
+      with:
+        username: virtualzone
+        password: ${{ secrets.CI_REGISTRY_PASSWORD }}
+        repository: virtualzone/compose-updater
+        readme-filepath: ./README.md
     - name: Create Release
       uses: actions/create-release@v1
       env:

README.md

@@ -9,7 +9,11 @@
 A solution for watching your Docker® containers running via Docker Compose for image updates and automatically restarting the compositions whenever an image is refreshed.
 
 ## Overview
-Compose Updater is an application which continuously monitors your running docker containers. When a base image is changed, the updated version gets pulled (or built via --pull) from the registry and the docker compose composition gets restarted (via down and up -d).
+Compose Updater is an application which continuously monitors your running docker containers. When an image is updated, the updated version gets pulled (or built via --pull) from the registry and the docker compose composition gets restarted (via down and up -d).
+
+Compose Updater is useful for your when you're using image tags which are updated regularly (such as ```image:latest``` or a specific major version like ```image:v3```).
+
+Currently, Compose Updater doesn't help you when your're using image tags that won't change (such as an unchangable SemVer, i.e. ```image:1.2.3```). It won't update your Docker Compose files to use newer image tags.
 
 ## Usage
 ### 1. Prepare your services
@@ -67,8 +71,32 @@ CLEANUP | -cleanup | 0 | Run docker system prune -a -f after each run
 ONCE | -once | 0 | Run once and exit
 PRINT_SETTINGS | -printSettings | 1 | Print settings on start
 UPDATE_LOG | -updateLog | '' | Log file for updates and restarts
-
-# License
+BUILD | -build | 0 | Build the image of a service with "build:" section in YAML file every run
+MQTT_BROKER | -mqttBroker | '' | MQTT Broker address (i.e. tcp://127.0.0.1:1883)
+MQTT_CLIENT_ID | -mqttClientId | composeupdater | MQTT Client ID
+MQTT_TOPIC_PREFIX | -mqttTopicPrefix | composeupdater | MQTT Topic Prefix
+MQTT_USERNAME | -mqttUsername | '' | MQTT Username
+MQTT_PASSWORD | -mqttPassword | '' | MQTT Password
+
+## Connecting an MQTT Broker
+You can connect Compose Updater to an MQTT Broker (such as Eclispe Mosquitto or HiveMQ). This way, the actions of each run (i.e. image pulls, composition restarts) are published to an MQTT topic. You can use these informations to send push notifications using a solution like [mqttwarn](https://github.com/jpmens/mqttwarn).
+
+To connect to an MQTT broker, specify the required connection parameters in the settings (see above).
+
+Compose Updater published the following topics:
+
+Topic | Corresponding event | Example content
+--- | --- | ---
+update | On update run start and done | 'start' or 'done'
+update/composition/start | On start checking for updates for a specific Docker Composition | YAML File Path
+update/composition/restart/dry | On skipping composition restart due to dry-run | YAML File Path
+update/composition/restart/skip | On skipping composition restart due to no updated found | YAML File Path
+update/composition/restart/start | On restarting a composition | ```{"composeFile": "/path/to/docker-compose.yml", "services":[{"name": "service1", "image": "image:tag"}]}```
+update/composition/restart/done | On finished restarting a composition | ```{"composeFile": "/path/to/docker-compose.yml", "services":[{"name": "service1", "image": "image:tag"}]}```
+update/composition/service/built | On service's image built | ```{"composeFile": "/path/to/docker-compose.yml", "services":[{"name": "service1", "image": "image:tag"}]}```
+update/composition/service/pulled | On service's image pulled | ```{"composeFile": "/path/to/docker-compose.yml", "services":[{"name": "service1", "image": "image:tag"}]}```
+
+## License
 GNU General Public License v3.0
 
 Docker® is a trademark of Docker, Inc.

src/go.mod

@@ -2,4 +2,7 @@ module github.com/virtualzone/compose-updater
 
 go 1.16
 
-require gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b
+require (
+	github.com/eclipse/paho.mqtt.golang v1.3.5 // indirect
+	gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b
+)

src/go.sum

@@ -1,3 +1,13 @@
+github.com/eclipse/paho.mqtt.golang v1.3.5 h1:sWtmgNxYM9P2sP+xEItMozsR3w0cqZFlqnNN1bdl41Y=
+github.com/eclipse/paho.mqtt.golang v1.3.5/go.mod h1:eTzb4gxwwyWpqBUHGQZ4ABAV7+Jgm1PklsYT/eo8Hcc=
+github.com/gorilla/websocket v1.4.2 h1:+/TMaTYc4QFitKJxsQ7Yye35DkWvkdLcvGKqM+x0Ufc=
+github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
+golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
+golang.org/x/net v0.0.0-20200425230154-ff2c4b7c35a0 h1:Jcxah/M+oLZ/R4/z5RzfPzGbPXnVDPkEDtf2JnuxN+U=
+golang.org/x/net v0.0.0-20200425230154-ff2c4b7c35a0/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
+golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20200323222414-85ca7c5b95cd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b h1:h8qDotaEPuJATrMmW04NCwg7v22aHH28wwpauUhK9Oo=

src/settings.go

@@ -10,15 +10,20 @@ import (
 
 // Settings holds the program runtime configuration
 type Settings struct {
-	Cleanup       bool
-	Dry           bool
-	Help          bool
-	Interval      int64
-	Once          bool
-	PrintSettings bool
-	UpdateLog     string
+	Cleanup         bool
+	Dry             bool
+	Help            bool
+	Interval        int64
+	Once            bool
+	PrintSettings   bool
+	UpdateLog       string
+	Build           bool
+	MqttBroker      string
+	MqttClientID    string
+	MqttTopicPrefix string
+	MqttUsername    string
+	MqttPassword    string
 	//CompleteStop  bool
-	Build bool
 }
 
 var GlobalSettings *Settings = nil
@@ -34,6 +39,11 @@ func ReadSettings() {
 	s.stringFlagEnv(&s.UpdateLog, "updateLog", "UPDATE_LOG", "", "update log file")
 	//s.boolFlagEnv(&s.CompleteStop, "completeStop", "COMPLETE_STOP", false, "Restart all services in docker-compose.yml (even unmanaged) after a new image is pulled")
 	s.boolFlagEnv(&s.Build, "build", "BUILD", false, "Rebuild images of services with 'build' definition")
+	s.stringFlagEnv(&s.MqttBroker, "mqttBroker", "MQTT_BROKER", "", "MQTT Broker address (i.e. tcp://127.0.0.1:1883)")
+	s.stringFlagEnv(&s.MqttClientID, "mqttClientId", "MQTT_CLIENT_ID", "composeupdater", "MQTT Client ID")
+	s.stringFlagEnv(&s.MqttTopicPrefix, "mqttTopicPrefix", "MQTT_TOPIC_PREFIX", "composeupdater", "MQTT Topic Prefix")
+	s.stringFlagEnv(&s.MqttUsername, "mqttUsername", "MQTT_USERNAME", "", "MQTT Username")
+	s.stringFlagEnv(&s.MqttPassword, "mqttPassword", "MQTT_PASSWORD", "", "MQTT Password")
 	flag.Parse()
 	GlobalSettings = s
 }
@@ -71,5 +81,9 @@ func (settings *Settings) Print() {
 	log.Printf("    printSettings ... %t\n", settings.PrintSettings)
 	//log.Printf("    completeStop .... %t\n", settings.CompleteStop)
 	log.Printf("    build ........... %t\n", settings.Build)
-	log.Printf("    updateLog ....... %s\n", settings.UpdateLog)
+	log.Printf("    mqttBroker ...... %s\n", settings.MqttBroker)
+	log.Printf("    mqttClientId .... %s\n", settings.MqttClientID)
+	log.Printf("    mqttTopicPrefix . %s\n", settings.MqttTopicPrefix)
+	log.Printf("    mqttUsername .... %s\n", settings.MqttUsername)
+	log.Printf("    mqttPassword .... %s\n", "(hidden)")
 }

src/update-event.go

@@ -1,41 +1,76 @@
 package main
 
-import "log"
+import (
+	"encoding/json"
+	"log"
+	"strings"
 
-type UpdateEvent struct{}
+	mqtt "github.com/eclipse/paho.mqtt.golang"
+)
+
+type UpdateEvent struct {
+	MqttClient mqtt.Client
+}
+
+type MqttComposeServiceEvent struct {
+	Name  string `json:"name"`
+	Image string `json:"image"`
+}
+
+type MqttComposeFileEvent struct {
+	ComposeFile string                     `json:"composeFile"`
+	Services    []*MqttComposeServiceEvent `json:"services"`
+}
 
 var EventBus *UpdateEvent = nil
 
 func CreateEventBus() {
-	EventBus = &UpdateEvent{}
+	e := &UpdateEvent{}
+	e.connectMqtt()
+	EventBus = e
 }
 
 func (e *UpdateEvent) OnPerformUpdatesStart() {
 	log.Println("Gathering details about running containers...")
+	e.publishMqtt("update", "start")
 }
 
 func (e *UpdateEvent) OnPerformUpdatesComplete() {
 	log.Println("Done.")
+	e.publishMqttWait("update", "done")
 }
 
 func (e *UpdateEvent) OnProcessComposeFileStart(composeFile *ComposeFile) {
 	log.Printf("Checking for updates of services in %s...\n", composeFile.YamlFilePath)
+	e.publishMqtt("update/composition/start", composeFile.YamlFilePath)
 }
 
 func (e *UpdateEvent) OnSkipRestartComposeFileDryMode(composeFile *ComposeFile) {
 	log.Printf("Dry-Mode enabled, not restarting services in %s\n", composeFile.YamlFilePath)
+	e.publishMqtt("update/composition/restart/dry", composeFile.YamlFilePath)
 }
 
 func (e *UpdateEvent) OnSkipRestartComposeFileNoUpdates(composeFile *ComposeFile) {
 	log.Printf("No need to restart services in %s\n", composeFile.YamlFilePath)
+	e.publishMqtt("update/composition/restart/skip", composeFile.YamlFilePath)
 }
 
 func (e *UpdateEvent) OnRestartComposeFile(composeFile *ComposeFile) {
 	log.Printf("Restarting services in %s...\n", composeFile.YamlFilePath)
+	msg := &MqttComposeFileEvent{
+		ComposeFile: composeFile.YamlFilePath,
+		Services:    e.servicesToStringArray(composeFile),
+	}
+	e.publishMqttJSON("update/composition/restart/start", msg)
 }
 
 func (e *UpdateEvent) OnRestartComposeFileComplete(composeFile *ComposeFile) {
 	log.Printf("Restarted services in %s\n", composeFile.YamlFilePath)
+	msg := &MqttComposeFileEvent{
+		ComposeFile: composeFile.YamlFilePath,
+		Services:    e.servicesToStringArray(composeFile),
+	}
+	e.publishMqttJSON("update/composition/restart/done", msg)
 }
 
 func (e *UpdateEvent) OnProcessServiceStart(service *ComposeService) {
@@ -44,12 +79,89 @@ func (e *UpdateEvent) OnProcessServiceStart(service *ComposeService) {
 
 func (e *UpdateEvent) OnServiceNewImageBuilt(service *ComposeService) {
 	log.Printf("Built new image for service %s\n", service.Name)
+	msg := &MqttComposeFileEvent{
+		ComposeFile: service.ComposeFile.YamlFilePath,
+		Services:    []*MqttComposeServiceEvent{e.servicesToString(service)},
+	}
+	e.publishMqttJSON("update/composition/service/built", msg)
 }
 
 func (e *UpdateEvent) OnServiceNewImagePulled(service *ComposeService) {
 	log.Printf("Pulled new image %s for service %s\n", service.ImageName, service.Name)
+	msg := &MqttComposeFileEvent{
+		ComposeFile: service.ComposeFile.YamlFilePath,
+		Services:    []*MqttComposeServiceEvent{e.servicesToString(service)},
+	}
+	e.publishMqttJSON("update/composition/service/pulled", msg)
 }
 
 func (e *UpdateEvent) OnImagePruneStart() {
 	log.Println("Removing unused images...")
 }
+
+func (e *UpdateEvent) OnMqttConnect(opts *mqtt.ClientOptions) {
+	log.Printf("Connecting to MQTT Broker %s...\n", opts.Servers[0])
+}
+
+func (e *UpdateEvent) connectMqtt() {
+	if GlobalSettings.MqttBroker == "" {
+		return
+	}
+	opts := mqtt.NewClientOptions().AddBroker(GlobalSettings.MqttBroker).SetClientID(GlobalSettings.MqttClientID)
+	if GlobalSettings.MqttUsername != "" {
+		opts.SetUsername(GlobalSettings.MqttUsername)
+	}
+	if GlobalSettings.MqttPassword != "" {
+		opts.SetPassword(GlobalSettings.MqttPassword)
+	}
+	e.OnMqttConnect(opts)
+	client := mqtt.NewClient(opts)
+	if token := client.Connect(); token.Wait() && token.Error() != nil {
+		log.Fatalf("Could not connect to MQTT broker: %s\n", token.Error())
+	}
+	GlobalSettings.MqttTopicPrefix = strings.TrimSuffix(GlobalSettings.MqttTopicPrefix, "/")
+	e.MqttClient = client
+}
+
+func (e *UpdateEvent) publishMqttInternal(topic string, s string, wait bool) {
+	if e.MqttClient == nil {
+		return
+	}
+	token := e.MqttClient.Publish(GlobalSettings.MqttTopicPrefix+"/"+topic, 0, false, s)
+	if wait {
+		token.Wait()
+	}
+}
+
+func (e *UpdateEvent) publishMqtt(topic string, s string) {
+	e.publishMqttInternal(topic, s, false)
+}
+
+func (e *UpdateEvent) publishMqttWait(topic string, s string) {
+	e.publishMqttInternal(topic, s, true)
+}
+
+func (e *UpdateEvent) publishMqttJSON(topic string, v interface{}) error {
+	json, err := json.Marshal(v)
+	if err != nil {
+		return err
+	}
+	e.publishMqttInternal(topic, string(json), false)
+	return nil
+}
+
+func (e *UpdateEvent) servicesToString(service *ComposeService) *MqttComposeServiceEvent {
+	item := &MqttComposeServiceEvent{
+		Name:  service.Name,
+		Image: service.ImageName,
+	}
+	return item
+}
+
+func (e *UpdateEvent) servicesToStringArray(composeFile *ComposeFile) []*MqttComposeServiceEvent {
+	services := make([]*MqttComposeServiceEvent, 0)
+	for _, service := range composeFile.Services {
+		services = append(services, e.servicesToString(service))
+	}
+	return services
+}

src/version.go

@@ -1,4 +1,4 @@
 package main
 
 // BuildVersion is the version
-const BuildVersion = "2.0.2"
+const BuildVersion = "2.1.0"

test/mosquitto.conf

@@ -0,0 +1,6 @@
+per_listener_settings true
+
+listener 1883
+allow_anonymous false
+password_file /mosquitto/config/mqpass
+require_certificate false

test/mqpass

@@ -0,0 +1 @@
+compose-updater:$7$101$/EAt7Hm36wbI6jXp$B+LItdHbG64HCjGg5Bbuz3k/r6HtZNaZozSHTs7RUmw6d/HuPt9k/q2ZVAj606rFphljf0XdWR5AZ9AtDbl7Kw==