Compare commits

...

3 Commits

Author SHA1 Message Date
bfe31c28c8 no need to export sessionCache methods
rename upsert to insert since we only call it if there isn't a session.
2024-10-23 17:22:03 +01:00
abc1ea9d3f rename Q3RCON_STALE_SESSION_TIMEOUT to Q3RCON_SESSION_TIMEOUT 2024-10-23 16:45:05 +01:00
82ca15f70e add sessionCache, use it to store sessions. 2024-10-23 16:44:35 +01:00
3 changed files with 60 additions and 27 deletions

View File

@ -31,19 +31,19 @@ func main() {
host = "0.0.0.0"
}
staleTimeout, err := getEnvInt("Q3RCON_STALE_SESSION_TIMEOUT")
sessionTimeout, err := getEnvInt("Q3RCON_SESSION_TIMEOUT")
if err != nil {
log.Fatalf("unable to parse Q3RCON_STALE_SESSION_TIMEOUT: %s", err.Error())
log.Fatalf("unable to parse Q3RCON_SESSION_TIMEOUT: %s", err.Error())
}
for _, proxy := range strings.Split(proxies, ";") {
go start(host, proxy, staleTimeout)
go start(host, proxy, sessionTimeout)
}
<-make(chan int)
}
func start(host, proxy string, staleTimeout int) {
func start(host, proxy string, sessionTimeout int) {
port, target := func() (string, string) {
x := strings.Split(proxy, ":")
return x[0], x[1]
@ -52,7 +52,7 @@ func start(host, proxy string, staleTimeout int) {
c, err := udpproxy.New(
fmt.Sprintf("%s:%s", host, port),
fmt.Sprintf("127.0.0.1:%s", target),
udpproxy.WithStaleTimeout(time.Duration(staleTimeout)*time.Minute))
udpproxy.WithSessionTimeout(time.Duration(sessionTimeout)*time.Minute))
if err != nil {
log.Fatal(err)
}

View File

@ -0,0 +1,41 @@
package udpproxy
import "sync"
// sessionCache tracks connection sessions
type sessionCache struct {
mu sync.RWMutex
data map[string]*session
}
// newSessionCache creates a usable sessionCache.
func newSessionCache() sessionCache {
return sessionCache{
data: make(map[string]*session),
}
}
// read returns the associated session for an addr
func (sc *sessionCache) read(addr string) (*session, bool) {
sc.mu.RLock()
defer sc.mu.RUnlock()
v, ok := sc.data[addr]
return v, ok
}
// insert adds a session for a given addr.
func (sc *sessionCache) insert(addr string, session *session) {
sc.mu.Lock()
defer sc.mu.Unlock()
sc.data[addr] = session
}
// delete removes the session for the given addr.
func (sc *sessionCache) delete(addr string) {
sc.mu.Lock()
defer sc.mu.Unlock()
delete(sc.data, addr)
}

View File

@ -2,7 +2,6 @@ package udpproxy
import (
"net"
"sync"
"time"
log "github.com/sirupsen/logrus"
@ -11,15 +10,15 @@ import (
// Option is a functional option type that allows us to configure the Client.
type Option func(*Client)
// WithStaleTimeout is a functional option to set the stale session timeout
func WithStaleTimeout(timeout time.Duration) Option {
// WithSessionTimeout is a functional option to set the session timeout
func WithSessionTimeout(timeout time.Duration) Option {
return func(c *Client) {
if timeout < time.Minute {
log.Warnf("cannot set stale session timeout to less than 1 minute.. defaulting to 5 minutes")
return
}
c.timeout = timeout
c.sessionTimeout = timeout
}
}
@ -29,10 +28,8 @@ type Client struct {
proxyConn *net.UDPConn
mutex sync.RWMutex
sessions map[string]*session
timeout time.Duration
sessionCache sessionCache
sessionTimeout time.Duration
}
func New(port, target string, options ...Option) (*Client, error) {
@ -47,11 +44,10 @@ func New(port, target string, options ...Option) (*Client, error) {
}
c := &Client{
laddr: laddr,
raddr: raddr,
mutex: sync.RWMutex{},
sessions: map[string]*session{},
timeout: 5 * time.Minute,
laddr: laddr,
raddr: raddr,
sessionCache: newSessionCache(),
sessionTimeout: 5 * time.Minute,
}
for _, o := range options {
@ -77,7 +73,7 @@ func (c *Client) ListenAndServe() error {
log.Error(err)
}
session, ok := c.sessions[caddr.String()]
session, ok := c.sessionCache.read(caddr.String())
if !ok {
session, err = newSession(caddr, c.raddr, c.proxyConn)
if err != nil {
@ -85,7 +81,7 @@ func (c *Client) ListenAndServe() error {
continue
}
c.sessions[caddr.String()] = session
c.sessionCache.insert(caddr.String(), session)
}
go session.proxyTo(buf[:n])
@ -95,16 +91,12 @@ func (c *Client) ListenAndServe() error {
func (c *Client) pruneSessions() {
ticker := time.NewTicker(1 * time.Minute)
// the locks here could be abusive and i dont even know if this is a real
// problem but we definitely need to clean up stale sessions
for range ticker.C {
for _, session := range c.sessions {
c.mutex.RLock()
if time.Since(session.updateTime) > c.timeout {
delete(c.sessions, session.caddr.String())
for _, session := range c.sessionCache.data {
if time.Since(session.updateTime) > c.sessionTimeout {
c.sessionCache.delete(session.caddr.String())
log.Tracef("session for %s deleted", session.caddr)
}
c.mutex.RUnlock()
}
}
}