首页 文章 精选 留言 我的

docker1.12-containerd源码分析

2017-11-13 695

     从原openstack转型至docker已有一段时间。更稳定的使用docker了解docker的各流程，从源代码层面了解下containerd。本文基于docker 1.12版本，从1.11开始docker已拆分docker daemon

containerd源码流程图

wKioL1iB3_XRqLZuAAJ80zXVg4o002.png

源码接口调用详情

A)第一步从ctr入口至API接口

checkpoint（用于快照，docker目前该功能不完善）

1
2
3
list -->  /types .API /ListCheckpoint 
create -->  /types .API /CreateCheckpoint 
delete -->  /types .API /DeleteCheckpoint


containers

1
2
3
4
5
6
7
8
9
list、state --> /types .API /State
pause、resume、update --> /types .API /UpdateContainer
create -->  /types .API /CreateContainer
stats -->  /types .API /Stats
watch  --> /types .API /State  、 /types .API /Events
exec  --> /types .API /Events  、 /types .API /AddProcess  、 /types .API /UpdateProcess
kill  --> /types .API /Signal
start --> /types .API /Events  、  /types .API /CreateContainer  、 /types .API /UpdateProcess
update --> /types .API /UpdateContainer

events

1
/types .API /Events

state

1
/types .API /State

version

1
/types .API /GetServerVersion  -- return  result

B)第二步从API接口至supervisor任务处理

1
注：API--server.go --> daemon – supervisor.go(handleTask func)

checkpoint

1
2
3
/types .API /ListCheckpoint  (supervisor.GetContainersTask)--> getContainers 
/types .API /CreateCheckpoint  --> createCheckpoint
/types .API /DeleteCheckpoint  --> deleteCheckpoint

containers

1
2
3
4
5
6
7
/types .API /State  /types .API /Stats  (supervisor.GetContainersTask)--> getContainers 
/types .API /UpdateContainer  (supervisor.UpdateTask)-->updateContainer
/types .API /CreateContainer  (supervisor.StartTask)-->start
/types .API /Events  --> Events -- return  result
/types .API /AddProcess  -->addProcess
/types .API /UpdateProcess  -->updateProcess
/types .API /Signal  -->signal

 

C)第三步从任务队列至runtime至runc

checkpoint

1
2
3
getContainers --  return  result
createCheckpoint -->(runtime)CheckPoint --> exec .Command(c.runtime,arg....)
deleteCheckpoint -->(runtime)DeleteCheckpoint --  return  result

containers

1
2
3
4
5
6
getContainers --  return  result
updateContainer -->(runtime)Resume Pause UpdateResources--> exec .Command(c.runtime,arg....)
start -->(runtime supervisor /worker .go) Start --> exec .Command(c.shim,c. id ,c.bundle,c.runtime)
addProcess -->(runtime)  exec  -->  exec .Command(c.shim,c. id ,c.bundle,c.runtime)
updateProcess --> return  result
signal --> return  result


createContainer示例

deamon启动监听tasks及startTasks进程

a)进入main.go main方法调用daemon方法 

1
2
3
4
5
6
7
app.Action = func(context *cli.Context) {
        if  err := daemon(context); err != nil {
               logrus.Fatal(err)
        }
  
  
}

b)进入main.go daemon方法

1
2
3
4
5
6
7
8
for  i := 0; i < 10; i++ {
        wg.Add(1)
        w := supervisor.NewWorker(sv, wg)
        go w.Start()
}
if  err := sv.Start(); err != nil {
        return  err
}

c)初始化supervisor/worker.go NewWorker并启动监听startTask并处理

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
func NewWorker(s *Supervisor, wg * sync .WaitGroup) Worker {
        return  &worker{
               s:  s,
               wg: wg,
        }
}
  
  
  
