Merge branch 'master' into mh/discovery

Author: mjh1

CHANGELOG_PENDING.md

@@ -16,4 +16,9 @@
 
 ### Bug Fixes 🐞
 
+#### General
+
+* [#3779](https://github.com/livepeer/go-livepeer/pull/3779) worker: Fix orphaned containers on node shutdown (@victorges)
+* [#3777](https://github.com/livepeer/go-livepeer/pull/3777) docker: Forcefully SIGKILL runners after timeout (@pwilczynskiclearcode)
+
 #### CLI

ai/worker/docker.go

@@ -34,6 +34,7 @@ const containerPort = "8000/tcp"
 const pollingInterval = 500 * time.Millisecond
 const externalContainerTimeout = 2 * time.Minute
 const optFlagsContainerTimeout = 5 * time.Minute
+const containerStopTimeout = 8 * time.Second
 const containerRemoveTimeout = 30 * time.Second
 const containerCreatorLabel = "creator"
 const containerCreator = "ai-worker"
@@ -411,12 +412,9 @@ func (m *DockerManager) createContainer(ctx context.Context, pipeline string, mo
 	}
 
 	restartPolicy := container.RestartPolicy{
-		Name:              "on-failure",
+		Name:              container.RestartPolicyOnFailure,
 		MaximumRetryCount: 3,
 	}
-	if keepWarm {
-		restartPolicy = container.RestartPolicy{Name: "always"}
-	}
 
 	hostConfig := &container.HostConfig{
 		Resources: container.Resources{
@@ -736,7 +734,8 @@ func dockerRemoveContainer(client DockerClient, containerID string) error {
 	ctx, cancel := context.WithTimeout(context.Background(), containerRemoveTimeout)
 	defer cancel()
 
-	err := client.ContainerStop(ctx, containerID, container.StopOptions{})
+	timeoutSec := int(containerStopTimeout.Seconds())
+	err := client.ContainerStop(ctx, containerID, container.StopOptions{Timeout: &timeoutSec})
 	// Ignore "not found" or "already stopped" errors
 	if err != nil && !docker.IsErrNotFound(err) && !errdefs.IsNotModified(err) {
 		return err
@@ -780,9 +779,26 @@ tickerLoop:
 				return err
 			}
 
+			// If the container is running, we're done.
 			if json.State.Running {
 				break tickerLoop
 			}
+
+			// Fail fast on states that won't become running after startup.
+			if json.State != nil {
+				status := strings.ToLower(json.State.Status)
+				// Consider exited/dead as terminal. "removing" will surface via
+				// inspect error or transition to exited/dead shortly.
+				if status == "exited" || status == "dead" {
+					return fmt.Errorf("container entered terminal state before running: %s (exitCode=%d)", json.State.Status, json.State.ExitCode)
+				}
+				if !json.State.Restarting && json.State.ExitCode != 0 {
+					return fmt.Errorf("container exited before running (status=%s, exitCode=%d)", json.State.Status, json.State.ExitCode)
+				}
+				if !json.State.Restarting && json.State.Error != "" {
+					return fmt.Errorf("container error before running: %s", json.State.Error)
+				}
+			}
 		}
 	}
 

ai/worker/docker_test.go

@@ -20,6 +20,9 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+var stopTimeout = 8
+var expectedContainerStopOptions = container.StopOptions{Timeout: &stopTimeout}
+
 type MockDockerClient struct {
 	mock.Mock
 }
@@ -136,15 +139,15 @@ func TestNewDockerManager(t *testing.T) {
 			{ID: "container2", Names: []string{"/container2"}, Labels: map[string]string{containerCreatorLabel: containerCreator}},
 		}
 		mockDockerClient.On("ContainerList", mock.Anything, mock.Anything).Return(existingContainers, nil)
-		mockDockerClient.On("ContainerStop", mock.Anything, "container1", mock.Anything).Return(nil)
-		mockDockerClient.On("ContainerStop", mock.Anything, "container2", mock.Anything).Return(nil)
+		mockDockerClient.On("ContainerStop", mock.Anything, "container1", expectedContainerStopOptions).Return(nil)
+		mockDockerClient.On("ContainerStop", mock.Anything, "container2", expectedContainerStopOptions).Return(nil)
 		mockDockerClient.On("ContainerRemove", mock.Anything, "container1", mock.Anything).Return(nil)
 		mockDockerClient.On("ContainerRemove", mock.Anything, "container2", mock.Anything).Return(nil)
 
 		// Verify that existing containers were stopped and removed.
 		createAndVerifyManager()
-		mockDockerClient.AssertCalled(t, "ContainerStop", mock.Anything, "container1", mock.Anything)
-		mockDockerClient.AssertCalled(t, "ContainerStop", mock.Anything, "container2", mock.Anything)
+		mockDockerClient.AssertCalled(t, "ContainerStop", mock.Anything, "container1", expectedContainerStopOptions)
+		mockDockerClient.AssertCalled(t, "ContainerStop", mock.Anything, "container2", expectedContainerStopOptions)
 		mockDockerClient.AssertCalled(t, "ContainerRemove", mock.Anything, "container1", mock.Anything)
 		mockDockerClient.AssertCalled(t, "ContainerRemove", mock.Anything, "container2", mock.Anything)
 		mockDockerClient.AssertExpectations(t)
@@ -236,7 +239,7 @@ func TestDockerManager_Stop(t *testing.T) {
 		},
 	}
 
-	MockDockerClient.On("ContainerStop", mock.Anything, containerID, container.StopOptions{Timeout: nil}).Return(nil)
+	MockDockerClient.On("ContainerStop", mock.Anything, containerID, expectedContainerStopOptions).Return(nil)
 	MockDockerClient.On("ContainerRemove", mock.Anything, containerID, container.RemoveOptions{}).Return(nil)
 	err := dockerManager.Stop(ctx)
 	require.NoError(t, err)
@@ -650,7 +653,7 @@ func TestDockerManager_allocGPU(t *testing.T) {
 				dockerManager.gpuContainers[rc.GPU] = rc
 				dockerManager.containers[rc.Name] = rc
 				// Mock client methods to simulate the removal of the warm container.
-				mockDockerClient.On("ContainerStop", mock.Anything, "container1", container.StopOptions{}).Return(nil)
+				mockDockerClient.On("ContainerStop", mock.Anything, "container1", expectedContainerStopOptions).Return(nil)
 				mockDockerClient.On("ContainerRemove", mock.Anything, "container1", container.RemoveOptions{}).Return(nil)
 			},
 			expectedAllocatedGPU: "gpu0",
@@ -694,7 +697,7 @@ func TestDockerManager_destroyContainer(t *testing.T) {
 	dockerManager.gpuContainers[gpu] = rc
 	dockerManager.containers[containerID] = rc
 
-	mockDockerClient.On("ContainerStop", mock.Anything, containerID, container.StopOptions{}).Return(nil)
+	mockDockerClient.On("ContainerStop", mock.Anything, containerID, expectedContainerStopOptions).Return(nil)
 	mockDockerClient.On("ContainerRemove", mock.Anything, containerID, container.RemoveOptions{}).Return(nil)
 
 	err := dockerManager.destroyContainer(rc, true)
@@ -844,7 +847,7 @@ func TestDockerManager_watchContainer(t *testing.T) {
 			tt.mockServerSetup(mockServer)
 
 			// Mock destroyContainer to verify it is called.
-			mockDockerClient.On("ContainerStop", mock.Anything, rc.Name, mock.Anything).Return(nil).Once()
+			mockDockerClient.On("ContainerStop", mock.Anything, rc.Name, expectedContainerStopOptions).Return(nil).Once()
 			mockDockerClient.On("ContainerRemove", mock.Anything, rc.Name, mock.Anything).Return(nil).Once()
 
 			done := make(chan struct{})
@@ -906,7 +909,7 @@ func TestDockerManager_watchContainer(t *testing.T) {
 		mockDockerClient.AssertNotCalled(t, "ContainerRemove", mock.Anything, rc.Name, mock.Anything)
 
 		// Mock destroyContainer to verify it is called.
-		mockDockerClient.On("ContainerStop", mock.Anything, rc.Name, mock.Anything).Return(nil).Once()
+		mockDockerClient.On("ContainerStop", mock.Anything, rc.Name, expectedContainerStopOptions).Return(nil).Once()
 		mockDockerClient.On("ContainerRemove", mock.Anything, rc.Name, mock.Anything).Return(nil).Once()
 
 		// after the first failure, there should only 1 more healthcheck for the container to be stopped
@@ -993,8 +996,8 @@ func TestRemoveExistingContainers(t *testing.T) {
 		{ID: "container2", Names: []string{"/container2"}, Labels: map[string]string{containerCreatorLabel: containerCreator}},
 	}
 	mockDockerClient.On("ContainerList", mock.Anything, mock.Anything).Return(existingContainers, nil)
-	mockDockerClient.On("ContainerStop", mock.Anything, "container1", mock.Anything).Return(nil)
-	mockDockerClient.On("ContainerStop", mock.Anything, "container2", mock.Anything).Return(nil)
+	mockDockerClient.On("ContainerStop", mock.Anything, "container1", expectedContainerStopOptions).Return(nil)
+	mockDockerClient.On("ContainerStop", mock.Anything, "container2", expectedContainerStopOptions).Return(nil)
 	mockDockerClient.On("ContainerRemove", mock.Anything, "container1", mock.Anything).Return(nil)
 	mockDockerClient.On("ContainerRemove", mock.Anything, "container2", mock.Anything).Return(nil)
 
@@ -1036,9 +1039,9 @@ func TestRemoveExistingContainers_InMemoryFilterLegacyAndOwnerID(t *testing.T) {
 			{ID: "mine-1", Names: []string{"/mine-1"}, Labels: map[string]string{containerCreatorLabel: containerCreator, containerCreatorIDLabel: "owner-A"}},   // match -> remove
 		}, nil).
 		Once()
-	mockDockerClient.On("ContainerStop", mock.Anything, "legacy-1", mock.Anything).Return(nil).Once()
+	mockDockerClient.On("ContainerStop", mock.Anything, "legacy-1", expectedContainerStopOptions).Return(nil).Once()
 	mockDockerClient.On("ContainerRemove", mock.Anything, "legacy-1", mock.Anything).Return(nil).Once()
-	mockDockerClient.On("ContainerStop", mock.Anything, "mine-1", mock.Anything).Return(nil).Once()
+	mockDockerClient.On("ContainerStop", mock.Anything, "mine-1", expectedContainerStopOptions).Return(nil).Once()
 	mockDockerClient.On("ContainerRemove", mock.Anything, "mine-1", mock.Anything).Return(nil).Once()
 
 	removed, err := RemoveExistingContainers(ctx, mockDockerClient, "owner-A")
@@ -1095,7 +1098,7 @@ func TestDockerContainerName(t *testing.T) {
 func TestDockerRemoveContainer(t *testing.T) {
 	mockDockerClient := new(MockDockerClient)
 
-	mockDockerClient.On("ContainerStop", mock.Anything, "container1", container.StopOptions{}).Return(nil)
+	mockDockerClient.On("ContainerStop", mock.Anything, "container1", expectedContainerStopOptions).Return(nil)
 	mockDockerClient.On("ContainerRemove", mock.Anything, "container1", container.RemoveOptions{}).Return(nil)
 
 	err := dockerRemoveContainer(mockDockerClient, "container1")
@@ -1165,6 +1168,63 @@ func TestDockerWaitUntilRunning(t *testing.T) {
 		require.Contains(t, err.Error(), "timed out waiting for managed container")
 		mockDockerClient.AssertExpectations(t)
 	})
+
+	t.Run("FailFastOnExited", func(t *testing.T) {
+		// If the container is immediately exited, we should fail fast instead of waiting.
+		mockDockerClient := new(MockDockerClient)
+		// Always return non-running, exited state
+		mockDockerClient.On("ContainerInspect", mock.Anything, containerID).Return(types.ContainerJSON{
+			ContainerJSONBase: &types.ContainerJSONBase{
+				State: &types.ContainerState{
+					Status:   "exited",
+					Running:  false,
+					ExitCode: 137,
+				},
+			},
+		}, nil)
+
+		err := dockerWaitUntilRunning(ctx, mockDockerClient, containerID, pollingInterval)
+		require.Error(t, err)
+		require.Contains(t, err.Error(), "terminal state")
+		mockDockerClient.AssertExpectations(t)
+	})
+
+	t.Run("FailFastOnDead", func(t *testing.T) {
+		mockDockerClient := new(MockDockerClient)
+		mockDockerClient.On("ContainerInspect", mock.Anything, containerID).Return(types.ContainerJSON{
+			ContainerJSONBase: &types.ContainerJSONBase{
+				State: &types.ContainerState{
+					Status:  "dead",
+					Running: false,
+					Error:   "killed",
+				},
+			},
+		}, nil)
+
+		err := dockerWaitUntilRunning(ctx, mockDockerClient, containerID, pollingInterval)
+		require.Error(t, err)
+		require.Contains(t, err.Error(), "container entered terminal state")
+		mockDockerClient.AssertExpectations(t)
+	})
+
+	t.Run("FailFastOnExitCodeNonZeroWithoutRestarting", func(t *testing.T) {
+		mockDockerClient := new(MockDockerClient)
+		mockDockerClient.On("ContainerInspect", mock.Anything, containerID).Return(types.ContainerJSON{
+			ContainerJSONBase: &types.ContainerJSONBase{
+				State: &types.ContainerState{
+					Status:     "created",
+					Running:    false,
+					Restarting: false,
+					ExitCode:   1,
+				},
+			},
+		}, nil)
+
+		err := dockerWaitUntilRunning(ctx, mockDockerClient, containerID, pollingInterval)
+		require.Error(t, err)
+		require.Contains(t, err.Error(), "exited before running")
+		mockDockerClient.AssertExpectations(t)
+	})
 }
 
 func TestHwGPU(t *testing.T) {

box/box.sh

@@ -1,8 +1,25 @@
 #!/bin/bash
 set -e
 
+DOCKER=${DOCKER:-false}
 FRONTEND=${FRONTEND:-false}
 
+if [ "$FRONTEND" = "true" && "$DOCKER" = "false" ]; then
+  echo "Running the box with FRONTEND=true requires DOCKER=true"
+  exit 1
+fi
+
+# Ensure backgrounded services are stopped gracefully when this script is interrupted or exits.
+cleanup() {
+  if [ "$DOCKER" = "true" ]; then
+    echo "Stopping dockerized services..."
+    docker stop orchestrator --time 15 >/dev/null 2>&1 || true
+    docker stop gateway --time 15 >/dev/null 2>&1 || true
+    docker stop mediamtx --time 15 >/dev/null 2>&1 || true
+  fi
+}
+trap cleanup INT TERM HUP EXIT
+
 # Start multiple processes and output their logs to the console
 gateway() {
   echo "Starting Gateway..."
@@ -34,9 +51,9 @@ gateway &
 orchestrator &
 mediamtx &
 
-if [ "$DOCKER" = "true" ]; then
+if [ "$FRONTEND" = "true" ]; then
   supabase &
-  mediamtx &
+  frontend &
 fi
 
 # Wait for all background processes to finish

cmd/livepeer/livepeer.go

@@ -10,6 +10,7 @@ import (
 	"os"
 	"os/signal"
 	"runtime"
+	"syscall"
 	"time"
 
 	"github.com/livepeer/go-livepeer/cmd/livepeer/starter"
@@ -20,6 +21,8 @@ import (
 	"github.com/livepeer/go-livepeer/core"
 )
 
+const shutdownTimeout = 10 * time.Second
+
 func main() {
 	// Override the default flag set since there are dependencies that
 	// incorrectly add their own flags (specifically, due to the 'testing'
@@ -76,17 +79,25 @@ func main() {
 	lc := make(chan struct{})
 
 	go func() {
+		defer close(lc)
 		starter.StartLivepeer(ctx, cfg)
-		lc <- struct{}{}
 	}()
 
-	c := make(chan os.Signal)
-	signal.Notify(c, os.Interrupt)
+	c := make(chan os.Signal, 1)
+	signal.Notify(c, os.Interrupt, syscall.SIGTERM)
+
 	select {
 	case sig := <-c:
 		glog.Infof("Exiting Livepeer: %v", sig)
 		cancel()
-		time.Sleep(time.Second * 2) //Give time for other processes to shut down completely
 	case <-lc:
+		// fallthrough to normal shutdown below
+	}
+	select {
+	case <-lc:
+		glog.Infof("Graceful shutdown complete")
+	case <-time.After(shutdownTimeout):
+		glog.Infof("Shutdown timed out, forcing exit")
+		os.Exit(1)
 	}
 }