Merge pull request #12 from Jigsaw-Code/fortuna-opt

Keep active ciphers at the front and add metrics

Author: fortuna

metrics/metrics.go

@@ -21,7 +21,7 @@ import (
 	"time"
 
 	onet "github.com/Jigsaw-Code/outline-ss-server/net"
-	"github.com/oschwald/geoip2-golang"
+	geoip2 "github.com/oschwald/geoip2-golang"
 	"github.com/prometheus/client_golang/prometheus"
 )
 
@@ -30,21 +30,25 @@ type ShadowsocksMetrics interface {
 	GetLocation(net.Addr) (string, error)
 
 	SetNumAccessKeys(numKeys int, numPorts int)
-	AddUDPPacketFromClient(clientLocation, accessKey, status string, clientProxyBytes, proxyTargetBytes int)
-	AddUDPPacketFromTarget(clientLocation, accessKey, status string, targetProxyBytes, proxyClientBytes int)
+
+	// TCP metrics
 	AddOpenTCPConnection(clientLocation string)
-	AddClosedTCPConnection(clientLocation, accessKey, status string, data ProxyMetrics, duration time.Duration)
+	AddClosedTCPConnection(clientLocation, accessKey, status string, data ProxyMetrics, timeToCipher, duration time.Duration)
 
-	AddUdpNatEntry()
-	RemoveUdpNatEntry()
+	// UDP metrics
+	AddUDPPacketFromClient(clientLocation, accessKey, status string, clientProxyBytes, proxyTargetBytes int, timeToCipher time.Duration)
+	AddUDPPacketFromTarget(clientLocation, accessKey, status string, targetProxyBytes, proxyClientBytes int)
+	AddUDPNatEntry()
+	RemoveUDPNatEntry()
 }
 
 type shadowsocksMetrics struct {
 	ipCountryDB *geoip2.Reader
 
-	accessKeys prometheus.Gauge
-	ports      prometheus.Gauge
-	dataBytes  *prometheus.CounterVec
+	accessKeys     prometheus.Gauge
+	ports          prometheus.Gauge
+	dataBytes      *prometheus.CounterVec
+	timeToCipherMs *prometheus.SummaryVec
 	// TODO: Add time to first byte.
 
 	tcpOpenConnections   *prometheus.CounterVec
@@ -87,14 +91,21 @@ func NewShadowsocksMetrics(ipCountryDB *geoip2.Reader) ShadowsocksMetrics {
 				Subsystem:  "tcp",
 				Name:       "connection_duration_ms",
 				Help:       "TCP connection duration distributions.",
-				Objectives: map[float64]float64{0.5: 0.05, 0.9: 0.01, 0.99: 0.001},
+				Objectives: map[float64]float64{0.5: 0.02, 0.9: 0.01, 0.99: 0.005},
 			}, []string{"location", "status", "access_key"}),
 		dataBytes: prometheus.NewCounterVec(
 			prometheus.CounterOpts{
 				Namespace: "shadowsocks",
 				Name:      "data_bytes",
 				Help:      "Bytes transferred by the proxy",
 			}, []string{"dir", "proto", "location", "status", "access_key"}),
+		timeToCipherMs: prometheus.NewSummaryVec(
+			prometheus.SummaryOpts{
+				Namespace:  "shadowsocks",
+				Name:       "time_to_cipher_ms",
+				Help:       "Time needed to find the cipher",
+				Objectives: map[float64]float64{0.5: 0.02, 0.9: 0.01, 0.99: 0.005},
+			}, []string{"proto", "location", "status", "access_key"}),
 		udpAddedNatEntries: prometheus.NewCounter(
 			prometheus.CounterOpts{
 				Namespace: "shadowsocks",
@@ -112,7 +123,7 @@ func NewShadowsocksMetrics(ipCountryDB *geoip2.Reader) ShadowsocksMetrics {
 	}
 	// TODO: Is it possible to pass where to register the collectors?
 	prometheus.MustRegister(m.accessKeys, m.ports, m.tcpOpenConnections, m.tcpClosedConnections, m.tcpConnectionDurationMs,
-		m.dataBytes, m.udpAddedNatEntries, m.udpRemovedNatEntries)
+		m.dataBytes, m.timeToCipherMs, m.udpAddedNatEntries, m.udpRemovedNatEntries)
 	return m
 }
 
@@ -163,16 +174,18 @@ func (m *shadowsocksMetrics) AddOpenTCPConnection(clientLocation string) {
 	m.tcpOpenConnections.WithLabelValues(clientLocation).Inc()
 }
 
