fix: Docker container event stream and monitoring

- Add timeout context for Docker event streams to prevent stale connections
- Implement max consecutive failures tracking with warning logs
- Add periodic event stream health checks and automatic reconnection
- Enhance error handling and backoff strategy for event stream failures
- Improve logging for connection status and reconnection attempts

commit: 9fac95c17204682c401d35b29eddd7801c02c77f
author: bamen · 19mo ago · 2025-03-10 16:10 UTC
parent: 086e0df
signature: unsigned

1 file · +43 -3

M pkg/docker/containers.go +43 -3


@@ -758,6 +758,7 @@ func ListenForContainerEvents(networkFilter string, callback func(string, string
758	758		maxDelay := 1 * time.Minute
759	759		currentDelay := initialDelay
760	760		consecutiveFailures := 0
	761	+	maxConsecutiveFailures := 10
761	762
762	763		for {
763	764		// Check if context was canceled (application is shutting down)

@@ -765,11 +766,15 @@ func ListenForContainerEvents(networkFilter string, callback func(string, string
765	766		return
766	767		}
767	768
	769	+	// 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)
	772	+
768	773		// Set up filters for the events we're interested in
769	774		filters := makeEventFilters(networkFilter)
770	775
771		-	// Start listening for events
772		-	messages, errs := dockerCli.Events(ctx, events.ListOptions{
	776	+	// Start listening for events with timeout context
	777	+	messages, errs := dockerCli.Events(eventCtx, events.ListOptions{
773	778		Filters: filters,
774	779		})
775	780

@@ -777,6 +782,7 @@ func ListenForContainerEvents(networkFilter string, callback func(string, string
777	782
778	783		// Process events in an inner loop
779	784		listenSuccess := false
	785	+	lastEventTime := time.Now()
780	786		eventStreamLoop:
781	787		for {
782	788		select {

@@ -784,15 +790,26 @@ func ListenForContainerEvents(networkFilter string, callback func(string, string
784	790		if err != nil && ctx.Err() == nil {
785	791		// Only log if this isn't due to context cancellation
786	792		log.Error("Error in container event stream", "error", err)
	793	+
	794	+	// Cancel the event context to clean up resources
	795	+	eventCancel()
	796	+
787	797		// Increment failure counter for backoff calculation
788	798		consecutiveFailures++
789	799
	800	+	// If we have too many consecutive failures, log a warning
	801	+	if consecutiveFailures >= maxConsecutiveFailures {
	802	+	log.Warn("Multiple consecutive failures in container event stream",
	803	+	"count", consecutiveFailures,
	804	+	"will_continue_trying", true)
	805	+	}
	806	+
790	807		// Calculate backoff delay with exponential increase
791	808		if listenSuccess {
792	809		// If we had successful events before failure, reset the delay
793	810		currentDelay = initialDelay
794	811		} else {
795		-	// Exponential backoff: double the delay each time, up to max
	812	+	// Exponential backoff: increase the delay each time, up to max
796	813		currentDelay = time.Duration(float64(currentDelay) * 1.5)
797	814		if currentDelay > maxDelay {
798	815		currentDelay = maxDelay

@@ -808,6 +825,7 @@ func ListenForContainerEvents(networkFilter string, callback func(string, string
808	825		}
809	826		// Context was canceled, exit completely
810	827		if ctx.Err() != nil {
	828	+	eventCancel()
811	829		return
812	830		}
813	831		case event := <-messages:

@@ -819,6 +837,9 @@ func ListenForContainerEvents(networkFilter string, callback func(string, string
819	837		log.Info("Successfully receiving container events")
820	838		}
821	839
	840	+	// Update last event time
	841	+	lastEventTime = time.Now()
	842	+
822	843		// Process container events
823	844		if event.Type == "container" {
824	845		containerID := event.Actor.ID

@@ -870,12 +891,31 @@ func ListenForContainerEvents(networkFilter string, callback func(string, string
870	891		// Add other event types as needed
871	892		}
872	893		}
	894	+	case <-time.After(30 * time.Second):
	895	+	// If we haven't received events for a while, check if the connection is still alive
	896	+	timeSinceLastEvent := time.Since(lastEventTime)
	897	+
	898	+	// If it's been more than 2 minutes since the last event, reconnect
	899	+	if timeSinceLastEvent > 2*time.Minute {
	900	+	log.Debug("No events received for too long, reconnecting",
	901	+	"time_since_last_event", timeSinceLastEvent)
	902	+
	903	+	// Cancel the event context to clean up resources
	904	+	eventCancel()
	905	+
	906	+	// Break out of the inner event loop to reconnect
	907	+	break eventStreamLoop
	908	+	}
873	909		case <-ctx.Done():
	910	+	eventCancel()
874	911		return
875	912		}
876	913		}
877	914
878	915		// If we get here, the event stream failed and we need to reconnect
	916	+	// Clean up the event context
	917	+	eventCancel()
	918	+
879	919		// Wait for the calculated delay before reconnecting
880	920		select {
881	921		case <-time.After(currentDelay):

 		maxDelay := 1 * time.Minute
 		currentDelay := initialDelay
 		consecutiveFailures := 0
+		maxConsecutiveFailures := 10
 		for {
 			// Check if context was canceled (application is shutting down)
 				return
 			}
+			// Add a timeout context for the Events call
+			// This helps prevent long-running connections that might become stale
+			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
+			// Start listening for events with timeout context
-			messages, errs := dockerCli.Events(ctx, events.ListOptions{
+			messages, errs := dockerCli.Events(eventCtx, events.ListOptions{
 				Filters: filters,
 			})
 			// Process events in an inner loop
 			listenSuccess := false
+			lastEventTime := time.Now()
 		eventStreamLoop:
 			for {
 				select {
 					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)
+						// Cancel the event context to clean up resources
+						eventCancel()
 						// Increment failure counter for backoff calculation
 						consecutiveFailures++
+						// If we have too many consecutive failures, log a warning
+						if consecutiveFailures >= maxConsecutiveFailures {
+							log.Warn("Multiple consecutive failures in container event stream",
+								"count", consecutiveFailures,
+								"will_continue_trying", true)
+						}
 						// Calculate backoff delay with exponential increase
 						if listenSuccess {
 							// If we had successful events before failure, reset the delay
 							currentDelay = initialDelay
 						} else {
-							// Exponential backoff: double the delay each time, up to max
+							// Exponential backoff: increase the delay each time, up to max
 							currentDelay = time.Duration(float64(currentDelay) * 1.5)
 							if currentDelay > maxDelay {
 								currentDelay = maxDelay
 					}
 					// Context was canceled, exit completely
 					if ctx.Err() != nil {
+						eventCancel()
 						return
 					}
 				case event := <-messages:
 						log.Info("Successfully receiving container events")
 					}
+					// Update last event time
+					lastEventTime = time.Now()
 					// Process container events
 					if event.Type == "container" {
 						containerID := event.Actor.ID
 							// Add other event types as needed
 						}
 					}
+				case <-time.After(30 * time.Second):
+					// If we haven't received events for a while, check if the connection is still alive
+					timeSinceLastEvent := time.Since(lastEventTime)
+					// If it's been more than 2 minutes since the last event, reconnect
+					if timeSinceLastEvent > 2*time.Minute {
+						log.Debug("No events received for too long, reconnecting",
+							"time_since_last_event", timeSinceLastEvent)
+						// Cancel the event context to clean up resources
+						eventCancel()
+						// Break out of the inner event loop to reconnect
+						break eventStreamLoop
+					}
 				case <-ctx.Done():
+					eventCancel()
 					return
 				}
 			}
 			// If we get here, the event stream failed and we need to reconnect
+			// Clean up the event context
+			eventCancel()
 			// Wait for the calculated delay before reconnecting
 			select {
 			case <-time.After(currentDelay):