Files
Abhishek Kondur 292af315dd NM-271: Scalability Improvements (#3921)
* feat(go): add user schema;

* feat(go): migrate to user schema;

* feat(go): add audit fields;

* feat(go): remove unused fields from the network model;

* feat(go): add network schema;

* feat(go): migrate to network schema;

* refactor(go): add comment to clarify migration logic;

* fix(go): test failures;

* fix(go): test failures;

* feat(go): change membership table to store memberships at all scopes;

* feat(go): add schema for access grants;

* feat(go): remove nameservers from new networks table; ensure db passed for schema functions;

* feat(go): set max conns for sqlite to 1;

* fix(go): issues updating user account status;

* NM-236: streamline operations in HA mode

* NM-236: only master pod should subscribe to updates from clients

* refactor(go): remove converters and access grants;

* refactor(go): add json tags in schema models;

* refactor(go): rename file to migrate_v1_6_0.go;

* refactor(go): add user groups and user roles tables; use schema tables;

* refactor(go): inline get and list from schema package;

* refactor(go): inline get network and list users from schema package;

* fix(go): staticcheck issues;

* fix(go): remove test not in use; fix test case;

* fix(go): validate network;

* fix(go): resolve static checks;

* fix(go): new models errors;

* fix(go): test errors;

* fix(go): handle no records;

* fix(go): add validations for user object;

* fix(go): set correct extclient status;

* fix(go): test error;

* feat(go): make schema the base package;

* feat(go): add host schema;

* feat(go): use schema host everywhere;

* feat(go): inline get host, list hosts and delete host;

* feat(go): use non-ptr value;

* feat(go): use save to upsert all fields;

* feat(go): use save to upsert all fields;

* feat(go): save turn endpoint as string;

* feat(go): check for gorm error record not found;

* fix(go): test failures;

* fix(go): update all network fields;

* fix(go): update all network fields;

* feat(go): add paginated list networks api;

* feat(go): add paginated list users api;

* feat(go): add paginated list hosts api;

* feat(go): add pagination to list groups api;

* fix(go): comment;

* fix(go): implement marshal and unmarshal text for custom types;

* fix(go): implement marshal and unmarshal json for custom types;

* fix(go): just use the old model for unmarshalling;

* fix(go): implement marshal and unmarshal json for custom types;

* NM-271:Import swap: compress/gzip replaced with github.com/klauspost/compress/gzip (2-4x faster, wire-compatible output). Added sync import.
Two sync.Pool variables (gzipWriterPool, bufferPool): reuse gzip.Writer and bytes.Buffer across calls instead of allocating fresh ones per publish.
compressPayload rewritten: pulls writer + buffer from pools, resets them, compresses at gzip.BestSpeed (level 1), copies the result out of the pooled buffer, and returns both objects to the pools.

* feat(go): remove paginated list networks api;

* feat(go): use custom paginated response object;

* NM-271: Improve server scalability under high host count

- Replace stdlib compress/gzip with klauspost/compress at BestSpeed and
  pool gzip writers and buffers via sync.Pool to eliminate compression
  as the dominant CPU hotspot.

- Debounce peer update broadcasts with a 500ms resettable window capped
  at 3s max-wait, coalescing rapid-fire PublishPeerUpdate calls into a
  single broadcast cycle.

- Cache HostPeerInfo (batch-refreshed by debounce worker) and
  HostPeerUpdate (stored as side-effect of each publish) so the pull API
  and peer_info API serve from pre-computed maps instead of triggering
  expensive per-host computations under thundering herd conditions.

- Warm both caches synchronously at startup before the first publish
  cycle so early pull requests are served instantly.

- Bound concurrent MQTT publishes to 5 via semaphore to prevent
  broker TCP buffer overflows that caused broken pipe disconnects.

- Remove manual Disconnect+SetupMQTT from ConnectionLostHandler and
  rely on the paho client's built-in AutoReconnect; add a 5s retry
  wait in publish() to ride out brief reconnection windows.

* NM-271: Reduce server CPU contention under high concurrent load

- Cache ServerSettings with atomic.Value to eliminate repeated DB reads
  on every pull request (was 32+ goroutines blocked on read lock)
- Batch UpdateNodeCheckin writes in memory, flush every 30s to reduce
  per-checkin write lock contention (was 88+ goroutines blocked)
- Enable SQLite WAL mode + busy_timeout and remove global dbMutex;
  let SQLite handle concurrency natively (reads no longer block writes)
- Move ResetFailedOverPeer/ResetAutoRelayedPeer to async in pull()
  handler since results don't affect the cached response
- Skip no-op UpsertNode writes in failover/relay reset functions
  (early return when node has no failover/relay state)
- Remove CheckHostPorts from hostUpdateFallback hot path
- Switch to pure-Go SQLite driver (glebarez/sqlite), set CGO_ENABLED=0

* fix(go): ensure default values for page and per_page are used when not passed;

* fix(go): rename v1.6.0 to v1.5.1;

* fix(go): check for gorm.ErrRecordNotFound instead of database.IsEmptyRecord;

* fix(go): use host id, not pending host id;

* NM-271: Revert pure-Go SQLite and FIPS disable to verify impact

Revert to CGO-based mattn/go-sqlite3 driver and re-enable FIPS to
isolate whether these changes are still needed now that the global
dbMutex has been removed and WAL mode is enabled. Keep WAL mode
pragma with mattn-compatible DSN format.

* feat(go): add filters to paginated apis;

* feat(go): add filters to paginated apis;

* feat(go): remove check for max username length;

* feat(go): add filters to count as well;

* feat(go): use library to check email address validity;

* feat(go): ignore pagination if params not passed;

* fix(go): pagination issues;

* fix(go): check exists before using;

* fix(go): remove debug log;

* NM-271: rm debug logs

* NM-271: check if caching is enabled

* NM-271: add server sync mq topic for HA mode

* NM-271: fix build

* NM-271: push metrics in batch to exproter over api

* NM-271: use basic auth for exporter metrics api

* fix(go): use gorm err record not found;

* NM-271: Add monitoring stack on demand

* NM-271: -m arg for install script should only add monitoring stack

* fix(go): use gorm err record not found;

* NM-271: update docker compose file for prometheus

* NM-271: update docker compose file for prometheus

* fix(go): use user principal name when creating pending user;

* fix(go): use schema package for consts;

* NM-236: rm duplicate network hook

* NM-271: add server topic to reset idp hooks on master node

* fix(go): prevent disabling superadmin user;

Co-authored-by: tenki-reviewer[bot] <262613592+tenki-reviewer[bot]@users.noreply.github.com>

* fix(go): swap is admin and is superadmin;

Co-authored-by: tenki-reviewer[bot] <262613592+tenki-reviewer[bot]@users.noreply.github.com>

* fix(go): remove dead code block;

https://github.com/gravitl/netmaker/pull/3910#discussion_r2928837937

* fix(go): incorrect message when trying to disable self;

https://github.com/gravitl/netmaker/pull/3910#discussion_r2928837934

* NM-271: fix stale peers on reset_failovered pull and add HTTP timeout to metrics exporter

Run the failover/relay reset synchronously in the pull handler so the
response reflects post-reset topology instead of serving stale cached
peers. Add a 30s timeout to the metrics exporter HTTP client to prevent
PushAllMetricsToExporter from blocking the Keepalive loop.

* NM-271: fix gzip pool corruption, MQTT topic mismatch, stale settings cache, and reduce redundant DB fetches

- Only return gzip.Writer to pool after successful Close to prevent
  silently malformed MQTT payloads from a previously errored writer.
- Fix serversync subscription to exact topic match since syncType is
  now in the message payload, not the topic path.
- Prevent zero-value ServerSettings from being cached indefinitely
  when the DB record is missing or unmarshal fails on startup.
- Return fetched hosts/nodes from RefreshHostPeerInfoCache so
  warmPeerCaches reuses them instead of querying the DB twice.
- Compute fresh HostPeerUpdate on reset_failovered pull instead of
  serving stale cache, and store result back for subsequent requests.

* NM-271: fix gzip writer pool leak, log checkin flush errors, and fix master pod ordinal parsing

- Reset gzip.Writer to io.Discard before returning to pool so errored
  writers are never leaked or silently reused with corrupt state.
- Track and log failed DB inserts in FlushNodeCheckins so operators
  have visibility when check-in timestamps are lost.
- Parse StatefulSet pod ordinal as integer instead of using HasSuffix
  to prevent netmaker-10 from being misidentified as master pod.

* NM-271: simplify masterpod logic

* fix(go): use correct header;

Co-authored-by: tenki-reviewer[bot] <262613592+tenki-reviewer[bot]@users.noreply.github.com>

* fix(go): return after error response;

Co-authored-by: tenki-reviewer[bot] <262613592+tenki-reviewer[bot]@users.noreply.github.com>

* fix(go): use correct order of params;

https://github.com/gravitl/netmaker/pull/3910#discussion_r2929593036

* fix(go): set default values for page and page size; use v2 instead of /list;

* NM-271: use host name

* Update mq/serversync.go

Co-authored-by: tenki-reviewer[bot] <262613592+tenki-reviewer[bot]@users.noreply.github.com>

* NM-271: fix duplicate serversynce case

* NM-271: streamline gw updates

* Update logic/auth.go

Co-authored-by: tenki-reviewer[bot] <262613592+tenki-reviewer[bot]@users.noreply.github.com>

* Update schema/user_roles.go

Co-authored-by: tenki-reviewer[bot] <262613592+tenki-reviewer[bot]@users.noreply.github.com>

* fix(go): syntax error;

* fix(go): set default values when page and per_page are not passed or 0;

* fix(go): use uuid.parse instead of uuid.must parse;

* fix(go): review errors;

* fix(go): review errors;

* Update controllers/user.go

Co-authored-by: tenki-reviewer[bot] <262613592+tenki-reviewer[bot]@users.noreply.github.com>

* Update controllers/user.go

Co-authored-by: tenki-reviewer[bot] <262613592+tenki-reviewer[bot]@users.noreply.github.com>

* NM-163: fix errors:

* Update db/types/options.go

Co-authored-by: tenki-reviewer[bot] <262613592+tenki-reviewer[bot]@users.noreply.github.com>

* fix(go): persist return user in event;

* Update db/types/options.go

Co-authored-by: tenki-reviewer[bot] <262613592+tenki-reviewer[bot]@users.noreply.github.com>

* NM-271: signal pull on ip changes

* NM-163: duplicate lines of code

* NM-163: fix(go): fix missing return and filter parsing in user controller

- Add missing return after error response in updateUserAccountStatus
  to prevent double-response and spurious ext-client side-effects
- Use switch statements in listUsers to skip unrecognized
  account_status and mfa_status filter values

* NM-271: signal pull req on node ip change

* fix(go): check for both min and max page size;

* NM-271: refresh node object before update

* fix(go): enclose transfer superadmin in transaction;

* fix(go): review errors;

* fix(go): remove free tier checks;

* fix(go): review fixes;

* NM-271: streamline ip pool ops

* NM-271: fix tests, set max idle conns

* NM-271: fix(go): fix data races in settings cache and peer update worker

- Use pointer type in atomic.Value for serverSettingsCache to avoid
  replacing the variable non-atomically in InvalidateServerSettingsCache
- Swap peerUpdateReplace flag before draining the channel to prevent
  a concurrent replacePeers=true from being consumed by the wrong cycle

---------

Co-authored-by: VishalDalwadi <dalwadivishal26@gmail.com>
Co-authored-by: Vishal Dalwadi <51291657+VishalDalwadi@users.noreply.github.com>
Co-authored-by: tenki-reviewer[bot] <262613592+tenki-reviewer[bot]@users.noreply.github.com>
2026-03-18 00:24:54 +05:30

216 lines
6.3 KiB
Go

package database
import (
"errors"
"time"
"github.com/gravitl/netmaker/logger"
"github.com/gravitl/netmaker/servercfg"
)
const (
// == Table Names ==
// NETWORKS_TABLE_NAME - networks table
NETWORKS_TABLE_NAME = "networks"
// NODES_TABLE_NAME - nodes table
NODES_TABLE_NAME = "nodes"
// DELETED_NODES_TABLE_NAME - deleted nodes table
DELETED_NODES_TABLE_NAME = "deletednodes"
// USERS_TABLE_NAME - users table
USERS_TABLE_NAME = "users"
// USER_PERMISSIONS_TABLE_NAME - user permissions table
USER_PERMISSIONS_TABLE_NAME = "user_permissions"
// CERTS_TABLE_NAME - certificates table
CERTS_TABLE_NAME = "certs"
// DNS_TABLE_NAME - dns table
DNS_TABLE_NAME = "dns"
// EXT_CLIENT_TABLE_NAME - ext client table
EXT_CLIENT_TABLE_NAME = "extclients"
// PEERS_TABLE_NAME - peers table
PEERS_TABLE_NAME = "peers"
// SERVERCONF_TABLE_NAME - stores server conf
SERVERCONF_TABLE_NAME = "serverconf"
// SERVER_UUID_TABLE_NAME - stores unique netmaker server data
SERVER_UUID_TABLE_NAME = "serveruuid"
// SERVER_UUID_RECORD_KEY - telemetry thing
SERVER_UUID_RECORD_KEY = "serveruuid"
// DATABASE_FILENAME - database file name
DATABASE_FILENAME = "netmaker.db"
// GENERATED_TABLE_NAME - stores server generated k/v
GENERATED_TABLE_NAME = "generated"
// NODE_ACLS_TABLE_NAME - stores the node ACL rules
NODE_ACLS_TABLE_NAME = "nodeacls"
// ACLS_TABLE_NAME - table for acls v2
ACLS_TABLE_NAME = "acls"
// SSO_STATE_CACHE - holds sso session information for OAuth2 sign-ins
SSO_STATE_CACHE = "ssostatecache"
// METRICS_TABLE_NAME - stores network metrics
METRICS_TABLE_NAME = "metrics"
// NETWORK_USER_TABLE_NAME - network user table tracks stats for a network user per network
NETWORK_USER_TABLE_NAME = "networkusers"
// USER_GROUPS_TABLE_NAME - table for storing usergroups
USER_GROUPS_TABLE_NAME = "usergroups"
// CACHE_TABLE_NAME - caching table
CACHE_TABLE_NAME = "cache"
// HOSTS_TABLE_NAME - the table name for hosts
HOSTS_TABLE_NAME = "hosts"
// ENROLLMENT_KEYS_TABLE_NAME - table name for enrollmentkeys
ENROLLMENT_KEYS_TABLE_NAME = "enrollmentkeys"
// HOST_ACTIONS_TABLE_NAME - table name for enrollmentkeys
HOST_ACTIONS_TABLE_NAME = "hostactions"
// PENDING_USERS_TABLE_NAME - table name for pending users
PENDING_USERS_TABLE_NAME = "pending_users"
// USER_INVITES - table for user invites
USER_INVITES_TABLE_NAME = "user_invites"
// TAG_TABLE_NAME - table for tags
TAG_TABLE_NAME = "tags"
// PEER_ACK_TABLE - table for failover peer ack
PEER_ACK_TABLE = "peer_ack"
// SERVER_SETTINGS - table for server settings
SERVER_SETTINGS = "server_settings"
// == ERROR CONSTS ==
// NO_RECORD - no singular result found
NO_RECORD = "no result found"
// NO_RECORDS - no results found
NO_RECORDS = "could not find any records"
// == DB Constants ==
// INIT_DB - initialize db
INIT_DB = "init"
// CREATE_TABLE - create table const
CREATE_TABLE = "createtable"
// INSERT - insert into db const
INSERT = "insert"
// INSERT_PEER - insert peer into db const
INSERT_PEER = "insertpeer"
// DELETE - delete db record const
DELETE = "delete"
// DELETE_ALL - delete a table const
DELETE_ALL = "deleteall"
// FETCH_ALL - fetch table contents const
FETCH_ALL = "fetchall"
// FETCH_ONE - fetch a single record const
FETCH_ONE = "fetchone"
// CLOSE_DB - graceful close of db const
CLOSE_DB = "closedb"
// isconnected
isConnected = "isconnected"
)
var Tables = []string{
NODES_TABLE_NAME,
CERTS_TABLE_NAME,
DELETED_NODES_TABLE_NAME,
DNS_TABLE_NAME,
EXT_CLIENT_TABLE_NAME,
PEERS_TABLE_NAME,
SERVERCONF_TABLE_NAME,
SERVER_UUID_TABLE_NAME,
GENERATED_TABLE_NAME,
NODE_ACLS_TABLE_NAME,
SSO_STATE_CACHE,
METRICS_TABLE_NAME,
NETWORK_USER_TABLE_NAME,
CACHE_TABLE_NAME,
ENROLLMENT_KEYS_TABLE_NAME,
HOST_ACTIONS_TABLE_NAME,
PENDING_USERS_TABLE_NAME,
USER_INVITES_TABLE_NAME,
TAG_TABLE_NAME,
ACLS_TABLE_NAME,
PEER_ACK_TABLE,
SERVER_SETTINGS,
// The following tables are to be migrated, but we still need them so that the migration function
// doesn't fail with table does not exist.
USERS_TABLE_NAME,
USER_GROUPS_TABLE_NAME,
USER_PERMISSIONS_TABLE_NAME,
NETWORKS_TABLE_NAME,
HOSTS_TABLE_NAME,
}
func getCurrentDB() map[string]interface{} {
switch servercfg.GetDB() {
case "rqlite":
return RQLITE_FUNCTIONS
case "sqlite":
return SQLITE_FUNCTIONS
case "postgres":
return PG_FUNCTIONS
default:
return SQLITE_FUNCTIONS
}
}
// InitializeDatabase - initializes database
func InitializeDatabase() error {
logger.Log(0, "connecting to", servercfg.GetDB())
tperiod := time.Now().Add(10 * time.Second)
for {
if err := getCurrentDB()[INIT_DB].(func() error)(); err != nil {
logger.Log(0, "unable to connect to db, retrying . . .")
if time.Now().After(tperiod) {
return err
}
} else {
break
}
time.Sleep(2 * time.Second)
}
createTables()
return nil
}
func createTables() {
for _, table := range Tables {
_ = CreateTable(table)
}
}
func CreateTable(tableName string) error {
return getCurrentDB()[CREATE_TABLE].(func(string) error)(tableName)
}
// Insert - inserts object into db
func Insert(key string, value string, tableName string) error {
if key != "" && value != "" {
return getCurrentDB()[INSERT].(func(string, string, string) error)(key, value, tableName)
} else {
return errors.New("invalid insert " + key + " : " + value)
}
}
// DeleteRecord - deletes a record from db
func DeleteRecord(tableName string, key string) error {
return getCurrentDB()[DELETE].(func(string, string) error)(tableName, key)
}
// DeleteAllRecords - removes a table and remakes
func DeleteAllRecords(tableName string) error {
err := getCurrentDB()[DELETE_ALL].(func(string) error)(tableName)
if err != nil {
return err
}
return CreateTable(tableName)
}
// FetchRecord - fetches a single record by key
func FetchRecord(tableName string, key string) (string, error) {
return getCurrentDB()[FETCH_ONE].(func(string, string) (string, error))(tableName, key)
}
// FetchRecords - fetches all records in given table
func FetchRecords(tableName string) (map[string]string, error) {
return getCurrentDB()[FETCH_ALL].(func(string) (map[string]string, error))(tableName)
}
// CloseDB - closes a database gracefully
func CloseDB() {
getCurrentDB()[CLOSE_DB].(func())()
}
// IsConnected - tell if the database is connected or not
func IsConnected() bool {
return getCurrentDB()[isConnected].(func() bool)()
}