Merge pull request #109 from getlantern/myles/tracing

Add basic tracing

Author: noahlevenson

egress/egress.go

@@ -18,8 +18,14 @@ import (
 
 	"github.com/elazarl/goproxy"
 	"github.com/getlantern/broflake/common"
+	"github.com/getlantern/telemetry"
 	"github.com/google/uuid"
 	"github.com/lucas-clemente/quic-go"
+	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/metric/global"
+	"go.opentelemetry.io/otel/metric/instrument"
+	"go.opentelemetry.io/otel/trace"
 	"nhooyr.io/websocket"
 )
 
@@ -29,6 +35,12 @@ import (
 var nClients uint64
 var nQUICStreams uint64
 
+// TODO: it'd be more elegant to use observers rather than counters, such that we could simply
+// observe the value of nClients and nQUICStreams instead of duplicating the increment/decrement
+// operations. However, the otel observer API seems more complicated than it's worth?
+var nClientsCounter instrument.Int64UpDownCounter
+var nQUICStreamsCounter instrument.Int64UpDownCounter
+
 // webSocketPacketConn wraps a websocket.Conn as a net.PacketConn
 type websocketPacketConn struct {
 	net.PacketConn
@@ -49,6 +61,7 @@ func (q websocketPacketConn) WriteTo(p []byte, addr net.Addr) (n int, err error)
 
 func (q websocketPacketConn) Close() error {
 	defer log.Printf("Closed a WebSocket connection! (%v total)\n", atomic.AddUint64(&nClients, ^uint64(0)))
+	defer nClientsCounter.Add(context.Background(), -1)
 	return q.w.Close(websocket.StatusNormalClosure, "")
 }
 
@@ -60,6 +73,7 @@ type proxyListener struct {
 	net.Listener
 	connections chan net.Conn
 	tlsConfig   *tls.Config
+	tracer      trace.Tracer
 }
 
 func (l proxyListener) Accept() (net.Conn, error) {
@@ -71,7 +85,6 @@ func (l proxyListener) Addr() net.Addr {
 	return common.DebugAddr("DEBUG NELSON WUZ HERE")
 }
 
-// TODO: Someone should scrutinize this
 func generateTLSConfig() *tls.Config {
 	key, err := rsa.GenerateKey(rand.Reader, 1024)
 	if err != nil {
@@ -100,6 +113,9 @@ func (l proxyListener) handleWebsocket(w http.ResponseWriter, r *http.Request) {
 	// patterns as strings using AcceptOptions.OriginPattern
 	// TODO: disabling compression is a workaround for a WebKit bug:
 	// https://github.com/getlantern/broflake/issues/45
+	ctx := r.Context()
+	ctx, span := l.tracer.Start(ctx, "handleWebsocket")
+	defer span.End()
 	c, err := websocket.Accept(
 		w,
 		r,
@@ -117,23 +133,25 @@ func (l proxyListener) handleWebsocket(w http.ResponseWriter, r *http.Request) {
 	defer wspconn.Close()
 
 	if err != nil {
-		// TODO: this is the idiom for our WebSocket library, but we should log the err better
-		log.Println(err)
+		span.RecordError(err)
 		return
 	}
 
 	log.Printf("Accepted a new WebSocket connection! (%v total)\n", atomic.AddUint64(&nClients, 1))
+	nClientsCounter.Add(r.Context(), 1)
 
 	listener, err := quic.Listen(wspconn, l.tlsConfig, &common.QUICCfg)
 	if err != nil {
 		log.Printf("Error creating QUIC listener: %v\n", err)
+		span.RecordError(err)
 		return
 	}
 
 	for {
 		conn, err := listener.Accept(context.Background())
 		if err != nil {
 			log.Printf("%v QUIC listener error (%v), closing!\n", wspconn.addr, err)
+			span.RecordError(err)
 			listener.Close()
 			break
 		}
@@ -154,21 +172,44 @@ func (l proxyListener) handleWebsocket(w http.ResponseWriter, r *http.Request) {
 				}
 
 				log.Printf("Accepted a new QUIC stream! (%v total)\n", atomic.AddUint64(&nQUICStreams, 1))
+				nQUICStreamsCounter.Add(r.Context(), 1)
 
 				l.connections <- common.QUICStreamNetConn{Stream: stream, OnClose: func() {
 					defer log.Printf("Closed a QUIC stream! (%v total)\n", atomic.AddUint64(&nQUICStreams, ^uint64(0)))
+					nQUICStreamsCounter.Add(r.Context(), -1)
 				}}
 			}
 		}()
 	}
 }
 
 func main() {
+	ctx := context.Background()
+	closeFuncTrace := telemetry.EnableOTELTracing(ctx)
+	closeFuncMetric := telemetry.EnableOTELMetrics(ctx)
+	defer func() {
+		_ = closeFuncTrace(ctx)
+		_ = closeFuncMetric(ctx)
+	}()
+
+	m := global.Meter("github.com/getlantern/broflake/egress")
+	var err error
+	nClientsCounter, err = m.Int64UpDownCounter("websocket-counter")
+	if err != nil {
+		panic(err)
+	}
+
+	nQUICStreamsCounter, err = m.Int64UpDownCounter("quic-stream-counter")
+	if err != nil {
+		panic(err)
+	}
+
 	// We use this wrapped listener to enable our local HTTP proxy to listen for WebSocket connections
 	l := proxyListener{
 		Listener:    &net.TCPListener{},
 		connections: make(chan net.Conn, 2048),
 		tlsConfig:   generateTLSConfig(),
+		tracer:      otel.Tracer("websocket-tracer"),
 	}
 
 	// Instantiate our local HTTP CONNECT proxy
@@ -213,9 +254,9 @@ func main() {
 		Addr:         fmt.Sprintf(":%v", port),
 	}
 
-	http.HandleFunc("/ws", l.handleWebsocket)
+	http.Handle("/ws", otelhttp.NewHandler(http.HandlerFunc(l.handleWebsocket), "/ws"))
 	log.Printf("Egress server listening for WebSocket connections on %v\n\n", srv.Addr)
-	err := srv.ListenAndServe()
+	err = srv.ListenAndServe()
 	if err != nil {
 		log.Println(err)
 	}

egress/go.mod

@@ -6,14 +6,39 @@ require (
 	github.com/elazarl/goproxy v0.0.0-20221015165544-a0805db90819
 	github.com/getlantern/broflake/common v0.0.0-00010101000000-000000000000
 	github.com/lucas-clemente/quic-go v0.31.1
+	go.opentelemetry.io/otel v1.13.0
+	go.opentelemetry.io/otel/trace v1.13.0
 	nhooyr.io/websocket v1.8.7
 )
 
 require (
+	github.com/cenkalti/backoff/v4 v4.2.0 // indirect
+	github.com/felixge/httpsnoop v1.0.3 // indirect
+	github.com/go-logr/logr v1.2.3 // indirect
+	github.com/go-logr/stdr v1.2.2 // indirect
+	github.com/golang/protobuf v1.5.2 // indirect
+	github.com/grpc-ecosystem/grpc-gateway/v2 v2.7.0 // indirect
+	go.opentelemetry.io/otel/exporters/otlp/internal/retry v1.13.0 // indirect
+	go.opentelemetry.io/otel/exporters/otlp/otlpmetric v0.36.0 // indirect
+	go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp v0.36.0 // indirect
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.12.0 // indirect
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.12.0 // indirect
+	go.opentelemetry.io/otel/metric v0.36.0 // indirect
+	go.opentelemetry.io/otel/sdk v1.13.0 // indirect
+	go.opentelemetry.io/otel/sdk/metric v0.36.0 // indirect
+	go.opentelemetry.io/proto/otlp v0.19.0 // indirect
+	golang.org/x/text v0.5.0 // indirect
+	google.golang.org/genproto v0.0.0-20221118155620-16455021b5e6 // indirect
+	google.golang.org/grpc v1.52.3 // indirect
+	google.golang.org/protobuf v1.28.1 // indirect
+)
+
+require (
+	github.com/getlantern/telemetry v0.0.0-20230227190802-faa666d3b3d5
 	github.com/go-task/slim-sprig v0.0.0-20210107165309-348f09dbbbc0 // indirect
 	github.com/golang/mock v1.6.0 // indirect
 	github.com/google/pprof v0.0.0-20210407192527-94a9f03dee38 // indirect
-	github.com/google/uuid v1.3.0 // indirect
+	github.com/google/uuid v1.3.0
 	github.com/klauspost/compress v1.10.3 // indirect
 	github.com/marten-seemann/qtls-go1-18 v0.1.3 // indirect
 	github.com/marten-seemann/qtls-go1-19 v0.1.1 // indirect
@@ -35,10 +60,11 @@ require (
 	github.com/pion/turn/v2 v2.0.8 // indirect
 	github.com/pion/udp v0.1.1 // indirect
 	github.com/pion/webrtc/v3 v3.1.50 // indirect
+	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.39.0
 	golang.org/x/crypto v0.0.0-20221010152910-d6f0a8c073c2 // indirect
 	golang.org/x/exp v0.0.0-20220722155223-a9213eeb770e // indirect
 	golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4 // indirect
-	golang.org/x/net v0.3.0 // indirect
+	golang.org/x/net v0.4.0 // indirect
 	golang.org/x/sys v0.3.0 // indirect
 	golang.org/x/tools v0.1.12 // indirect
 )