Merge pull request #112 from flant/improve_multitracker_output

Cli '--output-prefix' option; improve multitracker output

cmd/kubedog/main.go

@@ -29,6 +29,7 @@ func main() {
 	var logsSince string
 	var kubeContext string
 	var kubeConfig string
+	var outputPrefix string
 
 	makeTrackerOptions := func(mode string) tracker.Options {
 		// rollout track defaults
@@ -79,6 +80,7 @@ func main() {
 	rootCmd.PersistentFlags().StringVarP(&logsSince, "logs-since", "", "now", "A duration like 30s, 5m, or 2h to start log records from the past. 'all' to show all logs and 'now' to display only new records (default).")
 	rootCmd.PersistentFlags().StringVarP(&kubeContext, "kube-context", "", os.Getenv("KUBEDOG_KUBE_CONTEXT"), "The name of the kubeconfig context to use (can be set with $KUBEDOG_KUBE_CONTEXT).")
 	rootCmd.PersistentFlags().StringVarP(&kubeConfig, "kube-config", "", os.Getenv("KUBEDOG_KUBE_CONFIG"), "Path to the kubeconfig file (can be set with $KUBEDOG_KUBE_CONFIG).")
+	rootCmd.PersistentFlags().StringVarP(&outputPrefix, "output-prefix", "", "", "Arbitrary string which will be prefixed to kubedog output.")
 
 	versionCmd := &cobra.Command{
 		Use: "version",
@@ -95,6 +97,10 @@ func main() {
 		Run: func(cmd *cobra.Command, args []string) {
 			init()
 
+			if outputPrefix != "" {
+				logboek.SetPrefix(outputPrefix, logboek.ColorizeNone)
+			}
+
 			specsInput, err := ioutil.ReadAll(os.Stdin)
 			if err != nil {
 				fmt.Fprintf(os.Stderr, "Error reading stdin: %s\n", err)

go.mod

@@ -3,7 +3,7 @@ module github.com/flant/kubedog
 require (
 	github.com/acarl005/stripansi v0.0.0-20180116102854-5a71ef0e047d
 	github.com/fatih/color v1.7.0
-	github.com/flant/logboek v0.2.3
+	github.com/flant/logboek v0.2.4-0.20190702173639-9a253aa21697
 	github.com/gogo/protobuf v1.2.1 // indirect
 	github.com/golang/protobuf v1.3.1 // indirect
 	github.com/google/btree v1.0.0 // indirect

go.sum

@@ -22,6 +22,8 @@ github.com/flant/logboek v0.2.3-0.20190626194023-1580e4b6b6d9 h1:3DJ8TgWZfyi0nQt
 github.com/flant/logboek v0.2.3-0.20190626194023-1580e4b6b6d9/go.mod h1:eEQXX0UOWpyC8iaA9QOvzx8drZ64u0SRESiwQKnxF+I=
 github.com/flant/logboek v0.2.3 h1:NVc6TEXQeW0NEg0frSEvtw/fL7phnOnfjvjuJktLEUM=
 github.com/flant/logboek v0.2.3/go.mod h1:eEQXX0UOWpyC8iaA9QOvzx8drZ64u0SRESiwQKnxF+I=
+github.com/flant/logboek v0.2.4-0.20190702173639-9a253aa21697 h1:LxPW3KfHxVqnI+hqpwzAl+T0L7cjfMksXuQ6ghQgsM4=
+github.com/flant/logboek v0.2.4-0.20190702173639-9a253aa21697/go.mod h1:eEQXX0UOWpyC8iaA9QOvzx8drZ64u0SRESiwQKnxF+I=
 github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
 github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
 github.com/gogo/protobuf v1.2.1 h1:/s5zKNz0uPFCZ5hddgPdo2TK2TVrUNMn0OOX8/aZMTE=

pkg/trackers/rollout/multitrack/daemonset.go

@@ -61,9 +61,9 @@ func (mt *multitracker) TrackDaemonSet(kube kubernetes.Interface, spec Multitrac
 }
 
 func (mt *multitracker) daemonsetAdded(spec MultitrackSpec, feed daemonset.Feed, ready bool) error {
-	if ready {
-		mt.DaemonSetsStatuses[spec.ResourceName] = feed.GetStatus()
+	mt.DaemonSetsStatuses[spec.ResourceName] = feed.GetStatus()
 
+	if ready {
 		mt.displayResourceTrackerMessageF("ds", spec.ResourceName, "appears to be READY\n")
 
 		return mt.handleResourceReadyCondition(mt.TrackingDaemonSets, spec)
@@ -95,30 +95,16 @@ func (mt *multitracker) daemonsetEventMsg(spec MultitrackSpec, feed daemonset.Fe
 }
 
 func (mt *multitracker) daemonsetAddedReplicaSet(spec MultitrackSpec, feed daemonset.Feed, rs replicaset.ReplicaSet) error {
-	if !rs.IsNew {
-		return nil
-	}
-
 	mt.displayResourceTrackerMessageF("ds", spec.ResourceName, "rs/%s added\n", rs.Name)
-
 	return nil
 }
 
 func (mt *multitracker) daemonsetAddedPod(spec MultitrackSpec, feed daemonset.Feed, pod replicaset.ReplicaSetPod) error {
-	if !pod.ReplicaSet.IsNew {
-		return nil
-	}
-
 	mt.displayResourceTrackerMessageF("ds", spec.ResourceName, "po/%s added\n", pod.Name)
-
 	return nil
 }
 
 func (mt *multitracker) daemonsetPodError(spec MultitrackSpec, feed daemonset.Feed, podError replicaset.ReplicaSetPodError) error {
-	if !podError.ReplicaSet.IsNew {
-		return nil
-	}
-
 	reason := fmt.Sprintf("po/%s container/%s: %s", podError.PodName, podError.ContainerName, podError.Message)
 
 	mt.displayResourceErrorF("ds", spec.ResourceName, "%s\n", reason)
@@ -127,12 +113,14 @@ func (mt *multitracker) daemonsetPodError(spec MultitrackSpec, feed daemonset.Fe
 }
 
 func (mt *multitracker) daemonsetPodLogChunk(spec MultitrackSpec, feed daemonset.Feed, chunk *replicaset.ReplicaSetPodLogChunk) error {
-	if !chunk.ReplicaSet.IsNew {
-		return nil
+	controllerStatus := feed.GetStatus()
+	if podStatus, hasKey := controllerStatus.Pods[chunk.PodName]; hasKey {
+		if podStatus.IsReady {
+			return nil
+		}
 	}
 
 	mt.displayResourceLogChunk("ds", spec.ResourceName, podContainerLogChunkHeader(chunk.PodName, chunk.ContainerLogChunk), spec, chunk.ContainerLogChunk)
-
 	return nil
 }
 

pkg/trackers/rollout/multitrack/deployment.go

@@ -61,9 +61,9 @@ func (mt *multitracker) TrackDeployment(kube kubernetes.Interface, spec Multitra
 }
 
 func (mt *multitracker) deploymentAdded(spec MultitrackSpec, feed deployment.Feed, ready bool) error {
-	if ready {
-		mt.DeploymentsStatuses[spec.ResourceName] = feed.GetStatus()
+	mt.DeploymentsStatuses[spec.ResourceName] = feed.GetStatus()
 
+	if ready {
 		mt.displayResourceTrackerMessageF("deploy", spec.ResourceName, "appears to be READY\n")
 
 		return mt.handleResourceReadyCondition(mt.TrackingDeployments, spec)
@@ -129,6 +129,13 @@ func (mt *multitracker) deploymentPodLogChunk(spec MultitrackSpec, feed deployme
 		return nil
 	}
 
+	controllerStatus := feed.GetStatus()
+	if podStatus, hasKey := controllerStatus.Pods[chunk.PodName]; hasKey {
+		if podStatus.IsReady {
+			return nil
+		}
+	}
+
 	mt.displayResourceLogChunk("deploy", spec.ResourceName, podContainerLogChunkHeader(chunk.PodName, chunk.ContainerLogChunk), spec, chunk.ContainerLogChunk)
 
 	return nil

pkg/trackers/rollout/multitrack/job.go

@@ -56,6 +56,8 @@ func (mt *multitracker) TrackJob(kube kubernetes.Interface, spec MultitrackSpec,
 }
 
 func (mt *multitracker) jobAdded(spec MultitrackSpec, feed job.Feed) error {
+	mt.JobsStatuses[spec.ResourceName] = feed.GetStatus()
+
 	mt.displayResourceTrackerMessageF("job", spec.ResourceName, "added\n")
 
 	return nil
@@ -87,6 +89,13 @@ func (mt *multitracker) jobAddedPod(spec MultitrackSpec, feed job.Feed, podName
 }
 
 func (mt *multitracker) jobPodLogChunk(spec MultitrackSpec, feed job.Feed, chunk *pod.PodLogChunk) error {
+	controllerStatus := feed.GetStatus()
+	if podStatus, hasKey := controllerStatus.Pods[chunk.PodName]; hasKey {
+		if podStatus.IsReady {
+			return nil
+		}
+	}
+
 	mt.displayResourceLogChunk("job", spec.ResourceName, podContainerLogChunkHeader(chunk.PodName, chunk.ContainerLogChunk), spec, chunk.ContainerLogChunk)
 	return nil
 }

pkg/trackers/rollout/multitrack/multitrack.go

@@ -141,26 +141,21 @@ func Multitrack(kube kubernetes.Interface, specs MultitrackSpecs, opts Multitrac
 
 	errorChan := make(chan error, 0)
 	doneChan := make(chan struct{}, 0)
-	statusReportTicker := time.NewTicker(5 * time.Second)
-	defer statusReportTicker.Stop()
+	statusProgressTicker := time.NewTicker(5 * time.Second)
+	defer statusProgressTicker.Stop()
+
+	doDisplayStatusProgress := func() error {
+		mt.mux.Lock()
+		defer mt.mux.Unlock()
+		return mt.displayStatusProgress()
+	}
 
 	mt.Start(kube, specs, doneChan, errorChan, opts)
 
 	for {
 		select {
-		case <-statusReportTicker.C:
-			err := func() error {
-				mt.mux.Lock()
-				defer mt.mux.Unlock()
-
-				if err := mt.displayStatusProgress(); err != nil {
-					return err
-				}
-
-				return nil
-			}()
-
-			if err != nil {
+		case <-statusProgressTicker.C:
+			if err := doDisplayStatusProgress(); err != nil {
 				return err
 			}
 
@@ -244,7 +239,6 @@ func (mt *multitracker) Start(kube kubernetes.Interface, specs MultitrackSpecs,
 			defer mt.mux.Unlock()
 			return mt.displayStatusProgress()
 		}()
-
 		if err != nil {
 			errorChan <- err
 			return
@@ -323,6 +317,7 @@ type multitracker struct {
 	mux      sync.Mutex
 	isFailed bool
 
+	displayCalled                  bool
 	currentLogProcessHeader        string
 	debugDisplayMessagesByResource map[string][]string
 }

pkg/trackers/rollout/multitrack/multitrack_display.go

@@ -72,21 +72,21 @@ func (mt *multitracker) setLogProcess(header string) {
 	}
 }
 
+// resetLogProcess should be called every time something is about to be displayed
 func (mt *multitracker) resetLogProcess() {
+	mt.displayCalled = true
 	if mt.currentLogProcessHeader != "" {
 		logboek.LogProcessEnd(logboek.LogProcessEndOptions{WithoutLogOptionalLn: true, WithoutElapsedTime: true})
 		mt.currentLogProcessHeader = ""
 	}
 }
 
 func (mt *multitracker) displayResourceTrackerMessageF(resourceKind, resourceName string, format string, a ...interface{}) {
-	mt.resetLogProcess()
 	resource := fmt.Sprintf("%s/%s", resourceKind, resourceName)
 	mt.debugDisplayMessagesByResource[resource] = append(mt.debugDisplayMessagesByResource[resource], fmt.Sprintf(fmt.Sprintf("%s: %s", resource, format), a...))
 }
 
 func (mt *multitracker) displayResourceEventF(resourceKind, resourceName string, format string, a ...interface{}) {
-	mt.resetLogProcess()
 	resource := fmt.Sprintf("%s/%s", resourceKind, resourceName)
 	mt.debugDisplayMessagesByResource[resource] = append(mt.debugDisplayMessagesByResource[resource], fmt.Sprintf(fmt.Sprintf("%s event: %s", resource, format), a...))
 }
@@ -153,11 +153,18 @@ func (mt *multitracker) displayMultitrackErrorMessageF(format string, a ...inter
 }
 
 func (mt *multitracker) displayStatusProgress() error {
+	displayLn := false
+	if mt.displayCalled {
+		displayLn = true
+	}
+
 	mt.resetLogProcess()
 
-	caption := color.New(color.Bold).Sprint("Status progress")
+	if displayLn {
+		logboek.LogOptionalLn()
+	}
 
-	logboek.LogOptionalLn()
+	caption := color.New(color.Bold).Sprint("Status progress")
 
 	logboek.LogBlock(caption, logboek.LogBlockOptions{WithoutLogOptionalLn: true}, func() {
 		mt.displayDeploymentsStatusProgress()

pkg/trackers/rollout/multitrack/statefulset.go

@@ -61,9 +61,9 @@ func (mt *multitracker) TrackStatefulSet(kube kubernetes.Interface, spec Multitr
 }
 
 func (mt *multitracker) statefulsetAdded(spec MultitrackSpec, feed statefulset.Feed, ready bool) error {
-	if ready {
-		mt.StatefulSetsStatuses[spec.ResourceName] = feed.GetStatus()
+	mt.StatefulSetsStatuses[spec.ResourceName] = feed.GetStatus()
 
+	if ready {
 		mt.displayResourceTrackerMessageF("sts", spec.ResourceName, "appears to be READY\n")
 
 		return mt.handleResourceReadyCondition(mt.TrackingStatefulSets, spec)
@@ -94,30 +94,16 @@ func (mt *multitracker) statefulsetEventMsg(spec MultitrackSpec, feed statefulse
 }
 
 func (mt *multitracker) statefulsetAddedReplicaSet(spec MultitrackSpec, feed statefulset.Feed, rs replicaset.ReplicaSet) error {
-	if !rs.IsNew {
-		return nil
-	}
-
 	mt.displayResourceTrackerMessageF("sts", spec.ResourceName, "rs/%s added\n", rs.Name)
-
 	return nil
 }
 
 func (mt *multitracker) statefulsetAddedPod(spec MultitrackSpec, feed statefulset.Feed, pod replicaset.ReplicaSetPod) error {
-	if !pod.ReplicaSet.IsNew {
-		return nil
-	}
-
 	mt.displayResourceTrackerMessageF("sts", spec.ResourceName, "po/%s added\n", pod.Name)
-
 	return nil
 }
 
 func (mt *multitracker) statefulsetPodError(spec MultitrackSpec, feed statefulset.Feed, podError replicaset.ReplicaSetPodError) error {
-	if !podError.ReplicaSet.IsNew {
-		return nil
-	}
-
 	reason := fmt.Sprintf("po/%s container/%s: %s", podError.PodName, podError.ContainerName, podError.Message)
 
 	mt.displayResourceErrorF("sts", spec.ResourceName, "%s\n", reason)
@@ -126,12 +112,14 @@ func (mt *multitracker) statefulsetPodError(spec MultitrackSpec, feed statefulse
 }
 
 func (mt *multitracker) statefulsetPodLogChunk(spec MultitrackSpec, feed statefulset.Feed, chunk *replicaset.ReplicaSetPodLogChunk) error {
-	if !chunk.ReplicaSet.IsNew {
-		return nil
+	controllerStatus := feed.GetStatus()
+	if podStatus, hasKey := controllerStatus.Pods[chunk.PodName]; hasKey {
+		if podStatus.IsReady {
+			return nil
+		}
 	}
 
 	mt.displayResourceLogChunk("sts", spec.ResourceName, podContainerLogChunkHeader(chunk.PodName, chunk.ContainerLogChunk), spec, chunk.ContainerLogChunk)
-
 	return nil
 }