容器日志 | 南风未起｜云原生/Go技术分享博客

# 1.containerd

# 1.1.create

c.CreateContainer()会分配FIFO Pipe，初始化用于日志写出的stdin/stdout/stderr管道，后续该管道会绑定到container。

// CreateContainer creates a new container in the given PodSandbox.
func (c *criService) CreateContainer(ctx context.Context, r *runtime.CreateContainerRequest) (...) {
	...

    // 分配container FIFO Pipe
	containerIO, err := cio.NewContainerIO(id,
		cio.WithNewFIFOs(volatileContainerRootDir, config.GetTty(), config.GetStdin()))
	...
    
    // 初始化container meta数据存到boltdb
    cntr = c.client.NewContainer(ctx, id, opts...)
    ...
    
    // containerd container
	container, err := containerstore.NewContainer(meta,
		containerstore.WithStatus(status, containerRootDir),
		containerstore.WithContainer(cntr),
		containerstore.WithContainerIO(containerIO),
	)
	...

	// 缓存containerd container
	c.containerStore.Add(container)
    ...

	return &runtime.CreateContainerResponse{ContainerId: id}, nil
}

// NewContainerIO creates container io.
func NewContainerIO(id string, opts ...ContainerIOOpts) (_ *ContainerIO, err error) {
	...
    
    // 设置FIFO路径
	for _, opt := range opts {
		if err := opt(c); err != nil {
			return nil, err
		}
	}
	...
    
	// create actual fifos.
	stdio, closer, err := newStdioPipes(c.fifos)
	...
    
	c.stdioPipes = stdio
	c.closer = closer
	return c, nil
}

// newStdioPipes creates actual fifos for stdio.
func newStdioPipes(fifos *cio.FIFOSet) (_ *stdioPipes, _ *wgCloser, err error) {
	...

    // stdinPipe
	if fifos.Stdin != "" {
        // /run/containerd/io.containerd.grpc.v1.cri/containers/abc123/io/random-stdin
		f, err = openPipe(ctx, fifos.Stdin, syscall.O_WRONLY|syscall.O_CREAT|syscall.O_NONBLOCK, 0700)
        ...
		p.stdin = f
		set = append(set, f)
	}

    // stdoutPipe
	if fifos.Stdout != "" {
        // /run/containerd/io.containerd.grpc.v1.cri/containers/abc123/io/random-stdout
		f, err = openPipe(ctx, fifos.Stdout, syscall.O_RDONLY|syscall.O_CREAT|syscall.O_NONBLOCK, 0700)
        ...
		p.stdout = f
		set = append(set, f)
	}

    // stderrPipe
	if fifos.Stderr != "" {
        // // /run/containerd/io.containerd.grpc.v1.cri/containers/abc123/io/random-stderr
		f, err = openPipe(ctx, fifos.Stderr, syscall.O_RDONLY|syscall.O_CREAT|syscall.O_NONBLOCK, 0700)
        ...
		p.stderr = f
		set = append(set, f)
	}

	return p, &wgCloser{
		wg:     &sync.WaitGroup{},
		set:    set,
		ctx:    ctx,
		cancel: cancel,
	}, nil
}

注意

containerd FIFO Pipe路径会透传给shim-->runc，用于将read测绑定container Pipe的write侧

# 1.2.start

c.StartContainer()会初始化fileIO及logger，将FIFO Pipe-->logger-->file stdout/stderr，实现containerLog文件写入。

// StartContainer starts the container.
func (c *criService) StartContainer(ctx context.Context, r *runtime.StartContainerRequest) (...) {
	...
    
    // 获取containerd container
	cntr, err := c.containerStore.Get(r.GetContainerId())
	...

    // containerLog handler
	ioCreation := func(id string) (_ containerdio.IO, err error) {
        // 初始化file侧stdout/stderr
		stdoutWC, stderrWC, err := c.createContainerLoggers(meta.LogPath, config.GetTty())
        ...
        
        // stdout/stderr注册到stdGroup
		cntr.IO.AddOutput("log", stdoutWC, stderrWC)
        
        // containerd FIFO Pipe拷贝至stdout/stderr Group
		cntr.IO.Pipe()
		return cntr.IO, nil
	}
	...

	// container task
	task, err := cntr.Container.NewTask(ctx, ioCreation, taskOpts...)
	...

	// Start containerd task.
	task.Start(ctx)
	...

	return &runtime.StartContainerResponse{}, nil
}

