Add basic metrics

crew/metrics.go

@@ -0,0 +1,81 @@
+package crew
+
+import (
+	"sync/atomic"
+
+	"github.com/safing/portbase/api"
+	"github.com/safing/portbase/metrics"
+	"github.com/tevino/abool"
+)
+
+var (
+	connectOpDurationHistogram     *metrics.Histogram
+	connectOpDownloadDataHistogram *metrics.Histogram
+	connectOpUploadDataHistogram   *metrics.Histogram
+
+	metricsRegistered = abool.New()
+)
+
+func registerMetrics() (err error) {
+	// Only register metrics once.
+	if !metricsRegistered.SetToIf(false, true) {
+		return nil
+	}
+
+	// Connect Op Stats.
+
+	_, err = metrics.NewGauge(
+		"spn/op/connect/active/total",
+		nil,
+		getActiveConnectOpsStat,
+		&metrics.Options{
+			Name:       "SPN Active Connect Operations",
+			Permission: api.PermitUser,
+		},
+	)
+	if err != nil {
+		return err
+	}
+
+	connectOpDurationHistogram, err = metrics.NewHistogram(
+		"spn/op/connect/duration/seconds",
+		nil,
+		&metrics.Options{
+			Name:       "SPN Connect Operation Duration",
+			Permission: api.PermitUser,
+		},
+	)
+	if err != nil {
+		return err
+	}
+
+	connectOpDownloadDataHistogram, err = metrics.NewHistogram(
+		"spn/op/connect/download/bytes",
+		nil,
+		&metrics.Options{
+			Name:       "SPN Connect Operation Downloaded Data",
+			Permission: api.PermitUser,
+		},
+	)
+	if err != nil {
+		return err
+	}
+
+	connectOpUploadDataHistogram, err = metrics.NewHistogram(
+		"spn/op/connect/upload/bytes",
+		nil,
+		&metrics.Options{
+			Name:       "SPN Connect Operation Uploaded Data",
+			Permission: api.PermitUser,
+		},
+	)
+	if err != nil {
+		return err
+	}
+
+	return err
+}
+
+func getActiveConnectOpsStat() float64 {
+	return float64(atomic.LoadInt64(activeConnectOps))
+}

crew/module.go

@@ -7,5 +7,9 @@ import (
 var module *modules.Module
 
 func init() {
-	module = modules.Register("crew", nil, nil, nil, "navigator", "intel", "cabin")
+	module = modules.Register("crew", nil, start, nil, "navigator", "intel", "cabin")
+}
+
+func start() error {
+	return registerMetrics()
 }

crew/op_connect.go

@@ -7,6 +7,7 @@ import (
 	"io"
 	"net"
 	"strconv"
+	"sync/atomic"
 	"time"
 
 	"github.com/safing/portbase/container"
@@ -20,6 +21,10 @@ import (
 
 const ConnectOpType string = "connect"
 
+var (
+	activeConnectOps = new(int64)
+)
+
 type ConnectOp struct {
 	terminal.OpBase
 	*terminal.DuplexFlowQueue
@@ -29,6 +34,9 @@ type ConnectOp struct {
 	// cancelCtx cancels ctx.
 	cancelCtx context.CancelFunc
 
+	downloaded *uint64
+	uploaded   *uint64
+
 	t       terminal.OpTerminal
 	conn    net.Conn
 	entry   bool
@@ -160,6 +168,10 @@ func runConnectOp(t terminal.OpTerminal, opID uint32, data *container.Container)
 	op.ctx, op.cancelCtx = context.WithCancel(context.Background())
 	op.DuplexFlowQueue = terminal.NewDuplexFlowQueue(op, request.QueueSize, op.submitUpstream)
 
+	// Setup metrics.
+	op.downloaded = new(uint64)
+	op.uploaded = new(uint64)
+
 	// Start worker.
 	module.StartWorker("connect op conn reader", op.connReader)
 	module.StartWorker("connect op conn writer", op.connWriter)
@@ -177,6 +189,18 @@ func (op *ConnectOp) submitUpstream(c *container.Container) {
 }
 
 func (op *ConnectOp) connReader(_ context.Context) error {
+	// Metrics setup and submitting.
+	if !op.entry {
+		atomic.AddInt64(activeConnectOps, 1)
+		started := time.Now()
+		defer func() {
+			atomic.AddInt64(activeConnectOps, -1)
+			connectOpDurationHistogram.UpdateDuration(started)
+			connectOpDownloadDataHistogram.Update(float64(atomic.LoadUint64(op.downloaded)))
+			connectOpUploadDataHistogram.Update(float64(atomic.LoadUint64(op.uploaded)))
+		}()
+	}
+
 	for {
 		buf := make([]byte, 1500)
 		n, err := op.conn.Read(buf)
@@ -193,6 +217,11 @@ func (op *ConnectOp) connReader(_ context.Context) error {
 			continue
 		}
 
+		// Count downloaded data for metrics on the Hub.
+		if !op.entry {
+			atomic.AddUint64(op.downloaded, uint64(n))
+		}
+
 		tErr := op.DuplexFlowQueue.Send(container.New(buf[:n]))
 		if tErr != nil {
 			op.t.OpEnd(op, tErr.Wrap("failed to send data (dfq) read from %s", op.connectedType()))
@@ -228,6 +257,11 @@ writing:
 			continue writing
 		}
 
+		// Count uploaded data for metrics on the Hub.
+		if !op.entry {
+			atomic.AddUint64(op.uploaded, uint64(len(data)))
+		}
+
 		// Send all given data.
 		for {
 			n, err := op.conn.Write(data)

docks/crane.go

@@ -386,6 +386,9 @@ func (crane *Crane) unloadUntilFull(buf []byte) error {
 
 		// Return if buffer has been fully filled.
 		if bytesRead == len(buf) {
+			// Submit metrics.
+			totalIncomingTraffic.Add(bytesRead)
+
 			return nil
 		}
 	}
@@ -654,9 +657,15 @@ func (crane *Crane) load(c *container.Container) error {
 		return fmt.Errorf("failed to encrypt: %w", err)
 	}
 
-	// Load onto ship.
+	// Finalize data.
 	c.PrependLength()
-	err = crane.ship.Load(c.CompileData())
+	readyToSend := c.CompileData()
+
+	// Submit metrics.
+	totalOutgoingTraffic.Add(len(readyToSend))
+
+	// Load onto ship.
+	err = crane.ship.Load(readyToSend)
 	if err != nil {
 		return fmt.Errorf("failed to load ship: %w", err)
 	}