func (w *worker) Start() {
        defer w.wg.Done()
        for  t := range w.s.startTasks {
               started :=  time .Now()
               process, err := t.Container.Start(t.CheckpointPath, runtime.NewStdio(t.Stdin, t.Stdout, t.Stderr))
               if  err != nil {
                      logrus.WithFields(logrus.Fields{
                             "error" : err,
                             "id" :    t.Container.ID(),
                      }).Error( "containerd: start container" )
                      t.Err <- err
                      evt := &DeleteTask{
                             ID:      t.Container.ID(),
                             NoEvent:  true ,
                             Process: process,
                      }
                      w.s.SendTask(evt)
                      continue
               }

d)启动supervisor/supervisor.go task监听task并处理

1
2
3
4
5
6
7
8
9
10
11
12
13
func (s *Supervisor) Start() error {
        logrus.WithFields(logrus.Fields{
               "stateDir" :    s.stateDir,
               "runtime" :     s.runtime,
               "runtimeArgs" : s.runtimeArgs,
               "memory" :      s.machine.Memory,
               "cpus" :        s.machine.Cpus,
        }).Debug( "containerd: supervisor running" )
        go func() {
               for  i := range s.tasks {
                      s.handleTask(i)
  
               }

containers容器创建示例

Ctl控制台命令入口

ctr/main.go  containersCommand 

1
2
3
4
5
6
7
8
9
10
11
execCommand,
killCommand,
listCommand,
pauseCommand,
resumeCommand,
startCommand,
stateCommand,
statsCommand,
watchCommand,
  
updateCommand,

ctr/container.go

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
var startCommand = cli.Command{
        Name:       "start" ,
        Usage:      "start a container" ,
        ArgsUsage: "ID BundlePath”, ————…...
  
  
  
         events, err := c.Events(netcontext.Background(), &types.EventsRequest{})/*事件创建*/
        if  err != nil {
               fatal(err.Error(), 1)
        }
        if  _, err := c.CreateContainer(netcontext.Background(), r); err != nil {/*容器创建*/
               fatal(err.Error(), 1)
        }
        if  context.Bool( "attach" ) {
               go func() {
                      io.Copy(stdin, os.Stdin)
                      if  _, err := c.UpdateProcess(netcontext.Background(), &types.UpdateProcessRequest{/*更新进程*/
                             Id:          id ,
                             Pid:         "init" ,
                             CloseStdin:  true ,
                      }); err != nil {
                             fatal(err.Error(), 1)
                      }
                      restoreAndCloseStdin()
               }()
               if  tty  {
                      resize( id ,  "init" , c)
                      go func() {
                             s :=  make (chan os.Signal, 64)
                             signal.Notify(s, syscall.SIGWINCH)
                             for  range s {
                                    if  err := resize( id ,  "init" , c); err != nil {
                                           log.Println(err)
                                    }
                             }
                      }()
               }
               waitForExit(c, events,  id ,  "init" , restoreAndCloseStdin)
        }
  
},

api处理

api/grpc/types/api.pb.go 

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
func (c *aPIClient) Events(ctx context.Context,  in  *EventsRequest, opts ...grpc.CallOption) (API_EventsClient, error) {
        stream, err := grpc.NewClientStream(ctx, &_API_serviceDesc.Streams[0], c.cc,  "/types.API/Events" , opts...)
        if  err != nil {
               return  nil, err
        }
        x := &aPIEventsClient{stream}
        if  err := x.ClientStream.SendMsg( in ); err != nil {
               return  nil, err
        }
        if  err := x.ClientStream.CloseSend(); err != nil {
               return  nil, err
        }
        return  x, nil
}
  
  
  
func (c *aPIClient) CreateContainer(ctx context.Context,  in  *CreateContainerRequest, opts ...grpc.CallOption) (*CreateContainerResponse, error) {
        out := new(CreateContainerResponse)
        err := grpc.Invoke(ctx,  "/types.API/CreateContainer" ,  in , out, c.cc, opts...)
        if  err != nil {
               return  nil, err
        }
        return  out, nil
}
  
  
  
func (c *aPIClient) UpdateProcess(ctx context.Context,  in  *UpdateProcessRequest, opts ...grpc.CallOption) (*UpdateProcessResponse, error) {
        out := new(UpdateProcessResponse)
        err := grpc.Invoke(ctx,  "/types.API/UpdateProcess" ,  in , out, c.cc, opts...)
        if  err != nil {
               return  nil, err
        }
        return  out, nil
}

api/grpc/types/api.pb.go 

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
func _API_Events_Handler(srv interface{}, stream grpc.ServerStream) error {
        m := new(EventsRequest)
        if  err := stream.RecvMsg(m); err != nil {
               return  err
        }
        return  srv.(APIServer).Events(m, &aPIEventsServer{stream})
}
  
  
func _API_CreateContainer_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
        in  := new(CreateContainerRequest)
        if  err := dec( in ); err != nil {
               return  nil, err
        }
        if  interceptor == nil {
               return  srv.(APIServer).CreateContainer(ctx,  in )
        }
        info := &grpc.UnaryServerInfo{
               Server:     srv,
               FullMethod:  "/types.API/CreateContainer" ,
        }
        handler := func(ctx context.Context, req interface{}) (interface{}, error) {
               return  srv.(APIServer).CreateContainer(ctx, req.(*CreateContainerRequest))
        }
        return  interceptor(ctx,  in , info, handler)
}
  
  
  
func _API_UpdateProcess_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
        in  := new(UpdateProcessRequest)
        if  err := dec( in ); err != nil {
               return  nil, err
        }
        if  interceptor == nil {
               return  srv.(APIServer).UpdateProcess(ctx,  in )
        }
        info := &grpc.UnaryServerInfo{
               Server:     srv,
               FullMethod:  "/types.API/UpdateProcess" ,
        }
        handler := func(ctx context.Context, req interface{}) (interface{}, error) {
               return  srv.(APIServer).UpdateProcess(ctx, req.(*UpdateProcessRequest))
        }
        return  interceptor(ctx,  in , info, handler)

api/grpc/server/server.go 进入第一步中的tasks及sendTasks处理队列

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
func (s *apiServer) Events(r *types.EventsRequest, stream types.API_EventsServer) error {
  
events := s.sv.Events(t, r.StoredOnly, r.Id) 
  
  
  
func (s *apiServer) CreateContainer(ctx context.Context, c *types.CreateContainerRequest) (*types.CreateContainerResponse, error) {
  
s.sv.SendTask(e)
  
apiC, err := createAPIContainer(r.Container,  false ) 
  
  
  
func (s *apiServer) UpdateProcess(ctx context.Context, r *types.UpdateProcessRequest) (*types.UpdateProcessResponse, error) {
        e := &supervisor.UpdateProcessTask{}
        e.ID = r.Id
        e.PID = r.Pid
        e.Height = int(r.Height)
        e.Width = int(r.Width)
        e.CloseStdin = r.CloseStdin
        s.sv.SendTask(e)
        if  err := <-e.ErrorCh(); err != nil {
               return  nil, err
        }
        return  &types.UpdateProcessResponse{}, nil
}

supervisor/create.go

1
2
func (s *Supervisor) start(t *StartTask) error {
s.startTasks <- task

supervisor/worker.go

1
2
3
func (w *worker) Start() {
        defer w.wg.Done()
        for  t := range w.s.startTasks {

runtime/container.go

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
func (c *container) Start(checkpointPath string, s Stdio) (Process, error) {
        processRoot := filepath.Join(c.root, c. id , InitProcessID)
        if  err := os.Mkdir(processRoot, 0755); err != nil {
               return  nil, err
        }
        cmd :=  exec .Command(c.shim,
               c. id , c.bundle, c.runtime,
  
        ) ---执行 docker-containerd-shim命令
  
        cmd.Dir = processRoot
        cmd.SysProcAttr = &syscall.SysProcAttr{
               Setpgid:  true ,
        }
        spec, err := c.readSpec()
        if  err != nil {
               return  nil, err
        }
        config := &processConfig{
               checkpoint:  checkpointPath,
               root:        processRoot,
               id :          InitProcessID,
               c:           c,
               stdio:       s,
               spec:        spec,
               processSpec: specs.ProcessSpec(spec.Process),
        }
        p, err := newProcess(config)
        if  err != nil {
               return  nil, err
        }
        if  err := c.createCmd(InitProcessID, cmd, p); err != nil {
               return  nil, err
        }
        return  p, nil
}

containerd-shim接收后处理

containerd-shim/main.go 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
func start(log *os.File) error {
  
  
  
p, err := newProcess(flag.Arg(0), flag.Arg(1), flag.Arg(2))
if  err != nil {
        return  err
}
defer func() {
        if  err := p.Close(); err != nil {
               writeMessage(log,  "warn" , err)
        }
}()
if  err := p.create(); err != nil {
        p.delete()
        return  err
}

containerd-shim/process.go跳转执行runc命令 

1
2
3
func (p *process) create() error {
  
cmd :=  exec .Command(p.runtime, args...)


本文转自 qwjhq 51CTO博客，原文链接:http://blog.51cto.com/bingdian/1893470
上一篇 下一篇
优秀的个人博客,低调大师

微信关注我们

原文链接：https://yq.aliyun.com/articles/432476

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

openstack下service和endpoint

service_id=$(keystone --token ADMIN --endpoint http://localhost:35357/v2.0/ service-create --name=glance --type=image --description="Glance Image Service" | grep '^| id' | awk '{print $4}') keystone --token ADMIN --endpoint http://localhost:35357/v2.0/ endpoint-create --region $REGION --service_id=$service_id --publicurl=http://$g_p_ip:9292/v1 --internalurl=http://$g_i_ip:9292/v1 --adminurl=http://$g_a_ip:9292/v1 本文转自 jxwpx 51CTO博客，原文链接：http://blog.51cto.com/jxwpx/897496，如需转载请自行联系原作者
2017-11-14
590
上一篇

OpenStack 系列之Cinder multi backend

OpenStack-Cinder multi backend 补充内容： 1.修改 crushmap 有两种方式：在线修改和离线修改 2.为了保险起见一般都是采用离线修改，也就是导出来修改 3.导出默认的 crushmap，这个是二进制文件打不开 ceph osd getcrushmap -o {compiled-crushmap-filename} 4.将刚才的二进制文件转换成可视化的文本文件 crushtool -d {compiled-crushmap-filename} -o {decompiled-crushmap-filename} 5.转换成可视化的文本文件之后，建议不要把默认的二进制文件删掉避免修改 crushmap 造成 ceph 集群瘫痪，这样我们还留了一个备份 6.修改 crushmap，也就是下面讲述的 crushmap 详解，根据实际情况修改 7.将刚才修改的可视化文本文件转换成二进制文件 crushtool -c {decompiled-crush-map-filename} -o {compiled-crush-map-filename} 8.设置 OSD...
2017-11-14
540
下一篇

相关文章

发表评论

资源下载

更多资源
优质分享App

优质分享App

近一个月的开发和优化，本站点的第一个app全新上线。该app采用极致压缩，本体才4.36MB。系统里面做了大量数据访问、缓存优化。方便用户在手机上查看文章。后续会推出HarmonyOS的适配版本。

时间: 2025-12-03 大小: 4.36MB 下载: 23次
Mario

Mario

马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长，特征是大鼻子、头戴帽子、身穿背带裤，还留着胡子。与他的双胞胎兄弟路易基一起，长年担任任天堂的招牌角色。

时间: 2025-12-12 大小: 211.28KB 下载: 54次
Rocky Linux

Rocky Linux

Rocky Linux（中文名：洛基）是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版，作为CentOS稳定版停止维护后与RHEL（Red Hat Enterprise Linux）完全兼容的开源替代方案，由社区拥有并管理，支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性，采用模块化包装和SELinux安全架构，默认包含GNOME桌面环境及XFS文件系统，支持十年生命周期更新。

时间: 2026-01-03 大小: 1.42GB 下载: 2次
Sublime Text

Sublime Text

Sublime Text具有漂亮的用户界面和强大的功能，例如代码缩略图，Python的插件，代码段等。还可自定义键绑定，菜单和工具栏。Sublime Text 的主要功能包括：拼写检查，书签，完整的 Python API ， Goto 功能，即时项目切换，多选择，多窗口等等。Sublime Text 是一个跨平台的编辑器，同时支持Windows、Linux、Mac OS X等操作系统。

时间: 2025-12-03 大小: 9.84MB 下载: 39次
手机查看

扫码在手机上查看文章

扫描二维码，手机阅读更方便

热门标签
DeepSeek HarmonyOS6 Jdk25 Redis OpenAi SpringBoot4 Service Mesh Docker Kubernetes Nacos3 Gemini3 Rocky SpringCloud

欢迎您来访！

登录后，您将获得更多功能！
热门文章
联系我们

有任何问题或合作意向欢迎联系我们

Email: 99873273@qq.com

QQ: 99873273

用户登录
用户注册