feat(crr): implement TCPSocket preStop hook in ContainerRecreateRequest

apis/apps/v1alpha1/containerrecreaterequest_types.go

@@ -75,7 +75,6 @@ type ContainerRecreateRequestContainer struct {
 }
 
 // ProbeHandler defines a specific action that should be taken
-// TODO(FillZpp): improve the definition when openkruise/kruise updates to k8s 1.23
 type ProbeHandler struct {
 	// One and only one of the following should be specified.
 	// Exec specifies the action to take.
@@ -84,9 +83,10 @@ type ProbeHandler struct {
 	// HTTPGet specifies the http request to perform.
 	// +optional
 	HTTPGet *v1.HTTPGetAction `json:"httpGet,omitempty" protobuf:"bytes,2,opt,name=httpGet"`
-	// TCPSocket specifies an action involving a TCP port.
-	// TCP hooks not yet supported
-	// TODO: implement a realistic TCP lifecycle hook
+	// TCPSocket specifies a TCP port to dial as a pre-stop hook.
+	// The kruise daemon dials the port; a successful connection (which is then
+	// immediately closed) counts as success. This mirrors how Kubernetes handles
+	// TCPSocket liveness/readiness probes.
 	// +optional
 	TCPSocket *v1.TCPSocketAction `json:"tcpSocket,omitempty" protobuf:"bytes,3,opt,name=tcpSocket"`
 }

pkg/daemon/containerrecreate/crr_daemon_controller.go

@@ -311,7 +311,16 @@ func (c *Controller) manage(crr *appsv1beta1.ContainerRecreateRequest) error {
 		return c.completeCRRStatus(crr, fmt.Sprintf("failed to find runtime service: %v", err))
 	}
 
-	pod := convertCRRToPod(crr)
+	// Fetch the real Pod to obtain its IP for TCPSocket preStop hook resolution.
+	// convertCRRToPod builds a minimal fake Pod; without Status.PodIP the TCP
+	// dialer would fall back to localhost, which is incorrect when the hook
+	// should reach the container's own IP.
+	realPod := &v1.Pod{}
+	if err := c.runtimeClient.Get(context.TODO(), types.NamespacedName{Namespace: crr.Namespace, Name: crr.Spec.PodName}, realPod); err != nil {
+		return fmt.Errorf("failed to get Pod %s/%s: %v", crr.Namespace, crr.Spec.PodName, err)
+	}
+
+	pod := convertCRRToPod(crr, realPod.Status.PodIP)
 
 	podStatus, err := runtimeManager.GetPodStatus(context.TODO(), pod.UID, pod.Name, pod.Namespace)
 	if err != nil {

pkg/daemon/containerrecreate/crr_daemon_util.go

@@ -165,7 +165,11 @@ func getCRRSyncContainerStatuses(crr *appsv1beta1.ContainerRecreateRequest) map[
 	return statuses
 }
 
-func convertCRRToPod(crr *appsv1beta1.ContainerRecreateRequest) *v1.Pod {
+// convertCRRToPod builds a minimal fake v1.Pod from the CRR spec that is
+// sufficient for the runtime manager to execute preStop hooks and kill containers.
+// podIP should be the real pod's Status.PodIP so that TCPSocket preStop hooks
+// can dial the correct address when no explicit Host is configured.
+func convertCRRToPod(crr *appsv1beta1.ContainerRecreateRequest, podIP string) *v1.Pod {
 	podName := crr.Spec.PodName
 	podUID := types.UID(crr.Labels[appsv1beta1.ContainerRecreateRequestPodUIDKey])
 
@@ -175,7 +179,8 @@ func convertCRRToPod(crr *appsv1beta1.ContainerRecreateRequest) *v1.Pod {
 			Name:      podName,
 			UID:       podUID,
 		},
-		Spec: v1.PodSpec{},
+		Spec:   v1.PodSpec{},
+		Status: v1.PodStatus{PodIP: podIP},
 	}
 
 	if crr.Spec.Strategy != nil {

pkg/daemon/containerrecreate/crr_daemon_util_test.go

@@ -0,0 +1,152 @@
+/*
+Copyright 2025 The Kruise Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package containerrecreate
+
+import (
+	"testing"
+
+	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/util/intstr"
+	"k8s.io/utils/ptr"
+
+	appsv1beta1 "github.com/openkruise/kruise/apis/apps/v1beta1"
+)
+
+func makeCRR(podName, podUID string, containers []appsv1beta1.ContainerRecreateRequestContainer, gracePeriod *int64) *appsv1beta1.ContainerRecreateRequest {
+	crr := &appsv1beta1.ContainerRecreateRequest{}
+	crr.Namespace = "default"
+	crr.Name = "test-crr"
+	crr.Labels = map[string]string{
+		appsv1beta1.ContainerRecreateRequestPodUIDKey: podUID,
+	}
+	crr.Spec.PodName = podName
+	crr.Spec.Containers = containers
+	if gracePeriod != nil {
+		crr.Spec.Strategy = &appsv1beta1.ContainerRecreateRequestStrategy{
+			TerminationGracePeriodSeconds: gracePeriod,
+		}
+	}
+	return crr
+}
+
+// TestConvertCRRToPod_PodIPIsSet verifies that the pod IP passed to
+// convertCRRToPod is reflected in the resulting fake pod's Status.
+func TestConvertCRRToPod_PodIPIsSet(t *testing.T) {
+	crr := makeCRR("my-pod", "abc-123", nil, nil)
+	pod := convertCRRToPod(crr, "10.0.0.42")
+
+	if pod.Status.PodIP != "10.0.0.42" {
+		t.Errorf("expected PodIP 10.0.0.42, got %q", pod.Status.PodIP)
+	}
+	if pod.Name != "my-pod" {
+		t.Errorf("expected pod name my-pod, got %q", pod.Name)
+	}
+	if string(pod.UID) != "abc-123" {
+		t.Errorf("expected UID abc-123, got %q", pod.UID)
+	}
+}
+
+// TestConvertCRRToPod_EmptyPodIP verifies that an empty pod IP is handled
+// gracefully (TCPSocket host fallback logic lives in the daemon executor).
+func TestConvertCRRToPod_EmptyPodIP(t *testing.T) {
+	crr := makeCRR("my-pod", "abc-123", nil, nil)
+	pod := convertCRRToPod(crr, "")
+
+	if pod.Status.PodIP != "" {
+		t.Errorf("expected empty PodIP, got %q", pod.Status.PodIP)
+	}
+}
+
+// TestConvertCRRToPod_DefaultGracePeriod verifies that TerminationGracePeriodSeconds
+// defaults to 30s when the CRR strategy does not specify one.
+func TestConvertCRRToPod_DefaultGracePeriod(t *testing.T) {
+	crr := makeCRR("my-pod", "abc-123", nil, nil)
+	pod := convertCRRToPod(crr, "")
+
+	if pod.Spec.TerminationGracePeriodSeconds == nil {
+		t.Fatal("expected TerminationGracePeriodSeconds to be set")
+	}
+	if *pod.Spec.TerminationGracePeriodSeconds != 30 {
+		t.Errorf("expected 30, got %d", *pod.Spec.TerminationGracePeriodSeconds)
+	}
+}
+
+// TestConvertCRRToPod_CustomGracePeriod verifies that a custom grace period from
+// the CRR strategy is propagated to the fake pod.
+func TestConvertCRRToPod_CustomGracePeriod(t *testing.T) {
+	crr := makeCRR("my-pod", "abc-123", nil, ptr.To(int64(60)))
+	pod := convertCRRToPod(crr, "")
+
+	if pod.Spec.TerminationGracePeriodSeconds == nil || *pod.Spec.TerminationGracePeriodSeconds != 60 {
+		t.Errorf("expected grace period 60")
+	}
+}
+
+// TestConvertCRRToPod_TCPSocketPreStop verifies that a TCPSocket preStop hook is
+// correctly reconstructed in the fake pod's container lifecycle spec.
+func TestConvertCRRToPod_TCPSocketPreStop(t *testing.T) {
+	containers := []appsv1beta1.ContainerRecreateRequestContainer{
+		{
+			Name: "app",
+			PreStop: &appsv1beta1.CRRProbeHandler{
+				TCPSocket: &v1.TCPSocketAction{
+					Port: intstr.FromInt(9000),
+					Host: "127.0.0.1",
+				},
+			},
+			Ports: []v1.ContainerPort{
+				{Name: "shutdown", ContainerPort: 9000},
+			},
+		},
+	}
+	crr := makeCRR("my-pod", "abc-123", containers, nil)
+	pod := convertCRRToPod(crr, "10.0.0.1")
+
+	if len(pod.Spec.Containers) != 1 {
+		t.Fatalf("expected 1 container, got %d", len(pod.Spec.Containers))
+	}
+	c := pod.Spec.Containers[0]
+	if c.Lifecycle == nil || c.Lifecycle.PreStop == nil {
+		t.Fatal("expected Lifecycle.PreStop to be set")
+	}
+	if c.Lifecycle.PreStop.TCPSocket == nil {
+		t.Fatal("expected TCPSocket to be set in PreStop")
+	}
+	if c.Lifecycle.PreStop.TCPSocket.Port.IntValue() != 9000 {
+		t.Errorf("expected port 9000, got %v", c.Lifecycle.PreStop.TCPSocket.Port)
+	}
+	if len(c.Ports) != 1 || c.Ports[0].Name != "shutdown" {
+		t.Error("expected container port 'shutdown' to be set for named-port resolution")
+	}
+}
+
+// TestConvertCRRToPod_NoPreStop verifies that a container without a preStop hook
+// produces no Lifecycle entry in the fake pod.
+func TestConvertCRRToPod_NoPreStop(t *testing.T) {
+	containers := []appsv1beta1.ContainerRecreateRequestContainer{
+		{Name: "app"},
+	}
+	crr := makeCRR("my-pod", "abc-123", containers, nil)
+	pod := convertCRRToPod(crr, "")
+
+	if len(pod.Spec.Containers) != 1 {
+		t.Fatalf("expected 1 container, got %d", len(pod.Spec.Containers))
+	}
+	if pod.Spec.Containers[0].Lifecycle != nil {
+		t.Error("expected no Lifecycle for container without preStop")
+	}
+}

pkg/daemon/kuberuntime/kuberuntime_container.go

@@ -20,13 +20,16 @@ package kuberuntime
 import (
 	"context"
 	"fmt"
+	"net"
 	"sort"
+	"strconv"
 	"strings"
 	"time"
 
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/intstr"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	runtimeapi "k8s.io/cri-api/pkg/apis/runtime/v1"
 	"k8s.io/klog/v2"
@@ -236,7 +239,21 @@ func (m *genericRuntimeManager) executePreStopHook(pod *v1.Pod, containerID kube
 	go func() {
 		defer close(done)
 		defer utilruntime.HandleCrash()
-		if msg, err := m.runner.Run(context.TODO(), containerID, pod, containerSpec, containerSpec.Lifecycle.PreStop); err != nil {
+
+		handler := containerSpec.Lifecycle.PreStop
+		// The upstream HandlerRunner handles Exec, HTTPGet, and Sleep but does not
+		// implement TCPSocket for lifecycle hooks. We handle TCPSocket ourselves here
+		// so that containers relying on a TCP-based shutdown signal are properly notified.
+		if handler.TCPSocket != nil {
+			if err := executeTCPSocketHook(handler.TCPSocket, pod, containerSpec, gracePeriod); err != nil {
+				klog.ErrorS(err, "preStop TCPSocket hook for container failed", "name", containerSpec.Name)
+				m.recordContainerEvent(pod, containerSpec, containerID.ID, v1.EventTypeWarning, events.FailedPreStopHook,
+					fmt.Sprintf("TCPSocket preStop hook failed: %v", err))
+			}
+			return
+		}
+
+		if msg, err := m.runner.Run(context.TODO(), containerID, pod, containerSpec, handler); err != nil {
 			klog.ErrorS(err, "preStop hook for container failed", "name", containerSpec.Name)
 			m.recordContainerEvent(pod, containerSpec, containerID.ID, v1.EventTypeWarning, events.FailedPreStopHook, msg)
 		}
@@ -251,3 +268,70 @@ func (m *genericRuntimeManager) executePreStopHook(pod *v1.Pod, containerID kube
 
 	return int64(metav1.Now().Sub(start.Time).Seconds())
 }
+
+// executeTCPSocketHook dials a TCP connection to the address specified by the
+// TCPSocket action. Kubernetes defines a successful TCP lifecycle hook as a
+// connection that is established (and then immediately closed); no data needs to
+// be exchanged. A connect timeout equal to the remaining grace period is applied
+// so the hook never blocks beyond the container's termination window.
+//
+// Port resolution follows the same rules as the Kubernetes probing infrastructure:
+//   - An integer port is used directly.
+//   - A named port is resolved against the container's declared port list.
+func executeTCPSocketHook(action *v1.TCPSocketAction, pod *v1.Pod, containerSpec *v1.Container, gracePeriodSeconds int64) error {
+	port, err := resolveTCPSocketPort(action.Port, containerSpec)
+	if err != nil {
+		return fmt.Errorf("failed to resolve TCPSocket port: %w", err)
+	}
+
+	host := action.Host
+	if host == "" {
+		// Fall back to the pod IP when no explicit host is provided. This mirrors
+		// how the HTTPGet probe handler resolves the target address.
+		if pod.Status.PodIP != "" {
+			host = pod.Status.PodIP
+		} else {
+			host = "localhost"
+		}
+	}
+
+	address := net.JoinHostPort(host, strconv.Itoa(port))
+
+	// Use the full remaining grace period as the dial timeout so we never block
+	// longer than the container's termination window allows.
+	timeout := time.Duration(gracePeriodSeconds) * time.Second
+	if timeout <= 0 {
+		timeout = time.Second
+	}
+
+	klog.V(4).InfoS("Dialing TCPSocket preStop hook", "address", address, "timeout", timeout)
+
+	conn, err := net.DialTimeout("tcp", address, timeout)
+	if err != nil {
+		return fmt.Errorf("TCPSocket dial to %s failed: %w", address, err)
+	}
+	_ = conn.Close()
+	return nil
+}
+
+// resolveTCPSocketPort converts the IntOrString port in a TCPSocketAction to an
+// integer. Named ports are looked up in the container's Ports declaration.
+func resolveTCPSocketPort(port intstr.IntOrString, containerSpec *v1.Container) (int, error) {
+	switch port.Type {
+	case intstr.Int:
+		p := int(port.IntVal)
+		if p < 1 || p > 65535 {
+			return 0, fmt.Errorf("invalid port number %d", p)
+		}
+		return p, nil
+	case intstr.String:
+		for _, cp := range containerSpec.Ports {
+			if cp.Name == port.StrVal {
+				return int(cp.ContainerPort), nil
+			}
+		}
+		return 0, fmt.Errorf("named port %q not found in container %q", port.StrVal, containerSpec.Name)
+	default:
+		return 0, fmt.Errorf("unknown port type %v", port.Type)
+	}
+}