Compare commits
5 Commits
feature/st
...
feature/us
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
8d539e96ef | ||
|
|
726e162bee | ||
|
|
3fcf43114b | ||
|
|
488afe4082 | ||
|
|
f0e8cd578d |
@@ -73,8 +73,8 @@ func (p *RDCleanPathProxy) validateCertificateWithJS(conn *proxyConnection, cert
|
||||
}
|
||||
}
|
||||
|
||||
func (p *RDCleanPathProxy) getTLSConfigWithValidation(conn *proxyConnection, requiresCredSSP bool) *tls.Config {
|
||||
config := &tls.Config{
|
||||
func (p *RDCleanPathProxy) getTLSConfigWithValidation(conn *proxyConnection) *tls.Config {
|
||||
return &tls.Config{
|
||||
InsecureSkipVerify: true, // We'll validate manually after handshake
|
||||
VerifyConnection: func(cs tls.ConnectionState) error {
|
||||
var certChain [][]byte
|
||||
@@ -93,15 +93,4 @@ func (p *RDCleanPathProxy) getTLSConfigWithValidation(conn *proxyConnection, req
|
||||
return nil
|
||||
},
|
||||
}
|
||||
|
||||
// CredSSP (NLA) requires TLS 1.2 - it's incompatible with TLS 1.3
|
||||
if requiresCredSSP {
|
||||
config.MinVersion = tls.VersionTLS12
|
||||
config.MaxVersion = tls.VersionTLS12
|
||||
} else {
|
||||
config.MinVersion = tls.VersionTLS12
|
||||
config.MaxVersion = tls.VersionTLS13
|
||||
}
|
||||
|
||||
return config
|
||||
}
|
||||
|
||||
@@ -6,13 +6,11 @@ import (
|
||||
"context"
|
||||
"crypto/tls"
|
||||
"encoding/asn1"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"sync"
|
||||
"syscall/js"
|
||||
"time"
|
||||
|
||||
log "github.com/sirupsen/logrus"
|
||||
)
|
||||
@@ -21,34 +19,18 @@ const (
|
||||
RDCleanPathVersion = 3390
|
||||
RDCleanPathProxyHost = "rdcleanpath.proxy.local"
|
||||
RDCleanPathProxyScheme = "ws"
|
||||
|
||||
rdpDialTimeout = 15 * time.Second
|
||||
|
||||
GeneralErrorCode = 1
|
||||
WSAETimedOut = 10060
|
||||
WSAEConnRefused = 10061
|
||||
WSAEConnAborted = 10053
|
||||
WSAEConnReset = 10054
|
||||
WSAEGenericError = 10050
|
||||
)
|
||||
|
||||
type RDCleanPathPDU struct {
|
||||
Version int64 `asn1:"tag:0,explicit"`
|
||||
Error RDCleanPathErr `asn1:"tag:1,explicit,optional"`
|
||||
Destination string `asn1:"utf8,tag:2,explicit,optional"`
|
||||
ProxyAuth string `asn1:"utf8,tag:3,explicit,optional"`
|
||||
ServerAuth string `asn1:"utf8,tag:4,explicit,optional"`
|
||||
PreconnectionBlob string `asn1:"utf8,tag:5,explicit,optional"`
|
||||
X224ConnectionPDU []byte `asn1:"tag:6,explicit,optional"`
|
||||
ServerCertChain [][]byte `asn1:"tag:7,explicit,optional"`
|
||||
ServerAddr string `asn1:"utf8,tag:9,explicit,optional"`
|
||||
}
|
||||
|
||||
type RDCleanPathErr struct {
|
||||
ErrorCode int16 `asn1:"tag:0,explicit"`
|
||||
HTTPStatusCode int16 `asn1:"tag:1,explicit,optional"`
|
||||
WSALastError int16 `asn1:"tag:2,explicit,optional"`
|
||||
TLSAlertCode int8 `asn1:"tag:3,explicit,optional"`
|
||||
Version int64 `asn1:"tag:0,explicit"`
|
||||
Error []byte `asn1:"tag:1,explicit,optional"`
|
||||
Destination string `asn1:"utf8,tag:2,explicit,optional"`
|
||||
ProxyAuth string `asn1:"utf8,tag:3,explicit,optional"`
|
||||
ServerAuth string `asn1:"utf8,tag:4,explicit,optional"`
|
||||
PreconnectionBlob string `asn1:"utf8,tag:5,explicit,optional"`
|
||||
X224ConnectionPDU []byte `asn1:"tag:6,explicit,optional"`
|
||||
ServerCertChain [][]byte `asn1:"tag:7,explicit,optional"`
|
||||
ServerAddr string `asn1:"utf8,tag:9,explicit,optional"`
|
||||
}
|
||||
|
||||
type RDCleanPathProxy struct {
|
||||
@@ -228,13 +210,9 @@ func (p *RDCleanPathProxy) handleDirectRDP(conn *proxyConnection, firstPacket []
|
||||
destination := conn.destination
|
||||
log.Infof("Direct RDP mode: Connecting to %s via NetBird", destination)
|
||||
|
||||
ctx, cancel := context.WithTimeout(conn.ctx, rdpDialTimeout)
|
||||
defer cancel()
|
||||
|
||||
rdpConn, err := p.nbClient.Dial(ctx, "tcp", destination)
|
||||
rdpConn, err := p.nbClient.Dial(conn.ctx, "tcp", destination)
|
||||
if err != nil {
|
||||
log.Errorf("Failed to connect to %s: %v", destination, err)
|
||||
p.sendRDCleanPathError(conn, newWSAError(err))
|
||||
return
|
||||
}
|
||||
conn.rdpConn = rdpConn
|
||||
@@ -242,7 +220,6 @@ func (p *RDCleanPathProxy) handleDirectRDP(conn *proxyConnection, firstPacket []
|
||||
_, err = rdpConn.Write(firstPacket)
|
||||
if err != nil {
|
||||
log.Errorf("Failed to write first packet: %v", err)
|
||||
p.sendRDCleanPathError(conn, newWSAError(err))
|
||||
return
|
||||
}
|
||||
|
||||
@@ -250,7 +227,6 @@ func (p *RDCleanPathProxy) handleDirectRDP(conn *proxyConnection, firstPacket []
|
||||
n, err := rdpConn.Read(response)
|
||||
if err != nil {
|
||||
log.Errorf("Failed to read X.224 response: %v", err)
|
||||
p.sendRDCleanPathError(conn, newWSAError(err))
|
||||
return
|
||||
}
|
||||
|
||||
@@ -293,52 +269,3 @@ func (p *RDCleanPathProxy) sendToWebSocket(conn *proxyConnection, data []byte) {
|
||||
conn.wsHandlers.Call("send", uint8Array.Get("buffer"))
|
||||
}
|
||||
}
|
||||
|
||||
func (p *RDCleanPathProxy) sendRDCleanPathError(conn *proxyConnection, pdu RDCleanPathPDU) {
|
||||
data, err := asn1.Marshal(pdu)
|
||||
if err != nil {
|
||||
log.Errorf("Failed to marshal error PDU: %v", err)
|
||||
return
|
||||
}
|
||||
p.sendToWebSocket(conn, data)
|
||||
}
|
||||
|
||||
func errorToWSACode(err error) int16 {
|
||||
if err == nil {
|
||||
return WSAEGenericError
|
||||
}
|
||||
var netErr *net.OpError
|
||||
if errors.As(err, &netErr) && netErr.Timeout() {
|
||||
return WSAETimedOut
|
||||
}
|
||||
if errors.Is(err, context.DeadlineExceeded) {
|
||||
return WSAETimedOut
|
||||
}
|
||||
if errors.Is(err, context.Canceled) {
|
||||
return WSAEConnAborted
|
||||
}
|
||||
if errors.Is(err, io.EOF) {
|
||||
return WSAEConnReset
|
||||
}
|
||||
return WSAEGenericError
|
||||
}
|
||||
|
||||
func newWSAError(err error) RDCleanPathPDU {
|
||||
return RDCleanPathPDU{
|
||||
Version: RDCleanPathVersion,
|
||||
Error: RDCleanPathErr{
|
||||
ErrorCode: GeneralErrorCode,
|
||||
WSALastError: errorToWSACode(err),
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func newHTTPError(statusCode int16) RDCleanPathPDU {
|
||||
return RDCleanPathPDU{
|
||||
Version: RDCleanPathVersion,
|
||||
Error: RDCleanPathErr{
|
||||
ErrorCode: GeneralErrorCode,
|
||||
HTTPStatusCode: statusCode,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
@@ -3,7 +3,6 @@
|
||||
package rdp
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/tls"
|
||||
"encoding/asn1"
|
||||
"io"
|
||||
@@ -12,17 +11,11 @@ import (
|
||||
log "github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
const (
|
||||
// MS-RDPBCGR: confusingly named, actually means PROTOCOL_HYBRID (CredSSP)
|
||||
protocolSSL = 0x00000001
|
||||
protocolHybridEx = 0x00000008
|
||||
)
|
||||
|
||||
func (p *RDCleanPathProxy) processRDCleanPathPDU(conn *proxyConnection, pdu RDCleanPathPDU) {
|
||||
log.Infof("Processing RDCleanPath PDU: Version=%d, Destination=%s", pdu.Version, pdu.Destination)
|
||||
|
||||
if pdu.Version != RDCleanPathVersion {
|
||||
p.sendRDCleanPathError(conn, newHTTPError(400))
|
||||
p.sendRDCleanPathError(conn, "Unsupported version")
|
||||
return
|
||||
}
|
||||
|
||||
@@ -31,13 +24,10 @@ func (p *RDCleanPathProxy) processRDCleanPathPDU(conn *proxyConnection, pdu RDCl
|
||||
destination = pdu.Destination
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithTimeout(conn.ctx, rdpDialTimeout)
|
||||
defer cancel()
|
||||
|
||||
rdpConn, err := p.nbClient.Dial(ctx, "tcp", destination)
|
||||
rdpConn, err := p.nbClient.Dial(conn.ctx, "tcp", destination)
|
||||
if err != nil {
|
||||
log.Errorf("Failed to connect to %s: %v", destination, err)
|
||||
p.sendRDCleanPathError(conn, newWSAError(err))
|
||||
p.sendRDCleanPathError(conn, "Connection failed")
|
||||
p.cleanupConnection(conn)
|
||||
return
|
||||
}
|
||||
@@ -50,34 +40,6 @@ func (p *RDCleanPathProxy) processRDCleanPathPDU(conn *proxyConnection, pdu RDCl
|
||||
p.setupTLSConnection(conn, pdu)
|
||||
}
|
||||
|
||||
// detectCredSSPFromX224 checks if the X.224 response indicates NLA/CredSSP is required.
|
||||
// Per MS-RDPBCGR spec: byte 11 = TYPE_RDP_NEG_RSP (0x02), bytes 15-18 = selectedProtocol flags.
|
||||
// Returns (requiresTLS12, selectedProtocol, detectionSuccessful).
|
||||
func (p *RDCleanPathProxy) detectCredSSPFromX224(x224Response []byte) (bool, uint32, bool) {
|
||||
const minResponseLength = 19
|
||||
|
||||
if len(x224Response) < minResponseLength {
|
||||
return false, 0, false
|
||||
}
|
||||
|
||||
// Per X.224 specification:
|
||||
// x224Response[0] == 0x03: Length of X.224 header (3 bytes)
|
||||
// x224Response[5] == 0xD0: X.224 Data TPDU code
|
||||
if x224Response[0] != 0x03 || x224Response[5] != 0xD0 {
|
||||
return false, 0, false
|
||||
}
|
||||
|
||||
if x224Response[11] == 0x02 {
|
||||
flags := uint32(x224Response[15]) | uint32(x224Response[16])<<8 |
|
||||
uint32(x224Response[17])<<16 | uint32(x224Response[18])<<24
|
||||
|
||||
hasNLA := (flags & (protocolSSL | protocolHybridEx)) != 0
|
||||
return hasNLA, flags, true
|
||||
}
|
||||
|
||||
return false, 0, false
|
||||
}
|
||||
|
||||
func (p *RDCleanPathProxy) setupTLSConnection(conn *proxyConnection, pdu RDCleanPathPDU) {
|
||||
var x224Response []byte
|
||||
if len(pdu.X224ConnectionPDU) > 0 {
|
||||
@@ -85,7 +47,7 @@ func (p *RDCleanPathProxy) setupTLSConnection(conn *proxyConnection, pdu RDClean
|
||||
_, err := conn.rdpConn.Write(pdu.X224ConnectionPDU)
|
||||
if err != nil {
|
||||
log.Errorf("Failed to write X.224 PDU: %v", err)
|
||||
p.sendRDCleanPathError(conn, newWSAError(err))
|
||||
p.sendRDCleanPathError(conn, "Failed to forward X.224")
|
||||
return
|
||||
}
|
||||
|
||||
@@ -93,32 +55,21 @@ func (p *RDCleanPathProxy) setupTLSConnection(conn *proxyConnection, pdu RDClean
|
||||
n, err := conn.rdpConn.Read(response)
|
||||
if err != nil {
|
||||
log.Errorf("Failed to read X.224 response: %v", err)
|
||||
p.sendRDCleanPathError(conn, newWSAError(err))
|
||||
p.sendRDCleanPathError(conn, "Failed to read X.224 response")
|
||||
return
|
||||
}
|
||||
x224Response = response[:n]
|
||||
log.Debugf("Received X.224 Connection Confirm (%d bytes)", n)
|
||||
}
|
||||
|
||||
requiresCredSSP, selectedProtocol, detected := p.detectCredSSPFromX224(x224Response)
|
||||
if detected {
|
||||
if requiresCredSSP {
|
||||
log.Warnf("Detected NLA/CredSSP (selectedProtocol: 0x%08X), forcing TLS 1.2 for compatibility", selectedProtocol)
|
||||
} else {
|
||||
log.Warnf("No NLA/CredSSP detected (selectedProtocol: 0x%08X), allowing up to TLS 1.3", selectedProtocol)
|
||||
}
|
||||
} else {
|
||||
log.Warnf("Could not detect RDP security protocol, allowing up to TLS 1.3")
|
||||
}
|
||||
|
||||
tlsConfig := p.getTLSConfigWithValidation(conn, requiresCredSSP)
|
||||
tlsConfig := p.getTLSConfigWithValidation(conn)
|
||||
|
||||
tlsConn := tls.Client(conn.rdpConn, tlsConfig)
|
||||
conn.tlsConn = tlsConn
|
||||
|
||||
if err := tlsConn.Handshake(); err != nil {
|
||||
log.Errorf("TLS handshake failed: %v", err)
|
||||
p.sendRDCleanPathError(conn, newWSAError(err))
|
||||
p.sendRDCleanPathError(conn, "TLS handshake failed")
|
||||
return
|
||||
}
|
||||
|
||||
@@ -155,6 +106,47 @@ func (p *RDCleanPathProxy) setupTLSConnection(conn *proxyConnection, pdu RDClean
|
||||
p.cleanupConnection(conn)
|
||||
}
|
||||
|
||||
func (p *RDCleanPathProxy) setupPlainConnection(conn *proxyConnection, pdu RDCleanPathPDU) {
|
||||
if len(pdu.X224ConnectionPDU) > 0 {
|
||||
log.Debugf("Forwarding X.224 Connection Request (%d bytes)", len(pdu.X224ConnectionPDU))
|
||||
_, err := conn.rdpConn.Write(pdu.X224ConnectionPDU)
|
||||
if err != nil {
|
||||
log.Errorf("Failed to write X.224 PDU: %v", err)
|
||||
p.sendRDCleanPathError(conn, "Failed to forward X.224")
|
||||
return
|
||||
}
|
||||
|
||||
response := make([]byte, 1024)
|
||||
n, err := conn.rdpConn.Read(response)
|
||||
if err != nil {
|
||||
log.Errorf("Failed to read X.224 response: %v", err)
|
||||
p.sendRDCleanPathError(conn, "Failed to read X.224 response")
|
||||
return
|
||||
}
|
||||
|
||||
responsePDU := RDCleanPathPDU{
|
||||
Version: RDCleanPathVersion,
|
||||
X224ConnectionPDU: response[:n],
|
||||
ServerAddr: conn.destination,
|
||||
}
|
||||
|
||||
p.sendRDCleanPathPDU(conn, responsePDU)
|
||||
} else {
|
||||
responsePDU := RDCleanPathPDU{
|
||||
Version: RDCleanPathVersion,
|
||||
ServerAddr: conn.destination,
|
||||
}
|
||||
p.sendRDCleanPathPDU(conn, responsePDU)
|
||||
}
|
||||
|
||||
go p.forwardConnToWS(conn, conn.rdpConn, "TCP")
|
||||
go p.forwardWSToConn(conn, conn.rdpConn, "TCP")
|
||||
|
||||
<-conn.ctx.Done()
|
||||
log.Debug("TCP connection context done, cleaning up")
|
||||
p.cleanupConnection(conn)
|
||||
}
|
||||
|
||||
func (p *RDCleanPathProxy) sendRDCleanPathPDU(conn *proxyConnection, pdu RDCleanPathPDU) {
|
||||
data, err := asn1.Marshal(pdu)
|
||||
if err != nil {
|
||||
@@ -166,6 +158,21 @@ func (p *RDCleanPathProxy) sendRDCleanPathPDU(conn *proxyConnection, pdu RDClean
|
||||
p.sendToWebSocket(conn, data)
|
||||
}
|
||||
|
||||
func (p *RDCleanPathProxy) sendRDCleanPathError(conn *proxyConnection, errorMsg string) {
|
||||
pdu := RDCleanPathPDU{
|
||||
Version: RDCleanPathVersion,
|
||||
Error: []byte(errorMsg),
|
||||
}
|
||||
|
||||
data, err := asn1.Marshal(pdu)
|
||||
if err != nil {
|
||||
log.Errorf("Failed to marshal error PDU: %v", err)
|
||||
return
|
||||
}
|
||||
|
||||
p.sendToWebSocket(conn, data)
|
||||
}
|
||||
|
||||
func (p *RDCleanPathProxy) readWebSocketMessage(conn *proxyConnection) ([]byte, error) {
|
||||
msgChan := make(chan []byte)
|
||||
errChan := make(chan error)
|
||||
|
||||
1
go.mod
1
go.mod
@@ -45,6 +45,7 @@ require (
|
||||
github.com/eko/gocache/store/redis/v4 v4.2.2
|
||||
github.com/fsnotify/fsnotify v1.7.0
|
||||
github.com/gliderlabs/ssh v0.3.8
|
||||
github.com/go-gorm/caches/v4 v4.0.5
|
||||
github.com/godbus/dbus/v5 v5.1.0
|
||||
github.com/golang-jwt/jwt/v5 v5.3.0
|
||||
github.com/golang/mock v1.6.0
|
||||
|
||||
2
go.sum
2
go.sum
@@ -215,6 +215,8 @@ github.com/go-gl/glfw/v3.3/glfw v0.0.0-20191125211704-12ad95a8df72/go.mod h1:tQ2
|
||||
github.com/go-gl/glfw/v3.3/glfw v0.0.0-20200222043503-6f7a984d4dc4/go.mod h1:tQ2UAYgL5IevRw8kRxooKSPJfGvJ9fJQFa0TUsXzTg8=
|
||||
github.com/go-gl/glfw/v3.3/glfw v0.0.0-20240506104042-037f3cc74f2a h1:vxnBhFDDT+xzxf1jTJKMKZw3H0swfWk9RpWbBbDK5+0=
|
||||
github.com/go-gl/glfw/v3.3/glfw v0.0.0-20240506104042-037f3cc74f2a/go.mod h1:tQ2UAYgL5IevRw8kRxooKSPJfGvJ9fJQFa0TUsXzTg8=
|
||||
github.com/go-gorm/caches/v4 v4.0.5 h1:Sdj9vxbEM0sCmv5+s5o6GzoVMuraWF0bjJJvUU+7c1U=
|
||||
github.com/go-gorm/caches/v4 v4.0.5/go.mod h1:Ms8LnWVoW4GkTofpDzFH8OfDGNTjLxQDyxBmRN67Ujw=
|
||||
github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
|
||||
github.com/go-logr/logr v1.4.2 h1:6pFjapn8bFcIbiKo3XT4j/BhANplGihG6tvd+8rYgrY=
|
||||
github.com/go-logr/logr v1.4.2/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
|
||||
|
||||
@@ -17,7 +17,7 @@ type Peer struct {
|
||||
// ID is an internal ID of the peer
|
||||
ID string `gorm:"primaryKey"`
|
||||
// AccountID is a reference to Account that this object belongs
|
||||
AccountID string `json:"-" gorm:"index"`
|
||||
AccountID string `gorm:"index"`
|
||||
// WireGuard public key
|
||||
Key string `gorm:"index"`
|
||||
// IP address of the Peer
|
||||
|
||||
@@ -7,8 +7,9 @@ import (
|
||||
"regexp"
|
||||
|
||||
"github.com/hashicorp/go-version"
|
||||
"github.com/netbirdio/netbird/shared/management/http/api"
|
||||
|
||||
nbpeer "github.com/netbirdio/netbird/management/server/peer"
|
||||
"github.com/netbirdio/netbird/shared/management/http/api"
|
||||
"github.com/netbirdio/netbird/shared/management/status"
|
||||
)
|
||||
|
||||
@@ -45,7 +46,7 @@ type Checks struct {
|
||||
Description string
|
||||
|
||||
// AccountID is a reference to the Account that this object belongs
|
||||
AccountID string `json:"-" gorm:"index"`
|
||||
AccountID string `gorm:"index"`
|
||||
|
||||
// Checks is a set of objects that perform the actual checks
|
||||
Checks ChecksDefinition `gorm:"serializer:json"`
|
||||
|
||||
@@ -30,6 +30,7 @@ func TestChecks_MarshalJSON(t *testing.T) {
|
||||
},
|
||||
want: []byte(`
|
||||
{
|
||||
"AccountID":"acc1",
|
||||
"ID": "id1",
|
||||
"Name": "name1",
|
||||
"Description": "desc1",
|
||||
@@ -55,6 +56,7 @@ func TestChecks_MarshalJSON(t *testing.T) {
|
||||
},
|
||||
want: []byte(`
|
||||
{
|
||||
"AccountID":"",
|
||||
"ID": "",
|
||||
"Name": "",
|
||||
"Description": "",
|
||||
@@ -94,6 +96,7 @@ func TestChecks_UnmarshalJSON(t *testing.T) {
|
||||
name: "Valid JSON Posture Checks Unmarshal",
|
||||
in: []byte(`
|
||||
{
|
||||
"AccountID":"acc1",
|
||||
"ID": "id1",
|
||||
"Name": "name1",
|
||||
"Description": "desc1",
|
||||
@@ -106,6 +109,7 @@ func TestChecks_UnmarshalJSON(t *testing.T) {
|
||||
`),
|
||||
expected: &Checks{
|
||||
ID: "id1",
|
||||
AccountID: "acc1",
|
||||
Name: "name1",
|
||||
Description: "desc1",
|
||||
Checks: ChecksDefinition{
|
||||
|
||||
129
management/server/store/cache/dual_key_cache.go
vendored
129
management/server/store/cache/dual_key_cache.go
vendored
@@ -1,129 +0,0 @@
|
||||
package cache
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
)
|
||||
|
||||
// DualKeyCache provides a caching mechanism where each entry has two keys:
|
||||
// - Primary key (e.g., objectID): used for accessing and invalidating specific entries
|
||||
// - Secondary key (e.g., accountID): used for bulk invalidation of all entries with the same secondary key
|
||||
type DualKeyCache[K1 comparable, K2 comparable, V any] struct {
|
||||
mu sync.RWMutex
|
||||
primaryIndex map[K1]V // Primary key -> Value
|
||||
secondaryIndex map[K2]map[K1]struct{} // Secondary key -> Set of primary keys
|
||||
reverseLookup map[K1]K2 // Primary key -> Secondary key
|
||||
}
|
||||
|
||||
// NewDualKeyCache creates a new dual-key cache
|
||||
func NewDualKeyCache[K1 comparable, K2 comparable, V any]() *DualKeyCache[K1, K2, V] {
|
||||
return &DualKeyCache[K1, K2, V]{
|
||||
primaryIndex: make(map[K1]V),
|
||||
secondaryIndex: make(map[K2]map[K1]struct{}),
|
||||
reverseLookup: make(map[K1]K2),
|
||||
}
|
||||
}
|
||||
|
||||
// Get retrieves a value from the cache using the primary key
|
||||
func (c *DualKeyCache[K1, K2, V]) Get(ctx context.Context, primaryKey K1) (V, bool) {
|
||||
c.mu.RLock()
|
||||
defer c.mu.RUnlock()
|
||||
|
||||
value, ok := c.primaryIndex[primaryKey]
|
||||
return value, ok
|
||||
}
|
||||
|
||||
// Set stores a value in the cache with both primary and secondary keys
|
||||
func (c *DualKeyCache[K1, K2, V]) Set(ctx context.Context, primaryKey K1, secondaryKey K2, value V) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
|
||||
if oldSecondaryKey, exists := c.reverseLookup[primaryKey]; exists {
|
||||
if primaryKeys, ok := c.secondaryIndex[oldSecondaryKey]; ok {
|
||||
delete(primaryKeys, primaryKey)
|
||||
if len(primaryKeys) == 0 {
|
||||
delete(c.secondaryIndex, oldSecondaryKey)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
c.primaryIndex[primaryKey] = value
|
||||
c.reverseLookup[primaryKey] = secondaryKey
|
||||
|
||||
if _, exists := c.secondaryIndex[secondaryKey]; !exists {
|
||||
c.secondaryIndex[secondaryKey] = make(map[K1]struct{})
|
||||
}
|
||||
c.secondaryIndex[secondaryKey][primaryKey] = struct{}{}
|
||||
}
|
||||
|
||||
// InvalidateByPrimaryKey removes an entry using the primary key
|
||||
func (c *DualKeyCache[K1, K2, V]) InvalidateByPrimaryKey(ctx context.Context, primaryKey K1) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
|
||||
if secondaryKey, exists := c.reverseLookup[primaryKey]; exists {
|
||||
if primaryKeys, ok := c.secondaryIndex[secondaryKey]; ok {
|
||||
delete(primaryKeys, primaryKey)
|
||||
if len(primaryKeys) == 0 {
|
||||
delete(c.secondaryIndex, secondaryKey)
|
||||
}
|
||||
}
|
||||
delete(c.reverseLookup, primaryKey)
|
||||
}
|
||||
|
||||
delete(c.primaryIndex, primaryKey)
|
||||
}
|
||||
|
||||
// InvalidateBySecondaryKey removes all entries with the given secondary key
|
||||
func (c *DualKeyCache[K1, K2, V]) InvalidateBySecondaryKey(ctx context.Context, secondaryKey K2) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
|
||||
primaryKeys, exists := c.secondaryIndex[secondaryKey]
|
||||
if !exists {
|
||||
return
|
||||
}
|
||||
|
||||
for primaryKey := range primaryKeys {
|
||||
delete(c.primaryIndex, primaryKey)
|
||||
delete(c.reverseLookup, primaryKey)
|
||||
}
|
||||
|
||||
delete(c.secondaryIndex, secondaryKey)
|
||||
}
|
||||
|
||||
// InvalidateAll removes all entries from the cache
|
||||
func (c *DualKeyCache[K1, K2, V]) InvalidateAll(ctx context.Context) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
|
||||
c.primaryIndex = make(map[K1]V)
|
||||
c.secondaryIndex = make(map[K2]map[K1]struct{})
|
||||
c.reverseLookup = make(map[K1]K2)
|
||||
}
|
||||
|
||||
// Size returns the number of entries in the cache
|
||||
func (c *DualKeyCache[K1, K2, V]) Size() int {
|
||||
c.mu.RLock()
|
||||
defer c.mu.RUnlock()
|
||||
|
||||
return len(c.primaryIndex)
|
||||
}
|
||||
|
||||
// GetOrSet retrieves a value from the cache, or sets it using the provided function if not found
|
||||
// The loadFunc should return both the value and the secondary key (extracted from the value)
|
||||
func (c *DualKeyCache[K1, K2, V]) GetOrSet(ctx context.Context, primaryKey K1, loadFunc func() (V, K2, error)) (V, error) {
|
||||
if value, ok := c.Get(ctx, primaryKey); ok {
|
||||
return value, nil
|
||||
}
|
||||
|
||||
value, secondaryKey, err := loadFunc()
|
||||
if err != nil {
|
||||
var zero V
|
||||
return zero, err
|
||||
}
|
||||
|
||||
c.Set(ctx, primaryKey, secondaryKey, value)
|
||||
|
||||
return value, nil
|
||||
}
|
||||
53
management/server/store/cache/memory.go
vendored
Normal file
53
management/server/store/cache/memory.go
vendored
Normal file
@@ -0,0 +1,53 @@
|
||||
/*
|
||||
Example code copied from https://github.com/go-gorm/caches
|
||||
*/
|
||||
|
||||
package cache
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
|
||||
"github.com/go-gorm/caches/v4"
|
||||
)
|
||||
|
||||
type MemoryCacher struct {
|
||||
store *sync.Map
|
||||
}
|
||||
|
||||
func (c *MemoryCacher) init() {
|
||||
if c.store == nil {
|
||||
c.store = &sync.Map{}
|
||||
}
|
||||
}
|
||||
|
||||
func (c *MemoryCacher) Get(ctx context.Context, key string, q *caches.Query[any]) (*caches.Query[any], error) {
|
||||
c.init()
|
||||
val, ok := c.store.Load(key)
|
||||
if !ok {
|
||||
//nolint
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
if err := q.Unmarshal(val.([]byte)); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return q, nil
|
||||
}
|
||||
|
||||
func (c *MemoryCacher) Store(ctx context.Context, key string, val *caches.Query[any]) error {
|
||||
c.init()
|
||||
res, err := val.Marshal()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
c.store.Store(key, res)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *MemoryCacher) Invalidate(ctx context.Context) error {
|
||||
c.store = &sync.Map{}
|
||||
return nil
|
||||
}
|
||||
78
management/server/store/cache/redis.go
vendored
Normal file
78
management/server/store/cache/redis.go
vendored
Normal file
@@ -0,0 +1,78 @@
|
||||
/*
|
||||
Example code copied from https://github.com/go-gorm/caches
|
||||
*/
|
||||
|
||||
package cache
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/go-gorm/caches/v4"
|
||||
"github.com/redis/go-redis/v9"
|
||||
)
|
||||
|
||||
type RedisCacher struct {
|
||||
rdb *redis.Client
|
||||
}
|
||||
|
||||
func NewRedisCacher(rdb *redis.Client) *RedisCacher {
|
||||
return &RedisCacher{rdb: rdb}
|
||||
}
|
||||
|
||||
func (c *RedisCacher) Get(ctx context.Context, key string, q *caches.Query[any]) (*caches.Query[any], error) {
|
||||
res, err := c.rdb.Get(ctx, key).Result()
|
||||
if err == redis.Nil {
|
||||
//nolint
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if err := q.Unmarshal([]byte(res)); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return q, nil
|
||||
}
|
||||
|
||||
func (c *RedisCacher) Store(ctx context.Context, key string, val *caches.Query[any]) error {
|
||||
res, err := val.Marshal()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
c.rdb.Set(ctx, key, res, 300*time.Second) // Set proper cache time
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *RedisCacher) Invalidate(ctx context.Context) error {
|
||||
var (
|
||||
cursor uint64
|
||||
keys []string
|
||||
)
|
||||
for {
|
||||
var (
|
||||
k []string
|
||||
err error
|
||||
)
|
||||
k, cursor, err = c.rdb.Scan(ctx, cursor, fmt.Sprintf("%s*", caches.IdentifierPrefix), 0).Result()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
keys = append(keys, k...)
|
||||
if cursor == 0 {
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
if len(keys) > 0 {
|
||||
if _, err := c.rdb.Del(ctx, keys...).Result(); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -1,77 +0,0 @@
|
||||
package cache
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
)
|
||||
|
||||
// SingleKeyCache provides a simple caching mechanism with a single key
|
||||
type SingleKeyCache[K comparable, V any] struct {
|
||||
mu sync.RWMutex
|
||||
cache map[K]V // Key -> Value
|
||||
}
|
||||
|
||||
// NewSingleKeyCache creates a new single-key cache
|
||||
func NewSingleKeyCache[K comparable, V any]() *SingleKeyCache[K, V] {
|
||||
return &SingleKeyCache[K, V]{
|
||||
cache: make(map[K]V),
|
||||
}
|
||||
}
|
||||
|
||||
// Get retrieves a value from the cache using the key
|
||||
func (c *SingleKeyCache[K, V]) Get(ctx context.Context, key K) (V, bool) {
|
||||
c.mu.RLock()
|
||||
defer c.mu.RUnlock()
|
||||
|
||||
value, ok := c.cache[key]
|
||||
return value, ok
|
||||
}
|
||||
|
||||
// Set stores a value in the cache with the given key
|
||||
func (c *SingleKeyCache[K, V]) Set(ctx context.Context, key K, value V) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
|
||||
c.cache[key] = value
|
||||
}
|
||||
|
||||
// Invalidate removes an entry using the key
|
||||
func (c *SingleKeyCache[K, V]) Invalidate(ctx context.Context, key K) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
|
||||
delete(c.cache, key)
|
||||
}
|
||||
|
||||
// InvalidateAll removes all entries from the cache
|
||||
func (c *SingleKeyCache[K, V]) InvalidateAll(ctx context.Context) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
|
||||
c.cache = make(map[K]V)
|
||||
}
|
||||
|
||||
// Size returns the number of entries in the cache
|
||||
func (c *SingleKeyCache[K, V]) Size() int {
|
||||
c.mu.RLock()
|
||||
defer c.mu.RUnlock()
|
||||
|
||||
return len(c.cache)
|
||||
}
|
||||
|
||||
// GetOrSet retrieves a value from the cache, or sets it using the provided function if not found
|
||||
func (c *SingleKeyCache[K, V]) GetOrSet(ctx context.Context, key K, loadFunc func() (V, error)) (V, error) {
|
||||
if value, ok := c.Get(ctx, key); ok {
|
||||
return value, nil
|
||||
}
|
||||
|
||||
value, err := loadFunc()
|
||||
if err != nil {
|
||||
var zero V
|
||||
return zero, err
|
||||
}
|
||||
|
||||
c.Set(ctx, key, value)
|
||||
|
||||
return value, nil
|
||||
}
|
||||
242
management/server/store/cache/triple_key_cache.go
vendored
242
management/server/store/cache/triple_key_cache.go
vendored
@@ -1,242 +0,0 @@
|
||||
package cache
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
)
|
||||
|
||||
// TripleKeyCache provides a caching mechanism where each entry has three keys:
|
||||
// - Primary key (K1): used for accessing and invalidating specific entries
|
||||
// - Secondary key (K2): used for bulk invalidation of all entries with the same secondary key
|
||||
// - Tertiary key (K3): used for bulk invalidation of all entries with the same tertiary key
|
||||
type TripleKeyCache[K1 comparable, K2 comparable, K3 comparable, V any] struct {
|
||||
mu sync.RWMutex
|
||||
primaryIndex map[K1]V // Primary key -> Value
|
||||
secondaryIndex map[K2]map[K1]struct{} // Secondary key -> Set of primary keys
|
||||
tertiaryIndex map[K3]map[K1]struct{} // Tertiary key -> Set of primary keys
|
||||
reverseLookup map[K1]keyPair[K2, K3] // Primary key -> Secondary and Tertiary keys
|
||||
}
|
||||
|
||||
type keyPair[K2 comparable, K3 comparable] struct {
|
||||
secondary K2
|
||||
tertiary K3
|
||||
}
|
||||
|
||||
// NewTripleKeyCache creates a new triple-key cache
|
||||
func NewTripleKeyCache[K1 comparable, K2 comparable, K3 comparable, V any]() *TripleKeyCache[K1, K2, K3, V] {
|
||||
return &TripleKeyCache[K1, K2, K3, V]{
|
||||
primaryIndex: make(map[K1]V),
|
||||
secondaryIndex: make(map[K2]map[K1]struct{}),
|
||||
tertiaryIndex: make(map[K3]map[K1]struct{}),
|
||||
reverseLookup: make(map[K1]keyPair[K2, K3]),
|
||||
}
|
||||
}
|
||||
|
||||
// Get retrieves a value from the cache using the primary key
|
||||
func (c *TripleKeyCache[K1, K2, K3, V]) Get(ctx context.Context, primaryKey K1) (V, bool) {
|
||||
c.mu.RLock()
|
||||
defer c.mu.RUnlock()
|
||||
|
||||
value, ok := c.primaryIndex[primaryKey]
|
||||
return value, ok
|
||||
}
|
||||
|
||||
// Set stores a value in the cache with primary, secondary, and tertiary keys
|
||||
func (c *TripleKeyCache[K1, K2, K3, V]) Set(ctx context.Context, primaryKey K1, secondaryKey K2, tertiaryKey K3, value V) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
|
||||
if oldKeys, exists := c.reverseLookup[primaryKey]; exists {
|
||||
if primaryKeys, ok := c.secondaryIndex[oldKeys.secondary]; ok {
|
||||
delete(primaryKeys, primaryKey)
|
||||
if len(primaryKeys) == 0 {
|
||||
delete(c.secondaryIndex, oldKeys.secondary)
|
||||
}
|
||||
}
|
||||
if primaryKeys, ok := c.tertiaryIndex[oldKeys.tertiary]; ok {
|
||||
delete(primaryKeys, primaryKey)
|
||||
if len(primaryKeys) == 0 {
|
||||
delete(c.tertiaryIndex, oldKeys.tertiary)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
c.primaryIndex[primaryKey] = value
|
||||
c.reverseLookup[primaryKey] = keyPair[K2, K3]{
|
||||
secondary: secondaryKey,
|
||||
tertiary: tertiaryKey,
|
||||
}
|
||||
|
||||
if _, exists := c.secondaryIndex[secondaryKey]; !exists {
|
||||
c.secondaryIndex[secondaryKey] = make(map[K1]struct{})
|
||||
}
|
||||
c.secondaryIndex[secondaryKey][primaryKey] = struct{}{}
|
||||
|
||||
if _, exists := c.tertiaryIndex[tertiaryKey]; !exists {
|
||||
c.tertiaryIndex[tertiaryKey] = make(map[K1]struct{})
|
||||
}
|
||||
c.tertiaryIndex[tertiaryKey][primaryKey] = struct{}{}
|
||||
}
|
||||
|
||||
// InvalidateByPrimaryKey removes an entry using the primary key
|
||||
func (c *TripleKeyCache[K1, K2, K3, V]) InvalidateByPrimaryKey(ctx context.Context, primaryKey K1) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
|
||||
if keys, exists := c.reverseLookup[primaryKey]; exists {
|
||||
if primaryKeys, ok := c.secondaryIndex[keys.secondary]; ok {
|
||||
delete(primaryKeys, primaryKey)
|
||||
if len(primaryKeys) == 0 {
|
||||
delete(c.secondaryIndex, keys.secondary)
|
||||
}
|
||||
}
|
||||
if primaryKeys, ok := c.tertiaryIndex[keys.tertiary]; ok {
|
||||
delete(primaryKeys, primaryKey)
|
||||
if len(primaryKeys) == 0 {
|
||||
delete(c.tertiaryIndex, keys.tertiary)
|
||||
}
|
||||
}
|
||||
delete(c.reverseLookup, primaryKey)
|
||||
}
|
||||
|
||||
delete(c.primaryIndex, primaryKey)
|
||||
}
|
||||
|
||||
// InvalidateBySecondaryKey removes all entries with the given secondary key
|
||||
func (c *TripleKeyCache[K1, K2, K3, V]) InvalidateBySecondaryKey(ctx context.Context, secondaryKey K2) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
|
||||
primaryKeys, exists := c.secondaryIndex[secondaryKey]
|
||||
if !exists {
|
||||
return
|
||||
}
|
||||
|
||||
for primaryKey := range primaryKeys {
|
||||
if keys, ok := c.reverseLookup[primaryKey]; ok {
|
||||
if tertiaryPrimaryKeys, exists := c.tertiaryIndex[keys.tertiary]; exists {
|
||||
delete(tertiaryPrimaryKeys, primaryKey)
|
||||
if len(tertiaryPrimaryKeys) == 0 {
|
||||
delete(c.tertiaryIndex, keys.tertiary)
|
||||
}
|
||||
}
|
||||
}
|
||||
delete(c.primaryIndex, primaryKey)
|
||||
delete(c.reverseLookup, primaryKey)
|
||||
}
|
||||
|
||||
delete(c.secondaryIndex, secondaryKey)
|
||||
}
|
||||
|
||||
// InvalidateByTertiaryKey removes all entries with the given tertiary key
|
||||
func (c *TripleKeyCache[K1, K2, K3, V]) InvalidateByTertiaryKey(ctx context.Context, tertiaryKey K3) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
|
||||
primaryKeys, exists := c.tertiaryIndex[tertiaryKey]
|
||||
if !exists {
|
||||
return
|
||||
}
|
||||
|
||||
for primaryKey := range primaryKeys {
|
||||
if keys, ok := c.reverseLookup[primaryKey]; ok {
|
||||
if secondaryPrimaryKeys, exists := c.secondaryIndex[keys.secondary]; exists {
|
||||
delete(secondaryPrimaryKeys, primaryKey)
|
||||
if len(secondaryPrimaryKeys) == 0 {
|
||||
delete(c.secondaryIndex, keys.secondary)
|
||||
}
|
||||
}
|
||||
}
|
||||
delete(c.primaryIndex, primaryKey)
|
||||
delete(c.reverseLookup, primaryKey)
|
||||
}
|
||||
|
||||
delete(c.tertiaryIndex, tertiaryKey)
|
||||
}
|
||||
|
||||
// InvalidateAll removes all entries from the cache
|
||||
func (c *TripleKeyCache[K1, K2, K3, V]) InvalidateAll(ctx context.Context) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
|
||||
c.primaryIndex = make(map[K1]V)
|
||||
c.secondaryIndex = make(map[K2]map[K1]struct{})
|
||||
c.tertiaryIndex = make(map[K3]map[K1]struct{})
|
||||
c.reverseLookup = make(map[K1]keyPair[K2, K3])
|
||||
}
|
||||
|
||||
// Size returns the number of entries in the cache
|
||||
func (c *TripleKeyCache[K1, K2, K3, V]) Size() int {
|
||||
c.mu.RLock()
|
||||
defer c.mu.RUnlock()
|
||||
|
||||
return len(c.primaryIndex)
|
||||
}
|
||||
|
||||
// GetOrSet retrieves a value from the cache, or sets it using the provided function if not found
|
||||
// The loadFunc should return the value, secondary key, and tertiary key (extracted from the value)
|
||||
func (c *TripleKeyCache[K1, K2, K3, V]) GetOrSet(ctx context.Context, primaryKey K1, loadFunc func() (V, K2, K3, error)) (V, error) {
|
||||
if value, ok := c.Get(ctx, primaryKey); ok {
|
||||
return value, nil
|
||||
}
|
||||
|
||||
value, secondaryKey, tertiaryKey, err := loadFunc()
|
||||
if err != nil {
|
||||
var zero V
|
||||
return zero, err
|
||||
}
|
||||
|
||||
c.Set(ctx, primaryKey, secondaryKey, tertiaryKey, value)
|
||||
|
||||
return value, nil
|
||||
}
|
||||
|
||||
// GetOrSetBySecondaryKey retrieves a value from the cache using the secondary key, or sets it using the provided function if not found
|
||||
// The loadFunc should return the value, primary key, secondary key, and tertiary key
|
||||
func (c *TripleKeyCache[K1, K2, K3, V]) GetOrSetBySecondaryKey(ctx context.Context, secondaryKey K2, loadFunc func() (V, K1, K3, error)) (V, error) {
|
||||
c.mu.RLock()
|
||||
if primaryKeys, exists := c.secondaryIndex[secondaryKey]; exists && len(primaryKeys) > 0 {
|
||||
for primaryKey := range primaryKeys {
|
||||
if value, ok := c.primaryIndex[primaryKey]; ok {
|
||||
c.mu.RUnlock()
|
||||
return value, nil
|
||||
}
|
||||
}
|
||||
}
|
||||
c.mu.RUnlock()
|
||||
|
||||
value, primaryKey, tertiaryKey, err := loadFunc()
|
||||
if err != nil {
|
||||
var zero V
|
||||
return zero, err
|
||||
}
|
||||
|
||||
c.Set(ctx, primaryKey, secondaryKey, tertiaryKey, value)
|
||||
|
||||
return value, nil
|
||||
}
|
||||
|
||||
// GetOrSetByTertiaryKey retrieves a value from the cache using the tertiary key, or sets it using the provided function if not found
|
||||
// The loadFunc should return the value, primary key, secondary key, and tertiary key
|
||||
func (c *TripleKeyCache[K1, K2, K3, V]) GetOrSetByTertiaryKey(ctx context.Context, tertiaryKey K3, loadFunc func() (V, K1, K2, error)) (V, error) {
|
||||
c.mu.RLock()
|
||||
if primaryKeys, exists := c.tertiaryIndex[tertiaryKey]; exists && len(primaryKeys) > 0 {
|
||||
for primaryKey := range primaryKeys {
|
||||
if value, ok := c.primaryIndex[primaryKey]; ok {
|
||||
c.mu.RUnlock()
|
||||
return value, nil
|
||||
}
|
||||
}
|
||||
}
|
||||
c.mu.RUnlock()
|
||||
|
||||
value, primaryKey, secondaryKey, err := loadFunc()
|
||||
if err != nil {
|
||||
var zero V
|
||||
return zero, err
|
||||
}
|
||||
|
||||
c.Set(ctx, primaryKey, secondaryKey, tertiaryKey, value)
|
||||
|
||||
return value, nil
|
||||
}
|
||||
1010
management/server/store/cache_test.go
Normal file
1010
management/server/store/cache_test.go
Normal file
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
@@ -4,7 +4,6 @@ import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"go.opentelemetry.io/otel/attribute"
|
||||
"go.opentelemetry.io/otel/metric"
|
||||
)
|
||||
|
||||
@@ -15,8 +14,6 @@ type StoreMetrics struct {
|
||||
persistenceDurationMicro metric.Int64Histogram
|
||||
persistenceDurationMs metric.Int64Histogram
|
||||
transactionDurationMs metric.Int64Histogram
|
||||
queryDurationMs metric.Int64Histogram
|
||||
queryCounter metric.Int64Counter
|
||||
ctx context.Context
|
||||
}
|
||||
|
||||
@@ -62,29 +59,12 @@ func NewStoreMetrics(ctx context.Context, meter metric.Meter) (*StoreMetrics, er
|
||||
return nil, err
|
||||
}
|
||||
|
||||
queryDurationMs, err := meter.Int64Histogram("management.store.query.duration.ms",
|
||||
metric.WithUnit("milliseconds"),
|
||||
metric.WithDescription("Duration of database query operations with operation type and table name"),
|
||||
)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
queryCounter, err := meter.Int64Counter("management.store.query.count",
|
||||
metric.WithDescription("Count of database query operations with operation type, table name, and status"),
|
||||
)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return &StoreMetrics{
|
||||
globalLockAcquisitionDurationMicro: globalLockAcquisitionDurationMicro,
|
||||
globalLockAcquisitionDurationMs: globalLockAcquisitionDurationMs,
|
||||
persistenceDurationMicro: persistenceDurationMicro,
|
||||
persistenceDurationMs: persistenceDurationMs,
|
||||
transactionDurationMs: transactionDurationMs,
|
||||
queryDurationMs: queryDurationMs,
|
||||
queryCounter: queryCounter,
|
||||
ctx: ctx,
|
||||
}, nil
|
||||
}
|
||||
@@ -105,13 +85,3 @@ func (metrics *StoreMetrics) CountPersistenceDuration(duration time.Duration) {
|
||||
func (metrics *StoreMetrics) CountTransactionDuration(duration time.Duration) {
|
||||
metrics.transactionDurationMs.Record(metrics.ctx, duration.Milliseconds())
|
||||
}
|
||||
|
||||
// CountStoreOperation records a store operation with its method name, status, and duration
|
||||
func (metrics *StoreMetrics) CountStoreOperation(method string, duration time.Duration) {
|
||||
attrs := []attribute.KeyValue{
|
||||
attribute.String("method", method),
|
||||
}
|
||||
|
||||
metrics.queryDurationMs.Record(metrics.ctx, duration.Milliseconds(), metric.WithAttributes(attrs...))
|
||||
metrics.queryCounter.Add(metrics.ctx, 1, metric.WithAttributes(attrs...))
|
||||
}
|
||||
|
||||
7
management/server/testdata/store.sql
vendored
7
management/server/testdata/store.sql
vendored
@@ -4,7 +4,8 @@ CREATE TABLE `setup_keys` (`id` text,`account_id` text,`key` text,`name` text,`t
|
||||
CREATE TABLE `peers` (`id` text,`account_id` text,`key` text,`setup_key` text,`ip` text,`meta_hostname` text,`meta_go_os` text,`meta_kernel` text,`meta_core` text,`meta_platform` text,`meta_os` text,`meta_os_version` text,`meta_wt_version` text,`meta_ui_version` text,`meta_kernel_version` text,`meta_network_addresses` text,`meta_system_serial_number` text,`meta_system_product_name` text,`meta_system_manufacturer` text,`meta_environment` text,`meta_files` text,`name` text,`dns_label` text,`peer_status_last_seen` datetime,`peer_status_connected` numeric,`peer_status_login_expired` numeric,`peer_status_requires_approval` numeric,`user_id` text,`ssh_key` text,`ssh_enabled` numeric,`login_expiration_enabled` numeric,`last_login` datetime,`created_at` datetime,`ephemeral` numeric,`location_connection_ip` text,`location_country_code` text,`location_city_name` text,`location_geo_name_id` integer,PRIMARY KEY (`id`),CONSTRAINT `fk_accounts_peers_g` FOREIGN KEY (`account_id`) REFERENCES `accounts`(`id`));
|
||||
CREATE TABLE `users` (`id` text,`account_id` text,`role` text,`is_service_user` numeric,`non_deletable` numeric,`service_user_name` text,`auto_groups` text,`blocked` numeric,`last_login` datetime,`created_at` datetime,`issued` text DEFAULT "api",`integration_ref_id` integer,`integration_ref_integration_type` text,PRIMARY KEY (`id`),CONSTRAINT `fk_accounts_users_g` FOREIGN KEY (`account_id`) REFERENCES `accounts`(`id`));
|
||||
CREATE TABLE `personal_access_tokens` (`id` text,`user_id` text,`name` text,`hashed_token` text,`expiration_date` datetime,`created_by` text,`created_at` datetime,`last_used` datetime DEFAULT NULL,PRIMARY KEY (`id`),CONSTRAINT `fk_users_pa_ts_g` FOREIGN KEY (`user_id`) REFERENCES `users`(`id`));
|
||||
CREATE TABLE `groups` (`id` text,`account_id` text,`name` text,`issued` text,`peers` text,`integration_ref_id` integer,`integration_ref_integration_type` text,PRIMARY KEY (`id`),CONSTRAINT `fk_accounts_groups_g` FOREIGN KEY (`account_id`) REFERENCES `accounts`(`id`));
|
||||
CREATE TABLE `groups` (`id` text,`account_id` text,`name` text,`issued` text,`resources` text,`integration_ref_id` integer,`integration_ref_integration_type` text,PRIMARY KEY (`id`),CONSTRAINT `fk_accounts_groups_g` FOREIGN KEY (`account_id`) REFERENCES `accounts`(`id`));
|
||||
CREATE TABLE `group_peers` (`account_id` text,`group_id` text,`peer_id` text,PRIMARY KEY (`group_id`,`peer_id`),CONSTRAINT `fk_groups_group_peers` FOREIGN KEY (`group_id`) REFERENCES `groups`(`id`) ON DELETE CASCADE);
|
||||
CREATE TABLE `policies` (`id` text,`account_id` text,`name` text,`description` text,`enabled` numeric,`source_posture_checks` text,PRIMARY KEY (`id`),CONSTRAINT `fk_accounts_policies` FOREIGN KEY (`account_id`) REFERENCES `accounts`(`id`));
|
||||
CREATE TABLE `policy_rules` (`id` text,`policy_id` text,`name` text,`description` text,`enabled` numeric,`action` text,`destinations` text,`sources` text,`bidirectional` numeric,`protocol` text,`ports` text,`port_ranges` text,PRIMARY KEY (`id`),CONSTRAINT `fk_policies_rules` FOREIGN KEY (`policy_id`) REFERENCES `policies`(`id`) ON DELETE CASCADE);
|
||||
CREATE TABLE `routes` (`id` text,`account_id` text,`network` text,`domains` text,`keep_route` numeric,`net_id` text,`description` text,`peer` text,`peer_groups` text,`network_type` integer,`masquerade` numeric,`metric` integer,`enabled` numeric,`groups` text,`access_control_groups` text,PRIMARY KEY (`id`),CONSTRAINT `fk_accounts_routes_g` FOREIGN KEY (`account_id`) REFERENCES `accounts`(`id`));
|
||||
@@ -24,6 +25,7 @@ CREATE INDEX `idx_peers_account_id_ip` ON `peers`(`account_id`,`ip`);
|
||||
CREATE INDEX `idx_users_account_id` ON `users`(`account_id`);
|
||||
CREATE INDEX `idx_personal_access_tokens_user_id` ON `personal_access_tokens`(`user_id`);
|
||||
CREATE INDEX `idx_groups_account_id` ON `groups`(`account_id`);
|
||||
CREATE INDEX `idx_group_peers_account_id` ON `group_peers`(`account_id`);
|
||||
CREATE INDEX `idx_policies_account_id` ON `policies`(`account_id`);
|
||||
CREATE INDEX `idx_policy_rules_policy_id` ON `policy_rules`(`policy_id`);
|
||||
CREATE INDEX `idx_routes_account_id` ON `routes`(`account_id`);
|
||||
@@ -40,7 +42,8 @@ CREATE INDEX `idx_networks_account_id` ON `networks`(`account_id`);
|
||||
|
||||
INSERT INTO accounts VALUES('bf1c8084-ba50-4ce7-9439-34653001fc3b','edafee4e-63fb-11ec-90d6-0242ac120003','2024-10-02 16:03:06.778746+02:00','test.com','private',1,'af1c8024-ha40-4ce2-9418-34653101fc3c','{"IP":"100.64.0.0","Mask":"//8AAA=="}','',0,'[]',0,86400000000000,0,0,0,'',NULL,NULL,NULL);
|
||||
INSERT INTO accounts VALUES('9439-34653001fc3b-bf1c8084-ba50-4ce7','90d6-0242ac120003-edafee4e-63fb-11ec','2024-10-02 16:01:38.210000+02:00','test2.com','private',1,'af1c8024-ha40-4ce2-9418-34653101fc3c','{"IP":"100.64.0.0","Mask":"//8AAA=="}','',0,'[]',0,86400000000000,0,0,0,'',NULL,NULL,NULL);
|
||||
INSERT INTO account_onboardings VALUES('9439-34653001fc3b-bf1c8084-ba50-4ce7','2024-10-02 16:01:38.210000+02:00','2021-08-19 20:46:20.005936822+02:00',1,0);INSERT INTO "groups" VALUES('cs1tnh0hhcjnqoiuebeg','bf1c8084-ba50-4ce7-9439-34653001fc3b','All','api','[]',0,'');
|
||||
INSERT INTO account_onboardings VALUES('9439-34653001fc3b-bf1c8084-ba50-4ce7','2024-10-02 16:01:38.210000+02:00','2021-08-19 20:46:20.005936822+02:00',1,0);
|
||||
INSERT INTO "groups" VALUES('cs1tnh0hhcjnqoiuebeg','bf1c8084-ba50-4ce7-9439-34653001fc3b','All','api','[]',0,'');
|
||||
INSERT INTO setup_keys VALUES('','bf1c8084-ba50-4ce7-9439-34653001fc3b','A2C8E62B-38F5-4553-B31E-DD66C696CEBB','Default key','reusable','2021-08-19 20:46:20.005936822+02:00','2321-09-18 20:46:20.005936822+02:00','2021-08-19 20:46:20.005936822+02:00',0,0,NULL,'["cs1tnh0hhcjnqoiuebeg"]',0,0);
|
||||
INSERT INTO users VALUES('a23efe53-63fb-11ec-90d6-0242ac120003','bf1c8084-ba50-4ce7-9439-34653001fc3b','owner',0,0,'','[]',0,NULL,'2024-10-02 16:03:06.779156+02:00','api',0,'');
|
||||
INSERT INTO users VALUES('edafee4e-63fb-11ec-90d6-0242ac120003','bf1c8084-ba50-4ce7-9439-34653001fc3b','admin',0,0,'','[]',0,NULL,'2024-10-02 16:03:06.779156+02:00','api',0,'');
|
||||
|
||||
@@ -66,19 +66,19 @@ type Account struct {
|
||||
DomainCategory string
|
||||
IsDomainPrimaryAccount bool
|
||||
SetupKeys map[string]*SetupKey `gorm:"-"`
|
||||
SetupKeysG []SetupKey `json:"-" gorm:"foreignKey:AccountID;references:id"`
|
||||
SetupKeysG []SetupKey `gorm:"foreignKey:AccountID;references:id"`
|
||||
Network *Network `gorm:"embedded;embeddedPrefix:network_"`
|
||||
Peers map[string]*nbpeer.Peer `gorm:"-"`
|
||||
PeersG []nbpeer.Peer `json:"-" gorm:"foreignKey:AccountID;references:id"`
|
||||
PeersG []nbpeer.Peer `gorm:"foreignKey:AccountID;references:id"`
|
||||
Users map[string]*User `gorm:"-"`
|
||||
UsersG []User `json:"-" gorm:"foreignKey:AccountID;references:id"`
|
||||
UsersG []User `gorm:"foreignKey:AccountID;references:id"`
|
||||
Groups map[string]*Group `gorm:"-"`
|
||||
GroupsG []*Group `json:"-" gorm:"foreignKey:AccountID;references:id"`
|
||||
GroupsG []*Group `gorm:"foreignKey:AccountID;references:id"`
|
||||
Policies []*Policy `gorm:"foreignKey:AccountID;references:id"`
|
||||
Routes map[route.ID]*route.Route `gorm:"-"`
|
||||
RoutesG []route.Route `json:"-" gorm:"foreignKey:AccountID;references:id"`
|
||||
RoutesG []route.Route `gorm:"foreignKey:AccountID;references:id"`
|
||||
NameServerGroups map[string]*nbdns.NameServerGroup `gorm:"-"`
|
||||
NameServerGroupsG []nbdns.NameServerGroup `json:"-" gorm:"foreignKey:AccountID;references:id"`
|
||||
NameServerGroupsG []nbdns.NameServerGroup `gorm:"foreignKey:AccountID;references:id"`
|
||||
DNSSettings DNSSettings `gorm:"embedded;embeddedPrefix:dns_settings_"`
|
||||
PostureChecks []*posture.Checks `gorm:"foreignKey:AccountID;references:id"`
|
||||
// Settings is a dictionary of Account settings
|
||||
|
||||
@@ -17,7 +17,7 @@ type Group struct {
|
||||
ID string `gorm:"primaryKey"`
|
||||
|
||||
// AccountID is a reference to Account that this object belongs
|
||||
AccountID string `json:"-" gorm:"index"`
|
||||
AccountID string `gorm:"index"`
|
||||
|
||||
// Name visible in the UI
|
||||
Name string
|
||||
|
||||
@@ -55,7 +55,7 @@ type Policy struct {
|
||||
ID string `gorm:"primaryKey"`
|
||||
|
||||
// AccountID is a reference to Account that this object belongs
|
||||
AccountID string `json:"-" gorm:"index"`
|
||||
AccountID string `gorm:"index"`
|
||||
|
||||
// Name of the Policy
|
||||
Name string
|
||||
|
||||
@@ -43,7 +43,7 @@ type PolicyRule struct {
|
||||
ID string `gorm:"primaryKey"`
|
||||
|
||||
// PolicyID is a reference to Policy that this object belongs
|
||||
PolicyID string `json:"-" gorm:"index"`
|
||||
PolicyID string `gorm:"index"`
|
||||
|
||||
// Name of the rule visible in the UI
|
||||
Name string
|
||||
|
||||
@@ -33,7 +33,7 @@ type SetupKeyType string
|
||||
type SetupKey struct {
|
||||
Id string
|
||||
// AccountID is a reference to Account that this object belongs
|
||||
AccountID string `json:"-" gorm:"index"`
|
||||
AccountID string `gorm:"index"`
|
||||
Key string
|
||||
KeySecret string `gorm:"index"`
|
||||
Name string
|
||||
|
||||
@@ -72,7 +72,7 @@ type UserInfo struct {
|
||||
type User struct {
|
||||
Id string `gorm:"primaryKey"`
|
||||
// AccountID is a reference to Account that this object belongs
|
||||
AccountID string `json:"-" gorm:"index"`
|
||||
AccountID string `gorm:"index"`
|
||||
Role UserRole
|
||||
IsServiceUser bool
|
||||
// NonDeletable indicates whether the service user can be deleted
|
||||
@@ -82,7 +82,7 @@ type User struct {
|
||||
// AutoGroups is a list of Group IDs to auto-assign to peers registered by this user
|
||||
AutoGroups []string `gorm:"serializer:json"`
|
||||
PATs map[string]*PersonalAccessToken `gorm:"-"`
|
||||
PATsG []PersonalAccessToken `json:"-" gorm:"foreignKey:UserID;references:id;constraint:OnDelete:CASCADE;"`
|
||||
PATsG []PersonalAccessToken `gorm:"foreignKey:UserID;references:id;constraint:OnDelete:CASCADE;"`
|
||||
// Blocked indicates whether the user is blocked. Blocked users can't use the system.
|
||||
Blocked bool
|
||||
// PendingApproval indicates whether the user requires approval before being activated
|
||||
|
||||
Reference in New Issue
Block a user