lee/pentagi
1
0
Fork 0
mirror of https://github.com/vxcontrol/pentagi.git synced 2026-08-25 04:26:29 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(docker): update container error handling and logging

Browse Source 
- Replaced client.IsErrNotFound with cerrdefs.IsNotFound for consistent error handling across container operations.
- Updated probe image version to alpine:3.23.5.
- Enhanced logging for container removal failures in flow tools, ensuring better traceability of issues.
- Improved context usage in tests for better cancellation handling.
This commit is contained in:
Dmitry Ng
2026-08-04 10:54:58 +03:00
parent a5274d3c86
commit 4081292a25
3 changed files with 52 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/pkg/docker/client.go
+16 -7
View File
@@ -17,6 +17,7 @@ import (
"pentagi/pkg/config"
"pentagi/pkg/database"
cerrdefs "github.com/containerd/errdefs"
"github.com/docker/docker/api/types"
"github.com/docker/docker/api/types/container"
"github.com/docker/docker/api/types/filters"
@@ -453,7 +454,7 @@ func (dc *dockerClient) StopContainer(ctx context.Context, containerID string, d
stopErr := dc.client.ContainerStop(ctx, containerID, container.StopOptions{})
if stopErr != nil {
if client.IsErrNotFound(stopErr) {
if cerrdefs.IsNotFound(stopErr) {
logger.Warn("target container already removed or never existed")
} else {
return fmt.Errorf("container shutdown failed: %w", stopErr)
@@ -486,10 +487,11 @@ func (dc *dockerClient) RemoveContainer(ctx context.Context, containerID string,
Force: true,
}
if err := dc.client.ContainerRemove(ctx, containerID, options); err != nil {
if !client.IsErrNotFound(err) {
if !cerrdefs.IsNotFound(err) {
return fmt.Errorf("failed to remove container: %w", err)
}
// TODO: fix this case
// already gone (removed manually, or a prior call already succeeded);
// still mark it deleted below so the database row does not go stale.
logger.WithError(err).Warn("container not found")
}
@@ -600,16 +602,23 @@ func (dc *dockerClient) Cleanup(ctx context.Context) error {
func (dc *dockerClient) IsContainerRunning(ctx context.Context, containerID string) (bool, error) {
inspection, err := dc.client.ContainerInspect(ctx, containerID)
if err != nil {
if !client.IsErrNotFound(err) {
if !cerrdefs.IsNotFound(err) {
return false, fmt.Errorf("container inspection failed: %w", err)
}
// a removed container is missing, not an inspection failure
return false, nil
}
if inspection.State == nil {
// the daemon always populates State for a successfully inspected
// container; treat the unexpected absence as "not running" rather
// than panicking on the field access below.
return false, nil
}
// Check both Running state and health status if available
isOperational := inspection.State.Running
if inspection.State != nil && inspection.State.Health != nil && inspection.State.Health.Status != "" {
if inspection.State.Health != nil && inspection.State.Health.Status != "" {
isOperational = isOperational && inspection.State.Health.Status != "unhealthy"
}
@@ -974,7 +983,7 @@ func getHostDataDir(ctx context.Context, cli *client.Client, dataDir, workDir st
return "" // unexpected error
}
mounts := []types.MountPoint{}
mounts := []container.MountPoint{}
for _, container := range containers {
inspect, err := cli.ContainerInspect(ctx, container.ID)
if err != nil {
@@ -1001,7 +1010,7 @@ func getHostDataDir(ctx context.Context, cli *client.Client, dataDir, workDir st
}
// sort mounts by destination length to get the most accurate mount point
slices.SortFunc(mounts, func(a, b types.MountPoint) int {
slices.SortFunc(mounts, func(a, b container.MountPoint) int {
return len(b.Destination) - len(a.Destination)
})
backend/pkg/docker/client_test.go
+30 -34
View File
@@ -6,14 +6,18 @@ import (
"encoding/binary"
"errors"
"fmt"
"io"
"math/rand"
"strings"
"sync/atomic"
"testing"
"time"
cerrdefs "github.com/containerd/errdefs"
"github.com/docker/docker/api/types/container"
"github.com/docker/docker/client"
"github.com/sirupsen/logrus"
"github.com/stretchr/testify/require"
)
func TestStatContainerEntries_AllSucceed(t *testing.T) {
@@ -262,7 +266,7 @@ func failNames(failures []statFailure) map[string]bool {
return m
}
const probeImage = "alpine:3.20"
const probeImage = "alpine:3.23.5"
// newDaemonClient binds a client to the local daemon, skipping the test when
// none is reachable.
@@ -274,66 +278,58 @@ func newDaemonClient(t *testing.T) *dockerClient {
t.Skipf("docker daemon unavailable: %v", err)
}
ctx := context.Background()
ctx := t.Context()
cli.NegotiateAPIVersion(ctx)
if _, err := cli.Ping(ctx); err != nil {
t.Skipf("docker daemon unavailable: %v", err)
}
return &dockerClient{client: cli}
logger := logrus.New()
logger.SetOutput(io.Discard)
return &dockerClient{client: cli, logger: logger}
}
// A flow keeps the id of its primary container in the database. When that
// container is removed behind pentagi's back the id has to read as not running,
// otherwise the flow can never rebuild it.
func TestIsContainerRunningRemovedContainer(t *testing.T) {
dc := newDaemonClient(t)
ctx := context.Background()
ctx := t.Context()
created, err := dc.client.ContainerCreate(ctx, &container.Config{
Image: probeImage,
Entrypoint: []string{"tail", "-f", "/dev/null"},
}, nil, nil, nil, "")
if client.IsErrNotFound(err) {
if cerrdefs.IsNotFound(err) {
t.Skipf("%s is not present locally", probeImage)
}
if err != nil {
t.Fatalf("create probe container: %v", err)
}
require.NoError(t, err)
t.Cleanup(func() {
dc.client.ContainerRemove(context.Background(), created.ID, container.RemoveOptions{Force: true})
ctx := context.WithoutCancel(ctx)
// the happy path already removes the container below; only report a
// cleanup failure if it is still there for some other reason.
if err := dc.client.ContainerRemove(ctx, created.ID, container.RemoveOptions{Force: true}); err != nil && !cerrdefs.IsNotFound(err) {
t.Errorf("cleanup: failed to remove container %q: %v", created.ID, err)
}
})
if err := dc.client.ContainerStart(ctx, created.ID, container.StartOptions{}); err != nil {
t.Fatalf("start probe container: %v", err)
}
require.NoError(t, dc.client.ContainerStart(ctx, created.ID, container.StartOptions{}))
running, err := dc.IsContainerRunning(ctx, created.ID)
if err != nil || !running {
t.Fatalf("got running=%v err=%v, want true and no error", running, err)
}
require.NoError(t, err)
require.True(t, running)
if err := dc.client.ContainerRemove(ctx, created.ID, container.RemoveOptions{Force: true}); err != nil {
t.Fatalf("remove probe container: %v", err)
}
require.NoError(t, dc.client.ContainerRemove(ctx, created.ID, container.RemoveOptions{Force: true}))
running, err = dc.IsContainerRunning(ctx, created.ID)
if err != nil {
t.Fatalf("removed container: got error %v, want none", err)
}
if running {
t.Fatal("removed container reported as running")
}
require.NoError(t, err)
require.False(t, running)
}
func TestIsContainerRunningUnknownContainer(t *testing.T) {
dc := newDaemonClient(t)
running, err := dc.IsContainerRunning(context.Background(), "pentagi-container-that-does-not-exist")
if err != nil {
t.Fatalf("unknown container: got error %v, want none", err)
}
if running {
t.Fatal("unknown container reported as running")
}
running, err := dc.IsContainerRunning(t.Context(), "pentagi-container-that-does-not-exist")
require.NoError(t, err)
require.False(t, running)
}
backend/pkg/tools/tools.go
+6 -2
View File
@@ -481,9 +481,9 @@ func (fte *flowToolsExecutor) SetGraphitiClient(client *graphiti.Client) {
func (fte *flowToolsExecutor) Prepare(ctx context.Context) error {
if cnt, err := fte.db.GetFlowPrimaryContainer(ctx, fte.flowID); err == nil {
containerName := PrimaryTerminalName(fte.cfg.TenantPrefix(), fte.flowID)
// the stored status goes stale when the container is removed outside pentagi
if cnt.Status == database.ContainerStatusRunning {
containerName := PrimaryTerminalName(fte.cfg.TenantPrefix(), fte.flowID)
running, err := fte.docker.IsContainerRunning(ctx, cnt.LocalID.String)
if err != nil {
return fmt.Errorf("failed to inspect container '%s': %w", containerName, err)
@@ -498,7 +498,11 @@ func (fte *flowToolsExecutor) Prepare(ctx context.Context) error {
}
}
fte.docker.RemoveContainer(ctx, cnt.LocalID.String, cnt.ID)
if err := fte.docker.RemoveContainer(ctx, cnt.LocalID.String, cnt.ID); err != nil {
logrus.WithContext(ctx).WithError(err).WithFields(enrichLogrusFields(fte.flowID, nil, nil, logrus.Fields{
"container_name": containerName,
})).Warn("failed to remove stale primary container before rebuild")
}
}
// Explicit capability allow-list (CapDrop: ALL below): Docker's default 14