Simplify code to print logs
@@ -62,45 +62,43 @@ func NewLogAggregator(out io.Writer, podSelector PodSelector, colorPicker ColorP
}
func (a *LogAggregator) Start(ctx context.Context) error {
+ a.startTime = time.Now()
+
kubeclient, err := Client()
if err != nil {
return errors.Wrap(err, "getting k8s client")
}
client := kubeclient.CoreV1()
- go func() {
- retryLoop:
- for {
- a.startTime = time.Now()
-
+ var forever int64 = 3600 * 24 * 365 * 100
watcher, err := client.Pods("").Watch(meta_v1.ListOptions{
IncludeUninitialized: true,
+ TimeoutSeconds: &forever,
})
if err != nil {
- logrus.Errorf("initializing pod watcher %s", err)
- return
+ return errors.Wrap(err, "initializing pod watcher")
}
- eventLoop:
+ go func() {
+ defer watcher.Stop()
+
for {
select {
case <-ctx.Done():
- watcher.Stop()
return
case evt, ok := <-watcher.ResultChan():
if !ok {
- // expected: server connection timeout
- continue retryLoop
+ return
}
if evt.Type != watch.Added && evt.Type != watch.Modified {
- continue eventLoop
+ continue
}
pod, ok := evt.Object.(*v1.Pod)
if !ok {
- continue eventLoop
+ continue
}
if a.podSelector.Select(pod) {
@@ -108,7 +106,6 @@ func (a *LogAggregator) Start(ctx context.Context) error {
}
}
}
- }
}()
return nil
@@ -128,18 +125,16 @@ func (a *LogAggregator) streamLogs(ctx context.Context, pod *v1.Pod) {
logrus.Infof("Stream logs from pod: %s container: %s", pod.Name, container.Name)
- tr, tw := io.Pipe()
- go func() {
sinceSeconds := int64(time.Since(a.startTime).Seconds() + 0.5)
// 0s means all the logs
if sinceSeconds == 0 {
sinceSeconds = 1
}
+ tr, tw := io.Pipe()
cmd := exec.CommandContext(ctx, "kubectl", "logs", fmt.Sprintf("--since=%ds", sinceSeconds), "-f", pod.Name, "-c", container.Name, "--namespace", pod.Namespace)
cmd.Stdout = tw
- cmd.Run()
- }()
+ go cmd.Run()
color := a.colorPicker.Pick(pod)
prefix := prefix(pod, container)
