Merge branch 'main' into profile-ownership

This commit is contained in:
Theodor S. Midtlien
2026-10-26 11:08:28 +01:00
68 changed files with 2100 additions and 100 deletions
+1 -1
View File
@@ -1,4 +1,4 @@
FROM golang:1.26.7-bookworm FROM golang:1.26.10-bookworm
RUN apt-get update && export DEBIAN_FRONTEND=noninteractive \ RUN apt-get update && export DEBIAN_FRONTEND=noninteractive \
&& apt-get -y install --no-install-recommends\ && apt-get -y install --no-install-recommends\
+9 -1
View File
@@ -244,6 +244,14 @@ jobs:
NB_TEST_PKCS11_DISPOSABLE: "1" NB_TEST_PKCS11_DISPOSABLE: "1"
run: CGO_ENABLED=0 go test -tags pkcs11 -timeout 5m ./client/internal/pkcs11/... ./client/internal/certproof/... 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: test_client_on_docker:
name: "Client (Docker) / Unit" name: "Client (Docker) / Unit"
needs: [build-cache] needs: [build-cache]
@@ -303,7 +311,7 @@ jobs:
-e GOCACHE=${CONTAINER_GOCACHE} \ -e GOCACHE=${CONTAINER_GOCACHE} \
-e GOMODCACHE=${CONTAINER_GOMODCACHE} \ -e GOMODCACHE=${CONTAINER_GOMODCACHE} \
-e CONTAINER=${CONTAINER} \ -e CONTAINER=${CONTAINER} \
golang:1.26.7-alpine \ golang:1.26.10-alpine \
sh -c ' \ sh -c ' \
apk update; apk add --no-cache \ apk update; apk add --no-cache \
ca-certificates iptables ip6tables dbus dbus-dev libpcap-dev build-base; \ ca-certificates iptables ip6tables dbus dbus-dev libpcap-dev build-base; \
+12
View File
@@ -24,6 +24,15 @@ linters:
settings: settings:
errcheck: errcheck:
check-type-assertions: false 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: gocritic:
disabled-checks: disabled-checks:
- commentFormatting - commentFormatting
@@ -86,6 +95,9 @@ linters:
- linters: - linters:
- forbidigo - forbidigo
path: signal/cmd/root\.go path: signal/cmd/root\.go
- linters:
- forbidigo
path: client/ssh/server/setid_unix\.go
- linters: - linters:
- unused - unused
path: sharedsock/filter\.go path: sharedsock/filter\.go
+1 -1
View File
@@ -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 TARGETPLATFORM
ARG NETBIRD_BINARY=$TARGETPLATFORM/netbird ARG NETBIRD_BINARY=$TARGETPLATFORM/netbird
+6 -2
View File
@@ -20,6 +20,7 @@ import (
var ( var (
detailFlag bool detailFlag bool
tableFlag bool
ipv4Flag bool ipv4Flag bool
ipv6Flag bool ipv6Flag bool
jsonFlag bool jsonFlag bool
@@ -44,11 +45,12 @@ func init() {
ipsFilterMap = make(map[string]struct{}) ipsFilterMap = make(map[string]struct{})
prefixNamesFilterMap = 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(&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(&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(&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(&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.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(&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().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") 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 { switch {
case detailFlag: case detailFlag:
statusOutputString = outputInformationHolder.FullDetailSummary() statusOutputString = outputInformationHolder.FullDetailSummary()
case tableFlag:
statusOutputString = outputInformationHolder.PeersTable()
case jsonFlag: case jsonFlag:
statusOutputString, err = outputInformationHolder.JSON() statusOutputString, err = outputInformationHolder.JSON()
case yamlFlag: case yamlFlag:
@@ -264,7 +268,7 @@ func parseFilters() error {
} }
func enableDetailFlagWhenFilterFlag() { func enableDetailFlagWhenFilterFlag() {
if !detailFlag && !jsonFlag && !yamlFlag { if !detailFlag && !jsonFlag && !yamlFlag && !tableFlag {
detailFlag = true detailFlag = true
} }
} }
+59
View File
@@ -4,6 +4,7 @@ import (
"testing" "testing"
"github.com/stretchr/testify/assert" "github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
) )
func TestParsingOfIP(t *testing.T) { func TestParsingOfIP(t *testing.T) {
@@ -13,3 +14,61 @@ func TestParsingOfIP(t *testing.T) {
assert.Equal(t, "192.168.178.123\n", parsedIP) 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)
})
}
}
@@ -1,9 +1,9 @@
FROM golang:1.25-alpine AS build FROM golang:1.26.10-alpine AS build
WORKDIR /app WORKDIR /app
COPY go.mod main.go ./ COPY go.mod main.go ./
RUN CGO_ENABLED=0 go build -o ingest . RUN CGO_ENABLED=0 go build -o ingest .
FROM alpine:3.20 FROM alpine:3.24
RUN adduser -D -H ingest RUN adduser -D -H ingest
COPY --from=build /app/ingest /usr/local/bin/ingest COPY --from=build /app/ingest /usr/local/bin/ingest
USER ingest USER ingest
@@ -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
}
+5 -6
View File
@@ -11,7 +11,6 @@ import (
"path/filepath" "path/filepath"
"runtime" "runtime"
"strings" "strings"
"syscall"
log "github.com/sirupsen/logrus" 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) 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) 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) 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 // validatePrivilegeDropReversibility ensures privileges cannot be restored
func (pd *PrivilegeDropper) validatePrivilegeDropReversibility(targetUID, targetGID uint32, originalUID, originalGID int) error { func (pd *PrivilegeDropper) validatePrivilegeDropReversibility(targetUID, targetGID uint32, originalUID, originalGID int) error {
if originalGID != int(targetGID) { 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) return fmt.Errorf("privilege drop validation failed: able to restore original GID %d", originalGID)
} }
} }
if originalUID != int(targetUID) { 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) return fmt.Errorf("privilege drop validation failed: able to restore original UID %d", originalUID)
} }
} }
+108
View File
@@ -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)) })
}
+25
View File
@@ -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)
}
+55
View File
@@ -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
}
+176
View File
@@ -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)
}
}
+1 -1
View File
@@ -25,7 +25,7 @@ import (
// (.github/workflows/golang-test-linux.yml, test_client_on_docker). // (.github/workflows/golang-test-linux.yml, test_client_on_docker).
const ( const (
containerImage = "golang" containerImage = "golang"
containerTag = "1.26.7-alpine" containerTag = "1.26.10-alpine"
) )
const ( const (
+1 -1
View File
@@ -13,7 +13,7 @@
# docker run --rm -v $(pwd):/app wails-cross windows amd64 # docker run --rm -v $(pwd):/app wails-cross windows amd64
# docker run --rm -v $(pwd):/app wails-cross windows arm64 # docker run --rm -v $(pwd):/app wails-cross windows arm64
FROM golang:1.26.7-bookworm FROM golang:1.26.10-bookworm
ARG TARGETARCH ARG TARGETARCH
+2 -2
View File
@@ -2,7 +2,7 @@
# Multi-stage build for minimal image size # Multi-stage build for minimal image size
# Build stage # Build stage
FROM golang:1.26.7-alpine AS builder FROM golang:1.26.10-alpine AS builder
WORKDIR /app WORKDIR /app
@@ -22,7 +22,7 @@ RUN go mod tidy
RUN go build -tags server -ldflags="-s -w" -o server . RUN go build -tags server -ldflags="-s -w" -o server .
# Runtime stage - minimal image # Runtime stage - minimal image
FROM gcr.io/distroless/static-debian12 FROM gcr.io/distroless/static-debian13
# Copy the binary # Copy the binary
COPY --from=builder /app/server /server COPY --from=builder /app/server /server
+1 -1
View File
@@ -1,4 +1,4 @@
FROM golang:1.26.7-bookworm AS builder FROM golang:1.26.10-bookworm AS builder
WORKDIR /app WORKDIR /app
# Install build dependencies # Install build dependencies
+1 -1
View File
@@ -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 TARGETPLATFORM
ARG VERSION=dev ARG VERSION=dev
+20
View File
@@ -7,6 +7,7 @@ import (
"github.com/stretchr/testify/require" "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" 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, "postgres-dsn", os.Getenv("NB_STORE_ENGINE_POSTGRES_DSN"))
require.Equal(t, "store.db", os.Getenv("NB_STORE_ENGINE_SQLITE_FILE")) 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))
}
+1
View File
@@ -71,6 +71,7 @@ type ServerConfig struct {
Auth AuthConfig `yaml:"auth"` Auth AuthConfig `yaml:"auth"`
Store StoreConfig `yaml:"store"` Store StoreConfig `yaml:"store"`
ActivityStore StoreConfig `yaml:"activityStore"` ActivityStore StoreConfig `yaml:"activityStore"`
AccessLogStore StoreConfig `yaml:"accessLogStore"`
AuthStore StoreConfig `yaml:"authStore"` AuthStore StoreConfig `yaml:"authStore"`
ReverseProxy ReverseProxyConfig `yaml:"reverseProxy"` ReverseProxy ReverseProxyConfig `yaml:"reverseProxy"`
+16
View File
@@ -26,6 +26,7 @@ import (
"github.com/netbirdio/netbird/encryption" "github.com/netbirdio/netbird/encryption"
agentnetworkpricing "github.com/netbirdio/netbird/management/internals/modules/agentnetwork/pricing" 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" mgmtServer "github.com/netbirdio/netbird/management/internals/server"
nbconfig "github.com/netbirdio/netbird/management/internals/server/config" nbconfig "github.com/netbirdio/netbird/management/internals/server/config"
"github.com/netbirdio/netbird/management/server/telemetry" "github.com/netbirdio/netbird/management/server/telemetry"
@@ -152,6 +153,20 @@ func applyActivityStoreEnv(storeConfig StoreConfig) error {
return nil 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 { func initializeConfig() error {
var err error var err error
config, err = LoadConfig(configPath) config, err = LoadConfig(configPath)
@@ -172,6 +187,7 @@ func initializeConfig() error {
if err := applyActivityStoreEnv(config.Server.ActivityStore); err != nil { if err := applyActivityStoreEnv(config.Server.ActivityStore); err != nil {
return err return err
} }
applyAccessLogStoreEnv(config.Server.AccessLogStore)
log.Infof("Starting combined NetBird server") log.Infof("Starting combined NetBird server")
logConfig(config) logConfig(config)
+7
View File
@@ -121,6 +121,13 @@ server:
# dsn: "" # Connection string for postgres # dsn: "" # Connection string for postgres
# file: "" # Custom SQLite file path (optional, defaults to {dataDir}/events.db) # 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) # Auth (embedded IdP) store configuration (optional, defaults to sqlite3 in dataDir/idp.db)
# authStore: # authStore:
# engine: "sqlite3" # sqlite3 or postgres # engine: "sqlite3" # sqlite3 or postgres
+1 -1
View File
@@ -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 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. (`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 bind-mounts the repo and the host Go build/module caches, installs the
required packages, and runs `go test -tags 'devcert privileged'` over the required packages, and runs `go test -tags 'devcert privileged'` over the
client packages. client packages.
+1 -1
View File
@@ -3,7 +3,7 @@
# artifact), so this mirrors its alpine runtime + entrypoint while compiling the # artifact), so this mirrors its alpine runtime + entrypoint while compiling the
# CGO-free client inline. BuildKit cache mounts keep rebuilds incremental. # 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 WORKDIR /src
COPY go.mod go.sum ./ COPY go.mod go.sum ./
RUN --mount=type=cache,target=/go/pkg/mod go mod download RUN --mount=type=cache,target=/go/pkg/mod go mod download
+1 -1
View File
@@ -8,7 +8,7 @@
# that quietly hands it NET_ADMIN would make the suite prove the opposite of # that quietly hands it NET_ADMIN would make the suite prove the opposite of
# what it is for. # what it is for.
FROM golang:1.26.7-bookworm AS builder FROM golang:1.26.10-bookworm AS builder
WORKDIR /src WORKDIR /src
COPY go.mod go.sum ./ COPY go.mod go.sum ./
RUN --mount=type=cache,target=/go/pkg/mod go mod download RUN --mount=type=cache,target=/go/pkg/mod go mod download
+1 -1
View File
@@ -4,7 +4,7 @@ go 1.26.0
// Pin the toolchain to a patch release >= go1.26.2 // Pin the toolchain to a patch release >= go1.26.2
// See https://go.dev/issue/77875. // See https://go.dev/issue/77875.
toolchain go1.26.7 toolchain go1.26.10
require ( require (
cunicu.li/go-rosenpass v0.5.42 cunicu.li/go-rosenpass v0.5.42
@@ -10,14 +10,21 @@ import (
"path/filepath" "path/filepath"
"runtime" "runtime"
"strings" "strings"
"sync"
"testing" "testing"
"time"
"github.com/jackc/pgx/v5/pgxpool"
networkmapdb "github.com/netbirdio/netbird/management/internals/network_map_db" 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/integrations/integrated_validator"
"github.com/netbirdio/netbird/management/server/settings" "github.com/netbirdio/netbird/management/server/settings"
"github.com/netbirdio/netbird/management/server/types" "github.com/netbirdio/netbird/management/server/types"
log "github.com/sirupsen/logrus" log "github.com/sirupsen/logrus"
"github.com/stretchr/testify/assert" "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" "go.uber.org/mock/gomock"
) )
@@ -29,6 +36,19 @@ var goldenNMap string
const EnvUpdateGoldenData = "NMAP_UPDATE_GOLDEN_DATA" 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) { func TestGetNetworkMapData(t *testing.T) {
ctx := context.TODO() ctx := context.TODO()
@@ -51,11 +71,7 @@ func TestGetNetworkMapData(t *testing.T) {
IntegratedPeerValidator: peerValidators, IntegratedPeerValidator: peerValidators,
} }
for _, query := range strings.Split(nmapData, ";") { seedNetworkMapData(t, ctx)
if err := store(t).Exec(ctx, query); err != nil {
log.Fatalf("error initializing nmap test: %s", err.Error())
}
}
nmap, err := storeImpl.GetNetworkMapData(ctx, "account-33") nmap, err := storeImpl.GetNetworkMapData(ctx, "account-33")
assert.NoError(t, err) assert.NoError(t, err)
@@ -72,3 +88,52 @@ func TestGetNetworkMapData(t *testing.T) {
} }
assert.Equal(t, goldenNMap, string(serializedNMap)) 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")
}
+1 -1
View File
@@ -1,4 +1,4 @@
FROM golang:1.26.7-bookworm AS builder FROM golang:1.26.10-bookworm AS builder
WORKDIR /app WORKDIR /app
# Install build dependencies # Install build dependencies
+1 -1
View File
@@ -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 TARGETPLATFORM
ARG VERSION=dev ARG VERSION=dev
@@ -110,7 +110,7 @@ func NewController(ctx context.Context, store store.Store, metrics telemetry.App
if nmdataStore != nil { if nmdataStore != nil {
interval := requestbuffer.Interval(ctx, "NB_NETWORK_MAP_DATA_BUFFER_INTERVAL", defaultNetworkMapDataBufferInterval) 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) 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 return c
@@ -199,6 +199,10 @@ func (m *managerImpl) DeletePeers(ctx context.Context, accountID string, peerIDs
return err 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) log.WithContext(ctx).Debugf("DeletePeers: deleted peer %s", peerID)
if !(peer.ProxyMeta.Embedded || peer.Meta.KernelVersion == "wasm") { if !(peer.ProxyMeta.Embedded || peer.Meta.KernelVersion == "wasm") {
@@ -5,6 +5,8 @@ import (
"fmt" "fmt"
"testing" "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/peer"
"github.com/netbirdio/netbird/management/server/permissions" "github.com/netbirdio/netbird/management/server/permissions"
"github.com/netbirdio/netbird/management/server/permissions/modules" "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.Equal(t, 0, total)
assert.NoError(t, err) 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))
}
@@ -46,7 +46,7 @@ type AccessLogFilter struct {
SortOrder string // Sort order: asc or desc (default: desc) SortOrder string // Sort order: asc or desc (default: desc)
// Filtering parameters // 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 SourceIP *string // Filter by source IP address
Host *string // Filter by host header Host *string // Filter by host header
Path *string // Filter by request path (supports LIKE pattern) Path *string // Filter by request path (supports LIKE pattern)
@@ -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)}
}
@@ -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)
})
}
}
@@ -85,8 +85,8 @@ func applyFilters(query *gorm.DB, filter accesslogs.AccessLogFilter) *gorm.DB {
if filter.Search != nil { if filter.Search != nil {
searchPattern := "%" + *filter.Search + "%" searchPattern := "%" + *filter.Search + "%"
query = query.Where( 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 ?)", "id LIKE ? OR location_connection_ip LIKE ? OR host LIKE ? OR path LIKE ? OR CONCAT(host, path) LIKE ?",
searchPattern, searchPattern, searchPattern, searchPattern, searchPattern, searchPattern, searchPattern, searchPattern, searchPattern, searchPattern, searchPattern, searchPattern,
) )
} }
@@ -123,3 +123,37 @@ func TestSqlRepository_ListByAccount_StatusFilter(t *testing.T) {
assert.Equal(t, []string{"info", "notfound"}, listIDs("failed")) assert.Equal(t, []string{"info", "notfound"}, listIDs("failed"))
assert.Equal(t, []string{"ok"}, listIDs("success")) 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"))
}
@@ -6,9 +6,9 @@ import (
"testing" "testing"
"time" "time"
"go.uber.org/mock/gomock"
"github.com/stretchr/testify/assert" "github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require" "github.com/stretchr/testify/require"
"go.uber.org/mock/gomock"
"github.com/netbirdio/netbird/management/internals/modules/reverseproxy/proxy" "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/proxy"
rpservice "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/service" 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{ err = testStore.SaveAccount(ctx, &types.Account{
Id: testAccountID, Id: testAccountID,
CreatedBy: testUserID, CreatedBy: testUserID,
Network: types.NewNetwork(),
Settings: &types.Settings{ Settings: &types.Settings{
PeerExposeEnabled: true, PeerExposeEnabled: true,
PeerExposeGroups: []string{testGroupID}, PeerExposeGroups: []string{testGroupID},
@@ -363,6 +363,10 @@ func (m *Manager) persistNewService(ctx context.Context, accountID string, svc *
return fmt.Errorf("create service: %w", err) 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 return nil
}) })
} }
@@ -534,6 +538,10 @@ func (m *Manager) persistNewEphemeralService(ctx context.Context, accountID, pee
return fmt.Errorf("create service: %w", err) 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 return nil
}) })
} }
@@ -744,6 +752,10 @@ func (m *Manager) executeServiceUpdate(ctx context.Context, transaction store.St
return fmt.Errorf("update service: %w", err) 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 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) 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 return nil
}) })
if err != 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 return nil
}) })
if err != 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) 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 return nil
}) })
if err != 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 { if err = transaction.DeleteService(ctx, accountID, serviceID); err != nil {
return fmt.Errorf("delete service: %w", err) 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 deleted = true
return nil return nil
@@ -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) { func TestPersistNewService(t *testing.T) {
ctx := context.Background() ctx := context.Background()
accountID := "test-account" accountID := "test-account"
@@ -254,6 +261,9 @@ func TestPersistNewService(t *testing.T) {
txMock.EXPECT(). txMock.EXPECT().
CreateService(ctx, service). CreateService(ctx, service).
Return(nil) Return(nil)
txMock.EXPECT().
IncrementNetworkSerial(ctx, accountID).
Return(nil)
return fn(txMock) return fn(txMock)
}) })
@@ -465,6 +475,9 @@ func TestDeletePeerService_SourcePeerValidation(t *testing.T) {
txMock.EXPECT(). txMock.EXPECT().
DeleteService(ctx, accountID, serviceID). DeleteService(ctx, accountID, serviceID).
Return(nil) Return(nil)
txMock.EXPECT().
IncrementNetworkSerial(ctx, accountID).
Return(nil)
return fn(txMock) return fn(txMock)
}) })
mockStore.EXPECT(). mockStore.EXPECT().
@@ -570,6 +583,9 @@ func TestDeletePeerService_SourcePeerValidation(t *testing.T) {
txMock.EXPECT(). txMock.EXPECT().
DeleteService(ctx, accountID, serviceID). DeleteService(ctx, accountID, serviceID).
Return(nil) Return(nil)
txMock.EXPECT().
IncrementNetworkSerial(ctx, accountID).
Return(nil)
return fn(txMock) return fn(txMock)
}) })
mockStore.EXPECT(). mockStore.EXPECT().
@@ -617,6 +633,9 @@ func TestDeletePeerService_SourcePeerValidation(t *testing.T) {
txMock.EXPECT(). txMock.EXPECT().
DeleteService(ctx, accountID, serviceID). DeleteService(ctx, accountID, serviceID).
Return(nil) Return(nil)
txMock.EXPECT().
IncrementNetworkSerial(ctx, accountID).
Return(nil)
return fn(txMock) return fn(txMock)
}) })
mockStore.EXPECT(). mockStore.EXPECT().
@@ -679,6 +698,7 @@ func setupIntegrationTest(t *testing.T) (*Manager, store.Store) {
err = testStore.SaveAccount(ctx, &types.Account{ err = testStore.SaveAccount(ctx, &types.Account{
Id: testAccountID, Id: testAccountID,
CreatedBy: testUserID, CreatedBy: testUserID,
Network: types.NewNetwork(),
Settings: &types.Settings{ Settings: &types.Settings{
PeerExposeEnabled: true, PeerExposeEnabled: true,
PeerExposeGroups: []string{testGroupID}, 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, rpservice.SourceEphemeral, persisted.Source, "source should be ephemeral")
assert.Equal(t, testPeerID, persisted.SourcePeer, "source peer should be set") assert.Equal(t, testPeerID, persisted.SourcePeer, "source peer should be set")
assert.NotNil(t, persisted.Meta.LastRenewedAt, "last renewed 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) { 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) svc, err := testStore.GetServiceByDomain(ctx, resp.Domain)
require.NoError(t, err) require.NoError(t, err)
serial := networkSerial(t, testStore, testAccountID)
err = mgr.DeleteService(ctx, testAccountID, testUserID, svc.ID) err = mgr.DeleteService(ctx, testAccountID, testUserID, svc.ID)
require.NoError(t, err) 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) count, err = mgr.store.CountEphemeralServicesByPeer(ctx, store.LockingStrengthNone, testAccountID, testPeerID)
require.NoError(t, err) require.NoError(t, err)
@@ -1065,7 +1088,7 @@ func TestDeleteService_DeletesEphemeralExpose(t *testing.T) {
func TestDeleteAllServices_DeletesEphemeralExposes(t *testing.T) { func TestDeleteAllServices_DeletesEphemeralExposes(t *testing.T) {
ctx := context.Background() ctx := context.Background()
mgr, _ := setupIntegrationTest(t) mgr, testStore := setupIntegrationTest(t)
for i := range 3 { for i := range 3 {
_, err := mgr.CreateServiceFromPeer(ctx, testAccountID, testPeerID, &rpservice.ExposeServiceRequest{ _, err := mgr.CreateServiceFromPeer(ctx, testAccountID, testPeerID, &rpservice.ExposeServiceRequest{
@@ -1079,8 +1102,10 @@ func TestDeleteAllServices_DeletesEphemeralExposes(t *testing.T) {
require.NoError(t, err) require.NoError(t, err)
assert.Equal(t, int64(3), count, "all ephemeral services should exist") assert.Equal(t, int64(3), count, "all ephemeral services should exist")
serial := networkSerial(t, testStore, testAccountID)
err = mgr.DeleteAllServices(ctx, testAccountID, testUserID) err = mgr.DeleteAllServices(ctx, testAccountID, testUserID)
require.NoError(t, err) 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) count, err = mgr.store.CountEphemeralServicesByPeer(ctx, store.LockingStrengthNone, testAccountID, testPeerID)
require.NoError(t, err) 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") require.Len(t, targets, 1, "ephemeral peer-exposed service should have exactly one persisted target before reaping")
expireEphemeralService(t, testStore, testAccountID, resp.Domain) expireEphemeralService(t, testStore, testAccountID, resp.Domain)
serial := networkSerial(t, testStore, testAccountID)
err = mgr.deleteExpiredPeerService(ctx, testAccountID, testPeerID, svcID) err = mgr.deleteExpiredPeerService(ctx, testAccountID, testPeerID, svcID)
require.NoError(t, err) require.NoError(t, err)
assert.Equal(t, serial+1, networkSerial(t, testStore, testAccountID), "network serial should be incremented")
_, err = testStore.GetServiceByDomain(ctx, resp.Domain) _, err = testStore.GetServiceByDomain(ctx, resp.Domain)
require.Error(t, err, "expired peer-exposed service should be deleted") 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.NoError(t, err)
require.Len(t, targets, 1, "ephemeral peer-exposed service should have exactly one persisted target before stopping") 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) err = mgr.StopServiceFromPeer(ctx, testAccountID, testPeerID, svcID)
require.NoError(t, err) require.NoError(t, err)
assert.Equal(t, serial+1, networkSerial(t, testStore, testAccountID), "network serial should be incremented")
_, err = testStore.GetServiceByDomain(ctx, resp.Domain) _, err = testStore.GetServiceByDomain(ctx, resp.Domain)
require.Error(t, err, "stopped peer-exposed service should be deleted") 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.Contains(t, err.Error(), "svc-broken")
assert.NotContains(t, err.Error(), "svc-healthy") 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")
}
@@ -5,10 +5,15 @@ import (
"errors" "errors"
"fmt" "fmt"
"os" "os"
"time"
"github.com/jackc/pgx/v5/pgxpool"
"go.opentelemetry.io/otel/metric"
networkmapdb "github.com/netbirdio/netbird/management/internals/network_map_db" networkmapdb "github.com/netbirdio/netbird/management/internals/network_map_db"
networkmap_pgsql "github.com/netbirdio/netbird/management/internals/network_map_db/pgsql" networkmap_pgsql "github.com/netbirdio/netbird/management/internals/network_map_db/pgsql"
networkmap_sqlite "github.com/netbirdio/netbird/management/internals/network_map_db/sqlite" 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/integrations/integrated_validator"
"github.com/netbirdio/netbird/management/server/settings" "github.com/netbirdio/netbird/management/server/settings"
"github.com/netbirdio/netbird/management/server/store" "github.com/netbirdio/netbird/management/server/store"
@@ -16,7 +21,16 @@ import (
log "github.com/sirupsen/logrus" 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") var ErrNotSupportedStoreEngine = errors.New("unsupported store engine")
@@ -25,7 +39,9 @@ func NewNetworkMapDBStore(
kind types.Engine, kind types.Engine,
dataDir string, dataDir string,
integratedPeerValidator integrated_validator.IntegratedValidator, integratedPeerValidator integrated_validator.IntegratedValidator,
extraSettingsManager settings.Manager) (*networkmapdb.NetworkMapDBStoreImpl, error) { extraSettingsManager settings.Manager,
replicaPool *pgxpool.Pool,
meter metric.Meter) (*networkmapdb.NetworkMapDBStoreImpl, error) {
switch kind { switch kind {
case types.SqliteStoreEngine: case types.SqliteStoreEngine:
log.WithContext(ctx).Info("networkmap store is using SQLite") log.WithContext(ctx).Info("networkmap store is using SQLite")
@@ -54,11 +70,32 @@ func NewNetworkMapDBStore(
return nil, err return nil, err
} }
return &networkmapdb.NetworkMapDBStoreImpl{ impl := &networkmapdb.NetworkMapDBStoreImpl{
Store: store, Store: store,
IntegratedPeerValidator: integratedPeerValidator, IntegratedPeerValidator: integratedPeerValidator,
ExtraSettingsManager: extraSettingsManager, 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) return nil, fmt.Errorf("networkmap store doesn't support engine %s, %w", kind, ErrNotSupportedStoreEngine)
@@ -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())
}
@@ -2,7 +2,9 @@ package networkmapdb
import ( import (
"context" "context"
"errors"
"fmt" "fmt"
"time"
log "github.com/sirupsen/logrus" log "github.com/sirupsen/logrus"
"golang.org/x/exp/maps" "golang.org/x/exp/maps"
@@ -11,12 +13,147 @@ import (
"github.com/netbirdio/netbird/shared/management/networkmap/nmdata" "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) { 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 { if err != nil {
return nil, err 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) acctSettings, err := tx.GetAccountSettings(ctx, accountId)
if err != nil { if err != nil {
return rollbackAndReturnError(ctx, tx, fmt.Errorf("failed to get account settings: %w", err)) 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 { if err != nil {
return rollbackAndReturnError(ctx, tx, fmt.Errorf("failed to get network routers: %w", err)) 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) peers, _, err := tx.GetPeers(ctx, accountId)
if err != nil { if err != nil {
return rollbackAndReturnError(ctx, tx, fmt.Errorf("failed to get peers: %w", err)) 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, 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 return &toret, nil
} }
func rollbackAndReturnError(ctx context.Context, tx NetworkMapDBStoreConn, err error) (*networkmap.NetworkMapData, error) { func rollbackAndReturnError(ctx context.Context, tx NetworkMapDBStoreConn, err error) (*networkmap.NetworkMapData, error) {
if errr := tx.RollbackTx(ctx); errr != nil { rollbackTx(ctx, tx)
log.WithContext(ctx).Warnf("failed to rollback network map read transaction: %v", errr)
}
return nil, err 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 { func toMap[T any](all []T, id func(t T) string) map[string]*T {
toret := make(map[string]*T, len(all)) toret := make(map[string]*T, len(all))
for _, t := range all { for _, t := range all {
@@ -1,10 +1,21 @@
package networkmapdb package networkmapdb
import ( import (
"context"
"errors"
"sync/atomic"
"testing" "testing"
"testing/synctest"
"time"
"github.com/netbirdio/netbird/shared/management/networkmap/nmdata"
"github.com/stretchr/testify/assert" "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 // disabled network resource shouldn't be in the resulting map
@@ -156,3 +167,268 @@ func TestBuildResourcePolicies_ViaPolicyToDestinationGroupIdx(t *testing.T) {
}, },
}, resourceToPolicies) }, 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
}
@@ -7,6 +7,7 @@ import (
"errors" "errors"
"fmt" "fmt"
"reflect" "reflect"
"time"
"github.com/miekg/dns" "github.com/miekg/dns"
"github.com/netbirdio/netbird/management/server/integrations/integrated_validator" "github.com/netbirdio/netbird/management/server/integrations/integrated_validator"
@@ -17,6 +18,13 @@ import (
var ErrDnsUnsupportedRecordType = errors.New("unsupported record type") 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 type NetworkMapDBStore interface { //nolint:revive // established name across the codebase
BeginTx(ctx context.Context) (NetworkMapDBStoreConn, error) BeginTx(ctx context.Context) (NetworkMapDBStoreConn, error)
Exec(ctx context.Context, query string, args ...any) 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 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 IntegratedPeerValidator integrated_validator.IntegratedValidator
ExtraSettingsManager settings.Manager ExtraSettingsManager settings.Manager
} }
+43 -2
View File
@@ -8,12 +8,14 @@ import (
"errors" "errors"
"net/http" "net/http"
"net/netip" "net/netip"
"os"
"slices" "slices"
"time" "time"
"github.com/gorilla/mux" "github.com/gorilla/mux"
grpcMiddleware "github.com/grpc-ecosystem/go-grpc-middleware/v2" grpcMiddleware "github.com/grpc-ecosystem/go-grpc-middleware/v2"
"github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/realip" "github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/realip"
"github.com/jackc/pgx/v5/pgxpool"
"github.com/rs/cors" "github.com/rs/cors"
"github.com/rs/xid" "github.com/rs/xid"
log "github.com/sirupsen/logrus" log "github.com/sirupsen/logrus"
@@ -44,6 +46,7 @@ import (
"github.com/netbirdio/netbird/management/server/idp" "github.com/netbirdio/netbird/management/server/idp"
"github.com/netbirdio/netbird/management/server/store" "github.com/netbirdio/netbird/management/server/store"
"github.com/netbirdio/netbird/management/server/telemetry" "github.com/netbirdio/netbird/management/server/telemetry"
"github.com/netbirdio/netbird/management/server/types"
mgmtProto "github.com/netbirdio/netbird/shared/management/proto" mgmtProto "github.com/netbirdio/netbird/shared/management/proto"
"github.com/netbirdio/netbird/shared/ratelimit" "github.com/netbirdio/netbird/shared/ratelimit"
"github.com/netbirdio/netbird/util/crypt" "github.com/netbirdio/netbird/util/crypt"
@@ -130,6 +133,8 @@ func (s *BaseServer) NetworkMapStore() *networkmapdb.NetworkMapDBStoreImpl {
s.Config.Datadir, s.Config.Datadir,
s.IntegratedValidator(), s.IntegratedValidator(),
s.SettingsManager(), s.SettingsManager(),
s.NetworkMapReplicaPool(),
s.Metrics().GetMeter(),
) )
// networkmap db store supports postgres and sqlite backends only // networkmap db store supports postgres and sqlite backends only
// for other backends a fallback is used, so NotSupportedStoreEngineError // 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 { func (s *BaseServer) EventStore() activity.Store {
return Create(s, func() activity.Store { return Create(s, func() activity.Store {
var err error var err error
@@ -180,7 +204,8 @@ func (s *BaseServer) APIHandler() http.Handler {
s.GroupsManager(), s.GeoLocationManager(), s.PermissionsManager(), s.SettingsManager(), s.ZonesManager(), s.GroupsManager(), s.GeoLocationManager(), s.PermissionsManager(), s.SettingsManager(), s.ZonesManager(),
s.RecordsManager(), s.NetworkMapController(), s.IdpManager(), s.ServiceManager(), s.ReverseProxyDomainManager(), s.RecordsManager(), s.NetworkMapController(), s.IdpManager(), s.ServiceManager(), s.ReverseProxyDomainManager(),
s.AccessLogsManager(), s.ReverseProxyGRPCServer(), s.Config.ReverseProxy.TrustedHTTPProxies, s.AccessLogsManager(), s.ReverseProxyGRPCServer(), s.Config.ReverseProxy.TrustedHTTPProxies,
s.AgentNetworkManager(), nil) s.AgentNetworkManager(), nil,
)
if err != nil { if err != nil {
log.Fatalf("failed to create API handler: %v", err) 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 { func (s *BaseServer) AccessLogsManager() accesslogs.Manager {
return Create(s, func() 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( accessLogManager.StartPeriodicCleanup(
context.Background(), context.Background(),
s.Config.ReverseProxy.AccessLogRetentionDays, s.Config.ReverseProxy.AccessLogRetentionDays,
s.Config.ReverseProxy.AccessLogCleanupIntervalHours, s.Config.ReverseProxy.AccessLogCleanupIntervalHours,
) )
s.OnStop(accessLogManager.StopPeriodicCleanup)
return accessLogManager return accessLogManager
}) })
} }
+28 -4
View File
@@ -24,8 +24,8 @@ type TxMetrics interface {
CountTransactionDuration(duration time.Duration) CountTransactionDuration(duration time.Duration)
} }
// Conn is the database connection shared by all repositories: one gorm handle, // Conn is a database connection the repositories run on: one gorm handle, the
// the pgx pool of a Postgres deployment and the engine they talk to. // pgx pool of a Postgres deployment and the engine they talk to.
type Conn struct { type Conn struct {
db *gorm.DB db *gorm.DB
pool *pgxpool.Pool pool *pgxpool.Pool
@@ -34,15 +34,36 @@ type Conn struct {
metrics TxMetrics 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 // NewConn takes ownership of an open gorm handle and pool once it returns
// without error, applying the connection limits and transaction timeout // without error, applying the connection limits and transaction timeout
// configured through the environment. // configured through the environment and opts.
func NewConn(ctx context.Context, gormDB *gorm.DB, engine Engine, pool *pgxpool.Pool) (*Conn, error) { func NewConn(ctx context.Context, gormDB *gorm.DB, engine Engine, pool *pgxpool.Pool, opts ...Option) (*Conn, error) {
sqlDB, err := gormDB.DB() sqlDB, err := gormDB.DB()
if err != nil { if err != nil {
return nil, err return nil, err
} }
var options connOptions
for _, opt := range opts {
opt(&options)
}
txTimeout := defaultTransactionTimeout txTimeout := defaultTransactionTimeout
if v := os.Getenv("NB_STORE_TRANSACTION_TIMEOUT"); v != "" { if v := os.Getenv("NB_STORE_TRANSACTION_TIMEOUT"); v != "" {
if parsed, err := time.ParseDuration(v); err == nil { 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 { if connsConfigured {
conns = configuredConns conns = configuredConns
} }
if options.maxOpenConns > 0 {
conns = options.maxOpenConns
}
if engine == SqliteStoreEngine { if engine == SqliteStoreEngine {
if connsConfigured { if connsConfigured {
log.WithContext(ctx).Warnf("setting NB_SQL_MAX_OPEN_CONNS is not supported for sqlite, using default value 1") log.WithContext(ctx).Warnf("setting NB_SQL_MAX_OPEN_CONNS is not supported for sqlite, using default value 1")
@@ -146,3 +146,22 @@ func TestNewConn_MaxOpenConnsFromEnv(t *testing.T) {
require.NoError(t, err) require.NoError(t, err)
assert.Equal(t, 1, sqliteDB.Stats().MaxOpenConnections) 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")
}
+20 -4
View File
@@ -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. // 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()) gormDB, err := gorm.Open(postgres.Open(dsn), GormConfig())
if err != nil { if err != nil {
return nil, err return nil, err
} }
pgxPool, err := newPgxPool(ctx, dsn, pool) pgxPool, err := OpenPgxPool(ctx, dsn, pool)
if err != nil { if err != nil {
closeGorm(gormDB) closeGorm(gormDB)
return nil, err return nil, err
} }
conn, err := NewConn(ctx, gormDB, PostgresStoreEngine, pgxPool) conn, err := NewConn(ctx, gormDB, PostgresStoreEngine, pgxPool, opts...)
if err != nil { if err != nil {
pgxPool.Close() pgxPool.Close()
closeGorm(gormDB) closeGorm(gormDB)
@@ -120,7 +120,23 @@ func OpenPostgres(ctx context.Context, dsn string, pool PoolConfig) (*Conn, erro
return conn, nil 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) config, err := pgxpool.ParseConfig(dsn)
if err != nil { if err != nil {
return nil, fmt.Errorf("unable to parse database config: %w", err) return nil, fmt.Errorf("unable to parse database config: %w", err)
@@ -26,6 +26,9 @@ type Buffer[T any] struct {
mu sync.Mutex mu sync.Mutex
waiting map[string][]chan result[T] 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 { 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 // 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 // the current batch. The value is shared as is, so callers must treat it as
// read-only unless the fetch hands out copies. // 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) { func (b *Buffer[T]) flush(key string) {
b.mu.Lock() b.mu.Lock()
if _, running := b.inFlight[key]; running {
b.inFlight[key] = true
b.mu.Unlock()
return
}
waiting := b.waiting[key] waiting := b.waiting[key]
delete(b.waiting, key) delete(b.waiting, key)
if b.inFlight != nil && len(waiting) > 0 {
b.inFlight[key] = false
}
b.mu.Unlock() b.mu.Unlock()
if len(waiting) == 0 { if len(waiting) == 0 {
@@ -86,6 +106,17 @@ func (b *Buffer[T]) flush(key string) {
for _, ch := range waiting { for _, ch := range waiting {
ch <- result[T]{value: value, err: err} 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. // Interval reads a buffer interval from envVar, falling back to def.
@@ -6,6 +6,7 @@ import (
"sync" "sync"
"sync/atomic" "sync/atomic"
"testing" "testing"
"testing/synctest"
"time" "time"
"github.com/stretchr/testify/assert" "github.com/stretchr/testify/assert"
@@ -104,3 +105,73 @@ func TestBufferHonorsCallerContext(t *testing.T) {
_, err := buffer.Get(ctx, "account") _, err := buffer.Get(ctx, "account")
assert.ErrorIs(t, err, context.DeadlineExceeded) 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())
})
}
+13 -4
View File
@@ -1825,10 +1825,10 @@ func (am *DefaultAccountManager) SyncUserJWTGroups(ctx context.Context, userAuth
return fmt.Errorf("reconcile IPv6 for group changes: %w", err) return fmt.Errorf("reconcile IPv6 for group changes: %w", err)
} }
change.ChangedPeerIDs = append(change.ChangedPeerIDs, ipv6Changed...) change.ChangedPeerIDs = append(change.ChangedPeerIDs, ipv6Changed...)
}
if err = transaction.IncrementNetworkSerial(ctx, userAuth.AccountId); err != nil { if err = transaction.IncrementNetworkSerial(ctx, userAuth.AccountId); err != nil {
return fmt.Errorf("error incrementing network serial: %w", err) return fmt.Errorf("error incrementing network serial: %w", err)
}
} }
if snap, err = affectedpeers.Load(ctx, transaction, userAuth.AccountId, change); err != nil { 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) 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["old_ip"] = oldIP
eventMeta["ip"] = newIP.String() eventMeta["ip"] = newIP.String()
am.StoreEvent(ctx, userID, peer.ID, accountID, activity.PeerIPUpdated, eventMeta) 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. // 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) { 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 { if err != nil {
return false, fmt.Errorf("get network: %w", err) 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) 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 return true, nil
} }
+44 -1
View File
@@ -3690,7 +3690,7 @@ func createManagerWithNetworkMapStore(t testing.TB) (*DefaultAccountManager, *up
store, err := createStoreAt(t, dataDir) store, err := createStoreAt(t, dataDir)
require.NoError(t, err) 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) require.NoError(t, err)
sqliteStore, ok := nmdataStore.Store.(*networkmap_sqlite.SqliteStore) sqliteStore, ok := nmdataStore.Store.(*networkmap_sqlite.SqliteStore)
require.True(t, ok, "network map store is %T, want *networkmap_sqlite.SqliteStore", nmdataStore.Store) 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)
}
@@ -14,10 +14,12 @@ import (
"github.com/netbirdio/management-integrations/integrations" "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" accesslogsmanager "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/accesslogs/manager"
"github.com/netbirdio/netbird/management/internals/modules/reverseproxy/domain/manager" "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/domain/manager"
proxymanager "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/proxy/manager" proxymanager "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/proxy/manager"
reverseproxymanager "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/service/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" nbgrpc "github.com/netbirdio/netbird/management/internals/shared/grpc"
zonesManager "github.com/netbirdio/netbird/management/internals/modules/zones/manager" 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) 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) proxyTokenStore := nbgrpc.NewOneTimeTokenStore(ctx, cacheStore)
singleUseStore := nbgrpc.NewSingleUseStore(ctx, cacheStore) singleUseStore := nbgrpc.NewSingleUseStore(ctx, cacheStore)
noopMeter := noop.NewMeterProvider().Meter("") noopMeter := noop.NewMeterProvider().Meter("")
@@ -249,7 +251,7 @@ func BuildApiBlackBoxWithDBStateAndPeerChannel(t testing_tools.TB, sqlFile strin
t.Fatalf("Failed to create manager: %v", err) 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) proxyTokenStore := nbgrpc.NewOneTimeTokenStore(ctx, cacheStore)
singleUseStore := nbgrpc.NewSingleUseStore(ctx, cacheStore) singleUseStore := nbgrpc.NewSingleUseStore(ctx, cacheStore)
noopMeter := noop.NewMeterProvider().Meter("") noopMeter := noop.NewMeterProvider().Meter("")
@@ -299,6 +301,16 @@ func BuildApiBlackBoxWithDBStateAndPeerChannel(t testing_tools.TB, sqlFile strin
return apiHandler, am, updMsg 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) { func mockValidateAndParseToken(_ context.Context, token string) (auth.UserAuth, *jwt.Token, error) {
userAuth := auth.UserAuth{} userAuth := auth.UserAuth{}
+37 -5
View File
@@ -103,7 +103,9 @@ func (am *DefaultAccountManager) MarkPeerConnected(ctx context.Context, peerPubK
return err 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 { if err != nil {
am.metrics.AccountManagerMetrics().CountPeerStatusUpdate(telemetry.PeerStatusConnect, telemetry.PeerStatusError) am.metrics.AccountManagerMetrics().CountPeerStatusUpdate(telemetry.PeerStatusConnect, telemetry.PeerStatusError)
return err return err
@@ -140,6 +142,28 @@ func (am *DefaultAccountManager) MarkPeerConnected(ctx context.Context, peerPubK
return nil 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 // schedulePeerExpirations arms the account's login/inactivity expiration
// timers for an SSO peer that just connected. // timers for an SSO peer that just connected.
func (am *DefaultAccountManager) schedulePeerExpirations(ctx context.Context, accountID string, peer *nbpeer.Peer) error { 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 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 { if err != nil {
am.metrics.AccountManagerMetrics().CountPeerStatusUpdate(telemetry.PeerStatusDisconnect, telemetry.PeerStatusError) am.metrics.AccountManagerMetrics().CountPeerStatusUpdate(telemetry.PeerStatusDisconnect, telemetry.PeerStatusError)
return err return err
@@ -349,7 +375,7 @@ func (am *DefaultAccountManager) UpdatePeer(ctx context.Context, accountID, user
} }
if err = transaction.IncrementNetworkSerial(ctx, accountID); err != nil { 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) 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 { 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 return nil
@@ -956,7 +982,7 @@ func (am *DefaultAccountManager) AddPeer(ctx context.Context, accountID, setupKe
err = transaction.IncrementNetworkSerial(ctx, accountID) err = transaction.IncrementNetworkSerial(ctx, accountID)
if err != nil { if err != nil {
return fmt.Errorf("failed to increment network serial: %w", err) return fmt.Errorf("increment network serial: %w", err)
} }
if ephemeral { 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 { if err = transaction.SavePeer(ctx, accountID, peer); err != nil {
return err return err
} }
if err = transaction.IncrementNetworkSerial(ctx, accountID); err != nil {
return fmt.Errorf("increment network serial: %w", err)
}
} }
return nil 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 { if err = transaction.SavePeer(ctx, accountID, peer); err != nil {
return err return err
} }
if err = transaction.IncrementNetworkSerial(ctx, accountID); err != nil {
return fmt.Errorf("increment network serial: %w", err)
}
} }
return nil return nil
+68
View File
@@ -3255,3 +3255,71 @@ func TestAddPeer_ExhaustedIPAllocationFails(t *testing.T) {
require.Len(t, peers, 1) require.Len(t, peers, 1)
assert.Equal(t, netip.MustParseAddr("100.64.0.1"), peers[0].IP) 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")
})
}
+1 -2
View File
@@ -12,7 +12,6 @@ import (
nbdns "github.com/netbirdio/netbird/dns" nbdns "github.com/netbirdio/netbird/dns"
agentNetworkTypes "github.com/netbirdio/netbird/management/internals/modules/agentnetwork/types" 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/domain"
"github.com/netbirdio/netbird/management/internals/modules/reverseproxy/proxy" "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/proxy"
rpservice "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/service" 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{}, &installation{}, &types.ExtraSettings{}, &posture.Checks{}, &nbpeer.NetworkAddress{},
&networkTypes.Network{}, &routerTypes.NetworkRouter{}, &resourceTypes.NetworkResource{}, &types.AccountOnboarding{}, &networkTypes.Network{}, &routerTypes.NetworkRouter{}, &resourceTypes.NetworkResource{}, &types.AccountOnboarding{},
&types.Job{}, &zones.Zone{}, &records.Record{}, &types.UserInviteRecord{}, &rpservice.Service{}, &rpservice.Target{}, &domain.Domain{}, &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.Provider{}, &agentNetworkTypes.Policy{}, &agentNetworkTypes.Guardrail{}, &agentNetworkTypes.Settings{},
&agentNetworkTypes.Consumption{}, &agentNetworkTypes.AccountBudgetRule{}, &agentNetworkTypes.Consumption{}, &agentNetworkTypes.AccountBudgetRule{},
&agentNetworkTypes.AgentNetworkAccessLog{}, &agentNetworkTypes.AgentNetworkAccessLogGroup{}, &agentNetworkTypes.AgentNetworkAccessLog{}, &agentNetworkTypes.AgentNetworkAccessLogGroup{},
+6 -1
View File
@@ -460,6 +460,11 @@ func lookupDSNEnv(nbKey, legacyKey string) (string, bool) {
return os.LookupEnv(legacyKey) 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} var supportedEngines = []types.Engine{types.SqliteStoreEngine, types.PostgresStoreEngine}
func getStoreEngineFromEnv() types.Engine { 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) return db.OpenSqlite(ctx, dataDir)
case types.PostgresStoreEngine: case types.PostgresStoreEngine:
log.WithContext(ctx).Info("using Postgres store engine") log.WithContext(ctx).Info("using Postgres store engine")
dsn, ok := lookupDSNEnv(PostgresDsnEnv, PostgresDsnEnvLegacy) dsn, ok := LookupPostgresDSN()
if !ok { if !ok {
return nil, fmt.Errorf("%s is not set", PostgresDsnEnv) return nil, fmt.Errorf("%s is not set", PostgresDsnEnv)
} }
+10 -3
View File
@@ -670,6 +670,12 @@ func (am *DefaultAccountManager) SaveOrAddUsers(ctx context.Context, accountID,
var updatedUsersInfo = make([]*types.UserInfo, 0, len(usersToSave)) 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) userInfos, err := am.GetUsersFromAccount(ctx, accountID, initiatorUserID)
if err != nil { if err != nil {
return nil, err return nil, err
@@ -693,9 +699,6 @@ func (am *DefaultAccountManager) SaveOrAddUsers(ctx context.Context, accountID,
return nil, err return nil, err
} }
} else if len(usersToSave) > 0 { } 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()) 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) 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 return nil
}) })
if err != nil { if err != nil {
+22
View File
@@ -14,6 +14,7 @@ import (
"github.com/netbirdio/netbird/management/internals/controllers/network_map" "github.com/netbirdio/netbird/management/internals/controllers/network_map"
nbcache "github.com/netbirdio/netbird/management/server/cache" 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"
"github.com/netbirdio/netbird/management/server/permissions/modules" "github.com/netbirdio/netbird/management/server/permissions/modules"
"github.com/netbirdio/netbird/management/server/permissions/roles" "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) 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")
}
+1 -1
View File
@@ -1,4 +1,4 @@
FROM golang:1.26.7-alpine AS builder FROM golang:1.26.10-alpine AS builder
WORKDIR /app WORKDIR /app
RUN echo "netbird:x:1000:1000:netbird:/var/lib/netbird:/sbin/nologin" > /tmp/passwd && \ RUN echo "netbird:x:1000:1000:netbird:/var/lib/netbird:/sbin/nologin" > /tmp/passwd && \
+1 -1
View File
@@ -1,4 +1,4 @@
FROM golang:1.26.7-alpine AS builder FROM golang:1.26.10-alpine AS builder
WORKDIR /app WORKDIR /app
COPY go.mod go.sum ./ COPY go.mod go.sum ./
+1 -1
View File
@@ -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 TARGETPLATFORM
ARG VERSION=dev ARG VERSION=dev
+1 -1
View File
@@ -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 TARGETPLATFORM
ARG VERSION=dev ARG VERSION=dev
+1 -1
View File
@@ -9381,7 +9381,7 @@ paths:
name: search name: search
schema: schema:
type: string 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 - in: query
name: source_ip name: source_ip
schema: schema:
+1 -1
View File
@@ -6051,7 +6051,7 @@ type GetApiEventsProxyParams struct {
// SortOrder Sort order (ascending or descending) // SortOrder Sort order (ascending or descending)
SortOrder *GetApiEventsProxyParamsSortOrder `form:"sort_order,omitempty" json:"sort_order,omitempty"` 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"` Search *string `form:"search,omitempty" json:"search,omitempty"`
// SourceIp Filter by source IP address // SourceIp Filter by source IP address
+1 -1
View File
@@ -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 TARGETPLATFORM
ARG VERSION=dev ARG VERSION=dev
+6 -5
View File
@@ -16,13 +16,14 @@
# VARIANT=release|debug debug swaps the base for Chainguard busybox (a shell) # VARIANT=release|debug debug swaps the base for Chainguard busybox (a shell)
# VERSION stamped into the binary the same way goreleaser does # VERSION stamped into the binary the same way goreleaser does
# #
# Chainguard publishes only :latest for free, so the bases are pinned by digest # Images are pinned by digest alone. The builder is golang:1.26.10-bookworm and
# and moved by Dependabot. # the bases are Chainguard's :latest, the only free tag; bump them by hand with
# docker buildx imagetools inspect <image:tag>.
ARG VARIANT=release ARG VARIANT=release
# Pure Go: cross-compile from the build host instead of emulating the target. # 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 WORKDIR /app
COPY go.mod go.sum ./ COPY go.mod go.sum ./
@@ -39,8 +40,8 @@ RUN --mount=type=cache,target=/go/pkg/mod \
-o /out/netbird-upload ./upload-server \ -o /out/netbird-upload ./upload-server \
&& mkdir -p /out/var/lib/netbird && mkdir -p /out/var/lib/netbird
FROM cgr.dev/chainguard/static:latest@sha256:41e17ed83c594a64a9396b6ab96dd26d5ddc290dacf4c177464712ff21ad534f AS base-release FROM cgr.dev/chainguard/static@sha256:17367e4d3ae2b46c3290858e483fac09be085c92b372360fe3bf6a6150848fde AS base-release
FROM cgr.dev/chainguard/busybox:latest@sha256:b2953ab1cae4a6265e18cf675851bd99975211b150d7a014774911d76eb309ba AS base-debug FROM cgr.dev/chainguard/busybox@sha256:98e9e7306c899a8af252ef1ff50c54def91c2fe04677298eafdf0b902f2980ed AS base-debug
# hadolint ignore=DL3006 # hadolint ignore=DL3006
FROM base-${VARIANT} FROM base-${VARIANT}