-func (m *shadowsocksMetrics) AddClosedTCPConnection(clientLocation, accessKey, status string, data ProxyMetrics, duration time.Duration) {
+func (m *shadowsocksMetrics) AddClosedTCPConnection(clientLocation, accessKey, status string, data ProxyMetrics, timeToCipher, duration time.Duration) {
 	m.tcpClosedConnections.WithLabelValues(clientLocation, status, accessKey).Inc()
 	m.tcpConnectionDurationMs.WithLabelValues(clientLocation, status, accessKey).Observe(duration.Seconds() * 1000)
+	m.timeToCipherMs.WithLabelValues("tcp", clientLocation, status, accessKey).Observe(timeToCipher.Seconds() * 1000)
 	m.dataBytes.WithLabelValues("c>p", "tcp", clientLocation, status, accessKey).Add(float64(data.ClientProxy))
 	m.dataBytes.WithLabelValues("p>t", "tcp", clientLocation, status, accessKey).Add(float64(data.ProxyTarget))
 	m.dataBytes.WithLabelValues("p<t", "tcp", clientLocation, status, accessKey).Add(float64(data.TargetProxy))
 	m.dataBytes.WithLabelValues("c<p", "tcp", clientLocation, status, accessKey).Add(float64(data.ProxyClient))
 }
 
-func (m *shadowsocksMetrics) AddUDPPacketFromClient(clientLocation, accessKey, status string, clientProxyBytes, proxyTargetBytes int) {
+func (m *shadowsocksMetrics) AddUDPPacketFromClient(clientLocation, accessKey, status string, clientProxyBytes, proxyTargetBytes int, timeToCipher time.Duration) {
+	m.timeToCipherMs.WithLabelValues("udp", clientLocation, status, accessKey).Observe(timeToCipher.Seconds() * 1000)
 	m.dataBytes.WithLabelValues("c>p", "udp", clientLocation, status, accessKey).Add(float64(clientProxyBytes))
 	m.dataBytes.WithLabelValues("p>t", "udp", clientLocation, status, accessKey).Add(float64(proxyTargetBytes))
 }
@@ -182,11 +195,11 @@ func (m *shadowsocksMetrics) AddUDPPacketFromTarget(clientLocation, accessKey, s
 	m.dataBytes.WithLabelValues("c<p", "udp", clientLocation, status, accessKey).Add(float64(proxyClientBytes))
 }
 
-func (m *shadowsocksMetrics) AddUdpNatEntry() {
+func (m *shadowsocksMetrics) AddUDPNatEntry() {
 	m.udpAddedNatEntries.Inc()
 }
 
-func (m *shadowsocksMetrics) RemoveUdpNatEntry() {
+func (m *shadowsocksMetrics) RemoveUDPNatEntry() {
 	m.udpRemovedNatEntries.Inc()
 }
 

server.go

@@ -54,7 +54,7 @@ func init() {
 type SSPort struct {
 	tcpService shadowsocks.TCPService
 	udpService shadowsocks.UDPService
-	keys       map[string]shadowaead.Cipher
+	cipherList shadowsocks.CipherList
 }
 
 type SSServer struct {
@@ -73,10 +73,10 @@ func (s *SSServer) startPort(portNum int) error {
 		return fmt.Errorf("Failed to start UDP on port %v: %v", portNum, err)
 	}
 	logger.Infof("Listening TCP and UDP on port %v", portNum)
-	port := &SSPort{keys: make(map[string]shadowaead.Cipher)}
+	port := &SSPort{cipherList: shadowsocks.NewCipherList()}
 	// TODO: Register initial data metrics at zero.
-	port.tcpService = shadowsocks.NewTCPService(listener, &port.keys, s.m)
-	port.udpService = shadowsocks.NewUDPService(packetConn, s.natTimeout, &port.keys, s.m)
+	port.tcpService = shadowsocks.NewTCPService(listener, &port.cipherList, s.m)
+	port.udpService = shadowsocks.NewUDPService(packetConn, s.natTimeout, &port.cipherList, s.m)
 	s.ports[portNum] = port
 	go port.udpService.Start()
 	go port.tcpService.Start()
@@ -108,13 +108,13 @@ func (s *SSServer) loadConfig(filename string) error {
 	}
 
 	portChanges := make(map[int]int)
-	portKeys := make(map[int]map[string]shadowaead.Cipher)
+	portCiphers := make(map[int]shadowsocks.CipherList)
 	for _, keyConfig := range config.Keys {
 		portChanges[keyConfig.Port] = 1
-		keys, ok := portKeys[keyConfig.Port]
+		cipherList, ok := portCiphers[keyConfig.Port]
 		if !ok {
-			keys = make(map[string]shadowaead.Cipher)
-			portKeys[keyConfig.Port] = keys
+			cipherList = shadowsocks.NewCipherList()
+			portCiphers[keyConfig.Port] = cipherList
 		}
 		cipher, err := core.PickCipher(keyConfig.Cipher, nil, keyConfig.Secret)
 		if err != nil {
@@ -127,7 +127,7 @@ func (s *SSServer) loadConfig(filename string) error {
 		if !ok {
 			return fmt.Errorf("Only AEAD ciphers are supported. Found %v", keyConfig.Cipher)
 		}
-		keys[keyConfig.ID] = aead
+		cipherList.PushBack(keyConfig.ID, aead)
 	}
 	for port := range s.ports {
 		portChanges[port] = portChanges[port] - 1
@@ -143,11 +143,11 @@ func (s *SSServer) loadConfig(filename string) error {
 			}
 		}
 	}
-	for portNum, keys := range portKeys {
-		s.ports[portNum].keys = keys
+	for portNum, cipherList := range portCiphers {
+		s.ports[portNum].cipherList = cipherList
 	}
 	logger.Infof("Loaded %v access keys", len(config.Keys))
-	s.m.SetNumAccessKeys(len(config.Keys), len(portKeys))
+	s.m.SetNumAccessKeys(len(config.Keys), len(portCiphers))
 	return nil
 }
 

shadowsocks/cipher_cache.go

@@ -0,0 +1,69 @@
+// Copyright 2018 Jigsaw Operations LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package shadowsocks
+
+import (
+	"container/list"
+	"sync"
+
+	"github.com/shadowsocks/go-shadowsocks2/shadowaead"
+)
+
+// CipherEntry holds a Cipher with an identifier.
+type CipherEntry struct {
+	ID     string
+	Cipher shadowaead.Cipher
+}
+
+// CipherList is a list of CipherEntry elements that allows for thread-safe snapshotting and
+// moving to front.
+type CipherList interface {
+	PushBack(id string, cipher shadowaead.Cipher) *list.Element
+	SafeSnapshot() []*list.Element
+	SafeMoveToFront(e *list.Element)
+}
+
+type cipherList struct {
+	CipherList
+	list *list.List
+	mu   sync.RWMutex
+}
+
+// NewCipherList creates an empty CipherList
+func NewCipherList() CipherList {
+	return &cipherList{list: list.New()}
+}
+
+func (cl *cipherList) PushBack(id string, cipher shadowaead.Cipher) *list.Element {
+	return cl.list.PushBack(&CipherEntry{ID: id, Cipher: cipher})
+}
+
+func (cl *cipherList) SafeSnapshot() []*list.Element {
+	cl.mu.RLock()
+	defer cl.mu.RUnlock()
+	cipherArray := make([]*list.Element, cl.list.Len())
+	i := 0
+	for e := cl.list.Front(); e != nil; e = e.Next() {
+		cipherArray[i] = e
+		i++
+	}
+	return cipherArray
+}
+
+func (cl *cipherList) SafeMoveToFront(e *list.Element) {
+	cl.mu.Lock()
+	defer cl.mu.Unlock()
+	cl.list.MoveToFront(e)
+}

shadowsocks/cipher_list.go

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package testing
+package shadowsocks
 
 import (
 	"fmt"
@@ -21,16 +21,16 @@ import (
 	"github.com/shadowsocks/go-shadowsocks2/shadowaead"
 )
 
-func MakeTestCiphers(numCiphers int) (map[string]shadowaead.Cipher, error) {
-	cipherList := make(map[string]shadowaead.Cipher)
+func MakeTestCiphers(numCiphers int) (CipherList, error) {
+	cipherList := NewCipherList()
 	for i := 0; i < numCiphers; i++ {
 		cipherID := fmt.Sprintf("id-%v", i)
 		secret := fmt.Sprintf("secret-%v", i)
 		cipher, err := core.PickCipher("chacha20-ietf-poly1305", nil, secret)
 		if err != nil {
 			return nil, fmt.Errorf("Failed to create cipher %v: %v", i, err)
 		}
-		cipherList[cipherID] = cipher.(shadowaead.Cipher)
+		cipherList.PushBack(cipherID, cipher.(shadowaead.Cipher))
 	}
 	return cipherList, nil
 }