Merge pull request #136411 from olamilekan000/fix-cli-throwing-an-error-when-tailing-logs

fix cli throwing an error when trying to tail the logs for a Pod

staging/src/k8s.io/kubectl/pkg/cmd/logs/logs.go

@@ -32,6 +32,7 @@ import (
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/cli-runtime/pkg/genericclioptions"
 	"k8s.io/cli-runtime/pkg/genericiooptions"
 	"k8s.io/client-go/rest"
@@ -404,17 +405,11 @@ func (o LogsOptions) parallelConsumeRequest(ctx context.Context, requests map[co
 		go func(objRef corev1.ObjectReference, request rest.ResponseWrapper) {
 			defer wg.Done()
 			out := o.addPrefixIfNeeded(objRef, writer)
-			if err := o.ConsumeRequestFn(ctx, request, out); err != nil {
-				if !o.IgnoreLogErrors {
-					writer.CloseWithError(err)
-
-					// It's important to return here to propagate the error via the pipe
-					return
-				}
-
-				fmt.Fprintf(writer, "error: %v\n", err)
+			if err := o.consumeWithRetry(ctx, request, out, writer); err != nil {
+				writer.CloseWithError(err)
+				// It's important to return here to propagate the error via the pipe
+				return
 			}
-
 		}(objRef, request)
 	}
 
@@ -430,18 +425,45 @@ func (o LogsOptions) parallelConsumeRequest(ctx context.Context, requests map[co
 func (o LogsOptions) sequentialConsumeRequest(ctx context.Context, requests map[corev1.ObjectReference]rest.ResponseWrapper) error {
 	for objRef, request := range requests {
 		out := o.addPrefixIfNeeded(objRef, o.Out)
-		if err := o.ConsumeRequestFn(ctx, request, out); err != nil {
-			if !o.IgnoreLogErrors {
-				return err
-			}
-
-			fmt.Fprintf(o.Out, "error: %v\n", err)
+		if err := o.consumeWithRetry(ctx, request, out, o.Out); err != nil {
+			return err
 		}
 	}
 
 	return nil
 }
 
+func (o LogsOptions) consumeWithRetry(ctx context.Context, request rest.ResponseWrapper, out io.Writer, internalErrOut io.Writer) error {
+	if !o.Follow {
+		err := o.ConsumeRequestFn(ctx, request, out)
+		if err != nil && o.IgnoreLogErrors {
+			_, _ = fmt.Fprint(internalErrOut, "error: "+err.Error()+"\n")
+			return nil
+		}
+		return err
+	}
+
+	return wait.PollUntilContextTimeout(ctx, 1*time.Second, o.GetPodTimeout, true, func(_ context.Context) (bool, error) {
+		err := o.ConsumeRequestFn(ctx, request, out)
+		if err == nil {
+			return true, nil
+		}
+
+		if o.IgnoreLogErrors {
+			_, _ = fmt.Fprint(internalErrOut, "error: "+err.Error()+"\n")
+			return true, nil
+		}
+
+		_, _ = o.ErrOut.Write([]byte(err.Error() + "\n"))
+
+		if _, ok := err.(apierrors.APIStatus); ok {
+			return false, nil
+		}
+
+		return false, err
+	})
+}
+
 func (o LogsOptions) addPrefixIfNeeded(ref corev1.ObjectReference, writer io.Writer) io.Writer {
 	if !o.Prefix || ref.FieldPath == "" || ref.Name == "" {
 		return writer

staging/src/k8s.io/kubectl/pkg/cmd/logs/logs_test.go

@@ -64,6 +64,7 @@ func TestLog(t *testing.T) {
 				}
 
 				o := NewLogsOptions(streams)
+				o.GetPodTimeout = 5 * time.Millisecond
 				o.LogsForObject = mock.mockLogsForObject
 				o.ConsumeRequestFn = mock.mockConsumeRequest
 
@@ -85,6 +86,7 @@ func TestLog(t *testing.T) {
 				}
 
 				o := NewLogsOptions(streams)
+				o.GetPodTimeout = 5 * time.Millisecond
 				o.LogsForObject = mock.mockLogsForObject
 				o.ConsumeRequestFn = mock.mockConsumeRequest
 				o.Prefix = true
@@ -112,6 +114,7 @@ func TestLog(t *testing.T) {
 				}
 
 				o := NewLogsOptions(streams)
+				o.GetPodTimeout = 5 * time.Millisecond
 				o.LogsForObject = mock.mockLogsForObject
 				o.ConsumeRequestFn = mock.mockConsumeRequest
 				o.Prefix = true
@@ -136,6 +139,7 @@ func TestLog(t *testing.T) {
 				}
 
 				o := NewLogsOptions(streams)
+				o.GetPodTimeout = 5 * time.Millisecond
 				o.LogsForObject = mock.mockLogsForObject
 				o.ConsumeRequestFn = mock.mockConsumeRequest
 				o.Prefix = true
@@ -158,6 +162,7 @@ func TestLog(t *testing.T) {
 				}
 
 				o := NewLogsOptions(streams)
+				o.GetPodTimeout = 5 * time.Millisecond
 				o.LogsForObject = mock.mockLogsForObject
 				o.ConsumeRequestFn = mock.mockConsumeRequest
 				o.Prefix = true
@@ -190,6 +195,7 @@ func TestLog(t *testing.T) {
 				}
 
 				o := NewLogsOptions(streams)
+				o.GetPodTimeout = 5 * time.Millisecond
 				o.LogsForObject = mock.mockLogsForObject
 				o.ConsumeRequestFn = mock.mockConsumeRequest
 				return o
@@ -210,23 +216,36 @@ func TestLog(t *testing.T) {
 							Kind:      "Pod",
 							Name:      "some-pod-1",
 							FieldPath: "spec.containers{some-container-1}",
-						}: &responseWrapperMock{data: strings.NewReader("test log content from source 1\n")},
+						}: &statefulResponseWrapperMock{
+							results: []mockResult{
+								{data: strings.NewReader("test log content from source 1\n")},
+							},
+						},
 						{
 							Kind:      "Pod",
 							Name:      "some-pod-2",
 							FieldPath: "spec.containers{some-container-2}",
-						}: &responseWrapperMock{data: strings.NewReader("test log content from source 2\n")},
+						}: &statefulResponseWrapperMock{
+							results: []mockResult{
+								{data: strings.NewReader("test log content from source 2\n")},
+							},
+						},
 						{
 							Kind:      "Pod",
 							Name:      "some-pod-3",
 							FieldPath: "spec.containers{some-container-3}",
-						}: &responseWrapperMock{data: strings.NewReader("test log content from source 3\n")},
+						}: &statefulResponseWrapperMock{
+							results: []mockResult{
+								{data: strings.NewReader("test log content from source 3\n")},
+							},
+						},
 					},
 					wg: wg,
 				}
 				wg.Add(3)
 
 				o := NewLogsOptions(streams)
+				o.GetPodTimeout = 5 * time.Millisecond
 				o.LogsForObject = mock.mockLogsForObject
 				o.ConsumeRequestFn = mock.mockConsumeRequest
 				o.Follow = true
@@ -268,6 +287,7 @@ func TestLog(t *testing.T) {
 				o.LogsForObject = mock.mockLogsForObject
 				o.ConsumeRequestFn = mock.mockConsumeRequest
 				o.MaxFollowConcurrency = 2
+				o.GetPodTimeout = 5 * time.Millisecond
 				o.Follow = true
 				return o
 			},
@@ -277,6 +297,7 @@ func TestLog(t *testing.T) {
 			name: "fail if LogsForObject fails",
 			opts: func(streams genericiooptions.IOStreams) *LogsOptions {
 				o := NewLogsOptions(streams)
+				o.GetPodTimeout = 5 * time.Millisecond
 				o.LogsForObject = func(restClientGetter genericclioptions.RESTClientGetter, object, options runtime.Object, timeout time.Duration, allContainers bool) (map[corev1.ObjectReference]restclient.ResponseWrapper, error) {
 					return nil, errors.New("Error from the LogsForObject")
 				}
@@ -303,6 +324,7 @@ func TestLog(t *testing.T) {
 				}
 
 				o := NewLogsOptions(streams)
+				o.GetPodTimeout = 5 * time.Millisecond
 				o.LogsForObject = mock.mockLogsForObject
 				o.ConsumeRequestFn = func(ctx context.Context, req restclient.ResponseWrapper, out io.Writer) error {
 					return errors.New("Error from the ConsumeRequestFn")
@@ -321,17 +343,29 @@ func TestLog(t *testing.T) {
 							Kind:      "Pod",
 							Name:      "test-pod-1",
 							FieldPath: "spec.containers{test-container-1}",
-						}: &responseWrapperMock{data: strings.NewReader("test log content from source 1\n")},
+						}: &statefulResponseWrapperMock{
+							results: []mockResult{
+								{data: strings.NewReader("test log content from source 1\n")},
+							},
+						},
 						{
 							Kind:      "Pod",
 							Name:      "test-pod-2",
 							FieldPath: "spec.containers{test-container-2}",
-						}: &responseWrapperMock{data: strings.NewReader("test log content from source 2\n")},
+						}: &statefulResponseWrapperMock{
+							results: []mockResult{
+								{data: strings.NewReader("test log content from source 2\n")},
+							},
+						},
 						{
 							Kind:      "Pod",
 							Name:      "test-pod-3",
 							FieldPath: "spec.containers{test-container-3}",
-						}: &responseWrapperMock{data: strings.NewReader("test log content from source 3\n")},
+						}: &statefulResponseWrapperMock{
+							results: []mockResult{
+								{data: strings.NewReader("test log content from source 3\n")},
+							},
+						},
 					},
 					wg: wg,
 				}
@@ -340,6 +374,7 @@ func TestLog(t *testing.T) {
 				o := NewLogsOptions(streams)
 				o.LogsForObject = mock.mockLogsForObject
 				o.ConsumeRequestFn = mock.mockConsumeRequest
+				o.GetPodTimeout = 5 * time.Millisecond
 				o.Follow = true
 				o.Prefix = true
 				return o
@@ -360,17 +395,29 @@ func TestLog(t *testing.T) {
 							Kind:      "Pod",
 							Name:      "test-pod-1",
 							FieldPath: "spec.containers{test-container-1}",
-						}: &responseWrapperMock{},
+						}: &statefulResponseWrapperMock{
+							results: []mockResult{
+								{},
+							},
+						},
 						{
 							Kind:      "Pod",
 							Name:      "test-pod-2",
 							FieldPath: "spec.containers{test-container-2}",
-						}: &responseWrapperMock{},
+						}: &statefulResponseWrapperMock{
+							results: []mockResult{
+								{},
+							},
+						},
 						{
 							Kind:      "Pod",
 							Name:      "test-pod-3",
 							FieldPath: "spec.containers{test-container-3}",
-						}: &responseWrapperMock{},
+						}: &statefulResponseWrapperMock{
+							results: []mockResult{
+								{},
+							},
+						},
 					},
 					wg: wg,
 				}
