docker1.12-containerd源码分析
从原openstack转型至docker已有一段时间。更稳定的使用docker了解docker的各流程，从源代码层面了解下containerd。本文基于docker 1.12版本，从1.11开始docker已拆分docker daemon containerd源码流程图 源码接口调用详情 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(handleTaskfunc) 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)ResumePauseUpdateResources--> exec .Command(c.runtime,arg....) start-->(runtimesupervisor /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) gow.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 funcNewWorker(s*Supervisor,wg* sync .WaitGroup)Worker{ return &worker{ s:s, wg:wg, } } func(w*worker)Start(){ deferw.wg.Done() for t:=rangew.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:startcontainer" ) 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:supervisorrunning" ) gofunc(){ for i:=ranges.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 varstartCommand=cli.Command{ Name: "start" , Usage: "startacontainer" , ArgsUsage:"IDBundlePath”,————…... 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" ){ gofunc(){ 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) gofunc(){ s:= make (chanos.Signal,64) signal.Notify(s,syscall.SIGWINCH) for ranges{ 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(ctxcontext.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(ctxcontext.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(ctxcontext.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(srvinterface{},streamgrpc.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(srvinterface{},ctxcontext.Context,decfunc(interface{})error,interceptorgrpc.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(ctxcontext.Context,reqinterface{})(interface{},error){ return srv.(APIServer).CreateContainer(ctx,req.(*CreateContainerRequest)) } return interceptor(ctx, in ,info,handler) } func_API_UpdateProcess_Handler(srvinterface{},ctxcontext.Context,decfunc(interface{})error,interceptorgrpc.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(ctxcontext.Context,reqinterface{})(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,streamtypes.API_EventsServer)error{ events:=s.sv.Events(t,r.StoredOnly,r.Id) func(s*apiServer)CreateContainer(ctxcontext.Context,c*types.CreateContainerRequest)(*types.CreateContainerResponse,error){ s.sv.SendTask(e) apiC,err:=createAPIContainer(r.Container, false ) func(s*apiServer)UpdateProcess(ctxcontext.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(){ deferw.wg.Done() for t:=rangew.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(checkpointPathstring,sStdio)(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 funcstart(log*os.File)error{ p,err:=newProcess(flag.Arg(0),flag.Arg(1),flag.Arg(2)) if err!=nil{ return err } deferfunc(){ 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
