Merge pull request #1 from fgouteroux/feat/windows_service

feat: support windows service registration and event log

.golangci.yml

@@ -20,4 +20,6 @@ issues:
 
 linters-settings:
   errcheck:
-    exclude: scripts/errcheck_excludes.txt
+    exclude-functions:
+    - (net/http.ResponseWriter).Write
+    - (github.com/go-kit/log.Logger).Log

.promu.yml

@@ -5,6 +5,7 @@ repository:
 build:
     binaries:
         - name: promk
+          path: ./cmd/
     ldflags: |
         -X github.com/prometheus/common/version.Version={{.Version}}
         -X github.com/prometheus/common/version.Revision={{.Revision}}

CHANGELOG.md

@@ -1,3 +1,10 @@
+## 0.0.2 / 2024-07-22
+
+* [FEATURE] go 1.22
+* [FEATURE] support windows event log
+* [FEATURE] support windows service registration
+
+
 ## 0.0.1 / 2023-12-19
 
 * [FEATURE] first version

cmd/promk_unix.go

@@ -0,0 +1,13 @@
+//go:build !windows
+// +build !windows
+
+package main
+
+import (
+	"github.com/fgouteroux/promk/pkg/pusher"
+)
+
+func main() {
+	p := pusher.Setup()
+	p.Run()
+}

cmd/promk_windows.go

@@ -0,0 +1,43 @@
+//go:build windows
+// +build windows
+
+package main
+
+import (
+	"log"
+
+	"github.com/fgouteroux/promk/pkg/pusher"
+	win "github.com/fgouteroux/promk/pkg/windows"
+	"golang.org/x/sys/windows/svc"
+)
+
+func main() {
+	p := pusher.Setup()
+
+	isInteractive, err := svc.IsAnInteractiveSession()
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	stopCh := make(chan bool)
+	if !isInteractive {
+		go func() {
+			err = svc.Run("Puppet Agent Exporter", win.NewWindowsExporterService(stopCh))
+			if err != nil {
+				log.Fatalf("Failed to start service: %v", err)
+			}
+		}()
+	}
+
+	go func() {
+		p.Run()
+	}()
+
+	for {
+		if <-stopCh {
+			log.Printf("Shutting down %s", "Puppet Agent Exporter")
+			break
+		}
+	}
+
+}

go.mod

