Split streaming config from runtime config

Signed-off-by: Derek McGowan <[email protected]>

Author: dmcgowan

pkg/cri/config/config.go

@@ -26,6 +26,7 @@ import (
 	"github.com/containerd/log"
 	"github.com/pelletier/go-toml/v2"
 	runtime "k8s.io/cri-api/pkg/apis/runtime/v1"
+	"k8s.io/kubelet/pkg/cri/streaming"
 
 	runhcsoptions "github.com/Microsoft/hcsshim/cmd/containerd-shim-runhcs-v1/options"
 	runcoptions "github.com/containerd/containerd/v2/core/runtime/v2/runc/options"
@@ -312,31 +313,18 @@ type ImageConfig struct {
 	StatsCollectPeriod int `toml:"stats_collect_period" json:"statsCollectPeriod"`
 }
 
-// PluginConfig contains toml config related to CRI plugin,
+// RuntimeConfig contains toml config related to CRI plugin,
 // it is a subset of Config.
-type PluginConfig struct {
+type RuntimeConfig struct {
 	// ContainerdConfig contains config related to containerd
 	ContainerdConfig `toml:"containerd" json:"containerd"`
 	// CniConfig contains config related to cni
 	CniConfig `toml:"cni" json:"cni"`
-	// StreamServerAddress is the ip address streaming server is listening on.
-	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"`
 	// SelinuxCategoryRange allows the upper bound on the category range to be set.
 	// If not specified or set to 0, defaults to 1024 from the selinux package.
 	SelinuxCategoryRange int `toml:"selinux_category_range" json:"selinuxCategoryRange"`
-	// EnableTLSStreaming indicates to enable the TLS streaming support.
-	EnableTLSStreaming bool `toml:"enable_tls_streaming" json:"enableTLSStreaming"`
-	// X509KeyPairStreaming is a x509 key pair used for TLS streaming
-	X509KeyPairStreaming `toml:"x509_key_pair_streaming" json:"x509KeyPairStreaming"`
 	// MaxContainerLogLineSize is the maximum log line size in bytes for a container.
 	// Log line longer than the limit will be split into multiple lines. Non-positive
 	// value means no limit.
@@ -416,10 +404,10 @@ type X509KeyPairStreaming struct {
 	TLSKeyFile string `toml:"tls_key_file" json:"tlsKeyFile"`
 }
 
-// Config contains all configurations for cri server.
+// Config contains all configurations for CRI runtime plugin.
 type Config struct {
-	// PluginConfig is the config for CRI plugin.
-	PluginConfig
+	// RuntimeConfig is the config for CRI runtime.
+	RuntimeConfig
 	// ContainerdRootDir is the root directory path for containerd.
 	ContainerdRootDir string `json:"containerdRootDir"`
 	// ContainerdEndpoint is the containerd endpoint path.
@@ -431,10 +419,23 @@ type Config struct {
 	StateDir string `json:"stateDir"`
 }
 
-// ServiceConfig contains all the configuration for the CRI API server.
-type ServiceConfig struct {
+// ServerConfig contains all the configuration for the CRI API server.
+type ServerConfig struct {
 	// DisableTCPService disables serving CRI on the TCP server.
 	DisableTCPService bool `toml:"disable_tcp_service" json:"disableTCPService"`
+	// StreamServerAddress is the ip address streaming server is listening on.
+	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"`
+	// EnableTLSStreaming indicates to enable the TLS streaming support.
+	EnableTLSStreaming bool `toml:"enable_tls_streaming" json:"enableTLSStreaming"`
+	// X509KeyPairStreaming is a x509 key pair used for TLS streaming
+	X509KeyPairStreaming `toml:"x509_key_pair_streaming" json:"x509KeyPairStreaming"`
 }
 
 const (
@@ -498,8 +499,8 @@ func ValidateImageConfig(ctx context.Context, c *ImageConfig) ([]deprecation.War
 	return warnings, nil
 }
 
-// ValidatePluginConfig validates the given plugin configuration.
-func ValidatePluginConfig(ctx context.Context, c *PluginConfig) ([]deprecation.Warning, error) {
+// ValidateRuntimeConfig validates the given runtime configuration.
+func ValidateRuntimeConfig(ctx context.Context, c *RuntimeConfig) ([]deprecation.Warning, error) {
 	var warnings []deprecation.Warning
 	if c.ContainerdConfig.Runtimes == nil {
 		c.ContainerdConfig.Runtimes = make(map[string]Runtime)
@@ -524,13 +525,6 @@ func ValidatePluginConfig(ctx context.Context, c *PluginConfig) ([]deprecation.W
 		}
 	}
 
-	// Validation for stream_idle_timeout
-	if c.StreamIdleTimeout != "" {
-		if _, err := time.ParseDuration(c.StreamIdleTimeout); err != nil {
-			return warnings, fmt.Errorf("invalid stream idle timeout: %w", err)
-		}
-	}
-
 	// Validation for drain_exec_sync_io_timeout
 	if c.DrainExecSyncIOTimeout != "" {
 		if _, err := time.ParseDuration(c.DrainExecSyncIOTimeout); err != nil {
@@ -543,6 +537,18 @@ func ValidatePluginConfig(ctx context.Context, c *PluginConfig) ([]deprecation.W
 	return warnings, nil
 }
 
+// ValidateServerConfig validates the given server configuration.
+func ValidateServerConfig(ctx context.Context, c *ServerConfig) ([]deprecation.Warning, error) {
+	var warnings []deprecation.Warning
+	// Validation for stream_idle_timeout
+	if c.StreamIdleTimeout != "" {
+		if _, err := time.ParseDuration(c.StreamIdleTimeout); err != nil {
+			return warnings, fmt.Errorf("invalid stream idle timeout: %w", err)
+		}
+	}
+	return warnings, nil
+}
+
 func (config *Config) GetSandboxRuntime(podSandboxConfig *runtime.PodSandboxConfig, runtimeHandler string) (Runtime, error) {
 	if untrustedWorkload(podSandboxConfig) {
 		// If the untrusted annotation is provided, runtimeHandler MUST be empty.
@@ -631,3 +637,17 @@ func getRuntimeOptionsType(t string) interface{} {
 		return &runtimeoptions.Options{}
 	}
 }
+
+func DefaultServerConfig() ServerConfig {
+	return ServerConfig{
+		DisableTCPService:   true,
+		StreamServerAddress: "127.0.0.1",
+		StreamServerPort:    "0",
+		StreamIdleTimeout:   streaming.DefaultConfig.StreamIdleTimeout.String(), // 4 hour
+		EnableTLSStreaming:  false,
+		X509KeyPairStreaming: X509KeyPairStreaming{
+			TLSKeyFile:  "",
+			TLSCertFile: "",
+		},
+	}
+}

pkg/cri/config/config_kernel_linux.go

@@ -28,7 +28,7 @@ import (
 
 var kernelGreaterEqualThan = kernel.GreaterEqualThan
 
-func ValidateEnableUnprivileged(ctx context.Context, c *PluginConfig) error {
+func ValidateEnableUnprivileged(ctx context.Context, c *RuntimeConfig) error {
 	if c.EnableUnprivilegedICMP || c.EnableUnprivilegedPorts {
 		fourDotEleven := kernel.KernelVersion{Kernel: 4, Major: 11}
 		ok, err := kernelGreaterEqualThan(fourDotEleven)

pkg/cri/config/config_kernel_linux_test.go

@@ -32,13 +32,13 @@ func TestValidateEnableUnprivileged(t *testing.T) {
 
 	tests := []struct {
 		name          string
-		config        *PluginConfig
+		config        *RuntimeConfig
 		kernelGreater bool
 		expectedErr   string
 	}{
 		{
 			name: "disable unprivileged_icmp and unprivileged_port",
-			config: &PluginConfig{
+			config: &RuntimeConfig{
 				ContainerdConfig: ContainerdConfig{
 					DefaultRuntimeName: RuntimeDefault,
 					Runtimes: map[string]Runtime{
@@ -54,7 +54,7 @@ func TestValidateEnableUnprivileged(t *testing.T) {
 		},
 		{
 			name: "enable unprivileged_icmp or unprivileged_port, but kernel version is smaller than 4.11",
-			config: &PluginConfig{
+			config: &RuntimeConfig{
 				ContainerdConfig: ContainerdConfig{
 					DefaultRuntimeName: RuntimeDefault,
 					Runtimes: map[string]Runtime{
@@ -71,7 +71,7 @@ func TestValidateEnableUnprivileged(t *testing.T) {
 		},
 		{
 			name: "enable unprivileged_icmp or unprivileged_port, but kernel version is greater than or equal 4.11",
-			config: &PluginConfig{
+			config: &RuntimeConfig{
 				ContainerdConfig: ContainerdConfig{
 					DefaultRuntimeName: RuntimeDefault,
 					Runtimes: map[string]Runtime{

pkg/cri/config/config_kernel_other.go

@@ -22,6 +22,6 @@ import (
 	"context"
 )
 
-func ValidateEnableUnprivileged(ctx context.Context, c *PluginConfig) error {
+func ValidateEnableUnprivileged(ctx context.Context, c *RuntimeConfig) error {
 	return nil
 }

pkg/cri/config/config_test.go

@@ -28,37 +28,40 @@ import (
 
 func TestValidateConfig(t *testing.T) {
 	for desc, test := range map[string]struct {
-		config           *PluginConfig
-		expectedErr      string
-		expected         *PluginConfig
-		imageConfig      *ImageConfig
-		imageExpectedErr string
-		imageExpected    *ImageConfig
-		warnings         []deprecation.Warning
+		runtimeConfig      *RuntimeConfig
+		runtimeExpectedErr string
+		runtimeExpected    *RuntimeConfig
+		imageConfig        *ImageConfig
+		imageExpectedErr   string
+		imageExpected      *ImageConfig
+		serverConfig       *ServerConfig
+		serverExpectedErr  string
+		serverExpected     *ServerConfig
+		warnings           []deprecation.Warning
 	}{
 		"no default_runtime_name": {
-			config:      &PluginConfig{},
-			expectedErr: "`default_runtime_name` is empty",
+			runtimeConfig:      &RuntimeConfig{},
+			runtimeExpectedErr: "`default_runtime_name` is empty",
 		},
 		"no runtime[default_runtime_name]": {
-			config: &PluginConfig{
+			runtimeConfig: &RuntimeConfig{
 				ContainerdConfig: ContainerdConfig{
 					DefaultRuntimeName: RuntimeDefault,
 				},
 			},
-			expectedErr: "no corresponding runtime configured in `containerd.runtimes` for `containerd` `default_runtime_name = \"default\"",
+			runtimeExpectedErr: "no corresponding runtime configured in `containerd.runtimes` for `containerd` `default_runtime_name = \"default\"",
 		},
 
 		"deprecated auths": {
-			config: &PluginConfig{
+			runtimeConfig: &RuntimeConfig{
 				ContainerdConfig: ContainerdConfig{
 					DefaultRuntimeName: RuntimeDefault,
 					Runtimes: map[string]Runtime{
 						RuntimeDefault: {},
 					},
 				},
 			},
-			expected: &PluginConfig{
+			runtimeExpected: &RuntimeConfig{
 				ContainerdConfig: ContainerdConfig{
 					DefaultRuntimeName: RuntimeDefault,
 					Runtimes: map[string]Runtime{
@@ -92,18 +95,10 @@ func TestValidateConfig(t *testing.T) {
 			warnings: []deprecation.Warning{deprecation.CRIRegistryAuths},
 		},
 		"invalid stream_idle_timeout": {
-			config: &PluginConfig{
+			serverConfig: &ServerConfig{
 				StreamIdleTimeout: "invalid",
-				ContainerdConfig: ContainerdConfig{
-					DefaultRuntimeName: RuntimeDefault,
-					Runtimes: map[string]Runtime{
-						RuntimeDefault: {
-							Type: "default",
-						},
-					},
-				},
 			},
-			expectedErr: "invalid stream idle timeout",
+			serverExpectedErr: "invalid stream idle timeout",
 		},
 		"conflicting mirror registry config": {
 			imageConfig: &ImageConfig{
@@ -117,7 +112,7 @@ func TestValidateConfig(t *testing.T) {
 			imageExpectedErr: "`mirrors` cannot be set when `config_path` is provided",
 		},
 		"deprecated mirrors": {
-			config: &PluginConfig{
+			runtimeConfig: &RuntimeConfig{
 				ContainerdConfig: ContainerdConfig{
 					DefaultRuntimeName: RuntimeDefault,
 					Runtimes: map[string]Runtime{
@@ -132,7 +127,7 @@ func TestValidateConfig(t *testing.T) {
 					},
 				},
 			},
-			expected: &PluginConfig{
+			runtimeExpected: &RuntimeConfig{
 				ContainerdConfig: ContainerdConfig{
 					DefaultRuntimeName: RuntimeDefault,
 					Runtimes: map[string]Runtime{
@@ -152,7 +147,7 @@ func TestValidateConfig(t *testing.T) {
 			warnings: []deprecation.Warning{deprecation.CRIRegistryMirrors},
 		},
 		"deprecated configs": {
-			config: &PluginConfig{
+			runtimeConfig: &RuntimeConfig{
 				ContainerdConfig: ContainerdConfig{
 					DefaultRuntimeName: RuntimeDefault,
 					Runtimes: map[string]Runtime{
@@ -171,7 +166,7 @@ func TestValidateConfig(t *testing.T) {
 					},
 				},
 			},
-			expected: &PluginConfig{
+			runtimeExpected: &RuntimeConfig{
 				ContainerdConfig: ContainerdConfig{
 					DefaultRuntimeName: RuntimeDefault,
 					Runtimes: map[string]Runtime{
@@ -195,7 +190,7 @@ func TestValidateConfig(t *testing.T) {
 			warnings: []deprecation.Warning{deprecation.CRIRegistryConfigs},
 		},
 		"privileged_without_host_devices_all_devices_allowed without privileged_without_host_devices": {
-			config: &PluginConfig{
+			runtimeConfig: &RuntimeConfig{
 				ContainerdConfig: ContainerdConfig{
 					DefaultRuntimeName: RuntimeDefault,
 					Runtimes: map[string]Runtime{
@@ -207,10 +202,10 @@ func TestValidateConfig(t *testing.T) {
 					},
 				},
 			},
-			expectedErr: "`privileged_without_host_devices_all_devices_allowed` requires `privileged_without_host_devices` to be enabled",
+			runtimeExpectedErr: "`privileged_without_host_devices_all_devices_allowed` requires `privileged_without_host_devices` to be enabled",
 		},
 		"invalid drain_exec_sync_io_timeout input": {
-			config: &PluginConfig{
+			runtimeConfig: &RuntimeConfig{
 				ContainerdConfig: ContainerdConfig{
 					DefaultRuntimeName: RuntimeDefault,
 					Runtimes: map[string]Runtime{
@@ -221,18 +216,18 @@ func TestValidateConfig(t *testing.T) {
 				},
 				DrainExecSyncIOTimeout: "10",
 			},
-			expectedErr: "invalid `drain_exec_sync_io_timeout`",
+			runtimeExpectedErr: "invalid `drain_exec_sync_io_timeout`",
 		},
 	} {
 		t.Run(desc, func(t *testing.T) {
 			var warnings []deprecation.Warning
-			if test.config != nil {
-				w, err := ValidatePluginConfig(context.Background(), test.config)
-				if test.expectedErr != "" {
-					assert.Contains(t, err.Error(), test.expectedErr)
+			if test.runtimeConfig != nil {
+				w, err := ValidateRuntimeConfig(context.Background(), test.runtimeConfig)
+				if test.runtimeExpectedErr != "" {
+					assert.Contains(t, err.Error(), test.runtimeExpectedErr)
 				} else {
 					assert.NoError(t, err)
-					assert.Equal(t, test.expected, test.config)
+					assert.Equal(t, test.runtimeExpected, test.runtimeConfig)
 				}
 				warnings = append(warnings, w...)
 			}
@@ -246,6 +241,16 @@ func TestValidateConfig(t *testing.T) {
 				}
 				warnings = append(warnings, w...)
 			}
+			if test.serverConfig != nil {
+				w, err := ValidateServerConfig(context.Background(), test.serverConfig)
+				if test.serverExpectedErr != "" {
+					assert.Contains(t, err.Error(), test.serverExpectedErr)
+				} else {
+					assert.NoError(t, err)
+					assert.Equal(t, test.serverExpected, test.serverConfig)
+				}
+				warnings = append(warnings, w...)
+			}
 
 			if len(test.warnings) > 0 {
 				assert.ElementsMatch(t, test.warnings, warnings)