From 2fe076ea7ebaac58d3e731bcb6dcde84d3fe5719 Mon Sep 17 00:00:00 2001 From: ningmingxiao Date: Fri, 21 Mar 2025 18:31:06 +0800 Subject: [PATCH] cri:fix lost container exit events if they arrive before info is cached Signed-off-by: ningmingxiao (cherry picked from commit ead5c1ee6573c698b3776581f744a8e752e75770) Signed-off-by: ningmingxiao --- pkg/cri/sbserver/restart.go | 26 ++++++++++++++++---------- pkg/cri/server/restart.go | 26 ++++++++++++++++---------- 2 files changed, 32 insertions(+), 20 deletions(-) diff --git a/pkg/cri/sbserver/restart.go b/pkg/cri/sbserver/restart.go index f984e56779..7405ed25d2 100644 --- a/pkg/cri/sbserver/restart.go +++ b/pkg/cri/sbserver/restart.go @@ -150,7 +150,7 @@ func (c *criService) recover(ctx context.Context) error { for _, container := range containers { container := container eg.Go(func() error { - cntr, err := c.loadContainer(ctx2, container) + cntr, exitCh, pid, err := c.loadContainer(ctx2, container) if err != nil { log.G(ctx2).WithError(err).Errorf("Failed to load container %q", container.ID()) return nil @@ -159,6 +159,10 @@ func (c *criService) recover(ctx context.Context) error { if err := c.containerStore.Add(cntr); err != nil { return fmt.Errorf("failed to add container %q to store: %w", container.ID(), err) } + if exitCh != nil { + // Start the exit monitor. This should run after that container has been added to the container store. + c.eventMonitor.startContainerExitMonitor(context.Background(), cntr.ID, pid, exitCh) + } if err := c.containerNameIndex.Reserve(cntr.Name, cntr.ID); err != nil { return fmt.Errorf("failed to reserve container name %q: %w", cntr.Name, err) } @@ -228,7 +232,9 @@ func (c *criService) recover(ctx context.Context) error { const loadContainerTimeout = 10 * time.Second // loadContainer loads container from containerd and status checkpoint. -func (c *criService) loadContainer(ctx context.Context, cntr containerd.Container) (containerstore.Container, error) { +func (c *criService) loadContainer(ctx context.Context, cntr containerd.Container) (containerstore.Container, <-chan containerd.ExitStatus, uint32, error) { + var exitCh <-chan containerd.ExitStatus + var statusPid uint32 ctx, cancel := context.WithTimeout(ctx, loadContainerTimeout) defer cancel() id := cntr.ID() @@ -238,15 +244,15 @@ func (c *criService) loadContainer(ctx context.Context, cntr containerd.Containe // Load container metadata. exts, err := cntr.Extensions(ctx) if err != nil { - return container, fmt.Errorf("failed to get container extensions: %w", err) + return container, nil, 0, fmt.Errorf("failed to get container extensions: %w", err) } ext, ok := exts[containerMetadataExtension] if !ok { - return container, fmt.Errorf("metadata extension %q not found", containerMetadataExtension) + return container, nil, 0, fmt.Errorf("metadata extension %q not found", containerMetadataExtension) } data, err := typeurl.UnmarshalAny(ext) if err != nil { - return container, fmt.Errorf("failed to unmarshal metadata extension %q: %w", ext, err) + return container, nil, 0, fmt.Errorf("failed to unmarshal metadata extension %q: %w", ext, err) } meta := data.(*containerstore.Metadata) @@ -352,11 +358,13 @@ func (c *criService) loadContainer(ctx context.Context, cntr containerd.Containe // before status is checkpointed. status.StartedAt = time.Now().UnixNano() status.Pid = t.Pid() + statusPid = t.Pid() } // Wait for the task for exit monitor. // wait is a long running background request, no timeout needed. - exitCh, err := t.Wait(ctrdutil.NamespacedContext()) + exitCh, err = t.Wait(ctrdutil.NamespacedContext()) if err != nil { + exitCh = nil if !errdefs.IsNotFound(err) { return fmt.Errorf("failed to wait for task: %w", err) } @@ -365,9 +373,6 @@ func (c *criService) loadContainer(ctx context.Context, cntr containerd.Containe status.FinishedAt = time.Now().UnixNano() status.ExitCode = unknownExitCode status.Reason = unknownExitReason - } else { - // Start exit monitor. - c.eventMonitor.startContainerExitMonitor(context.Background(), id, status.Pid, exitCh) } case containerd.Stopped: // Task is stopped. Update status and delete the task. @@ -396,7 +401,8 @@ func (c *criService) loadContainer(ctx context.Context, cntr containerd.Containe if containerIO != nil { opts = append(opts, containerstore.WithContainerIO(containerIO)) } - return containerstore.NewContainer(*meta, opts...) + container, err = containerstore.NewContainer(*meta, opts...) + return container, exitCh, statusPid, err } // podSandboxRecover is an additional interface implemented by podsandbox/ controller to handle diff --git a/pkg/cri/server/restart.go b/pkg/cri/server/restart.go index 76471471be..67ab1561fd 100644 --- a/pkg/cri/server/restart.go +++ b/pkg/cri/server/restart.go @@ -90,7 +90,7 @@ func (c *criService) recover(ctx context.Context) error { for _, container := range containers { container := container eg.Go(func() error { - cntr, err := c.loadContainer(ctx2, container) + cntr, exitCh, pid, err := c.loadContainer(ctx2, container) if err != nil { log.G(ctx2).WithError(err).Errorf("Failed to load container %q", container.ID()) return nil @@ -99,6 +99,10 @@ func (c *criService) recover(ctx context.Context) error { if err := c.containerStore.Add(cntr); err != nil { return fmt.Errorf("failed to add container %q to store: %w", container.ID(), err) } + if exitCh != nil { + // Start the exit monitor. This should run after that container has been added to the container store. + c.eventMonitor.startContainerExitMonitor(context.Background(), cntr.ID, pid, exitCh) + } if err := c.containerNameIndex.Reserve(cntr.Name, cntr.ID); err != nil { return fmt.Errorf("failed to reserve container name %q: %w", cntr.Name, err) } @@ -168,7 +172,9 @@ func (c *criService) recover(ctx context.Context) error { const loadContainerTimeout = 10 * time.Second // loadContainer loads container from containerd and status checkpoint. -func (c *criService) loadContainer(ctx context.Context, cntr containerd.Container) (containerstore.Container, error) { +func (c *criService) loadContainer(ctx context.Context, cntr containerd.Container) (containerstore.Container, <-chan containerd.ExitStatus, uint32, error) { + var exitCh <-chan containerd.ExitStatus + var statusPid uint32 ctx, cancel := context.WithTimeout(ctx, loadContainerTimeout) defer cancel() id := cntr.ID() @@ -178,15 +184,15 @@ func (c *criService) loadContainer(ctx context.Context, cntr containerd.Containe // Load container metadata. exts, err := cntr.Extensions(ctx) if err != nil { - return container, fmt.Errorf("failed to get container extensions: %w", err) + return container, nil, 0, fmt.Errorf("failed to get container extensions: %w", err) } ext, ok := exts[containerMetadataExtension] if !ok { - return container, fmt.Errorf("metadata extension %q not found", containerMetadataExtension) + return container, nil, 0, fmt.Errorf("metadata extension %q not found", containerMetadataExtension) } data, err := typeurl.UnmarshalAny(ext) if err != nil { - return container, fmt.Errorf("failed to unmarshal metadata extension %q: %w", ext, err) + return container, nil, 0, fmt.Errorf("failed to unmarshal metadata extension %q: %w", ext, err) } meta := data.(*containerstore.Metadata) @@ -292,11 +298,13 @@ func (c *criService) loadContainer(ctx context.Context, cntr containerd.Containe // before status is checkpointed. status.StartedAt = time.Now().UnixNano() status.Pid = t.Pid() + statusPid = t.Pid() } // Wait for the task for exit monitor. // wait is a long running background request, no timeout needed. - exitCh, err := t.Wait(ctrdutil.NamespacedContext()) + exitCh, err = t.Wait(ctrdutil.NamespacedContext()) if err != nil { + exitCh = nil if !errdefs.IsNotFound(err) { return fmt.Errorf("failed to wait for task: %w", err) } @@ -305,9 +313,6 @@ func (c *criService) loadContainer(ctx context.Context, cntr containerd.Containe status.FinishedAt = time.Now().UnixNano() status.ExitCode = unknownExitCode status.Reason = unknownExitReason - } else { - // Start exit monitor. - c.eventMonitor.startContainerExitMonitor(context.Background(), id, status.Pid, exitCh) } case containerd.Stopped: // Task is stopped. Update status and delete the task. @@ -336,7 +341,8 @@ func (c *criService) loadContainer(ctx context.Context, cntr containerd.Containe if containerIO != nil { opts = append(opts, containerstore.WithContainerIO(containerIO)) } - return containerstore.NewContainer(*meta, opts...) + container, err = containerstore.NewContainer(*meta, opts...) + return container, exitCh, statusPid, err } // loadSandbox loads sandbox from containerd.