Force log order (#423)

Author: vishalbollu

pkg/operator/workloads/logs.go

@@ -188,9 +188,7 @@ func GetLogKeys(pod kcore.Pod) strset.Set {
 	containerStatuses := append(pod.Status.InitContainerStatuses, pod.Status.ContainerStatuses...)
 	logKeys := strset.New()
 	for _, status := range containerStatuses {
-		if status.State.Terminated != nil && (status.State.Terminated.ExitCode != 0 || status.State.Terminated.StartedAt.After(time.Now().Add(-newPodCheckInterval))) {
-			logKeys.Add(GetLogKey(pod, status).String())
-		} else if status.State.Running != nil {
+		if status.State.Terminated != nil || status.State.Running != nil {
 			logKeys.Add(GetLogKey(pod, status).String())
 		}
 	}
@@ -235,6 +233,10 @@ func podCheck(podCheckCancel chan struct{}, socket *websocket.Conn, initialPodLi
 	defer outw.Close()
 	defer outr.Close()
 
+	procAttr := os.ProcAttr{
+		Files: []*os.File{inr, outw, outw},
+	}
+
 	socketWriterError := make(chan error, 1)
 	defer close(socketWriterError)
 
@@ -286,10 +288,53 @@ func podCheck(podCheckCancel chan struct{}, socket *websocket.Conn, initialPodLi
 				wrotePending = false
 			}
 
+			initProcesses := strset.New()
+			tfServingProcesses := strset.New()
+			apiProcesses := strset.New()
+
 			for logProcess := range processesToAdd {
-				process, err := createKubectlProcess(StringToLogKey(logProcess), &os.ProcAttr{
-					Files: []*os.File{inr, outw, outw},
-				})
+				logKey := StringToLogKey(logProcess)
+				switch logKey.ContainerName {
+				case downloaderInitContainerName:
+					initProcesses.Add(logProcess)
+				case tfServingContainerName:
+					tfServingProcesses.Add(logProcess)
+				case apiContainerName:
+					apiProcesses.Add(logProcess)
+				default:
+					socketWriterError <- errors.New("unexpected container type encountered " + logKey.ContainerName) // unexpected
+					return
+				}
+			}
+
+			for logProcess := range initProcesses {
+				process, err := createKubectlProcess(StringToLogKey(logProcess), &procAttr)
+				if err != nil {
+					socketWriterError <- err
+					return
+				}
+				processMap[logProcess] = process
+			}
+
+			if len(tfServingProcesses) != 0 && len(initProcesses) != 0 {
+				time.Sleep(100 * time.Millisecond)
+			}
+
+			for logProcess := range tfServingProcesses {
+				process, err := createKubectlProcess(StringToLogKey(logProcess), &procAttr)
+				if err != nil {
+					socketWriterError <- err
+					return
+				}
+				processMap[logProcess] = process
+			}
+
+			if len(apiProcesses) != 0 && (len(tfServingProcesses) != 0 || len(initProcesses) != 0) {
+				time.Sleep(100 * time.Millisecond)
+			}
+
+			for logProcess := range apiProcesses {
+				process, err := createKubectlProcess(StringToLogKey(logProcess), &procAttr)
 				if err != nil {
 					socketWriterError <- err
 					return
@@ -387,7 +432,7 @@ func pumpStdout(socket *websocket.Conn, socketWriterError chan error, reader io.
 	select {
 	case err := <-socketWriterError:
 		if err != nil {
-			writeSocket(err.Error(), socket)
+			writeSocket(err.Error()+"\n", socket)
 		}
 	default:
 	}