// createContainerLoggers creates container loggers and return write closer for stdout and stderr.
func (c *criService) createContainerLoggers(logPath string, tty bool) (...) {
	if logPath != "" {
        // container log标准流(/var/log/pods/<ns>_<pod-name>_<pod-uid>/<container-name>/<restart-count>.log)
		f, err := openLogFile(logPath)
		...
		wc := cioutil.NewSerialWriteCloser(f)
        
        // stdout logger
		stdout, stdoutCh = cio.NewCRILogger(logPath, wc, cio.Stdout, c.config.MaxContainerLogLineSize)
        
		// Only redirect stderr when there is no tty.
        // linux tty模式stdout和stderr已经一起输出
		if !tty {
			stderr, stderrCh = cio.NewCRILogger(logPath, wc, cio.Stderr, c.config.MaxContainerLogLineSize)
		}
        
        // 退出关闭file
		go func() {
			if stdoutCh != nil {
				<-stdoutCh
			}
			if stderrCh != nil {
				<-stderrCh
			}
			f.Close()
		}()
	}
    ...
    
	return
}

// returns writeCloser which redirect container log into log file, and decorate line into CRI defined format.
func NewCRILogger(path string, w io.Writer, stream StreamType, maxLen int) (io.WriteCloser, <-chan struct{}) {
	// 初始化pipe对象
    prc, pwc := io.Pipe()
	...
    
	go func() {
        // 日志处理协程
		redirectLogs(path, prc, w, stream, maxLen)
		close(stop)
	}()
    
	return pwc, stop
}

// Pipe creates container fifos and pipe container output
// to output stream.
func (c *ContainerIO) Pipe() {
	...
	if c.stdout != nil {
		...
		go func() {
            // FIFO stdoutPipe-->file stdout
			io.Copy(c.stdoutGroup, c.stdout)
            ...
		}()
	}

	if !c.fifos.Terminal && c.stderr != nil {
		...
		go func() {
            // FIFO stderrPipe-->file stderr
			io.Copy(c.stderrGroup, c.stderr)
            ...
		}()
	}
}

注意

start阶段主要初始化file stdIO及logger，将FIFO Pipe与file stdIO绑定输出

# 1.3.task

c.NewTask()是容器启动的前置阶段，负责分配及激活container shim，基于shim执行runc create创建容器及绑定containerIO。

