dashboard / tuns / feat: metric drain #34 rss

 1diff --git a/Makefile b/Makefile
 2index 8b36002..31e617e 100644
 3--- a/Makefile
 4+++ b/Makefile
 5@@ -19,3 +19,7 @@ docs-prod: ssg
 6 dev:
 7 	go run main.go --http-address localhost:3000 --domain testing.ssi.sh
 8 .PHONY: dev
 9+
10+fmt:
11+	go fmt ./...
12+.PHONY: fmt

 1diff --git a/cmd/sish.go b/cmd/sish.go
 2index 159f02e..bba3f45 100644
 3--- a/cmd/sish.go
 4+++ b/cmd/sish.go
 5@@ -17,6 +17,7 @@ import (
 6 	"github.com/spf13/viper"
 7 	"gopkg.in/natefinch/lumberjack.v2"
 8 
 9+	"github.com/picosh/pubsub"
10 	pipeLogger "github.com/picosh/pubsub/log"
11 	picoUtils "github.com/picosh/utils"
12 )
13@@ -206,17 +207,21 @@ func initConfig() {
14 		slog.NewTextHandler(multiWriter, &slog.HandlerOptions{AddSource: true, Level: logLevel}),
15 	)
16 
17-	realLogger, err := pipeLogger.SendLogRegister(logger, &pipeLogger.PubSubConnectionInfo{
18+	info := &pubsub.RemoteClientInfo{
19 		RemoteHost:     picoUtils.GetEnv("PICO_PIPE_ENDPOINT", "send.pico.sh:22"),
20 		KeyLocation:    picoUtils.GetEnv("PICO_PIPE_KEY", "ssh_data/term_info_ed25519"),
21 		KeyPassphrase:  picoUtils.GetEnv("PICO_PIPE_PASSPHRASE", ""),
22 		RemoteHostname: picoUtils.GetEnv("PICO_PIPE_REMOTE_HOST", "send.pico.sh"),
23 		RemoteUser:     picoUtils.GetEnv("PICO_PIPE_USER", "pico"),
24-	}, 100)
25+	}
26+	realLogger, err := pipeLogger.SendLogRegister(logger, info, 100)
27 	if err != nil {
28 		slog.Error("unable to set real logger", slog.Any("error", err))
29 	}
30 
31+	metricDrain := pubsub.NewRemoteClientWriter(info, logger, 100)
32+	go metricDrain.KeepAlive("pub metric-drain -b=false")
33+
34 	trueLogger := logger
35 
36 	if realLogger != nil {

 1diff --git a/go.mod b/go.mod
 2index 32259c2..043199f 100644
 3--- a/go.mod
 4+++ b/go.mod
 5@@ -2,6 +2,8 @@ module github.com/antoniomika/sish
 6 
 7 go 1.23.1
 8 
 9+replace github.com/picosh/pubsub => /home/erock/dev/pico/pubsub
10+
11 require (
12 	github.com/ScaleFT/sshkeys v1.2.0
13 	github.com/antoniomika/multilistener v0.0.0-20240307222635-f0dc097d8acc

 1diff --git a/httpmuxer/httpmuxer.go b/httpmuxer/httpmuxer.go
 2index 7ab9094..1888123 100644
 3--- a/httpmuxer/httpmuxer.go
 4+++ b/httpmuxer/httpmuxer.go
 5@@ -31,6 +31,16 @@ import (
 6 	"github.com/gin-gonic/gin"
 7 )
 8 
 9+type MetricDrainVisit struct {
10+	UserID    string `json:"user_id"`
11+	Host      string `json:"host"`
12+	Path      string `json:"path"`
13+	IpAddress string `json:"ip_adress"`
14+	UserAgent string `json:"user_agent"`
15+	Referer   string `json:"referer"`
16+	Status    int    `json:"status"`
17+}
18+
19 // Start initializes the HTTP service.
20 func Start(state *utils.State) {
21 	releaseMode := gin.ReleaseMode
22@@ -106,6 +116,24 @@ func Start(state *utils.State) {
23 			param.ErrorMessage,
24 		)
25 
26+		visit := MetricDrainVisit{
27+			UserID:    "",
28+			Host:      param.Request.Host,
29+			IpAddress: param.ClientIP,
30+			Status:    param.StatusCode,
31+			Path:      originalURI,
32+			UserAgent: param.Request.UserAgent(),
33+			Referer:   param.Request.Referer(),
34+		}
35+		data, err := json.Marshal(visit)
36+		if err != nil {
37+			slog.Error("could not json marshall visit record", "err", err)
38+		}
39+		_, err = state.MetricDrain.Write(data)
40+		if err != nil {
41+			slog.Error("could not send to visit to metric-drain", "err", err)
42+		}
43+
44 		slog.Info(logLine)
45 
46 		if viper.GetBool("log-to-client") && param.Keys["httpHolder"] != nil {

 1diff --git a/sshmuxer/sshmuxer.go b/sshmuxer/sshmuxer.go
 2index e200bb7..1a6c014 100644
 3--- a/sshmuxer/sshmuxer.go
 4+++ b/sshmuxer/sshmuxer.go
 5@@ -16,6 +16,8 @@ import (
 6 	"github.com/antoniomika/sish/httpmuxer"
 7 	"github.com/antoniomika/sish/utils"
 8 	"github.com/antoniomika/syncmap"
 9+	"github.com/picosh/pubsub"
10+	picoUtils "github.com/picosh/utils"
11 	"github.com/pires/go-proxyproto"
12 	"github.com/spf13/viper"
13 	"golang.org/x/crypto/ssh"
14@@ -70,10 +72,21 @@ func Start() {
15 
16 	utils.WatchKeys()
17 
18+	info := &pubsub.RemoteClientInfo{
19+		RemoteHost:     picoUtils.GetEnv("PICO_PIPE_ENDPOINT", "send.pico.sh:22"),
20+		KeyLocation:    picoUtils.GetEnv("PICO_PIPE_KEY", "ssh_data/term_info_ed25519"),
21+		KeyPassphrase:  picoUtils.GetEnv("PICO_PIPE_PASSPHRASE", ""),
22+		RemoteHostname: picoUtils.GetEnv("PICO_PIPE_REMOTE_HOST", "send.pico.sh"),
23+		RemoteUser:     picoUtils.GetEnv("PICO_PIPE_USER", "pico"),
24+	}
25+	metricDrain := pubsub.NewRemoteClientWriter(info, slog.Default(), 100)
26+	go metricDrain.KeepAlive("pub metric-drain -b=false")
27+
28 	state := utils.NewState()
29 	state.Ports.HTTPPort = httpPort
30 	state.Ports.HTTPSPort = httpsPort
31 	state.Ports.SSHPort = sshPort
32+	state.MetricDrain = metricDrain
33 
34 	state.Console.State = state
35 

 1diff --git a/utils/conn.go b/utils/conn.go
 2index 4ce0ebc..57ef9e6 100644
 3--- a/utils/conn.go
 4+++ b/utils/conn.go
 5@@ -99,6 +99,16 @@ func (s *SSHConnection) User() string {
 6 	return user
 7 }
 8 
 9+// User returns the user of the session, either using the extensions user or the passed user.
10+func (s *SSHConnection) UserId() string {
11+	user, ok := s.SSHConn.Permissions.Extensions["userId"]
12+	if !ok {
13+		user = ""
14+	}
15+
16+	return user
17+}
18+
19 // TeeConn represents a simple net.Conn interface for SNI Processing.
20 type TeeConn struct {
21 	Conn     net.Conn

 1diff --git a/utils/state.go b/utils/state.go
 2index 594c299..f27e201 100644
 3--- a/utils/state.go
 4+++ b/utils/state.go
 5@@ -231,6 +231,7 @@ type State struct {
 6 	ActiveConnections prometheus.GaugeFunc
 7 	ActiveTunnels     *prometheus.GaugeVec
 8 	ConnectionTime    *prometheus.GaugeVec
 9+	MetricDrain       io.Writer
10 }
11 
12 // NewState returns a new State struct.

 1diff --git a/utils/utils.go b/utils/utils.go
 2index 013c13c..aa3c403 100644
 3--- a/utils/utils.go
 4+++ b/utils/utils.go
 5@@ -597,7 +597,13 @@ func checkAuthenticationKeyRequest(authUrl string, authKey []byte, addr net.Addr
 6 			return false, extensionsInfo, nil
 7 		}
 8 
 9+		userId, ok := userData["id"].(string)
10+		if !ok {
11+			return false, extensionsInfo, nil
12+		}
13+
14 		extensionsInfo["user"] = userName
15+		extensionsInfo["userId"] = userId
16 	}
17 
18 	return true, extensionsInfo, nil

 1diff --git a/httpmuxer/httpmuxer.go b/httpmuxer/httpmuxer.go
 2index 1888123..8bf5317 100644
 3--- a/httpmuxer/httpmuxer.go
 4+++ b/httpmuxer/httpmuxer.go
 5@@ -116,25 +116,30 @@ func Start(state *utils.State) {
 6 			param.ErrorMessage,
 7 		)
 8 
 9-		visit := MetricDrainVisit{
10-			UserID:    "",
11-			Host:      param.Request.Host,
12-			IpAddress: param.ClientIP,
13-			Status:    param.StatusCode,
14-			Path:      originalURI,
15-			UserAgent: param.Request.UserAgent(),
16-			Referer:   param.Request.Referer(),
17-		}
18-		data, err := json.Marshal(visit)
19-		if err != nil {
20-			slog.Error("could not json marshall visit record", "err", err)
21-		}
22-		_, err = state.MetricDrain.Write(data)
23-		if err != nil {
24-			slog.Error("could not send to visit to metric-drain", "err", err)
25-		}
26-
27-		slog.Info(logLine)
28+		state.SSHConnections.Range(func(I string, conn *utils.SSHConnection) bool {
29+			user := conn.User()
30+			userId := conn.UserId()
31+			slog.Info(logLine, "user", user, "userId", userId)
32+
33+			visit := MetricDrainVisit{
34+				UserID:    userId,
35+				Host:      param.Request.Host,
36+				IpAddress: param.ClientIP,
37+				Status:    param.StatusCode,
38+				Path:      originalURI,
39+				UserAgent: param.Request.UserAgent(),
40+				Referer:   param.Request.Referer(),
41+			}
42+			data, err := json.Marshal(visit)
43+			if err != nil {
44+				slog.Error("could not json marshall visit record", "err", err)
45+			}
46+			_, err = state.MetricDrain.Write(data)
47+			if err != nil {
48+				slog.Error("could not send to visit to metric-drain", "err", err)
49+			}
50+			return true
51+		})
52 
53 		if viper.GetBool("log-to-client") && param.Keys["httpHolder"] != nil {
54 			currentListener := param.Keys["httpHolder"].(*utils.HTTPHolder)