Merge branch 'main' into feature/relay-integration

signal/README.md

@@ -18,6 +18,8 @@ Flags:
       --letsencrypt-domain string   a domain to issue Let's Encrypt certificate for. Enables TLS using Let's Encrypt. Will fetch and renew certificate, and run the server with TLS
       --port int                    Server port to listen on (e.g. 10000) (default 10000)
       --ssl-dir string              server ssl directory location. *Required only for Let's Encrypt certificates. (default "/var/lib/netbird/")
+      --cert-file string            Location of your SSL certificate. Can be used when you have an existing certificate and don't want a new certificate be generated automatically. If letsencrypt-domain is specified this property has no effect
+      --cert-key string             Location of your SSL certificate private key. Can be used when you have an existing certificate and don't want a new certificate be generated automatically. If letsencrypt-domain is specified this property has no effect
 
 Global Flags:
       --log-file string    sets Netbird log path. If console is specified the the log will be output to stdout (default "/var/log/netbird/signal.log")
@@ -90,6 +92,9 @@ The Signal Server exposes the following metrics in Prometheus format:
 - **registration_delay_milliseconds**: A Histogram metric that measures the time
   it took to register a peer in
   milliseconds.
+- **get_registration_delay_milliseconds**: A Histogram metric that measures the time
+  it took to get a peer registration in
+  milliseconds.
 - **messages_forwarded_total**: A Counter metric that counts the total number of
   messages forwarded between peers.
 - **message_forward_failures_total**: A Counter metric that counts the total

signal/client/grpc.go

