Merge pull request #57 from redsift/feature/beacons

add beacon checks metrics

cmd/salt-exporter/config.go

@@ -38,6 +38,7 @@ type Config struct {
 	ListenAddress string `mapstructure:"listen-address"`
 	ListenPort    int    `mapstructure:"listen-port"`
 	IPCFile       string `mapstructure:"ipc-file"`
+	PKIDir        string `mapstructure:"pki-dir"`
 	TLS           struct {
 		Enabled     bool
 		Key         string
@@ -86,6 +87,7 @@ func setDefaults(healthMinions bool) {
 	viper.SetDefault("log-level", defaultLogLevel)
 	viper.SetDefault("listen-port", defaultPort)
 	viper.SetDefault("ipc-file", listener.DefaultIPCFilepath)
+	viper.SetDefault("pki-dir", listener.DefaultPKIDirpath)
 	viper.SetDefault("metrics.health-minions", defaultHealthMinion)
 	viper.SetDefault("metrics.salt_new_job_total.enabled", true)
 	viper.SetDefault("metrics.salt_expected_responses_total.enabled", true)

cmd/salt-exporter/config_test.go

@@ -28,6 +28,7 @@ func TestReadConfigFlagOnly(t *testing.T) {
 				ListenAddress: "127.0.0.1",
 				ListenPort:    8080,
 				IPCFile:       listener.DefaultIPCFilepath,
+				PKIDir:        listener.DefaultPKIDirpath,
 				TLS: struct {
 					Enabled     bool
 					Key         string
@@ -119,6 +120,7 @@ func TestReadConfigFlagOnly(t *testing.T) {
 				ListenAddress: "127.0.0.1",
 				ListenPort:    8080,
 				IPCFile:       "/dev/null",
+				PKIDir:        "/etc/salt/pki",
 				TLS: struct {
 					Enabled     bool
 					Key         string
@@ -228,6 +230,7 @@ func TestConfigFileOnly(t *testing.T) {
 		ListenAddress: "127.0.0.1",
 		ListenPort:    2113,
 		IPCFile:       "/dev/null",
+		PKIDir:        "/tmp/pki",
 		TLS: struct {
 			Enabled     bool
 			Key         string
@@ -338,6 +341,7 @@ func TestConfigFileWithFlags(t *testing.T) {
 		ListenAddress: "127.0.0.1",
 		ListenPort:    8080,
 		IPCFile:       "/somewhere",
+		PKIDir:        "/tmp/pki",
 		TLS: struct {
 			Enabled     bool
 			Key         string

cmd/salt-exporter/config_test.yml

@@ -2,6 +2,7 @@ listen-address: "127.0.0.1"
 listen-port: 2113
 
 ipc-file: /dev/null
+pki-dir: /tmp/pki
 
 log-level: "info"
 tls:

cmd/salt-exporter/main.go

@@ -55,14 +55,23 @@ func start(config Config) {
 
 	log.Info().Msg("listening for events...")
 	eventChan := make(chan event.SaltEvent)
+	watchChan := make(chan event.WatchEvent)
 
 	// listen and expose metric
 	parser := parser.NewEventParser(false)
 	eventListener := listener.NewEventListener(ctx, parser, eventChan)
 	eventListener.SetIPCFilepath(config.IPCFile)
 
+	if config.Metrics.HealthMinions {
+		pkiWatcher, err := listener.NewPKIWatcher(ctx, config.PKIDir, watchChan)
+		if err != nil {
+			log.Fatal().Msgf("unable to watch PKI for minions change: %v", err)
+		}
+
+		go pkiWatcher.StartWatching()
+	}
 	go eventListener.ListenEvents()
-	go metrics.ExposeMetrics(ctx, eventChan, config.Metrics)
+	go metrics.ExposeMetrics(ctx, eventChan, watchChan, config.Metrics)
 
 	// start http server
 	log.Info().Msg("exposing metrics on " + listenSocket + "/metrics")

internal/metrics/metrics.go

@@ -4,6 +4,7 @@ import (
 	"context"
 
 	"github.com/kpetremann/salt-exporter/pkg/event"
+	evt "github.com/kpetremann/salt-exporter/pkg/event"
 	"github.com/rs/zerolog/log"
 )
 
@@ -14,7 +15,15 @@ func boolToFloat64(b bool) float64 {
 	return 0.0
 }
 
-func eventToMetrics(event event.SaltEvent, r Registry) {
+func eventToMetrics(event event.SaltEvent, r *Registry) {
+	if event.Module == evt.BeaconModule {
+		if event.Type != "status" {
+			return
+		}
+		r.UpdateLastHeartbeat(event.Data.Id)
+		return
+	}
+
 	switch event.Type {
 	case "new":
 		state := event.ExtractState()
@@ -48,14 +57,21 @@ func eventToMetrics(event event.SaltEvent, r Registry) {
 	}
 }
 
-func ExposeMetrics(ctx context.Context, eventChan <-chan event.SaltEvent, config Config) {
+func ExposeMetrics(ctx context.Context, eventChan <-chan event.SaltEvent, watchChan <-chan event.WatchEvent, config Config) {
 	registry := NewRegistry(config)
 
 	for {
 		select {
 		case <-ctx.Done():
 			log.Info().Msg("stopping event listener")
 			return
+		case event := <-watchChan:
+			if event.Op == evt.Accepted {
+				registry.AddObservableMinion(event.MinionName)
+			}
+			if event.Op == evt.Removed {
+				registry.DeleteObservableMinion(event.MinionName)
+			}
 		case event := <-eventChan:
 			if config.Global.Filters.IgnoreTest && event.IsTest {
 				return
@@ -64,7 +80,7 @@ func ExposeMetrics(ctx context.Context, eventChan <-chan event.SaltEvent, config
 				return
 			}
 
-			eventToMetrics(event, registry)
+			eventToMetrics(event, &registry)
 		}
 	}
 }

internal/metrics/registry.go

@@ -2,6 +2,7 @@ package metrics
 
 import (
 	"strconv"
+	"time"
 
 	"github.com/kpetremann/salt-exporter/internal/filters"
 	"github.com/prometheus/client_golang/prometheus"
@@ -11,6 +12,8 @@ import (
 type Registry struct {
 	config Config
 
+	observedMinions int32
+
 	newJobTotal            *prometheus.CounterVec
 	expectedResponsesTotal *prometheus.CounterVec
 
@@ -19,6 +22,9 @@ type Registry struct {
 
 	responseTotal  *prometheus.CounterVec
 	functionStatus *prometheus.GaugeVec
+
+	statusLastResponse *prometheus.GaugeVec
+	minionsTotal       *prometheus.GaugeVec
 }
 
 func NewRegistry(config Config) Registry {
@@ -35,6 +41,7 @@ func NewRegistry(config Config) Registry {
 	return Registry{
 		config: config,
 
+		observedMinions: 0,
 		newJobTotal: promauto.NewCounterVec(
 			prometheus.CounterOpts{
 				Name: "salt_new_job_total",
@@ -82,9 +89,39 @@ func NewRegistry(config Config) Registry {
 			},
 			[]string{"minion", "function", "state"},
 		),
+		statusLastResponse: promauto.NewGaugeVec(
+			prometheus.GaugeOpts{
+				Name: "salt_health_last_heartbeat",
+				Help: "Last status beacon received. Unix timestamp",
+			},
+			[]string{"minion"},
+		),
+		minionsTotal: promauto.NewGaugeVec(
+			prometheus.GaugeOpts{
+				Name: "salt_health_minions_total",
+				Help: "Total number of observed minions via status beacon",
+			}, []string{},
+		),
 	}
 }
 
+func (r *Registry) UpdateLastHeartbeat(minion string) {
+	timestamp := time.Now().Unix()
+	r.statusLastResponse.WithLabelValues(minion).Set(float64(timestamp))
+}
+
+func (r *Registry) AddObservableMinion(minion string) {
+	r.observedMinions += 1
+	r.UpdateLastHeartbeat(minion)
+	r.minionsTotal.WithLabelValues().Set(float64(r.observedMinions))
+}
+
+func (r *Registry) DeleteObservableMinion(minion string) {
+	r.statusLastResponse.DeleteLabelValues(minion)
+	r.observedMinions -= 1
+	r.minionsTotal.WithLabelValues().Set(float64(r.observedMinions))
+}
+
 func (r *Registry) IncreaseNewJobTotal(function, state string) {
 	if r.config.SaltNewJobTotal.Enabled {
 		r.newJobTotal.WithLabelValues(function, state).Inc()

pkg/event/event.go

@@ -3,11 +3,33 @@ package event
 import (
 	"encoding/json"
 	"errors"
+	"strings"
 
 	"github.com/vmihailenco/msgpack/v5"
 	"gopkg.in/yaml.v3"
 )
 
+type EventModule int
+
+type WatchOp uint32
+
+const (
+	UnknownModule EventModule = iota
+	RunnerModule
+	JobModule
+	BeaconModule
+)
+
+const (
+	Accepted WatchOp = iota
+	Removed
+)
+
+type WatchEvent struct {
+	MinionName string
+	Op         WatchOp
+}
+
 type EventData struct {
 	Arg       []interface{} `msgpack:"arg"`
 	Cmd       string        `msgpack:"cmd"`
@@ -32,6 +54,7 @@ type EventData struct {
 type SaltEvent struct {
 	Tag                string
 	Type               string
+	Module             EventModule
 	TargetNumber       int
 	Data               EventData
 	IsScheduleJob      bool
@@ -74,6 +97,23 @@ func (e SaltEvent) RawToYAML() ([]byte, error) {
 	return yaml.Marshal(data)
 }
 
+func GetEventModule(tag string) EventModule {
+	tagParts := strings.Split(tag, "/")
+	if len(tagParts) < 2 {
+		return UnknownModule
+	}
+	switch tagParts[1] {
+	case "run":
+		return RunnerModule
+	case "job":
+		return JobModule
+	case "beacon":
+		return BeaconModule
+	default:
+		return UnknownModule
+	}
+}
+
 // extractStateFromArgs extracts embedded state info
 func extractStateFromArgs(args interface{}, key string) string {
 	// args only

pkg/event/event_test.go

@@ -10,6 +10,7 @@ func getNewStateEvent() event.SaltEvent {
 	return event.SaltEvent{
 		Tag:          "salt/job/20220630000f000000000/new",
 		Type:         "new",
+		Module:       event.JobModule,
 		TargetNumber: 1,
 		Data: event.EventData{
 			Timestamp: "2022-06-30T00:00:00.000000",