limit nodagent container runtime mutation operation concurrency (#149)

Author: guowenjian90

cmd/runtimetest/main.go

@@ -25,7 +25,7 @@ import (
 )
 
 func main() {
-	runtimeService, err := dependency.InitRuntimeService(config.DefaultContainerRuntimeEndpoint)
+	runtimeService, err := dependency.InitRuntimeService(config.DefaultContainerRuntimeEndpoint, 1)
 	if err != nil {
 		klog.ErrorS(err, "Failed to init runtime service")
 		os.Exit(-1)

pkg/fornaxcore/application/application_session.go

@@ -96,7 +96,7 @@ func (am *ApplicationManager) onApplicationSessionAddEvent(obj interface{}) {
 
 	klog.InfoS("Application session created", "session", util.Name(session))
 	if v := pool.getSession(string(session.GetUID())); v != nil {
-		am.onApplicationSessionUpdateEvent(v.session, session)
+		am.onApplicationSessionUpdateEvent(v.session, v)
 		return
 	} else {
 		if !util.SessionInTerminalState(session) {
@@ -119,13 +119,12 @@ func (am *ApplicationManager) onApplicationSessionUpdateEvent(old, cur interface
 
 	if v := pool.getSession(string(newCopy.GetUID())); v != nil {
 		updateSessionPool(pool, newCopy)
-		am.enqueueApplication(applicationKey)
 	} else {
 		if !util.SessionInTerminalState(newCopy) {
 			updateSessionPool(pool, newCopy)
 		}
-		am.enqueueApplication(applicationKey)
 	}
+	am.enqueueApplication(applicationKey)
 }
 
 // callback from Application informer when ApplicationSession is physically deleted
@@ -143,8 +142,7 @@ func (am *ApplicationManager) onApplicationSessionDeleteEvent(obj interface{}) {
 	if pool == nil {
 		return
 	}
-	oldCopy := pool.getSession(string(session.GetUID()))
-	if oldCopy != nil {
+	if oldCopy := pool.getSession(string(session.GetUID())); oldCopy != nil {
 		if oldCopy.session.DeletionTimestamp == nil {
 			oldCopy.session.DeletionTimestamp = util.NewCurrentMetaTime()
 		}
@@ -168,14 +166,15 @@ func (ps PendingSessions) Swap(i, j int) {
 	ps[i], ps[j] = ps[j], ps[i]
 }
 
-// deployApplicationSessions grab a list of pending session and try to allocate them to pods and call OpenSession on choosen pod.
-// session status change in memory to SessionStatusStarting, but do not update etcd to avoid unnecessary resync.
-// session status will be changed in etcd until pod report back, if fornax core restart and lost these memory state, it rely on pod to report back.
-// It also cleanup session when a session is in Starting or Pending state for more than a timeout duration.
+// deployApplicationSessions group session into pending, timeout, deleting states, and
+// 1, assign pending session to idle pods and call OpenSession on choosen pod.
+// session status change in memory to SessionStatusStarting, session is store in node and report back,
+// if fornax core restart and lost these memory state, it rely on pod to report back.
+// 2, It cleanup timeout session which stuck in pending or starting session for more than a timeout duration.
+// and call node to close session if session is in starting state which was sent to a pod before.
 // session is changed to SessionStatusTimeout, session client need to create a new session.
-// It also cleanup session in deletingSessions when a session is in Starting or Pending state for more than a timeout duration.
-// session is changed to SessionStatusClosed, session client need to create a new session.
-// session timedout and closed are removed from application pool's session list, so, syncApplicationPods do not need to consider these sessions anymore
+// 3, if a session is being deleted by client(aka, close session), it call node to close session session,
+// timedout and closed session are removed from application's session pool
 func (am *ApplicationManager) deployApplicationSessions(pool *ApplicationPool, application *fornaxv1.Application) error {
 	pendingSessions, deletingSessions, timeoutSessions := pool.getNonRunningSessions()
 	// get 5 more in case some pods assigment failed
@@ -306,7 +305,7 @@ func (am *ApplicationManager) bindSessionToPod(pool *ApplicationPool, pod *v1.Po
 // cleanupSessionOnDeletedPod handle pod is terminated unexpectedly, e.g. node crash
 // in normal cases,session should be closed before pod is terminated and deleted.
 // It update open session to closed and pending session to timedout,
-// and does not try to call node to close session, as session does not exist at all on node when pod deleted
+// and does not try to call node to close session, as session does not exist at all on node when pod terminated on node
 func (am *ApplicationManager) cleanupSessionOnDeletedPod(pool *ApplicationPool, podName string) {
 	podSessions := pool.getPodSessions(podName)
 	for _, sess := range podSessions {
@@ -315,9 +314,9 @@ func (am *ApplicationManager) cleanupSessionOnDeletedPod(pool *ApplicationPool,
 	}
 }
 
-// cleanupSessionOfApplication if a application is being deleted,
-// close all sessions which are still alive and delete sessions from application sessions pool if they are still pending
-// when alive session reported as closed by Node Agent, then session can be eventually deleted
+// cleanupSessionOfApplication if a application is being deleted, it
+// call node to close all sessions which are still open, when session reported as closed from node, then session can be eventually deleted
+// if session is still pending assigned to pod, delete sessions from application sessions pool directly
 func (am *ApplicationManager) cleanupSessionOfApplication(pool *ApplicationPool) error {
 	deleteErrors := []error{}
 

pkg/nodeagent/config/config.go

@@ -73,6 +73,7 @@ const (
 	KubeletPluginsDirSELinuxLabel     = "system_u:object_r:container_file_t:s0"
 	DefaultPodCgroupName              = "containers"
 	DefaultRuntimeHandler             = "runc"
+	DefaultPodConcurrency             = 5
 )
 
 type NodeConfiguration struct {
@@ -111,6 +112,7 @@ type NodeConfiguration struct {
 	SeccompDefault           bool
 	NodePortStartingNo       int32
 	SessionServicePort       int32
+	PodConcurrency           int
 }
 
 func DefaultNodeConfiguration() (*NodeConfiguration, error) {
@@ -142,6 +144,7 @@ func DefaultNodeConfiguration() (*NodeConfiguration, error) {
 		NodeAgentCgroupName:      DefaultNodeAgentCgroupName,
 		OOMScoreAdj:              -999,
 		QOSReserved:              map[v1.ResourceName]int64{},
+		PodConcurrency:           DefaultPodConcurrency,
 		PodLogRootPath:           DefaultPodLogsRootPath,
 		PodPidLimits:             DefaultPodPidLimits,
 		PodsPerCore:              DefaultPodsPerCore,

pkg/nodeagent/dependency/dependency.go

@@ -80,7 +80,7 @@ func InitBasicDependencies(ctx context.Context, nodeConfig config.NodeConfigurat
 	dependencies.NetworkProvider = InitNetworkProvider(nodeConfig.Hostname)
 
 	// Runtime
-	dependencies.RuntimeService, err = InitRuntimeService(nodeConfig.ContainerRuntimeEndpoint)
+	dependencies.RuntimeService, err = InitRuntimeService(nodeConfig.ContainerRuntimeEndpoint, nodeConfig.PodConcurrency)
 	if err != nil {
 		klog.ErrorS(err, "failed to init container runtime client")
 		return nil, err
@@ -104,8 +104,8 @@ func InitBasicDependencies(ctx context.Context, nodeConfig config.NodeConfigurat
 	return &dependencies, nil
 }
 
-func InitRuntimeService(endpoint string) (runtime.RuntimeService, error) {
-	return runtime.NewRemoteRuntimeService(endpoint, runtime.DefaultTimeout)
+func InitRuntimeService(endpoint string, concurrency int) (runtime.RuntimeService, error) {
+	return runtime.NewRemoteRuntimeService(endpoint, runtime.DefaultTimeout, concurrency)
 }
 
 func InitImageService(endpoint string) (images.ImageManager, error) {
@@ -169,7 +169,7 @@ func (n *Dependencies) Complete(node *v1.Node, nodeConfig config.NodeConfigurati
 
 	// CRIRuntime
 	if n.RuntimeService == nil {
-		n.RuntimeService, err = InitRuntimeService(nodeConfig.ContainerRuntimeEndpoint)
+		n.RuntimeService, err = InitRuntimeService(nodeConfig.ContainerRuntimeEndpoint, nodeConfig.PodConcurrency)
 		if err != nil {
 			klog.ErrorS(err, "Failed to init runtime service")
 			return err

pkg/nodeagent/pod/pod_actor.go

@@ -309,7 +309,7 @@ func (a *PodActor) podHouseKeeping() (err error) {
 	}
 
 	switch {
-	case pod.FornaxPodState == types.PodStateTerminating:
+	case pod.FornaxPodState == types.PodStateTerminating || pod.FornaxPodState == types.PodStateFailed:
 		err = a.terminate(true)
 	case pod.RuntimePod == nil:
 		// pod create failed to create a sandbox

pkg/nodeagent/runtime/runtime.go

@@ -23,6 +23,7 @@ import (
 	"time"
 
 	grpc_util "centaurusinfra.io/fornax-serverless/pkg/util"
+	"golang.org/x/sync/semaphore"
 
 	"github.com/containerd/containerd"
 	"github.com/containerd/containerd/api/services/tasks/v1"
@@ -42,6 +43,7 @@ var _ RuntimeService = &remoteRuntimeManager{}
 type remoteRuntimeManager struct {
 	runtimeService    criapi.RuntimeService
 	containerdService *containerd.Client
+	podConcurrency    *semaphore.Weighted
 }
 
 // GetPodSandbox implements RuntimeService
@@ -77,12 +79,14 @@ func (r *remoteRuntimeManager) GetRuntimeStatus() (*criv1.RuntimeStatus, error)
 	return resp.GetStatus(), nil
 }
 
-func (m *remoteRuntimeManager) CreateContainer(podSandboxID string, containerConfig *criv1.ContainerConfig, podSandboxConfig *criv1.PodSandboxConfig) (*Container, error) {
+func (r *remoteRuntimeManager) CreateContainer(podSandboxID string, containerConfig *criv1.ContainerConfig, podSandboxConfig *criv1.PodSandboxConfig) (*Container, error) {
+	r.podConcurrency.Acquire(context.Background(), 1)
+	defer r.podConcurrency.Release(1)
 	klog.InfoS("Create container", "PodSandboxID", podSandboxID, "ContainerConfig", containerConfig)
 
 	var containerId string
 	var err error
-	containerId, err = m.runtimeService.CreateContainer(podSandboxID, containerConfig, podSandboxConfig)
+	containerId, err = r.runtimeService.CreateContainer(podSandboxID, containerConfig, podSandboxConfig)
 	if err != nil {
 		return nil, err
 	}
@@ -93,7 +97,7 @@ func (m *remoteRuntimeManager) CreateContainer(podSandboxID string, containerCon
 	}
 
 	var containers []*criv1.Container
-	containers, err = m.runtimeService.ListContainers(&criv1.ContainerFilter{
+	containers, err = r.runtimeService.ListContainers(&criv1.ContainerFilter{
 		Id:           containerId,
 		PodSandboxId: podSandboxID,
 	})
@@ -113,6 +117,8 @@ func (m *remoteRuntimeManager) CreateContainer(podSandboxID string, containerCon
 // CreateSandbox implements cri.RuntimeService
 // if RunPodSandbox succeeded but failed in following steps, return it still, it will be cleaned by node
 func (r *remoteRuntimeManager) CreateSandbox(sandboxConfig *criv1.PodSandboxConfig, runtimeClassName string) (*Pod, error) {
+	r.podConcurrency.Acquire(context.Background(), 1)
+	defer r.podConcurrency.Release(1)
 	klog.InfoS("Run pod sandbox", "SandboxConfig", sandboxConfig)
 	podSandBoxID, err := r.runtimeService.RunPodSandbox(sandboxConfig, runtimeClassName)
 	if err != nil {
@@ -268,6 +274,8 @@ func (r *remoteRuntimeManager) GetPods(includeContainers bool) ([]*Pod, error) {
 
 // StopContainer implements RuntimeService
 func (r *remoteRuntimeManager) StopContainer(containerID string, timeout time.Duration) error {
+	r.podConcurrency.Acquire(context.Background(), 1)
+	defer r.podConcurrency.Release(1)
 	klog.InfoS("Stop container", "ContainerID", containerID)
 
 	err := r.runtimeService.StopContainer(containerID, int64(timeout.Seconds()))
@@ -280,6 +288,8 @@ func (r *remoteRuntimeManager) StopContainer(containerID string, timeout time.Du
 
 // StartContainer implements cri.RuntimeService
 func (r *remoteRuntimeManager) StartContainer(containerID string) error {
+	r.podConcurrency.Acquire(context.Background(), 1)
+	defer r.podConcurrency.Release(1)
 	klog.InfoS("Start container", "ContainerID", containerID)
 
 	err := r.runtimeService.StartContainer(containerID)
@@ -291,6 +301,8 @@ func (r *remoteRuntimeManager) StartContainer(containerID string) error {
 
 // TerminateContainer implements cri.RuntimeService
 func (r *remoteRuntimeManager) TerminateContainer(containerID string) error {
+	r.podConcurrency.Acquire(context.Background(), 1)
+	defer r.podConcurrency.Release(1)
 	klog.InfoS("Terminate container, stop immediately without gracePeriod", "ContainerID", containerID)
 
 	status, err := r.GetContainerStatus(containerID)
@@ -317,6 +329,8 @@ func (r *remoteRuntimeManager) TerminateContainer(containerID string) error {
 
 // TerminatePod implements cri.RuntimeService
 func (r *remoteRuntimeManager) TerminatePod(podSandboxID string, containerIDs []string) error {
+	r.podConcurrency.Acquire(context.Background(), 1)
+	defer r.podConcurrency.Release(1)
 	klog.InfoS("Terminate pod", "PodSandboxID", podSandboxID)
 	var err error
 	if len(containerIDs) == 0 {
@@ -327,7 +341,7 @@ func (r *remoteRuntimeManager) TerminatePod(podSandboxID string, containerIDs []
 		for _, v := range containers {
 			if v.State != criv1.ContainerState_CONTAINER_EXITED {
 				klog.InfoS("Terminate container in pod", "PodSandboxID", podSandboxID, "ContainerID", v)
-				err = r.TerminateContainer(v.GetId())
+				err = r.runtimeService.StopContainer(v.GetId(), 0)
 				if err != nil {
 					return err
 				}
@@ -430,7 +444,7 @@ func (r *remoteRuntimeManager) getPodContainers(podSandboxID string) ([]*criv1.C
 	return containers, nil
 }
 
-func NewRemoteRuntimeService(endpoint string, connectionTimeout time.Duration) (*remoteRuntimeManager, error) {
+func NewRemoteRuntimeService(endpoint string, connectionTimeout time.Duration, concurrency int) (*remoteRuntimeManager, error) {
 	klog.InfoS("Connecting to runtime service", "endpoint", endpoint)
 	remoteService, err := remote.NewRemoteRuntimeService(endpoint, connectionTimeout)
 	if err != nil {
@@ -446,6 +460,7 @@ func NewRemoteRuntimeService(endpoint string, connectionTimeout time.Duration) (
 	service := &remoteRuntimeManager{
 		runtimeService:    remoteService,
 		containerdService: containerdClient,
+		podConcurrency:    semaphore.NewWeighted(int64(concurrency)),
 	}
 
 	return service, nil