@@ -2,7 +2,6 @@ package client
 
 import (
 	"context"
-	"crypto/tls"
 	"fmt"
 	"io"
 	"sync"
@@ -14,9 +13,6 @@ import (
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/connectivity"
-	"google.golang.org/grpc/credentials"
-	"google.golang.org/grpc/credentials/insecure"
-	"google.golang.org/grpc/keepalive"
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/status"
 
@@ -64,28 +60,21 @@ func (c *GrpcClient) Close() error {
 
 // NewClient creates a new Signal client
 func NewClient(ctx context.Context, addr string, key wgtypes.Key, tlsEnabled bool) (*GrpcClient, error) {
+	var conn *grpc.ClientConn
 
-	transportOption := grpc.WithTransportCredentials(insecure.NewCredentials())
-
-	if tlsEnabled {
-		transportOption = grpc.WithTransportCredentials(credentials.NewTLS(&tls.Config{}))
+	operation := func() error {
+		var err error
+		conn, err = nbgrpc.CreateConnection(addr, tlsEnabled)
+		if err != nil {
+			log.Printf("createConnection error: %v", err)
+			return err
+		}
+		return nil
 	}
 
-	sigCtx, cancel := context.WithTimeout(ctx, client.ConnectTimeout)
-	defer cancel()
-	conn, err := grpc.DialContext(
-		sigCtx,
-		addr,
-		transportOption,
-		nbgrpc.WithCustomDialer(),
-		grpc.WithBlock(),
-		grpc.WithKeepaliveParams(keepalive.ClientParameters{
-			Time:    30 * time.Second,
-			Timeout: 10 * time.Second,
-		}))
-
+	err := backoff.Retry(operation, nbgrpc.Backoff(ctx))
 	if err != nil {
-		log.Errorf("failed to connect to the signalling server %v", err)
+		log.Errorf("failed to connect to the signalling server: %v", err)
 		return nil, err
 	}
 
@@ -408,7 +397,7 @@ func (c *GrpcClient) receive(stream proto.SignalExchange_ConnectStreamClient,
 
 		if err != nil {
 			log.Errorf("error while handling message of Peer [key: %s] error: [%s]", msg.Key, err.Error())
-			//todo send something??
+			// todo send something??
 		}
 	}
 }

signal/cmd/run.go

@@ -2,15 +2,12 @@ package cmd
 
 import (
 	"context"
+	"crypto/tls"
 	"errors"
 	"flag"
 	"fmt"
-	"io"
-	"io/fs"
 	"net"
 	"net/http"
-	"os"
-	"path"
 	"strings"
 	"time"
 
@@ -41,7 +38,8 @@ var (
 	signalLetsencryptDomain string
 	signalSSLDir            string
 	defaultSignalSSLDir     string
-	tlsEnabled              bool
+	signalCertFile          string
+	signalCertKey           string
 
 	signalKaep = grpc.KeepaliveEnforcementPolicy(keepalive.EnforcementPolicy{
 		MinTime:             5 * time.Second,
@@ -56,12 +54,22 @@ var (
 	})
 
 	runCmd = &cobra.Command{
-		Use:   "run",
-		Short: "start NetBird Signal Server daemon",
+		Use:          "run",
+		Short:        "start NetBird Signal Server daemon",
+		SilenceUsage: true,
 		PreRun: func(cmd *cobra.Command, args []string) {
+			err := util.InitLog(logLevel, logFile)
+			if err != nil {
+				log.Fatalf("failed initializing log %v", err)
+			}
+
+			flag.Parse()
+
 			// detect whether user specified a port
 			userPort := cmd.Flag("port").Changed
-			if signalLetsencryptDomain != "" {
+
+			tlsEnabled := false
+			if signalLetsencryptDomain != "" || (signalCertFile != "" && signalCertKey != "") {
 				tlsEnabled = true
 			}
 
@@ -77,33 +85,12 @@ var (
 		RunE: func(cmd *cobra.Command, args []string) error {
 			flag.Parse()
 
-			err := util.InitLog(logLevel, logFile)
+			opts, certManager, err := getTLSConfigurations()
 			if err != nil {
-				log.Fatalf("failed initializing log %v", err)
+				return err
 			}
 
-			if signalSSLDir == "" {
-				oldPath := "/var/lib/wiretrustee"
-				if migrateToNetbird(oldPath, defaultSignalSSLDir) {
-					if err := cpDir(oldPath, defaultSignalSSLDir); err != nil {
-						log.Fatal(err)
-					}
-				}
-			}
-
-			var opts []grpc.ServerOption
-			var certManager *autocert.Manager
-			if tlsEnabled {
-				// Let's encrypt enabled -> generate certificate automatically
-				certManager, err = encryption.CreateCertManager(signalSSLDir, signalLetsencryptDomain)
-				if err != nil {
-					return err
-				}
-				transportCredentials := credentials.NewTLS(certManager.TLSConfig())
-				opts = append(opts, grpc.Creds(transportCredentials))
-			}
-
-			metricsServer := metrics.NewServer(metricsPort, "")
+			metricsServer, err := metrics.NewServer(metricsPort, "")
 			if err != nil {
 				return fmt.Errorf("setup metrics: %v", err)
 			}
@@ -124,7 +111,25 @@ var (
 			}
 			proto.RegisterSignalExchangeServer(grpcServer, srv)
 
+			grpcRootHandler := grpcHandlerFunc(grpcServer)
+
+			if certManager != nil {
+				startServerWithCertManager(certManager, grpcRootHandler)
+			}
+
 			var compatListener net.Listener
+			var grpcListener net.Listener
+			var httpListener net.Listener
+
+			// If certManager is configured and signalPort == 443, then the gRPC server has already been started
+			if certManager == nil || signalPort != 443 {
+				grpcListener, err = serveGRPC(grpcServer, signalPort)
+				if err != nil {
+					return err
+				}
+				log.Infof("running gRPC server: %s", grpcListener.Addr().String())
+			}
+
 			if signalPort != 10000 {
 				// The Signal gRPC server was running on port 10000 previously. Old agents that are already connected to Signal
 				// are using port 10000. For compatibility purposes we keep running a 2nd gRPC server on port 10000.
@@ -135,28 +140,6 @@ var (
 				log.Infof("running gRPC backward compatibility server: %s", compatListener.Addr().String())
 			}
 
-			var grpcListener net.Listener
-			var httpListener net.Listener
-			if tlsEnabled {
-				httpListener = certManager.Listener()
-				if signalPort == 443 {
-					// running gRPC and HTTP cert manager on the same port
-					serveHTTP(httpListener, certManager.HTTPHandler(grpcHandlerFunc(grpcServer)))
-					log.Infof("running HTTP server (LetsEncrypt challenge handler) and gRPC server on the same port: %s", httpListener.Addr().String())
-				} else {
-					serveHTTP(httpListener, certManager.HTTPHandler(nil))
-					log.Infof("running HTTP server (LetsEncrypt challenge handler): %s", httpListener.Addr().String())
-				}
-			}
-
-			if signalPort != 443 || !tlsEnabled {
-				grpcListener, err = serveGRPC(grpcServer, signalPort)
-				if err != nil {
-					return err
-				}
-				log.Infof("running gRPC server: %s", grpcListener.Addr().String())
-			}
-
 			log.Infof("signal server version %s", version.NetbirdVersion())
 			log.Infof("started Signal Service")
 
@@ -190,6 +173,58 @@ var (
 	}
 )
 
+func getTLSConfigurations() ([]grpc.ServerOption, *autocert.Manager, error) {
+	var (
+		err         error
+		certManager *autocert.Manager
+		tlsConfig   *tls.Config
+	)
+
+	if signalLetsencryptDomain == "" && signalCertFile == "" && signalCertKey == "" {
+		log.Infof("running without TLS")
+		return nil, nil, nil
+	}
+
+	if signalLetsencryptDomain != "" {
+		certManager, err = encryption.CreateCertManager(signalSSLDir, signalLetsencryptDomain)
+		if err != nil {
+			return nil, certManager, err
+		}
+		tlsConfig = certManager.TLSConfig()
+		log.Infof("setting up TLS with LetsEncrypt.")
+	} else {
+		if signalCertFile == "" || signalCertKey == "" {
+			log.Errorf("both cert-file and cert-key must be provided when not using LetsEncrypt")
+			return nil, certManager, errors.New("both cert-file and cert-key must be provided when not using LetsEncrypt")
+		}
+
+		tlsConfig, err = loadTLSConfig(signalCertFile, signalCertKey)
+		if err != nil {
+			log.Errorf("cannot load TLS credentials: %v", err)
+			return nil, certManager, err
+		}
+		log.Infof("setting up TLS with custom certificates.")
+	}
+
+	transportCredentials := credentials.NewTLS(tlsConfig)
+
+	return []grpc.ServerOption{grpc.Creds(transportCredentials)}, certManager, err
+}
+
+func startServerWithCertManager(certManager *autocert.Manager, grpcRootHandler http.Handler) {
+	// a call to certManager.Listener() always creates a new listener so we do it once
+	httpListener := certManager.Listener()
+	if signalPort == 443 {
+		// running gRPC and HTTP cert manager on the same port
+		serveHTTP(httpListener, certManager.HTTPHandler(grpcRootHandler))
+		log.Infof("running HTTP server (LetsEncrypt challenge handler) and gRPC server on the same port: %s", httpListener.Addr().String())
+	} else {
+		// Start the HTTP cert manager server separately
+		serveHTTP(httpListener, certManager.HTTPHandler(nil))
+		log.Infof("running HTTP server (LetsEncrypt challenge handler): %s", httpListener.Addr().String())
+	}
+}
+
 func grpcHandlerFunc(grpcServer *grpc.Server) http.Handler {
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		grpcHeader := strings.HasPrefix(r.Header.Get("Content-Type"), "application/grpc") ||
@@ -232,95 +267,29 @@ func serveGRPC(grpcServer *grpc.Server, port int) (net.Listener, error) {
 	return listener, nil
 }
 
-func cpFile(src, dst string) error {
-	var err error
-	var srcfd *os.File
-	var dstfd *os.File
-	var srcinfo os.FileInfo
-
-	if srcfd, err = os.Open(src); err != nil {
-		return err
-	}
-	defer srcfd.Close()
-
-	if dstfd, err = os.Create(dst); err != nil {
-		return err
-	}
-	defer dstfd.Close()
-
-	if _, err = io.Copy(dstfd, srcfd); err != nil {
-		return err
-	}
-	if srcinfo, err = os.Stat(src); err != nil {
-		return err
-	}
-	return os.Chmod(dst, srcinfo.Mode())
-}
-
-func copySymLink(source, dest string) error {
-	link, err := os.Readlink(source)
+func loadTLSConfig(certFile string, certKey string) (*tls.Config, error) {
+	// Load server's certificate and private key
+	serverCert, err := tls.LoadX509KeyPair(certFile, certKey)
 	if err != nil {
-		return err
-	}
-	return os.Symlink(link, dest)
-}
-
-func cpDir(src string, dst string) error {
-	var err error
-	var fds []os.DirEntry
-	var srcinfo os.FileInfo
-
-	if srcinfo, err = os.Stat(src); err != nil {
-		return err
+		return nil, err
 	}
 
-	if err = os.MkdirAll(dst, srcinfo.Mode()); err != nil {
-		return err
+	// NewDefaultAppMetrics the credentials and return it
+	config := &tls.Config{
+		Certificates: []tls.Certificate{serverCert},
+		ClientAuth:   tls.NoClientCert,
+		NextProtos: []string{
+			"h2", "http/1.1", // enable HTTP/2
+		},
 	}
 
-	if fds, err = os.ReadDir(src); err != nil {
-		return err
-	}
-	for _, fd := range fds {
-		srcfp := path.Join(src, fd.Name())
-		dstfp := path.Join(dst, fd.Name())
-
-		fileInfo, err := os.Stat(srcfp)
-		if err != nil {
-			log.Fatalf("Couldn't get fileInfo; %v", err)
-		}
-
-		switch fileInfo.Mode() & os.ModeType {
-		case os.ModeSymlink:
-			if err = copySymLink(srcfp, dstfp); err != nil {
-				log.Fatalf("Failed to copy from %s to %s; %v", srcfp, dstfp, err)
-			}
-		case os.ModeDir:
-			if err = cpDir(srcfp, dstfp); err != nil {
-				log.Fatalf("Failed to copy from %s to %s; %v", srcfp, dstfp, err)
-			}
-		default:
-			if err = cpFile(srcfp, dstfp); err != nil {
-				log.Fatalf("Failed to copy from %s to %s; %v", srcfp, dstfp, err)
-			}
-		}
-	}
-	return nil
-}
-
-func migrateToNetbird(oldPath, newPath string) bool {
-	_, errOld := os.Stat(oldPath)
-	_, errNew := os.Stat(newPath)
-
-	if errors.Is(errOld, fs.ErrNotExist) || errNew == nil {
-		return false
-	}
-
-	return true
+	return config, nil
 }
 
 func init() {
 	runCmd.PersistentFlags().IntVar(&signalPort, "port", 80, "Server port to listen on (defaults to 443 if TLS is enabled, 80 otherwise")
 	runCmd.Flags().StringVar(&signalSSLDir, "ssl-dir", defaultSignalSSLDir, "server ssl directory location. *Required only for Let's Encrypt certificates.")
 	runCmd.Flags().StringVar(&signalLetsencryptDomain, "letsencrypt-domain", "", "a domain to issue Let's Encrypt certificate for. Enables TLS using Let's Encrypt. Will fetch and renew certificate, and run the server with TLS")
+	runCmd.Flags().StringVar(&signalCertFile, "cert-file", "", "Location of your SSL certificate. Can be used when you have an existing certificate and don't want a new certificate be generated automatically. If letsencrypt-domain is specified this property has no effect")
+	runCmd.Flags().StringVar(&signalCertKey, "cert-key", "", "Location of your SSL certificate private key. Can be used when you have an existing certificate and don't want a new certificate be generated automatically. If letsencrypt-domain is specified this property has no effect")
 }

signal/metrics/app.go

@@ -15,6 +15,7 @@ type AppMetrics struct {
 	Deregistrations      metric.Int64Counter
 	RegistrationFailures metric.Int64Counter
 	RegistrationDelay    metric.Float64Histogram
+	GetRegistrationDelay metric.Float64Histogram
 
 	MessagesForwarded      metric.Int64Counter
 	MessageForwardFailures metric.Int64Counter
@@ -54,6 +55,12 @@ func NewAppMetrics(meter metric.Meter) (*AppMetrics, error) {
 		return nil, err
 	}
 
+	getRegistrationDelay, err := meter.Float64Histogram("get_registration_delay_milliseconds",
+		metric.WithExplicitBucketBoundaries(getStandardBucketBoundaries()...))
+	if err != nil {
+		return nil, err
+	}
+
 	messagesForwarded, err := meter.Int64Counter("messages_forwarded_total")
 	if err != nil {
 		return nil, err
@@ -80,6 +87,7 @@ func NewAppMetrics(meter metric.Meter) (*AppMetrics, error) {
 		Deregistrations:      deregistrations,
 		RegistrationFailures: registrationFailures,
 		RegistrationDelay:    registrationDelay,
+		GetRegistrationDelay: getRegistrationDelay,
 
 		MessagesForwarded:      messagesForwarded,
 		MessageForwardFailures: messageForwardFailures,

signal/metrics/metrics.go

@@ -26,10 +26,10 @@ type Metrics struct {
 }
 
 // NewServer initializes and returns a new Metrics instance
-func NewServer(port int, endpoint string) *Metrics {
+func NewServer(port int, endpoint string) (*Metrics, error) {
 	exporter, err := prometheus.New()
 	if err != nil {
-		return nil
+		return nil, err
 	}
 
 	provider := metric.NewMeterProvider(metric.WithReader(exporter))
@@ -57,7 +57,7 @@ func NewServer(port int, endpoint string) *Metrics {
 		provider: provider,
 		Endpoint: endpoint,
 		Server:   server,
-	}
+	}, nil
 }
 
 // Shutdown stops the metrics server

signal/server/signal.go

@@ -23,11 +23,17 @@ const (
 	labelTypeError         = "error"
 	labelTypeNotConnected  = "not_connected"
 	labelTypeNotRegistered = "not_registered"
+	labelTypeStream        = "stream"
+	labelTypeMessage       = "message"
 
 	labelError             = "error"
 	labelErrorMissingId    = "missing_id"
 	labelErrorMissingMeta  = "missing_meta"
 	labelErrorFailedHeader = "failed_header"
+
+	labelRegistrationStatus   = "status"
+	labelRegistrationFound    = "found"
+	labelRegistrationNotFound = "not_found"
 )
 
 // Server an instance of a Signal server
@@ -61,7 +67,11 @@ func (s *Server) Send(ctx context.Context, msg *proto.EncryptedMessage) (*proto.
 		return nil, fmt.Errorf("peer %s is not registered", msg.Key)
 	}
 
+	getRegistrationStart := time.Now()
+
 	if dstPeer, found := s.registry.Get(msg.RemoteKey); found {
+		s.metrics.GetRegistrationDelay.Record(ctx, float64(time.Since(getRegistrationStart).Nanoseconds())/1e6, metric.WithAttributes(attribute.String(labelType, labelTypeMessage), attribute.String(labelRegistrationStatus, labelRegistrationFound)))
+		start := time.Now()
 		//forward the message to the target peer
 		if err := dstPeer.Stream.Send(msg); err != nil {
 			log.Errorf("error while forwarding message from peer [%s] to peer [%s] %v", msg.Key, msg.RemoteKey, err)
@@ -69,9 +79,11 @@ func (s *Server) Send(ctx context.Context, msg *proto.EncryptedMessage) (*proto.
 
 			s.metrics.MessageForwardFailures.Add(ctx, 1, metric.WithAttributes(attribute.String(labelType, labelTypeError)))
 		} else {
+			s.metrics.MessageForwardLatency.Record(ctx, float64(time.Since(start).Nanoseconds())/1e6, metric.WithAttributes(attribute.String(labelType, labelTypeMessage)))
 			s.metrics.MessagesForwarded.Add(context.Background(), 1)
 		}
 	} else {
+		s.metrics.GetRegistrationDelay.Record(ctx, float64(time.Since(getRegistrationStart).Nanoseconds())/1e6, metric.WithAttributes(attribute.String(labelType, labelTypeMessage), attribute.String(labelRegistrationStatus, labelRegistrationNotFound)))
 		log.Debugf("message from peer [%s] can't be forwarded to peer [%s] because destination peer is not connected", msg.Key, msg.RemoteKey)
 		//todo respond to the sender?
 
@@ -118,28 +130,30 @@ func (s *Server) ConnectStream(stream proto.SignalExchange_ConnectStreamServer)
 		} else if err != nil {
 			return err
 		}
-		start := time.Now()
 
 		log.Debugf("received a new message from peer [%s] to peer [%s]", p.Id, msg.RemoteKey)
 
+		getRegistrationStart := time.Now()
+
 		// lookup the target peer where the message is going to
 		if dstPeer, found := s.registry.Get(msg.RemoteKey); found {
+			s.metrics.GetRegistrationDelay.Record(stream.Context(), float64(time.Since(getRegistrationStart).Nanoseconds())/1e6, metric.WithAttributes(attribute.String(labelType, labelTypeStream), attribute.String(labelRegistrationStatus, labelRegistrationFound)))
+			start := time.Now()
 			//forward the message to the target peer
 			if err := dstPeer.Stream.Send(msg); err != nil {
 				log.Errorf("error while forwarding message from peer [%s] to peer [%s] %v", p.Id, msg.RemoteKey, err)
 				//todo respond to the sender?
-
-				// in milliseconds
-				s.metrics.MessageForwardLatency.Record(stream.Context(), float64(time.Since(start).Nanoseconds())/1e6)
-				s.metrics.MessagesForwarded.Add(stream.Context(), 1)
-			} else {
 				s.metrics.MessageForwardFailures.Add(stream.Context(), 1, metric.WithAttributes(attribute.String(labelType, labelTypeError)))
+			} else {
+				// in milliseconds
+				s.metrics.MessageForwardLatency.Record(stream.Context(), float64(time.Since(start).Nanoseconds())/1e6, metric.WithAttributes(attribute.String(labelType, labelTypeStream)))
+				s.metrics.MessagesForwarded.Add(stream.Context(), 1)
 			}
 		} else {
+			s.metrics.GetRegistrationDelay.Record(stream.Context(), float64(time.Since(getRegistrationStart).Nanoseconds())/1e6, metric.WithAttributes(attribute.String(labelType, labelTypeStream), attribute.String(labelRegistrationStatus, labelRegistrationNotFound)))
+			s.metrics.MessageForwardFailures.Add(stream.Context(), 1, metric.WithAttributes(attribute.String(labelType, labelTypeNotConnected)))
 			log.Debugf("message from peer [%s] can't be forwarded to peer [%s] because destination peer is not connected", p.Id, msg.RemoteKey)
 			//todo respond to the sender?
-
-			s.metrics.MessageForwardFailures.Add(stream.Context(), 1, metric.WithAttributes(attribute.String(labelType, labelTypeNotConnected)))
 		}
 	}
 	<-stream.Context().Done()