Timemachine/mc-router
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Allow mc-router to scale a backend StatefulSet while routing traffic to a proxy via proxyServerName (#512)

Browse Source
This commit is contained in:
Chris Farhood
2026-02-13 08:07:27 -05:00
committed by GitHub
parent 1f86f88536
commit 21f349c2da
16 changed files with 527 additions and 135 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/api_server.go
+2 -2
View File
@@ -81,7 +81,7 @@ func routesCreateHandler(writer http.ResponseWriter, request *http.Request) {
return
}
Routes.CreateMapping(definition.ServerAddress, definition.Backend, nil, nil, "")
Routes.CreateMapping(definition.ServerAddress, definition.Backend, "", nil, nil, "")
RoutesConfigLoader.SaveRoutes()
writer.WriteHeader(http.StatusCreated)
}
@@ -102,7 +102,7 @@ func routesSetDefault(writer http.ResponseWriter, request *http.Request) {
return
}
Routes.SetDefaultRoute(body.Backend, nil, nil, "")
Routes.SetDefaultRoute(body.Backend, "", nil, nil, "")
RoutesConfigLoader.SaveRoutes()
writer.WriteHeader(http.StatusOK)
}
server/connector.go
+13 -8
View File
@@ -77,6 +77,7 @@ func NewConnector(ctx context.Context, metrics *ConnectorMetrics, sendProxyProto
recordLogins: recordLogins,
autoScaleUpAllowDenyConfig: autoScaleUpAllowDenyConfig,
activeConnections: NewActiveConnections(),
scaleActiveConnections: NewActiveConnections(),
}
}
@@ -95,6 +96,7 @@ type Connector struct {
trustedProxyNets []*net.IPNet
totalActiveConnections int32
activeConnections *ActiveConnections
scaleActiveConnections *ActiveConnections
connectionsCond *sync.Cond
ngrok NgrokConnector
clientFilter *ClientFilter
@@ -484,7 +486,7 @@ func (c *Connector) readPlayerInfo(protocolVersion mcproto.ProtocolVersion, buff
}
}
func (c *Connector) cleanupBackendConnection(clientAddr net.Addr, serverAddress string, playerInfo *PlayerInfo, backendHostPort string, cleanupMetrics bool, checkScaleDown bool) {
func (c *Connector) cleanupBackendConnection(clientAddr net.Addr, serverAddress string, playerInfo *PlayerInfo, backendHostPort string, scalingTarget string, cleanupMetrics bool, checkScaleDown bool) {
if c.connectionNotifier != nil {
err := c.connectionNotifier.NotifyDisconnected(c.ctx, clientAddr, serverAddress, playerInfo, backendHostPort)
if err != nil {
@@ -501,6 +503,8 @@ func (c *Connector) cleanupBackendConnection(clientAddr net.Addr, serverAddress
With("server_address", serverAddress).
Set(float64(c.activeConnections.GetCount(backendHostPort)))
c.scaleActiveConnections.Decrement(scalingTarget)
if c.recordLogins && playerInfo != nil {
c.metrics.ServerActivePlayer.
With("player_name", playerInfo.Name).
@@ -514,8 +518,8 @@ func (c *Connector) cleanupBackendConnection(clientAddr net.Addr, serverAddress
WithField("backendHostPort", backendHostPort).
WithField("connectionCount", c.activeConnections.GetCount(backendHostPort)).
Info("Closed connection to backend")
if checkScaleDown && c.activeConnections.GetCount(backendHostPort) <= 0 {
DownScaler.Begin(backendHostPort)
if checkScaleDown && c.scaleActiveConnections.GetCount(scalingTarget) <= 0 {
DownScaler.Begin(scalingTarget)
}
c.connectionsCond.Signal()
}
@@ -523,12 +527,12 @@ func (c *Connector) cleanupBackendConnection(clientAddr net.Addr, serverAddress
func (c *Connector) findAndConnectBackend(frontendConn net.Conn,
clientAddr net.Addr, preReadContent io.Reader, serverAddress string, playerInfo *PlayerInfo, nextState mcproto.State, isLegacy bool, clientProtocol int) {
backendHostPort, resolvedHost, waker, _ := Routes.FindBackendForServerAddress(c.ctx, serverAddress)
backendHostPort, resolvedHost, scalingTarget, waker, _ := Routes.FindBackendForServerAddress(c.ctx, serverAddress)
cleanupMetrics := false
cleanupCheckScaleDown := false
defer func() {
c.cleanupBackendConnection(clientAddr, serverAddress, playerInfo, backendHostPort, cleanupMetrics, cleanupCheckScaleDown)
c.cleanupBackendConnection(clientAddr, serverAddress, playerInfo, backendHostPort, scalingTarget, cleanupMetrics, cleanupCheckScaleDown)
}()
if waker != nil && nextState > mcproto.StateStatus {
@@ -541,8 +545,8 @@ func (c *Connector) findAndConnectBackend(frontendConn net.Conn,
Debug("checked if player is allowed to wake up the server")
if serverAllowsPlayer {
// Cancel down scaler if active before scale up
if backendHostPort != "" {
DownScaler.Cancel(backendHostPort)
if scalingTarget != "" {
DownScaler.Cancel(scalingTarget)
}
cleanupCheckScaleDown = true
logrus.WithField("serverAddress", serverAddress).Info("Waking up backend server")
@@ -558,7 +562,7 @@ func (c *Connector) findAndConnectBackend(frontendConn net.Conn,
return
}
// Cancel again in case any routes were changed during wake up
DownScaler.Cancel(newBackendHostPort)
DownScaler.Cancel(scalingTarget)
backendHostPort = newBackendHostPort
logrus.WithFields(logrus.Fields{
"serverAddress": serverAddress,
@@ -642,6 +646,7 @@ func (c *Connector) findAndConnectBackend(frontendConn net.Conn,
atomic.AddInt32(&c.totalActiveConnections, 1)))
c.activeConnections.Increment(backendHostPort)
c.scaleActiveConnections.Increment(scalingTarget)
c.metrics.ServerActiveConnections.
With("server_address", serverAddress).
Set(float64(c.activeConnections.GetCount(backendHostPort)))
server/docker.go
+7 -7
View File
@@ -186,9 +186,9 @@ func (w *dockerWatcherImpl) monitorContainers(ctx context.Context) error {
wakerFunc := w.makeWakerFunc(rs)
sleeperFunc := w.makeSleeperFunc(rs)
if rs.externalContainerName != "" {
Routes.CreateMapping(rs.externalContainerName, rs.containerEndpoint, wakerFunc, sleeperFunc, rs.autoScaleAsleepMOTD)
Routes.CreateMapping(rs.externalContainerName, rs.containerEndpoint, "", wakerFunc, sleeperFunc, rs.autoScaleAsleepMOTD)
} else {
Routes.SetDefaultRoute(rs.containerEndpoint, wakerFunc, sleeperFunc, rs.autoScaleAsleepMOTD)
Routes.SetDefaultRoute(rs.containerEndpoint, "", wakerFunc, sleeperFunc, rs.autoScaleAsleepMOTD)
}
} else if oldRs.containerEndpoint != rs.containerEndpoint ||
oldRs.containerID != rs.containerID ||
@@ -200,9 +200,9 @@ func (w *dockerWatcherImpl) monitorContainers(ctx context.Context) error {
sleeperFunc := w.makeSleeperFunc(rs)
if rs.externalContainerName != "" {
Routes.DeleteMapping(rs.externalContainerName)
Routes.CreateMapping(rs.externalContainerName, rs.containerEndpoint, wakerFunc, sleeperFunc, rs.autoScaleAsleepMOTD)
Routes.CreateMapping(rs.externalContainerName, rs.containerEndpoint, "", wakerFunc, sleeperFunc, rs.autoScaleAsleepMOTD)
} else {
Routes.SetDefaultRoute(rs.containerEndpoint, wakerFunc, sleeperFunc, rs.autoScaleAsleepMOTD)
Routes.SetDefaultRoute(rs.containerEndpoint, "", wakerFunc, sleeperFunc, rs.autoScaleAsleepMOTD)
}
logrus.WithFields(logrus.Fields{"old": oldRs, "new": rs}).Debug("UPDATE")
}
@@ -214,7 +214,7 @@ func (w *dockerWatcherImpl) monitorContainers(ctx context.Context) error {
if rs.externalContainerName != "" {
Routes.DeleteMapping(rs.externalContainerName)
} else {
Routes.SetDefaultRoute("", nil, nil, "")
Routes.SetDefaultRoute("", "", nil, nil, "")
}
logrus.WithField("routableContainer", rs).Debug("DELETE")
}
@@ -256,9 +256,9 @@ func (w *dockerWatcherImpl) Start(ctx context.Context) error {
wakerFunc := w.makeWakerFunc(c)
sleeperFunc := w.makeSleeperFunc(c)
if c.externalContainerName != "" {
Routes.CreateMapping(c.externalContainerName, c.containerEndpoint, wakerFunc, sleeperFunc, c.autoScaleAsleepMOTD)
Routes.CreateMapping(c.externalContainerName, c.containerEndpoint, "", wakerFunc, sleeperFunc, c.autoScaleAsleepMOTD)
} else {
Routes.SetDefaultRoute(c.containerEndpoint, wakerFunc, sleeperFunc, c.autoScaleAsleepMOTD)
Routes.SetDefaultRoute(c.containerEndpoint, "", wakerFunc, sleeperFunc, c.autoScaleAsleepMOTD)
}
}
server/docker_swarm.go
+7 -7
View File
@@ -92,9 +92,9 @@ func (w *dockerSwarmWatcherImpl) Start(ctx context.Context) error {
wakerFunc := w.makeWakerFunc(s)
sleeperFunc := w.makeSleeperFunc(s)
if s.externalServiceName != "" {
Routes.CreateMapping(s.externalServiceName, s.containerEndpoint, wakerFunc, sleeperFunc, "")
Routes.CreateMapping(s.externalServiceName, s.containerEndpoint, "", wakerFunc, sleeperFunc, "")
} else {
Routes.SetDefaultRoute(s.containerEndpoint, wakerFunc, sleeperFunc, "")
Routes.SetDefaultRoute(s.containerEndpoint, "", wakerFunc, sleeperFunc, "")
}
}
@@ -116,9 +116,9 @@ func (w *dockerSwarmWatcherImpl) Start(ctx context.Context) error {
wakerFunc := w.makeWakerFunc(rs)
sleeperFunc := w.makeSleeperFunc(rs)
if rs.externalServiceName != "" {
Routes.CreateMapping(rs.externalServiceName, rs.containerEndpoint, wakerFunc, sleeperFunc, "")
Routes.CreateMapping(rs.externalServiceName, rs.containerEndpoint, "", wakerFunc, sleeperFunc, "")
} else {
Routes.SetDefaultRoute(rs.containerEndpoint, wakerFunc, sleeperFunc, "")
Routes.SetDefaultRoute(rs.containerEndpoint, "", wakerFunc, sleeperFunc, "")
}
} else if oldRs.containerEndpoint != rs.containerEndpoint {
serviceMap[rs.externalServiceName] = rs
@@ -126,9 +126,9 @@ func (w *dockerSwarmWatcherImpl) Start(ctx context.Context) error {
sleeperFunc := w.makeSleeperFunc(rs)
if rs.externalServiceName != "" {
Routes.DeleteMapping(rs.externalServiceName)
Routes.CreateMapping(rs.externalServiceName, rs.containerEndpoint, wakerFunc, sleeperFunc, "")
Routes.CreateMapping(rs.externalServiceName, rs.containerEndpoint, "", wakerFunc, sleeperFunc, "")
} else {
Routes.SetDefaultRoute(rs.containerEndpoint, wakerFunc, sleeperFunc, "")
Routes.SetDefaultRoute(rs.containerEndpoint, "", wakerFunc, sleeperFunc, "")
}
logrus.WithFields(logrus.Fields{"old": oldRs, "new": rs}).Debug("UPDATE")
}
@@ -140,7 +140,7 @@ func (w *dockerSwarmWatcherImpl) Start(ctx context.Context) error {
if rs.externalServiceName != "" {
Routes.DeleteMapping(rs.externalServiceName)
} else {
Routes.SetDefaultRoute("", nil, nil, "")
Routes.SetDefaultRoute("", "", nil, nil, "")
}
logrus.WithField("routableService", rs).Debug("DELETE")
}
server/k8s.go
+68 -19
View File
@@ -11,10 +11,10 @@ import (
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
apps "k8s.io/api/apps/v1"
autoscaling "k8s.io/api/autoscaling/v1"
core "k8s.io/api/core/v1"
meta "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/fields"
"k8s.io/apimachinery/pkg/types"
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/rest"
"k8s.io/client-go/tools/cache"
@@ -26,6 +26,7 @@ const (
AnnotationDefaultServer = "mc-router.itzg.me/defaultServer"
AnnotationAutoScaleUp = "mc-router.itzg.me/autoScaleUp"
AnnotationAutoScaleDown = "mc-router.itzg.me/autoScaleDown"
AnnotationProxyServerName = "mc-router.itzg.me/proxyServerName"
)
// K8sWatcher is a RouteFinder that can find routes from kubernetes services.
@@ -184,9 +185,9 @@ func (w *K8sWatcher) handleUpdate(oldObj interface{}, newObj interface{}) {
"new": newRoutableService,
}).Debug("UPDATE")
if newRoutableService.externalServiceName != "" {
w.routesHandler.CreateMapping(newRoutableService.externalServiceName, newRoutableService.containerEndpoint, newRoutableService.autoScaleUp, newRoutableService.autoScaleDown, "")
w.routesHandler.CreateMapping(newRoutableService.externalServiceName, newRoutableService.containerEndpoint, newRoutableService.scalingTarget, newRoutableService.autoScaleUp, newRoutableService.autoScaleDown, "")
} else {
w.routesHandler.SetDefaultRoute(newRoutableService.containerEndpoint, newRoutableService.autoScaleUp, newRoutableService.autoScaleDown, "")
w.routesHandler.SetDefaultRoute(newRoutableService.containerEndpoint, newRoutableService.scalingTarget, newRoutableService.autoScaleUp, newRoutableService.autoScaleDown, "")
}
}
}
@@ -201,7 +202,7 @@ func (w *K8sWatcher) handleDelete(obj interface{}) {
if routableService.externalServiceName != "" {
w.routesHandler.DeleteMapping(routableService.externalServiceName)
} else {
w.routesHandler.SetDefaultRoute("", nil, nil, "")
w.routesHandler.SetDefaultRoute("", "", nil, nil, "")
}
}
}
@@ -215,9 +216,9 @@ func (w *K8sWatcher) handleAdd(obj interface{}) {
logrus.WithField("routableService", routableService).Debug("ADD")
if routableService.externalServiceName != "" {
w.routesHandler.CreateMapping(routableService.externalServiceName, routableService.containerEndpoint, routableService.autoScaleUp, routableService.autoScaleDown, "")
w.routesHandler.CreateMapping(routableService.externalServiceName, routableService.containerEndpoint, routableService.scalingTarget, routableService.autoScaleUp, routableService.autoScaleDown, "")
} else {
w.routesHandler.SetDefaultRoute(routableService.containerEndpoint, routableService.autoScaleUp, routableService.autoScaleDown, "")
w.routesHandler.SetDefaultRoute(routableService.containerEndpoint, routableService.scalingTarget, routableService.autoScaleUp, routableService.autoScaleDown, "")
}
}
}
@@ -226,6 +227,7 @@ func (w *K8sWatcher) handleAdd(obj interface{}) {
type routableService struct {
externalServiceName string
containerEndpoint string
scalingTarget string
autoScaleUp WakerFunc
autoScaleDown SleeperFunc
}
@@ -273,11 +275,25 @@ func (w *K8sWatcher) buildDetails(service *core.Service, externalServiceName str
port = mcPort
}
endpoint := net.JoinHostPort(clusterIp, port)
routingEndpoint := endpoint
scalingTarget := endpoint // Default to service endpoint for scaling
if proxyServerName, exists := service.Annotations[AnnotationProxyServerName]; exists && proxyServerName != "" {
// Ensure the proxy address has a port
if _, _, err := net.SplitHostPort(proxyServerName); err != nil {
proxyServerName = net.JoinHostPort(proxyServerName, "25565")
}
routingEndpoint = proxyServerName
// scalingTarget remains the service endpoint (already set above)
}
wakerFunc := w.buildScaleFunction(service, 0, 1)
rs := &routableService{
externalServiceName: externalServiceName,
containerEndpoint: endpoint,
autoScaleUp: buildWakerFromSleeper(endpoint, wakerFunc),
containerEndpoint: routingEndpoint,
scalingTarget: scalingTarget,
autoScaleUp: buildWakerFromSleeper(routingEndpoint, wakerFunc),
autoScaleDown: w.buildScaleFunction(service, 1, 0),
}
return rs
@@ -332,6 +348,7 @@ func (w *K8sWatcher) buildScaleFunction(service *core.Service, from int32, to in
return func(ctx context.Context) error {
serviceName := service.Name
if statefulSetName, exists := w.mappings[serviceName]; exists {
// Get current replicas to check if scaling is needed
if scale, err := w.clientset.AppsV1().StatefulSets(service.Namespace).GetScale(ctx, statefulSetName, meta.GetOptions{}); err == nil {
replicas := scale.Status.Replicas
logrus.WithFields(logrus.Fields{
@@ -339,25 +356,57 @@ func (w *K8sWatcher) buildScaleFunction(service *core.Service, from int32, to in
"statefulSet": statefulSetName,
"replicas": replicas,
}).Debug("StatefulSet of Service Replicas")
if replicas == from {
if _, err := w.clientset.AppsV1().StatefulSets(service.Namespace).UpdateScale(ctx, statefulSetName, &autoscaling.Scale{
ObjectMeta: meta.ObjectMeta{
Name: scale.Name,
Namespace: scale.Namespace,
UID: scale.UID,
ResourceVersion: scale.ResourceVersion,
},
Spec: autoscaling.ScaleSpec{Replicas: to}}, meta.UpdateOptions{},
); err == nil {
// Use Patch instead of Update to avoid optimistic concurrency errors
// This doesn't require resourceVersion and is atomic
patchData := fmt.Sprintf(`{"spec":{"replicas":%d}}`, to)
_, err := w.clientset.AppsV1().StatefulSets(service.Namespace).Patch(
ctx,
statefulSetName,
types.StrategicMergePatchType,
[]byte(patchData),
meta.PatchOptions{},
)
if err == nil {
logrus.WithFields(logrus.Fields{
"service": serviceName,
"statefulSet": statefulSetName,
"replicas": replicas,
}).Infof("StatefulSet Replicas Autoscaled from %d to %d", from, to)
} else {
return errors.Wrapf(err, "UpdateScale for Replicas=%d failed for StatefulSet: %s", to, statefulSetName)
return nil
}
// Fallback to UpdateScale if Patch fails due to RBAC permissions
// This maintains backward compatibility with existing RBAC configurations
if strings.Contains(err.Error(), "forbidden") {
logrus.WithFields(logrus.Fields{
"service": serviceName,
"statefulSet": statefulSetName,
}).Warn("Patch operation forbidden - falling back to UpdateScale. Consider updating RBAC to allow 'patch' verb for better concurrency handling")
scale.Spec.Replicas = to
if _, updateErr := w.clientset.AppsV1().StatefulSets(service.Namespace).UpdateScale(
ctx,
statefulSetName,
scale,
meta.UpdateOptions{},
); updateErr == nil {
logrus.WithFields(logrus.Fields{
"service": serviceName,
"statefulSet": statefulSetName,
"replicas": replicas,
}).Infof("StatefulSet Replicas Autoscaled from %d to %d (via UpdateScale fallback)", from, to)
return nil
} else {
return errors.Wrapf(updateErr, "UpdateScale fallback for Replicas=%d failed for StatefulSet: %s", to, statefulSetName)
}
}
return errors.Wrapf(err, "Patch for Replicas=%d failed for StatefulSet: %s", to, statefulSetName)
}
// Replicas already at desired state
return nil
} else {
return fmt.Errorf("GetScale failed for StatefulSet %s: %w", statefulSetName, err)
}
server/k8s_test.go
+151 -10
View File
@@ -28,16 +28,16 @@ func (m *MockedRoutesHandler) GetBackendForServer(server string) string {
}
}
func (m *MockedRoutesHandler) CreateMapping(serverAddress string, backend string, waker WakerFunc, sleeper SleeperFunc, asleepMOTD string) {
m.MethodCalled("CreateMapping", serverAddress, backend, waker, sleeper, asleepMOTD)
func (m *MockedRoutesHandler) CreateMapping(serverAddress string, backend string, scaleKey string, waker WakerFunc, sleeper SleeperFunc, asleepMOTD string) {
m.MethodCalled("CreateMapping", serverAddress, backend, scaleKey, waker, sleeper, asleepMOTD)
if m.routes == nil {
m.routes = make(map[string]string)
}
m.routes[serverAddress] = backend
}
func (m *MockedRoutesHandler) SetDefaultRoute(backend string, waker WakerFunc, sleeper SleeperFunc, asleepMOTD string) {
m.MethodCalled("SetDefaultRoute", backend, waker, sleeper, asleepMOTD)
func (m *MockedRoutesHandler) SetDefaultRoute(backend string, scaleKey string, waker WakerFunc, sleeper SleeperFunc, asleepMOTD string) {
m.MethodCalled("SetDefaultRoute", backend, scaleKey, waker, sleeper, asleepMOTD)
if m.routes == nil {
m.routes = make(map[string]string)
}
@@ -183,8 +183,8 @@ func TestK8sWatcherImpl_handleAddThenUpdate(t *testing.T) {
DownScaler = NewDownScaler(context.Background(), false, 1*time.Second)
routesHandler := new(MockedRoutesHandler)
routesHandler.On("CreateMapping", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return()
routesHandler.On("SetDefaultRoute", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return()
routesHandler.On("CreateMapping", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return()
routesHandler.On("SetDefaultRoute", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return()
routesHandler.On("GetAsleepMOTD", mock.Anything).Return("")
routesHandler.On("DeleteMapping", mock.Anything).Return(true)
@@ -264,8 +264,8 @@ func TestK8sWatcherImpl_handleAddThenDelete(t *testing.T) {
DownScaler = NewDownScaler(context.Background(), false, 1*time.Second)
routesHandler := new(MockedRoutesHandler)
routesHandler.On("CreateMapping", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return()
routesHandler.On("SetDefaultRoute", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return()
routesHandler.On("CreateMapping", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return()
routesHandler.On("SetDefaultRoute", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return()
routesHandler.On("GetAsleepMOTD", mock.Anything).Return("")
routesHandler.On("DeleteMapping", mock.Anything).Return(true)
@@ -363,8 +363,8 @@ func TestK8s_externalName(t *testing.T) {
DownScaler = NewDownScaler(context.Background(), false, 1*time.Second)
routesHandler := new(MockedRoutesHandler)
routesHandler.On("CreateMapping", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return()
routesHandler.On("SetDefaultRoute", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return()
routesHandler.On("CreateMapping", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return()
routesHandler.On("SetDefaultRoute", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return()
routesHandler.On("GetAsleepMOTD", mock.Anything).Return("")
routesHandler.On("DeleteMapping", mock.Anything).Return(true)
@@ -393,3 +393,144 @@ func TestK8s_externalName(t *testing.T) {
})
}
}
func TestK8s_proxyServerName(t *testing.T) {
type scenario struct {
server string
backend string
}
tests := []struct {
name string
svc string
scenarios []scenario
}{
{
name: "proxy routes to proxy address",
svc: `{"metadata": {"annotations": {"mc-router.itzg.me/externalServerName": "mc.example.com", "mc-router.itzg.me/proxyServerName": "velocity-proxy:25577"}}, "spec":{"clusterIP": "10.0.0.5"}}`,
scenarios: []scenario{
{server: "mc.example.com", backend: "velocity-proxy:25577"},
},
},
{
name: "proxy without port gets default 25565",
svc: `{"metadata": {"annotations": {"mc-router.itzg.me/externalServerName": "mc.example.com", "mc-router.itzg.me/proxyServerName": "velocity-proxy"}}, "spec":{"clusterIP": "10.0.0.5"}}`,
scenarios: []scenario{
{server: "mc.example.com", backend: "velocity-proxy:25565"},
},
},
{
name: "no proxy annotation routes to ClusterIP",
svc: `{"metadata": {"annotations": {"mc-router.itzg.me/externalServerName": "mc.example.com"}}, "spec":{"clusterIP": "10.0.0.5"}}`,
scenarios: []scenario{
{server: "mc.example.com", backend: "10.0.0.5:25565"},
},
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
DownScaler = NewDownScaler(context.Background(), false, 1*time.Second)
routesHandler := new(MockedRoutesHandler)
routesHandler.On("CreateMapping", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return()
routesHandler.On("SetDefaultRoute", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return()
routesHandler.On("GetAsleepMOTD", mock.Anything).Return("")
routesHandler.On("DeleteMapping", mock.Anything).Return(true)
watcher := &K8sWatcher{
routesHandler: routesHandler,
}
svc := v1.Service{}
err := json.Unmarshal([]byte(test.svc), &svc)
require.NoError(t, err)
watcher.handleAdd(&svc)
for _, s := range test.scenarios {
backend := routesHandler.GetBackendForServer(s.server)
assert.Equal(t, s.backend, backend, "given=%s", s.server)
}
})
}
}
func TestK8s_proxyServerNameScaleEndpoint(t *testing.T) {
DownScaler = NewDownScaler(context.Background(), false, 1*time.Second)
routesHandler := new(MockedRoutesHandler)
routesHandler.On("CreateMapping", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return()
routesHandler.On("SetDefaultRoute", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return()
routesHandler.On("GetAsleepMOTD", mock.Anything).Return("")
routesHandler.On("DeleteMapping", mock.Anything).Return(true)
watcher := &K8sWatcher{
routesHandler: routesHandler,
}
svc := v1.Service{}
err := json.Unmarshal([]byte(`{"metadata": {"annotations": {"mc-router.itzg.me/externalServerName": "mc.example.com", "mc-router.itzg.me/proxyServerName": "velocity:25577"}}, "spec":{"clusterIP": "10.0.0.5"}}`), &svc)
require.NoError(t, err)
watcher.handleAdd(&svc)
// Verify CreateMapping was called with the correct scaleKey (original endpoint)
routesHandler.AssertCalled(t, "CreateMapping", "mc.example.com", "velocity:25577", "10.0.0.5:25565", mock.Anything, mock.Anything, mock.Anything)
}
func TestK8s_proxyServerNameUpdate(t *testing.T) {
DownScaler = NewDownScaler(context.Background(), false, 1*time.Second)
routesHandler := new(MockedRoutesHandler)
routesHandler.On("CreateMapping", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return()
routesHandler.On("SetDefaultRoute", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return()
routesHandler.On("GetAsleepMOTD", mock.Anything).Return("")
routesHandler.On("DeleteMapping", mock.Anything).Return(true)
watcher := &K8sWatcher{
routesHandler: routesHandler,
}
// Start with proxy
initialSvc := v1.Service{}
err := json.Unmarshal([]byte(`{"metadata": {"annotations": {"mc-router.itzg.me/externalServerName": "mc.example.com", "mc-router.itzg.me/proxyServerName": "velocity:25577"}}, "spec":{"clusterIP": "10.0.0.5"}}`), &initialSvc)
require.NoError(t, err)
watcher.handleAdd(&initialSvc)
assert.Equal(t, "velocity:25577", routesHandler.GetBackendForServer("mc.example.com"))
// Update to remove proxy
updatedSvc := v1.Service{}
err = json.Unmarshal([]byte(`{"metadata": {"annotations": {"mc-router.itzg.me/externalServerName": "mc.example.com"}}, "spec":{"clusterIP": "10.0.0.5"}}`), &updatedSvc)
require.NoError(t, err)
watcher.handleUpdate(&initialSvc, &updatedSvc)
assert.Equal(t, "10.0.0.5:25565", routesHandler.GetBackendForServer("mc.example.com"))
}
func TestK8s_autoScaleWithoutProxy(t *testing.T) {
DownScaler = NewDownScaler(context.Background(), false, 1*time.Second)
routesHandler := new(MockedRoutesHandler)
routesHandler.On("CreateMapping", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return()
routesHandler.On("SetDefaultRoute", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return()
routesHandler.On("GetAsleepMOTD", mock.Anything).Return("")
routesHandler.On("DeleteMapping", mock.Anything).Return(true)
watcher := &K8sWatcher{
autoScaleUp: true,
autoScaleDown: true,
routesHandler: routesHandler,
}
// Service WITHOUT proxyServerName but WITH autoScaleUp/Down annotations
svc := v1.Service{}
err := json.Unmarshal([]byte(`{"metadata": {"annotations": {"mc-router.itzg.me/externalServerName": "atm-10.example.com", "mc-router.itzg.me/autoScaleUp": "true", "mc-router.itzg.me/autoScaleDown": "true"}}, "spec":{"clusterIP": "10.0.0.10"}}`), &svc)
require.NoError(t, err)
watcher.handleAdd(&svc)
// Verify routes to ClusterIP (not proxy)
assert.Equal(t, "10.0.0.10:25565", routesHandler.GetBackendForServer("atm-10.example.com"))
// CRITICAL: Verify scaleKey is set to the service endpoint (not empty)
// This ensures auto-scaling targets the correct StatefulSet
routesHandler.AssertCalled(t, "CreateMapping", "atm-10.example.com", "10.0.0.10:25565", "10.0.0.10:25565", mock.Anything, mock.Anything, mock.Anything)
}
server/routes.go
+36 -27
View File
@@ -36,8 +36,8 @@ type RouteFinder interface {
}
type RoutesHandler interface {
CreateMapping(serverAddress string, backend string, waker WakerFunc, sleeper SleeperFunc, asleepMOTD string)
SetDefaultRoute(backend string, waker WakerFunc, sleeper SleeperFunc, asleepMOTD string)
CreateMapping(serverAddress string, backend string, scalingTarget string, waker WakerFunc, sleeper SleeperFunc, asleepMOTD string)
SetDefaultRoute(backend string, scalingTarget string, waker WakerFunc, sleeper SleeperFunc, asleepMOTD string)
// DeleteMapping requests that the serverAddress be removed from routes.
// Returns true if the route existed.
DeleteMapping(serverAddress string) bool
@@ -50,13 +50,14 @@ type IRoutes interface {
RegisterAll(mappings map[string]string)
// FindBackendForServerAddress returns the host:port for the external server address, if registered.
// Otherwise, an empty string is returned. Also returns the normalized version of the given serverAddress.
// The 3rd value returned is an (optional) "waker" function which a caller must invoke to wake up serverAddress.
// The 4th value returned is an (optional) "sleeper" function which a caller must invoke to shut down serverAddress.
// The 3rd value returned is the scalingTarget which indicates what endpoint to scale (may differ from backend when using proxy).
// The 4th value returned is an (optional) "waker" function which a caller must invoke to wake up serverAddress.
// The 5th value returned is an (optional) "sleeper" function which a caller must invoke to shut down serverAddress.
HasRoute(serverAddress string) bool
FindBackendForServerAddress(ctx context.Context, serverAddress string) (string, string, WakerFunc, SleeperFunc)
GetSleepers(backend string) []SleeperFunc
FindBackendForServerAddress(ctx context.Context, serverAddress string) (string, string, string, WakerFunc, SleeperFunc)
GetSleepers(scalingTarget string) []SleeperFunc
GetMappings() map[string]string
GetDefaultRoute() (string, WakerFunc, SleeperFunc)
GetDefaultRoute() (string, string, WakerFunc, SleeperFunc)
GetAsleepMOTD(serverAddress string) string
SimplifySRV(srvEnabled bool)
}
@@ -73,15 +74,16 @@ func NewRoutes() IRoutes {
func (r *routesImpl) RegisterAll(mappings map[string]string) {
for k, v := range mappings {
r.CreateMapping(k, v, nil, nil, "")
r.CreateMapping(k, v, "", nil, nil, "")
}
}
type mapping struct {
backend string
waker WakerFunc
sleeper SleeperFunc
asleepMOTD string
backend string
waker WakerFunc
sleeper SleeperFunc
asleepMOTD string
scalingTarget string // The endpoint to scale (may differ from backend when using proxy)
}
type routesImpl struct {
@@ -96,16 +98,19 @@ func (r *routesImpl) Reset() {
DownScaler.Reset()
}
func (r *routesImpl) SetDefaultRoute(backend string, waker WakerFunc, sleeper SleeperFunc, asleepMOTD string) {
r.defaultRoute = mapping{backend: backend, waker: waker, sleeper: sleeper, asleepMOTD: asleepMOTD}
func (r *routesImpl) SetDefaultRoute(backend string, scalingTarget string, waker WakerFunc, sleeper SleeperFunc, asleepMOTD string) {
if scalingTarget == "" {
scalingTarget = backend
}
r.defaultRoute = mapping{backend: backend, scalingTarget: scalingTarget, waker: waker, sleeper: sleeper, asleepMOTD: asleepMOTD}
logrus.WithFields(logrus.Fields{
"backend": backend,
}).Info("Using default route")
}
func (r *routesImpl) GetDefaultRoute() (string, WakerFunc, SleeperFunc) {
return r.defaultRoute.backend, r.defaultRoute.waker, r.defaultRoute.sleeper
func (r *routesImpl) GetDefaultRoute() (string, string, WakerFunc, SleeperFunc) {
return r.defaultRoute.backend, r.defaultRoute.scalingTarget, r.defaultRoute.waker, r.defaultRoute.sleeper
}
func (r *routesImpl) GetAsleepMOTD(serverAddress string) string {
@@ -134,7 +139,7 @@ func (r *routesImpl) HasRoute(serverAddress string) bool {
return exists
}
func (r *routesImpl) FindBackendForServerAddress(_ context.Context, serverAddress string) (string, string, WakerFunc, SleeperFunc) {
func (r *routesImpl) FindBackendForServerAddress(_ context.Context, serverAddress string) (string, string, string, WakerFunc, SleeperFunc) {
r.RLock()
defer r.RUnlock()
@@ -173,23 +178,23 @@ func (r *routesImpl) FindBackendForServerAddress(_ context.Context, serverAddres
if r.mappings != nil {
if mapping, exists := r.mappings[serverAddress]; exists {
return mapping.backend, serverAddress, mapping.waker, mapping.sleeper
return mapping.backend, serverAddress, mapping.scalingTarget, mapping.waker, mapping.sleeper
}
}
return r.defaultRoute.backend, serverAddress, r.defaultRoute.waker, r.defaultRoute.sleeper
return r.defaultRoute.backend, serverAddress, r.defaultRoute.scalingTarget, r.defaultRoute.waker, r.defaultRoute.sleeper
}
func (r *routesImpl) GetSleepers(backend string) []SleeperFunc {
func (r *routesImpl) GetSleepers(scalingTarget string) []SleeperFunc {
r.RLock()
defer r.RUnlock()
var sleepers []SleeperFunc
for _, m := range r.mappings {
if m.backend == backend && m.sleeper != nil {
if m.scalingTarget == scalingTarget && m.sleeper != nil {
sleepers = append(sleepers, m.sleeper)
}
}
if r.defaultRoute.backend == backend && r.defaultRoute.sleeper != nil {
if r.defaultRoute.scalingTarget == scalingTarget && r.defaultRoute.sleeper != nil {
sleepers = append(sleepers, r.defaultRoute.sleeper)
}
return sleepers
@@ -212,7 +217,7 @@ func (r *routesImpl) DeleteMapping(serverAddress string) bool {
logrus.WithField("serverAddress", serverAddress).Info("Deleting route")
if m, ok := r.mappings[serverAddress]; ok {
DownScaler.Cancel(m.backend)
DownScaler.Cancel(m.scalingTarget)
delete(r.mappings, serverAddress)
return true
} else {
@@ -220,20 +225,24 @@ func (r *routesImpl) DeleteMapping(serverAddress string) bool {
}
}
func (r *routesImpl) CreateMapping(serverAddress string, backend string, waker WakerFunc, sleeper SleeperFunc, asleepMOTD string) {
func (r *routesImpl) CreateMapping(serverAddress string, backend string, scalingTarget string, waker WakerFunc, sleeper SleeperFunc, asleepMOTD string) {
r.Lock()
defer r.Unlock()
serverAddress = strings.ToLower(serverAddress)
if scalingTarget == "" {
scalingTarget = backend
}
logrus.WithFields(logrus.Fields{
"serverAddress": serverAddress,
"backend": backend,
}).Info("Created route mapping")
r.mappings[serverAddress] = mapping{backend: backend, waker: waker, sleeper: sleeper, asleepMOTD: asleepMOTD}
r.mappings[serverAddress] = mapping{backend: backend, scalingTarget: scalingTarget, waker: waker, sleeper: sleeper, asleepMOTD: asleepMOTD}
// Trigger auto scale down when mapping is created to ensure servers are shut down if router restarts
if DownScaler != nil && backend != "" {
DownScaler.Begin(backend)
if DownScaler != nil && scalingTarget != "" {
DownScaler.Begin(scalingTarget)
}
}
server/routes_config_loader.go
+3 -3
View File
@@ -44,7 +44,7 @@ func (r *routesConfigLoader) Load(routesConfigFileName string) error {
}
Routes.RegisterAll(config.Mappings)
Routes.SetDefaultRoute(config.DefaultServer, nil, nil, "")
Routes.SetDefaultRoute(config.DefaultServer, "", nil, nil, "")
return nil
}
@@ -62,7 +62,7 @@ func (r *routesConfigLoader) Reload() error {
logrus.WithField("routesConfig", r.fileName).Info("Re-loading routes config file")
Routes.Reset()
Routes.RegisterAll(config.Mappings)
Routes.SetDefaultRoute(config.DefaultServer, nil, nil, "")
Routes.SetDefaultRoute(config.DefaultServer, "", nil, nil, "")
return nil
}
@@ -135,7 +135,7 @@ func (r *routesConfigLoader) SaveRoutes() {
return
}
server, _, _ := Routes.GetDefaultRoute()
server, _, _, _ := Routes.GetDefaultRoute()
err := r.writeFile(&RoutesConfigSchema{
DefaultServer: server,
Mappings: Routes.GetMappings(),
server/routes_test.go
+69 -2
View File
@@ -3,8 +3,10 @@ package server
import (
"context"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func Test_routesImpl_FindBackendForServerAddress(t *testing.T) {
@@ -66,9 +68,9 @@ func Test_routesImpl_FindBackendForServerAddress(t *testing.T) {
t.Run(tt.name, func(t *testing.T) {
r := NewRoutes()
r.CreateMapping(tt.mapping.serverAddress, tt.mapping.backend, nil, nil, "")
r.CreateMapping(tt.mapping.serverAddress, tt.mapping.backend, "", nil, nil, "")
if got, server, _, _ := r.FindBackendForServerAddress(context.Background(), tt.args.serverAddress); got != tt.want {
if got, server, _, _, _ := r.FindBackendForServerAddress(context.Background(), tt.args.serverAddress); got != tt.want {
t.Errorf("routesImpl.FindBackendForServerAddress() = %v, want %v", got, tt.want)
} else {
assert.Equal(t, tt.mapping.serverAddress, server)
@@ -76,3 +78,68 @@ func Test_routesImpl_FindBackendForServerAddress(t *testing.T) {
})
}
}
func Test_routesImpl_ScaleKey(t *testing.T) {
DownScaler = NewDownScaler(context.Background(), false, 1*time.Second)
t.Run("scaleKey defaults to backend when empty", func(t *testing.T) {
r := NewRoutes()
r.CreateMapping("mc.example.com", "backend:25565", "", nil, nil, "")
_, _, scaleKey, _, _ := r.FindBackendForServerAddress(context.Background(), "mc.example.com")
assert.Equal(t, "backend:25565", scaleKey)
})
t.Run("scaleKey is set when provided", func(t *testing.T) {
r := NewRoutes()
r.CreateMapping("mc.example.com", "proxy:25577", "10.0.0.5:25565", nil, nil, "")
backend, _, scaleKey, _, _ := r.FindBackendForServerAddress(context.Background(), "mc.example.com")
assert.Equal(t, "proxy:25577", backend)
assert.Equal(t, "10.0.0.5:25565", scaleKey)
})
t.Run("GetSleepers matches on scaleKey not backend", func(t *testing.T) {
r := NewRoutes()
called := false
sleeper := func(ctx context.Context) error {
called = true
return nil
}
// Two routes with same proxy backend but different scaleKeys
r.CreateMapping("mc1.example.com", "proxy:25577", "10.0.0.1:25565", nil, sleeper, "")
r.CreateMapping("mc2.example.com", "proxy:25577", "10.0.0.2:25565", nil, nil, "")
sleepers := r.GetSleepers("10.0.0.1:25565")
require.Len(t, sleepers, 1)
_ = sleepers[0](context.Background())
assert.True(t, called)
// No sleeper for the second scaleKey since it has nil sleeper
sleepers = r.GetSleepers("10.0.0.2:25565")
assert.Empty(t, sleepers)
// No sleeper when querying by proxy backend address
sleepers = r.GetSleepers("proxy:25577")
assert.Empty(t, sleepers)
})
t.Run("default route scaleKey", func(t *testing.T) {
r := NewRoutes()
r.SetDefaultRoute("proxy:25577", "10.0.0.5:25565", nil, nil, "")
backend, scaleKey, _, _ := r.GetDefaultRoute()
assert.Equal(t, "proxy:25577", backend)
assert.Equal(t, "10.0.0.5:25565", scaleKey)
})
t.Run("default route scaleKey defaults to backend", func(t *testing.T) {
r := NewRoutes()
r.SetDefaultRoute("backend:25565", "", nil, nil, "")
backend, scaleKey, _, _ := r.GetDefaultRoute()
assert.Equal(t, "backend:25565", backend)
assert.Equal(t, "backend:25565", scaleKey)
})
}
server/server.go
+1 -1
View File
@@ -73,7 +73,7 @@ func NewServer(ctx context.Context, config *Config) (*Server, error) {
Routes.RegisterAll(config.Mapping)
if config.Default != "" {
Routes.SetDefaultRoute(config.Default, nil, nil, "")
Routes.SetDefaultRoute(config.Default, "", nil, nil, "")
}
if config.ConnectionRateLimit < 1 {