Add GET /telegraf for admin socket

Author: iBug

pkg/server/server.go

@@ -122,6 +122,36 @@ func (s *Server) snapshotConnections() []ConnectionInfo {
 	return out
 }
 
+func (s *Server) connectionCounts() (active, queued int) {
+	for _, info := range s.snapshotConnections() {
+		if info.QueuePos > 0 {
+			queued++
+			continue
+		}
+		active++
+	}
+	return active, queued
+}
+
+func escapeInfluxTagValue(value string) string {
+	replacer := strings.NewReplacer(",", `\,`, " ", `\ `, "=", `\=`)
+	return replacer.Replace(value)
+}
+
+func formatTelegrafLine(host string, active, queued int, timestamp time.Time) string {
+	return fmt.Sprintf("git-queue,host=%s active=%di,queued=%di %d\n",
+		escapeInfluxTagValue(host), active, queued, timestamp.UnixNano())
+}
+
+func (s *Server) telegrafMetrics() string {
+	host, err := os.Hostname()
+	if err != nil || host == "" {
+		host = "<unknown>"
+	}
+	active, queued := s.connectionCounts()
+	return formatTelegrafLine(host, active, queued, time.Now())
+}
+
 func remoteParts(attrs map[string]string, fallback net.Addr) (string, string) {
 	host := attrs["REMOTE_ADDR"]
 	port := attrs["REMOTE_PORT"]
@@ -287,16 +317,16 @@ func (s *Server) acceptLoop() {
 
 func (s *Server) adminServeLoop() {
 	mux := http.NewServeMux()
-	mux.HandleFunc("/connections", func(w http.ResponseWriter, r *http.Request) {
-		if r.Method != http.MethodGet {
-			http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
-			return
-		}
+	mux.HandleFunc("GET /connections", func(w http.ResponseWriter, r *http.Request) {
 		w.Header().Set("Content-Type", "application/json")
 		if err := json.NewEncoder(w).Encode(s.snapshotConnections()); err != nil {
 			log.Printf("Admin write: %v", err)
 		}
 	})
+	mux.HandleFunc("GET /telegraf", func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "text/plain; charset=utf-8")
+		_, _ = io.WriteString(w, s.telegrafMetrics())
+	})
 	if err := http.Serve(s.adminL, mux); err != nil && !errors.Is(err, net.ErrClosed) {
 		log.Printf("Admin serve: %v", err)
 	}

pkg/server/server_test.go

@@ -0,0 +1,29 @@
+package server
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestConnectionCounts(t *testing.T) {
+	s := &Server{
+		conns: map[uint64]ConnectionInfo{
+			1: {Index: 1, QueuePos: 0},
+			2: {Index: 2, QueuePos: 3},
+			3: {Index: 3, QueuePos: 0},
+			4: {Index: 4, QueuePos: 1},
+		},
+	}
+
+	active, queued := s.connectionCounts()
+	require.Equal(t, 2, active)
+	assert.Equal(t, 2, queued)
+}
+
+func TestFormatTelegrafLine(t *testing.T) {
+	line := formatTelegrafLine("host name,prod=1", 10, 15, time.Unix(0, 1778837100123456789))
+	assert.Equal(t, "git-queue,host=host\\ name\\,prod\\=1 active=10i,queued=15i 1778837100123456789\n", line)
+}