@@ -382,6 +429,7 @@ func TestLog(t *testing.T) {
 					return errors.New("Error from the ConsumeRequestFn")
 				}
 				o.Follow = true
+				o.GetPodTimeout = 5 * time.Millisecond
 				return o
 			},
 			expectedErr: "Error from the ConsumeRequestFn",
@@ -405,6 +453,7 @@ func TestLog(t *testing.T) {
 					return errors.New("Error from the ConsumeRequestFn")
 				}
 				o.Follow = true
+				o.GetPodTimeout = 5 * time.Millisecond
 				return o
 			},
 			expectedErr: "Error from the ConsumeRequestFn",
@@ -433,6 +482,7 @@ func TestLog(t *testing.T) {
 				}
 
 				o := NewLogsOptions(streams)
+				o.GetPodTimeout = 5 * time.Millisecond
 				o.LogsForObject = mock.mockLogsForObject
 				o.ConsumeRequestFn = mock.mockConsumeRequest
 				o.IgnoreLogErrors = true
@@ -463,6 +513,7 @@ func TestLog(t *testing.T) {
 				}
 
 				o := NewLogsOptions(streams)
+				o.GetPodTimeout = 5 * time.Millisecond
 				o.LogsForObject = mock.mockLogsForObject
 				o.ConsumeRequestFn = mock.mockConsumeRequest
 				return o
