Reworked base

2026-09-01 10:44:02 +03:00 · 2019-08-29 13:16:08 +03:00
parent 09c7ce45d2
commit 07985cf418
26 changed files with 1481 additions and 271 deletions
@@ -0,0 +1,93 @@
+package newstats
+
+import (
+	"net"
+	"net/http"
+
+	"github.com/juju/errors"
+
+	"github.com/9seconds/mtg/newconfig"
+	"github.com/9seconds/mtg/newprotocol"
+)
+
+type Stats interface {
+	IngressTraffic(int)
+	EgressTraffic(int)
+	ClientConnected(newprotocol.ConnectionType, *net.TCPAddr)
+	ClientDisconnected(newprotocol.ConnectionType, *net.TCPAddr)
+	Crash()
+	AntiReplayDetected()
+}
+
+type multiStats []Stats
+
+func (m multiStats) IngressTraffic(traffic int) {
+	for i := range m {
+		go m[i].IngressTraffic(traffic)
+	}
+}
+
+func (m multiStats) EgressTraffic(traffic int) {
+	for i := range m {
+		go m[i].EgressTraffic(traffic)
+	}
+}
+
+func (m multiStats) ClientConnected(connectionType newprotocol.ConnectionType, addr *net.TCPAddr) {
+	for i := range m {
+		go m[i].ClientConnected(connectionType, addr)
+	}
+}
+
+func (m multiStats) ClientDisconnected(connectionType newprotocol.ConnectionType, addr *net.TCPAddr) {
+	for i := range m {
+		go m[i].ClientDisconnected(connectionType, addr)
+	}
+}
+
+func (m multiStats) Crash() {
+	for i := range m {
+		go m[i].Crash()
+	}
+}
+
+func (m multiStats) AntiReplayDetected() {
+	for i := range m {
+		go m[i].AntiReplayDetected()
+	}
+}
+
+var S Stats
+
+func Init() error {
+	mux := http.NewServeMux()
+
+	instanceJSON := newStatsJSON(mux)
+	instancePrometheus, err := newStatsPrometheus(mux)
+	if err != nil {
+		return errors.Annotate(err, "Cannot initialize Prometheus")
+	}
+
+	stats := []Stats{instanceJSON, instancePrometheus}
+	if newconfig.C.StatsdStats.Addr.IP != nil {
+		instanceStatsd, err := newStatsStatsd()
+		if err != nil {
+			return errors.Annotate(err, "Cannot initialize StatsD")
+		}
+		stats = append(stats, instanceStatsd)
+	}
+
+	listener, err := net.Listen("tcp", newconfig.C.StatsAddr.String())
+	if err != nil {
+		return errors.Annotate(err, "Cannot initialize stats server")
+	}
+
+	srv := http.Server{
+		Handler: mux,
+	}
+	go srv.Serve(listener) // nolint: errcheck
+
+	S = multiStats(stats)
+
+	return nil
+}
@@ -0,0 +1,127 @@
+package newstats
+
+import (
+	"encoding/json"
+	"net"
+	"net/http"
+	"sync/atomic"
+	"time"
+
+	"go.uber.org/zap"
+
+	"github.com/9seconds/mtg/newprotocol"
+)
+
+type statsJSON struct {
+	Connections statsJSONConnections `json:"connections"`
+	Traffic     statsJSONTraffic     `json:"traffic"`
+	Uptime      statsJSONUptime      `json:"uptime"`
+	Crashes     uint32               `json:"crashes"`
+	AntiReplays uint32               `json:"anti_replay_detected"`
+}
+
+type statsBaseJSONConnections struct {
+	All          statsJSONConnectionType `json:"all"`
+	Abridged     statsJSONConnectionType `json:"abridged"`
+	Intermediate statsJSONConnectionType `json:"intermediate"`
+	Secured      statsJSONConnectionType `json:"secured"`
+}
+
+type statsJSONConnections struct {
+	statsBaseJSONConnections
+}
+
+type statsJSONConnectionType struct {
+	IPv4 uint32 `json:"ipv4"`
+	IPv6 uint32 `json:"ipv6"`
+}
+
+func (c statsJSONConnections) MarshalJSON() ([]byte, error) {
+	c.All.IPv4 = c.Abridged.IPv4 + c.Intermediate.IPv4 + c.Secured.IPv4
+	c.All.IPv6 = c.Abridged.IPv6 + c.Intermediate.IPv6 + c.Secured.IPv6
+
+	return json.Marshal(c.statsBaseJSONConnections)
+}
+
+type statsJSONTraffic struct {
+	Ingress uint64 `json:"ingress"`
+	Egress  uint64 `json:"egress"`
+}
+
+type statsJSONUptime time.Time
+
+func (s statsJSONUptime) MarshalJSON() ([]byte, error) {
+	return json.Marshal(time.Since(time.Time(s)).Seconds())
+}
+
+func (s *statsJSON) IngressTraffic(traffic int) {
+	atomic.AddUint64(&s.Traffic.Ingress, uint64(traffic))
+}
+
+func (s *statsJSON) EgressTraffic(traffic int) {
+	atomic.AddUint64(&s.Traffic.Egress, uint64(traffic))
+}
+
+func (s *statsJSON) ClientConnected(connectionType newprotocol.ConnectionType, addr *net.TCPAddr) {
+	s.changeConnections(connectionType, addr, 1)
+}
+
+func (s *statsJSON) ClientDisconnected(connectionType newprotocol.ConnectionType, addr *net.TCPAddr) {
+	s.changeConnections(connectionType, addr, ^uint32(0))
+}
+
+func (s *statsJSON) changeConnections(connectionType newprotocol.ConnectionType, addr *net.TCPAddr, value uint32) {
+	var connections *statsJSONConnectionType
+
+	switch connectionType {
+	case newprotocol.ConnectionTypeAbridged:
+		connections = &s.Connections.Abridged
+	case newprotocol.ConnectionTypeSecure:
+		connections = &s.Connections.Secured
+	default:
+		connections = &s.Connections.Intermediate
+	}
+
+	if addr.IP.To4() == nil {
+		atomic.AddUint32(&connections.IPv4, value)
+	} else {
+		atomic.AddUint32(&connections.IPv6, value)
+	}
+}
+
+func (s *statsJSON) Crash() {
+	atomic.AddUint32(&s.Crashes, 1)
+}
+
+func (s *statsJSON) AntiReplayDetected() {
+	atomic.AddUint32(&s.AntiReplays, 1)
+}
+
+func newStatsJSON(mux *http.ServeMux) Stats {
+	instance := &statsJSON{}
+	logger := zap.S().Named("stats")
+
+	mux.HandleFunc("/", func(w http.ResponseWriter, _ *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		first, err := json.Marshal(instance)
+		if err != nil {
+			logger.Errorw("Cannot encode json", "error", err)
+			http.Error(w, "Internal server error", http.StatusServiceUnavailable)
+			return
+		}
+
+		interim := map[string]interface{}{}
+		if err := json.Unmarshal(first, &interim); err != nil {
+			panic(err)
+		}
+
+		encoder := json.NewEncoder(w)
+		encoder.SetEscapeHTML(false)
+		encoder.SetIndent("", "  ")
+		if err := encoder.Encode(interim); err != nil {
+			logger.Errorw("Cannot encode json", "error", err)
+		}
+	})
+
+	return instance
+}
@@ -0,0 +1,110 @@
+package newstats
+
+import (
+	"net"
+	"net/http"
+
+	"github.com/juju/errors"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+
+	"github.com/9seconds/mtg/newconfig"
+	"github.com/9seconds/mtg/newprotocol"
+)
+
+type statsPrometheus struct {
+	connections *prometheus.GaugeVec
+	traffic     *prometheus.GaugeVec
+	crashes     prometheus.Gauge
+	antiReplays prometheus.Gauge
+}
+
+func (s *statsPrometheus) IngressTraffic(traffic int) {
+	s.traffic.WithLabelValues("ingress").Add(float64(traffic))
+}
+
+func (s *statsPrometheus) EgressTraffic(traffic int) {
+	s.traffic.WithLabelValues("egress").Add(float64(traffic))
+}
+
+func (s *statsPrometheus) ClientConnected(connectionType newprotocol.ConnectionType, addr *net.TCPAddr) {
+	s.changeConnections(connectionType, addr, 1.0)
+}
+
+func (s *statsPrometheus) ClientDisconnected(connectionType newprotocol.ConnectionType, addr *net.TCPAddr) {
+	s.changeConnections(connectionType, addr, -1.0)
+}
+
+func (s *statsPrometheus) changeConnections(connectionType newprotocol.ConnectionType,
+	addr *net.TCPAddr,
+	increment float64) {
+	var labels [2]string
+
+	switch connectionType {
+	case newprotocol.ConnectionTypeAbridged:
+		labels[0] = "abridged"
+	case newprotocol.ConnectionTypeSecure:
+		labels[0] = "secured"
+	default:
+		labels[0] = "intermediate"
+	}
+
+	labels[1] = "ipv4"
+	if addr.IP.To4() == nil {
+		labels[1] = "ipv6"
+	}
+
+	s.connections.WithLabelValues(labels[:]...).Add(increment)
+}
+
+func (s *statsPrometheus) Crash() {
+	s.crashes.Inc()
+}
+
+func (s *statsPrometheus) AntiReplayDetected() {
+	s.antiReplays.Inc()
+}
+
+func newStatsPrometheus(mux *http.ServeMux) (Stats, error) {
+	registry := prometheus.NewRegistry()
+	instance := &statsPrometheus{
+		connections: prometheus.NewGaugeVec(prometheus.GaugeOpts{
+			Namespace: newconfig.C.PrometheusStats.Prefix,
+			Name:      "connections",
+			Help:      "Current number of connections to the proxy.",
+		}, []string{"type", "protocol"}),
+		traffic: prometheus.NewGaugeVec(prometheus.GaugeOpts{
+			Namespace: newconfig.C.PrometheusStats.Prefix,
+			Name:      "traffic",
+			Help:      "Traffic passed through the proxy in bytes.",
+		}, []string{"direction"}),
+		crashes: prometheus.NewGauge(prometheus.GaugeOpts{
+			Namespace: newconfig.C.PrometheusStats.Prefix,
+			Name:      "crashes",
+			Help:      "How many crashes happened.",
+		}),
+		antiReplays: prometheus.NewGauge(prometheus.GaugeOpts{
+			Namespace: newconfig.C.PrometheusStats.Prefix,
+			Name:      "anti_replays",
+			Help:      "How many anti replay attacks were prevented.",
+		}),
+	}
+
+	if err := registry.Register(instance.connections); err != nil {
+		return nil, errors.Annotate(err, "Cannot register metrics for connections")
+	}
+	if err := registry.Register(instance.traffic); err != nil {
+		return nil, errors.Annotate(err, "Cannot register metrics for traffic")
+	}
+	if err := registry.Register(instance.crashes); err != nil {
+		return nil, errors.Annotate(err, "Cannot register metrics for crashes")
+	}
+	if err := registry.Register(instance.antiReplays); err != nil {
+		return nil, errors.Annotate(err, "Cannot register metrics for anti replays")
+	}
+
+	handler := promhttp.HandlerFor(registry, promhttp.HandlerOpts{})
+	mux.Handle("/prometheus", handler)
+
+	return instance, nil
+}
@@ -0,0 +1,87 @@
+package newstats
+
+import (
+	"net"
+	"strings"
+
+	"gopkg.in/alexcesaro/statsd.v2"
+
+	"github.com/9seconds/mtg/newconfig"
+	"github.com/9seconds/mtg/newprotocol"
+	"github.com/juju/errors"
+)
+
+type statsStatsd struct {
+	client *statsd.Client
+}
+
+func (s *statsStatsd) IngressTraffic(traffic int) {
+	s.client.Count("traffic.ingress", traffic)
+}
+
+func (s *statsStatsd) EgressTraffic(traffic int) {
+	s.client.Count("traffic.egress", traffic)
+}
+
+func (s *statsStatsd) ClientConnected(connectionType newprotocol.ConnectionType, addr *net.TCPAddr) {
+	s.changeConnections(connectionType, addr, 1)
+}
+
+func (s *statsStatsd) ClientDisconnected(connectionType newprotocol.ConnectionType, addr *net.TCPAddr) {
+	s.changeConnections(connectionType, addr, -1)
+}
+
+func (s *statsStatsd) changeConnections(connectionType newprotocol.ConnectionType, addr *net.TCPAddr, value int) {
+	var labels [3]string
+
+	labels[0] = "connections"
+	switch connectionType {
+	case newprotocol.ConnectionTypeAbridged:
+		labels[1] = "abridged"
+	case newprotocol.ConnectionTypeSecure:
+		labels[1] = "secured"
+	default:
+		labels[1] = "intermediate"
+	}
+
+	labels[2] = "ipv4"
+	if addr.IP.To4() == nil {
+		labels[2] = "ipv6"
+	}
+
+	s.client.Count(strings.Join(labels[:], "."), value)
+}
+
+func (s *statsStatsd) Crash() {
+	s.client.Increment("crashes")
+}
+
+func (s *statsStatsd) AntiReplayDetected() {
+	s.client.Increment("anti_replays")
+}
+
+func newStatsStatsd() (Stats, error) {
+	options := []statsd.Option{
+		statsd.Prefix(newconfig.C.StatsdStats.Prefix),
+		statsd.Network(newconfig.C.StatsdStats.Addr.Network()),
+		statsd.Address(newconfig.C.StatsdStats.Addr.String()),
+		statsd.TagsFormat(newconfig.C.StatsdStats.TagsFormat),
+	}
+
+	if len(newconfig.C.StatsdStats.Tags) > 0 {
+		tags := make([]string, len(newconfig.C.StatsdStats.Tags)*2)
+		for k, v := range newconfig.C.StatsdStats.Tags {
+			tags = append(tags, k, v)
+		}
+		options = append(options, statsd.Tags(tags...))
+	}
+
+	client, err := statsd.New(options...)
+	if err != nil {
+		return nil, errors.Annotate(err, "Cannot initialize a client")
+	}
+
+	return &statsStatsd{
+		client: client,
+	}, nil
+}