add telegram integration

Author: reddec

.goxc.json

@@ -3,7 +3,7 @@
 		"default",
 		"publish-github"
 	],
-	"PackageVersion": "0.1.0",
+	"PackageVersion": "0.1.1",
 	"TaskSettings": {
 		"publish-github": {
 			"owner": "reddec",

README.md

@@ -10,6 +10,7 @@ It’s tool for controlling processes like a supervisord but with some important
 * Easy to use - no dependencies. Just a single binary file pre-compilled for most major platforms
 * Easy to hack - monexec can be used as a Golang library with clean and simple architecture
 * Integrated with Consul - optionally, monexec can register all running processes as services and deregister on fail
+* Optional notification to Telegram
 * Supports gracefull and fast shutdown by signals
 * Developed for used inside Docker containers
 * Different strategies for processes

config.go

@@ -20,14 +20,15 @@ import (
 
 type Config struct {
 	Services []monexec.Executable                       `yaml:"services"`
-	Critical []string                           `yaml:"critical,omitempty"`
+	Critical []string                                   `yaml:"critical,omitempty"`
 	Consul struct {
 		URL                       string        `yaml:"url"`
 		TTL                       time.Duration `yaml:"ttl"`
 		AutoDeregistrationTimeout time.Duration `yaml:"timeout"`
 		Dynamic                   []string      `yaml:"register,omitempty"`
 		Permanent                 []string      `yaml:"permanent,omitempty"`
 	}                                           `yaml:"consul"`
+	Telegram *Telegram                          `yaml:"telegram,omitempty"`
 }
 
 func (c *Config) MergeFrom(other *Config) error {
@@ -38,24 +39,32 @@ func (c *Config) MergeFrom(other *Config) error {
 
 	if c.Consul.URL == def.Consul.URL {
 		c.Consul.URL = other.Consul.URL
-	} else if c.Consul.URL != def.Consul.URL && other.Consul.URL != def.Consul.URL {
+	} else if c.Consul.URL != def.Consul.URL && other.Consul.URL != def.Consul.URL && other.Consul.URL != c.Consul.URL {
 		return errors.New("Different CONSUL definition (different URL) - specify same or only once")
 	}
 
 	if c.Consul.TTL == def.Consul.TTL {
 		c.Consul.TTL = other.Consul.TTL
-	} else if c.Consul.TTL != def.Consul.TTL && other.Consul.TTL != def.Consul.TTL {
+	} else if c.Consul.TTL != def.Consul.TTL && other.Consul.TTL != def.Consul.TTL && other.Consul.TTL != c.Consul.TTL {
 		return errors.New("Different CONSUL definition (different TTL) - specify same or only once")
 	}
 
 	if c.Consul.AutoDeregistrationTimeout == def.Consul.AutoDeregistrationTimeout {
 		c.Consul.AutoDeregistrationTimeout = other.Consul.AutoDeregistrationTimeout
-	} else if c.Consul.AutoDeregistrationTimeout != def.Consul.AutoDeregistrationTimeout && other.Consul.AutoDeregistrationTimeout != def.Consul.AutoDeregistrationTimeout {
+	} else if c.Consul.AutoDeregistrationTimeout != def.Consul.AutoDeregistrationTimeout &&
+		other.Consul.AutoDeregistrationTimeout != def.Consul.AutoDeregistrationTimeout &&
+		other.Consul.AutoDeregistrationTimeout != c.Consul.AutoDeregistrationTimeout {
 		return errors.New("Different CONSUL definition (different AutoDeregistrationTimeout) - specify same or only once")
 	}
 
 	c.Consul.Permanent = append(c.Consul.Permanent, other.Consul.Permanent...)
 	c.Consul.Dynamic = append(c.Consul.Dynamic, other.Consul.Dynamic...)
+
+	merged, err := mergeTelegram(c.Telegram, other.Telegram)
+	if err != nil {
+		return err
+	}
+	c.Telegram = merged
 	return nil
 }
 
@@ -87,6 +96,8 @@ func (config *Config) Run(sv container.Supervisor, ctx context.Context) error {
 	critical := plugin.NewCritical(sv, log.New(os.Stderr, "[critical-plugin] ", log.LstdFlags), config.Critical...)
 	sv.Events().AddHandler(critical)
 
+	// Initialize plugins
+	// -- consul
 	consulConfig := api.DefaultConfig()
 	consulConfig.Address = config.Consul.URL
 
@@ -104,8 +115,19 @@ func (config *Config) Run(sv container.Supervisor, ctx context.Context) error {
 	consulLogger := log.New(os.Stderr, "[consul-plugin] ", log.LstdFlags)
 	consulService := plugin.NewConsul(consul, config.Consul.TTL, config.Consul.AutoDeregistrationTimeout, consulLogger, consulRegs)
 	defer consulService.Close()
-
 	sv.Events().AddHandler(consulService)
+
+	// -- telegram
+	if config.Telegram != nil {
+		err := config.Telegram.Prepare()
+		if err != nil {
+			log.Println("telegram plugin ont initialized due to", err)
+		} else {
+			sv.Events().AddHandler(config.Telegram)
+		}
+	}
+
+	// Run
 	wg := sync.WaitGroup{}
 	for _, exec := range config.Services {
 		FillDefaultExecutable(&exec)
@@ -114,7 +136,6 @@ func (config *Config) Run(sv container.Supervisor, ctx context.Context) error {
 			defer wg.Done()
 			container.Wait(sv.Watch(ctx, exec.Factory, exec.Restart, exec.RestartTimeout, false))
 		}(exec)
-
 	}
 
 	wg.Wait()

docs/index.md

@@ -31,7 +31,7 @@ Consul is a service registry and service discover system. MONEXEC can automatica
 
 Auto(de)registration available for `run` or `start` commands.
 
-Use general flag `--consul` (or env var `MONEXEC_CONSUL=true`) for enable Consul integration. Monexec will try register and update status of service in Consul local agent. 
+Use general flag `--consul` (or env var `MONEXEC_CONSUL=true`) for enable Consul integration. Monexec will try register and update status of service in Consul local agent.
 
 Monexec will continue work even if Consul becomes unavailable.
 
@@ -62,6 +62,53 @@ Suppose Consul agent is running in host `registry`
 monexec run --consul --consul-address "http://registry:8500" -l srv1 -- nc -l 9000
 ```
 
+# How to integrate with Telegram
+
+Since `0.1.1` you can receive notifications over Telegram.
+
+You have to know:
+
+* BOT token : can be obtained here http://t.me/botfather
+* Receipients ChatID's : can be obtained here http://t.me/MyTelegramID_bot
+
+Message template (based on Golang templates) also required. We recommend use this:
+
+```
+*{{.label}}*
+Service {{.label}} {{.action}}
+{{if .error}}⚠️ *Error:*  {{.error}}{{end}}
+_time: {{.time}}_
+_host: {{.hostname}}_
+```
+
+Available params:
+
+* `.label` - name of service
+* `.action` - servce action. Can be `spawned` or `stopped`
+* `.time` - current time in UTC format with timezone
+* `.error` - error message available only on `stopped` action
+* `.hostname` - current hostname
+
+Configuration avaiable only from .yaml files:
+
+```yaml
+telegram:
+  # BOT token
+  token: "123456789:AAAAAAAAAAAAAAAAAAAAAA_BBBBBBBBBBBB"
+  services:
+      # services that will be monitored
+      - "listener2"
+  recipients:
+      # List of telegrams chat id
+      - 123456789
+  template: |
+    *{{.label}}*
+    Service {{.label}} {{.action}}
+    {{if .error}}⚠️ *Error:*  {{.error}}{{end}}
+    _time: {{.time}}_
+    _host: {{.hostname}}_
+```
+
 # Usage
 
 `monexec <command> [command-flags...] [args,...]`
@@ -210,4 +257,4 @@ consul:
 critical:
   - consul
 
-```
+```

sample/sample2.yaml

@@ -10,3 +10,16 @@ services:
 consul:
   permanent:
   - listener2
+
+telegram:
+  token: "123456789:AAAAAAAAAAAAAAAAAAAAAA_BBBBBBBBBBBB"
+  services:
+      - "listener2"
+  recipients:
+      - 123456789
+  template: |
+    *{{.label}}*
+    Service {{.label}} {{.action}}
+    {{if .error}}⚠️ *Error:*  {{.error}}{{end}}
+    _time: {{.time}}_
+    _host: {{.hostname}}_

telegram.go

@@ -0,0 +1,120 @@
+package main
+
+import (
+	"github.com/reddec/container"
+	"text/template"
+	"log"
+	"os"
+	"bytes"
+	"gopkg.in/telegram-bot-api.v4"
+	"github.com/pkg/errors"
+	"time"
+)
+
+type Telegram struct {
+	Token      string   `yaml:"token"`
+	Recipients []int64  `yaml:"recipients"`
+	Services   []string `yaml:"services"`
+	Template   string   `yaml:"template"`
+
+	servicesSet map[string]bool     `yaml:"-"`
+	templateBin *template.Template  `yaml:"-"`
+	logger      *log.Logger         `yaml:"-"`
+	bot         *tgbotapi.BotAPI    `yaml:"-"`
+	hostname    string
+}
+
+func (c *Telegram) Prepare() error {
+	c.servicesSet = make(map[string]bool)
+	for _, srv := range c.Services {
+		c.servicesSet[srv] = true
+	}
+	t, err := template.New("").Parse(c.Template)
+	if err != nil {
+		return err
+	}
+	c.templateBin = t
+	c.logger = log.New(os.Stderr, "[telegram] ", log.LstdFlags)
+	bot, err := tgbotapi.NewBotAPI(c.Token)
+	if err != nil {
+		return err
+	}
+	c.bot = bot
+	c.hostname, _ = os.Hostname()
+	return nil
+}
+
+func (c *Telegram) Stopped(runnable container.Runnable, id container.ID, err error) {
+	if c.servicesSet[runnable.Label()] {
+		params := map[string]interface{}{
+			"action":   "stopped",
+			"id":       id,
+			"error":    err,
+			"label":    runnable.Label(),
+			"hostname": c.hostname,
+			"time":     time.Now().String(),
+		}
+		c.renderAndSend(params)
+	}
+}
+
+func (c *Telegram) renderAndSend(params map[string]interface{}) {
+	message := &bytes.Buffer{}
+	renderErr := c.templateBin.Execute(message, params)
+	if renderErr != nil {
+		c.logger.Println("failed render:", renderErr, "; params:", params)
+	} else {
+		msg := tgbotapi.NewMessage(0, message.String())
+		msg.ParseMode = "markdown"
+		for _, r := range c.Recipients {
+			msg.ChatID = r
+			_, err := c.bot.Send(msg)
+			if err != nil {
+				c.logger.Println("failed send message to", r, "due to", err)
+			}
+		}
+	}
+}
+
+func (c *Telegram) Spawned(runnable container.Runnable, id container.ID) {
+	if c.servicesSet[runnable.Label()] {
+		params := map[string]interface{}{
+			"action":   "spawned",
+			"id":       id,
+			"label":    runnable.Label(),
+			"hostname": c.hostname,
+			"time":     time.Now().String(),
+		}
+		c.renderAndSend(params)
+	}
+}
+
+func mergeTelegram(a *Telegram, b *Telegram) (*Telegram, error) {
+	if a == nil {
+		return b, nil
+	}
+	if b == nil {
+		return a, nil
+	}
+	if a.Token == "" {
+		a.Token = b.Token
+	}
+	if b.Token == "" {
+		b.Token = a.Token
+	}
+	if a.Token != b.Token {
+		return nil, errors.New("token are different")
+	}
+	if a.Template == "" {
+		a.Template = b.Template
+	}
+	if b.Template == "" {
+		b.Template = a.Template
+	}
+	if a.Template != b.Template {
+		return nil, errors.New("different templates")
+	}
+	a.Recipients = append(a.Recipients, b.Recipients...)
+	a.Services = append(a.Services, b.Services...)
+	return a, nil
+}