当前位置: 首页 > 工具软件 > runv > 使用案例 >

runv containerd 流程分析

空夕
2023-12-01

当runv需要启动一个容器的时候,首先需要启动containrd,作为该容器的daemon。因此,启动containerd的相关代码也是从runv/start.go开始。最终,启动containerd的命令行参数如下所示:

runv --kernel /var/lib/hyper/kernel --initrd /var/lib/hyper/hyper-initrd.img --default_cpus 1 --default_memory 128 containerd 
--solo-namespaced --containerd-dir /run/runv-namespace-457712864 --state-dir /run/runv --listen /run/runv-namespace-457712864/namespaced.sock

  

1、runv/containerd/containerd.go

Action: func(context *cli.Context)

(1)、对各个flag进行加载,包括driver, kernel, initrd,template,stateDir等等,上面的命令行都描述得很清晰了

(2)、调用hypervisor.HDriver, err = driverloader.Probe(driver)加载hypervisor的驱动,一般为qemu

(3)、调用f = factory.NewFromConfigs(kernel, initrd, nil)获得一个工厂实例

(4)、调用sv, err := supervisor.New(stateDir, containerdDir, f, context.GlobalInt("default_cpus"), context.GlobalInt("default_memory")),获得一个supervisor实例

(5)、如果指定了solo-namespaced参数,则调用go namespaceShare(sv, containerdDir, stateDir)

(6)、创建daemon,调用daemon(sv, context.String("listen"))

(7)、如果指定了solo-namespaced,则进行清除工作,os.RemoveAll(containerdDir)

 

2、runv/driverloader_linux.go

func Probe(driver string) (hypervisor.HypervisorDriver, error)

该函数针对不同的driver做初始化操作,这里我们只对driver为qemu的情况进行讨论,因此直接调用qd := qemu.InitDriver(),并且return qd。而qemu的InitDriver的操作很简单,仅仅是找到qemu的绝对路径,然后return &QemuDriver{executable: cmd.,}。

 

3、runv/factory/factory.go

func NewFromConfigs(kernel, initrd string, configs []FactoryConfig) Factory

因为在初始化的过程中configs为nil,因此仅仅只是调用return single.New(direct.New(1000000, 1000000, kernel, initrd))。其中direct.New返回了一个base.Factory,首先构建一个hypervisor.BootConfig的实例,b := hypervisor.BootConfig{CPU: cpu, Memory: mem, HotAddCpuMem: true, Kernel: kernel, Initrd: initrd}, 在包装一下b,return &directFactory{config: b}。而single.New(b base.Factory)仅仅只是返回 Factory{Factory: b}

 

Supervisor结构如下所示:

type Supervisor struct {

  StateDir    string
  Factory    factory.Factory
  defaultCpus  int
  defaultMemory int

  Events    SvEvents

  sync.RWMutex

  containers  map[string]*Container

}

  

 

4、runv/supervisor/supervisor.go

func New(stateDir, eventLogDir string, f factory.Factory, defaultCpus int , defaultMemory int) (*Supervisor, error)

首先创建stateDir目录和eventLogDir目录,接着填充数据结构Supervisor, sv := &Supervisor{...}。创建sv.Events.subscribers = make(map[chan Event]struct{})。接着go sv.reaper(),最后,return sv, sv.Events.setupEventLog(eventLogDir) // eventLogDir其实就是containerdDir。例如,/run/runv-namespace-457712864

 

5、runv/containerd/containerd.go

func namaspaceShare(sv *supervisor.Supervisor, namespace, state string) 

该函数主要通过从events := sv.Events.Events(time.Time{})接收来的事件来对容器进行计数

(1)、当接收到的事件e的Type为EventContainerStart时,调用os.Symlink(namespace, filepath.Join(state, e.Id, "namespace")),链接两个容器的namespace,最后,containerCount++

(2)、当接收到的事件e的Type为EventExit并且e.Pid为init时,containerCount--,当containerCount为0时,调用syscall.Kill(0, syscall.SIGQUIT)(注:kill的pid为0,表示向进程组所在的所有进程发送信号)。最后,containerd的reaper接收到syscall.SIGQUIT时,退出。

 

