initial version of monitoring

Author: numkem

cmd/server/cmd.go

@@ -1,6 +1,11 @@
 package main
 
-import "github.com/spf13/cobra"
+import (
+	"os"
+
+	log "github.com/sirupsen/logrus"
+	"github.com/spf13/cobra"
+)
 
 var (
 	rootCmd = &cobra.Command{
@@ -10,6 +15,11 @@ var (
 )
 
 func init() {
+	if os.Getenv("DEBUG") != "" {
+		log.Info("debug level set")
+		log.SetLevel(log.DebugLevel)
+	}
+
 	rootCmd.PersistentFlags().StringP("config", "c", "traffikey.json", "json configuration filename")
 }
 

cmd/server/cmd_monitor.go

@@ -1,7 +1,11 @@
 package main
 
 import (
+	"context"
 	"fmt"
+	"os"
+	"os/signal"
+	"time"
 
 	"github.com/labstack/echo/v4"
 	logrusmiddleware "github.com/numkem/echo-logrusmiddleware"
@@ -12,7 +16,7 @@ import (
 var monitorCmd = &cobra.Command{
 	Use:   "monitor",
 	Short: "monitor configured endpoints",
-	Long:  "monitor configured endpoints. If one become unreachable, it will redirect to itself and show an error message explaining the target is not available.",
+	Long:  "monitor configured endpoints. If one become unreachable, it will redirect to itself and show an error message",
 	Run:   monitorCmdRun,
 }
 
@@ -24,12 +28,36 @@ func init() {
 }
 
 func monitorCmdRun(cmd *cobra.Command, args []string) {
+	configFilename := cmd.Flag("config").Value.String()
+	mon, err := NewMonitor(configFilename)
+	if err != nil {
+		log.Fatalf("failed to read configuration: %v", err)
+	}
+
+	h := &handler{monitor: mon}
+
+	// echo init
 	e := echo.New()
+	e.HideBanner = true
 	e.Logger = logrusmiddleware.Logger{Logger: log.StandardLogger()}
 
-	e.GET("/", func(c echo.Context) error {
-		return nil
-	})
+	e.GET("/", h.List)
+
+	go func() {
+		mon.Start()
+		e.Logger.Fatal(e.Start(fmt.Sprintf("%s", cmd.Flag("bind").Value)))
+	}()
+
+	// Listen for signal and quit
+	sig := make(chan os.Signal, 1)
+	signal.Notify(sig, os.Interrupt)
+
+	select {
+	case <-sig:
+		ctx, cancel := context.WithTimeout(cmd.Context(), 2*time.Second)
 
-	e.Logger.Fatal(e.Start(fmt.Sprintf("%s", cmd.Flag("bind").Value)))
+		mon.Stop()
+		e.Shutdown(ctx)
+		cancel()
+	}
 }

cmd/server/handler.go

@@ -0,0 +1,13 @@
+package main
+
+import (
+	"github.com/labstack/echo/v4"
+)
+
+type handler struct {
+	monitor *Monitor
+}
+
+func (h *handler) List(c echo.Context) error {
+	return c.JSON(200, nil)
+}

cmd/server/monitor.go

@@ -0,0 +1,142 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"sync"
+	"time"
+
+	"github.com/gofrs/uuid"
+	traffikey "github.com/numkem/traffikey"
+	"github.com/numkem/traffikey/keymate"
+
+	log "github.com/sirupsen/logrus"
+)
+
+type Monitor struct {
+	currentTargets *sync.Map
+	manager        keymate.KeymateConnector
+	cfg            *traffikey.Config
+}
+
+func NewMonitor(configFilename string) (*Monitor, error) {
+	// Read configuration file
+	cfg, err := traffikey.NewConfig(configFilename)
+	if err != nil {
+		return nil, fmt.Errorf("Failed to read configuraiton: %v", err)
+	}
+
+	// Check if defaults are set
+	if cfg.Traefik.DefaultPrefix == "" {
+		log.Warn("default traefik prefix isn't set, this could cause errors")
+	}
+
+	if cfg.Traefik.DefaultEntrypoint == "" {
+		log.Warn("default Traefik entrypoint isn't set, this could cause errors")
+	}
+
+	// Create manager connection
+	mgr, err := keymate.NewEtcdManager(cfg)
+	if err != nil {
+		log.Fatalf("failed to create manager: %v", err)
+	}
+
+	return &Monitor{cfg: cfg, currentTargets: &sync.Map{}, manager: mgr}, nil
+}
+
+func testTarget(tgt *traffikey.Target) []string {
+	// Attempt a test connection to the target
+	switch tgt.Type {
+	case "http":
+		return testHTTPTarget(tgt)
+	default:
+		log.WithField("target", tgt.Name).Warnf("invalid target type %s", tgt.Type)
+	}
+
+	return []string{}
+}
+
+func testHTTPTarget(tgt *traffikey.Target) []string {
+	// Try to connect to the http endpoint with a very small timeout
+	aliveUrls := []string{}
+	for _, url := range tgt.ServerURLs {
+		client := http.Client{
+			Timeout: 1 * time.Second,
+		}
+
+		_, err := client.Get(url)
+		// TODO: Check the HTTP response code
+		if err == nil {
+			aliveUrls = append(aliveUrls, url)
+		}
+	}
+
+	return aliveUrls
+}
+
+type monitoredTarget struct {
+	ID      string
+	Context context.Context
+	Cancel  context.CancelFunc
+	Target  *traffikey.Target
+}
+
+func (m *Monitor) Start() {
+	// Start a monitor goroutine for each target
+	for _, tgt := range m.cfg.Targets {
+		if !tgt.Monitored {
+			log.WithField("target", tgt.Name).Infof("target %s isn't monitored", tgt.Name)
+			continue
+		}
+
+		id, err := uuid.NewV4()
+		if err != nil {
+			log.WithField("target", tgt.Name).Errorf("failed to generate UUID: %v", err)
+		}
+
+		ctx, cancel := context.WithCancel(context.Background())
+		mt := &monitoredTarget{
+			ID:      id.String(),
+			Context: ctx,
+			Cancel:  cancel,
+			Target:  tgt,
+		}
+		m.currentTargets.Store(id, mt)
+
+		log.WithField("target", tgt.Name).Debug("starting monitoring of target")
+		go watchTarget(mt)
+	}
+}
+
+func (m *Monitor) Stop() {
+	m.currentTargets.Range(func(key, value interface{}) bool {
+		mt := value.(*monitoredTarget)
+
+		log.Debugf("Stopping monitoring of target ID %s :: %s", mt.ID, mt.Target.Name)
+		mt.Cancel()
+		return true
+	})
+}
+
+func watchTarget(m *monitoredTarget) {
+	for {
+		ticker := time.NewTicker(15 * time.Second)
+
+		aliveUrls := testTarget(m.Target)
+		switch len(aliveUrls) {
+		case 0:
+			log.WithField("target", m.Target.Name).Infof("Target is DOWN (0/%d)", len(m.Target.ServerURLs))
+		default:
+			upNb := len(m.Target.ServerURLs) - len(aliveUrls)
+			log.WithField("target", m.Target.Name).Infof("Target is UP (%d/%d)", upNb, len(m.Target.ServerURLs))
+		}
+
+		select {
+		case <-m.Context.Done():
+			return
+
+		case <-ticker.C:
+		}
+	}
+}

cmd/server/monitor_test.go

@@ -0,0 +1,32 @@
+package main
+
+import (
+	"testing"
+
+	"github.com/numkem/traffikey"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestHTTPTarget(t *testing.T) {
+	aliveUrls := testHTTPTarget(&traffikey.Target{
+		Name: "google.com",
+		Type: "http",
+		ServerURLs: []string{
+			"https://google.com",
+		},
+	})
+
+	assert.NotEmpty(t, aliveUrls, "testHTTPTarget should return one url")
+}
+
+func TestHTTPTargetFailed(t *testing.T) {
+	aliveUrls := testHTTPTarget(&traffikey.Target{
+		Name: "failed.com",
+		Type: "http",
+		ServerURLs: []string{
+			"http://falksjdfalskjfaldkjf.com",
+		},
+	})
+
+	assert.Empty(t, aliveUrls, "testHTTPTarget should not return any url")
+}

config.go

@@ -4,6 +4,8 @@ import (
 	"encoding/json"
 	"fmt"
 	"os"
+
+	log "github.com/sirupsen/logrus"
 )
 
 type Config struct {
@@ -23,6 +25,8 @@ type traefikConfig struct {
 }
 
 func NewConfig(filename string) (*Config, error) {
+	log.WithField("filename", filename).Debugf("reading configuration file")
+
 	f, err := os.Open(filename)
 	if err != nil {
 		if os.IsNotExist(err) {

dev/example.json

@@ -1,21 +1,29 @@
 {
   "etcd": {
-    "endpoints": [
-      "http://172.31.99.4:2379",
-      "http://172.31.99.5:2379",
-      "http://172.31.99.3:2769"
-    ]
+    "endpoints": ["http://127.0.0.1:2379"]
   },
   "targets": [
     {
-      "name": "test",
-      "urls": ["http://192.168.0.104:8080"],
+      "name": "forge",
+      "urls": ["http://192.168.0.7:8080"],
       "entrypoint": "web",
       "prefix": "traefik",
-      "rule": "test.numkem.org"
+      "rule": "test.numkem.org",
+      "type": "http",
+      "monitored": true
+    },
+    {
+      "name": "pylon",
+      "urls": ["http://192.168.0.2:8080"],
+      "entrypoint": "web",
+      "prefix": "traefik",
+      "rule": "test.numkem.org",
+      "type": "http",
+      "monitored": true
     }
   ],
   "traefik": {
-    "default_entrypoint": "web"
+    "default_entrypoint": "web",
+    "default_prefix": "traefik"
   }
 }