feat: add TunnelTime metric

README.md

@@ -42,11 +42,7 @@ To fetch and update MMDB files from [DB-IP](https://db-ip.com), you can do somet
 
 ## Full Working Example: Try It!
 
-Fetch dependencies for this demo:
-```
-GO111MODULE=off go get github.com/prometheus/prometheus/cmd/...
-```
-If that doesn't work, download the [prometheus](https://prometheus.io/download/) binary directly.
+Download the [Prometheus](https://prometheus.io/download/) binary.
 
 
 ### Run the server
@@ -60,7 +56,7 @@ In production, you may want to specify `-ip_country_db` to get per-country metri
 ### Run the Prometheus scraper for metrics collection
 On Terminal 2, start prometheus scraper for metrics collection:
 ```
-$(go env GOPATH)/bin/prometheus --config.file=cmd/outline-ss-server/prometheus_example.yml
+prometheus --config.file=cmd/outline-ss-server/prometheus_example.yml
 ```
 
 ### Run the SOCKS-to-Shadowsocks client

cmd/outline-ss-server/main.go

@@ -227,21 +227,23 @@ func readConfig(filename string) (*Config, error) {
 
 func main() {
 	var flags struct {
-		ConfigFile    string
-		MetricsAddr   string
-		IPCountryDB   string
-		IPASNDB       string
-		natTimeout    time.Duration
-		replayHistory int
-		Verbose       bool
-		Version       bool
+		ConfigFile                     string
+		MetricsAddr                    string
+		IPCountryDB                    string
+		IPASNDB                        string
+		natTimeout                     time.Duration
+		replayHistory                  int
+		EnableIPKeyConnectivityMetrics bool
+		Verbose                        bool
+		Version                        bool
 	}
 	flag.StringVar(&flags.ConfigFile, "config", "", "Configuration filename")
 	flag.StringVar(&flags.MetricsAddr, "metrics", "", "Address for the Prometheus metrics")
 	flag.StringVar(&flags.IPCountryDB, "ip_country_db", "", "Path to the ip-to-country mmdb file")
 	flag.StringVar(&flags.IPASNDB, "ip_asn_db", "", "Path to the ip-to-ASN mmdb file")
 	flag.DurationVar(&flags.natTimeout, "udptimeout", defaultNatTimeout, "UDP tunnel timeout")
 	flag.IntVar(&flags.replayHistory, "replay_history", 0, "Replay buffer size (# of handshakes)")
+	flag.BoolVar(&flags.EnableIPKeyConnectivityMetrics, "enable_ip_key_connectivity_metrics", false, "Enables the collection and reporting of IPKey connectivity metrics")
 	flag.BoolVar(&flags.Verbose, "verbose", false, "Enables verbose logging output")
 	flag.BoolVar(&flags.Version, "version", false, "The version of the server")
 
@@ -280,11 +282,11 @@ func main() {
 	}
 	ip2info, err := ipinfo.NewMMDBIPInfoMap(flags.IPCountryDB, flags.IPASNDB)
 	if err != nil {
-		logger.Fatalf("Could create IP info map: %v. Aborting", err)
+		logger.Fatalf("Failed to create IP info map: %v. Aborting", err)
 	}
 	defer ip2info.Close()
 
-	m := newPrometheusOutlineMetrics(ip2info, prometheus.DefaultRegisterer)
+	m := newPrometheusOutlineMetrics(ip2info, prometheus.DefaultRegisterer, flags.EnableIPKeyConnectivityMetrics)
 	m.SetBuildInfo(version)
 	_, err = RunSSServer(flags.ConfigFile, flags.natTimeout, m, flags.replayHistory)
 	if err != nil {

cmd/outline-ss-server/metrics.go

@@ -16,6 +16,7 @@ package main
 
 import (
 	"fmt"
+	"net"
 	"strconv"
 	"time"
 
@@ -25,8 +26,16 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 )
 
+const (
+	// How often to report the active IP key TunnelTime.
+	activeIPKeyTrackerReportingInterval = 5 * time.Second
+)
+
+var since = time.Since
+
 type outlineMetrics struct {
 	ipinfo.IPInfoMap
+	*activeIPKeyTracker
 
 	buildInfo            *prometheus.GaugeVec
 	accessKeys           prometheus.Gauge
@@ -36,6 +45,9 @@ type outlineMetrics struct {
 	timeToCipherMs       *prometheus.HistogramVec
 	// TODO: Add time to first byte.
 
+	IPKeyTimePerKey      *prometheus.CounterVec
+	IPKeyTimePerLocation *prometheus.CounterVec
+
 	tcpProbes               *prometheus.HistogramVec
 	tcpOpenConnections      *prometheus.CounterVec
 	tcpClosedConnections    *prometheus.CounterVec
@@ -49,11 +61,100 @@ type outlineMetrics struct {
 var _ service.TCPMetrics = (*outlineMetrics)(nil)
 var _ service.UDPMetrics = (*outlineMetrics)(nil)
 
+type activeClient struct {
+	IPKey           IPKey
+	connectionCount int
+	startTime       time.Time
+}
+
+func (c *activeClient) IsActive() bool {
+	return c.connectionCount > 0
+}
+
+type IPKey struct {
+	ip        string
+	accessKey string
+}
+
+type activeIPKeyTracker struct {
+	activeClients   map[IPKey]activeClient
+	metricsCallback func(IPKey, time.Duration)
+}
+
+// Reports time connected for all active clients, called at a regular interval.
+func (t *activeIPKeyTracker) reportAll() {
+	if len(t.activeClients) == 0 {
+		logger.Debugf("No active clients. No IPKey activity to report.")
+		return
+	}
+	for _, c := range t.activeClients {
+		t.reportDuration(c)
+	}
+}
+
+// Reports time connected for a given active client.
+func (t *activeIPKeyTracker) reportDuration(c activeClient) {
+	connDuration := since(c.startTime)
+	logger.Debugf("Reporting activity for key `%v`, duration: %v", c.IPKey.accessKey, connDuration)
+	t.metricsCallback(c.IPKey, connDuration)
+
+	// Reset the start time now that it's been reported.
+	c.startTime = time.Now()
+	t.activeClients[c.IPKey] = c
+}
+
+// Registers a new active connection for a client [net.Addr] and access key.
+func (t *activeIPKeyTracker) startConnection(addr net.Addr, accessKey string) {
+	hostname, _, _ := net.SplitHostPort(addr.String())
+	ipKey := IPKey{ip: hostname, accessKey: accessKey}
+
+	c, exists := t.activeClients[ipKey]
+	if !exists {
+		c = activeClient{ipKey, 0, time.Now()}
+	}
+	c.connectionCount++
+	t.activeClients[ipKey] = c
+}
+
+// Removes an active connection for a client [net.Addr] and access key.
+func (t *activeIPKeyTracker) stopConnection(addr net.Addr, accessKey string) {
+	hostname, _, _ := net.SplitHostPort(addr.String())
+	ipKey := IPKey{ip: hostname, accessKey: accessKey}
+
+	c := t.activeClients[ipKey]
+	c.connectionCount--
+	if !c.IsActive() {
+		t.reportDuration(c)
+		delete(t.activeClients, ipKey)
+		return
+	}
+	t.activeClients[ipKey] = c
+}
+
+func newActiveIPKeyTracker(callback func(IPKey, time.Duration)) *activeIPKeyTracker {
+	t := &activeIPKeyTracker{activeClients: make(map[IPKey]activeClient), metricsCallback: callback}
+	ticker := time.NewTicker(activeIPKeyTrackerReportingInterval)
+	done := make(chan struct{})
+	go func() {
+		for {
+			select {
+			case <-ticker.C:
+				t.reportAll()
+			case <-done:
+				logger.Debugf("done channel %p closed", done)
+				ticker.Stop()
+				return
+			}
+		}
+	}()
+	return t
+}
+
 // newPrometheusOutlineMetrics constructs a metrics object that uses
-// `ipCountryDB` to convert IP addresses to countries, and reports all
-// metrics to Prometheus via `registerer`.  `ipCountryDB` may be nil, but
+// `ip2info` to convert IP addresses to countries, and reports all
+// metrics to Prometheus via `registerer`. `ip2info` may be nil, but
 // `registerer` must not be.
-func newPrometheusOutlineMetrics(ip2info ipinfo.IPInfoMap, registerer prometheus.Registerer) *outlineMetrics {
+func newPrometheusOutlineMetrics(ip2info ipinfo.IPInfoMap, registerer prometheus.Registerer, enableIPKeyConnectivity bool) *outlineMetrics {
 	m := &outlineMetrics{
 		IPInfoMap: ip2info,
 		buildInfo: prometheus.NewGaugeVec(prometheus.GaugeOpts{
@@ -104,6 +205,16 @@ func newPrometheusOutlineMetrics(ip2info ipinfo.IPInfoMap, registerer prometheus
 					float64(7 * 24 * time.Hour.Milliseconds()), // Week
 				},
 			}, []string{"status"}),
+		IPKeyTimePerKey: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Namespace: "shadowsocks",
+			Name:      "ip_key_connectivity_seconds",
+			Help:      "Time at least 1 connection was open for a (IP, access key) pair, per key",
+		}, []string{"access_key"}),
+		IPKeyTimePerLocation: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Namespace: "shadowsocks",
+			Name:      "ip_key_connectivity_seconds_per_location",
+			Help:      "Time at least 1 connection was open for a (IP, access key) pair, per location",
+		}, []string{"location", "asn"}),
 		dataBytes: prometheus.NewCounterVec(
 			prometheus.CounterOpts{
 				Namespace: "shadowsocks",
@@ -145,10 +256,15 @@ func newPrometheusOutlineMetrics(ip2info ipinfo.IPInfoMap, registerer prometheus
 				Help:      "Entries removed from the UDP NAT table",
 			}),
 	}
+	if enableIPKeyConnectivity {
+		m.activeIPKeyTracker = newActiveIPKeyTracker(m.reportIPKeyActivity)
+	}
+	logger.Debugf("tracker: %v", m.activeIPKeyTracker)
 
 	// TODO: Is it possible to pass where to register the collectors?
 	registerer.MustRegister(m.buildInfo, m.accessKeys, m.ports, m.tcpProbes, m.tcpOpenConnections, m.tcpClosedConnections, m.tcpConnectionDurationMs,
-		m.dataBytes, m.dataBytesPerLocation, m.timeToCipherMs, m.udpPacketsFromClientPerLocation, m.udpAddedNatEntries, m.udpRemovedNatEntries)
+		m.dataBytes, m.dataBytesPerLocation, m.timeToCipherMs, m.udpPacketsFromClientPerLocation, m.udpAddedNatEntries, m.udpRemovedNatEntries,
+		m.IPKeyTimePerKey, m.IPKeyTimePerLocation)
 	return m
 }
 
@@ -161,10 +277,32 @@ func (m *outlineMetrics) SetNumAccessKeys(numKeys int, ports int) {
 	m.ports.Set(float64(ports))
 }
 
-func (m *outlineMetrics) AddOpenTCPConnection(clientInfo ipinfo.IPInfo) {
+func (m *outlineMetrics) AddOpenTCPConnection(addr net.Addr) {
+	clientInfo, err := ipinfo.GetIPInfoFromAddr(m.IPInfoMap, addr)
+	if err != nil {
+		logger.Warningf("Failed client info lookup: %v", err)
+	}
+	logger.Debugf("Got info \"%#v\" for IP %v", clientInfo, addr.String())
 	m.tcpOpenConnections.WithLabelValues(clientInfo.CountryCode.String(), asnLabel(clientInfo.ASN)).Inc()
 }
 
+// Reports total time connected, by access key and by country.
+func (m *outlineMetrics) reportIPKeyActivity(ipKey IPKey, duration time.Duration) {
+	m.IPKeyTimePerKey.WithLabelValues(ipKey.accessKey).Add(duration.Seconds())
+	ip := net.ParseIP(ipKey.ip)
+	clientInfo, err := ipinfo.GetIPInfoFromIP(m.IPInfoMap, ip)
+	if err != nil {
+		logger.Warningf("Failed client info lookup: %v", err)
+	}
+	m.IPKeyTimePerLocation.WithLabelValues(clientInfo.CountryCode.String(), asnLabel(clientInfo.ASN)).Add(duration.Seconds())
+}
+
+func (m *outlineMetrics) AddAuthenticatedTCPConnection(addr net.Addr, accessKey string) {
+	if m.activeIPKeyTracker != nil {
+		m.activeIPKeyTracker.startConnection(addr, accessKey)
+	}
+}
+
 // addIfNonZero helps avoid the creation of series that are always zero.
 func addIfNonZero(value int64, counterVec *prometheus.CounterVec, lvs ...string) {
 	if value > 0 {
@@ -179,7 +317,12 @@ func asnLabel(asn int) string {
 	return fmt.Sprint(asn)
 }
 
-func (m *outlineMetrics) AddClosedTCPConnection(clientInfo ipinfo.IPInfo, accessKey, status string, data metrics.ProxyMetrics, duration time.Duration) {
+func (m *outlineMetrics) AddClosedTCPConnection(addr net.Addr, accessKey, status string, data metrics.ProxyMetrics, duration time.Duration) {
+	clientInfo, err := ipinfo.GetIPInfoFromAddr(m.IPInfoMap, addr)
+	if err != nil {
+		logger.Warningf("Failed client info lookup: %v", err)
+	}
+	logger.Debugf("Got info \"%#v\" for IP %v", clientInfo, addr.String())
 	m.tcpClosedConnections.WithLabelValues(clientInfo.CountryCode.String(), asnLabel(clientInfo.ASN), status, accessKey).Inc()
 	m.tcpConnectionDurationMs.WithLabelValues(status).Observe(duration.Seconds() * 1000)
 	addIfNonZero(data.ClientProxy, m.dataBytes, "c>p", "tcp", accessKey)
@@ -190,6 +333,10 @@ func (m *outlineMetrics) AddClosedTCPConnection(clientInfo ipinfo.IPInfo, access
 	addIfNonZero(data.TargetProxy, m.dataBytesPerLocation, "p<t", "tcp", clientInfo.CountryCode.String(), asnLabel(clientInfo.ASN))
 	addIfNonZero(data.ProxyClient, m.dataBytes, "c<p", "tcp", accessKey)
 	addIfNonZero(data.ProxyClient, m.dataBytesPerLocation, "c<p", "tcp", clientInfo.CountryCode.String(), asnLabel(clientInfo.ASN))
+
+	if m.activeIPKeyTracker != nil {
+		m.activeIPKeyTracker.stopConnection(addr, accessKey)
+	}
 }
 
 func (m *outlineMetrics) AddUDPPacketFromClient(clientInfo ipinfo.IPInfo, accessKey, status string, clientProxyBytes, proxyTargetBytes int) {
@@ -207,12 +354,20 @@ func (m *outlineMetrics) AddUDPPacketFromTarget(clientInfo ipinfo.IPInfo, access
 	addIfNonZero(int64(proxyClientBytes), m.dataBytesPerLocation, "c<p", "udp", clientInfo.CountryCode.String(), asnLabel(clientInfo.ASN))
 }
 
-func (m *outlineMetrics) AddUDPNatEntry() {
+func (m *outlineMetrics) AddUDPNatEntry(addr net.Addr, accessKey string) {
 	m.udpAddedNatEntries.Inc()
+
+	if m.activeIPKeyTracker != nil {
+		m.activeIPKeyTracker.startConnection(addr, accessKey)
+	}
 }
 
-func (m *outlineMetrics) RemoveUDPNatEntry() {
+func (m *outlineMetrics) RemoveUDPNatEntry(addr net.Addr, accessKey string) {
 	m.udpRemovedNatEntries.Inc()
+
+	if m.activeIPKeyTracker != nil {
+		m.activeIPKeyTracker.stopConnection(addr, accessKey)
+	}
 }
 
 func (m *outlineMetrics) AddTCPProbe(status, drainResult string, port int, clientProxyBytes int64) {