Method SingleStream

pkg/podlogs/cluster_writer.go:171–233

SingleStream streams the cluster's pod logs and shunts them to a single io.Writer

(ctx context.Context, writer io.Writer)

// SingleStream streams the cluster's pod logs and shunts them to a single io.Writer
func (csr *ClusterWriter) SingleStream(ctx context.Context, writer io.Writer) error {
	client := csr.getKubernetesClient()
	streamSet := newActiveSet()
	defer func() {
		// try to cancel the streaming goroutines
		ctx.Done()
	}()
	isFirstScan := true

	for {
		var (
			podList *corev1.PodList
			err     error
		)
		if isFirstScan || csr.Options.Follow {
			podList, err = client.CoreV1().Pods(csr.getClusterNamespace()).List(ctx, metav1.ListOptions{
				LabelSelector: utils.ClusterLabelName + "=" + csr.getClusterName(),
			})
			if err != nil {
				return err
			}
			isFirstScan = false
		} else {
			streamSet.wait()
			return nil
		}
		if len(podList.Items) == 0 && streamSet.isZero() {
			log.Printf("no pods to log in namespace %s", csr.getClusterNamespace())
			return nil
		}

		wrappedWriter := safeWriterFrom(writer)
		for _, pod := range podList.Items {
			for _, container := range pod.Status.ContainerStatuses {
				if container.State.Running != nil {
					streamName := fmt.Sprintf("%s-%s", pod.Name, container.Name)
					if streamSet.has(streamName) {
						continue
					}

					streamSet.add(streamName)
					go csr.streamInGoroutine(
						ctx,
						pod.Name,
						container.Name,
						client,
						streamSet,
						wrappedWriter,
					)
				}
			}
		}
		if !csr.Options.Follow && streamSet.isZero() {
			return nil
		}

		select {
		case <-ctx.Done():