Fix log line container not found in logging api (#6399)

* fix prev iteration

* use owner references

* fix NPE

* fix rpc error rpc error: code = Unknown desc = Error: No such container: fa7802b2206f84f4f1e166a7a640523a281031c4c95d1709d38d62680391b97c

* don't add containers for deleted pods

* better comments

pkg/skaffold/kubernetes/logger/log.go

@@ -26,6 +26,7 @@ import (
 
 	"github.com/sirupsen/logrus"
 	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/watch"
 
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/graph"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubectl"
@@ -138,6 +139,10 @@ func (a *LogAggregator) Start(ctx context.Context, out io.Writer) error {
 
 				// TODO(dgageot): Add EphemeralContainerStatuses
 				pod := evt.Pod
+				if evt.Type == watch.Deleted {
+					continue
+				}
+
 				for _, c := range append(pod.Status.InitContainerStatuses, pod.Status.ContainerStatuses...) {
 					if c.ContainerID == "" {
 						if c.State.Waiting != nil && c.State.Waiting.Message != "" {
@@ -219,7 +224,6 @@ func (a *LogAggregator) Unmute() {
 		// Logs are not activated.
 		return
 	}
-
 	atomic.StoreInt32(&a.muted, 0)
 }
 

pkg/skaffold/log/stream/stream.go

@@ -21,6 +21,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"strings"
 
 	"github.com/sirupsen/logrus"
 
@@ -38,8 +39,12 @@ func StreamRequest(ctx context.Context, out io.Writer, formatter log.Formatter,
 		default:
 			// Read up to newline
 			line, err := r.ReadString('\n')
+			// As per https://github.com/kubernetes/kubernetes/blob/017b359770e333eacd3efcb4174f1d464c208400/test/e2e/storage/podlogs/podlogs.go#L214
+			// Filter out the expected "end of stream" error message and
+			// attempts to read logs from a container that was deleted due to re-deploy or
+			// attempts to read logs from a container that is not ready yet.
 			if err == io.EOF {
-				if line != "" {
+				if !isEmptyOrContainerNotReady(line) {
 					formatter.PrintLine(out, line)
 				}
 				return nil
@@ -51,3 +56,10 @@ func StreamRequest(ctx context.Context, out io.Writer, formatter log.Formatter,
 		}
 	}
 }
+
+func isEmptyOrContainerNotReady(line string) bool {
+	return line == "" ||
+		strings.HasPrefix(line, "rpc error: code = Unknown desc = Error: No such container:") ||
+		strings.HasPrefix(line, "unable to retrieve container logs for ") ||
+		strings.HasPrefix(line, "Unable to retrieve container logs for ")
+}

pkg/skaffold/log/stream/stream_test.go

@@ -0,0 +1,61 @@
+/*
+Copyright 2021 The Skaffold Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package stream
+
+import (
+	"testing"
+
+	"github.com/GoogleContainerTools/skaffold/testutil"
+)
+
+func TestIsEmptyOrContainerNotReady(t *testing.T) {
+	tests := []struct {
+		description string
+		line        string
+		expected    bool
+	}{
+		{
+			description: "empty line",
+			line:        "",
+			expected:    true,
+		},
+		{
+			description: "container not ready",
+			line:        "rpc error: code = Unknown desc = Error: No such container: fa7802b2206f84f4f1e166a7a640523a281031c4c95d1709d38d62680391b97c",
+			expected:    true,
+		},
+		{
+			description: "logs could not be retrieved",
+			line:        "unable to retrieve container logs for fa7802b2206f84f4f1e166a7a640523a281031c4c95d1709d38d62680391b97c",
+			expected:    true,
+		},
+		{
+			description: "logs could not be retrieved",
+			line:        "Unable to retrieve container logs for fa7802b2206f84f4f1e166a7a640523a281031c4c95d1709d38d62680391b97c",
+			expected:    true,
+		},
+		{
+			description: "actual log",
+			line:        "log line",
+		},
+	}
+	for _, test := range tests {
+		testutil.Run(t, test.description, func(t *testutil.T) {
+			t.CheckDeepEqual(test.expected, isEmptyOrContainerNotReady(test.line))
+		})
+	}
+}