mirror of
https://github.com/slackhq/nebula.git
synced 2026-08-15 19:47:01 +02:00
Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 3cdb94b2a0 |
@@ -7,7 +7,6 @@ require (
|
||||
filippo.io/bigmod v0.1.0
|
||||
github.com/anmitsu/go-shlex v0.0.0-20200514113438-38f4b401e2be
|
||||
github.com/armon/go-radix v1.0.0
|
||||
github.com/cyberdelia/go-metrics-graphite v0.0.0-20161219230853-39f87cc3b432
|
||||
github.com/flynn/noise v1.1.0
|
||||
github.com/gaissmai/bart v0.28.0
|
||||
github.com/gogo/protobuf v1.3.2
|
||||
|
||||
@@ -19,8 +19,6 @@ github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6r
|
||||
github.com/cespare/xxhash/v2 v2.1.1/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
|
||||
github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
|
||||
github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
|
||||
github.com/cyberdelia/go-metrics-graphite v0.0.0-20161219230853-39f87cc3b432 h1:M5QgkYacWj0Xs8MhpIK/5uwU02icXpEoSo9sM2aRCps=
|
||||
github.com/cyberdelia/go-metrics-graphite v0.0.0-20161219230853-39f87cc3b432/go.mod h1:xwIwAxMvYnVrGJPe2FKx5prTrnAjGOD8zvDOnxnrrkM=
|
||||
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
|
||||
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
|
||||
+117
@@ -0,0 +1,117 @@
|
||||
package nebula
|
||||
|
||||
// This file is a trimmed, inlined copy of the graphite exporter from
|
||||
// github.com/cyberdelia/go-metrics-graphite, retaining only the Config type and
|
||||
// the Once entrypoint that Nebula uses. The upstream package has been
|
||||
// unmaintained for 10+ years, so it was vendored here to drop the dependency.
|
||||
// See https://github.com/slackhq/nebula/issues/1831.
|
||||
//
|
||||
// Copyright 2015 Timothée Peignier. All rights reserved.
|
||||
//
|
||||
// Redistribution and use in source and binary forms, with or without
|
||||
// modification, are permitted provided that the following conditions are met:
|
||||
//
|
||||
// 1. Redistributions of source code must retain the above copyright notice,
|
||||
// this list of conditions and the following disclaimer.
|
||||
//
|
||||
// 2. Redistributions in binary form must reproduce the above copyright notice,
|
||||
// this list of conditions and the following disclaimer in the documentation
|
||||
// and/or other materials provided with the distribution.
|
||||
//
|
||||
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
|
||||
// AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
|
||||
// IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
|
||||
// DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
|
||||
// FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
|
||||
// DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
|
||||
// SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
|
||||
// CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
|
||||
// OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
|
||||
// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"fmt"
|
||||
"net"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/rcrowley/go-metrics"
|
||||
)
|
||||
|
||||
// graphiteConfigExport provides a container with configuration parameters for
|
||||
// the Graphite exporter.
|
||||
type graphiteConfigExport struct {
|
||||
Addr *net.TCPAddr // Network address to connect to
|
||||
Registry metrics.Registry // Registry to be exported
|
||||
FlushInterval time.Duration // Flush interval
|
||||
DurationUnit time.Duration // Time conversion unit for durations
|
||||
Prefix string // Prefix to be prepended to metric names
|
||||
Percentiles []float64 // Percentiles to export from timers and histograms
|
||||
}
|
||||
|
||||
// graphiteOnce performs a single submission to Graphite, returning a non-nil
|
||||
// error on failed connections.
|
||||
func graphiteOnce(c graphiteConfigExport) error {
|
||||
now := time.Now().Unix()
|
||||
du := float64(c.DurationUnit)
|
||||
flushSeconds := float64(c.FlushInterval) / float64(time.Second)
|
||||
conn, err := net.DialTCP("tcp", nil, c.Addr)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer conn.Close()
|
||||
w := bufio.NewWriter(conn)
|
||||
c.Registry.Each(func(name string, i any) {
|
||||
switch metric := i.(type) {
|
||||
case metrics.Counter:
|
||||
count := metric.Count()
|
||||
fmt.Fprintf(w, "%s.%s.count %d %d\n", c.Prefix, name, count, now)
|
||||
fmt.Fprintf(w, "%s.%s.count_ps %.2f %d\n", c.Prefix, name, float64(count)/flushSeconds, now)
|
||||
case metrics.Gauge:
|
||||
fmt.Fprintf(w, "%s.%s.value %d %d\n", c.Prefix, name, metric.Value(), now)
|
||||
case metrics.GaugeFloat64:
|
||||
fmt.Fprintf(w, "%s.%s.value %f %d\n", c.Prefix, name, metric.Value(), now)
|
||||
case metrics.Histogram:
|
||||
h := metric.Snapshot()
|
||||
ps := h.Percentiles(c.Percentiles)
|
||||
fmt.Fprintf(w, "%s.%s.count %d %d\n", c.Prefix, name, h.Count(), now)
|
||||
fmt.Fprintf(w, "%s.%s.min %d %d\n", c.Prefix, name, h.Min(), now)
|
||||
fmt.Fprintf(w, "%s.%s.max %d %d\n", c.Prefix, name, h.Max(), now)
|
||||
fmt.Fprintf(w, "%s.%s.mean %.2f %d\n", c.Prefix, name, h.Mean(), now)
|
||||
fmt.Fprintf(w, "%s.%s.std-dev %.2f %d\n", c.Prefix, name, h.StdDev(), now)
|
||||
for psIdx, psKey := range c.Percentiles {
|
||||
key := strings.Replace(strconv.FormatFloat(psKey*100.0, 'f', -1, 64), ".", "", 1)
|
||||
fmt.Fprintf(w, "%s.%s.%s-percentile %.2f %d\n", c.Prefix, name, key, ps[psIdx], now)
|
||||
}
|
||||
case metrics.Meter:
|
||||
m := metric.Snapshot()
|
||||
fmt.Fprintf(w, "%s.%s.count %d %d\n", c.Prefix, name, m.Count(), now)
|
||||
fmt.Fprintf(w, "%s.%s.one-minute %.2f %d\n", c.Prefix, name, m.Rate1(), now)
|
||||
fmt.Fprintf(w, "%s.%s.five-minute %.2f %d\n", c.Prefix, name, m.Rate5(), now)
|
||||
fmt.Fprintf(w, "%s.%s.fifteen-minute %.2f %d\n", c.Prefix, name, m.Rate15(), now)
|
||||
fmt.Fprintf(w, "%s.%s.mean %.2f %d\n", c.Prefix, name, m.RateMean(), now)
|
||||
case metrics.Timer:
|
||||
t := metric.Snapshot()
|
||||
ps := t.Percentiles(c.Percentiles)
|
||||
count := t.Count()
|
||||
fmt.Fprintf(w, "%s.%s.count %d %d\n", c.Prefix, name, count, now)
|
||||
fmt.Fprintf(w, "%s.%s.count_ps %.2f %d\n", c.Prefix, name, float64(count)/flushSeconds, now)
|
||||
fmt.Fprintf(w, "%s.%s.min %d %d\n", c.Prefix, name, t.Min()/int64(du), now)
|
||||
fmt.Fprintf(w, "%s.%s.max %d %d\n", c.Prefix, name, t.Max()/int64(du), now)
|
||||
fmt.Fprintf(w, "%s.%s.mean %.2f %d\n", c.Prefix, name, t.Mean()/du, now)
|
||||
fmt.Fprintf(w, "%s.%s.std-dev %.2f %d\n", c.Prefix, name, t.StdDev()/du, now)
|
||||
for psIdx, psKey := range c.Percentiles {
|
||||
key := strings.Replace(strconv.FormatFloat(psKey*100.0, 'f', -1, 64), ".", "", 1)
|
||||
fmt.Fprintf(w, "%s.%s.%s-percentile %.2f %d\n", c.Prefix, name, key, ps[psIdx]/du, now)
|
||||
}
|
||||
fmt.Fprintf(w, "%s.%s.one-minute %.2f %d\n", c.Prefix, name, t.Rate1(), now)
|
||||
fmt.Fprintf(w, "%s.%s.five-minute %.2f %d\n", c.Prefix, name, t.Rate5(), now)
|
||||
fmt.Fprintf(w, "%s.%s.fifteen-minute %.2f %d\n", c.Prefix, name, t.Rate15(), now)
|
||||
fmt.Fprintf(w, "%s.%s.mean-rate %.2f %d\n", c.Prefix, name, t.RateMean(), now)
|
||||
}
|
||||
w.Flush()
|
||||
})
|
||||
return nil
|
||||
}
|
||||
@@ -22,12 +22,9 @@ type m = map[string]any
|
||||
|
||||
func Main(c *config.C, configTest bool, buildVersion string, l *slog.Logger, deviceFactory overlay.DeviceFactory) (retcon *Control, reterr error) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
// The goroutines started below stop only when this context does, and only a caller holding the
|
||||
// Control can arrange that. Cancel whenever we are not handing one back, which covers an error
|
||||
// and a config test alike: a config test used to leave the lighthouse query worker, and a
|
||||
// hostname resolver per dns named static host, running for the life of the process.
|
||||
// Automatically cancel the context if Main returns an error, to signal all created goroutines to quit.
|
||||
defer func() {
|
||||
if retcon == nil {
|
||||
if reterr != nil {
|
||||
cancel()
|
||||
}
|
||||
}()
|
||||
|
||||
@@ -1,82 +0,0 @@
|
||||
package nebula
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"net/netip"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/slackhq/nebula/cert"
|
||||
cert_test "github.com/slackhq/nebula/cert_test"
|
||||
"github.com/slackhq/nebula/config"
|
||||
"github.com/slackhq/nebula/test"
|
||||
"github.com/stretchr/testify/require"
|
||||
"go.uber.org/goleak"
|
||||
)
|
||||
|
||||
// TestMain_ConfigTestReleasesItsGoroutines pins the rule that Main only leaves goroutines running
|
||||
// when it hands back a Control to stop them with.
|
||||
//
|
||||
// A config test gets no Control, so anything it started had nothing to stop it: the lighthouse
|
||||
// query worker, and a hostname resolver per dns named static host, ran for the life of the
|
||||
// process. That matters to every embedder that validates a config in process rather than by
|
||||
// exec'ing, dnclient and the apple clients included, because they do it on each config load and
|
||||
// the leak accumulates.
|
||||
func TestMain_ConfigTestReleasesItsGoroutines(t *testing.T) {
|
||||
defer goleak.VerifyNone(t, goleak.IgnoreCurrent())
|
||||
|
||||
l := test.NewLogger()
|
||||
dir := t.TempDir()
|
||||
|
||||
before := time.Now().Add(-time.Hour)
|
||||
after := time.Now().Add(time.Hour)
|
||||
ca, _, caKey, caPEM := cert_test.NewTestCaCert(cert.Version2, cert.Curve_CURVE25519, before, after, nil, nil, nil)
|
||||
networks := []netip.Prefix{netip.MustParsePrefix("10.0.0.1/24")}
|
||||
_, _, keyPEM, certPEM := cert_test.NewTestCert(
|
||||
cert.Version2, cert.Curve_CURVE25519, ca, caKey, "config-test", before, after, networks, nil, nil)
|
||||
|
||||
caPath := filepath.Join(dir, "ca.pem")
|
||||
certPath := filepath.Join(dir, "cert.pem")
|
||||
keyPath := filepath.Join(dir, "key.pem")
|
||||
require.NoError(t, os.WriteFile(caPath, caPEM, 0o600))
|
||||
require.NoError(t, os.WriteFile(certPath, certPEM, 0o600))
|
||||
require.NoError(t, os.WriteFile(keyPath, keyPEM, 0o600))
|
||||
|
||||
// A static host by address, not by name: the query worker is the goroutine under test and a
|
||||
// hostname would drag a real dns lookup into a unit test.
|
||||
configBody := fmt.Sprintf(`
|
||||
pki:
|
||||
ca: %s
|
||||
cert: %s
|
||||
key: %s
|
||||
static_host_map:
|
||||
"10.0.0.2": ["192.0.2.1:4242"]
|
||||
lighthouse:
|
||||
hosts:
|
||||
- "10.0.0.2"
|
||||
listen:
|
||||
host: 127.0.0.1
|
||||
port: 0
|
||||
tun:
|
||||
disabled: true
|
||||
firewall:
|
||||
outbound:
|
||||
- port: any
|
||||
proto: any
|
||||
host: any
|
||||
inbound:
|
||||
- port: any
|
||||
proto: any
|
||||
host: any
|
||||
`, caPath, certPath, keyPath)
|
||||
require.NoError(t, os.WriteFile(filepath.Join(dir, "config.yml"), []byte(configBody), 0o600))
|
||||
|
||||
c := config.NewC(l)
|
||||
require.NoError(t, c.Load(dir))
|
||||
|
||||
ctrl, err := Main(c, true, "config-test", l, nil)
|
||||
require.NoError(t, err)
|
||||
require.Nil(t, ctrl, "a config test hands back nothing to stop, so it must stop itself")
|
||||
}
|
||||
+3
-61
@@ -30,10 +30,7 @@ type tun struct {
|
||||
Routes atomic.Pointer[[]Route]
|
||||
routeTree atomic.Pointer[bart.Table[routing.Gateways]]
|
||||
linkAddr *netroute.LinkAddr
|
||||
// hostOwned means the fd arrived from the OS, which has already configured addressing, mtu
|
||||
// and routes for it. NEPacketTunnelProvider on darwin does this.
|
||||
hostOwned bool
|
||||
l *slog.Logger
|
||||
l *slog.Logger
|
||||
}
|
||||
|
||||
type ifReq struct {
|
||||
@@ -153,48 +150,8 @@ func (t *tun) deviceBytes() (o [16]byte) {
|
||||
return
|
||||
}
|
||||
|
||||
// newTunFromFd adopts a utun the host already created and configured, which is how a darwin
|
||||
// network extension is handed its device. Everything about moving packets is shared with newTun,
|
||||
// only the setup differs: the host owns addressing and routing here.
|
||||
func newTunFromFd(c *config.C, l *slog.Logger, deviceFd int, vpnNetworks []netip.Prefix) (*tun, error) {
|
||||
if err := unix.SetNonblock(deviceFd, true); err != nil {
|
||||
// We own the fd from the moment it is handed to us
|
||||
_ = unix.Close(deviceFd)
|
||||
return nil, fmt.Errorf("failed to set the tun fd to non-blocking mode: %w", err)
|
||||
}
|
||||
|
||||
file := os.NewFile(uintptr(deviceFd), "/dev/tun")
|
||||
t := &tun{
|
||||
f: file,
|
||||
Device: utunNameFromFd(deviceFd),
|
||||
vpnNetworks: vpnNetworks,
|
||||
DefaultMTU: c.GetInt("tun.mtu", DefaultMTU),
|
||||
hostOwned: true,
|
||||
l: l,
|
||||
}
|
||||
|
||||
if err := t.reload(c, true); err != nil {
|
||||
_ = file.Close()
|
||||
return nil, err
|
||||
}
|
||||
|
||||
c.RegisterReloadCallback(func(c *config.C) {
|
||||
if err := t.reload(c, false); err != nil {
|
||||
util.LogWithContextIfNeeded("failed to reload tun device", err, t.l)
|
||||
}
|
||||
})
|
||||
|
||||
return t, nil
|
||||
}
|
||||
|
||||
// utunNameFromFd asks the socket what interface it is, for logs. A blank name is not worth
|
||||
// failing a tunnel over, so an error just leaves it empty.
|
||||
func utunNameFromFd(fd int) string {
|
||||
name, err := unix.GetsockoptString(fd, unix.AF_SYS_CONTROL, _UTUN_OPT_IFNAME)
|
||||
if err != nil {
|
||||
return ""
|
||||
}
|
||||
return name
|
||||
func newTunFromFd(_ *config.C, _ *slog.Logger, _ int, _ []netip.Prefix) (*tun, error) {
|
||||
return nil, fmt.Errorf("newTunFromFd not supported in Darwin")
|
||||
}
|
||||
|
||||
func (t *tun) Close() error {
|
||||
@@ -205,12 +162,6 @@ func (t *tun) Close() error {
|
||||
}
|
||||
|
||||
func (t *tun) Activate() error {
|
||||
// The host handed us a configured device. Its addresses, mtu and routes come from the network
|
||||
// settings it applied, and a sandboxed extension cannot change them anyway.
|
||||
if t.hostOwned {
|
||||
return nil
|
||||
}
|
||||
|
||||
devName := t.deviceBytes()
|
||||
|
||||
s, err := unix.Socket(
|
||||
@@ -424,11 +375,6 @@ func getLinkAddr(name string) (*netroute.LinkAddr, error) {
|
||||
}
|
||||
|
||||
func (t *tun) addRoutes(logErrors bool) error {
|
||||
// The route tree is still ours, the system routing table is not
|
||||
if t.hostOwned {
|
||||
return nil
|
||||
}
|
||||
|
||||
routes := *t.Routes.Load()
|
||||
|
||||
for _, r := range routes {
|
||||
@@ -458,10 +404,6 @@ func (t *tun) addRoutes(logErrors bool) error {
|
||||
}
|
||||
|
||||
func (t *tun) removeRoutes(routes []Route) error {
|
||||
if t.hostOwned {
|
||||
return nil
|
||||
}
|
||||
|
||||
for _, r := range routes {
|
||||
if !r.Install {
|
||||
continue
|
||||
|
||||
@@ -13,7 +13,6 @@ import (
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
graphite "github.com/cyberdelia/go-metrics-graphite"
|
||||
mp "github.com/nbrownus/go-metrics-prometheus"
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
"github.com/prometheus/client_golang/prometheus/promhttp"
|
||||
@@ -253,7 +252,7 @@ func (s *statsServer) buildRuntime(cfg statsConfig) ([]func(), *http.Server) {
|
||||
// loadStatsConfig already resolved and validated the address; re-parse
|
||||
// the resolved form (no DNS lookup) to get a *net.TCPAddr.
|
||||
addr, _ := net.ResolveTCPAddr(cfg.graphite.protocol, cfg.graphite.resolvedAddr)
|
||||
gcfg := graphite.Config{
|
||||
gcfg := graphiteConfigExport{
|
||||
Addr: addr,
|
||||
Registry: metrics.DefaultRegistry,
|
||||
FlushInterval: cfg.interval,
|
||||
@@ -262,7 +261,7 @@ func (s *statsServer) buildRuntime(cfg statsConfig) ([]func(), *http.Server) {
|
||||
Percentiles: []float64{0.5, 0.75, 0.95, 0.99, 0.999},
|
||||
}
|
||||
captureFns = append(captureFns, func() {
|
||||
if err := graphite.Once(gcfg); err != nil {
|
||||
if err := graphiteOnce(gcfg); err != nil {
|
||||
s.l.Error("Graphite export failed", "error", err)
|
||||
}
|
||||
})
|
||||
|
||||
+1
-1
@@ -371,7 +371,7 @@ func waitForListening(t *testing.T, addr string) {
|
||||
})
|
||||
}
|
||||
|
||||
// graphiteSink is a minimal TCP accept-and-discard server so graphite.Once
|
||||
// graphiteSink is a minimal TCP accept-and-discard server so graphiteOnce
|
||||
// calls in tests don't spam error logs or wedge on connection refused.
|
||||
type graphiteSink struct {
|
||||
ln net.Listener
|
||||
|
||||
Reference in New Issue
Block a user