monoadmin/the-other-dude
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(poller): wire tunnel manager and SSH relay into main

Browse Source 
Add TunnelManager, TunnelResponder, SSH relay server, and SSH relay HTTP
server to the poller startup sequence with env-configurable port ranges,
idle timeouts, and session limits. Extends graceful shutdown to cover the
HTTP server (5s context), tunnel manager, and SSH relay server via defer.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Jason Staack
2026-03-12 15:35:55 -05:00
parent c73466c5e0
commit cb427272ed
2 changed files with 94 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

60
poller/cmd/poller/main.go
View File

@@ -9,6 +9,7 @@ package main
import (
"context"
"log/slog"
"net/http"
"os"
"os/signal"
"syscall"
@@ -21,7 +22,9 @@ import (
"github.com/mikrotik-portal/poller/internal/config"
"github.com/mikrotik-portal/poller/internal/observability"
"github.com/mikrotik-portal/poller/internal/poller"
"github.com/mikrotik-portal/poller/internal/sshrelay"
"github.com/mikrotik-portal/poller/internal/store"
"github.com/mikrotik-portal/poller/internal/tunnel"
"github.com/mikrotik-portal/poller/internal/vault"
)
@@ -186,6 +189,56 @@ func main() {
defer credentialSub.Stop()
slog.Info("NATS credential subscriber started (device.credential_changed.>)")
// -----------------------------------------------------------------------
// Initialize WinBox tunnel manager
// -----------------------------------------------------------------------
tunnelMgr := tunnel.NewManager(
cfg.TunnelPortMin,
cfg.TunnelPortMax,
time.Duration(cfg.TunnelIdleTimeout)*time.Second,
deviceStore,
credentialCache,
)
defer tunnelMgr.Shutdown()
slog.Info("tunnel manager initialized",
"port_min", cfg.TunnelPortMin,
"port_max", cfg.TunnelPortMax,
"idle_timeout_s", cfg.TunnelIdleTimeout,
)
// -----------------------------------------------------------------------
// Subscribe NATS tunnel responder
// -----------------------------------------------------------------------
tunnelResp := bus.NewTunnelResponder(publisher.Conn(), tunnelMgr, deviceStore, credentialCache)
if err := tunnelResp.Subscribe(); err != nil {
slog.Error("failed to subscribe tunnel responder", "error", err)
os.Exit(1)
}
defer tunnelResp.Stop()
slog.Info("NATS tunnel responder started (tunnel.*)")
// -----------------------------------------------------------------------
// Initialize SSH relay server and HTTP listener
// -----------------------------------------------------------------------
sshServer := sshrelay.NewServer(redisClient, credentialCache, deviceStore, sshrelay.Config{
IdleTimeout: time.Duration(cfg.SSHIdleTimeout) * time.Second,
MaxSessions: cfg.SSHMaxSessions,
MaxPerUser: cfg.SSHMaxPerUser,
MaxPerDevice: cfg.SSHMaxPerDevice,
})
defer sshServer.Shutdown()
httpServer := &http.Server{
Addr: ":" + cfg.SSHRelayPort,
Handler: sshServer.Handler(),
}
go func() {
slog.Info("SSH relay HTTP server starting", "port", cfg.SSHRelayPort)
if err := httpServer.ListenAndServe(); err != nil && err != http.ErrServerClosed {
slog.Error("SSH relay HTTP server error", "error", err)
}
}()
// -----------------------------------------------------------------------
// Start observability HTTP server (Prometheus metrics + health endpoint)
// -----------------------------------------------------------------------
@@ -227,5 +280,12 @@ func main() {
os.Exit(1)
}
// Gracefully shut down the SSH relay HTTP server.
shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 5*time.Second)
defer shutdownCancel()
if err := httpServer.Shutdown(shutdownCtx); err != nil {
slog.Warn("SSH relay HTTP server shutdown error", "error", err)
}
slog.Info("poller shutdown complete")
}