refactor: Improve Docker event stream resilience and logging

- Enhance container event listener with advanced EOF and error handling
- Add configurable reconnection strategies for different error types
- Implement more granular logging for event stream connections
- Reduce log noise with intelligent logging thresholds
- Add Podman socket compatibility in Docker client initialization
- Improve debug logging for Docker client and event stream operations

commit: d95b77e02d0a460390f424640b1c1dd19daaaeb0
author: bamen · 19mo ago · 2025-03-10 16:52 UTC
parent: 9fac95c
signature: unsigned

2 files · +77 -26

M pkg/docker/client.go +8 -1


@@ -6,6 +6,7 @@ import (
6	6		"os"
7	7		"time"
8	8
	9	+	"github.com/charmbracelet/log"
9	10		"github.com/docker/docker/api/types"
10	11		"github.com/docker/docker/api/types/container"
11	12		"github.com/docker/docker/client"

@@ -30,13 +31,19 @@ func InitializeClient(config *Config) error {
30	31		return err
31	32		}
32	33
	34	+	// For Podman compatibility, we need to ensure we're using the right connection approach
	35	+	host := "unix://" + config.Sock
	36	+
	37	+	// Create the Docker client with appropriate options
33	38		cli, err := client.NewClientWithOpts(
34	39		client.WithAPIVersionNegotiation(),
35		-	client.WithHost("unix://"+config.Sock),
	40	+	client.WithHost(host),
36	41		)
37	42		if err != nil {
38	43		return fmt.Errorf("error initializing Docker client: %s", err)
39	44		}
	45	+
	46	+	log.Debug("Docker client initialized", "socket", config.Sock, "podman_enabled", config.PodmanEnable)
40	47		dockerCli = cli
41	48		currentConfig = config
42	49		return nil

 	"os"
 	"time"
+	"github.com/charmbracelet/log"
 	"github.com/docker/docker/api/types"
 	"github.com/docker/docker/api/types/container"
 	"github.com/docker/docker/client"
 		return err
 	}
+	// For Podman compatibility, we need to ensure we're using the right connection approach
+	host := "unix://" + config.Sock
+	// Create the Docker client with appropriate options
 	cli, err := client.NewClientWithOpts(
 		client.WithAPIVersionNegotiation(),
-		client.WithHost("unix://"+config.Sock),
+		client.WithHost(host),
 	)
 	if err != nil {
 		return fmt.Errorf("error initializing Docker client: %s", err)
 	}
+	log.Debug("Docker client initialized", "socket", config.Sock, "podman_enabled", config.PodmanEnable)
 	dockerCli = cli
 	currentConfig = config
 	return nil

M pkg/docker/containers.go +69 -25


