refactor(metrics): make IO reporting and country lookup asynchronous

Author: garmr-ulfr

cmd/cmd_run.go

@@ -122,10 +122,7 @@ func create(configPath string, datacapURL string) (*box.Box, context.CancelFunc,
 	if _, ok := mp.(noop.MeterProvider); ok {
 		log.Info("Metrics not enabled, no meter provider configured")
 	} else {
-		metricsTracker, err := metrics.NewTracker()
-		if err != nil {
-			return nil, nil, fmt.Errorf("create metrics tracker: %w", err)
-		}
+		metricsTracker := metrics.NewTracker(ctx)
 		instance.Router().AppendTracker(metricsTracker)
 		log.Info("Metric Tracking Enabled")
 	}

cmd/main.go

@@ -5,13 +5,16 @@ import (
 	"fmt"
 	"os"
 	"strings"
+	"time"
 
-	box "github.com/getlantern/lantern-box"
-	"github.com/getlantern/lantern-box/otel"
-	"github.com/getlantern/lantern-box/tracker/metrics"
+	"github.com/getlantern/geo"
 	"github.com/spf13/cobra"
 	sdkotel "go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/metric/noop"
+
+	box "github.com/getlantern/lantern-box"
+	"github.com/getlantern/lantern-box/otel"
+	"github.com/getlantern/lantern-box/tracker/metrics"
 )
 
 type ProxyInfo struct {
@@ -81,7 +84,8 @@ func preRun(cmd *cobra.Command, args []string) {
 		ProxyProtocol:    proxyInfo.Protocol,
 	})
 
-	metrics.SetupMetricsManager(geoCityURL, cityDatabaseName)
+	geolookup := geo.FromWeb(geoCityURL, cityDatabaseName, 24*time.Hour, cityDatabaseName, geo.CountryCode)
+	metrics.SetupMetricsManager(geolookup)
 }
 
 func main() {

go.mod

@@ -1,6 +1,6 @@
 module github.com/getlantern/lantern-box
 
-go 1.24.1
+go 1.25.1
 
 replace github.com/sagernet/sing => github.com/getlantern/sing v0.7.18-lantern
 

tracker/metrics/conn.go

@@ -1,39 +1,33 @@
 package metrics
 
 import (
-	"context"
 	"net"
 	"time"
-
-	"github.com/sagernet/sing-box/adapter"
-	"go.opentelemetry.io/otel/attribute"
-	"go.opentelemetry.io/otel/metric"
 )
 
 // Conn wraps a net.Conn and tracks metrics such as bytes sent and received.
 type Conn struct {
 	net.Conn
-	attributes []attribute.KeyValue
-	startTime  time.Time
+	attrs     *attributes
+	tracker   *MetricsTracker
+	startTime time.Time
 }
 
 // NewConn creates a new Conn instance.
-func NewConn(conn net.Conn, metadata *adapter.InboundContext) net.Conn {
-	attributes := metadataToAttributes(metadata)
-	metrics.Connections.Add(context.Background(), 1, metric.WithAttributes(attributes...))
+func NewConn(conn net.Conn, attrs *attributes, tracker *MetricsTracker) net.Conn {
 	return &Conn{
-		Conn:       conn,
-		attributes: attributes,
-		startTime:  time.Now(),
+		Conn:      conn,
+		attrs:     attrs,
+		tracker:   tracker,
+		startTime: time.Now(),
 	}
 }
 
 // Read overrides net.Conn's Read method to track received bytes.
 func (c *Conn) Read(b []byte) (n int, err error) {
 	n, err = c.Conn.Read(b)
 	if n > 0 {
-		attrs := append(c.attributes, attribute.KeyValue{Key: "direction", Value: attribute.StringValue("receive")})
-		metrics.ProxyIO.Add(context.Background(), int64(n), metric.WithAttributes(attrs...))
+		c.tracker.TrackIO(rx, n, c.attrs)
 	}
 	return
 }
@@ -42,17 +36,15 @@ func (c *Conn) Read(b []byte) (n int, err error) {
 func (c *Conn) Write(b []byte) (n int, err error) {
 	n, err = c.Conn.Write(b)
 	if n > 0 {
-		attrs := append(c.attributes, attribute.KeyValue{Key: "direction", Value: attribute.StringValue("transmit")})
-		metrics.ProxyIO.Add(context.Background(), int64(n), metric.WithAttributes(attrs...))
+		c.tracker.TrackIO(tx, n, c.attrs)
 	}
 	return
 }
 
 // Close overrides net.Conn's Close method to track connection duration.
 func (c *Conn) Close() error {
 	duration := time.Since(c.startTime).Milliseconds()
-	metrics.duration.Record(context.Background(), duration, metric.WithAttributes(c.attributes...))
-	metrics.conns.Add(context.Background(), -1, metric.WithAttributes(c.attributes...))
+	c.tracker.Leave(duration, c.attrs)
 	return c.Conn.Close()
 }
 

tracker/metrics/metrics.go

@@ -6,13 +6,8 @@
 package metrics
 
 import (
-	"time"
-
 	"github.com/getlantern/geo"
-	"github.com/sagernet/sing-box/adapter"
-	"github.com/sagernet/sing-box/log"
 	"go.opentelemetry.io/otel"
-	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/metric"
 	"go.opentelemetry.io/otel/metric/noop"
 )
@@ -27,55 +22,39 @@ type metricsManager struct {
 	countryLookup geo.CountryLookup
 }
 
-var metrics = &metricsManager{}
+var metrics = &metricsManager{
+	ProxyIO:       &noop.Int64Counter{},
+	Connections:   &noop.Int64Counter{},
+	conns:         &noop.Int64UpDownCounter{},
+	duration:      &noop.Int64Histogram{},
+	countryLookup: geo.NoLookup{},
+}
 
-func SetupMetricsManager(geolite2CityURL, cityDBFile string) {
+func SetupMetricsManager(countryLookup geo.CountryLookup) {
 	meter := otel.GetMeterProvider().Meter("lantern-box")
 
-	pIO, err := meter.Int64Counter("proxy.io", metric.WithUnit("bytes"))
-	if err != nil {
-		pIO = &noop.Int64Counter{}
+	if pIO, err := meter.Int64Counter("proxy.io", metric.WithUnit("bytes")); err == nil {
+		metrics.ProxyIO = pIO
 	}
 
-	connections, err := meter.Int64Counter("proxy.connections")
-	if err != nil {
-		connections = &noop.Int64Counter{}
+	if connections, err := meter.Int64Counter("proxy.connections"); err == nil {
+		metrics.Connections = connections
 	}
 
 	// Track the number of connections.
 	conns, err := meter.Int64UpDownCounter("sing.connections", metric.WithDescription("Number of connections"))
-	if err != nil {
-		conns = &noop.Int64UpDownCounter{}
+	if err == nil {
+		metrics.conns = conns
 	}
 
 	// Track connection duration.
 	duration, err := meter.Int64Histogram("sing.connection_duration", metric.WithDescription("Connection duration"))
-	if err != nil {
-		duration = &noop.Int64Histogram{}
+	if err == nil {
+		metrics.duration = duration
 	}
 
-	metrics.meter = meter
-	metrics.ProxyIO = pIO
-	metrics.duration = duration
-	metrics.Connections = connections
-	metrics.conns = conns
-
-	metrics.countryLookup = geo.FromWeb(geolite2CityURL, cityDBFile, 24*time.Hour, cityDBFile, geo.CountryCode)
-	if metrics.countryLookup == nil {
-		metrics.countryLookup = geo.NoLookup{}
-	}
-
-	log.Info("metrics manager set up completed")
-}
-
-func metadataToAttributes(metadata *adapter.InboundContext) []attribute.KeyValue {
-	// Convert metadata to attributes
-	fromCountry := metrics.countryLookup.CountryCode(metadata.Source.IPAddr().IP)
-	return []attribute.KeyValue{
-		attribute.String("country", fromCountry),
-		attribute.String("protocol", metadata.Protocol),
-		attribute.String("inbound", metadata.Inbound),
-		attribute.String("inbound_type", metadata.InboundType),
-		attribute.String("outbound", metadata.Outbound),
+	if countryLookup != nil {
+		metrics.countryLookup = countryLookup
 	}
+	metrics.meter = meter
 }

tracker/metrics/packet_conn.go

@@ -1,32 +1,27 @@
 package metrics
 
 import (
-	"context"
 	"time"
 
-	"github.com/sagernet/sing-box/adapter"
 	"github.com/sagernet/sing/common/buf"
 	M "github.com/sagernet/sing/common/metadata"
 	N "github.com/sagernet/sing/common/network"
-	"go.opentelemetry.io/otel/attribute"
-	"go.opentelemetry.io/otel/metric"
 )
 
 // PacketConn wraps a sing-box network.PacketConn and tracks metrics such as bytes sent and received.
 type PacketConn struct {
 	N.PacketConn
-	attributes []attribute.KeyValue
-	startTime  time.Time
+	attrs     *attributes
+	tracker   *MetricsTracker
+	startTime time.Time
 }
 
 // NewPacketConn creates a new PacketConn instance.
-func NewPacketConn(conn N.PacketConn, metadata *adapter.InboundContext) N.PacketConn {
-	attributes := metadataToAttributes(metadata)
-	metrics.Connections.Add(context.Background(), 1, metric.WithAttributes(attributes...))
-
+func NewPacketConn(conn N.PacketConn, attrs *attributes, tracker *MetricsTracker) N.PacketConn {
 	return &PacketConn{
 		PacketConn: conn,
-		attributes: attributes,
+		attrs:      attrs,
+		tracker:    tracker,
 		startTime:  time.Now(),
 	}
 }
@@ -38,26 +33,23 @@ func (c *PacketConn) ReadPacket(buffer *buf.Buffer) (destination M.Socksaddr, er
 		return dest, err
 	}
 	if buffer.Len() > 0 {
-		attrs := append(c.attributes, attribute.KeyValue{Key: "direction", Value: attribute.StringValue("receive")})
-		metrics.ProxyIO.Add(context.Background(), int64(buffer.Len()), metric.WithAttributes(attrs...))
+		c.tracker.TrackIO(rx, buffer.Len(), c.attrs)
 	}
 	return dest, nil
 }
 
 // WritePacket overrides network.PacketConn's WritePacket method to track sent bytes.
 func (c *PacketConn) WritePacket(buffer *buf.Buffer, destination M.Socksaddr) error {
 	if buffer.Len() > 0 {
-		attrs := append(c.attributes, attribute.KeyValue{Key: "direction", Value: attribute.StringValue("transmit")})
-		metrics.ProxyIO.Add(context.Background(), int64(buffer.Len()), metric.WithAttributes(attrs...))
+		c.tracker.TrackIO(tx, buffer.Len(), c.attrs)
 	}
 	return c.PacketConn.WritePacket(buffer, destination)
 }
 
 // Close overrides net.PacketConn's Close method to track connection duration.
 func (c *PacketConn) Close() error {
 	duration := time.Since(c.startTime).Milliseconds()
-	metrics.duration.Record(context.Background(), duration, metric.WithAttributes(c.attributes...))
-	metrics.conns.Add(context.Background(), -1, metric.WithAttributes(c.attributes...))
+	c.tracker.Leave(duration, c.attrs)
 	return c.PacketConn.Close()
 }