func (c *container) NewTask(ctx context.Context, ioCreate cio.Creator, opts ...NewTaskOpts) (_ Task, err error){
	// 执行ioCreation处理日志
    i, err := ioCreate(c.id)
	...
    
    // FIFO Pipe流路径获取
	cfg := i.Config()
	request := &tasks.CreateTaskRequest{
		ContainerID: c.id,
		Terminal:    cfg.Terminal,
        // containerd分配给container的Pipe流路径
		Stdin:       cfg.Stdin,
		Stdout:      cfg.Stdout,
		Stderr:      cfg.Stderr,
	}
	...
    // container创建数据绑定到request
    ...
    
	t := &task{
		client: c.client,
		io:     i,
		id:     c.id,
		c:      c,
	}
	...
    
    // grpc调用task模块创建container
	response, err := c.client.TaskService().Create(ctx, request)
	...
    
    // 记录container PID
	t.pid = response.Pid
	return t, nil
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

注意

这里比较明确，container创建请求会携带FIFO Pipe的IO路径

# 1.4.redirect

redirectLogs()会读取container pipe日志输出，将日志转为CRI格式写入到/var/log/containers/*.log，日志超出16k会进行切割。

func redirectLogs(path string, rc io.ReadCloser, w io.Writer, s StreamType, maxLen int) {
	...
    
    // Make sure bufSize <= maxLen(maxLen可配置,默认16K,最低4K)
	if maxLen > 0 && maxLen < bufSize {
		bufSize = maxLen
	}
    
    // lineReader(containerd向container分配的命名Pipe)
	r := bufio.NewReaderSize(rc, bufSize)
    
    // container吐的日志写入/var/log/xxx
	writeLineBuffer := func(tag []byte, lineBytes [][]byte) {
        // 追加当前时间
		timeBuffer = time.Now().AppendFormat(timeBuffer[:0], timestampFormat)
        // header-->时间-staout/stderr-P/F
        // tag由日志长度决定
		headers := [][]byte{timeBuffer, stream, tag}

        // 重置buf区
		lineBuffer.Reset()
        
        // header写入buf
		for _, h := range headers {
			lineBuffer.Write(h)
            // 时间 staout/stderr P/F
			lineBuffer.Write(delimiter)
		}
        
        // 日志写入buf
		for _, l := range lineBytes {
			lineBuffer.Write(l)
		}
        
        // 换行符
		lineBuffer.WriteByte(eol)
        
        // 写入日志文件
		lineBuffer.WriteTo(w)
        ...
	}
    
	for {
        // 基于换行读取一条日志
		newLine, isPrefix, err := readLine(r)
		// NOTE(random-liu): readLine can return actual content even if there is an error.
		if len(newLine) > 0 {
			...
			// Buffer returned by ReadLine will change after next read, copy it.
			l := make([]byte, len(newLine))
			copy(l, newLine)
			buf = append(buf, l)
			length += len(l)
		}
        
        // EOF
		if err != nil {
			...
            
            // No content left to write, break.
			if length == 0 {
				break
			}
            
			// Stop after writing the content left in buffer.
			stop = true
		}
        
        // 超出16k
		if maxLen > 0 && length > maxLen {
			exceedLen := length - maxLen
			last := buf[len(buf)-1]
			...
            
            // 超出16k的部分留到下一次写出
			buf[len(buf)-1] = last[:len(last)-exceedLen]
            // 时间 staout/stderr P content写到日志文件
			writeLineBuffer(partial, buf)
			...
            
            // 写出的部分清理
			buf = [][]byte{last[len(last)-exceedLen:]}
			length = exceedLen
		}
        
        // 日志行未读完
		if isPrefix {
			continue
		}
        
        // readLine报错
		if stop {
			// 时间 staout/stderr P content写到日志文件
			writeLineBuffer(partial, buf)
        // 日志行前面的部分已经读完,剩余部分未超出16k
		} else {
            // 时间 staout/stderr F content写到日志文件
			writeLineBuffer(full, buf)
		}
        
        // 重置buf/length
		buf = nil
		length = 0
        
        // readLine报错,最后内容写完就退出
		if stop {
			break
		}
	}
}

// bufio.ReadLine in golang eats both read errors and tailing newlines
// there is a newline at the end, for example:
// 1) When reading "CONTENT\n", it returns "CONTENT" without error;
// 2) When reading "CONTENT", it also returns "CONTENT" without error.
//
// To differentiate these 2 cases, we need to write a readLine function
// ourselves to not ignore the error.
func readLine(b *bufio.Reader) (line []byte, isPrefix bool, err error) {
    // bufio.ReadSlice读一行日志
	line, err = b.ReadSlice('\n')
    // 超出4k
	if err == bufio.ErrBufferFull {
		...
		return line, true, nil
	}

    // EOF
	if len(line) == 0 {
		if err != nil {
			line = nil
		}
		return
	}

    // 换行处理
	if line[len(line)-1] == '\n' {
		...
		line = line[:len(line)-drop]
	}
	return
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142

注意

containerLog限制16k，基于\n结尾读取一行，根据4k循环读，截断日志追加P标记，结尾日志标记F标记，供filebeat恢复

# 2.shim

# 2.1.create

local.Create()基于请求分配及激活container shim作为容器watcher，利用shim执行runc create创建容器及绑定containerIO。

func (l *local) Create(ctx context.Context, r *api.CreateTaskRequest, _ ...grpc.CallOption) (...) {
    // 加载boltdb container meta
	container, err := l.getContainer(ctx, r.ContainerID)
	...
	
    // shim请求
	opts := runtime.CreateOpts{
		Spec: container.Spec,
		IO: runtime.IO{
            // 这里就是containerd FIFO Pipe的流路径
			Stdin:    r.Stdin,
			Stdout:   r.Stdout,
			Stderr:   r.Stderr,
			Terminal: r.Terminal,
		},
		...
	}
	...
    
    // start shim及创建container
	c, err := rtime.Create(ctx, r.ContainerID, opts)
	...

	pid, err := c.PID(ctx)
	...
    
	return &api.CreateTaskResponse{
		ContainerID: r.ContainerID,
		Pid:         pid,
	}, nil
}

// Create launches new shim instance and creates new task
func (m *TaskManager) Create(ctx context.Context, taskID string, opts runtime.CreateOpts) (runtime.Task, error){
	// 分配container shim及启动
    process, err := m.manager.Start(ctx, taskID, opts)
	...

	// 基于shim创建容器
	shim := process.(*shimTask)
	t, err := shim.Create(ctx, opts)
	...

	return t, nil
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

问题

shimLog会基于Pipe吐到/run/containerd/io.containerd.runtime.v2.task/k8s.io/<container-id>/log供containerd读取

# 2.2.shimtask

shimTask.Create()本质执行runc命令创建容器，InitPID/shim socket均会写入上面的目录，runc create会重定向输出到Pipe。

func (s *shimTask) Create(ctx context.Context, opts runtime.CreateOpts) (runtime.Task, error) {
	...
	request := &task.CreateTaskRequest{
		ID:         s.ID(),
        // /run/containerd/io.containerd.runtime.v2.task/<namespace>/<container-id>
		Bundle:     s.bundle.Path,
        // containerd FIFO Pipe流路径
		Stdin:      opts.IO.Stdin,
		Stdout:     opts.IO.Stdout,
		Stderr:     opts.IO.Stderr,
		Terminal:   opts.IO.Terminal,
		Checkpoint: opts.Checkpoint,
		Options:    topts,
	}
	...

    // 基于shim调用runc创建container
	_, err := s.task.Create(ctx, request)
	...

	return s, nil
}

// Create a new initial process and container with the underlying OCI runtime
func (s *service) Create(ctx context.Context, r *taskAPI.CreateTaskRequest) (...) {
	...

    // runc create container
	container, err := runc.NewContainer(ctx, s.platform, r)
	...
	s.containers[r.ID] = container
	...

	// 获取container process
	proc, _ := container.Process("")
    // 更新container状态
	handleStarted(container, proc)

	return &taskAPI.CreateTaskResponse{
		Pid: uint32(container.Pid()),
	}, nil
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

注意

/run/containerd/io.containerd.runtime.v2.task/k8s.io/<container-id>是容器运行期间bundle数据存放地

# 2.3.newcntr

NewContainer()开支真正创建容器，基于bundle执行runc create及重定向containerIO至Pipe，实现容器启动日志采集和消费罗盘。

// NewContainer returns a new runc container
func NewContainer(ctx context.Context, platform stdio.Platform, r *task.CreateTaskRequest) (...) {
	...

	config := &process.CreateConfig{
		ID:               r.ID,
		Bundle:           r.Bundle,
		Runtime:          opts.BinaryName,
		Rootfs:           mounts,
		Terminal:         r.Terminal,
        // containerd FIFO Pipe流路径
		Stdin:            r.Stdin,
		Stdout:           r.Stdout,
		Stderr:           r.Stderr,
		...
	}
	...

    // 初始化process对象
	p, err := newInit(ctx, r.Bundle, filepath.Join(r.Bundle, "work"), ns, platform, config, &opts, rootfs)
	...
    // 执行runc create,containerIO绑定FIFO Pipe
	p.Create(ctx, config)
    ...
    
	container := &Container{
		ID:              r.ID,
		Bundle:          r.Bundle,
		process:         p,
		processes:       make(map[string]process.Process),
		reservedProcess: make(map[string]struct{}),
	}
	...    

	return container, nil
}

// Create the process with the provided config
func (p *Init) Create(ctx context.Context, r *CreateConfig) error {
	...

    // exec进程,tty模式
	if r.Terminal {
        // socketIO绑定,直接基于socket吐日志到远程
		socket = runc.NewTempConsoleSocket()
        ...
		defer socket.Close()
    // 普通进程
	} else {
        // 初始化shim Pipe,shim Pipe属主会调整,避免container stdIO无权限写入
		pio, err = createIO(ctx, p.id, p.IoUID, p.IoGID, p.stdio)
        ...
		p.io = pio
	}
	...
    
	opts := &runc.CreateOpts{
        // /run/containerd/io.containerd.runtime.v2.task/<namespace>/<container-id>/init.id
        // 用于保存container initPID
		PidFile:      pidFile.Path(),
		NoPivot:      p.NoPivotRoot,
		NoNewKeyring: p.NoNewKeyring,
	}
    
    // 设置容器IO
	if p.io != nil {
		opts.IO = p.io.IO()
	}
    
    // 设置socketIO,先不管
	if socket != nil {
		opts.ConsoleSocket = socket
	}
    
    // runc create --bundle <bundle-path> <container-id>...
    // containerIO-->shim Pipe
	p.runtime.Create(ctx, r.ID, r.Bundle, opts)
    ...
    
	if r.Stdin != "" {
		p.openStdin(r.Stdin)
        ...
	}
	...
    
    // socketIO
	if socket != nil {
		console, err := socket.ReceiveMaster()
		...
        
        // socketIO<--->remoteIO
		console, err = p.Platform.CopyConsole(ctx, console, p.id, r.Stdin, r.Stdout, r.Stderr, &p.wg)
		...
		p.console = console
	} else {
        //shim Pipe--->contained Pipe IO
		pio.Copy(ctx, &p.wg)
        ...
	}
    
	pid, err := pidFile.Read()
	...
    
	p.pid = pid
	return nil
}

注意

run create输出会重定向到shim Pipe，由shimPipe转到containerdPipe，实现日志类型直达containerd效果

# 2.4.createio

这里的createIO创建的是shim Pipe，相应的stdout/stderr Pipe会调整uid/gid，确保container重定向的输出有权写入Pipe管道。

func createIO(ctx context.Context, id string, ioUID, ioGID int, stdio stdio.Stdio) (*processIO, error) {
	pio := &processIO{
		stdio: stdio,
	}
    
    // IO设置为/dev/null
	if stdio.IsNull() {
		i, err := runc.NewNullIO()
		...
		pio.io = i
		return pio, nil
	}
    
    // stdout流
	u, err := url.Parse(stdio.Stdout)
	...
    
    // 未设置协议默认为FIFO Pipe
	if u.Scheme == "" {
		u.Scheme = "fifo"
	}
    
	pio.uri = u
	switch u.Scheme {
    // FIFO Pipe
	case "fifo":
		pio.copy = true
		pio.io, err = runc.NewPipeIO(ioUID, ioGID, withConditionalIO(stdio))
	...
	}
	...
    
    // 现在pio握着containerPipe可以获取到容器日志,握着containerdPipe可以转发日志
	return pio, nil
}

// NewPipeIO creates pipe pairs to be used with runc
func NewPipeIO(uid, gid int, opts ...IOOpt) (i IO, err error) {
	...
	
    // stdin
	if option.OpenStdin {
		stdin, err = newPipe()
        ...
        
		pipes = append(pipes, stdin)
        // stdin属主设置为container(权限)
		unix.Fchown(int(stdin.r.Fd()), uid, gid)
        ...
	}
    
    // stdout
	if option.OpenStdout {
		stdout, err = newPipe()
        ...
		pipes = append(pipes, stdout)
        // stdout属主设置为container(权限)
		iunix.Fchown(int(stdout.w.Fd()), uid, gid)
        ...
	}
    
    // stderr
	if option.OpenStderr {
		stderr, err = newPipe()
        ...
        
		pipes = append(pipes, stderr)
        // stderr属主设置为container(权限)
		unix.Fchown(int(stderr.w.Fd()), uid, gid)
        ...
	}
    
	return &pipeIO{
		in:  stdin,
		out: stdout,
		err: stderr,
	}, nil
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78

注意

shimIO支持FIFO/Binary/File三种类型，这里只关注FIFO类型，其它原理类同

# 2.5.runc

runc.Create()本质就是执行runc create --bundle...命令，未设置stdIO的命令输出不会重定向，否则会重定向输出到shim Pipe。

// Create creates a new container and returns its pid if it was created successfully
func (r *Runc) Create(context context.Context, id, bundle string, opts *CreateOpts) error {
	...
    
	cmd := r.command(context, append(args, id)...)
    // 设置container stdIO
    if opts != nil && opts.IO != nil {
		opts.Set(cmd)
	}
	...
	cmd.ExtraFiles = opts.ExtraFiles

    // /dev/null类型输出
	if cmd.Stdout == nil && cmd.Stderr == nil {
        // 直接创建container
		data, err := cmdOutput(cmd, true, nil)
		...
        
		return nil
	}
    
    // 执行cmd,watch容器退出
	ec, err := Monitor.Start(cmd)
	...
	status, err := Monitor.Wait(cmd, ec)
	...
    
	return err
}

// Set sets the io to the exec.Cmd
func (i *pipeIO) Set(cmd *exec.Cmd) {
	if i.in != nil {
		cmd.Stdin = i.in.r
	}
	if i.out != nil {
		cmd.Stdout = i.out.w
	}
	if i.err != nil {
		cmd.Stderr = i.err.w
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

注意

到此为止，containerLog就经历了stdIO-->shimPipe-->containerdPipe-->logger.chan-->logFile采集过程

# 3.kubelet

# 2.1.manager

containerLogManager用于管理容器日志，负责压缩及轮转shim采出的容器日志，以及Pod驱逐和容器回收时清理未使用日志和移除多余日志。

// ContainerLogManager manages lifecycle of all container logs.
type ContainerLogManager interface {
	// Start container log manager.
	Start()
	// Clean removes all logs of specified container.
	Clean(containerID string) error
}

type containerLogManager struct {
    // 运行时
	runtimeService internalapi.RuntimeService
    // os文件操作接口
	osInterface    kubecontainer.OSInterface
    // 日志轮转策略
	policy         LogRotatePolicy
	clock          clock.Clock
	mutex          sync.Mutex
}

// NewContainerLogManager creates a new container log manager.
func NewContainerLogManager(...) (ContainerLogManager, error) {
	...
	// policy LogRotatePolicy
	return &containerLogManager{
		osInterface:    osInterface,
		runtimeService: runtimeService,
		policy: LogRotatePolicy{
            // 日志大小(10M)
			MaxSize:  parsedMaxSize,
            // 日志数量(5)
			MaxFiles: maxFiles,
		},
		clock: clock.RealClock{},
		mutex: sync.Mutex{},
	}, nil
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

注意

1.containerLogManager默认最大保留5个日志文件

2.containerLogManager默认日志文件最大10M

# 2.2.clean

containerLogManager清理日志前调用CRI获取运行时containerStatus，基于containerStatus数据获取日志文件路径执行日志清理。

// Clean removes all logs of specified container (including rotated one).
func (c *containerLogManager) Clean(containerID string) error {
	c.mutex.Lock()
	defer c.mutex.Unlock()
    // 获取运行时容器状态
	resp, err := c.runtimeService.ContainerStatus(containerID, false)
	...
	// 构建日志目录
	pattern := fmt.Sprintf("%s*", resp.GetStatus().GetLogPath())
    // 获取目录下所有日志文件
	logs, err := c.osInterface.Glob(pattern)
	...
    // 依次清理
	for _, l := range logs {
		c.osInterface.Remove(l)
        ...
	}

	return nil
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

注意

1.container状态会记录容器日志目录，containerLogManager基于此路径获取所有日志文件

2.containerGC回收容器时调用日志清理

# 2.3.rotate

containerLogManager.Start()激活后台协程间隔10s不停执行rotateLogs()，以实现日志压缩及轮转，控制日志文件大小及数量。

// Start the container log manager.
func (c *containerLogManager) Start() {
	// Start a goroutine periodically does container log rotation.
    // 10s触发依次
	go wait.Forever(func() {
		c.rotateLogs()
        ...
	}, logMonitorPeriod)
}

// 日志轮转,防止日志文件膨胀
func (c *containerLogManager) rotateLogs() error {
	c.mutex.Lock()
	defer c.mutex.Unlock()
	// 调用运行时获取所有容器
	containers, err := c.runtimeService.ListContainers(&runtimeapi.ContainerFilter{})
	...
	// NOTE(random-liu): Figure out whether we need to rotate container logs in parallel.
	for _, container := range containers {
		// Only rotate logs for running containers. Non-running containers won't
		// generate new output, it doesn't make sense to keep an empty latest log.
		if container.GetState() != runtimeapi.ContainerState_CONTAINER_RUNNING {
			continue
		}
		id := container.GetId()
		// 获取容器运行时状态
		resp, err := c.runtimeService.ContainerStatus(id, false)
		...
		// 获取容器日志路径
		path := resp.GetStatus().GetLogPath()
        // 获取日志文件信息
		info, err := c.osInterface.Stat(path)
		if err != nil {
            // 日志文件不存在
			if !os.IsNotExist(err) {
				continue
			}
            // 重新生成stdout和stderr logger,向日志文件写入CRI规范日志(旧的关闭)
			if err := c.runtimeService.ReopenContainerLog(id); err != nil {
				continue
			}
			// 获取日志文件信息.
			info, err = c.osInterface.Stat(path)
			if err != nil {
				continue
			}
		}
        // 日志大小未超出10M
		if info.Size() < c.policy.MaxSize {
			continue
		}
		// 超出,开始轮转日志
		c.rotateLog(id, path)
        ...
	}
	return nil
}

func (c *containerLogManager) rotateLog(id, log string) error {
    // 匹配容器的已轮转日志文件(restartcount.log.xxx)
	pattern := fmt.Sprintf("%s.*", log)
	logs, err := filepath.Glob(pattern)
	...

    // 清理未使用日志(临时文件/已压缩文件)
	logs, err = c.cleanupUnusedLogs(logs)
	...

    // 清理超出保留数量日志
	logs, err = c.removeExcessLogs(logs)
	...

    // 压缩未压缩日志
	for _, l := range logs {
        // 跳过已压缩的
		if strings.HasSuffix(l, compressSuffix) {
			continue
		}
        // 压缩日志文件
		c.compressLog(l)
        ...
	}

    // 轮转日志
	c.rotateLatestLog(id, log)
    ...

	return nil
}

// removeExcessLogs removes old logs to make sure there are only at most MaxFiles log files.
func (c *containerLogManager) removeExcessLogs(logs []string) ([]string, error) {
	// 日志文件由旧到新排序
	sort.Strings(logs)
	// 计算可保留的已轮转日志
    // 1份是正在写的当前日志
    // 1份是刚要切分的轮转日志
    // 其余的才是可保留的已轮转日志
	maxRotatedFiles := c.policy.MaxFiles - 2
	if maxRotatedFiles < 0 {
		maxRotatedFiles = 0
	}
	i := 0
    // 依次删除最旧的日志
	for ; i < len(logs)-maxRotatedFiles; i++ {
		c.osInterface.Remove(logs[i])
        ...
	}
	logs = logs[i:]
	return logs, nil
}

// compressLog compresses a log to log.gz with gzip.
func (c *containerLogManager) compressLog(log string) error {
    // 只读方式打开日志文件
	r, err := c.osInterface.Open(log)
	...
    
	defer r.Close()
    
    // 只写模式创建临时文件
	tmpLog := log + tmpSuffix
	f, err := c.osInterface.OpenFile(tmpLog, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644)
	...
    
	defer func() {
		// 退出时清理临时文件
		c.osInterface.Remove(tmpLog)
	}()
	defer f.Close()
    
    // 创建临时文件压缩流
	w := gzip.NewWriter(f)
	defer w.Close()
    // 流拷贝(原本日志-->压缩临时流)
	io.Copy(w, r)
    ...
	// The archive needs to be closed before renaming, otherwise an error will occur on Windows.
	w.Close()
	f.Close()
    // 压缩日志文件名
	compressedLog := log + compressSuffix
	// 压缩的临时文件重命名
    c.osInterface.Rename(tmpLog, compressedLog)
    ...
	// Remove old log file.
	r.Close()
	c.osInterface.Remove(log)
    ...
	return nil
}

// rotateLatestLog rotates latest log without compression, so that container can still write
// and fluentd can finish reading.
func (c *containerLogManager) rotateLatestLog(id, log string) error 
	// 生成轮转日志名称
	timestamp := c.clock.Now().Format(timestampFormat)
	rotated := fmt.Sprintf("%s.%s", log, timestamp)
	// 旧日志改名
	c.osInterface.Rename(log, rotated)
	...

	// 调用CRI重置日志文件stdout和stderr句柄
	c.runtimeService.ReopenContainerLog(id)
	...
	return nil
}

注意

1.containerLogManager会清理已压缩的轮转日志

2.containerLogManager会限制容器日志数量最大5个

3.containerLogManager会轮转超出10M的日志文件

4.containerLogManager的日志轮转就是替换containerd的日志文件标准流

5.containerd创建容器时初始化命名管道和日志写出器，runc启动的容器标准流指向命令管道，containerd拷贝管道输出到日志文件