@@ -10,6 +10,7 @@ require (
 	github.com/prometheus/client_model v0.6.1
 	github.com/prometheus/common v0.55.0
 	github.com/prometheus/prometheus v0.52.1
+	golang.org/x/sys v0.21.0
 )
 
 require (
@@ -56,7 +57,6 @@ require (
 	golang.org/x/crypto v0.24.0 // indirect
 	golang.org/x/net v0.26.0 // indirect
 	golang.org/x/oauth2 v0.21.0 // indirect
-	golang.org/x/sys v0.21.0 // indirect
 	golang.org/x/text v0.16.0 // indirect
 	golang.org/x/time v0.5.0 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20240415180920-8c6c420018be // indirect

pkg/log/log.go

@@ -0,0 +1,13 @@
+//go:build !windows
+// +build !windows
+
+package log
+
+import (
+	"github.com/go-kit/log"
+	"github.com/prometheus/common/promlog"
+)
+
+func InitLogger(cfg *promlog.Config) (log.Logger, error) {
+	return promlog.New(cfg), nil
+}

pkg/log/log_windows.go

@@ -0,0 +1,136 @@
+//go:build windows
+// +build windows
+
+package log
+
+import (
+	"runtime"
+	"strings"
+
+	"github.com/prometheus/common/promlog"
+
+	"github.com/go-kit/log/level"
+
+	"github.com/go-kit/log"
+	"golang.org/x/sys/windows/svc"
+	el "golang.org/x/sys/windows/svc/eventlog"
+)
+
+const ServiceName = "Promk"
+
+var levelMap = map[string]level.Option{
+	"error": level.AllowError(),
+	"warn":  level.AllowWarn(),
+	"info":  level.AllowInfo(),
+	"debug": level.AllowDebug(),
+}
+
+// IsWindowsService returns whether the current process is running as a Windows
+// Service. On non-Windows platforms, this always returns false.
+func IsWindowsService() bool {
+	isService, err := svc.IsWindowsService()
+	if err != nil {
+		return false
+	}
+	return isService
+}
+
+// InitLogger returns Windows Event Logger if running as a service under windows
+func InitLogger(cfg *promlog.Config) (log.Logger, error) {
+	if IsWindowsService() {
+		return NewWindowsEventLogger(cfg)
+	} else {
+		return promlog.New(cfg), nil
+	}
+}
+
+func NewWindowsEventLogger(cfg *promlog.Config) (log.Logger, error) {
+	// Setup the log in windows events
+	err := el.InstallAsEventCreate(ServiceName, el.Error|el.Info|el.Warning)
+
+	// Agent should expect an error of 'already exists' if the Event Log sink has already previously been installed
+	if err != nil && !strings.Contains(err.Error(), "already exists") {
+		return nil, err
+	}
+	il, err := el.Open(ServiceName)
+	if err != nil {
+		return nil, err
+	}
+
+	// Ensure the logger gets closed when the GC runs. It's valid to have more than one win logger open concurrently.
+	runtime.SetFinalizer(il, func(l *el.Log) {
+		l.Close()
+	})
+
+	// These are setup to be writers for each Windows log level
+	// Setup this way so we can utilize all the benefits of logformatter
+	infoLogger := newWinLogWrapper(cfg.Format.String(), func(p []byte) error {
+		return il.Info(1, string(p))
+	})
+	warningLogger := newWinLogWrapper(cfg.Format.String(), func(p []byte) error {
+		return il.Warning(1, string(p))
+	})
+
+	errorLogger := newWinLogWrapper(cfg.Format.String(), func(p []byte) error {
+		return il.Error(1, string(p))
+	})
+
+	wl := &winLogger{
+		errorLogger:   errorLogger,
+		infoLogger:    infoLogger,
+		warningLogger: warningLogger,
+	}
+	return level.NewFilter(wl, levelMap[cfg.Level.String()]), nil
+}
+
+// Looks through the key value pairs in the log for level and extract the value
+func getLevel(keyvals ...interface{}) level.Value {
+	for i := 0; i < len(keyvals); i++ {
+		if vo, ok := keyvals[i].(level.Value); ok {
+			return vo
+		}
+	}
+	return nil
+}
+
+func newWinLogWrapper(format string, write func(p []byte) error) log.Logger {
+	infoWriter := &winLogWriter{writer: write}
+	infoLogger := log.NewLogfmtLogger(infoWriter)
+	if format == "json" {
+		infoLogger = log.NewJSONLogger(infoWriter)
+	}
+	return infoLogger
+}
+
+type winLogger struct {
+	errorLogger   log.Logger
+	infoLogger    log.Logger
+	warningLogger log.Logger
+}
+
+func (w *winLogger) Log(keyvals ...interface{}) error {
+	lvl := getLevel(keyvals...)
+	// 3 different loggers are used so that agent can utilize the formatting features of go-kit logging
+	// if agent did not use this then the windows logger uses different function calls for different levels
+	// this is paired with the fact that the io.Writer interface only gives a byte array.
+	switch lvl {
+	case level.DebugValue():
+		return w.infoLogger.Log(keyvals...)
+	case level.InfoValue():
+		return w.infoLogger.Log(keyvals...)
+	case level.WarnValue():
+		return w.warningLogger.Log(keyvals...)
+	case level.ErrorValue():
+		return w.errorLogger.Log(keyvals...)
+	default:
+		return w.infoLogger.Log(keyvals...)
+	}
+}
+
+type winLogWriter struct {
+	writer func(p []byte) error
+}
+
+func (i *winLogWriter) Write(p []byte) (n int, err error) {
+	return len(p), i.writer(p)
+}

client.go → pkg/pusher/client.go

@@ -1,4 +1,4 @@
-package main
+package pusher
 
 import (
 	"crypto/tls"

collect_push.go → pkg/pusher/collect_push.go

@@ -1,4 +1,4 @@
-package main
+package pusher
 
 import (
 	"context"

main.go → pkg/pusher/pusher.go

@@ -1,4 +1,4 @@
-package main
+package pusher
 
 import (
 	"os"
@@ -7,14 +7,27 @@ import (
 
 	"github.com/alecthomas/kingpin/v2"
 
+	"github.com/go-kit/log"
 	"github.com/go-kit/log/level"
 
 	"github.com/prometheus/common/promlog"
 	"github.com/prometheus/common/promlog/flag"
 	"github.com/prometheus/common/version"
+
+	"github.com/prometheus/prometheus/storage/remote"
+
+	customlog "github.com/fgouteroux/promk/pkg/log"
 )
 
-func main() {
+type Pusher struct {
+	logger   log.Logger
+	client   *remote.Client
+	interval time.Duration
+	jobLabel string
+	labels   map[string]string
+}
+
+func Setup() (p *Pusher) {
 	app := kingpin.New(filepath.Base(os.Args[0]), "Prometheus Keepalive Agent.").UsageWriter(os.Stdout)
 	baseURL := app.Flag("remote-write-url", "Prometheus remote-write url").Envar("PROMK_URL").Required().URL()
 	username := app.Flag("basic-auth.username", "Prometheus remote-write username").Envar("PROMK_USERNAME").String()
@@ -36,7 +49,12 @@ func main() {
 
 	kingpin.MustParse(app.Parse(os.Args[1:]))
 
-	logger := promlog.New(promlogConfig)
+	logger, err := customlog.InitLogger(promlogConfig)
+	if err != nil {
+		var logger log.Logger
+		level.Error(logger).Log("Failed to init custom logger", err)
+		os.Exit(1)
+	}
 
 	level.Info(logger).Log("msg", "Starting promk", "version", version.Info())              // #nosec G104
 	level.Info(logger).Log("msg", "Build context", "build_context", version.BuildContext()) // #nosec G104
@@ -52,17 +70,27 @@ func main() {
 		os.Exit(1)
 	}
 
+	return &Pusher{
+		logger:   logger,
+		client:   remoteWriteClient,
+		interval: *pushInterval,
+		jobLabel: *jobLabel,
+		labels:   *labels,
+	}
+}
+
+func (p *Pusher) Run() {
 	// create a new Ticker
-	tk := time.NewTicker(*pushInterval)
+	tk := time.NewTicker(p.interval)
 
 	// start the ticker
 	for range tk.C {
-		metric := CollectAndEncode(logger, *jobLabel, *labels)
-		err := Push(remoteWriteClient, metric)
+		metric := CollectAndEncode(p.logger, p.jobLabel, p.labels)
+		err := Push(p.client, metric)
 		if err != nil {
-			level.Error(logger).Log("msg", "Could not push to the remote write", "err", err) // #nosec G104
+			level.Error(p.logger).Log("msg", "Could not push to the remote write", "err", err) // #nosec G104
 		} else {
-			level.Info(logger).Log("msg", "Successful push to the remote write") // #nosec G104
+			level.Info(p.logger).Log("msg", "Successful push to the remote write") // #nosec G104
 		}
 	}
 }

pkg/windows/service.go

@@ -0,0 +1,45 @@
+//go:build windows
+// +build windows
+
+package windows
+
+import (
+	"fmt"
+	"log"
+
+	"golang.org/x/sys/windows/svc"
+)
+
+// WindowsExporterService channel for service stop
+type WindowsExporterService struct {
+	stopCh chan<- bool
+}
+
+// NewWindowsExporterService return new WindowsExporterService
+func NewWindowsExporterService(ch chan<- bool) *WindowsExporterService {
+	return &WindowsExporterService{stopCh: ch}
+}
+
+// Execute run programm directly or for service
+func (s *WindowsExporterService) Execute(args []string, r <-chan svc.ChangeRequest, changes chan<- svc.Status) (ssec bool, errno uint32) {
+	const cmdsAccepted = svc.AcceptStop | svc.AcceptShutdown
+	changes <- svc.Status{State: svc.StartPending}
+	changes <- svc.Status{State: svc.Running, Accepts: cmdsAccepted}
+loop:
+	for {
+		select {
+		case c := <-r:
+			switch c.Cmd {
+			case svc.Interrogate:
+				changes <- c.CurrentStatus
+			case svc.Stop, svc.Shutdown:
+				s.stopCh <- true
+				break loop
+			default:
+				log.Fatalf(fmt.Sprintf("unexpected control request #%d", c))
+			}
+		}
+	}
+	changes <- svc.Status{State: svc.StopPending}
+	return
+}