Merge pull request #52503 from joelsmith/journald-log-fallback

Automatic merge from submit-queue (batch tested with PRs 54040, 52503). If you want to cherry-pick this change to another branch, please follow the instructions <a href="https://github.com/kubernetes/community/blob/master/contributors/devel/cherry-picks.md">here</a>.

Get fallback termination msg from docker when using journald log driver

**What this PR does / why we need it**:

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.

**Which issue this PR fixes** fixes #52502

**Special notes for your reviewer**:

**Release note**:
```release-note
Fixed log fallback termination messages when using docker with journald log driver
```

pkg/kubelet/dockershim/BUILD

@@ -64,6 +64,7 @@ go_library(
         "//pkg/security/apparmor:go_default_library",
         "//pkg/util/hash:go_default_library",
         "//pkg/util/parsers: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",
@@ -74,6 +75,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/client-go/tools/remotecommand: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"
@@ -515,6 +517,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

@@ -602,6 +602,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:
@@ -637,6 +639,7 @@ func NewMainKubelet(kubeCfg *kubeletconfiginternal.KubeletConfiguration,
 			}
 			if !supported {
 				klet.dockerLegacyService = dockershim.NewDockerLegacyService(kubeDeps.DockerClient)
+				legacyLogProvider = dockershim.NewLegacyLogProvider(klet.dockerLegacyService)
 			}
 		case kubetypes.RemoteContainerRuntime:
 			// No-op.
@@ -667,6 +670,7 @@ func NewMainKubelet(kubeCfg *kubeletconfiginternal.KubeletConfiguration,
 			runtimeService,
 			imageService,
 			kubeDeps.ContainerManager.InternalContainerLifecycle(),
+			legacyLogProvider,
 		)
 		if err != nil {
 			return nil, err

pkg/kubelet/kuberuntime/kuberuntime_container.go

@@ -427,8 +427,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)