Metrics server plugin (#622)

* Metrics server plugin

* Link to metrics from root README

plugin/metrics/README.md

@@ -0,0 +1,23 @@
+# metrics plugin for sshpiperd
+
+The metrics plugin provides a server to report metrics on active pipes and connection failures.
+By default, only the sshpiper_pipe_open_connections{remote_addr="...",user="..."} metric is collected.
+
+## Metrics
+
+Name                            | Type    | Labels                    | Description
+------------------------------- | ------- | ------------------------- | ------------
+sshpiper_pipe_open_connections  | Gauge   | remote_addr, user         | Incremented each time a pipe is successfully started, decremented on close
+sshpiper_pipe_create_errors     | Counter | remote_addr               | Incremented each time a pipe fails to be created (disabled by default)
+sshpiper_upstream_auth_failures | Counter | remote_addr, user, method | Incremented each time an upstream rejects the authentication method (disabled by default)
+
+## Usage
+
+Note: this is a metrics server plugin. 📈 you must use it with other routing/auth plugins.
+
+```
+sshpiperd other-plugin --other-option -- metrics --port <metrics-port>
+```
+
+Start the plugin with --collect-pipe-create-errors to enable sshpiper_pipe_create_errors
+Start the plugin with --collect-upstream-auth-failures to enable sshpiper_upstream_auth_failures

plugin/metrics/main.go

@@ -0,0 +1,172 @@
+package main
+
+import (
+	"fmt"
+	"net/http"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+	log "github.com/sirupsen/logrus"
+	"github.com/tg123/sshpiper/libplugin"
+	"github.com/urfave/cli/v2"
+)
+
+func main() {
+	libplugin.CreateAndRunPluginTemplate(&libplugin.PluginTemplate{
+		Name:  "metrics",
+		Usage: "sshpiperd metrics plugin, expose prometheus metrics after login",
+		Flags: []cli.Flag{
+			&cli.StringFlag{
+				Name:     "address",
+				Usage:    "Metrics server listen address",
+				Required: false,
+				EnvVars:  []string{"SSHPIPERD_METRICS_ADDRESS"},
+			},
+			&cli.IntFlag{
+				Name:     "port",
+				Usage:    "Metrics server listen port",
+				Required: false,
+				Value:    9000,
+				EnvVars:  []string{"SSHPIPERD_METRICS_PORT"},
+			},
+			&cli.BoolFlag{
+				Name:     "collect-pipe-create-errors",
+				Usage:    "Collect metrics on pipe creation errors",
+				Required: false,
+				Value:    false,
+				EnvVars:  []string{"SSHPIPERD_METRICS_COLLECT_PIPE_CREATE_ERRORS"},
+			},
+			&cli.BoolFlag{
+				Name:     "collect-upstream-auth-failures",
+				Usage:    "Collect metrics on upstream auth failures",
+				Required: false,
+				Value:    false,
+				EnvVars:  []string{"SSHPIPERD_METRICS_COLLECT_UPSTREAM_AUTH_FAILURES"},
+			},
+		},
+		CreateConfig: func(c *cli.Context) (*libplugin.SshPiperPluginConfig, error) {
+			port := c.Int("port")
+			address := c.String("address")
+			bindAddress := fmt.Sprintf("%v:%v", address, port)
+			metrics, config := newPrometheusMetrics(
+				c.Bool("collect-pipe-create-errors"), c.Bool("collect-upstream-auth-failures"),
+			)
+			go func(metrics *prometheusMetrics, bindAddress string) {
+				if err := metrics.ListenAndServe(bindAddress); err != nil {
+					log.Error("Metrics server error:", err)
+				}
+			}(metrics, bindAddress)
+			log.Info("Metrics server is listening on: ", bindAddress)
+			return config, nil
+		},
+	})
+}
+
+func newPrometheusMetrics(collectPipeCreateErrors, collectUpstreamAuthFailures bool) (*prometheusMetrics, *libplugin.SshPiperPluginConfig) {
+	registry := prometheus.NewRegistry()
+	openConnections := prometheus.NewGaugeVec(
+		// sshpiper_pipe_open_connections
+		prometheus.GaugeOpts{
+			Namespace: "sshpiper",
+			Subsystem: "pipe",
+			Name:      "open_connections",
+			Help:      "Number of open connections that currently exist partitioned by remote_addr and user",
+		},
+		[]string{"remote_addr", "username"},
+	)
+	registry.MustRegister(openConnections)
+
+	metrics := &prometheusMetrics{
+		registry:        registry,
+		openConnections: openConnections,
+	}
+	config := &libplugin.SshPiperPluginConfig{
+		PipeStartCallback: metrics.pipeStartCallback,
+		PipeErrorCallback: metrics.pipeErrorCallback,
+	}
+
+	// Optional metrics
+	if collectPipeCreateErrors {
+		metrics.pipeCreateErrors = prometheus.NewCounterVec(
+			// sshpiper_pipe_create_errors
+			prometheus.CounterOpts{
+				Namespace: "sshpiper",
+				Subsystem: "pipe",
+				Name:      "create_errors",
+				Help:      "Number of create pipe errors partitioned by remote_addr",
+			},
+			[]string{"remote_addr"},
+		)
+		registry.MustRegister(metrics.pipeCreateErrors)
+		config.PipeCreateErrorCallback = metrics.pipeCreateErrorCallback
+	}
+	if collectUpstreamAuthFailures {
+		metrics.upstreamAuthFailures = prometheus.NewCounterVec(
+			// sshpiper_upstream_auth_failures
+			prometheus.CounterOpts{
+				Namespace: "sshpiper",
+				Subsystem: "upstream",
+				Name:      "auth_failures",
+				Help:      "Number of upstream auth failures partitioned by remote_addr, user, and method",
+			},
+			[]string{"remote_addr", "user", "method"},
+		)
+		registry.MustRegister(metrics.upstreamAuthFailures)
+		config.UpstreamAuthFailureCallback = metrics.upstreamAuthFailureCallback
+	}
+	return metrics, config
+}
+
+type prometheusMetrics struct {
+	registry *prometheus.Registry
+
+	openConnections      *prometheus.GaugeVec
+	pipeCreateErrors     *prometheus.CounterVec
+	upstreamAuthFailures *prometheus.CounterVec
+}
+
+func (ms *prometheusMetrics) ListenAndServe(addr string) error {
+	http.Handle("/metrics", promhttp.InstrumentMetricHandler(
+		ms.registry, promhttp.HandlerFor(ms.registry, promhttp.HandlerOpts{
+			ErrorLog: errorLogger{},
+		}),
+	))
+	return http.ListenAndServe(addr, nil)
+}
+
+func (ms *prometheusMetrics) pipeStartCallback(conn libplugin.ConnMetadata) {
+	gauge, err := ms.openConnections.GetMetricWithLabelValues(conn.RemoteAddr(), conn.User())
+	if err != nil {
+		log.Error("Failed to fetch gauge for pipe start callback: ", err)
+		return
+	}
+	gauge.Inc()
+}
+
+func (ms *prometheusMetrics) pipeErrorCallback(conn libplugin.ConnMetadata, _ error) {
+	ms.openConnections.DeleteLabelValues(conn.RemoteAddr(), conn.User())
+}
+
+func (ms *prometheusMetrics) pipeCreateErrorCallback(remoteAddr string, _ error) {
+	counter, err := ms.pipeCreateErrors.GetMetricWithLabelValues(remoteAddr)
+	if err != nil {
+		log.Error("Failed to get counter for pipe create error callback: ", err)
+		return
+	}
+	counter.Inc()
+}
+
+func (ms *prometheusMetrics) upstreamAuthFailureCallback(conn libplugin.ConnMetadata, method string, _ error, _ []string) {
+	counter, err := ms.upstreamAuthFailures.GetMetricWithLabelValues(conn.RemoteAddr(), conn.User(), method)
+	if err != nil {
+		log.Error("Failed to get counter for upstream auth failure callback: ", err)
+		return
+	}
+	counter.Inc()
+}
+
+type errorLogger struct{}
+
+func (l errorLogger) Println(v ...any) {
+	log.Error(v...)
+}