@@ -494,6 +545,7 @@ func TestLog(t *testing.T) {
 
 				o := NewLogsOptions(streams)
 				o.LogsForObject = mock.mockLogsForObject
+				o.GetPodTimeout = 5 * time.Millisecond
 				o.ConsumeRequestFn = mock.mockConsumeRequest
 				o.IgnoreLogErrors = true
 				o.Follow = true
@@ -527,10 +579,73 @@ func TestLog(t *testing.T) {
 				o.LogsForObject = mock.mockLogsForObject
 				o.ConsumeRequestFn = mock.mockConsumeRequest
 				o.Follow = true
+				o.GetPodTimeout = 5 * time.Second
 				return o
 			},
 			expectedErr: "error-container",
 		},
+		{
+			name: "follow logs with retry on waiting container",
+			opts: func(streams genericiooptions.IOStreams) *LogsOptions {
+				mock := &logTestMock{
+					logsForObjectRequests: map[corev1.ObjectReference]restclient.ResponseWrapper{
+						{
+							Kind:      "Pod",
+							Name:      "some-pod",
+							FieldPath: "spec.containers{some-container}",
+						}: &statefulResponseWrapperMock{
+							results: []mockResult{
+								{
+									err: apierrors.NewBadRequest("container \"some-container\" in pod \"some-pod\" is waiting to start: ContainerCreating"),
+								},
+								{
+									data: strings.NewReader("test log content\n"),
+								},
+							},
+						},
+					},
+				}
+				o := NewLogsOptions(streams)
+				o.LogsForObject = mock.mockLogsForObject
+				o.ConsumeRequestFn = mock.mockConsumeRequest
+				o.Follow = true
+				o.GetPodTimeout = 5 * time.Second
+				return o
+			},
+			expectedOutSubstrings: []string{
+				"test log content\n",
+			},
+		},
+		{
+			name: "follow logs with retry and timeout",
+			opts: func(streams genericiooptions.IOStreams) *LogsOptions {
+				mock := &logTestMock{
+					logsForObjectRequests: map[corev1.ObjectReference]restclient.ResponseWrapper{
+						{
+							Kind:      "Pod",
+							Name:      "some-pod",
+							FieldPath: "spec.containers{some-container}",
+						}: &statefulResponseWrapperMock{
+							results: []mockResult{
+								{
+									err: apierrors.NewBadRequest("container \"some-container\" in pod \"some-pod\" is waiting to start: ContainerCreating"),
+								},
+								{
+									err: apierrors.NewBadRequest("container \"some-container\" in pod \"some-pod\" is waiting to start: ContainerCreating"),
+								},
+							},
+						},
+					},
+				}
+				o := NewLogsOptions(streams)
+				o.LogsForObject = mock.mockLogsForObject
+				o.ConsumeRequestFn = mock.mockConsumeRequest
+				o.Follow = true
+				o.GetPodTimeout = 1 * time.Millisecond
+				return o
+			},
+			expectedErr: "context deadline exceeded",
+		},
 	}
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
@@ -538,6 +653,7 @@ func TestLog(t *testing.T) {
 			defer tf.Cleanup()
 
 			streams, _, buf, _ := genericiooptions.NewTestIOStreams()
+			streams.ErrOut = &threadSafeWriter{writer: streams.ErrOut}
 
 			opts := test.opts(streams)
 			opts.Namespace = "test"
@@ -968,3 +1084,43 @@ func (l *logTestMock) mockLogsForObject(restClientGetter genericclioptions.RESTC
 		return nil, fmt.Errorf("cannot get the logs from %T", object)
 	}
 }
+
+type mockResult struct {
+	err  error
+	data io.Reader
+}
+
+type statefulResponseWrapperMock struct {
+	lock    sync.Mutex
+	calls   int
+	results []mockResult
+}
+
+func (r *statefulResponseWrapperMock) DoRaw(context.Context) ([]byte, error) {
+	return nil, nil
+}
+
+func (r *statefulResponseWrapperMock) Stream(context.Context) (io.ReadCloser, error) {
+	r.lock.Lock()
+	defer r.lock.Unlock()
+	if r.calls >= len(r.results) {
+		return nil, fmt.Errorf("unexpected call %d", r.calls)
+	}
+	res := r.results[r.calls]
+	r.calls++
+	if res.err != nil {
+		return nil, res.err
+	}
+	return io.NopCloser(res.data), nil
+}
+
+type threadSafeWriter struct {
+	mu     sync.Mutex
+	writer io.Writer
+}
+
+func (tsw *threadSafeWriter) Write(p []byte) (n int, err error) {
+	tsw.mu.Lock()
+	defer tsw.mu.Unlock()
+	return tsw.writer.Write(p)
+}