Support stream idle timeout.

Signed-off-by: Lantao Liu <lantaol@google.com>

cri.go

@@ -19,6 +19,7 @@ package cri
 import (
 	"flag"
 	"path/filepath"
+	"time"
 
 	"github.com/containerd/containerd"
 	"github.com/containerd/containerd/api/services/containers/v1"
@@ -118,6 +119,11 @@ func validateConfig(c *criconfig.Config) error {
 		}
 	}
 
+	if c.StreamIdleTimeout != "" {
+		if _, err := time.ParseDuration(c.StreamIdleTimeout); err != nil {
+			return errors.Wrap(err, "invalid stream idle timeout")
+		}
+	}
 	return nil
 }
 

docs/config.md

@@ -12,6 +12,12 @@ The explanation and default value of each configuration item are as follows:
   # stream_server_port is the port streaming server is listening on.
   stream_server_port = "0"
 
+  # stream_idle_timeout is the maximum time a streaming connection can be
+  # idle before the connection is automatically closed.
+  # The string is in the golang duration format, see:
+  #   https://golang.org/pkg/time/#ParseDuration
+  stream_idle_timeout = "4h"
+
   # enable_selinux indicates to enable the selinux support.
   enable_selinux = false
 

pkg/config/config.go

@@ -19,6 +19,7 @@ package config
 import (
 	"github.com/BurntSushi/toml"
 	"github.com/containerd/containerd"
+	"k8s.io/kubernetes/pkg/kubelet/server/streaming"
 )
 
 // Runtime struct to contain the type(ID), engine, and root variables for a default runtime
@@ -124,6 +125,11 @@ type PluginConfig struct {
 	StreamServerAddress string `toml:"stream_server_address" json:"streamServerAddress"`
 	// StreamServerPort is the port streaming server is listening on.
 	StreamServerPort string `toml:"stream_server_port" json:"streamServerPort"`
+	// StreamIdleTimeout is the maximum time a streaming connection
+	// can be idle before the connection is automatically closed.
+	// The string is in the golang duration format, see:
+	//   https://golang.org/pkg/time/#ParseDuration
+	StreamIdleTimeout string `toml:"stream_idle_timeout" json:"streamIdleTimeout"`
 	// EnableSelinux indicates to enable the selinux support.
 	EnableSelinux bool `toml:"enable_selinux" json:"enableSelinux"`
 	// SandboxImage is the image used by sandbox container.
@@ -196,6 +202,7 @@ func DefaultConfig() PluginConfig {
 		},
 		StreamServerAddress: "127.0.0.1",
 		StreamServerPort:    "0",
+		StreamIdleTimeout:   streaming.DefaultConfig.StreamIdleTimeout.String(), // 4 hour
 		EnableSelinux:       false,
 		EnableTLSStreaming:  false,
 		X509KeyPairStreaming: X509KeyPairStreaming{

pkg/server/service.go

@@ -159,7 +159,7 @@ func NewCRIService(config criconfig.Config, client *containerd.Client) (CRIServi
 		logrus.WithError(err).Error("Failed to load cni during init, please check CRI plugin status before setting up network for pods")
 	}
 	// prepare streaming server
-	c.streamServer, err = newStreamServer(c, config.StreamServerAddress, config.StreamServerPort)
+	c.streamServer, err = newStreamServer(c, config.StreamServerAddress, config.StreamServerPort, config.StreamIdleTimeout)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to create stream server")
 	}

pkg/server/streaming.go

@@ -22,6 +22,7 @@ import (
 	"math"
 	"net"
 	"os"
+	"time"
 
 	"github.com/pkg/errors"
 	k8snet "k8s.io/apimachinery/pkg/util/net"
@@ -64,7 +65,7 @@ func getStreamListenerMode(c *criService) (streamListenerMode, error) {
 	return withoutTLS, nil
 }
 
-func newStreamServer(c *criService, addr, port string) (streaming.Server, error) {
+func newStreamServer(c *criService, addr, port, streamIdleTimeout string) (streaming.Server, error) {
 	if addr == "" {
 		a, err := k8snet.ChooseBindAddress(nil)
 		if err != nil {
@@ -73,6 +74,13 @@ func newStreamServer(c *criService, addr, port string) (streaming.Server, error)
 		addr = a.String()
 	}
 	config := streaming.DefaultConfig
+	if streamIdleTimeout != "" {
+		var err error
+		config.StreamIdleTimeout, err = time.ParseDuration(streamIdleTimeout)
+		if err != nil {
+			return nil, errors.Wrap(err, "invalid stream idle timeout")
+		}
+	}
 	config.Addr = net.JoinHostPort(addr, port)
 	run := newStreamRuntime(c)
 	tlsMode, err := getStreamListenerMode(c)