From 4df81f032dee948518e5a23ec3a38a66dae925ba Mon Sep 17 00:00:00 2001 From: Lenart Kos <39205323+Lenart12@users.noreply.github.com> Date: Thu, 22 Jan 2026 00:09:35 +0100 Subject: [PATCH] Fix race condition during first connection (#509) --- server/docker.go | 120 +++++++++++++++++++++++++++-------------------- 1 file changed, 70 insertions(+), 50 deletions(-) diff --git a/server/docker.go b/server/docker.go index ebbdf75..6d0a05a 100644 --- a/server/docker.go +++ b/server/docker.go @@ -62,8 +62,10 @@ func NewDockerWatcher(socket string, timeoutSeconds int, refreshIntervalSeconds type dockerWatcherImpl struct { sync.RWMutex - config dockerWatcherConfig - client *client.Client + config dockerWatcherConfig + client *client.Client + containerMap map[string]*routableContainer + monitorLock sync.Mutex } func (w *dockerWatcherImpl) makeWakerFunc(rc *routableContainer) WakerFunc { @@ -108,6 +110,13 @@ func (w *dockerWatcherImpl) makeWakerFunc(rc *routableContainer) WakerFunc { } endpoint := net.JoinHostPort(data.ip, strconv.Itoa(int(data.port))) + // Update the route mappings + err = w.monitorContainers(ctx) + if err != nil { + logrus.WithError(err).Error("Docker monitoring failed") + return "", err + } + // Wait until the container is reachable deadline := time.Now().Add(60 * time.Second) for { @@ -158,6 +167,61 @@ func (w *dockerWatcherImpl) makeSleeperFunc(rc *routableContainer) SleeperFunc { } } +func (w *dockerWatcherImpl) monitorContainers(ctx context.Context) error { + w.monitorLock.Lock() + defer w.monitorLock.Unlock() + + logrus.Trace("Listing Docker containers") + containers, err := w.listContainers(ctx) + if err != nil { + logrus.WithError(err).Error("Docker failed to list containers") + return err + } + + visited := map[string]struct{}{} + for _, rs := range containers { + if oldRs, ok := w.containerMap[rs.externalContainerName]; !ok { + w.containerMap[rs.externalContainerName] = rs + logrus.WithField("routableContainer", rs).Debug("ADD") + wakerFunc := w.makeWakerFunc(rs) + sleeperFunc := w.makeSleeperFunc(rs) + if rs.externalContainerName != "" { + Routes.CreateMapping(rs.externalContainerName, rs.containerEndpoint, wakerFunc, sleeperFunc, rs.autoScaleAsleepMOTD) + } else { + Routes.SetDefaultRoute(rs.containerEndpoint, wakerFunc, sleeperFunc, rs.autoScaleAsleepMOTD) + } + } else if oldRs.containerEndpoint != rs.containerEndpoint || + oldRs.containerID != rs.containerID || + oldRs.autoScaleUp != rs.autoScaleUp || + oldRs.autoScaleDown != rs.autoScaleDown || + oldRs.autoScaleAsleepMOTD != rs.autoScaleAsleepMOTD { + w.containerMap[rs.externalContainerName] = rs + wakerFunc := w.makeWakerFunc(rs) + sleeperFunc := w.makeSleeperFunc(rs) + if rs.externalContainerName != "" { + Routes.DeleteMapping(rs.externalContainerName) + Routes.CreateMapping(rs.externalContainerName, rs.containerEndpoint, wakerFunc, sleeperFunc, rs.autoScaleAsleepMOTD) + } else { + Routes.SetDefaultRoute(rs.containerEndpoint, wakerFunc, sleeperFunc, rs.autoScaleAsleepMOTD) + } + logrus.WithFields(logrus.Fields{"old": oldRs, "new": rs}).Debug("UPDATE") + } + visited[rs.externalContainerName] = struct{}{} + } + for _, rs := range w.containerMap { + if _, ok := visited[rs.externalContainerName]; !ok { + delete(w.containerMap, rs.externalContainerName) + if rs.externalContainerName != "" { + Routes.DeleteMapping(rs.externalContainerName) + } else { + Routes.SetDefaultRoute("", nil, nil, "") + } + logrus.WithField("routableContainer", rs).Debug("DELETE") + } + } + return nil +} + func (w *dockerWatcherImpl) Start(ctx context.Context) error { var err error @@ -186,9 +250,9 @@ func (w *dockerWatcherImpl) Start(ctx context.Context) error { return err } - containerMap := map[string]*routableContainer{} + w.containerMap = map[string]*routableContainer{} for _, c := range initialContainers { - containerMap[c.externalContainerName] = c + w.containerMap[c.externalContainerName] = c wakerFunc := w.makeWakerFunc(c) sleeperFunc := w.makeSleeperFunc(c) if c.externalContainerName != "" { @@ -202,55 +266,11 @@ func (w *dockerWatcherImpl) Start(ctx context.Context) error { for { select { case <-ticker.C: - logrus.Trace("Listing Docker containers") - containers, err := w.listContainers(ctx) + err := w.monitorContainers(ctx) if err != nil { - logrus.WithError(err).Error("Docker failed to list containers") + logrus.WithError(err).Error("Docker monitoring failed") return } - - visited := map[string]struct{}{} - for _, rs := range containers { - if oldRs, ok := containerMap[rs.externalContainerName]; !ok { - containerMap[rs.externalContainerName] = rs - logrus.WithField("routableContainer", rs).Debug("ADD") - wakerFunc := w.makeWakerFunc(rs) - sleeperFunc := w.makeSleeperFunc(rs) - if rs.externalContainerName != "" { - Routes.CreateMapping(rs.externalContainerName, rs.containerEndpoint, wakerFunc, sleeperFunc, rs.autoScaleAsleepMOTD) - } else { - Routes.SetDefaultRoute(rs.containerEndpoint, wakerFunc, sleeperFunc, rs.autoScaleAsleepMOTD) - } - } else if oldRs.containerEndpoint != rs.containerEndpoint || - oldRs.containerID != rs.containerID || - oldRs.autoScaleUp != rs.autoScaleUp || - oldRs.autoScaleDown != rs.autoScaleDown || - oldRs.autoScaleAsleepMOTD != rs.autoScaleAsleepMOTD { - containerMap[rs.externalContainerName] = rs - wakerFunc := w.makeWakerFunc(rs) - sleeperFunc := w.makeSleeperFunc(rs) - if rs.externalContainerName != "" { - Routes.DeleteMapping(rs.externalContainerName) - Routes.CreateMapping(rs.externalContainerName, rs.containerEndpoint, wakerFunc, sleeperFunc, rs.autoScaleAsleepMOTD) - } else { - Routes.SetDefaultRoute(rs.containerEndpoint, wakerFunc, sleeperFunc, rs.autoScaleAsleepMOTD) - } - logrus.WithFields(logrus.Fields{"old": oldRs, "new": rs}).Debug("UPDATE") - } - visited[rs.externalContainerName] = struct{}{} - } - for _, rs := range containerMap { - if _, ok := visited[rs.externalContainerName]; !ok { - delete(containerMap, rs.externalContainerName) - if rs.externalContainerName != "" { - Routes.DeleteMapping(rs.externalContainerName) - } else { - Routes.SetDefaultRoute("", nil, nil, "") - } - logrus.WithField("routableContainer", rs).Debug("DELETE") - } - } - case <-ctx.Done(): logrus.Debug("Stopping Docker monitoring") ticker.Stop()