UPSTREAM: 52503: Get fallback termination msg from docker when using journald log driver

When using the legacy docker container runtime and when a container has
terminationMessagePolicy=FallbackToLogsOnError and when docker is
configured with a log driver other than json-log (such as journald),
the kubelet should not try to get the container's log from the
json log file (since it's not there) but should instead ask docker for
the logs.

Author: joelsmith

pkg/kubelet/dockershim/BUILD

@@ -59,6 +59,7 @@ go_library(
         "//pkg/util/hash:go_default_library",
         "//pkg/util/parsers:go_default_library",
         "//pkg/util/term:go_default_library",
+        "//vendor/github.com/armon/circbuf:go_default_library",
         "//vendor/github.com/blang/semver:go_default_library",
         "//vendor/github.com/docker/docker/api/types:go_default_library",
         "//vendor/github.com/docker/docker/api/types/container:go_default_library",
@@ -69,6 +70,7 @@ go_library(
         "//vendor/github.com/golang/glog:go_default_library",
         "//vendor/k8s.io/api/core/v1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/types:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/util/errors:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/util/sets:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/util/wait:go_default_library",

pkg/kubelet/dockershim/docker_service.go

@@ -24,12 +24,14 @@ import (
 	"sync"
 	"time"
 
+	"github.com/armon/circbuf"
 	"github.com/blang/semver"
 	dockertypes "github.com/docker/docker/api/types"
 	"github.com/golang/glog"
 
 	"k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	kubetypes "k8s.io/apimachinery/pkg/types"
 	internalapi "k8s.io/kubernetes/pkg/kubelet/apis/cri"
 	runtimeapi "k8s.io/kubernetes/pkg/kubelet/apis/cri/v1alpha1/runtime"
 	"k8s.io/kubernetes/pkg/kubelet/apis/kubeletconfig"
@@ -524,6 +526,36 @@ func (d *dockerLegacyService) GetContainerLogs(pod *v1.Pod, containerID kubecont
 	return d.client.Logs(containerID.ID, opts, sopts)
 }
 
+// LegacyLogProvider implements the kuberuntime.LegacyLogProvider interface
+type LegacyLogProvider struct {
+	dls DockerLegacyService
+}
+
+func NewLegacyLogProvider(dls DockerLegacyService) LegacyLogProvider {
+	return LegacyLogProvider{dls: dls}
+}
+
+// GetContainerLogTail attempts to read up to MaxContainerTerminationMessageLogLength
+// from the end of the log when docker is configured with a log driver other than json-log.
+// It reads up to MaxContainerTerminationMessageLogLines lines.
+func (l LegacyLogProvider) GetContainerLogTail(uid kubetypes.UID, name, namespace string, containerId kubecontainer.ContainerID) (string, error) {
+	value := int64(kubecontainer.MaxContainerTerminationMessageLogLines)
+	buf, _ := circbuf.NewBuffer(kubecontainer.MaxContainerTerminationMessageLogLength)
+	// Although this is not a full spec pod, dockerLegacyService.GetContainerLogs() currently completely ignores its pod param
+	pod := &v1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			UID:       uid,
+			Name:      name,
+			Namespace: namespace,
+		},
+	}
+	err := l.dls.GetContainerLogs(pod, containerId, &v1.PodLogOptions{TailLines: &value}, buf, buf)
+	if err != nil {
+		return "", err
+	}
+	return buf.String(), nil
+}
+
 // criSupportedLogDrivers are log drivers supported by native CRI integration.
 var criSupportedLogDrivers = []string{"json-file"}
 

pkg/kubelet/kubelet.go

@@ -582,6 +582,8 @@ func NewMainKubelet(kubeCfg *kubeletconfiginternal.KubeletConfiguration,
 		// It's easier to always probe and initialize plugins till cri
 		// becomes the default.
 		klet.networkPlugin = nil
+		// if left at nil, that means it is unneeded
+		var legacyLogProvider kuberuntime.LegacyLogProvider
 
 		switch kubeCfg.ContainerRuntime {
 		case kubetypes.DockerContainerRuntime:
@@ -617,6 +619,7 @@ func NewMainKubelet(kubeCfg *kubeletconfiginternal.KubeletConfiguration,
 			}
 			if !supported {
 				klet.dockerLegacyService = dockershim.NewDockerLegacyService(kubeDeps.DockerClient)
+				legacyLogProvider = dockershim.NewLegacyLogProvider(klet.dockerLegacyService)
 			}
 		case kubetypes.RemoteContainerRuntime:
 			// No-op.
@@ -647,6 +650,7 @@ func NewMainKubelet(kubeCfg *kubeletconfiginternal.KubeletConfiguration,
 			runtimeService,
 			imageService,
 			kubeDeps.ContainerManager.InternalContainerLifecycle(),
+			legacyLogProvider,
 		)
 		if err != nil {
 			return nil, err

pkg/kubelet/kuberuntime/kuberuntime_container.go

@@ -425,8 +425,16 @@ func (m *kubeGenericRuntimeManager) getPodContainerStatuses(uid kubetypes.UID, n
 			fallbackToLogs := annotatedInfo.TerminationMessagePolicy == v1.TerminationMessageFallbackToLogsOnError && cStatus.ExitCode != 0
 			tMessage, checkLogs := getTerminationMessage(status, annotatedInfo.TerminationMessagePath, fallbackToLogs)
 			if checkLogs {
-				path := buildFullContainerLogsPath(uid, labeledInfo.ContainerName, annotatedInfo.RestartCount)
-				tMessage = m.readLastStringFromContainerLogs(path)
+				// if dockerLegacyService is populated, we're supposed to use it to fetch logs
+				if m.legacyLogProvider != nil {
+					tMessage, err = m.legacyLogProvider.GetContainerLogTail(uid, name, namespace, kubecontainer.ContainerID{Type: m.runtimeName, ID: c.Id})
+					if err != nil {
+						tMessage = fmt.Sprintf("Error reading termination message from logs: %v", err)
+					}
+				} else {
+					path := buildFullContainerLogsPath(uid, labeledInfo.ContainerName, annotatedInfo.RestartCount)
+					tMessage = m.readLastStringFromContainerLogs(path)
+				}
 			}
 			// Use the termination message written by the application is not empty
 			if len(tMessage) != 0 {

pkg/kubelet/kuberuntime/kuberuntime_manager.go

@@ -113,6 +113,9 @@ type kubeGenericRuntimeManager struct {
 
 	// Internal lifecycle event handlers for container resource management.
 	internalLifecycle cm.InternalContainerLifecycle
+
+	// A shim to legacy functions for backward compatibility.
+	legacyLogProvider LegacyLogProvider
 }
 
 type KubeGenericRuntime interface {
@@ -121,6 +124,12 @@ type KubeGenericRuntime interface {
 	kubecontainer.ContainerCommandRunner
 }
 
+// LegacyLogProvider gives the ability to use unsupported docker log drivers (e.g. journald)
+type LegacyLogProvider interface {
+	// Get the last few lines of the logs for a specific container.
+	GetContainerLogTail(uid kubetypes.UID, name, namespace string, containerID kubecontainer.ContainerID) (string, error)
+}
+
 // NewKubeGenericRuntimeManager creates a new kubeGenericRuntimeManager
 func NewKubeGenericRuntimeManager(
 	recorder record.EventRecorder,
@@ -140,6 +149,7 @@ func NewKubeGenericRuntimeManager(
 	runtimeService internalapi.RuntimeService,
 	imageService internalapi.ImageManagerService,
 	internalLifecycle cm.InternalContainerLifecycle,
+	legacyLogProvider LegacyLogProvider,
 ) (KubeGenericRuntime, error) {
 	kubeRuntimeManager := &kubeGenericRuntimeManager{
 		recorder:            recorder,
@@ -154,6 +164,7 @@ func NewKubeGenericRuntimeManager(
 		imageService:        newInstrumentedImageManagerService(imageService),
 		keyring:             credentialprovider.NewDockerKeyring(),
 		internalLifecycle:   internalLifecycle,
+		legacyLogProvider:   legacyLogProvider,
 	}
 
 	typedVersion, err := kubeRuntimeManager.runtimeService.Version(kubeRuntimeAPIVersion)