Repository: kubernetes/kubernetes
File: staging/src/k8s.io/kubectl/pkg/cmd/logs/logs.go (Go)

Code before:
	}

	if o.Follow && len(requests) > 1 {
		if len(requests) > o.MaxFollowConcurrency {
			return fmt.Errorf(
				"you are attempting to follow %d log streams, but maximum allowed concurrency is %d, use --max-log-requests to increase the limit",
				len(requests), o.MaxFollowConcurrency,
			)
		}
	}

	if o.Follow && len(requests) > 1 {
		return o.parallelConsumeRequest(ctx, requests)
	}
	return o.sequentialConsumeRequest(ctx, requests)
}

func (o LogsOptions) parallelConsumeRequest(ctx context.Context, requests map[corev1.ObjectReference]rest.ResponseWrapper) error {
	reader, writer := io.Pipe()
	wg := &sync.WaitGroup{}
	wg.Add(len(requests))
	for objRef, request := range requests {
		go func(objRef corev1.ObjectReference, request rest.ResponseWrapper) {
			defer wg.Done()
			out := o.addPrefixIfNeeded(objRef, writer)
			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)
	}

	go func() {
		wg.Wait()
		writer.Close()
	}()

	_, err := io.Copy(o.Out, reader)
	return err
}

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.consumeWithRetry(ctx, request, out, o.Out); err != nil {
			return err
		}
	}


Review comment (refactor):
`kubectl logs` has a `--pod-running-timeout` which defaults to 20s. Currently this value is being passed only when retrieving the log request in https://github.com/kubernetes/kubernetes/blob/2274f69d9a00b52e076c3df5b82827f756ec8c81/staging/src/k8s.io/kubectl/pkg/cmd/logs/logs.go#L376-L380

When you look under the covers of `logsForobjectWithClient` https://github.com/kubernetes/kubernetes/blob/2274f69d9a00b52e076c3df5b82827f756ec8c81/staging/src/k8s.io/kubectl/pkg/polymorphichelpers/logsforobject.go#L64 you'll quickly realize that the timeout there is not being used at all. It's just being passed everywhere without ever being used. 

So I'm proposing the following:
1. Wire up the `--pod-running-timeout` into your retries logic, such that it will actually wait up to that limit, especially that this value is being controller by a user. 
2. Don't check for a specific error, I'd say anything from `Pending`, through starting is valid reason for us to wait for a pod. 


Code after:
	}

	if o.Follow && len(requests) > 1 {
		if len(requests) > o.MaxFollowConcurrency {
			return fmt.Errorf(
				"you are attempting to follow %d log streams, but maximum allowed concurrency is %d, use --max-log-requests to increase the limit",
				len(requests), o.MaxFollowConcurrency,
			)
		}
	}

	if o.Follow && len(requests) > 1 {
		return o.parallelConsumeRequest(ctx, requests)
	}
	return o.sequentialConsumeRequest(ctx, requests)
}

func (o LogsOptions) parallelConsumeRequest(ctx context.Context, requests map[corev1.ObjectReference]rest.ResponseWrapper) error {
	reader, writer := io.Pipe()
	wg := &sync.WaitGroup{}
	wg.Add(len(requests))
	for objRef, request := range requests {
		go func(objRef corev1.ObjectReference, request rest.ResponseWrapper) {
			defer wg.Done()
			out := o.addPrefixIfNeeded(objRef, writer)
			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)
	}

	go func() {
		wg.Wait()
		writer.Close()
	}()

	_, err := io.Copy(o.Out, reader)
	return err
}

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.consumeWithRetry(ctx, request, out, o.Out); err != nil {
			return err
		}
	}

	return nil