watchAllLogsOfJobContainers.go

func watchLogsOfAllJobContainers(
	podClient clientCoreV1.PodInterface,
	podListOptions *meta_v1.ListOptions,
	overallChannel *chan OverallEvent,
) (err error) {
	jobContainers := []podContainer{}
	var logsWaitGroup sync.WaitGroup

	podList, err := podClient.List(*podListOptions)
	// 遍历所有 pod 的容器, 将每个容器从属的 pod 名称以及本身的名称保存到总的数组中
	for _, pod := range podList.Items {
		podContainers := listPodContainers(pod)

		for _, containerName := range podContainers {
			jobContainers = append(jobContainers, podContainer{Pod: pod.Name, Container: containerName})
		}
	}

	containersNum := len(jobContainers)
	tailLines := int64(64)

	logsWaitGroup.Add(containersNum)

	for i := 0; i < containersNum; i++ {
		pc := jobContainers[i]
		logsRequest := podClient.GetLogs(pc.Pod, &coreV1.PodLogOptions{
			Container: pc.Container,
			Follow:    true,
			TailLines: &tailLines,
		})

		logsReader, err := logsRequest.Stream()
		if err != nil {
			sugarLogger.Error("Get logs stream failed", err)

			return err
		}

		go func() {
			defer logsWaitGroup.Done()

			logsBytes, err := ioutil.ReadAll(logsReader)
			if err != nil {
				sugarLogger.Error("Read all bytes from logsReader failed", err)

				return
			}

			*overallChannel <- OverallEvent{
				Type: OverallEventContainerLog,
				Log:  podContainer{Pod: pc.Pod, Container: pc.Container, LogBytes: logsBytes},
			}

			sugarLogger.Debug("Received bytes string", string(logsBytes[:]))
		}()
	}

	logsWaitGroup.Wait()

	sugarLogger.Debug("watchLogsOfAllJobContainers finished")

	return
}