@@ -760,6 +760,13 @@ func ListenForContainerEvents(networkFilter string, callback func(string, string
760	760		consecutiveFailures := 0
761	761		maxConsecutiveFailures := 10
762	762
	763	+	// Track EOFs for reduced logging
	764	+	eofCount := 0
	765	+	lastEOFLogTime := time.Now().Add(-10 * time.Minute) // Initialize to past time
	766	+
	767	+	// Fixed delay for EOFs to prevent rapid reconnection
	768	+	eofReconnectDelay := 5 * time.Second
	769	+
763	770		for {
764	771		// Check if context was canceled (application is shutting down)
765	772		if ctx.Err() != nil {

@@ -767,18 +774,24 @@ func ListenForContainerEvents(networkFilter string, callback func(string, string
767	774		}
768	775
769	776		// Add a timeout context for the Events call
770		-	// This helps prevent long-running connections that might become stale
771		-	eventCtx, eventCancel := context.WithTimeout(ctx, 5*time.Minute)
	777	+	eventCtx, eventCancel := context.WithCancel(ctx)
772	778
773	779		// Set up filters for the events we're interested in
774	780		filters := makeEventFilters(networkFilter)
775	781
776		-	// Start listening for events with timeout context
	782	+	log.Debug("connecting to container event stream", "network_filter", networkFilter)
	783	+
	784	+	// Start listening for events
777	785		messages, errs := dockerCli.Events(eventCtx, events.ListOptions{
778	786		Filters: filters,
779	787		})
780	788
781		-	log.Info("Started container event listener", "network_filter", networkFilter)
	789	+	// Only log the first connection and every 10th reconnection for EOF
	790	+	if eofCount == 0 {
	791	+	log.Info("Started container event listener", "network_filter", networkFilter)
	792	+	} else {
	793	+	log.Debug("Reconnected container event listener", "network_filter", networkFilter, "reconnect_count", eofCount)
	794	+	}
782	795
783	796		// Process events in an inner loop
784	797		listenSuccess := false

@@ -789,7 +802,24 @@ func ListenForContainerEvents(networkFilter string, callback func(string, string
789	802		case err := <-errs:
790	803		if err != nil && ctx.Err() == nil {
791	804		// Only log if this isn't due to context cancellation
792		-	log.Error("Error in container event stream", "error", err)
	805	+	if err.Error() == "EOF" {
	806	+	// Increment EOF counter
	807	+	eofCount++
	808	+
	809	+	// Log EOF errors less frequently to reduce spam
	810	+	now := time.Now()
	811	+	if eofCount == 1 \|\| eofCount%10 == 0 \|\| now.Sub(lastEOFLogTime) > time.Minute {
	812	+	log.Debug("Docker event stream closed with EOF - will reconnect",
	813	+	"count", eofCount,
	814	+	"next_reconnect_delay", eofReconnectDelay)
	815	+	lastEOFLogTime = now
	816	+	}
	817	+	} else {
	818	+	// Non-EOF errors are still important, log them as errors
	819	+	log.Error("Error in container event stream", "error", err)
	820	+	// Reset EOF counter for non-EOF errors
	821	+	eofCount = 0
	822	+	}
793	823
794	824		// Cancel the event context to clean up resources
795	825		eventCancel()

@@ -798,28 +828,26 @@ func ListenForContainerEvents(networkFilter string, callback func(string, string
798	828		consecutiveFailures++
799	829
800	830		// If we have too many consecutive failures, log a warning
801		-	if consecutiveFailures >= maxConsecutiveFailures {
	831	+	if consecutiveFailures >= maxConsecutiveFailures && consecutiveFailures%maxConsecutiveFailures == 0 {
802	832		log.Warn("Multiple consecutive failures in container event stream",
803	833		"count", consecutiveFailures,
804	834		"will_continue_trying", true)
805	835		}
806	836
807		-	// Calculate backoff delay with exponential increase
808		-	if listenSuccess {
809		-	// If we had successful events before failure, reset the delay
810		-	currentDelay = initialDelay
811		-	} else {
812		-	// Exponential backoff: increase the delay each time, up to max
813		-	currentDelay = time.Duration(float64(currentDelay) * 1.5)
814		-	if currentDelay > maxDelay {
815		-	currentDelay = maxDelay
	837	+	// Calculate backoff delay with exponential increase for non-EOF errors
	838	+	if err.Error() != "EOF" {
	839	+	if listenSuccess {
	840	+	// If we had successful events before failure, reset the delay
	841	+	currentDelay = initialDelay
	842	+	} else {
	843	+	// Exponential backoff: increase the delay each time, up to max
	844	+	currentDelay = time.Duration(float64(currentDelay) * 1.5)
	845	+	if currentDelay > maxDelay {
	846	+	currentDelay = maxDelay
	847	+	}
816	848		}
817	849		}
818	850
819		-	log.Debug("Will attempt to reconnect to container event stream",
820		-	"delay", currentDelay,
821		-	"consecutive_failures", consecutiveFailures)
822		-
823	851		// Break out of the inner event loop to reconnect
824	852		break eventStreamLoop
825	853		}

@@ -834,7 +862,10 @@ func ListenForContainerEvents(networkFilter string, callback func(string, string
834	862		listenSuccess = true
835	863		// Reset failure counter on successful event
836	864		consecutiveFailures = 0
837		-	log.Info("Successfully receiving container events")
	865	+	// Only log first successful connection or after errors
	866	+	if eofCount <= 1 {
	867	+	log.Info("Successfully receiving container events")
	868	+	}
838	869		}
839	870
840	871		// Update last event time

@@ -858,7 +889,7 @@ func ListenForContainerEvents(networkFilter string, callback func(string, string
858	889		switch event.Action {
859	890		case "start":
860	891		// Container started - need to get its IP and update any proxy routes
861		-	log.Debug("Container started event",
	892	+	log.Debug("container started event",
862	893		"container_id", containerID,
863	894		"container_name", containerName)
864	895

@@ -876,14 +907,14 @@ func ListenForContainerEvents(networkFilter string, callback func(string, string
876	907
877	908		case "die", "kill", "destroy":
878	909		// Container stopped - might need to mark routes as inactive
879		-	log.Debug("Container stopped event",
	910	+	log.Debug("container stopped event",
880	911		"container_id", containerID,
881	912		"container_name", containerName,
882	913		"action", event.Action)
883	914
884	915		case "rename":
885	916		// Container renamed - update related proxy routes
886		-	log.Debug("Container renamed event",
	917	+	log.Debug("container renamed event",
887	918		"container_id", containerID,
888	919		"container_name", containerName,
889	920		"new_name", event.Actor.Attributes["name"])

@@ -916,10 +947,23 @@ func ListenForContainerEvents(networkFilter string, callback func(string, string
916	947		// Clean up the event context
917	948		eventCancel()
918	949
	950	+	// Use different reconnection delays for different error types
	951	+	var reconnectDelay time.Duration
	952	+	if eofCount > 0 {
	953	+	// Use fixed delay for EOF errors to prevent rapid reconnection cycles
	954	+	reconnectDelay = eofReconnectDelay
	955	+	} else {
	956	+	// Use exponential backoff for other errors
	957	+	reconnectDelay = currentDelay
	958	+	}
	959	+
919	960		// Wait for the calculated delay before reconnecting
920	961		select {
921		-	case <-time.After(currentDelay):
922		-	log.Info("Attempting to reconnect to container event stream")
	962	+	case <-time.After(reconnectDelay):
	963	+	// Only log reconnection attempts for non-EOF errors or occasional EOF errors
	964	+	if eofCount == 0 \|\| eofCount%10 == 0 {
	965	+	log.Debug("Attempting to reconnect to container event stream", "delay", reconnectDelay)
	966	+	}
923	967		case <-ctx.Done():
924	968		return
925	969		}

 		consecutiveFailures := 0
 		maxConsecutiveFailures := 10
+		// Track EOFs for reduced logging
+		eofCount := 0
+		lastEOFLogTime := time.Now().Add(-10 * time.Minute) // Initialize to past time
+		// Fixed delay for EOFs to prevent rapid reconnection
+		eofReconnectDelay := 5 * time.Second
 		for {
 			// Check if context was canceled (application is shutting down)
 			if ctx.Err() != nil {
 			}
 			// Add a timeout context for the Events call
-			// This helps prevent long-running connections that might become stale
+			eventCtx, eventCancel := context.WithCancel(ctx)
-			eventCtx, eventCancel := context.WithTimeout(ctx, 5*time.Minute)
 			// Set up filters for the events we're interested in
 			filters := makeEventFilters(networkFilter)
-			// Start listening for events with timeout context
+			log.Debug("connecting to container event stream", "network_filter", networkFilter)
+			// Start listening for events
 			messages, errs := dockerCli.Events(eventCtx, events.ListOptions{
 				Filters: filters,
 			})
-			log.Info("Started container event listener", "network_filter", networkFilter)
+			// Only log the first connection and every 10th reconnection for EOF
+			if eofCount == 0 {
+				log.Info("Started container event listener", "network_filter", networkFilter)
+			} else {
+				log.Debug("Reconnected container event listener", "network_filter", networkFilter, "reconnect_count", eofCount)
+			}
 			// Process events in an inner loop
 			listenSuccess := false
 				case err := <-errs:
 					if err != nil && ctx.Err() == nil {
 						// Only log if this isn't due to context cancellation
-						log.Error("Error in container event stream", "error", err)
+						if err.Error() == "EOF" {
+							// Increment EOF counter
+							eofCount++
+							// Log EOF errors less frequently to reduce spam
+							now := time.Now()
+							if eofCount == 1 || eofCount%10 == 0 || now.Sub(lastEOFLogTime) > time.Minute {
+								log.Debug("Docker event stream closed with EOF - will reconnect",
+									"count", eofCount,
+									"next_reconnect_delay", eofReconnectDelay)
+								lastEOFLogTime = now
+							}
+						} else {
+							// Non-EOF errors are still important, log them as errors
+							log.Error("Error in container event stream", "error", err)
+							// Reset EOF counter for non-EOF errors
+							eofCount = 0
+						}
 						// Cancel the event context to clean up resources
 						eventCancel()
 						consecutiveFailures++
 						// If we have too many consecutive failures, log a warning
-						if consecutiveFailures >= maxConsecutiveFailures {
+						if consecutiveFailures >= maxConsecutiveFailures && consecutiveFailures%maxConsecutiveFailures == 0 {
 							log.Warn("Multiple consecutive failures in container event stream",
 								"count", consecutiveFailures,
 								"will_continue_trying", true)
 						}
-						// Calculate backoff delay with exponential increase
+						// Calculate backoff delay with exponential increase for non-EOF errors
-						if listenSuccess {
+						if err.Error() != "EOF" {
-							// If we had successful events before failure, reset the delay
+							if listenSuccess {
-							currentDelay = initialDelay
+								// If we had successful events before failure, reset the delay
-						} else {
+								currentDelay = initialDelay
-							// Exponential backoff: increase the delay each time, up to max
+							} else {
-							currentDelay = time.Duration(float64(currentDelay) * 1.5)
+								// Exponential backoff: increase the delay each time, up to max
-							if currentDelay > maxDelay {
+								currentDelay = time.Duration(float64(currentDelay) * 1.5)
-								currentDelay = maxDelay
+								if currentDelay > maxDelay {
+									currentDelay = maxDelay
+								}
 							}
 						}
-						log.Debug("Will attempt to reconnect to container event stream",
-							"delay", currentDelay,
-							"consecutive_failures", consecutiveFailures)
 						// Break out of the inner event loop to reconnect
 						break eventStreamLoop
 					}
 						listenSuccess = true
 						// Reset failure counter on successful event
 						consecutiveFailures = 0
-						log.Info("Successfully receiving container events")
+						// Only log first successful connection or after errors
+						if eofCount <= 1 {
+							log.Info("Successfully receiving container events")
+						}
 					}
 					// Update last event time
 						switch event.Action {
 						case "start":
 							// Container started - need to get its IP and update any proxy routes
-							log.Debug("Container started event",
+							log.Debug("container started event",
 								"container_id", containerID,
 								"container_name", containerName)
 						case "die", "kill", "destroy":
 							// Container stopped - might need to mark routes as inactive
-							log.Debug("Container stopped event",
+							log.Debug("container stopped event",
 								"container_id", containerID,
 								"container_name", containerName,
 								"action", event.Action)
 						case "rename":
 							// Container renamed - update related proxy routes
-							log.Debug("Container renamed event",
+							log.Debug("container renamed event",
 								"container_id", containerID,
 								"container_name", containerName,
 								"new_name", event.Actor.Attributes["name"])
 			// Clean up the event context
 			eventCancel()
+			// Use different reconnection delays for different error types
+			var reconnectDelay time.Duration
+			if eofCount > 0 {
+				// Use fixed delay for EOF errors to prevent rapid reconnection cycles
+				reconnectDelay = eofReconnectDelay
+			} else {
+				// Use exponential backoff for other errors
+				reconnectDelay = currentDelay
+			}
 			// Wait for the calculated delay before reconnecting
 			select {
-			case <-time.After(currentDelay):
+			case <-time.After(reconnectDelay):
-				log.Info("Attempting to reconnect to container event stream")
+				// Only log reconnection attempts for non-EOF errors or occasional EOF errors
+				if eofCount == 0 || eofCount%10 == 0 {
+					log.Debug("Attempting to reconnect to container event stream", "delay", reconnectDelay)
+				}
 			case <-ctx.Done():
 				return
 			}