diff --git a/.devcontainer/Dockerfile b/.devcontainer/Dockerfile index 9b6a0edfd..dc7475a06 100644 --- a/.devcontainer/Dockerfile +++ b/.devcontainer/Dockerfile @@ -1,4 +1,4 @@ -FROM golang:1.26.7-bookworm +FROM golang:1.26.10-bookworm RUN apt-get update && export DEBIAN_FRONTEND=noninteractive \ && apt-get -y install --no-install-recommends\ diff --git a/.github/workflows/golang-test-linux.yml b/.github/workflows/golang-test-linux.yml index b3e4dd4f7..e4388bfe3 100644 --- a/.github/workflows/golang-test-linux.yml +++ b/.github/workflows/golang-test-linux.yml @@ -244,6 +244,14 @@ jobs: NB_TEST_PKCS11_DISPOSABLE: "1" run: CGO_ENABLED=0 go test -tags pkcs11 -timeout 5m ./client/internal/pkcs11/... ./client/internal/certproof/... + - name: Test SSH privilege drop + # purego marks the runtime as cgo without libc's setxid hooks, so the syscall + # package's setuid family panics in this build. Drop privileges as root to cover it. + run: | + CGO_ENABLED=0 go test -c -tags 'pkcs11 privileged' -o "$RUNNER_TEMP/ssh-server.test" ./client/ssh/server/ + sudo "$RUNNER_TEMP/ssh-server.test" -test.run '^TestDropPrivilegesAllThreads$' -test.v + CGO_ENABLED=0 GOARCH=arm64 go test -c -tags 'pkcs11 privileged' -o "$RUNNER_TEMP/ssh-server-arm64.test" ./client/ssh/server/ + test_client_on_docker: name: "Client (Docker) / Unit" needs: [build-cache] @@ -303,7 +311,7 @@ jobs: -e GOCACHE=${CONTAINER_GOCACHE} \ -e GOMODCACHE=${CONTAINER_GOMODCACHE} \ -e CONTAINER=${CONTAINER} \ - golang:1.26.7-alpine \ + golang:1.26.10-alpine \ sh -c ' \ apk update; apk add --no-cache \ ca-certificates iptables ip6tables dbus dbus-dev libpcap-dev build-base; \ diff --git a/.golangci.yaml b/.golangci.yaml index e350b9de7..f52f822dd 100644 --- a/.golangci.yaml +++ b/.golangci.yaml @@ -24,6 +24,15 @@ linters: settings: errcheck: check-type-assertions: false + forbidigo: + analyze-types: true + forbid: + - pattern: ^(fmt\.Print(|f|ln)|print|println)$ + # Builds that link libc through purego mark the runtime as cgo without + # registering the libc setxid hooks, so these panic in AllThreadsSyscall. + - pattern: ^(syscall|unix)\.(Setgroups|Set(e|re|res)?[ug]id|AllThreadsSyscall6?)$ + pkg: ^(syscall|golang\.org/x/sys/unix)$ + msg: use the setgroups/setgid/setuid helpers in client/ssh/server, which go through libc in purego builds gocritic: disabled-checks: - commentFormatting @@ -86,6 +95,9 @@ linters: - linters: - forbidigo path: signal/cmd/root\.go + - linters: + - forbidigo + path: client/ssh/server/setid_unix\.go - linters: - unused path: sharedsock/filter\.go diff --git a/client/Dockerfile-rootless.ubi b/client/Dockerfile-rootless.ubi index 4701728c1..396e0d562 100644 --- a/client/Dockerfile-rootless.ubi +++ b/client/Dockerfile-rootless.ubi @@ -1,4 +1,4 @@ -FROM registry.access.redhat.com/ubi9/ubi-minimal@sha256:7fbeae18dc9476399f565e68255f602a3374ea8614ba3d14843565131a13ff93 +FROM registry.access.redhat.com/ubi9/ubi-minimal@sha256:2d4c4551af6831dcd26bbf86108c4385de5183ec8fe6b5868f6f9640fe47f4ed ARG TARGETPLATFORM ARG NETBIRD_BINARY=$TARGETPLATFORM/netbird diff --git a/client/cmd/status.go b/client/cmd/status.go index 7abf189d0..2dc308bac 100644 --- a/client/cmd/status.go +++ b/client/cmd/status.go @@ -20,6 +20,7 @@ import ( var ( detailFlag bool + tableFlag bool ipv4Flag bool ipv6Flag bool jsonFlag bool @@ -44,11 +45,12 @@ func init() { ipsFilterMap = make(map[string]struct{}) prefixNamesFilterMap = make(map[string]struct{}) statusCmd.PersistentFlags().BoolVarP(&detailFlag, "detail", "d", false, "display detailed status information in human-readable format") + statusCmd.PersistentFlags().BoolVarP(&tableFlag, "table", "t", false, "display peers in a table, supports the --filter-by-* flags") statusCmd.PersistentFlags().BoolVarP(&jsonFlag, "json", "j", false, "display detailed status information in json format") statusCmd.PersistentFlags().BoolVarP(&yamlFlag, "yaml", "y", false, "display detailed status information in yaml format") statusCmd.PersistentFlags().BoolVarP(&ipv4Flag, "ipv4", "4", false, "display only NetBird IPv4 of this peer, e.g., --ipv4 will output 100.64.0.33") statusCmd.PersistentFlags().BoolVarP(&ipv6Flag, "ipv6", "6", false, "display only NetBird IPv6 of this peer") - statusCmd.MarkFlagsMutuallyExclusive("detail", "json", "yaml", "ipv4", "ipv6") + statusCmd.MarkFlagsMutuallyExclusive("detail", "table", "json", "yaml", "ipv4", "ipv6") statusCmd.PersistentFlags().StringSliceVarP(&ipsFilter, "filter-by-ips", "I", []string{}, "filters the detailed output by a list of one or more IPs (v4 or v6), e.g., --filter-by-ips 100.64.0.100,fd00::1") statusCmd.PersistentFlags().StringSliceVarP(&prefixNamesFilter, "filter-by-names", "N", []string{}, "filters the detailed output by a list of one or more peer FQDN or hostnames, e.g., --filter-by-names peer-a,peer-b.netbird.cloud") statusCmd.PersistentFlags().StringVarP(&statusFilter, "filter-by-status", "S", "", "filters the detailed output by connection status(idle|connecting|connected), e.g., --filter-by-status connected") @@ -172,6 +174,8 @@ func statusFunc(cmd *cobra.Command, args []string) error { switch { case detailFlag: statusOutputString = outputInformationHolder.FullDetailSummary() + case tableFlag: + statusOutputString = outputInformationHolder.PeersTable() case jsonFlag: statusOutputString, err = outputInformationHolder.JSON() case yamlFlag: @@ -264,7 +268,7 @@ func parseFilters() error { } func enableDetailFlagWhenFilterFlag() { - if !detailFlag && !jsonFlag && !yamlFlag { + if !detailFlag && !jsonFlag && !yamlFlag && !tableFlag { detailFlag = true } } diff --git a/client/cmd/status_test.go b/client/cmd/status_test.go index 03608eab0..ca0fd97dd 100644 --- a/client/cmd/status_test.go +++ b/client/cmd/status_test.go @@ -4,6 +4,7 @@ import ( "testing" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" ) func TestParsingOfIP(t *testing.T) { @@ -13,3 +14,61 @@ func TestParsingOfIP(t *testing.T) { assert.Equal(t, "192.168.178.123\n", parsedIP) } + +// resetStatusFlags restores only the status-specific output flags. The global +// root flags (setup-key and friends) are shared with other tests in the +// package, so they must not be touched here. +func resetStatusFlags(t *testing.T) { + t.Helper() + t.Cleanup(func() { + for _, name := range []string{"detail", "table", "json", "yaml", "ipv4", "ipv6"} { + f := statusCmd.Flags().Lookup(name) + require.NotNil(t, f, "flag %s must exist", name) + require.NoError(t, f.Value.Set(f.DefValue)) + f.Changed = false + } + }) +} + +func TestStatusTableFlagExclusive(t *testing.T) { + for _, other := range []string{"detail", "json", "yaml", "ipv4", "ipv6"} { + t.Run(other, func(t *testing.T) { + resetStatusFlags(t) + require.NoError(t, statusCmd.Flags().Set("table", "true")) + require.NoError(t, statusCmd.Flags().Set(other, "true")) + + assert.Error(t, statusCmd.ValidateFlagGroups(), "--table with --%s must be rejected", other) + }) + } + + t.Run("alone", func(t *testing.T) { + resetStatusFlags(t) + require.NoError(t, statusCmd.Flags().Set("table", "true")) + + assert.NoError(t, statusCmd.ValidateFlagGroups()) + }) +} + +func TestFilterEnablesDetailOnlyWithoutOutputFormat(t *testing.T) { + tests := []struct { + name string + table bool + json bool + wantDetail bool + }{ + {name: "no format", wantDetail: true}, + {name: "table", table: true, wantDetail: false}, + {name: "json", json: true, wantDetail: false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + resetStatusFlags(t) + tableFlag, jsonFlag = tt.table, tt.json + + enableDetailFlagWhenFilterFlag() + + assert.Equal(t, tt.wantDetail, detailFlag) + }) + } +} diff --git a/client/internal/metrics/infra/ingest/Dockerfile b/client/internal/metrics/infra/ingest/Dockerfile index 3620c524b..1b4862a7d 100644 --- a/client/internal/metrics/infra/ingest/Dockerfile +++ b/client/internal/metrics/infra/ingest/Dockerfile @@ -1,9 +1,9 @@ -FROM golang:1.25-alpine AS build +FROM golang:1.26.10-alpine AS build WORKDIR /app COPY go.mod main.go ./ RUN CGO_ENABLED=0 go build -o ingest . -FROM alpine:3.20 +FROM alpine:3.24 RUN adduser -D -H ingest COPY --from=build /app/ingest /usr/local/bin/ingest USER ingest diff --git a/client/ssh/server/executor_drop_privileged_linux_test.go b/client/ssh/server/executor_drop_privileged_linux_test.go new file mode 100644 index 000000000..27cc1384b --- /dev/null +++ b/client/ssh/server/executor_drop_privileged_linux_test.go @@ -0,0 +1,102 @@ +//go:build linux && privileged + +package server + +import ( + "bufio" + "fmt" + "os" + "os/exec" + "path/filepath" + "runtime" + "strings" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const envDropPrivilegesChild = "NB_TEST_DROP_PRIVILEGES_CHILD" + +const ( + dropTargetUID = 65534 + dropTargetGID = 65534 + dropExtraGID = 65533 +) + +// TestDropPrivilegesAllThreads drops privileges in a child process, because the drop +// cannot be undone, and checks that every OS thread of that process carries the target +// credentials. A drop applied to the calling thread only would leave the others as root. +func TestDropPrivilegesAllThreads(t *testing.T) { + if os.Getenv(envDropPrivilegesChild) == "1" { + dropPrivilegesChild(t) + return + } + if os.Geteuid() != 0 { + t.Skip("requires root") + } + + cmd := exec.Command(os.Args[0], "-test.run=^TestDropPrivilegesAllThreads$", "-test.v") + cmd.Env = append(os.Environ(), envDropPrivilegesChild+"=1") + out, err := cmd.CombinedOutput() + t.Logf("child output:\n%s", out) + require.NoError(t, err, "child must drop privileges on all threads") + assert.Contains(t, string(out), "--- PASS: TestDropPrivilegesAllThreads", "child test must run and pass") +} + +func dropPrivilegesChild(t *testing.T) { + // Park goroutines on their own OS threads so the process has threads other than + // the one that performs the drop. + release := make(chan struct{}) + defer close(release) + ready := make(chan struct{}) + for range 4 { + go func() { + runtime.LockOSThread() + defer runtime.UnlockOSThread() + ready <- struct{}{} + <-release + }() + <-ready + } + + pd := NewPrivilegeDropper() + require.NoError(t, pd.DropPrivileges(dropTargetUID, dropTargetGID, []uint32{dropTargetGID, dropExtraGID})) + + tasks, err := filepath.Glob("/proc/self/task/*/status") + require.NoError(t, err) + require.Greater(t, len(tasks), 4, "process must have several threads for the check to mean anything") + + want := map[string]string{ + "Uid": fmt.Sprintf("%d\t%d\t%d\t%d", dropTargetUID, dropTargetUID, dropTargetUID, dropTargetUID), + "Gid": fmt.Sprintf("%d\t%d\t%d\t%d", dropTargetGID, dropTargetGID, dropTargetGID, dropTargetGID), + "Groups": fmt.Sprintf("%d %d", dropExtraGID, dropTargetGID), + } + for _, task := range tasks { + got := readStatusFields(t, task) + for field, value := range want { + assert.Equal(t, value, got[field], "%s of %s", field, task) + } + } + + assert.Error(t, seteuid(0), "euid 0 must not be restorable after the drop") + assert.Error(t, setegid(0), "egid 0 must not be restorable after the drop") +} + +func readStatusFields(t *testing.T, path string) map[string]string { + f, err := os.Open(path) + require.NoError(t, err) + defer f.Close() + + fields := make(map[string]string) + scanner := bufio.NewScanner(f) + for scanner.Scan() { + key, value, ok := strings.Cut(scanner.Text(), ":") + if !ok { + continue + } + fields[key] = strings.TrimSpace(value) + } + require.NoError(t, scanner.Err()) + return fields +} diff --git a/client/ssh/server/executor_unix.go b/client/ssh/server/executor_unix.go index ee0b0ff78..5b42190ab 100644 --- a/client/ssh/server/executor_unix.go +++ b/client/ssh/server/executor_unix.go @@ -11,7 +11,6 @@ import ( "path/filepath" "runtime" "strings" - "syscall" log "github.com/sirupsen/logrus" ) @@ -151,15 +150,15 @@ func (pd *PrivilegeDropper) setGroupsAndIDs(targetUID, targetGID uint32, supplem } } - if err := syscall.Setgroups(groups); err != nil { + if err := setgroups(groups); err != nil { return fmt.Errorf("setgroups to %v: %w", groups, err) } - if err := syscall.Setgid(int(targetGID)); err != nil { + if err := setgid(int(targetGID)); err != nil { return fmt.Errorf("setgid to %d: %w", targetGID, err) } - if err := syscall.Setuid(int(targetUID)); err != nil { + if err := setuid(int(targetUID)); err != nil { return fmt.Errorf("setuid to %d: %w", targetUID, err) } @@ -182,12 +181,12 @@ func (pd *PrivilegeDropper) validatePrivilegeDropSuccess(targetUID, targetGID ui // validatePrivilegeDropReversibility ensures privileges cannot be restored func (pd *PrivilegeDropper) validatePrivilegeDropReversibility(targetUID, targetGID uint32, originalUID, originalGID int) error { if originalGID != int(targetGID) { - if err := syscall.Setegid(originalGID); err == nil { + if err := setegid(originalGID); err == nil { return fmt.Errorf("privilege drop validation failed: able to restore original GID %d", originalGID) } } if originalUID != int(targetUID) { - if err := syscall.Seteuid(originalUID); err == nil { + if err := seteuid(originalUID); err == nil { return fmt.Errorf("privilege drop validation failed: able to restore original UID %d", originalUID) } } diff --git a/client/ssh/server/setid_libc_linux.go b/client/ssh/server/setid_libc_linux.go new file mode 100644 index 000000000..13a9f254c --- /dev/null +++ b/client/ssh/server/setid_libc_linux.go @@ -0,0 +1,108 @@ +//go:build pkcs11 && linux && !android && (amd64 || arm64) + +package server + +import ( + "fmt" + "runtime" + "sync" + "syscall" + + "github.com/ebitengine/purego" +) + +// This build links libc through purego, whose fakecgo runtime marks the binary as cgo +// without registering the libc hooks a real cgo build provides. syscall's setuid family +// then panics in AllThreadsSyscall, so the calls go to libc directly: glibc applies them +// to every thread, and all Go threads here are created through pthread_create. + +type libcSetID struct { + setgroups func(size uintptr, list *uint32) int32 + setgid func(gid uint32) int32 + setuid func(uid uint32) int32 + setegid func(egid uint32) int32 + seteuid func(euid uint32) int32 + errnoLocation func() *int32 +} + +var ( + libcSetIDOnce sync.Once + libcSetIDFns libcSetID + libcSetIDErr error +) + +func loadLibcSetID() (*libcSetID, error) { + libcSetIDOnce.Do(func() { + syms := []struct { + fn any + name string + }{ + {&libcSetIDFns.setgroups, "setgroups"}, + {&libcSetIDFns.setgid, "setgid"}, + {&libcSetIDFns.setuid, "setuid"}, + {&libcSetIDFns.setegid, "setegid"}, + {&libcSetIDFns.seteuid, "seteuid"}, + {&libcSetIDFns.errnoLocation, "__errno_location"}, + } + for _, s := range syms { + addr, err := purego.Dlsym(purego.RTLD_DEFAULT, s.name) + if err != nil { + libcSetIDErr = fmt.Errorf("resolve libc %s: %w", s.name, err) + return + } + purego.RegisterFunc(s.fn, addr) + } + }) + if libcSetIDErr != nil { + return nil, libcSetIDErr + } + return &libcSetIDFns, nil +} + +// libcCall runs call on a locked thread so errno is read from the thread that set it. +func libcCall(call func(*libcSetID) int32) error { + fns, err := loadLibcSetID() + if err != nil { + return err + } + + runtime.LockOSThread() + defer runtime.UnlockOSThread() + + if call(fns) == 0 { + return nil + } + if errno := *fns.errnoLocation(); errno != 0 { + return syscall.Errno(errno) + } + return syscall.EINVAL +} + +func setgroups(gids []int) error { + list := make([]uint32, len(gids)) + for i, g := range gids { + list[i] = uint32(g) + } + return libcCall(func(f *libcSetID) int32 { + if len(list) == 0 { + return f.setgroups(0, nil) + } + return f.setgroups(uintptr(len(list)), &list[0]) + }) +} + +func setgid(gid int) error { + return libcCall(func(f *libcSetID) int32 { return f.setgid(uint32(gid)) }) +} + +func setuid(uid int) error { + return libcCall(func(f *libcSetID) int32 { return f.setuid(uint32(uid)) }) +} + +func setegid(egid int) error { + return libcCall(func(f *libcSetID) int32 { return f.setegid(uint32(egid)) }) +} + +func seteuid(euid int) error { + return libcCall(func(f *libcSetID) int32 { return f.seteuid(uint32(euid)) }) +} diff --git a/client/ssh/server/setid_unix.go b/client/ssh/server/setid_unix.go new file mode 100644 index 000000000..bd9fe3856 --- /dev/null +++ b/client/ssh/server/setid_unix.go @@ -0,0 +1,25 @@ +//go:build unix && !(pkcs11 && linux && !android && (amd64 || arm64)) + +package server + +import "syscall" + +func setgroups(gids []int) error { + return syscall.Setgroups(gids) +} + +func setgid(gid int) error { + return syscall.Setgid(gid) +} + +func setuid(uid int) error { + return syscall.Setuid(uid) +} + +func setegid(egid int) error { + return syscall.Setegid(egid) +} + +func seteuid(euid int) error { + return syscall.Seteuid(euid) +} diff --git a/client/status/table.go b/client/status/table.go new file mode 100644 index 000000000..f1cd7e01d --- /dev/null +++ b/client/status/table.go @@ -0,0 +1,55 @@ +package status + +import ( + "fmt" + "strings" + "text/tabwriter" + "time" +) + +// PeersTable renders the peers as an aligned table, one row per peer. +func (o *OutputOverview) PeersTable() string { + var sb strings.Builder + fmt.Fprintf(&sb, "Peers (%d/%d connected):\n", o.Peers.Connected, o.Peers.Total) + + tw := tabwriter.NewWriter(&sb, 0, 0, 3, ' ', 0) + fmt.Fprintln(tw, "NAME\tIP\tSTATUS\tCONN\tLATENCY\tRX / TX") + for _, p := range o.Peers.Details { + fmt.Fprintf(tw, "%s\t%s\t%s\t%s\t%s\t%s / %s\n", + shortName(p.FQDN), + dashIfEmpty(p.IP), + p.Status, + dashIfEmpty(p.ConnType), + tableLatency(p), + toIEC(p.TransferReceived), + toIEC(p.TransferSent), + ) + } + // The writer targets a strings.Builder, which never returns an error. + _ = tw.Flush() + + return sb.String() +} + +// shortName drops the domain suffix, which is the same for every peer. +func shortName(fqdn string) string { + name, _, _ := strings.Cut(fqdn, ".") + return dashIfEmpty(name) +} + +func tableLatency(p PeerStateDetailOutput) string { + if p.Latency <= 0 { + return "-" + } + if p.Latency >= time.Millisecond { + return p.Latency.Round(time.Millisecond).String() + } + return p.Latency.Round(time.Microsecond).String() +} + +func dashIfEmpty(s string) string { + if s == "" { + return "-" + } + return s +} diff --git a/client/status/table_test.go b/client/status/table_test.go new file mode 100644 index 000000000..c60e57917 --- /dev/null +++ b/client/status/table_test.go @@ -0,0 +1,176 @@ +package status + +import ( + "strings" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + protobuf "google.golang.org/protobuf/proto" + + "github.com/netbirdio/netbird/client/anonymize" + "github.com/netbirdio/netbird/client/proto" +) + +func TestPeersTable(t *testing.T) { + overview := OutputOverview{Peers: PeersStateOutput{ + Total: 2, + Connected: 1, + Details: []PeerStateDetailOutput{ + { + FQDN: "web-01.netbird.cloud", + IP: "100.64.0.12", + PubKey: "k3Vx9pQ2abcdefghijklmnopqrstuvwxyz0123456789=", + Status: "Connected", + ConnType: "P2P", + Latency: 12 * time.Millisecond, + TransferReceived: 1024, + TransferSent: 2048, + }, + { + FQDN: "old-nas.netbird.cloud", + IP: "100.64.0.77", + Status: "Idle", + }, + }, + }} + + lines := strings.Split(strings.TrimRight(overview.PeersTable(), "\n"), "\n") + require.Len(t, lines, 4, "title, header and one line per peer") + + assert.Equal(t, "Peers (1/2 connected):", lines[0]) + assert.Equal(t, + []string{"NAME", "IP", "STATUS", "CONN", "LATENCY", "RX", "/", "TX"}, + strings.Fields(lines[1])) + + connected := strings.Fields(lines[2]) + assert.Equal(t, "web-01", connected[0], "domain suffix must be dropped") + assert.Equal(t, "Connected", connected[2]) + assert.Equal(t, "P2P", connected[3]) + assert.Equal(t, "12ms", connected[4]) + assert.NotContains(t, lines[2], "k3Vx9pQ2", "public key must not be shown") + assert.Contains(t, lines[2], "1.0 KiB / 2.0 KiB") + + idle := strings.Fields(lines[3]) + assert.Equal(t, []string{"old-nas", "100.64.0.77", "Idle", "-", "-", "0", "B", "/", "0", "B"}, idle, + "missing values render as dashes") +} + +func TestPeersTable_NoPeers(t *testing.T) { + out := (&OutputOverview{}).PeersTable() + assert.True(t, strings.HasPrefix(out, "Peers (0/0 connected):\n")) + assert.Contains(t, out, "NAME") +} + +func tableRows(t *testing.T, opts ConvertOptions) []string { + t.Helper() + + opts.DaemonVersion = resp.GetDaemonVersion() + opts.DaemonStatus = ParseDaemonStatus(resp.GetStatus()) + // The anonymizer rewrites slices in place, so work on a copy to keep the + // shared fixture intact for the other tests. + fullStatus := protobuf.Clone(resp.GetFullStatus()).(*proto.FullStatus) + overview := ConvertToStatusOutputOverview(fullStatus, opts) + out := overview.PeersTable() + + lines := strings.Split(strings.TrimRight(out, "\n"), "\n") + require.GreaterOrEqual(t, len(lines), 2, "title and header are always present") + return lines +} + +// The fixture holds peer-1 (connected, P2P) and peer-2 (connected, relayed). +func TestPeersTable_Filters(t *testing.T) { + tests := []struct { + name string + opts ConvertOptions + wantTitle string + wantPeers []string + }{ + { + name: "no filter lists every peer sorted by IP", + wantTitle: "Peers (2/2 connected):", + wantPeers: []string{"peer-1", "peer-2"}, + }, + { + name: "connection type P2P", + opts: ConvertOptions{ConnectionTypeFilter: "P2P"}, + wantTitle: "Peers (1/1 connected):", + wantPeers: []string{"peer-1"}, + }, + { + name: "connection type Relayed", + opts: ConvertOptions{ConnectionTypeFilter: "relayed"}, + wantTitle: "Peers (1/1 connected):", + wantPeers: []string{"peer-2"}, + }, + { + name: "status with no match", + opts: ConvertOptions{StatusFilter: "idle"}, + wantTitle: "Peers (0/0 connected):", + }, + { + name: "name prefix", + opts: ConvertOptions{ + PrefixNamesFilter: []string{"peer-2"}, + PrefixNamesFilterMap: map[string]struct{}{"peer-2": {}}, + }, + wantTitle: "Peers (1/1 connected):", + wantPeers: []string{"peer-2"}, + }, + { + name: "IPv4", + opts: ConvertOptions{IPsFilter: map[string]struct{}{"192.168.178.101": {}}}, + wantTitle: "Peers (1/1 connected):", + wantPeers: []string{"peer-1"}, + }, + { + name: "IPv6", + opts: ConvertOptions{IPsFilter: map[string]struct{}{"fd00::1": {}}}, + wantTitle: "Peers (1/1 connected):", + wantPeers: []string{"peer-1"}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + lines := tableRows(t, tt.opts) + + assert.Equal(t, tt.wantTitle, lines[0]) + rows := lines[2:] + require.Len(t, rows, len(tt.wantPeers), "one row per matching peer") + for i, name := range tt.wantPeers { + assert.True(t, strings.HasPrefix(rows[i], name+" "), "row %d should be %s, got %q", i, name, rows[i]) + } + }) + } +} + +// The table shows whatever the anonymizer left in the overview, so it must +// carry the redacted names rather than the originals. +func TestPeersTable_Anonymized(t *testing.T) { + lines := tableRows(t, ConvertOptions{Anonymize: true, AnonymizeLevel: anonymize.LevelStrict}) + + for _, row := range lines[2:] { + assert.NotContains(t, row, "awesome-domain.com", "FQDN must be anonymized") + assert.NotContains(t, row, "peer-", "peer name must be anonymized") + } +} + +func TestShortName(t *testing.T) { + assert.Equal(t, "web-01", shortName("web-01.netbird.cloud")) + assert.Equal(t, "web-01", shortName("web-01")) + assert.Equal(t, "-", shortName("")) +} + +func TestTableLatency(t *testing.T) { + tests := map[time.Duration]string{ + 0: "-", + 145939625 * time.Nanosecond: "146ms", + 12 * time.Millisecond: "12ms", + 420*time.Microsecond + 700: "421µs", + } + for in, want := range tests { + assert.Equal(t, want, tableLatency(PeerStateDetailOutput{Latency: in}), "latency %s", in) + } +} diff --git a/client/testutil/privileged/runner_test.go b/client/testutil/privileged/runner_test.go index 157005d3e..c7bde1316 100644 --- a/client/testutil/privileged/runner_test.go +++ b/client/testutil/privileged/runner_test.go @@ -25,7 +25,7 @@ import ( // (.github/workflows/golang-test-linux.yml, test_client_on_docker). const ( containerImage = "golang" - containerTag = "1.26.7-alpine" + containerTag = "1.26.10-alpine" ) const ( diff --git a/client/ui/build/docker/Dockerfile.cross b/client/ui/build/docker/Dockerfile.cross index 55c0d69e1..650087fb3 100644 --- a/client/ui/build/docker/Dockerfile.cross +++ b/client/ui/build/docker/Dockerfile.cross @@ -13,7 +13,7 @@ # docker run --rm -v $(pwd):/app wails-cross windows amd64 # docker run --rm -v $(pwd):/app wails-cross windows arm64 -FROM golang:1.26.7-bookworm +FROM golang:1.26.10-bookworm ARG TARGETARCH diff --git a/client/ui/build/docker/Dockerfile.server b/client/ui/build/docker/Dockerfile.server index 57183f1d2..7a7f36575 100644 --- a/client/ui/build/docker/Dockerfile.server +++ b/client/ui/build/docker/Dockerfile.server @@ -2,7 +2,7 @@ # Multi-stage build for minimal image size # Build stage -FROM golang:1.26.7-alpine AS builder +FROM golang:1.26.10-alpine AS builder WORKDIR /app @@ -22,7 +22,7 @@ RUN go mod tidy RUN go build -tags server -ldflags="-s -w" -o server . # Runtime stage - minimal image -FROM gcr.io/distroless/static-debian12 +FROM gcr.io/distroless/static-debian13 # Copy the binary COPY --from=builder /app/server /server diff --git a/combined/Dockerfile.multistage b/combined/Dockerfile.multistage index 011379c2f..0551516fd 100644 --- a/combined/Dockerfile.multistage +++ b/combined/Dockerfile.multistage @@ -1,4 +1,4 @@ -FROM golang:1.26.7-bookworm AS builder +FROM golang:1.26.10-bookworm AS builder WORKDIR /app # Install build dependencies diff --git a/combined/Dockerfile.ubi b/combined/Dockerfile.ubi index 66ef55a34..dbbb920c9 100644 --- a/combined/Dockerfile.ubi +++ b/combined/Dockerfile.ubi @@ -1,4 +1,4 @@ -FROM registry.access.redhat.com/ubi9/ubi-minimal@sha256:7fbeae18dc9476399f565e68255f602a3374ea8614ba3d14843565131a13ff93 +FROM registry.access.redhat.com/ubi9/ubi-minimal@sha256:2d4c4551af6831dcd26bbf86108c4385de5183ec8fe6b5868f6f9640fe47f4ed ARG TARGETPLATFORM ARG VERSION=dev diff --git a/combined/cmd/admin_config_test.go b/combined/cmd/admin_config_test.go index 5cec26552..c1ea2e8da 100644 --- a/combined/cmd/admin_config_test.go +++ b/combined/cmd/admin_config_test.go @@ -7,6 +7,7 @@ import ( "github.com/stretchr/testify/require" + accesslogsmanager "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/accesslogs/manager" nbconfig "github.com/netbirdio/netbird/management/internals/server/config" ) @@ -41,3 +42,22 @@ func TestApplyServerStoreEnv(t *testing.T) { require.Equal(t, "postgres-dsn", os.Getenv("NB_STORE_ENGINE_POSTGRES_DSN")) require.Equal(t, "store.db", os.Getenv("NB_STORE_ENGINE_SQLITE_FILE")) } + +func TestApplyAccessLogStoreEnv(t *testing.T) { + keys := []string{accesslogsmanager.StoreEngineEnv, accesslogsmanager.PostgresDSNEnv, accesslogsmanager.SqliteFileEnv} + for _, key := range keys { + t.Setenv(key, "") + require.NoError(t, os.Unsetenv(key)) + } + + applyAccessLogStoreEnv(StoreConfig{}) + for _, key := range keys { + _, ok := os.LookupEnv(key) + require.False(t, ok, "%s should stay unset for an empty accessLogStore block", key) + } + + applyAccessLogStoreEnv(StoreConfig{Engine: "Postgres", DSN: "postgres-dsn", File: "access_logs.db"}) + require.Equal(t, "postgres", os.Getenv(accesslogsmanager.StoreEngineEnv)) + require.Equal(t, "postgres-dsn", os.Getenv(accesslogsmanager.PostgresDSNEnv)) + require.Equal(t, "access_logs.db", os.Getenv(accesslogsmanager.SqliteFileEnv)) +} diff --git a/combined/cmd/config.go b/combined/cmd/config.go index ca3b25617..0c7302fe5 100644 --- a/combined/cmd/config.go +++ b/combined/cmd/config.go @@ -71,6 +71,7 @@ type ServerConfig struct { Auth AuthConfig `yaml:"auth"` Store StoreConfig `yaml:"store"` ActivityStore StoreConfig `yaml:"activityStore"` + AccessLogStore StoreConfig `yaml:"accessLogStore"` AuthStore StoreConfig `yaml:"authStore"` ReverseProxy ReverseProxyConfig `yaml:"reverseProxy"` diff --git a/combined/cmd/root.go b/combined/cmd/root.go index fedb0e1cc..7f4549643 100644 --- a/combined/cmd/root.go +++ b/combined/cmd/root.go @@ -26,6 +26,7 @@ import ( "github.com/netbirdio/netbird/encryption" agentnetworkpricing "github.com/netbirdio/netbird/management/internals/modules/agentnetwork/pricing" + accesslogsmanager "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/accesslogs/manager" mgmtServer "github.com/netbirdio/netbird/management/internals/server" nbconfig "github.com/netbirdio/netbird/management/internals/server/config" "github.com/netbirdio/netbird/management/server/telemetry" @@ -152,6 +153,20 @@ func applyActivityStoreEnv(storeConfig StoreConfig) error { return nil } +// applyAccessLogStoreEnv leaves unset fields to follow the main store, so a +// postgres engine without a DSN reuses the main store DSN. +func applyAccessLogStoreEnv(storeConfig StoreConfig) { + if engine := storeConfig.Engine; engine != "" { + os.Setenv(accesslogsmanager.StoreEngineEnv, strings.ToLower(engine)) + } + if dsn := storeConfig.DSN; dsn != "" { + os.Setenv(accesslogsmanager.PostgresDSNEnv, dsn) + } + if file := storeConfig.File; file != "" { + os.Setenv(accesslogsmanager.SqliteFileEnv, file) + } +} + func initializeConfig() error { var err error config, err = LoadConfig(configPath) @@ -172,6 +187,7 @@ func initializeConfig() error { if err := applyActivityStoreEnv(config.Server.ActivityStore); err != nil { return err } + applyAccessLogStoreEnv(config.Server.AccessLogStore) log.Infof("Starting combined NetBird server") logConfig(config) diff --git a/combined/config.yaml.example b/combined/config.yaml.example index 68dbf701e..7f2d85210 100644 --- a/combined/config.yaml.example +++ b/combined/config.yaml.example @@ -121,6 +121,13 @@ server: # dsn: "" # Connection string for postgres # file: "" # Custom SQLite file path (optional, defaults to {dataDir}/events.db) + # Reverse proxy access log store configuration (optional, follows the store engine: + # sqlite uses its own file in dataDir, postgres reuses the store DSN with a separate pool) + # accessLogStore: + # engine: "postgres" # sqlite or postgres + # dsn: "" # Connection string for a dedicated postgres database + # file: "" # Custom SQLite file path (optional, defaults to {dataDir}/access_logs.db) + # Auth (embedded IdP) store configuration (optional, defaults to sqlite3 in dataDir/idp.db) # authStore: # engine: "sqlite3" # sqlite3 or postgres diff --git a/docs/testing-privileged.md b/docs/testing-privileged.md index 939770de9..7d9552c4a 100644 --- a/docs/testing-privileged.md +++ b/docs/testing-privileged.md @@ -32,7 +32,7 @@ list; both are optional and default to the full privileged suite. 1. Skips immediately when it detects it is already inside the container (`DOCKER_CI=true`), so the privileged tests run in place instead of recursing. -2. Otherwise spins up a `golang:1.26.7-alpine` container (matching CI), +2. Otherwise spins up a `golang:1.26.10-alpine` container (matching CI), bind-mounts the repo and the host Go build/module caches, installs the required packages, and runs `go test -tags 'devcert privileged'` over the client packages. diff --git a/e2e/harness/Dockerfile.client b/e2e/harness/Dockerfile.client index 4c76b95c6..4bcb2b079 100644 --- a/e2e/harness/Dockerfile.client +++ b/e2e/harness/Dockerfile.client @@ -3,7 +3,7 @@ # artifact), so this mirrors its alpine runtime + entrypoint while compiling the # CGO-free client inline. BuildKit cache mounts keep rebuilds incremental. -FROM golang:1.26.7-bookworm AS builder +FROM golang:1.26.10-bookworm AS builder WORKDIR /src COPY go.mod go.sum ./ RUN --mount=type=cache,target=/go/pkg/mod go mod download diff --git a/e2e/harness/Dockerfile.nblink b/e2e/harness/Dockerfile.nblink index dac3aceec..0628aa083 100644 --- a/e2e/harness/Dockerfile.nblink +++ b/e2e/harness/Dockerfile.nblink @@ -8,7 +8,7 @@ # that quietly hands it NET_ADMIN would make the suite prove the opposite of # what it is for. -FROM golang:1.26.7-bookworm AS builder +FROM golang:1.26.10-bookworm AS builder WORKDIR /src COPY go.mod go.sum ./ RUN --mount=type=cache,target=/go/pkg/mod go mod download diff --git a/go.mod b/go.mod index f93141d5c..0d1492731 100644 --- a/go.mod +++ b/go.mod @@ -4,7 +4,7 @@ go 1.26.0 // Pin the toolchain to a patch release >= go1.26.2 // See https://go.dev/issue/77875. -toolchain go1.26.7 +toolchain go1.26.10 require ( cunicu.li/go-rosenpass v0.5.42 diff --git a/integration_tests/management/network_map_db/network_map_data_test.go b/integration_tests/management/network_map_db/network_map_data_test.go index 00c0ec03f..96172bd49 100644 --- a/integration_tests/management/network_map_db/network_map_data_test.go +++ b/integration_tests/management/network_map_db/network_map_data_test.go @@ -10,14 +10,21 @@ import ( "path/filepath" "runtime" "strings" + "sync" "testing" + "time" + "github.com/jackc/pgx/v5/pgxpool" networkmapdb "github.com/netbirdio/netbird/management/internals/network_map_db" "github.com/netbirdio/netbird/management/server/integrations/integrated_validator" "github.com/netbirdio/netbird/management/server/settings" "github.com/netbirdio/netbird/management/server/types" log "github.com/sirupsen/logrus" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel/attribute" + sdkmetric "go.opentelemetry.io/otel/sdk/metric" + "go.opentelemetry.io/otel/sdk/metric/metricdata" "go.uber.org/mock/gomock" ) @@ -29,6 +36,19 @@ var goldenNMap string const EnvUpdateGoldenData = "NMAP_UPDATE_GOLDEN_DATA" +var seedNetworkMapDataOnce sync.Once + +func seedNetworkMapData(t *testing.T, ctx context.Context) { + t.Helper() + seedNetworkMapDataOnce.Do(func() { + for _, query := range strings.Split(nmapData, ";") { + if err := store(t).Exec(ctx, query); err != nil { + log.Fatalf("error initializing nmap test: %s", err.Error()) + } + } + }) +} + func TestGetNetworkMapData(t *testing.T) { ctx := context.TODO() @@ -51,11 +71,7 @@ func TestGetNetworkMapData(t *testing.T) { IntegratedPeerValidator: peerValidators, } - for _, query := range strings.Split(nmapData, ";") { - if err := store(t).Exec(ctx, query); err != nil { - log.Fatalf("error initializing nmap test: %s", err.Error()) - } - } + seedNetworkMapData(t, ctx) nmap, err := storeImpl.GetNetworkMapData(ctx, "account-33") assert.NoError(t, err) @@ -72,3 +88,52 @@ func TestGetNetworkMapData(t *testing.T) { } assert.Equal(t, goldenNMap, string(serializedNMap)) } + +func TestGetNetworkMapData_FromReplica(t *testing.T) { + if engine != string(types.PostgresStoreEngine) { + t.Skip("replicas exist only for Postgres") + } + ctx := context.TODO() + + extraSettingsManager := settings.NewMockManager(gomock.NewController(t)) + extraSettingsManager.EXPECT().GetExtraSettings(gomock.Any(), gomock.Any()).Return(&types.ExtraSettings{}, nil) + peerValidators := integrated_validator.NewMockIntegratedValidator(gomock.NewController(t)) + peerValidators.EXPECT().GetValidatedPeers(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(map[string]struct{}{}, nil) + + seedNetworkMapData(t, ctx) + + reader := sdkmetric.NewManualReader() + metrics, err := networkmapdb.NewMetrics(sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader)).Meter("test"), + map[string]*pgxpool.Pool{"replica": pgstore.Pool}) + require.NoError(t, err) + + primary := store(t) + storeImpl := networkmapdb.NetworkMapDBStoreImpl{ + Metrics: metrics, + Store: primary, + Replica: primary, + ReplicaWait: time.Second, + ExtraSettingsManager: extraSettingsManager, + IntegratedPeerValidator: peerValidators, + } + + nmap, err := storeImpl.GetNetworkMapData(ctx, "account-33") + require.NoError(t, err) + assert.NotEmpty(t, nmap.Peers, "data must be read through the replica transaction after its attempt deadline is released") + + var rm metricdata.ResourceMetrics + require.NoError(t, reader.Collect(ctx, &rm)) + var totalConns int64 + for _, sm := range rm.ScopeMetrics { + for _, m := range sm.Metrics { + if gauge, ok := m.Data.(metricdata.Gauge[int64]); ok { + for _, dp := range gauge.DataPoints { + if state, _ := dp.Attributes.Value(attribute.Key("state")); state.AsString() == "total" { + totalConns += dp.Value + } + } + } + } + } + assert.Positive(t, totalConns, "the replica pool gauge reports its open connections") +} diff --git a/management/Dockerfile.multistage b/management/Dockerfile.multistage index 5d037f1b1..1e9d11227 100644 --- a/management/Dockerfile.multistage +++ b/management/Dockerfile.multistage @@ -1,4 +1,4 @@ -FROM golang:1.26.7-bookworm AS builder +FROM golang:1.26.10-bookworm AS builder WORKDIR /app # Install build dependencies diff --git a/management/Dockerfile.ubi b/management/Dockerfile.ubi index 628d1a086..163302aa9 100644 --- a/management/Dockerfile.ubi +++ b/management/Dockerfile.ubi @@ -1,4 +1,4 @@ -FROM registry.access.redhat.com/ubi9/ubi-minimal@sha256:7fbeae18dc9476399f565e68255f602a3374ea8614ba3d14843565131a13ff93 +FROM registry.access.redhat.com/ubi9/ubi-minimal@sha256:2d4c4551af6831dcd26bbf86108c4385de5183ec8fe6b5868f6f9640fe47f4ed ARG TARGETPLATFORM ARG VERSION=dev diff --git a/management/internals/controllers/network_map/controller/controller.go b/management/internals/controllers/network_map/controller/controller.go index 08d018e5f..0554811db 100644 --- a/management/internals/controllers/network_map/controller/controller.go +++ b/management/internals/controllers/network_map/controller/controller.go @@ -110,7 +110,7 @@ func NewController(ctx context.Context, store store.Store, metrics telemetry.App if nmdataStore != nil { interval := requestbuffer.Interval(ctx, "NB_NETWORK_MAP_DATA_BUFFER_INTERVAL", defaultNetworkMapDataBufferInterval) log.WithContext(ctx).Infof("set network map data request buffer interval to %s", interval) - c.nmdataBuffer = requestbuffer.New(ctx, "network map data request buffer", interval, c.fetchNetworkMapData) + c.nmdataBuffer = requestbuffer.NewSerialized(ctx, "network map data request buffer", interval, c.fetchNetworkMapData) } return c diff --git a/management/internals/modules/peers/manager.go b/management/internals/modules/peers/manager.go index 6f9705efe..b49667948 100644 --- a/management/internals/modules/peers/manager.go +++ b/management/internals/modules/peers/manager.go @@ -199,6 +199,10 @@ func (m *managerImpl) DeletePeers(ctx context.Context, accountID string, peerIDs return err } + if err = transaction.IncrementNetworkSerial(ctx, accountID); err != nil { + return fmt.Errorf("increment network serial: %w", err) + } + log.WithContext(ctx).Debugf("DeletePeers: deleted peer %s", peerID) if !(peer.ProxyMeta.Embedded || peer.Meta.KernelVersion == "wasm") { diff --git a/management/internals/modules/peers/manager_test.go b/management/internals/modules/peers/manager_test.go index 2645276b7..eba565fbf 100644 --- a/management/internals/modules/peers/manager_test.go +++ b/management/internals/modules/peers/manager_test.go @@ -5,6 +5,8 @@ import ( "fmt" "testing" + "github.com/netbirdio/netbird/management/internals/controllers/network_map" + "github.com/netbirdio/netbird/management/server/account" "github.com/netbirdio/netbird/management/server/peer" "github.com/netbirdio/netbird/management/server/permissions" "github.com/netbirdio/netbird/management/server/permissions/modules" @@ -195,3 +197,33 @@ func TestManager_GetPeersPaginated_NoPermission_User_IsRestrictable_ViewBlocked( assert.Equal(t, 0, total) assert.NoError(t, err) } + +func TestManager_DeletePeers_IncrementsNetworkSerial(t *testing.T) { + ctrl := gomock.NewController(t) + mockStore := store.NewMockStore(ctrl) + txStore := store.NewMockStore(ctrl) + accountManager := account.NewMockManager(ctrl) + networkMapController := network_map.NewMockController(ctrl) + + m := NewManager(mockStore, permissions.NewMockManager(ctrl)).(*managerImpl) + m.accountManager = accountManager + m.networkMapController = networkMapController + + ctx := context.Background() + mockStore.EXPECT().GetAccountSettings(gomock.Any(), store.LockingStrengthNone, "test-account").Return(&types.Settings{}, nil) + networkMapController.EXPECT().GetDNSDomain(gomock.Any()).Return("netbird.selfhosted") + mockStore.EXPECT().ExecuteInTransaction(gomock.Any(), gomock.Any()).Times(2). + DoAndReturn(func(_ context.Context, fn func(store.Store) error) error { return fn(txStore) }) + + for _, id := range []string{"peer-1", "peer-2"} { + txStore.EXPECT().GetPeerByID(gomock.Any(), store.LockingStrengthNone, "test-account", id).Return(&peer.Peer{ID: id, Status: &peer.PeerStatus{}}, nil) + txStore.EXPECT().RemovePeerFromAllGroups(gomock.Any(), id).Return(nil) + txStore.EXPECT().GetPolicyRulesByResourceID(gomock.Any(), store.LockingStrengthNone, "test-account", id).Return(nil, nil) + txStore.EXPECT().DeletePeer(gomock.Any(), "test-account", id).Return(nil) + } + txStore.EXPECT().IncrementNetworkSerial(gomock.Any(), "test-account").Times(2).Return(nil) + accountManager.EXPECT().StoreEvent(gomock.Any(), "test-user", gomock.Any(), "test-account", gomock.Any(), gomock.Any()).Times(2) + accountManager.EXPECT().UpdateAccountPeers(gomock.Any(), "test-account", gomock.Any()) + + assert.NoError(t, m.DeletePeers(ctx, "test-account", []string{"peer-1", "peer-2"}, "test-user", false)) +} diff --git a/management/internals/modules/reverseproxy/accesslogs/filter.go b/management/internals/modules/reverseproxy/accesslogs/filter.go index a1fa28312..996828c51 100644 --- a/management/internals/modules/reverseproxy/accesslogs/filter.go +++ b/management/internals/modules/reverseproxy/accesslogs/filter.go @@ -46,7 +46,7 @@ type AccessLogFilter struct { SortOrder string // Sort order: asc or desc (default: desc) // Filtering parameters - Search *string // General search across log ID, host, path, source IP, and user fields + Search *string // General search across log ID, host, path, and source IP SourceIP *string // Filter by source IP address Host *string // Filter by host header Path *string // Filter by request path (supports LIKE pattern) diff --git a/management/internals/modules/reverseproxy/accesslogs/manager/conn.go b/management/internals/modules/reverseproxy/accesslogs/manager/conn.go new file mode 100644 index 000000000..b1ca05d16 --- /dev/null +++ b/management/internals/modules/reverseproxy/accesslogs/manager/conn.go @@ -0,0 +1,92 @@ +package manager + +import ( + "context" + "fmt" + "os" + "strconv" + "strings" + + log "github.com/sirupsen/logrus" + + "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/accesslogs" + "github.com/netbirdio/netbird/management/internals/shared/db" + "github.com/netbirdio/netbird/management/server/store" +) + +const ( + // StoreEngineEnv selects the access log store engine. Unset follows the main store engine. + StoreEngineEnv = "NB_ACCESS_LOG_STORE_ENGINE" + // PostgresDSNEnv is the access log Postgres DSN. Unset reuses the main store DSN. + PostgresDSNEnv = "NB_ACCESS_LOG_POSTGRES_DSN" + // SqliteFileEnv is the access log SQLite file, resolved against the data directory when relative. + SqliteFileEnv = "NB_ACCESS_LOG_SQLITE_FILE" + // MaxOpenConnsEnv sizes the access log connection pool. Unset applies NB_SQL_MAX_OPEN_CONNS. + MaxOpenConnsEnv = "NB_ACCESS_LOG_SQL_MAX_OPEN_CONNS" + + sqliteFileName = "access_logs.db" +) + +// OpenConn opens and migrates the access log database. Unless configured +// otherwise, it follows mainEngine: SQLite uses its own file in dataDir and +// Postgres reuses the main DSN. +func OpenConn(ctx context.Context, dataDir string, mainEngine db.Engine) (*db.Conn, error) { + engine := mainEngine + if v := os.Getenv(StoreEngineEnv); v != "" { + engine = db.Engine(strings.ToLower(v)) + } + + conn, err := openConn(ctx, dataDir, engine) + if err != nil { + return nil, err + } + if err := conn.AutoMigrate(&accesslogs.AccessLogEntry{}); err != nil { + _ = conn.Close() + return nil, fmt.Errorf("migrate access log store: %w", err) + } + return conn, nil +} + +func openConn(ctx context.Context, dataDir string, engine db.Engine) (*db.Conn, error) { + switch engine { + case db.SqliteStoreEngine: + log.WithContext(ctx).Info("using SQLite access log store engine") + file := sqliteFileName + if v := os.Getenv(SqliteFileEnv); v != "" { + file = v + } + return db.OpenSqliteFile(ctx, dataDir, file) + case db.PostgresStoreEngine: + log.WithContext(ctx).Info("using Postgres access log store engine") + dsn, err := postgresDSN() + if err != nil { + return nil, err + } + return db.OpenPostgresWithoutPool(ctx, dsn, connOptions(ctx)...) + default: + return nil, fmt.Errorf("unsupported access log store engine: %s", engine) + } +} + +func postgresDSN() (string, error) { + if dsn := os.Getenv(PostgresDSNEnv); dsn != "" { + return dsn, nil + } + if dsn, ok := store.LookupPostgresDSN(); ok && dsn != "" { + return dsn, nil + } + return "", fmt.Errorf("neither %s nor %s is set", PostgresDSNEnv, store.PostgresDsnEnv) +} + +func connOptions(ctx context.Context) []db.Option { + val := os.Getenv(MaxOpenConnsEnv) + if val == "" { + return nil + } + conns, err := strconv.Atoi(val) + if err != nil || conns <= 0 { + log.WithContext(ctx).Warnf("ignoring invalid %s value %q", MaxOpenConnsEnv, val) + return nil + } + return []db.Option{db.WithMaxOpenConns(conns)} +} diff --git a/management/internals/modules/reverseproxy/accesslogs/manager/conn_test.go b/management/internals/modules/reverseproxy/accesslogs/manager/conn_test.go new file mode 100644 index 000000000..90fdf8ffa --- /dev/null +++ b/management/internals/modules/reverseproxy/accesslogs/manager/conn_test.go @@ -0,0 +1,142 @@ +package manager + +import ( + "context" + "os" + "path/filepath" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "gorm.io/driver/sqlite" + "gorm.io/gorm" + + "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/accesslogs" + "github.com/netbirdio/netbird/management/internals/shared/db" + "github.com/netbirdio/netbird/management/server/store" + "github.com/netbirdio/netbird/management/server/testutil" +) + +func unsetStoreEnv(t *testing.T) { + t.Helper() + for _, key := range []string{StoreEngineEnv, PostgresDSNEnv, SqliteFileEnv, MaxOpenConnsEnv, store.PostgresDsnEnv, store.PostgresDsnEnvLegacy} { + t.Setenv(key, "") + require.NoError(t, os.Unsetenv(key)) + } +} + +func openTestConn(t *testing.T, dataDir string, mainEngine db.Engine) *db.Conn { + t.Helper() + conn, err := OpenConn(context.Background(), dataDir, mainEngine) + require.NoError(t, err) + t.Cleanup(func() { _ = conn.Close() }) + return conn +} + +func assertStoresEntries(t *testing.T, conn *db.Conn) { + t.Helper() + ctx := context.Background() + repo := NewRepository(conn) + require.NoError(t, repo.Create(ctx, newEntry("entry", "acc", "GET", time.Minute))) + + logs, total, err := repo.ListByAccount(ctx, db.LockingStrengthNone, "acc", accesslogs.AccessLogFilter{Page: 1, PageSize: 10}) + require.NoError(t, err) + assert.EqualValues(t, 1, total, "the stored entry should be listed") + require.Len(t, logs, 1) + assert.Equal(t, "entry", logs[0].ID) +} + +func TestOpenConn_SqliteUsesItsOwnFile(t *testing.T) { + unsetStoreEnv(t) + dataDir := t.TempDir() + + conn := openTestConn(t, dataDir, db.SqliteStoreEngine) + + assert.Equal(t, db.SqliteStoreEngine, conn.Engine()) + assert.FileExists(t, filepath.Join(dataDir, sqliteFileName)) + assert.NoFileExists(t, filepath.Join(dataDir, db.SqliteFileName), "the main store file must not be opened") + assertStoresEntries(t, conn) +} + +func TestOpenConn_SqliteFileFromEnv(t *testing.T) { + unsetStoreEnv(t) + dataDir := t.TempDir() + t.Setenv(SqliteFileEnv, "custom_logs.db") + + conn := openTestConn(t, dataDir, db.SqliteStoreEngine) + + assert.FileExists(t, filepath.Join(dataDir, "custom_logs.db")) + assert.NoFileExists(t, filepath.Join(dataDir, sqliteFileName)) + assertStoresEntries(t, conn) +} + +func TestOpenConn_EngineFromEnvOverridesMainEngine(t *testing.T) { + unsetStoreEnv(t) + t.Setenv(StoreEngineEnv, "SQLite") + + conn := openTestConn(t, t.TempDir(), db.PostgresStoreEngine) + + assert.Equal(t, db.SqliteStoreEngine, conn.Engine(), "the configured engine should win over the main store engine") +} + +func TestOpenConn_PostgresWithoutDSNFails(t *testing.T) { + unsetStoreEnv(t) + + _, err := OpenConn(context.Background(), t.TempDir(), db.PostgresStoreEngine) + + require.Error(t, err) + assert.Contains(t, err.Error(), PostgresDSNEnv) +} + +func TestOpenConn_UnsupportedEngineFails(t *testing.T) { + unsetStoreEnv(t) + t.Setenv(StoreEngineEnv, "mysql") + + _, err := OpenConn(context.Background(), t.TempDir(), db.SqliteStoreEngine) + + require.Error(t, err) + assert.Contains(t, err.Error(), "mysql") +} + +func TestOpenConn_PostgresReusesMainDSNWithItsOwnPool(t *testing.T) { + if os.Getenv("NETBIRD_STORE_ENGINE") != string(db.PostgresStoreEngine) { + t.Skip("requires NETBIRD_STORE_ENGINE=postgres") + } + unsetStoreEnv(t) + cleanup, dsn, err := testutil.CreatePostgresTestContainer() + require.NoError(t, err) + t.Cleanup(cleanup) + t.Setenv(store.PostgresDsnEnv, dsn) + t.Setenv(MaxOpenConnsEnv, "3") + + conn := openTestConn(t, t.TempDir(), db.PostgresStoreEngine) + + assert.Equal(t, db.PostgresStoreEngine, conn.Engine()) + assert.Nil(t, conn.Pool(nil), "access logs never take the pgx read paths, so no pgx pool should be opened") + sqlDB, err := conn.DB(nil).DB() + require.NoError(t, err) + assert.Equal(t, 3, sqlDB.Stats().MaxOpenConnections, "the pool should be sized by %s", MaxOpenConnsEnv) + assertStoresEntries(t, conn) +} + +func TestConnOptions_SizesPoolFromEnv(t *testing.T) { + // Invalid values fall back to the store-wide setting instead of failing boot. + cases := map[string]int{"3": 3, "": 7, "zero": 7, "0": 7, "-1": 7} + for value, want := range cases { + t.Run(value, func(t *testing.T) { + t.Setenv("NB_SQL_MAX_OPEN_CONNS", "7") + t.Setenv(MaxOpenConnsEnv, value) + + gormDB, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), "pool.db")), db.GormConfig()) + require.NoError(t, err) + conn, err := db.NewConn(context.Background(), gormDB, db.PostgresStoreEngine, nil, connOptions(context.Background())...) + require.NoError(t, err) + t.Cleanup(func() { _ = conn.Close() }) + + sqlDB, err := conn.DB(nil).DB() + require.NoError(t, err) + assert.Equal(t, want, sqlDB.Stats().MaxOpenConnections, "max open connections for %s=%q", MaxOpenConnsEnv, value) + }) + } +} diff --git a/management/internals/modules/reverseproxy/accesslogs/manager/repository.go b/management/internals/modules/reverseproxy/accesslogs/manager/repository.go index 105696ab7..3967442d8 100644 --- a/management/internals/modules/reverseproxy/accesslogs/manager/repository.go +++ b/management/internals/modules/reverseproxy/accesslogs/manager/repository.go @@ -85,8 +85,8 @@ func applyFilters(query *gorm.DB, filter accesslogs.AccessLogFilter) *gorm.DB { if filter.Search != nil { searchPattern := "%" + *filter.Search + "%" query = query.Where( - "id LIKE ? OR location_connection_ip LIKE ? OR host LIKE ? OR path LIKE ? OR CONCAT(host, path) LIKE ? OR user_id IN (SELECT id FROM users WHERE email LIKE ? OR name LIKE ?)", - searchPattern, searchPattern, searchPattern, searchPattern, searchPattern, searchPattern, searchPattern, + "id LIKE ? OR location_connection_ip LIKE ? OR host LIKE ? OR path LIKE ? OR CONCAT(host, path) LIKE ?", + searchPattern, searchPattern, searchPattern, searchPattern, searchPattern, ) } diff --git a/management/internals/modules/reverseproxy/accesslogs/manager/repository_test.go b/management/internals/modules/reverseproxy/accesslogs/manager/repository_test.go index 9931cc27c..7fea12a85 100644 --- a/management/internals/modules/reverseproxy/accesslogs/manager/repository_test.go +++ b/management/internals/modules/reverseproxy/accesslogs/manager/repository_test.go @@ -123,3 +123,37 @@ func TestSqlRepository_ListByAccount_StatusFilter(t *testing.T) { assert.Equal(t, []string{"info", "notfound"}, listIDs("failed")) assert.Equal(t, []string{"ok"}, listIDs("success")) } + +func TestSqlRepository_ListByAccount_Search(t *testing.T) { + repo, _ := newTestRepository(t) + ctx := context.Background() + api := newEntry("req-api", "acc", "GET", time.Hour) + api.Host = "api.example.com" + api.Path = "/v1/peers" + web := newEntry("req-web", "acc", "GET", 2*time.Hour) + web.Host = "web.example.com" + web.Path = "/login" + foreign := newEntry("foreign", "other", "GET", time.Hour) + foreign.Host = "api.example.com" + for _, entry := range []*accesslogs.AccessLogEntry{api, web, foreign} { + require.NoError(t, repo.Create(ctx, entry)) + } + + search := func(term string) []string { + logs, total, err := repo.ListByAccount(ctx, db.LockingStrengthNone, "acc", accesslogs.AccessLogFilter{Page: 1, PageSize: 10, Search: &term}) + require.NoError(t, err) + require.EqualValues(t, len(logs), total) + ids := make([]string, 0, len(logs)) + for _, entry := range logs { + ids = append(ids, entry.ID) + } + return ids + } + + assert.Equal(t, []string{"req-api"}, search("api.example"), "search should match the host") + assert.Equal(t, []string{"req-web"}, search("/login"), "search should match the path") + assert.Equal(t, []string{"req-api"}, search("example.com/v1"), "search should match host and path joined") + assert.Equal(t, []string{"req-web"}, search("req-web"), "search should match the log ID") + assert.Equal(t, []string{"req-api", "req-web"}, search("example.com"), "search should list every match newest first") + assert.Empty(t, search("nothing-matches")) +} diff --git a/management/internals/modules/reverseproxy/service/manager/l4_port_test.go b/management/internals/modules/reverseproxy/service/manager/l4_port_test.go index a44e759c4..5ec34a073 100644 --- a/management/internals/modules/reverseproxy/service/manager/l4_port_test.go +++ b/management/internals/modules/reverseproxy/service/manager/l4_port_test.go @@ -6,9 +6,9 @@ import ( "testing" "time" - "go.uber.org/mock/gomock" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "go.uber.org/mock/gomock" "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/proxy" rpservice "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/service" @@ -38,6 +38,7 @@ func setupL4Test(t *testing.T, customPortsSupported *bool) (*Manager, store.Stor err = testStore.SaveAccount(ctx, &types.Account{ Id: testAccountID, CreatedBy: testUserID, + Network: types.NewNetwork(), Settings: &types.Settings{ PeerExposeEnabled: true, PeerExposeGroups: []string{testGroupID}, diff --git a/management/internals/modules/reverseproxy/service/manager/manager.go b/management/internals/modules/reverseproxy/service/manager/manager.go index ebb52c74b..6173a381d 100644 --- a/management/internals/modules/reverseproxy/service/manager/manager.go +++ b/management/internals/modules/reverseproxy/service/manager/manager.go @@ -363,6 +363,10 @@ func (m *Manager) persistNewService(ctx context.Context, accountID string, svc * return fmt.Errorf("create service: %w", err) } + if err := transaction.IncrementNetworkSerial(ctx, accountID); err != nil { + return fmt.Errorf("increment network serial: %w", err) + } + return nil }) } @@ -534,6 +538,10 @@ func (m *Manager) persistNewEphemeralService(ctx context.Context, accountID, pee return fmt.Errorf("create service: %w", err) } + if err := transaction.IncrementNetworkSerial(ctx, accountID); err != nil { + return fmt.Errorf("increment network serial: %w", err) + } + return nil }) } @@ -744,6 +752,10 @@ func (m *Manager) executeServiceUpdate(ctx context.Context, transaction store.St return fmt.Errorf("update service: %w", err) } + if err := transaction.IncrementNetworkSerial(ctx, accountID); err != nil { + return fmt.Errorf("increment network serial: %w", err) + } + return nil } @@ -1027,6 +1039,10 @@ func (m *Manager) DeleteService(ctx context.Context, accountID, userID, serviceI return fmt.Errorf("failed to delete service: %w", err) } + if err = transaction.IncrementNetworkSerial(ctx, accountID); err != nil { + return fmt.Errorf("increment network serial: %w", err) + } + return nil }) if err != nil { @@ -1069,6 +1085,12 @@ func (m *Manager) DeleteAllServices(ctx context.Context, accountID, userID strin } } + if len(services) > 0 { + if err = transaction.IncrementNetworkSerial(ctx, accountID); err != nil { + return fmt.Errorf("increment network serial: %w", err) + } + } + return nil }) if err != nil { @@ -1425,6 +1447,10 @@ func (m *Manager) deletePeerService(ctx context.Context, accountID, peerID, serv return fmt.Errorf("delete service: %w", err) } + if err = transaction.IncrementNetworkSerial(ctx, accountID); err != nil { + return fmt.Errorf("increment network serial: %w", err) + } + return nil }) if err != nil { @@ -1477,6 +1503,11 @@ func (m *Manager) deleteExpiredPeerService(ctx context.Context, accountID, peerI if err = transaction.DeleteService(ctx, accountID, serviceID); err != nil { return fmt.Errorf("delete service: %w", err) } + + if err = transaction.IncrementNetworkSerial(ctx, accountID); err != nil { + return fmt.Errorf("increment network serial: %w", err) + } + deleted = true return nil diff --git a/management/internals/modules/reverseproxy/service/manager/manager_test.go b/management/internals/modules/reverseproxy/service/manager/manager_test.go index f9273d594..85d7c4574 100644 --- a/management/internals/modules/reverseproxy/service/manager/manager_test.go +++ b/management/internals/modules/reverseproxy/service/manager/manager_test.go @@ -227,6 +227,13 @@ func TestCheckDomainAvailable_EdgeCases(t *testing.T) { }) } +func networkSerial(t *testing.T, s store.Store, accountID string) uint64 { + t.Helper() + account, err := s.GetAccount(context.Background(), accountID) + require.NoError(t, err) + return account.Network.Serial +} + func TestPersistNewService(t *testing.T) { ctx := context.Background() accountID := "test-account" @@ -254,6 +261,9 @@ func TestPersistNewService(t *testing.T) { txMock.EXPECT(). CreateService(ctx, service). Return(nil) + txMock.EXPECT(). + IncrementNetworkSerial(ctx, accountID). + Return(nil) return fn(txMock) }) @@ -465,6 +475,9 @@ func TestDeletePeerService_SourcePeerValidation(t *testing.T) { txMock.EXPECT(). DeleteService(ctx, accountID, serviceID). Return(nil) + txMock.EXPECT(). + IncrementNetworkSerial(ctx, accountID). + Return(nil) return fn(txMock) }) mockStore.EXPECT(). @@ -570,6 +583,9 @@ func TestDeletePeerService_SourcePeerValidation(t *testing.T) { txMock.EXPECT(). DeleteService(ctx, accountID, serviceID). Return(nil) + txMock.EXPECT(). + IncrementNetworkSerial(ctx, accountID). + Return(nil) return fn(txMock) }) mockStore.EXPECT(). @@ -617,6 +633,9 @@ func TestDeletePeerService_SourcePeerValidation(t *testing.T) { txMock.EXPECT(). DeleteService(ctx, accountID, serviceID). Return(nil) + txMock.EXPECT(). + IncrementNetworkSerial(ctx, accountID). + Return(nil) return fn(txMock) }) mockStore.EXPECT(). @@ -679,6 +698,7 @@ func setupIntegrationTest(t *testing.T) (*Manager, store.Store) { err = testStore.SaveAccount(ctx, &types.Account{ Id: testAccountID, CreatedBy: testUserID, + Network: types.NewNetwork(), Settings: &types.Settings{ PeerExposeEnabled: true, PeerExposeGroups: []string{testGroupID}, @@ -843,6 +863,7 @@ func TestCreateServiceFromPeer(t *testing.T) { assert.Equal(t, rpservice.SourceEphemeral, persisted.Source, "source should be ephemeral") assert.Equal(t, testPeerID, persisted.SourcePeer, "source peer should be set") assert.NotNil(t, persisted.Meta.LastRenewedAt, "last renewed should be set") + assert.Equal(t, uint64(1), networkSerial(t, testStore, testAccountID), "network serial should be incremented") }) t.Run("creates service with custom domain", func(t *testing.T) { @@ -1049,8 +1070,10 @@ func TestDeleteService_DeletesEphemeralExpose(t *testing.T) { svc, err := testStore.GetServiceByDomain(ctx, resp.Domain) require.NoError(t, err) + serial := networkSerial(t, testStore, testAccountID) err = mgr.DeleteService(ctx, testAccountID, testUserID, svc.ID) require.NoError(t, err) + assert.Equal(t, serial+1, networkSerial(t, testStore, testAccountID), "network serial should be incremented") count, err = mgr.store.CountEphemeralServicesByPeer(ctx, store.LockingStrengthNone, testAccountID, testPeerID) require.NoError(t, err) @@ -1065,7 +1088,7 @@ func TestDeleteService_DeletesEphemeralExpose(t *testing.T) { func TestDeleteAllServices_DeletesEphemeralExposes(t *testing.T) { ctx := context.Background() - mgr, _ := setupIntegrationTest(t) + mgr, testStore := setupIntegrationTest(t) for i := range 3 { _, err := mgr.CreateServiceFromPeer(ctx, testAccountID, testPeerID, &rpservice.ExposeServiceRequest{ @@ -1079,8 +1102,10 @@ func TestDeleteAllServices_DeletesEphemeralExposes(t *testing.T) { require.NoError(t, err) assert.Equal(t, int64(3), count, "all ephemeral services should exist") + serial := networkSerial(t, testStore, testAccountID) err = mgr.DeleteAllServices(ctx, testAccountID, testUserID) require.NoError(t, err) + assert.Equal(t, serial+1, networkSerial(t, testStore, testAccountID), "network serial should be incremented") count, err = mgr.store.CountEphemeralServicesByPeer(ctx, store.LockingStrengthNone, testAccountID, testPeerID) require.NoError(t, err) @@ -1223,8 +1248,10 @@ func TestDeleteExpiredPeerService_DeletesTargets(t *testing.T) { require.Len(t, targets, 1, "ephemeral peer-exposed service should have exactly one persisted target before reaping") expireEphemeralService(t, testStore, testAccountID, resp.Domain) + serial := networkSerial(t, testStore, testAccountID) err = mgr.deleteExpiredPeerService(ctx, testAccountID, testPeerID, svcID) require.NoError(t, err) + assert.Equal(t, serial+1, networkSerial(t, testStore, testAccountID), "network serial should be incremented") _, err = testStore.GetServiceByDomain(ctx, resp.Domain) require.Error(t, err, "expired peer-exposed service should be deleted") @@ -1253,8 +1280,10 @@ func TestDeleteServiceFromPeer_DeletesTargets(t *testing.T) { require.NoError(t, err) require.Len(t, targets, 1, "ephemeral peer-exposed service should have exactly one persisted target before stopping") + serial := networkSerial(t, testStore, testAccountID) err = mgr.StopServiceFromPeer(ctx, testAccountID, testPeerID, svcID) require.NoError(t, err) + assert.Equal(t, serial+1, networkSerial(t, testStore, testAccountID), "network serial should be incremented") _, err = testStore.GetServiceByDomain(ctx, resp.Domain) require.Error(t, err, "stopped peer-exposed service should be deleted") @@ -1521,3 +1550,30 @@ func TestReloadAllServicesForAccount_ContinuesAfterFailedService(t *testing.T) { assert.Contains(t, err.Error(), "svc-broken") assert.NotContains(t, err.Error(), "svc-healthy") } + +func TestServiceWrites_IncrementNetworkSerial(t *testing.T) { + ctx := context.Background() + mgr, testStore, _ := setupL4Test(t, nil) + + created, err := mgr.CreateService(ctx, testAccountID, testUserID, &rpservice.Service{ + Name: "serial", Domain: "serial.test.netbird.io", Mode: rpservice.ModeHTTP, Enabled: true, + Targets: []*rpservice.Target{{ + AccountID: testAccountID, TargetId: testPeerID, + TargetType: rpservice.TargetTypePeer, Protocol: "http", Port: 8080, Enabled: true, + }}, + }) + require.NoError(t, err) + serial := networkSerial(t, testStore, testAccountID) + assert.Equal(t, uint64(1), serial, "create should increment the network serial") + + updated := created.Copy() + updated.Name = "serial renamed" + _, err = mgr.UpdateService(ctx, testAccountID, testUserID, updated) + require.NoError(t, err) + assert.Equal(t, serial+1, networkSerial(t, testStore, testAccountID), "update should increment the network serial") + serial++ + + require.NoError(t, mgr.SetStatus(ctx, testAccountID, created.ID, rpservice.StatusActive)) + require.NoError(t, mgr.SetCertificateIssuedAt(ctx, testAccountID, created.ID)) + assert.Equal(t, serial, networkSerial(t, testStore, testAccountID), "meta-only writes must not increment the network serial") +} diff --git a/management/internals/network_map_db/factory/db_store.go b/management/internals/network_map_db/factory/db_store.go index 3eea0ae69..e60c1f6f9 100644 --- a/management/internals/network_map_db/factory/db_store.go +++ b/management/internals/network_map_db/factory/db_store.go @@ -5,10 +5,15 @@ import ( "errors" "fmt" "os" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + "go.opentelemetry.io/otel/metric" networkmapdb "github.com/netbirdio/netbird/management/internals/network_map_db" networkmap_pgsql "github.com/netbirdio/netbird/management/internals/network_map_db/pgsql" networkmap_sqlite "github.com/netbirdio/netbird/management/internals/network_map_db/sqlite" + "github.com/netbirdio/netbird/management/internals/shared/requestbuffer" "github.com/netbirdio/netbird/management/server/integrations/integrated_validator" "github.com/netbirdio/netbird/management/server/settings" "github.com/netbirdio/netbird/management/server/store" @@ -16,7 +21,16 @@ import ( log "github.com/sirupsen/logrus" ) -const storeSqliteFileName = "store.db" +const ( + storeSqliteFileName = "store.db" + + // ReplicaDsnEnv names the Postgres DSN of a read replica serving network map data reads. + ReplicaDsnEnv = "NB_STORE_ENGINE_POSTGRES_REPLICA_DSN" + // replicaWaitEnv bounds how long a network map data read waits for the replica. + replicaWaitEnv = "NB_NETWORK_MAP_DATA_REPLICA_WAIT" + + defaultReplicaWait = 2 * time.Second +) var ErrNotSupportedStoreEngine = errors.New("unsupported store engine") @@ -25,7 +39,9 @@ func NewNetworkMapDBStore( kind types.Engine, dataDir string, integratedPeerValidator integrated_validator.IntegratedValidator, - extraSettingsManager settings.Manager) (*networkmapdb.NetworkMapDBStoreImpl, error) { + extraSettingsManager settings.Manager, + replicaPool *pgxpool.Pool, + meter metric.Meter) (*networkmapdb.NetworkMapDBStoreImpl, error) { switch kind { case types.SqliteStoreEngine: log.WithContext(ctx).Info("networkmap store is using SQLite") @@ -54,11 +70,32 @@ func NewNetworkMapDBStore( return nil, err } - return &networkmapdb.NetworkMapDBStoreImpl{ + impl := &networkmapdb.NetworkMapDBStoreImpl{ Store: store, IntegratedPeerValidator: integratedPeerValidator, ExtraSettingsManager: extraSettingsManager, - }, nil + } + + pools := map[string]*pgxpool.Pool{"primary": store.Pool} + if replicaPool != nil { + impl.Replica = &networkmap_pgsql.PgStore{Pool: replicaPool} + impl.ReplicaWait = requestbuffer.Interval(ctx, replicaWaitEnv, defaultReplicaWait) + pools["replica"] = replicaPool + replicaConfig := replicaPool.Config().ConnConfig + log.WithContext(ctx).Infof("networkmap store reads from Postgres replica %s:%d/%s, waiting up to %s for replication", + replicaConfig.Host, replicaConfig.Port, replicaConfig.Database, impl.ReplicaWait) + } else { + log.WithContext(ctx).Info("networkmap store reads from the primary, no replica configured") + } + + if meter != nil { + impl.Metrics, err = networkmapdb.NewMetrics(meter, pools) + if err != nil { + return nil, fmt.Errorf("create network map data metrics: %w", err) + } + } + + return impl, nil } return nil, fmt.Errorf("networkmap store doesn't support engine %s, %w", kind, ErrNotSupportedStoreEngine) diff --git a/management/internals/network_map_db/metrics.go b/management/internals/network_map_db/metrics.go new file mode 100644 index 000000000..0cfd8f92d --- /dev/null +++ b/management/internals/network_map_db/metrics.go @@ -0,0 +1,96 @@ +package networkmapdb + +import ( + "context" + "fmt" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/metric" +) + +const ( + readSourceReplica = "replica" + readSourcePrimary = "primary" + + readReasonNoReplica = "no_replica" + readReasonCaughtUp = "caught_up" + readReasonLagTimeout = "lag_timeout" + readReasonReplicaError = "replica_error" + readReasonSerialReadError = "serial_read_error" +) + +// Metrics records which database serves network map data reads, how long reads +// wait for the replica and the state of the connection pools. +type Metrics struct { + reads metric.Int64Counter + replicaWait metric.Int64Histogram +} + +// NewMetrics creates the network map data metrics on meter and observes the +// connection pools keyed by pool name. +func NewMetrics(meter metric.Meter, pools map[string]*pgxpool.Pool) (*Metrics, error) { + reads, err := meter.Int64Counter("management.nmdata.reads", + metric.WithUnit("1"), + metric.WithDescription("Network map data reads by the database that served them and the reason it was chosen"), + ) + if err != nil { + return nil, fmt.Errorf("create reads counter: %w", err) + } + + replicaWait, err := meter.Int64Histogram("management.nmdata.replica.wait.ms", + metric.WithUnit("milliseconds"), + metric.WithDescription("Time a network map data read waited for the replica to replicate the primary's network serial"), + ) + if err != nil { + return nil, fmt.Errorf("create replica wait histogram: %w", err) + } + + connections, err := meter.Int64ObservableGauge("management.nmdata.pool.connections", + metric.WithUnit("1"), + metric.WithDescription("Connections of the network map data connection pools by state"), + ) + if err != nil { + return nil, fmt.Errorf("create pool connections gauge: %w", err) + } + + _, err = meter.RegisterCallback(func(_ context.Context, observer metric.Observer) error { + for name, pool := range pools { + stat := pool.Stat() + observePool(observer, connections, name, "acquired", stat.AcquiredConns()) + observePool(observer, connections, name, "idle", stat.IdleConns()) + observePool(observer, connections, name, "total", stat.TotalConns()) + } + return nil + }, connections) + if err != nil { + return nil, fmt.Errorf("create pool connections callback: %w", err) + } + + return &Metrics{reads: reads, replicaWait: replicaWait}, nil +} + +func observePool(observer metric.Observer, gauge metric.Int64ObservableGauge, pool, state string, value int32) { + observer.ObserveInt64(gauge, int64(value), metric.WithAttributes( + attribute.String("pool", pool), + attribute.String("state", state), + )) +} + +func (m *Metrics) countRead(ctx context.Context, source, reason string) { + if m == nil { + return + } + m.reads.Add(ctx, 1, metric.WithAttributes( + attribute.String("source", source), + attribute.String("reason", reason), + )) +} + +func (m *Metrics) recordReplicaWait(ctx context.Context, wait time.Duration) { + if m == nil { + return + } + m.replicaWait.Record(ctx, wait.Milliseconds()) +} diff --git a/management/internals/network_map_db/network_map_data.go b/management/internals/network_map_db/network_map_data.go index 1e83bfb34..2c1e7abcc 100644 --- a/management/internals/network_map_db/network_map_data.go +++ b/management/internals/network_map_db/network_map_data.go @@ -2,7 +2,9 @@ package networkmapdb import ( "context" + "errors" "fmt" + "time" log "github.com/sirupsen/logrus" "golang.org/x/exp/maps" @@ -11,12 +13,147 @@ import ( "github.com/netbirdio/netbird/shared/management/networkmap/nmdata" ) +// GetNetworkMapData reads the network map data of the account. With a Replica +// configured it reads from the replica once that has replicated the account's +// current network serial, and from Store when it does not catch up within +// ReplicaWait or the replica read fails. func (s *NetworkMapDBStoreImpl) GetNetworkMapData(ctx context.Context, accountId string) (*networkmap.NetworkMapData, error) { - tx, err := s.Store.BeginTx(ctx) + toret, err := s.readNetworkMapData(ctx, accountId) if err != nil { return nil, err } + extraSettings, err := s.ExtraSettingsManager.GetExtraSettings(ctx, accountId) + if err != nil { + return nil, err + } + + toret.ValidatedPeers, err = s.IntegratedPeerValidator.GetValidatedPeers(ctx, accountId, maps.Values(toret.Groups), maps.Values(toret.Peers), extraSettings) + if err != nil { + return nil, err + } + + return toret, nil +} + +func (s *NetworkMapDBStoreImpl) readNetworkMapData(ctx context.Context, accountId string) (*networkmap.NetworkMapData, error) { + reason := readReasonNoReplica + if s.Replica != nil { + toret, err := s.readFromReplica(ctx, accountId) + if err == nil { + s.Metrics.countRead(ctx, readSourceReplica, readReasonCaughtUp) + return toret, nil + } + if ctx.Err() != nil { + return nil, ctx.Err() + } + reason = fallbackReason(err) + if reason == readReasonLagTimeout { + log.WithContext(ctx).Debugf("reading network map data of account %s from primary: %v", accountId, err) + } else { + log.WithContext(ctx).Warnf("reading network map data of account %s from primary: %v", accountId, err) + } + } + + tx, network, err := beginRead(ctx, s.Store, accountId) + if err != nil { + return nil, err + } + toret, err := readFromTx(ctx, tx, accountId, network) + if err != nil { + return nil, err + } + s.Metrics.countRead(ctx, readSourcePrimary, reason) + return toret, nil +} + +func fallbackReason(err error) string { + switch { + case errors.Is(err, errReplicaLagging): + return readReasonLagTimeout + case errors.Is(err, errPrimarySerialRead): + return readReasonSerialReadError + default: + return readReasonReplicaError + } +} + +func (s *NetworkMapDBStoreImpl) readFromReplica(ctx context.Context, accountId string) (*networkmap.NetworkMapData, error) { + tx, network, err := s.beginOnReplica(ctx, accountId) + if err != nil { + return nil, err + } + return readFromTx(ctx, tx, accountId, network) +} + +// beginOnReplica opens a read transaction on the replica whose snapshot holds at +// least the account's network serial on the primary at the time of the call. +func (s *NetworkMapDBStoreImpl) beginOnReplica(ctx context.Context, accountId string) (NetworkMapDBStoreConn, nmdata.Network, error) { + primaryTx, primaryNetwork, err := beginRead(ctx, s.Store, accountId) + if err != nil { + return nil, nmdata.Network{}, fmt.Errorf("%w: %w", errPrimarySerialRead, err) + } + rollbackTx(ctx, primaryTx) + + start := time.Now() + deadline := start.Add(s.ReplicaWait) + + for { + tx, network, err := beginReadBefore(ctx, s.Replica, accountId, deadline) + if err != nil { + if ctx.Err() == nil && errors.Is(err, context.DeadlineExceeded) { + s.Metrics.recordReplicaWait(ctx, time.Since(start)) + return nil, nmdata.Network{}, fmt.Errorf("%w: no answer within %s: %w", errReplicaLagging, s.ReplicaWait, err) + } + return nil, nmdata.Network{}, err + } + if network.Serial >= primaryNetwork.Serial { + s.Metrics.recordReplicaWait(ctx, time.Since(start)) + return tx, network, nil + } + rollbackTx(ctx, tx) + + remaining := time.Until(deadline) + if remaining <= 0 { + s.Metrics.recordReplicaWait(ctx, time.Since(start)) + return nil, nmdata.Network{}, fmt.Errorf("%w: at network serial %d after %s, primary at %d", errReplicaLagging, network.Serial, s.ReplicaWait, primaryNetwork.Serial) + } + + select { + case <-ctx.Done(): + return nil, nmdata.Network{}, ctx.Err() + case <-time.After(min(replicaPollInterval, remaining)): + } + } +} + +// beginReadBefore runs beginRead with its connect, begin and network read bounded +// by deadline, so a stalled replica cannot hold the read past it. +func beginReadBefore(ctx context.Context, store NetworkMapDBStore, accountId string, deadline time.Time) (NetworkMapDBStoreConn, nmdata.Network, error) { + attemptCtx, cancel := context.WithDeadline(ctx, deadline) + defer cancel() + return beginRead(attemptCtx, store, accountId) +} + +// beginRead opens a read transaction and reads the account's network as its +// first statement, so the network serial belongs to the transaction's snapshot. +func beginRead(ctx context.Context, store NetworkMapDBStore, accountId string) (NetworkMapDBStoreConn, nmdata.Network, error) { + tx, err := store.BeginTx(ctx) + if err != nil { + return nil, nmdata.Network{}, err + } + + network, err := tx.GetNetwork(ctx, accountId) + if err != nil { + rollbackTx(ctx, tx) + return nil, nmdata.Network{}, fmt.Errorf("get network: %w", err) + } + + return tx, network, nil +} + +// readFromTx reads the remaining network map data in tx and ends it. +func readFromTx(ctx context.Context, tx NetworkMapDBStoreConn, accountId string, network nmdata.Network) (*networkmap.NetworkMapData, error) { acctSettings, err := tx.GetAccountSettings(ctx, accountId) if err != nil { return rollbackAndReturnError(ctx, tx, fmt.Errorf("failed to get account settings: %w", err)) @@ -41,10 +178,6 @@ func (s *NetworkMapDBStoreImpl) GetNetworkMapData(ctx context.Context, accountId if err != nil { return rollbackAndReturnError(ctx, tx, fmt.Errorf("failed to get network routers: %w", err)) } - network, err := tx.GetNetwork(ctx, accountId) - if err != nil { - return rollbackAndReturnError(ctx, tx, fmt.Errorf("failed to get network: %w", err)) - } peers, _, err := tx.GetPeers(ctx, accountId) if err != nil { return rollbackAndReturnError(ctx, tx, fmt.Errorf("failed to get peers: %w", err)) @@ -111,26 +244,20 @@ func (s *NetworkMapDBStoreImpl) GetNetworkMapData(ctx context.Context, accountId ProxyTargetedDomainResourceIDs: proxyTargetedDomainResourceIDs, } - extraSettings, err := s.ExtraSettingsManager.GetExtraSettings(ctx, accountId) - if err != nil { - return nil, err - } - - toret.ValidatedPeers, err = s.IntegratedPeerValidator.GetValidatedPeers(ctx, accountId, maps.Values(toret.Groups), maps.Values(toret.Peers), extraSettings) - if err != nil { - return nil, err - } - return &toret, nil } func rollbackAndReturnError(ctx context.Context, tx NetworkMapDBStoreConn, err error) (*networkmap.NetworkMapData, error) { - if errr := tx.RollbackTx(ctx); errr != nil { - log.WithContext(ctx).Warnf("failed to rollback network map read transaction: %v", errr) - } + rollbackTx(ctx, tx) return nil, err } +func rollbackTx(ctx context.Context, tx NetworkMapDBStoreConn) { + if err := tx.RollbackTx(ctx); err != nil { + log.WithContext(ctx).Warnf("failed to rollback network map read transaction: %v", err) + } +} + func toMap[T any](all []T, id func(t T) string) map[string]*T { toret := make(map[string]*T, len(all)) for _, t := range all { diff --git a/management/internals/network_map_db/network_map_data_test.go b/management/internals/network_map_db/network_map_data_test.go index 2b694f5bc..d2d24ecc6 100644 --- a/management/internals/network_map_db/network_map_data_test.go +++ b/management/internals/network_map_db/network_map_data_test.go @@ -1,10 +1,21 @@ package networkmapdb import ( + "context" + "errors" + "sync/atomic" "testing" + "testing/synctest" + "time" - "github.com/netbirdio/netbird/shared/management/networkmap/nmdata" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel/attribute" + sdkmetric "go.opentelemetry.io/otel/sdk/metric" + "go.opentelemetry.io/otel/sdk/metric/metricdata" + + "github.com/netbirdio/netbird/shared/management/networkmap" + "github.com/netbirdio/netbird/shared/management/networkmap/nmdata" ) // disabled network resource shouldn't be in the resulting map @@ -156,3 +167,268 @@ func TestBuildResourcePolicies_ViaPolicyToDestinationGroupIdx(t *testing.T) { }, }, resourceToPolicies) } + +type fakeNmdataStore struct { + name string + serial atomic.Int64 + readErr error + stall bool + begins atomic.Int32 + finishes atomic.Int32 +} + +func (f *fakeNmdataStore) BeginTx(ctx context.Context) (NetworkMapDBStoreConn, error) { + if f.stall { + <-ctx.Done() + return nil, ctx.Err() + } + f.begins.Add(1) + return &fakeNmdataConn{store: f}, nil +} + +func (f *fakeNmdataStore) Exec(context.Context, string, ...any) error { return nil } + +type fakeNmdataConn struct { + store *fakeNmdataStore +} + +func (c *fakeNmdataConn) GetNetwork(context.Context, string) (nmdata.Network, error) { + return nmdata.Network{Identifier: c.store.name, Serial: c.store.serial.Load()}, nil +} + +func (c *fakeNmdataConn) GetPeers(context.Context, string) ([]nmdata.Peer, map[string][]*nmdata.Peer, error) { + return nil, nil, c.store.readErr +} + +func (c *fakeNmdataConn) GetGroups(context.Context, string) ([]nmdata.Group, map[string]map[string]any, error) { + return nil, nil, nil +} +func (c *fakeNmdataConn) GetDomains(context.Context, string) ([]Domain, error) { return nil, nil } +func (c *fakeNmdataConn) GetPolicies(context.Context, string) ([]nmdata.Policy, map[string]map[string]any, map[string]map[string]any, error) { + return nil, nil, nil, nil +} +func (c *fakeNmdataConn) GetRoutes(context.Context, string) ([]nmdata.Route, error) { return nil, nil } +func (c *fakeNmdataConn) GetNameServerGroups(context.Context, string) ([]nmdata.NameServerGroup, error) { + return nil, nil +} +func (c *fakeNmdataConn) GetNetworkResources(context.Context, string) ([]nmdata.NetworkResource, error) { + return nil, nil +} +func (c *fakeNmdataConn) GetNetworkRouters(context.Context, string) (map[string]map[string]*nmdata.NetworkRouter, error) { + return map[string]map[string]*nmdata.NetworkRouter{}, nil +} +func (c *fakeNmdataConn) GetAppliedZoneCandidates(context.Context, string) ([]networkmap.AppliedZoneCandidate, error) { + return nil, nil +} +func (c *fakeNmdataConn) GetAccountSettings(context.Context, string) (nmdata.AccountSettingsInfo, error) { + return nmdata.AccountSettingsInfo{}, nil +} +func (c *fakeNmdataConn) GetPostureChecks(context.Context, string) ([]nmdata.PostureChecks, map[string]string, error) { + return nil, nil, nil +} +func (c *fakeNmdataConn) GetAllowedUsers(context.Context, string) (map[string]struct{}, map[string][]string, error) { + return nil, nil, nil +} +func (c *fakeNmdataConn) GetDnsSettings(context.Context, string) (nmdata.DNSSettings, error) { + return nmdata.DNSSettings{}, nil +} +func (c *fakeNmdataConn) GetNetworkXIDToPublicIdMap(context.Context, string) (map[string]string, error) { + return map[string]string{}, nil +} +func (c *fakeNmdataConn) GetPrivateServices(context.Context, string) ([]Service, error) { + return nil, nil +} +func (c *fakeNmdataConn) GetProxyTargetedDomainResourceIDs(context.Context, string) (map[string]struct{}, error) { + return map[string]struct{}{}, nil +} + +func (c *fakeNmdataConn) CommitTx(context.Context) error { + c.store.finishes.Add(1) + return nil +} + +func (c *fakeNmdataConn) RollbackTx(context.Context) error { + c.store.finishes.Add(1) + return nil +} + +func newReplicaTestStores(primarySerial, replicaSerial int64) (*NetworkMapDBStoreImpl, *fakeNmdataStore, *fakeNmdataStore) { + primary := &fakeNmdataStore{name: "primary"} + primary.serial.Store(primarySerial) + replica := &fakeNmdataStore{name: "replica"} + replica.serial.Store(replicaSerial) + return &NetworkMapDBStoreImpl{Store: primary, Replica: replica, ReplicaWait: time.Second}, primary, replica +} + +func TestReadNetworkMapData_ReplicaCaughtUp(t *testing.T) { + s, primary, replica := newReplicaTestStores(5, 5) + + data, err := s.readNetworkMapData(context.Background(), "account") + require.NoError(t, err) + + assert.Equal(t, "replica", data.Network.Identifier) + assert.Equal(t, int32(1), replica.begins.Load()) + assert.Equal(t, primary.begins.Load(), primary.finishes.Load()) + assert.Equal(t, replica.begins.Load(), replica.finishes.Load()) +} + +func TestReadNetworkMapData_WaitsForReplica(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + s, _, replica := newReplicaTestStores(5, 4) + + time.AfterFunc(100*time.Millisecond, func() { replica.serial.Store(5) }) + + start := time.Now() + data, err := s.readNetworkMapData(context.Background(), "account") + require.NoError(t, err) + + assert.Equal(t, "replica", data.Network.Identifier) + assert.Equal(t, int64(5), data.Network.Serial) + assert.GreaterOrEqual(t, time.Since(start), 100*time.Millisecond) + assert.Equal(t, replica.begins.Load(), replica.finishes.Load()) + }) +} + +func TestReadNetworkMapData_FallsBackToPrimaryWhenReplicaLags(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + s, primary, replica := newReplicaTestStores(5, 4) + + start := time.Now() + data, err := s.readNetworkMapData(context.Background(), "account") + require.NoError(t, err) + + assert.Equal(t, "primary", data.Network.Identifier) + assert.GreaterOrEqual(t, time.Since(start), s.ReplicaWait) + assert.Equal(t, primary.begins.Load(), primary.finishes.Load()) + assert.Equal(t, replica.begins.Load(), replica.finishes.Load()) + }) +} + +func TestReadNetworkMapData_FallsBackToPrimaryOnReplicaReadError(t *testing.T) { + s, primary, replica := newReplicaTestStores(5, 5) + replica.readErr = errors.New("canceling statement due to conflict with recovery") + + data, err := s.readNetworkMapData(context.Background(), "account") + require.NoError(t, err) + + assert.Equal(t, "primary", data.Network.Identifier) + assert.Equal(t, primary.begins.Load(), primary.finishes.Load()) + assert.Equal(t, replica.begins.Load(), replica.finishes.Load()) +} + +func TestReadNetworkMapData_WithoutReplicaReadsPrimaryOnce(t *testing.T) { + primary := &fakeNmdataStore{name: "primary"} + s := &NetworkMapDBStoreImpl{Store: primary} + + data, err := s.readNetworkMapData(context.Background(), "account") + require.NoError(t, err) + + assert.Equal(t, "primary", data.Network.Identifier) + assert.Equal(t, int32(1), primary.begins.Load()) + assert.Equal(t, int32(1), primary.finishes.Load()) +} + +func TestReadNetworkMapData_CancelledWhileWaitingForReplica(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + s, primary, replica := newReplicaTestStores(5, 4) + + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + + _, err := s.readNetworkMapData(ctx, "account") + require.ErrorIs(t, err, context.DeadlineExceeded) + + assert.Equal(t, int32(1), primary.begins.Load(), "primary should only be read for the serial") + assert.Equal(t, primary.begins.Load(), primary.finishes.Load()) + assert.Equal(t, replica.begins.Load(), replica.finishes.Load()) + }) +} + +func TestReadNetworkMapData_FallsBackToPrimaryWhenReplicaStalls(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + reader := sdkmetric.NewManualReader() + metrics, err := NewMetrics(sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader)).Meter("test"), nil) + require.NoError(t, err) + + s, primary, replica := newReplicaTestStores(5, 5) + s.Metrics = metrics + replica.stall = true + + start := time.Now() + data, err := s.readNetworkMapData(context.Background(), "account") + require.NoError(t, err) + + assert.Equal(t, "primary", data.Network.Identifier) + assert.Equal(t, s.ReplicaWait, time.Since(start), "a stalled replica must be abandoned at the wait deadline") + assert.Equal(t, primary.begins.Load(), primary.finishes.Load()) + + reads, waits := collectReadMetrics(t, reader) + assert.Equal(t, map[[2]string]int64{{readSourcePrimary, readReasonLagTimeout}: 1}, reads) + assert.Equal(t, uint64(1), waits) + }) +} + +func TestReadNetworkMapData_CountsReadsBySourceAndReason(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + reader := sdkmetric.NewManualReader() + metrics, err := NewMetrics(sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader)).Meter("test"), nil) + require.NoError(t, err) + + s, _, replica := newReplicaTestStores(5, 5) + s.Metrics = metrics + + _, err = s.readNetworkMapData(context.Background(), "account") + require.NoError(t, err) + + replica.serial.Store(4) + _, err = s.readNetworkMapData(context.Background(), "account") + require.NoError(t, err) + + replica.serial.Store(5) + replica.readErr = errors.New("canceling statement due to conflict with recovery") + _, err = s.readNetworkMapData(context.Background(), "account") + require.NoError(t, err) + + s.Replica = nil + _, err = s.readNetworkMapData(context.Background(), "account") + require.NoError(t, err) + + reads, waits := collectReadMetrics(t, reader) + + assert.Equal(t, map[[2]string]int64{ + {readSourceReplica, readReasonCaughtUp}: 1, + {readSourcePrimary, readReasonLagTimeout}: 1, + {readSourcePrimary, readReasonReplicaError}: 1, + {readSourcePrimary, readReasonNoReplica}: 1, + }, reads) + assert.Equal(t, uint64(3), waits, "every replica attempt that reached the serial check records its wait") + }) +} + +// collectReadMetrics returns the read counts keyed by source and reason and the +// number of recorded replica waits. +func collectReadMetrics(t *testing.T, reader *sdkmetric.ManualReader) (map[[2]string]int64, uint64) { + t.Helper() + var rm metricdata.ResourceMetrics + require.NoError(t, reader.Collect(context.Background(), &rm)) + + reads := map[[2]string]int64{} + var waits uint64 + for _, sm := range rm.ScopeMetrics { + for _, m := range sm.Metrics { + switch data := m.Data.(type) { + case metricdata.Sum[int64]: + for _, dp := range data.DataPoints { + source, _ := dp.Attributes.Value(attribute.Key("source")) + reason, _ := dp.Attributes.Value(attribute.Key("reason")) + reads[[2]string{source.AsString(), reason.AsString()}] = dp.Value + } + case metricdata.Histogram[int64]: + for _, dp := range data.DataPoints { + waits += dp.Count + } + } + } + } + return reads, waits +} diff --git a/management/internals/network_map_db/shared_types.go b/management/internals/network_map_db/shared_types.go index dc065073c..ca92d4eea 100644 --- a/management/internals/network_map_db/shared_types.go +++ b/management/internals/network_map_db/shared_types.go @@ -7,6 +7,7 @@ import ( "errors" "fmt" "reflect" + "time" "github.com/miekg/dns" "github.com/netbirdio/netbird/management/server/integrations/integrated_validator" @@ -17,6 +18,13 @@ import ( var ErrDnsUnsupportedRecordType = errors.New("unsupported record type") +const replicaPollInterval = 10 * time.Millisecond + +var ( + errReplicaLagging = errors.New("replica lagging") + errPrimarySerialRead = errors.New("read primary network serial") +) + type NetworkMapDBStore interface { //nolint:revive // established name across the codebase BeginTx(ctx context.Context) (NetworkMapDBStoreConn, error) Exec(ctx context.Context, query string, args ...any) error @@ -46,7 +54,14 @@ type NetworkMapDBStoreConn interface { //nolint:revive // established name acros } type NetworkMapDBStoreImpl struct { //nolint:revive // established name across the codebase - Store NetworkMapDBStore + Store NetworkMapDBStore + // Replica is an optional read replica of Store that serves reads once it has + // replicated the account's current network serial. + Replica NetworkMapDBStore + // ReplicaWait bounds how long a read waits for Replica before it falls back to Store. + ReplicaWait time.Duration + // Metrics records how reads are served. It is optional. + Metrics *Metrics IntegratedPeerValidator integrated_validator.IntegratedValidator ExtraSettingsManager settings.Manager } diff --git a/management/internals/server/boot.go b/management/internals/server/boot.go index e466005fe..64d5ee4bc 100644 --- a/management/internals/server/boot.go +++ b/management/internals/server/boot.go @@ -8,12 +8,14 @@ import ( "errors" "net/http" "net/netip" + "os" "slices" "time" "github.com/gorilla/mux" grpcMiddleware "github.com/grpc-ecosystem/go-grpc-middleware/v2" "github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/realip" + "github.com/jackc/pgx/v5/pgxpool" "github.com/rs/cors" "github.com/rs/xid" log "github.com/sirupsen/logrus" @@ -44,6 +46,7 @@ import ( "github.com/netbirdio/netbird/management/server/idp" "github.com/netbirdio/netbird/management/server/store" "github.com/netbirdio/netbird/management/server/telemetry" + "github.com/netbirdio/netbird/management/server/types" mgmtProto "github.com/netbirdio/netbird/shared/management/proto" "github.com/netbirdio/netbird/shared/ratelimit" "github.com/netbirdio/netbird/util/crypt" @@ -130,6 +133,8 @@ func (s *BaseServer) NetworkMapStore() *networkmapdb.NetworkMapDBStoreImpl { s.Config.Datadir, s.IntegratedValidator(), s.SettingsManager(), + s.NetworkMapReplicaPool(), + s.Metrics().GetMeter(), ) // networkmap db store supports postgres and sqlite backends only // for other backends a fallback is used, so NotSupportedStoreEngineError @@ -141,6 +146,25 @@ func (s *BaseServer) NetworkMapStore() *networkmapdb.NetworkMapDBStoreImpl { }) } +// NetworkMapReplicaPool opens the pgx pool to the Postgres read replica serving +// network map data reads. It is nil unless the store engine is Postgres and +// networkmapdbfactory.ReplicaDsnEnv is set. +func (s *BaseServer) NetworkMapReplicaPool() *pgxpool.Pool { + return CreateNamed(s, "NetworkMapReplica", func() *pgxpool.Pool { + dsn := os.Getenv(networkmapdbfactory.ReplicaDsnEnv) + if dsn == "" || s.Config.StoreConfig.Engine != types.PostgresStoreEngine { + return nil + } + pool, err := db.OpenPgxPool(context.Background(), dsn, db.DefaultPoolConfig) + if err != nil { + log.Errorf("network map data reads use the primary, connecting to the replica failed: %v", err) + return nil + } + s.OnStop(pool.Close) + return pool + }) +} + func (s *BaseServer) EventStore() activity.Store { return Create(s, func() activity.Store { var err error @@ -180,7 +204,8 @@ func (s *BaseServer) APIHandler() http.Handler { s.GroupsManager(), s.GeoLocationManager(), s.PermissionsManager(), s.SettingsManager(), s.ZonesManager(), s.RecordsManager(), s.NetworkMapController(), s.IdpManager(), s.ServiceManager(), s.ReverseProxyDomainManager(), s.AccessLogsManager(), s.ReverseProxyGRPCServer(), s.Config.ReverseProxy.TrustedHTTPProxies, - s.AgentNetworkManager(), nil) + s.AgentNetworkManager(), nil, + ) if err != nil { log.Fatalf("failed to create API handler: %v", err) } @@ -360,14 +385,30 @@ func (s *BaseServer) ProxyActivityManager() proxyactivity.Manager { }) } +func (s *BaseServer) AccessLogDBConn() *db.Conn { + return CreateNamed(s, "accesslogs", func() *db.Conn { + conn, err := accesslogsmanager.OpenConn(context.Background(), s.Config.Datadir, s.DBConn().Engine()) + if err != nil { + log.Fatalf("failed to open access log database connection: %v", err) + } + s.OnStop(func() { + if err := conn.Close(); err != nil { + log.Debugf("failed to close access log database connection: %v", err) + } + }) + return conn + }) +} + func (s *BaseServer) AccessLogsManager() accesslogs.Manager { return Create(s, func() accesslogs.Manager { - accessLogManager := accesslogsmanager.NewManager(accesslogsmanager.NewRepository(s.DBConn()), s.Store(), s.PermissionsManager(), s.GeoLocationManager()) + accessLogManager := accesslogsmanager.NewManager(accesslogsmanager.NewRepository(s.AccessLogDBConn()), s.Store(), s.PermissionsManager(), s.GeoLocationManager()) accessLogManager.StartPeriodicCleanup( context.Background(), s.Config.ReverseProxy.AccessLogRetentionDays, s.Config.ReverseProxy.AccessLogCleanupIntervalHours, ) + s.OnStop(accessLogManager.StopPeriodicCleanup) return accessLogManager }) } diff --git a/management/internals/shared/db/conn.go b/management/internals/shared/db/conn.go index 8edaa1e4e..b21b066e2 100644 --- a/management/internals/shared/db/conn.go +++ b/management/internals/shared/db/conn.go @@ -24,8 +24,8 @@ type TxMetrics interface { CountTransactionDuration(duration time.Duration) } -// Conn is the database connection shared by all repositories: one gorm handle, -// the pgx pool of a Postgres deployment and the engine they talk to. +// Conn is a database connection the repositories run on: one gorm handle, the +// pgx pool of a Postgres deployment and the engine they talk to. type Conn struct { db *gorm.DB pool *pgxpool.Pool @@ -34,15 +34,36 @@ type Conn struct { metrics TxMetrics } +// Option tunes a Conn as it is opened. +type Option func(*connOptions) + +type connOptions struct { + maxOpenConns int +} + +// WithMaxOpenConns sizes the gorm connection pool of this Conn, taking +// precedence over NB_SQL_MAX_OPEN_CONNS. A pgx pool keeps its PoolConfig size, +// and SQLite always uses one connection. +func WithMaxOpenConns(n int) Option { + return func(o *connOptions) { + o.maxOpenConns = n + } +} + // NewConn takes ownership of an open gorm handle and pool once it returns // without error, applying the connection limits and transaction timeout -// configured through the environment. -func NewConn(ctx context.Context, gormDB *gorm.DB, engine Engine, pool *pgxpool.Pool) (*Conn, error) { +// configured through the environment and opts. +func NewConn(ctx context.Context, gormDB *gorm.DB, engine Engine, pool *pgxpool.Pool, opts ...Option) (*Conn, error) { sqlDB, err := gormDB.DB() if err != nil { return nil, err } + var options connOptions + for _, opt := range opts { + opt(&options) + } + txTimeout := defaultTransactionTimeout if v := os.Getenv("NB_STORE_TRANSACTION_TIMEOUT"); v != "" { if parsed, err := time.ParseDuration(v); err == nil { @@ -57,6 +78,9 @@ func NewConn(ctx context.Context, gormDB *gorm.DB, engine Engine, pool *pgxpool. if connsConfigured { conns = configuredConns } + if options.maxOpenConns > 0 { + conns = options.maxOpenConns + } if engine == SqliteStoreEngine { if connsConfigured { log.WithContext(ctx).Warnf("setting NB_SQL_MAX_OPEN_CONNS is not supported for sqlite, using default value 1") diff --git a/management/internals/shared/db/conn_test.go b/management/internals/shared/db/conn_test.go index 4f1a234b2..cd7488422 100644 --- a/management/internals/shared/db/conn_test.go +++ b/management/internals/shared/db/conn_test.go @@ -146,3 +146,22 @@ func TestNewConn_MaxOpenConnsFromEnv(t *testing.T) { require.NoError(t, err) assert.Equal(t, 1, sqliteDB.Stats().MaxOpenConnections) } + +func TestNewConn_MaxOpenConnsOptionOverridesEnv(t *testing.T) { + t.Setenv("NB_SQL_MAX_OPEN_CONNS", "7") + + open := func(engine Engine, opts ...Option) int { + gormDB, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), "store.db")), GormConfig()) + require.NoError(t, err) + conn, err := NewConn(context.Background(), gormDB, engine, nil, opts...) + require.NoError(t, err) + t.Cleanup(func() { _ = conn.Close() }) + sqlDB, err := conn.DB(nil).DB() + require.NoError(t, err) + return sqlDB.Stats().MaxOpenConnections + } + + assert.Equal(t, 3, open(PostgresStoreEngine, WithMaxOpenConns(3)), "option should take precedence over the environment") + assert.Equal(t, 7, open(PostgresStoreEngine, WithMaxOpenConns(0)), "a non-positive option should fall back to the environment") + assert.Equal(t, 1, open(SqliteStoreEngine, WithMaxOpenConns(3)), "sqlite should stay on a single connection") +} diff --git a/management/internals/shared/db/open.go b/management/internals/shared/db/open.go index c573e8c08..f813b0ce8 100644 --- a/management/internals/shared/db/open.go +++ b/management/internals/shared/db/open.go @@ -101,17 +101,17 @@ func OpenSqliteFile(ctx context.Context, dataDir, storeFile string) (*Conn, erro } // OpenPostgres opens a Postgres database through gorm and a pgx pool sized by pool. -func OpenPostgres(ctx context.Context, dsn string, pool PoolConfig) (*Conn, error) { +func OpenPostgres(ctx context.Context, dsn string, pool PoolConfig, opts ...Option) (*Conn, error) { gormDB, err := gorm.Open(postgres.Open(dsn), GormConfig()) if err != nil { return nil, err } - pgxPool, err := newPgxPool(ctx, dsn, pool) + pgxPool, err := OpenPgxPool(ctx, dsn, pool) if err != nil { closeGorm(gormDB) return nil, err } - conn, err := NewConn(ctx, gormDB, PostgresStoreEngine, pgxPool) + conn, err := NewConn(ctx, gormDB, PostgresStoreEngine, pgxPool, opts...) if err != nil { pgxPool.Close() closeGorm(gormDB) @@ -120,7 +120,23 @@ func OpenPostgres(ctx context.Context, dsn string, pool PoolConfig) (*Conn, erro return conn, nil } -func newPgxPool(ctx context.Context, dsn string, cfg PoolConfig) (*pgxpool.Pool, error) { +// OpenPostgresWithoutPool opens a Postgres database through gorm alone, for +// repositories that never take the pgx read paths. +func OpenPostgresWithoutPool(ctx context.Context, dsn string, opts ...Option) (*Conn, error) { + gormDB, err := gorm.Open(postgres.Open(dsn), GormConfig()) + if err != nil { + return nil, err + } + conn, err := NewConn(ctx, gormDB, PostgresStoreEngine, nil, opts...) + if err != nil { + closeGorm(gormDB) + return nil, err + } + return conn, nil +} + +// OpenPgxPool opens a pgx pool to dsn sized by cfg and verifies it with a ping. +func OpenPgxPool(ctx context.Context, dsn string, cfg PoolConfig) (*pgxpool.Pool, error) { config, err := pgxpool.ParseConfig(dsn) if err != nil { return nil, fmt.Errorf("unable to parse database config: %w", err) diff --git a/management/internals/shared/requestbuffer/buffer.go b/management/internals/shared/requestbuffer/buffer.go index c3823776c..22f9d1952 100644 --- a/management/internals/shared/requestbuffer/buffer.go +++ b/management/internals/shared/requestbuffer/buffer.go @@ -26,6 +26,9 @@ type Buffer[T any] struct { mu sync.Mutex waiting map[string][]chan result[T] + // inFlight holds the keys with a running fetch and whether their next flush + // is due once it ends. It is nil unless fetches are serialized. + inFlight map[string]bool } type result[T any] struct { @@ -45,6 +48,15 @@ func New[T any](ctx context.Context, name string, interval time.Duration, fetch } } +// NewSerialized returns a Buffer like New that runs at most one fetch per key at +// a time. Requests arriving during a fetch form the next batch, which is fetched +// once both its window closed and the running fetch ended. +func NewSerialized[T any](ctx context.Context, name string, interval time.Duration, fetch FetchFunc[T]) *Buffer[T] { + b := New(ctx, name, interval, fetch) + b.inFlight = make(map[string]bool) + return b +} + // Get returns the value for key, sharing one fetch with the other callers of // the current batch. The value is shared as is, so callers must treat it as // read-only unless the fetch hands out copies. @@ -71,8 +83,16 @@ func (b *Buffer[T]) Get(ctx context.Context, key string) (T, error) { func (b *Buffer[T]) flush(key string) { b.mu.Lock() + if _, running := b.inFlight[key]; running { + b.inFlight[key] = true + b.mu.Unlock() + return + } waiting := b.waiting[key] delete(b.waiting, key) + if b.inFlight != nil && len(waiting) > 0 { + b.inFlight[key] = false + } b.mu.Unlock() if len(waiting) == 0 { @@ -86,6 +106,17 @@ func (b *Buffer[T]) flush(key string) { for _, ch := range waiting { ch <- result[T]{value: value, err: err} } + + if b.inFlight == nil { + return + } + b.mu.Lock() + due := b.inFlight[key] + delete(b.inFlight, key) + b.mu.Unlock() + if due { + go b.flush(key) + } } // Interval reads a buffer interval from envVar, falling back to def. diff --git a/management/internals/shared/requestbuffer/buffer_test.go b/management/internals/shared/requestbuffer/buffer_test.go index 9e356145e..3bfb73de4 100644 --- a/management/internals/shared/requestbuffer/buffer_test.go +++ b/management/internals/shared/requestbuffer/buffer_test.go @@ -6,6 +6,7 @@ import ( "sync" "sync/atomic" "testing" + "testing/synctest" "time" "github.com/stretchr/testify/assert" @@ -104,3 +105,73 @@ func TestBufferHonorsCallerContext(t *testing.T) { _, err := buffer.Get(ctx, "account") assert.ErrorIs(t, err, context.DeadlineExceeded) } + +func TestSerializedBufferQueuesRequestsDuringFetch(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + var fetches, running, maxRunning atomic.Int32 + buffer := NewSerialized(context.Background(), "test", 10*time.Millisecond, + func(ctx context.Context, key string) (int32, error) { + n := running.Add(1) + defer running.Add(-1) + if n > maxRunning.Load() { + maxRunning.Store(n) + } + time.Sleep(time.Second) + return fetches.Add(1), nil + }) + + var wg sync.WaitGroup + get := func(want int32) { + wg.Add(1) + go func() { + defer wg.Done() + value, err := buffer.Get(context.Background(), "account") + assert.NoError(t, err) + assert.Equal(t, want, value) + }() + } + + get(1) + time.Sleep(100 * time.Millisecond) + synctest.Wait() + require.Equal(t, int32(1), running.Load(), "first fetch should be running") + + for range 5 { + get(2) + } + wg.Wait() + + assert.Equal(t, int32(2), fetches.Load(), "requests queued during the fetch should share one fetch") + assert.Equal(t, int32(1), maxRunning.Load(), "fetches of a key should not overlap") + }) +} + +func TestUnserializedBufferOverlapsFetches(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + var running, maxRunning atomic.Int32 + buffer := New(context.Background(), "test", 10*time.Millisecond, + func(ctx context.Context, key string) (string, error) { + n := running.Add(1) + defer running.Add(-1) + if n > maxRunning.Load() { + maxRunning.Store(n) + } + time.Sleep(time.Second) + return key, nil + }) + + var wg sync.WaitGroup + wg.Go(func() { + _, err := buffer.Get(context.Background(), "account") + assert.NoError(t, err) + }) + time.Sleep(100 * time.Millisecond) + wg.Go(func() { + _, err := buffer.Get(context.Background(), "account") + assert.NoError(t, err) + }) + wg.Wait() + + assert.Equal(t, int32(2), maxRunning.Load()) + }) +} diff --git a/management/server/account.go b/management/server/account.go index ddc8d4bbc..21b9fa431 100644 --- a/management/server/account.go +++ b/management/server/account.go @@ -1825,10 +1825,10 @@ func (am *DefaultAccountManager) SyncUserJWTGroups(ctx context.Context, userAuth return fmt.Errorf("reconcile IPv6 for group changes: %w", err) } change.ChangedPeerIDs = append(change.ChangedPeerIDs, ipv6Changed...) + } - if err = transaction.IncrementNetworkSerial(ctx, userAuth.AccountId); err != nil { - return fmt.Errorf("error incrementing network serial: %w", err) - } + if err = transaction.IncrementNetworkSerial(ctx, userAuth.AccountId); err != nil { + return fmt.Errorf("error incrementing network serial: %w", err) } if snap, err = affectedpeers.Load(ctx, transaction, userAuth.AccountId, change); err != nil { @@ -2844,6 +2844,10 @@ func (am *DefaultAccountManager) savePeerIPUpdate(ctx context.Context, transacti return fmt.Errorf("save peer: %w", err) } + if err = transaction.IncrementNetworkSerial(ctx, accountID); err != nil { + return fmt.Errorf("increment network serial: %w", err) + } + eventMeta["old_ip"] = oldIP eventMeta["ip"] = newIP.String() am.StoreEvent(ctx, userID, peer.ID, accountID, activity.PeerIPUpdated, eventMeta) @@ -2884,7 +2888,8 @@ func (am *DefaultAccountManager) UpdatePeerIPv6(ctx context.Context, accountID, // updatePeerIPv6InTransaction validates and applies an IPv6 address change within a store transaction. func (am *DefaultAccountManager) updatePeerIPv6InTransaction(ctx context.Context, transaction store.Store, accountID, peerID string, newIPv6 netip.Addr) (bool, error) { - network, err := transaction.GetAccountNetwork(ctx, store.LockingStrengthShare, accountID) + // Locked for update because the serial increment below writes the same row. + network, err := transaction.GetAccountNetwork(ctx, store.LockingStrengthUpdate, accountID) if err != nil { return false, fmt.Errorf("get network: %w", err) } @@ -2928,6 +2933,10 @@ func (am *DefaultAccountManager) updatePeerIPv6InTransaction(ctx context.Context return false, fmt.Errorf("save peer: %w", err) } + if err := transaction.IncrementNetworkSerial(ctx, accountID); err != nil { + return false, fmt.Errorf("increment network serial: %w", err) + } + return true, nil } diff --git a/management/server/account_test.go b/management/server/account_test.go index 7acc72d1f..c343575b8 100644 --- a/management/server/account_test.go +++ b/management/server/account_test.go @@ -3690,7 +3690,7 @@ func createManagerWithNetworkMapStore(t testing.TB) (*DefaultAccountManager, *up store, err := createStoreAt(t, dataDir) require.NoError(t, err) - nmdataStore, err := networkmapdbfactory.NewNetworkMapDBStore(context.Background(), types.SqliteStoreEngine, dataDir, MockIntegratedValidator{}, newSettingsMockManager(t)) + nmdataStore, err := networkmapdbfactory.NewNetworkMapDBStore(context.Background(), types.SqliteStoreEngine, dataDir, MockIntegratedValidator{}, newSettingsMockManager(t), nil, nil) require.NoError(t, err) sqliteStore, ok := nmdataStore.Store.(*networkmap_sqlite.SqliteStore) require.True(t, ok, "network map store is %T, want *networkmap_sqlite.SqliteStore", nmdataStore.Store) @@ -5110,3 +5110,46 @@ func TestDefaultAccountManager_ExpirationJobsStopForDeletedAccount(t *testing.T) }) } } + +func networkSerial(t *testing.T, manager *DefaultAccountManager, accountID string) uint64 { + t.Helper() + + network, err := manager.Store.GetAccountNetwork(context.Background(), store.LockingStrengthNone, accountID) + require.NoError(t, err) + return network.Serial +} + +func TestDefaultAccountManager_UpdatePeerIP_BumpsNetworkSerial(t *testing.T) { + manager, _, account, peer1, peer2, peer3 := setupNetworkMapTest(t) + + newIP, err := types.AllocatePeerIP(netip.MustParsePrefix(account.Network.Net.String()), []netip.Addr{peer1.IP, peer2.IP, peer3.IP}) + require.NoError(t, err) + + before := networkSerial(t, manager, account.Id) + require.NoError(t, manager.UpdatePeerIP(context.Background(), account.Id, userID, peer1.ID, newIP)) + assert.Greater(t, networkSerial(t, manager, account.Id), before) +} + +func TestDefaultAccountManager_SyncUserJWTGroups_BumpsNetworkSerialWithoutPropagation(t *testing.T) { + manager, _, err := createManager(t) + require.NoError(t, err) + + accountID, err := manager.GetAccountIDByUserID(context.Background(), auth.UserAuth{UserId: "jwt-user", Domain: "test.domain"}) + require.NoError(t, err) + acc, err := manager.Store.GetAccount(context.Background(), accountID) + require.NoError(t, err) + acc.Settings.JWTGroupsEnabled = true + acc.Settings.JWTGroupsClaimName = "idp-groups" + acc.Settings.GroupsPropagationEnabled = false + require.NoError(t, manager.Store.SaveAccount(context.Background(), acc)) + + before := networkSerial(t, manager, accountID) + err = manager.SyncUserJWTGroups(context.Background(), auth.UserAuth{ + AccountId: accountID, + Domain: "test.domain", + UserId: "jwt-user", + Groups: []string{"group1"}, + }) + require.NoError(t, err) + assert.Greater(t, networkSerial(t, manager, accountID), before) +} diff --git a/management/server/http/testing/testing_tools/channel/channel.go b/management/server/http/testing/testing_tools/channel/channel.go index 88ab70b9c..326dfb9a0 100644 --- a/management/server/http/testing/testing_tools/channel/channel.go +++ b/management/server/http/testing/testing_tools/channel/channel.go @@ -14,10 +14,12 @@ import ( "github.com/netbirdio/management-integrations/integrations" + "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/accesslogs" accesslogsmanager "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/accesslogs/manager" "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/domain/manager" proxymanager "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/proxy/manager" reverseproxymanager "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/service/manager" + "github.com/netbirdio/netbird/management/internals/shared/db" nbgrpc "github.com/netbirdio/netbird/management/internals/shared/grpc" zonesManager "github.com/netbirdio/netbird/management/internals/modules/zones/manager" @@ -107,7 +109,7 @@ func BuildApiBlackBoxWithDBState(t testing_tools.TB, sqlFile string, expectedPee t.Fatalf("Failed to create manager: %v", err) } - accessLogsManager := accesslogsmanager.NewManager(accesslogsmanager.NewRepository(store.(*nbstore.SqlStore).Conn()), store, permissionsManager, nil) + accessLogsManager := accesslogsmanager.NewManager(newAccessLogsRepository(t), store, permissionsManager, nil) proxyTokenStore := nbgrpc.NewOneTimeTokenStore(ctx, cacheStore) singleUseStore := nbgrpc.NewSingleUseStore(ctx, cacheStore) noopMeter := noop.NewMeterProvider().Meter("") @@ -249,7 +251,7 @@ func BuildApiBlackBoxWithDBStateAndPeerChannel(t testing_tools.TB, sqlFile strin t.Fatalf("Failed to create manager: %v", err) } - accessLogsManager := accesslogsmanager.NewManager(accesslogsmanager.NewRepository(store.(*nbstore.SqlStore).Conn()), store, permissionsManager, nil) + accessLogsManager := accesslogsmanager.NewManager(newAccessLogsRepository(t), store, permissionsManager, nil) proxyTokenStore := nbgrpc.NewOneTimeTokenStore(ctx, cacheStore) singleUseStore := nbgrpc.NewSingleUseStore(ctx, cacheStore) noopMeter := noop.NewMeterProvider().Meter("") @@ -299,6 +301,16 @@ func BuildApiBlackBoxWithDBStateAndPeerChannel(t testing_tools.TB, sqlFile strin return apiHandler, am, updMsg } +func newAccessLogsRepository(t testing_tools.TB) accesslogs.Repository { + t.Helper() + conn, err := accesslogsmanager.OpenConn(context.Background(), t.TempDir(), db.SqliteStoreEngine) + if err != nil { + t.Fatalf("Failed to open access log store: %v", err) + } + t.Cleanup(func() { _ = conn.Close() }) + return accesslogsmanager.NewRepository(conn) +} + func mockValidateAndParseToken(_ context.Context, token string) (auth.UserAuth, *jwt.Token, error) { userAuth := auth.UserAuth{} diff --git a/management/server/peer.go b/management/server/peer.go index 5b33ca9f8..c3454179e 100644 --- a/management/server/peer.go +++ b/management/server/peer.go @@ -103,7 +103,9 @@ func (am *DefaultAccountManager) MarkPeerConnected(ctx context.Context, peerPubK return err } - updated, err := am.Store.MarkPeerConnectedIfNewerSession(ctx, accountID, peer.ID, sessionStartedAt) + updated, err := am.writePeerStatus(ctx, accountID, peer, func(st store.Store) (bool, error) { + return st.MarkPeerConnectedIfNewerSession(ctx, accountID, peer.ID, sessionStartedAt) + }) if err != nil { am.metrics.AccountManagerMetrics().CountPeerStatusUpdate(telemetry.PeerStatusConnect, telemetry.PeerStatusError) return err @@ -140,6 +142,28 @@ func (am *DefaultAccountManager) MarkPeerConnected(ctx context.Context, peerPubK return nil } +// writePeerStatus runs a peer status write. The connection state of an embedded +// proxy peer is part of the network map, so for those peers an applied write also +// increments the account network serial in the same transaction. +func (am *DefaultAccountManager) writePeerStatus(ctx context.Context, accountID string, peer *nbpeer.Peer, write func(store.Store) (bool, error)) (bool, error) { + if !peer.ProxyMeta.Embedded { + return write(am.Store) + } + + var updated bool + err := am.Store.ExecuteInTransaction(ctx, func(transaction store.Store) error { + var err error + if updated, err = write(transaction); err != nil || !updated { + return err + } + if err = transaction.IncrementNetworkSerial(ctx, accountID); err != nil { + return fmt.Errorf("increment network serial: %w", err) + } + return nil + }) + return updated, err +} + // schedulePeerExpirations arms the account's login/inactivity expiration // timers for an SSO peer that just connected. func (am *DefaultAccountManager) schedulePeerExpirations(ctx context.Context, accountID string, peer *nbpeer.Peer) error { @@ -179,7 +203,9 @@ func (am *DefaultAccountManager) MarkPeerDisconnected(ctx context.Context, peerP return err } - updated, err := am.Store.MarkPeerDisconnectedIfSameSession(ctx, accountID, peer.ID, sessionStartedAt) + updated, err := am.writePeerStatus(ctx, accountID, peer, func(st store.Store) (bool, error) { + return st.MarkPeerDisconnectedIfSameSession(ctx, accountID, peer.ID, sessionStartedAt) + }) if err != nil { am.metrics.AccountManagerMetrics().CountPeerStatusUpdate(telemetry.PeerStatusDisconnect, telemetry.PeerStatusError) return err @@ -349,7 +375,7 @@ func (am *DefaultAccountManager) UpdatePeer(ctx context.Context, accountID, user } if err = transaction.IncrementNetworkSerial(ctx, accountID); err != nil { - return fmt.Errorf("failed to increment network serial: %w", err) + return fmt.Errorf("increment network serial: %w", err) } return transaction.SavePeer(ctx, accountID, peer) @@ -618,7 +644,7 @@ func (am *DefaultAccountManager) deletePeerInTransaction(ctx context.Context, ac } if err = transaction.IncrementNetworkSerial(ctx, accountID); err != nil { - return fmt.Errorf("failed to increment network serial: %w", err) + return fmt.Errorf("increment network serial: %w", err) } return nil @@ -956,7 +982,7 @@ func (am *DefaultAccountManager) AddPeer(ctx context.Context, accountID, setupKe err = transaction.IncrementNetworkSerial(ctx, accountID) if err != nil { - return fmt.Errorf("failed to increment network serial: %w", err) + return fmt.Errorf("increment network serial: %w", err) } if ephemeral { @@ -1097,6 +1123,9 @@ func (am *DefaultAccountManager) SyncPeer(ctx context.Context, sync types.PeerSy if err = transaction.SavePeer(ctx, accountID, peer); err != nil { return err } + if err = transaction.IncrementNetworkSerial(ctx, accountID); err != nil { + return fmt.Errorf("increment network serial: %w", err) + } } return nil }) @@ -1272,6 +1301,9 @@ func (am *DefaultAccountManager) LoginPeer(ctx context.Context, login types.Peer if err = transaction.SavePeer(ctx, accountID, peer); err != nil { return err } + if err = transaction.IncrementNetworkSerial(ctx, accountID); err != nil { + return fmt.Errorf("increment network serial: %w", err) + } } return nil diff --git a/management/server/peer_test.go b/management/server/peer_test.go index b75dd1b72..bf8093c1d 100644 --- a/management/server/peer_test.go +++ b/management/server/peer_test.go @@ -3255,3 +3255,71 @@ func TestAddPeer_ExhaustedIPAllocationFails(t *testing.T) { require.Len(t, peers, 1) assert.Equal(t, netip.MustParseAddr("100.64.0.1"), peers[0].IP) } + +func TestPeerWrites_BumpNetworkSerial(t *testing.T) { + manager, _, account, peer1, peer2, _ := setupNetworkMapTest(t) + ctx := context.Background() + + t.Run("SyncPeer with changed meta bumps, unchanged does not", func(t *testing.T) { + current, err := manager.Store.GetPeerByID(ctx, store.LockingStrengthNone, account.Id, peer1.ID) + require.NoError(t, err) + + before := networkSerial(t, manager, account.Id) + _, _, _, _, err = manager.SyncPeer(ctx, types.PeerSync{WireGuardPubKey: peer1.Key, Meta: current.Meta}, account.Id) + require.NoError(t, err) + assert.Equal(t, before, networkSerial(t, manager, account.Id)) + + meta := current.Meta + meta.Hostname = "renamed-host" + _, _, _, _, err = manager.SyncPeer(ctx, types.PeerSync{WireGuardPubKey: peer1.Key, Meta: meta}, account.Id) + require.NoError(t, err) + assert.Greater(t, networkSerial(t, manager, account.Id), before) + }) + + t.Run("LoginPeer with changed ssh key bumps, unchanged does not", func(t *testing.T) { + current, err := manager.Store.GetPeerByID(ctx, store.LockingStrengthNone, account.Id, peer1.ID) + require.NoError(t, err) + + before := networkSerial(t, manager, account.Id) + _, _, _, _, err = manager.LoginPeer(ctx, types.PeerLogin{WireGuardPubKey: peer1.Key, SSHKey: current.SSHKey, Meta: current.Meta}) + require.NoError(t, err) + assert.Equal(t, before, networkSerial(t, manager, account.Id)) + + _, _, _, _, err = manager.LoginPeer(ctx, types.PeerLogin{WireGuardPubKey: peer1.Key, SSHKey: "ssh-ed25519 AAAAnew", Meta: current.Meta}) + require.NoError(t, err) + assert.Greater(t, networkSerial(t, manager, account.Id), before) + }) + + t.Run("MarkPeerConnected and MarkPeerDisconnected", func(t *testing.T) { + session := time.Now().UTC().UnixNano() + + before := networkSerial(t, manager, account.Id) + require.NoError(t, manager.MarkPeerConnected(ctx, peer2.Key, account.Id, session, nil)) + require.NoError(t, manager.MarkPeerDisconnected(ctx, peer2.Key, account.Id, session)) + assert.Equal(t, before, networkSerial(t, manager, account.Id), "regular peers must not bump") + + expired, err := manager.Store.GetPeerByID(ctx, store.LockingStrengthNone, account.Id, peer2.ID) + require.NoError(t, err) + expired.Status.LoginExpired = true + require.NoError(t, manager.Store.SavePeer(ctx, account.Id, expired)) + + session++ + require.NoError(t, manager.MarkPeerConnected(ctx, peer2.Key, account.Id, session, nil)) + afterExpired := networkSerial(t, manager, account.Id) + assert.Equal(t, before, afterExpired, "login-expired reconnect writes no network map data") + + embedded, err := manager.Store.GetPeerByID(ctx, store.LockingStrengthNone, account.Id, peer2.ID) + require.NoError(t, err) + embedded.Status.LoginExpired = false + embedded.ProxyMeta.Embedded = true + require.NoError(t, manager.Store.SavePeer(ctx, account.Id, embedded)) + + require.NoError(t, manager.MarkPeerDisconnected(ctx, peer2.Key, account.Id, session)) + afterDisconnect := networkSerial(t, manager, account.Id) + assert.Greater(t, afterDisconnect, afterExpired, "embedded proxy disconnect must bump") + + session++ + require.NoError(t, manager.MarkPeerConnected(ctx, peer2.Key, account.Id, session, nil)) + assert.Greater(t, networkSerial(t, manager, account.Id), afterDisconnect, "embedded proxy connect must bump") + }) +} diff --git a/management/server/store/sql_store.go b/management/server/store/sql_store.go index 1734129c3..dc1edca88 100644 --- a/management/server/store/sql_store.go +++ b/management/server/store/sql_store.go @@ -12,7 +12,6 @@ import ( nbdns "github.com/netbirdio/netbird/dns" agentNetworkTypes "github.com/netbirdio/netbird/management/internals/modules/agentnetwork/types" - "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/accesslogs" "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/domain" "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/proxy" rpservice "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/service" @@ -83,7 +82,7 @@ func NewSqlStore(ctx context.Context, conn *db.Conn, metrics telemetry.AppMetric &installation{}, &types.ExtraSettings{}, &posture.Checks{}, &nbpeer.NetworkAddress{}, &networkTypes.Network{}, &routerTypes.NetworkRouter{}, &resourceTypes.NetworkResource{}, &types.AccountOnboarding{}, &types.Job{}, &zones.Zone{}, &records.Record{}, &types.UserInviteRecord{}, &rpservice.Service{}, &rpservice.Target{}, &domain.Domain{}, - &accesslogs.AccessLogEntry{}, &proxy.Proxy{}, + &proxy.Proxy{}, &agentNetworkTypes.Provider{}, &agentNetworkTypes.Policy{}, &agentNetworkTypes.Guardrail{}, &agentNetworkTypes.Settings{}, &agentNetworkTypes.Consumption{}, &agentNetworkTypes.AccountBudgetRule{}, &agentNetworkTypes.AgentNetworkAccessLog{}, &agentNetworkTypes.AgentNetworkAccessLogGroup{}, diff --git a/management/server/store/store.go b/management/server/store/store.go index 6d20c1e2b..670bd3956 100644 --- a/management/server/store/store.go +++ b/management/server/store/store.go @@ -460,6 +460,11 @@ func lookupDSNEnv(nbKey, legacyKey string) (string, bool) { return os.LookupEnv(legacyKey) } +// LookupPostgresDSN returns the DSN the main store opens a Postgres engine with. +func LookupPostgresDSN() (string, bool) { + return lookupDSNEnv(PostgresDsnEnv, PostgresDsnEnvLegacy) +} + var supportedEngines = []types.Engine{types.SqliteStoreEngine, types.PostgresStoreEngine} func getStoreEngineFromEnv() types.Engine { @@ -530,7 +535,7 @@ func OpenConn(ctx context.Context, kind types.Engine, dataDir string) (*db.Conn, return db.OpenSqlite(ctx, dataDir) case types.PostgresStoreEngine: log.WithContext(ctx).Info("using Postgres store engine") - dsn, ok := lookupDSNEnv(PostgresDsnEnv, PostgresDsnEnvLegacy) + dsn, ok := LookupPostgresDSN() if !ok { return nil, fmt.Errorf("%s is not set", PostgresDsnEnv) } diff --git a/management/server/user.go b/management/server/user.go index 3e84fcd03..89149367d 100644 --- a/management/server/user.go +++ b/management/server/user.go @@ -670,6 +670,12 @@ func (am *DefaultAccountManager) SaveOrAddUsers(ctx context.Context, accountID, var updatedUsersInfo = make([]*types.UserInfo, 0, len(usersToSave)) + if len(usersToSave) > 0 { + if err = am.Store.IncrementNetworkSerial(ctx, accountID); err != nil { + return nil, fmt.Errorf("increment network serial: %w", err) + } + } + userInfos, err := am.GetUsersFromAccount(ctx, accountID, initiatorUserID) if err != nil { return nil, err @@ -693,9 +699,6 @@ func (am *DefaultAccountManager) SaveOrAddUsers(ctx context.Context, accountID, return nil, err } } else if len(usersToSave) > 0 { - if err = am.Store.IncrementNetworkSerial(ctx, accountID); err != nil { - return nil, fmt.Errorf("failed to increment network serial: %w", err) - } go am.dispatchAffectedPeerIDs(ctx, accountID, affected.ids()) } @@ -1425,6 +1428,10 @@ func (am *DefaultAccountManager) deleteRegularUser(ctx context.Context, accountI return fmt.Errorf("failed to delete user: %s %w", targetUserInfo.ID, err) } + if err = transaction.IncrementNetworkSerial(ctx, accountID); err != nil { + return fmt.Errorf("increment network serial: %w", err) + } + return nil }) if err != nil { diff --git a/management/server/user_test.go b/management/server/user_test.go index 305ed5a24..2e1b51572 100644 --- a/management/server/user_test.go +++ b/management/server/user_test.go @@ -14,6 +14,7 @@ import ( "github.com/netbirdio/netbird/management/internals/controllers/network_map" nbcache "github.com/netbirdio/netbird/management/server/cache" + nbcontext "github.com/netbirdio/netbird/management/server/context" "github.com/netbirdio/netbird/management/server/permissions" "github.com/netbirdio/netbird/management/server/permissions/modules" "github.com/netbirdio/netbird/management/server/permissions/roles" @@ -2352,3 +2353,24 @@ func TestUser_Operations_WithEmbeddedIDP(t *testing.T) { t.Logf("Duplicate email error: %v", err) }) } + +func TestDefaultAccountManager_UserMutations_BumpNetworkSerial(t *testing.T) { + manager, _, account, _, _, _ := setupNetworkMapTest(t) + + account, err := manager.Store.GetAccount(context.Background(), account.Id) + require.NoError(t, err) + account.Users["regular-user"] = types.NewRegularUser("regular-user", "", "") + require.NoError(t, manager.Store.SaveAccount(context.Background(), account)) + + ctx := nbcontext.WithRole(context.Background(), string(types.UserRoleOwner)) + + before := networkSerial(t, manager, account.Id) + _, err = manager.SaveUser(ctx, account.Id, userID, &types.User{Id: "regular-user", Role: types.UserRoleUser, Blocked: true}) + require.NoError(t, err) + afterSave := networkSerial(t, manager, account.Id) + assert.Greater(t, afterSave, before, "SaveUser must bump the serial") + + _, err = manager.deleteRegularUser(ctx, account.Id, userID, &types.UserInfo{ID: "regular-user"}) + require.NoError(t, err) + assert.Greater(t, networkSerial(t, manager, account.Id), afterSave, "deleteRegularUser must bump the serial") +} diff --git a/proxy/Dockerfile b/proxy/Dockerfile index 5944d944e..e4f9a0239 100644 --- a/proxy/Dockerfile +++ b/proxy/Dockerfile @@ -1,4 +1,4 @@ -FROM golang:1.26.7-alpine AS builder +FROM golang:1.26.10-alpine AS builder WORKDIR /app RUN echo "netbird:x:1000:1000:netbird:/var/lib/netbird:/sbin/nologin" > /tmp/passwd && \ diff --git a/proxy/Dockerfile.multistage b/proxy/Dockerfile.multistage index d1db32296..ff46280d2 100644 --- a/proxy/Dockerfile.multistage +++ b/proxy/Dockerfile.multistage @@ -1,4 +1,4 @@ -FROM golang:1.26.7-alpine AS builder +FROM golang:1.26.10-alpine AS builder WORKDIR /app COPY go.mod go.sum ./ diff --git a/proxy/Dockerfile.ubi b/proxy/Dockerfile.ubi index a74280a49..013eb423d 100644 --- a/proxy/Dockerfile.ubi +++ b/proxy/Dockerfile.ubi @@ -1,4 +1,4 @@ -FROM registry.access.redhat.com/ubi9/ubi-minimal@sha256:7fbeae18dc9476399f565e68255f602a3374ea8614ba3d14843565131a13ff93 +FROM registry.access.redhat.com/ubi9/ubi-minimal@sha256:2d4c4551af6831dcd26bbf86108c4385de5183ec8fe6b5868f6f9640fe47f4ed ARG TARGETPLATFORM ARG VERSION=dev diff --git a/relay/Dockerfile.ubi b/relay/Dockerfile.ubi index 86b0cbc01..b4987a424 100644 --- a/relay/Dockerfile.ubi +++ b/relay/Dockerfile.ubi @@ -1,4 +1,4 @@ -FROM registry.access.redhat.com/ubi9/ubi-minimal@sha256:7fbeae18dc9476399f565e68255f602a3374ea8614ba3d14843565131a13ff93 +FROM registry.access.redhat.com/ubi9/ubi-minimal@sha256:2d4c4551af6831dcd26bbf86108c4385de5183ec8fe6b5868f6f9640fe47f4ed ARG TARGETPLATFORM ARG VERSION=dev diff --git a/shared/management/http/api/openapi.yml b/shared/management/http/api/openapi.yml index f129ada26..e64c67ce0 100644 --- a/shared/management/http/api/openapi.yml +++ b/shared/management/http/api/openapi.yml @@ -9381,7 +9381,7 @@ paths: name: search schema: type: string - description: General search across request ID, host, path, source IP, user email, and user name + description: General search across request ID, host, path, and source IP - in: query name: source_ip schema: diff --git a/shared/management/http/api/types.gen.go b/shared/management/http/api/types.gen.go index a5ffe4145..6a7450153 100644 --- a/shared/management/http/api/types.gen.go +++ b/shared/management/http/api/types.gen.go @@ -6051,7 +6051,7 @@ type GetApiEventsProxyParams struct { // SortOrder Sort order (ascending or descending) SortOrder *GetApiEventsProxyParamsSortOrder `form:"sort_order,omitempty" json:"sort_order,omitempty"` - // Search General search across request ID, host, path, source IP, user email, and user name + // Search General search across request ID, host, path, and source IP Search *string `form:"search,omitempty" json:"search,omitempty"` // SourceIp Filter by source IP address diff --git a/signal/Dockerfile.ubi b/signal/Dockerfile.ubi index f0a6552da..5b4036469 100644 --- a/signal/Dockerfile.ubi +++ b/signal/Dockerfile.ubi @@ -1,4 +1,4 @@ -FROM registry.access.redhat.com/ubi9/ubi-minimal@sha256:7fbeae18dc9476399f565e68255f602a3374ea8614ba3d14843565131a13ff93 +FROM registry.access.redhat.com/ubi9/ubi-minimal@sha256:2d4c4551af6831dcd26bbf86108c4385de5183ec8fe6b5868f6f9640fe47f4ed ARG TARGETPLATFORM ARG VERSION=dev diff --git a/upload-server/Dockerfile b/upload-server/Dockerfile index 8098a0186..16fd79871 100644 --- a/upload-server/Dockerfile +++ b/upload-server/Dockerfile @@ -16,13 +16,14 @@ # VARIANT=release|debug debug swaps the base for Chainguard busybox (a shell) # VERSION stamped into the binary the same way goreleaser does # -# Chainguard publishes only :latest for free, so the bases are pinned by digest -# and moved by Dependabot. +# Images are pinned by digest alone. The builder is golang:1.26.10-bookworm and +# the bases are Chainguard's :latest, the only free tag; bump them by hand with +# docker buildx imagetools inspect . ARG VARIANT=release # Pure Go: cross-compile from the build host instead of emulating the target. -FROM --platform=$BUILDPLATFORM golang:1.26.7-bookworm@sha256:e8c859f5632dcfde7b32d2012b4351728f6437930887c2f6a91ea242459e5514 AS builder +FROM --platform=$BUILDPLATFORM golang@sha256:632dff7351799a5a9954a48dc06670ade72ee44e8c351cdc0e5a57af2fea21af AS builder WORKDIR /app COPY go.mod go.sum ./ @@ -39,8 +40,8 @@ RUN --mount=type=cache,target=/go/pkg/mod \ -o /out/netbird-upload ./upload-server \ && mkdir -p /out/var/lib/netbird -FROM cgr.dev/chainguard/static:latest@sha256:41e17ed83c594a64a9396b6ab96dd26d5ddc290dacf4c177464712ff21ad534f AS base-release -FROM cgr.dev/chainguard/busybox:latest@sha256:b2953ab1cae4a6265e18cf675851bd99975211b150d7a014774911d76eb309ba AS base-debug +FROM cgr.dev/chainguard/static@sha256:17367e4d3ae2b46c3290858e483fac09be085c92b372360fe3bf6a6150848fde AS base-release +FROM cgr.dev/chainguard/busybox@sha256:98e9e7306c899a8af252ef1ff50c54def91c2fe04677298eafdf0b902f2980ed AS base-debug # hadolint ignore=DL3006 FROM base-${VARIANT}