feat: update validation helpers

Signed-off-by: Jian Zeng <anonymousknight96@gmail.com>

pkg/apis/core/v1/validation/validation.go

@@ -24,6 +24,7 @@ import (
 	"k8s.io/apimachinery/pkg/api/resource"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/validation/field"
+
 	"k8s.io/kubernetes/pkg/apis/core"
 	"k8s.io/kubernetes/pkg/apis/core/helper"
 	v1helper "k8s.io/kubernetes/pkg/apis/core/v1/helper"
@@ -124,6 +125,12 @@ func validateResourceName(value core.ResourceName, fldPath *field.Path) field.Er
 	return allErrs
 }
 
+var validLogStreams = sets.New[string](
+	v1.LogStreamStdout,
+	v1.LogStreamStderr,
+	v1.LogStreamAll,
+)
+
 // ValidatePodLogOptions checks if options that are set are at the correct
 // value. Any incorrect value will be returned to the ErrorList.
 func ValidatePodLogOptions(opts *v1.PodLogOptions) field.ErrorList {
@@ -142,6 +149,15 @@ func ValidatePodLogOptions(opts *v1.PodLogOptions) field.ErrorList {
 			allErrs = append(allErrs, field.Invalid(field.NewPath("sinceSeconds"), *opts.SinceSeconds, "must be greater than 0"))
 		}
 	}
+	// opts.Stream can be nil because defaulting might not apply if no URL params are provided.
+	if opts.Stream != nil {
+		if !validLogStreams.Has(*opts.Stream) {
+			allErrs = append(allErrs, field.NotSupported(field.NewPath("stream"), *opts.Stream, validLogStreams.UnsortedList()))
+		}
+		if *opts.Stream != v1.LogStreamAll && opts.TailLines != nil {
+			allErrs = append(allErrs, field.Forbidden(field.NewPath(""), "`tailLines` and specific `stream` are mutually exclusive for now"))
+		}
+	}
 	return allErrs
 }
 

pkg/apis/core/v1/validation/validation_test.go

@@ -25,6 +25,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/validation/field"
+
 	"k8s.io/kubernetes/pkg/apis/core"
 )
 