6、runv/containerd/containerd.go

func daemon(sv *supervisor.Supervisor, address)

首先创建一个reaper, s := make(chan os.Signal, 2048),signal.Notify(s, syscall.SIGCHLD, syscall.SIGTERM, syscall.SIGINT, syscall.SIGQUIT)。之后,调用server, err := startServer(address, sv)启动server。最后,处理来自s的信号,若信号为syscall.SIGCHLD只调用osutils.Reap()进行简单的处理,其他信号则直接调用server.Stop()关闭containerd

 

7、runv/containerd/containerd.go

func startServer(address string, sv *supervisor.Supervisor) (*grpc.Server, error)

该函数主要用于创建grpc server

(1)、l, err := net.Listen(defautlListenType, address)

(2)、types.RegisterAPIServer(s, server.NewServer(sv))

(3)、go func() { s.server(l) }()

 

-------------------------------------------------------------------------- containerd 对 event的处理---------------------------------------------------------

1、runv/supervisor/supervisor.go

func (sv *Supervisor) reaper()

首先调用events := sv.Events.Events(time.Time{})获得一个channel,用于获取container最新的状态。之后,调用一个for循环,从events中得到事件e,当e.Type 为EventExit时,调用go sv.reap(e.ID, e.PID)

 

2、runv/supervisor/supervisor.go

func (sv *Supervisor) reap(container, processId string)

根据container和processId,调用go p.reap(),删除c.ownerPod.Processes中和c.Processes对应的processId, 如果c中的Process为0了,则调用go c.reap(),再依次删除c.ownerPod.Containers和sv.Containers中的container。最后,若c.ownerPod.Containers为0了,则调用go c.ownerPod.reap()

 

3、runv/supervisor/supervisor.go

func (se *SvEvents) setupEventLog(logDir string) error

首先调用se.readEventLog(logDir),将events.log中的内容都读取到se.eventLog中,接着调用events := se.Events(time.Time{})获取events的channel,最后将channel中读取的event写入events.log中并且也添加到se.eventLog中。其实这个函数所做的工作,就是将events.log中原有的log添加到se.eventLog中,并且将新的event添加到se.eventLog和events.log中。

 

4、runv/supervisor/supervisor.go

// notifySubscribers will send the provided event to the external subscriber of the events channel,就是向se.subscriber中注册的各个channel发送event

func (se *SvEvents) notifySubscribers(e Event)

仅仅只是一个简单的for循环,遍历subscribers:

for sub := range se.subscribers{

  select{

  // do a non-blocking send for the channel (non-blocking是指,如果sub<-e阻塞,就直接default?)

  case sub <- e:

  default:

    glog.Infof("containerd: event not sent to the subscriber")

  }

}

 

-------------------------------------------------------------------- process, container, pod的reap 操作 --------------------------------------------------------------------------------------

1、runv/supervisor/process.go

func (p *Process) reap()

该函数只是简单的调用p.closeStdin而已

 

2、runv/supervisor/process.go

func (p *Process) closeStdin() error

当p.stdinCloser不为空时,调用p.stdinCloser.Close(),最后将p.stdinCloser置为nil即可

 

3、runv/supervisor/container.go

func (c *Container) reap()

首先,containerShareDir := filepath.Join(hypervisor.BaseDir, c.ownerPod.vm.Id, hypervisor.ShareDirTag, c.Id),然后将containerShareDir中的rootfs umount,最后,删除containerShareDir和filepath.Join(c.ownerPod.sv.StateDir, c.Id)

 

4、runv/supervisor/hyperpod.go

func (hp *HyperPod) reap()

首先调用Response := hp.vm.StopPod(hp.podStatus),接着调用hp.stopNsListener(),最后删除目录filepath.Join(hypervisor.BaseDir, hp.vm.Id)

转载于:https://www.cnblogs.com/YaoDD/p/5984517.html

 类似资料: