Migrate to CRI container log format

.github/workflows/integration.yml

@@ -2,7 +2,7 @@ name: Integration Tests
 
 on:
   push:
-    branches: [ master ]
+    branches: [master]
   pull_request:
     branches: [ master ]
   workflow_call:
@@ -12,7 +12,6 @@ jobs:
     runs-on: ubuntu-20.04
     timeout-minutes: 60
     steps:
-
       - name: Set up Go 1.19.10
         uses: actions/setup-go@v1
         with:
@@ -71,7 +70,6 @@ jobs:
         with:
           repository: kubernetes-sigs/cri-tools
           path: src/sigs.k8s.io/cri-tools
-          ref: 5fd98895f3bbf8a3ba2d25e93fa95ba1e2ae0923
 
       - name: Build cri-tools
         working-directory: src/sigs.k8s.io/cri-tools

Makefile

@@ -79,3 +79,4 @@ docs:
 integration:
 	sudo critest -runtime-endpoint=unix:///var/run/cri-dockerd.sock -ginkgo.skip="runtime should support apparmor|runtime should support reopening container log|runtime should support execSync with timeout|runtime should support selinux|.*should support propagation.*"
 
+

core/container_start.go

@@ -19,6 +19,7 @@ package core
 import (
 	"context"
 	"fmt"
+
 	v1 "k8s.io/cri-api/pkg/apis/runtime/v1"
 )
 
@@ -30,7 +31,7 @@ func (ds *dockerService) StartContainer(
 	err := ds.client.StartContainer(r.ContainerId)
 
 	// Create container log symlink for all containers (including failed ones).
-	if linkError := ds.createContainerLogSymlink(r.ContainerId); linkError != nil {
+	if linkError := ds.createContainerKubeLogFile(r.ContainerId); linkError != nil {
 		// Do not stop the container if we failed to create symlink because:
 		//   1. This is not a critical failure.
 		//   2. We don't have enough information to properly stop container here.

core/logs.go

@@ -18,6 +18,7 @@ package core
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"io"
@@ -26,6 +27,7 @@ import (
 	"time"
 
 	"github.com/Mirantis/cri-dockerd/config"
+	"github.com/nxadm/tail"
 
 	"github.com/armon/circbuf"
 	dockertypes "github.com/docker/docker/api/types"
@@ -37,6 +39,7 @@ import (
 	runtimeapi "k8s.io/cri-api/pkg/apis/runtime/v1"
 
 	"github.com/Mirantis/cri-dockerd/libdocker"
+	"github.com/docker/docker/daemon/logger/jsonfilelog/jsonlog"
 )
 
 // ReopenContainerLog reopens the container log file.
@@ -85,6 +88,7 @@ func (ds *dockerService) GetContainerLogs(
 		stderr = SharedLimitWriter(stderr, &max)
 		stdout = SharedLimitWriter(stdout, &max)
 	}
+
 	sopts := libdocker.StreamOptions{
 		OutputStream: stdout,
 		ErrorStream:  stderr,
@@ -155,6 +159,7 @@ func (ds *dockerService) getContainerLogPath(containerID string) (string, string
 	if err != nil {
 		return "", "", fmt.Errorf("failed to inspect container %q: %v", containerID, err)
 	}
+
 	return info.Config.Labels[containerLogPathLabelKey], info.LogPath, nil
 }
 
@@ -222,3 +227,101 @@ func (ds *dockerService) removeContainerLogSymlink(containerID string) error {
 	}
 	return nil
 }
+
+// createContainerKubeLogFile creates the kube log file for docker container log.
+func (ds *dockerService) createContainerKubeLogFile(containerID string) error {
+	kubePath, dockerPath, err := ds.getContainerLogPath(containerID)
+	if err != nil {
+		return fmt.Errorf("failed to get container %q log path: %v", containerID, err)
+	}
+
+	if kubePath == "" {
+		logrus.Debugf("Container log path for Container ID %s isn't specified, will not create kubepath", containerID)
+		return nil
+	}
+
+	if dockerPath != "" {
+		// Only create the kube log file when container log path is specified and log file exists.
+		// Delete possibly existing file first
+		if err = ds.os.Remove(kubePath); err == nil {
+			logrus.Debugf("Deleted previously existing kube log file: %s", kubePath)
+		}
+
+		_, err = os.Create(kubePath)
+		if err != nil {
+			return fmt.Errorf(
+				"failed to create the kube log file %q to the container log file %q for container %q: %v",
+				kubePath,
+				dockerPath,
+				containerID,
+				err,
+			)
+
+		}
+
+		go func() {
+			// Open the kube file for output
+			kubeFile, err := os.OpenFile(kubePath, os.O_APPEND|os.O_WRONLY|os.O_CREATE, 0600)
+			if err != nil {
+				logrus.Errorf(
+					"failed to open kube log file %s: %v",
+					kubePath,
+					err,
+				)
+				panic(err)
+			}
+			defer kubeFile.Close()
+
+			// Tail the docker file for input
+			t, err := tail.TailFile(dockerPath, tail.Config{
+				Follow: true,
+				ReOpen: true})
+			if err != nil {
+				logrus.Errorf(
+					"failed to tail docker log file %s: %v",
+					dockerPath,
+					err,
+				)
+				panic(err)
+			}
+			// Watch for changes to be copied over
+			for line := range t.Lines {
+				logLine := log{}
+				json.Unmarshal([]byte(line.Text), &logLine)
+
+				_, err := kubeFile.WriteString(logLine.CRIFormat())
+				if err != nil {
+					logrus.Errorf(
+						"failed to write to kube log file %s: %v",
+						kubePath,
+						err,
+					)
+					return
+				}
+			}
+		}()
+	} else {
+		supported, err := ds.IsCRISupportedLogDriver()
+		if err != nil {
+			logrus.Errorf("Failed to check supported logging driver for CRI: %v", err)
+			return nil
+		}
+
+		if supported {
+			logrus.Info("Cannot create kube log file because container log file doesn't exist!")
+		} else {
+			logrus.Debug("Unsupported logging driver by CRI")
+		}
+	}
+
+	return nil
+}
+
+type log struct {
+	jsonlog.JSONLog
+}
+
+// CRIFormat returns the log in the CRI format
+func (l log) CRIFormat() string {
+	return fmt.Sprintf("%s %s %s %s", l.Created.Format(time.RFC3339), l.Stream, l.Attrs, l.Log)
+}

core/sandbox_helpers.go

@@ -18,13 +18,14 @@ package core
 
 import (
 	"fmt"
+	"os"
+	"strings"
+	"time"
+
 	"github.com/Mirantis/cri-dockerd/libdocker"
 	"github.com/Mirantis/cri-dockerd/utils"
 	"github.com/Mirantis/cri-dockerd/utils/errors"
 	"k8s.io/kubernetes/pkg/credentialprovider"
-	"os"
-	"strings"
-	"time"
 
 	"github.com/Mirantis/cri-dockerd/config"
 	dockertypes "github.com/docker/docker/api/types"
@@ -376,7 +377,7 @@ func recoverFromCreationConflictIfNeeded(
 	client libdocker.DockerClientInterface,
 	createConfig dockertypes.ContainerCreateConfig,
 	err error,
-) (*dockercontainer.ContainerCreateCreatedBody, error) {
+) (*dockercontainer.CreateResponse, error) {
 	matches := conflictRE.FindStringSubmatch(err.Error())
 	if len(matches) != 2 {
 		return nil, err

go.mod

@@ -9,11 +9,12 @@ require (
 	github.com/containernetworking/cni v1.1.2
 	github.com/coreos/go-systemd/v22 v22.5.0
 	github.com/davecgh/go-spew v1.1.1
-	github.com/docker/distribution v2.8.3+incompatible
-	github.com/docker/docker v23.0.3+incompatible
+	github.com/docker/distribution v2.8.2-beta.1+incompatible
+	github.com/docker/docker v24.0.6+incompatible
 	github.com/docker/go-connections v0.4.0
 	github.com/emicklei/go-restful v2.16.0+incompatible
 	github.com/golang/mock v1.6.0
+	github.com/nxadm/tail v1.4.8
 	github.com/opencontainers/go-digest v1.0.0
 	github.com/opencontainers/image-spec v1.1.0-rc5
 	github.com/opencontainers/runc v1.1.10
@@ -38,6 +39,7 @@ require (
 )
 
 require (
+	github.com/AdaLogics/go-fuzz-headers v0.0.0-20230811130428-ced1acdcaa24 // indirect
 	github.com/Azure/go-ansiterm v0.0.0-20210617225240-d185dfc1b5a1 // indirect
 	github.com/JeffAshton/win_pdh v0.0.0-20161109143554-76bb4ee9f0ab // indirect
 	github.com/Microsoft/go-winio v0.4.17 // indirect
@@ -47,7 +49,6 @@ require (
 	github.com/containerd/cgroups v1.0.1 // indirect
 	github.com/coreos/go-semver v0.3.0 // indirect
 	github.com/cyphar/filepath-securejoin v0.2.3 // indirect
-	github.com/distribution/reference v0.5.0 // indirect
 	github.com/docker/go-units v0.5.0 // indirect
 	github.com/evanphx/json-patch v4.12.0+incompatible // indirect
 	github.com/felixge/httpsnoop v1.0.1 // indirect
@@ -126,6 +127,7 @@ require (
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20230822172742-b8732ec3820d // indirect
 	google.golang.org/protobuf v1.31.0 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
+	gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 	k8s.io/cloud-provider v0.22.8 // indirect

go.sum

@@ -44,6 +44,8 @@ cloud.google.com/go/storage v1.8.0/go.mod h1:Wv1Oy7z6Yz3DshWRJFhqM/UCfaWIRTdp0RX
 cloud.google.com/go/storage v1.10.0/go.mod h1:FLPqc6j+Ki4BU591ie1oL6qBQGu2Bl/tZ9ullr3+Kg0=
 dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU=
 dmitri.shuralyov.com/gpu/mtl v0.0.0-20201218220906-28db891af037/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU=
+github.com/AdaLogics/go-fuzz-headers v0.0.0-20230811130428-ced1acdcaa24 h1:bvDV9vkmnHYOMsOr4WLk+Vo07yKIzd94sVoIqshQ4bU=
+github.com/AdaLogics/go-fuzz-headers v0.0.0-20230811130428-ced1acdcaa24/go.mod h1:8o94RPi1/7XTJvwPpRSzSUedZrtlirdB3r9Z20bi2f8=
 github.com/Azure/azure-sdk-for-go v55.0.0+incompatible/go.mod h1:9XXNKU+eRnpl9moKnB4QOLf1HestfXbmab5FXxiDBjc=
 github.com/Azure/go-ansiterm v0.0.0-20170929234023-d6e3b3328b78/go.mod h1:LmzpDX56iTiv29bbRTIsUNlaFfuhWRQBWjQdVyAevI8=
 github.com/Azure/go-ansiterm v0.0.0-20210608223527-2377c96fe795/go.mod h1:LmzpDX56iTiv29bbRTIsUNlaFfuhWRQBWjQdVyAevI8=
@@ -194,18 +196,16 @@ github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSs
 github.com/daviddengcn/go-colortext v0.0.0-20160507010035-511bcaf42ccd/go.mod h1:dv4zxwHi5C/8AeI+4gX4dCWOIvNi7I6JCSX0HvlKPgE=
 github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZmtrrCbhqsmaPHjLKYnJCaQ=
 github.com/dgryski/go-sip13 v0.0.0-20181026042036-e10d5fee7954/go.mod h1:vAd38F8PWV+bWy6jNmig1y/TA+kYO4g3RSRF0IAv0no=
-github.com/distribution/reference v0.5.0 h1:/FUIFXtfc/x2gpa5/VGfiGLuOIdYa1t65IKK2OFGvA0=
-github.com/distribution/reference v0.5.0/go.mod h1:BbU0aIcezP1/5jX/8MP0YiH4SdvB5Y4f/wlDRiLyi3E=
 github.com/dnaeon/go-vcr v1.0.1/go.mod h1:aBB1+wY4s93YsC3HHjMBMrwTj2R9FHDzUr9KyGc8n1E=
 github.com/docker/distribution v2.7.1-0.20190205005809-0d3efadf0154+incompatible/go.mod h1:J2gT2udsDAN96Uj4KfcMRqY0/ypR+oyYUYmja8H+y+w=
 github.com/docker/distribution v2.7.1+incompatible/go.mod h1:J2gT2udsDAN96Uj4KfcMRqY0/ypR+oyYUYmja8H+y+w=
 github.com/docker/distribution v2.8.0+incompatible/go.mod h1:J2gT2udsDAN96Uj4KfcMRqY0/ypR+oyYUYmja8H+y+w=
-github.com/docker/distribution v2.8.3+incompatible h1:AtKxIZ36LoNK51+Z6RpzLpddBirtxJnzDrHLEKxTAYk=
-github.com/docker/distribution v2.8.3+incompatible/go.mod h1:J2gT2udsDAN96Uj4KfcMRqY0/ypR+oyYUYmja8H+y+w=
+github.com/docker/distribution v2.8.2-beta.1+incompatible h1:gILO60VLD2v28ozemv4aAwDb8ds5U2O/vD/sBXbd7Rw=
+github.com/docker/distribution v2.8.2-beta.1+incompatible/go.mod h1:J2gT2udsDAN96Uj4KfcMRqY0/ypR+oyYUYmja8H+y+w=
 github.com/docker/docker v20.10.2+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk=
 github.com/docker/docker v20.10.12+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk=
-github.com/docker/docker v23.0.3+incompatible h1:9GhVsShNWz1hO//9BNg/dpMnZW25KydO4wtVxWAIbho=
-github.com/docker/docker v23.0.3+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk=
+github.com/docker/docker v24.0.6+incompatible h1:hceabKCtUgDqPu+qm0NgsaXf28Ljf4/pWFL7xjWWDgE=
+github.com/docker/docker v24.0.6+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk=
 github.com/docker/go-connections v0.4.0 h1:El9xVISelRB7BuFusrZozjnkIM5YnzCViNKohAFqRJQ=
 github.com/docker/go-connections v0.4.0/go.mod h1:Gbd7IOopHjR8Iph03tsViu4nIes5XhDvyHbTtUxmeec=
 github.com/docker/go-units v0.4.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk=

libdocker/client.go

@@ -48,7 +48,7 @@ type DockerClientInterface interface {
 	InspectContainerWithSize(id string) (*dockertypes.ContainerJSON, error)
 	CreateContainer(
 		dockertypes.ContainerCreateConfig,
-	) (*dockercontainer.ContainerCreateCreatedBody, error)
+	) (*dockercontainer.CreateResponse, error)
 	StartContainer(id string) error
 	StopContainer(id string, timeout time.Duration) error
 	UpdateContainerResources(id string, updateConfig dockercontainer.UpdateConfig) error

libdocker/fake_client.go

@@ -480,7 +480,7 @@ func GetFakeContainerID(name string) string {
 // It adds an entry "create" to the internal method call record.
 func (f *FakeDockerClient) CreateContainer(
 	c dockertypes.ContainerCreateConfig,
-) (*dockercontainer.ContainerCreateCreatedBody, error) {
+) (*dockercontainer.CreateResponse, error) {
 	f.Lock()
 	defer f.Unlock()
 	f.appendCalled(CalledDetail{name: "create"})
@@ -508,7 +508,7 @@ func (f *FakeDockerClient) CreateContainer(
 
 	f.normalSleep(100, 25, 25)
 
-	return &dockercontainer.ContainerCreateCreatedBody{ID: id}, nil
+	return &dockercontainer.CreateResponse{ID: id}, nil
 }
 
 // StartContainer is a test-spy implementation of DockerClientInterface.StartContainer.

libdocker/instrumented_client.go

@@ -93,7 +93,7 @@ func (in instrumentedInterface) InspectContainerWithSize(
 
 func (in instrumentedInterface) CreateContainer(
 	opts dockertypes.ContainerCreateConfig,
-) (*dockercontainer.ContainerCreateCreatedBody, error) {
+) (*dockercontainer.CreateResponse, error) {
 	const operation = "create_container"
 	defer recordOperation(operation, time.Now())
 

libdocker/kube_docker_client.go

@@ -142,7 +142,7 @@ func (d *kubeDockerClient) InspectContainerWithSize(id string) (*dockertypes.Con
 
 func (d *kubeDockerClient) CreateContainer(
 	opts dockertypes.ContainerCreateConfig,
-) (*dockercontainer.ContainerCreateCreatedBody, error) {
+) (*dockercontainer.CreateResponse, error) {
 	ctx, cancel := context.WithTimeout(context.Background(), d.timeout)
 	defer cancel()
 	// we provide an explicit default shm size as to not depend on docker daemon.