@@ -216,6 +217,10 @@ func TestValidatePodLogOptions(t *testing.T) {
 		sinceSecondsGreaterThan1 = int64(10)
 		sinceSecondsLessThan1    = int64(0)
 		timestamp                = metav1.Now()
+		stdoutStream             = v1.LogStreamStdout
+		stderrStream             = v1.LogStreamStderr
+		allStream                = v1.LogStreamAll
+		invalidStream            = "invalid"
 	)
 
 	successCase := []struct {
@@ -252,6 +257,24 @@ func TestValidatePodLogOptions(t *testing.T) {
 			TailLines:    &positiveLine,
 			SinceSeconds: &sinceSecondsGreaterThan1,
 		},
+	}, {
+		name: "PodLogOptions with stdout Stream",
+		podLogOptions: v1.PodLogOptions{
+			Stream: &stdoutStream,
+		},
+	}, {
+		name: "PodLogOptions with stderr Stream and Follow",
+		podLogOptions: v1.PodLogOptions{
+			Stream: &stderrStream,
+			Follow: true,
+		},
+	}, {
+		name: "PodLogOptions with All Stream, TailLines and LimitBytes",
+		podLogOptions: v1.PodLogOptions{
+			Stream:     &allStream,
+			TailLines:  &positiveLine,
+			LimitBytes: &limitBytesGreaterThan1,
+		},
 	}}
 	for _, tc := range successCase {
 		t.Run(tc.name, func(t *testing.T) {
@@ -293,6 +316,23 @@ func TestValidatePodLogOptions(t *testing.T) {
 			SinceSeconds: &sinceSecondsGreaterThan1,
 			SinceTime:    &timestamp,
 		},
+	}, {
+		name: "Invalid podLogOptions with invalid Stream",
+		podLogOptions: v1.PodLogOptions{
+			Stream: &invalidStream,
+		},
+	}, {
+		name: "Invalid podLogOptions with stdout Stream and TailLines set",
+		podLogOptions: v1.PodLogOptions{
+			Stream:    &stdoutStream,
+			TailLines: &positiveLine,
+		},
+	}, {
+		name: "Invalid podLogOptions with stderr Stream and TailLines set",
+		podLogOptions: v1.PodLogOptions{
+			Stream:    &stderrStream,
+			TailLines: &positiveLine,
+		},
 	}}
 	for _, tc := range errorCase {
 		t.Run(tc.name, func(t *testing.T) {

pkg/apis/core/validation/validation.go

@@ -31,6 +31,8 @@ import (
 	"unicode/utf8"
 
 	"github.com/google/go-cmp/cmp"
+	netutils "k8s.io/utils/net"
+
 	v1 "k8s.io/api/core/v1"
 	apiequality "k8s.io/apimachinery/pkg/api/equality"
 	"k8s.io/apimachinery/pkg/api/resource"
@@ -47,6 +49,7 @@ import (
 	utilsysctl "k8s.io/component-helpers/node/util/sysctl"
 	schedulinghelper "k8s.io/component-helpers/scheduling/corev1"
 	kubeletapis "k8s.io/kubelet/pkg/apis"
+
 	apiservice "k8s.io/kubernetes/pkg/api/service"
 	"k8s.io/kubernetes/pkg/apis/core"
 	"k8s.io/kubernetes/pkg/apis/core/helper"
@@ -57,7 +60,6 @@ import (
 	"k8s.io/kubernetes/pkg/cluster/ports"
 	"k8s.io/kubernetes/pkg/features"
 	"k8s.io/kubernetes/pkg/fieldpath"
-	netutils "k8s.io/utils/net"
 )
 
 const isNegativeErrorMsg string = apimachineryvalidation.IsNegativeErrorMsg
@@ -7522,7 +7524,13 @@ func validateOS(podSpec *core.PodSpec, fldPath *field.Path, opts PodValidationOp
 	return allErrs
 }
 
-func ValidatePodLogOptions(opts *core.PodLogOptions) field.ErrorList {
+var validLogStreams = sets.New[string](
+	core.LogStreamStdout,
+	core.LogStreamStderr,
+	core.LogStreamAll,
+)
+
+func ValidatePodLogOptions(opts *core.PodLogOptions, allowStreamSelection bool) field.ErrorList {
 	allErrs := field.ErrorList{}
 	if opts.TailLines != nil && *opts.TailLines < 0 {
 		allErrs = append(allErrs, field.Invalid(field.NewPath("tailLines"), *opts.TailLines, isNegativeErrorMsg))
@@ -7538,6 +7546,20 @@ func ValidatePodLogOptions(opts *core.PodLogOptions) field.ErrorList {
 			allErrs = append(allErrs, field.Invalid(field.NewPath("sinceSeconds"), *opts.SinceSeconds, "must be greater than 0"))
 		}
 	}
+	if allowStreamSelection {
+		if opts.Stream == nil {
+			allErrs = append(allErrs, field.Required(field.NewPath("stream"), "must be specified"))
+		} else {
+			if !validLogStreams.Has(*opts.Stream) {
+				allErrs = append(allErrs, field.NotSupported(field.NewPath("stream"), *opts.Stream, validLogStreams.UnsortedList()))
+			}
+			if *opts.Stream != core.LogStreamAll && opts.TailLines != nil {
+				allErrs = append(allErrs, field.Forbidden(field.NewPath(""), "`tailLines` and specific `stream` are mutually exclusive for now"))
+			}
+		}
+	} else if opts.Stream != nil {
+		allErrs = append(allErrs, field.Forbidden(field.NewPath("stream"), "may not be specified"))
+	}
 	return allErrs
 }
 

pkg/apis/core/validation/validation_test.go

@@ -21022,29 +21022,79 @@ func TestValidPodLogOptions(t *testing.T) {
 	negative := int64(-1)
 	zero := int64(0)
 	positive := int64(1)
+	stdoutStream := core.LogStreamStdout
+	stderrStream := core.LogStreamStderr
+	allStream := core.LogStreamAll
+	invalidStream := "invalid"
 	tests := []struct {
-		opt  core.PodLogOptions
+		opt                  core.PodLogOptions
-		errs int
+		errs                 int
+		allowStreamSelection bool
 	}{
-		{core.PodLogOptions{}, 0},
+		{core.PodLogOptions{}, 0, false},
-		{core.PodLogOptions{Previous: true}, 0},
+		{core.PodLogOptions{Previous: true}, 0, false},
-		{core.PodLogOptions{Follow: true}, 0},
+		{core.PodLogOptions{Follow: true}, 0, false},
-		{core.PodLogOptions{TailLines: &zero}, 0},
+		{core.PodLogOptions{TailLines: &zero}, 0, false},
-		{core.PodLogOptions{TailLines: &negative}, 1},
+		{core.PodLogOptions{TailLines: &negative}, 1, false},
-		{core.PodLogOptions{TailLines: &positive}, 0},
+		{core.PodLogOptions{TailLines: &positive}, 0, false},
-		{core.PodLogOptions{LimitBytes: &zero}, 1},
+		{core.PodLogOptions{LimitBytes: &zero}, 1, false},
-		{core.PodLogOptions{LimitBytes: &negative}, 1},
+		{core.PodLogOptions{LimitBytes: &negative}, 1, false},
-		{core.PodLogOptions{LimitBytes: &positive}, 0},
+		{core.PodLogOptions{LimitBytes: &positive}, 0, false},
-		{core.PodLogOptions{SinceSeconds: &negative}, 1},
+		{core.PodLogOptions{SinceSeconds: &negative}, 1, false},
-		{core.PodLogOptions{SinceSeconds: &positive}, 0},
+		{core.PodLogOptions{SinceSeconds: &positive}, 0, false},
-		{core.PodLogOptions{SinceSeconds: &zero}, 1},
+		{core.PodLogOptions{SinceSeconds: &zero}, 1, false},
-		{core.PodLogOptions{SinceTime: &now}, 0},
+		{core.PodLogOptions{SinceTime: &now}, 0, false},
+		{
+			opt: core.PodLogOptions{
+				Stream: &stdoutStream,
+			},
+			allowStreamSelection: false,
+			errs:                 1,
+		},
+		{
+			opt: core.PodLogOptions{
+				Stream: &stdoutStream,
+			},
+			allowStreamSelection: true,
+		},
+		{
+			opt: core.PodLogOptions{
+				Stream: &invalidStream,
+			},
+			allowStreamSelection: true,
+			errs:                 1,
+		},
+		{
+			opt: core.PodLogOptions{
+				Stream:    &stderrStream,
+				TailLines: &positive,
+			},
+			allowStreamSelection: true,
+			errs:                 1,
+		},
+		{
+			opt: core.PodLogOptions{
+				Stream:    &allStream,
+				TailLines: &positive,
+			},
+			allowStreamSelection: true,
+		},
+		{
+			opt: core.PodLogOptions{
+				Stream:     &stdoutStream,
+				LimitBytes: &positive,
+				SinceTime:  &now,
+			},
+			allowStreamSelection: true,
+		},
 	}
 	for i, test := range tests {
-		errs := ValidatePodLogOptions(&test.opt)
+		t.Run(fmt.Sprintf("case-%d", i), func(t *testing.T) {
-		if test.errs != len(errs) {
+			errs := ValidatePodLogOptions(&test.opt, test.allowStreamSelection)
-			t.Errorf("%d: Unexpected errors: %v", i, errs)
+			if test.errs != len(errs) {
-		}
+				t.Errorf("%d: Unexpected errors: %v", i, errs)
+			}
+		})
 	}
 }