画布 1
containerd阅读
  cmd/containerd/main.go
    server, err := server.New(ctx, config)
      server/server.go
plugins, err := loadPlugins(config)
        162 // load all plugins into containerd
163 if err := plugin.Load(filepath.Join(config.Root, "plugins")); err != nil {
164 return nil, err
165 }
        166 // load additional plugins that don't automatically register themselves
167 plugin.Register(&plugin.Registration{
168 Type: plugin.ContentPlugin, // io.containerd.content.v1
169 ID: "content",
170 Init: func(ic *plugin.InitContext) (interface{}, error) {
171 return local.NewStore(ic.Root)
172 },
173 })
          return local.NewStore(ic.Root)
            content/local/store.go:
35 type store struct {
36 root string
37 }
38
39 func NewStore(root string) (content.Store, error) {
40 if err := os.MkdirAll(filepath.Join(root, "ingest"), 0777); err != nil && !os.IsExist(err) {
41 return nil, err
42 }
43
44 return &store{
45 root: root, // "/var/lib/containerd/io.containerd.content.v1.content"
46 }, nil
47 }
local.store实现了content.Store接口
              content/local/store.go:
104 type Store interface {
105 Manager
106 Provider
107 IngestManager
108 Ingester
109 }
        174 plugin.Register(&plugin.Registration{
175 Type: plugin.MetadataPlugin, //io.containerd.metadata.v1
176 ID: "bolt",
177 Init: func(ic *plugin.InitContext) (interface{}, error) {
178 if err := os.MkdirAll(ic.Root, 0711); err != nil {
179 return nil, err
180 }
181 return bolt.Open(filepath.Join(ic.Root, "meta.db"), 0644, nil)
// 返回了一个bolt.DB
182 },
183 })
        return plugin.Graph(), nil
// 所有plugin都返回了。
      rpc := grpc.NewServer(
grpc.UnaryInterceptor(interceptor),
grpc.StreamInterceptor(grpc_prometheus.StreamServerInterceptor),
)
s = &Server{
rpc: rpc,
events: events.NewExchange(),
}
      for _, p := range plugins {
id := p.URI()
initContext := plugin.NewContext(
ctx,
initialized,
config.Root,
config.State,
id,
)
initContext.Events = s.events
initContext.Address = config.GRPC.Address
pluginConfig, err := config.Decode(p.ID, p.Config)
initContext.Config = pluginConfig
instance, err := p.Init(initContext)
if types, ok := initialized[p.Type]; ok {
types[p.ID] = instance
} else {
initialized[p.Type] = map[string]interface{}{
p.ID: instance,
}
}
        differ/differ.go
func init() {
plugin.Register(&plugin.Registration{
Type: plugin.DiffPlugin, //io.containerd.differ.v1
ID: "base-diff",
Requires: []plugin.PluginType{
plugin.ContentPlugin,
plugin.MetadataPlugin,
},
Init: func(ic *plugin.InitContext) (interface{}, error) {
c, err := ic.Get(plugin.ContentPlugin)
if err != nil {
return nil, err
}
md, err := ic.Get(plugin.MetadataPlugin)
if err != nil {
return nil, err
}
return NewBaseDiff(metadata.NewContentStore(md.(*bolt.DB), c.(content.Store)))
},
})
}
          metadata.NewContentStore(md.(*bolt.DB), c.(content.Store))
            metadata/content.go:
type contentStore struct {
content.Store
db *bolt.DB
}
// NewContentStore returns a namespaced content store using an existing
// content store interface.
func NewContentStore(db *bolt.DB, cs content.Store) content.Store {
return &contentStore{
Store: cs,
db: db,
}
}
// 基于现有的local.Store封装成一个新的content.Store.
          type BaseDiff struct {
store content.Store
}
func NewBaseDiff(store content.Store) (*BaseDiff, error) {
return &BaseDiff{
store: store,
}, nil
}
BaseDiff实现了plugin.Diff
        service/container/service.go:
func init() {
plugin.Register(&plugin.Registration{
Type: plugin.GRPCPlugin,
ID: "containers",
Requires: []plugin.PluginType{
plugin.MetadataPlugin,
},
Init: func(ic *plugin.InitContext) (interface{}, error) {
m, err := ic.Get(plugin.MetadataPlugin)
if err != nil {
return nil, err
}
return NewService(m.(*bolt.DB), ic.Events), nil
},
})
}
          type Service struct {
db *bolt.DB
publisher events.Publisher
}
func NewService(db *bolt.DB, publisher events.Publisher) api.ContainersServer {
return &Service{db: db, publisher: publisher}
}
// Register接口决定了这个是一个服务
func (s *Service) Register(server *grpc.Server) error {
api.RegisterContainersServer(server, s)
return nil
}
// 实现了api/services/containers/v1/containers.proto里面的service接口
        略过：
io.containerd.grpc.v1.content, diff, events, healthcheck, images, namespaces
        func init() {
plugin.Register(&plugin.Registration{
Type: plugin.SnapshotPlugin, //io.containerd.snapshotter.v1
ID: "overlayfs",
Init: func(ic *plugin.InitContext) (interface{}, error) {
return NewSnapshotter(ic.Root)
},
})
}
          // NewSnapshotter returns a Snapshotter which uses overlayfs. The overlayfs
// diffs are stored under the provided root. A metadata file is stored under
// the root.
func NewSnapshotter(root string) (snapshot.Snapshotter, error) {
if err := os.MkdirAll(root, 0700); err != nil {
return nil, err
}
supportsDType, err := fs.SupportsDType(root)
if err != nil {
return nil, err
}
if !supportsDType {
return nil, fmt.Errorf("%s does not support d_type. If the backing filesystem is xfs, please reformat with ftype=1 to enable d_type support.", root)
}
ms, err := storage.NewMetaStore(filepath.Join(root, "metadata.db"))
if err != nil {
return nil, err
}
if err := os.Mkdir(filepath.Join(root, "snapshots"), 0700); err != nil && !os.IsExist(err) {
return nil, err
}
return &snapshotter{
root: root, // "/var/lib/containerd/io.containerd.snapshotter.v1.overlayfs/"
ms: ms,
}, nil
}
            type MetaStore struct {
dbfile string
}
func NewMetaStore(dbfile string) (*MetaStore, error) {
return &MetaStore{
dbfile: dbfile, // "/var/lib/containerd/io.containerd.snapshotter.v1.overlayfs/metadata.db"
}, nil
}
        service/snapshot/service.go
func init() {
plugin.Register(&plugin.Registration{
Type: plugin.GRPCPlugin,
ID: "snapshots",
Requires: []plugin.PluginType{
plugin.SnapshotPlugin,
plugin.MetadataPlugin,
},
Init: newService,
})
}
//这个也是个service
          type service struct {
snapshotters map[string]snapshot.Snapshotter
publisher events.Publisher
}
func newService(ic *plugin.InitContext) (interface{}, error) {
rawSnapshotters, err := ic.GetAll(plugin.SnapshotPlugin) //获得snapshot.Snapshotter接口
if err != nil {
return nil, err
}
md, err := ic.Get(plugin.MetadataPlugin) //返回boltdb
if err != nil {
return nil, err
}
snapshotters := make(map[string]snapshot.Snapshotter)
for name, sn := range rawSnapshotters {
snapshotters[name] = metadata.NewSnapshotter(md.(*bolt.DB), name, sn.(snapshot.Snapshotter))
}
if len(snapshotters) == 0 {
return nil, errors.Errorf("failed to create snapshotter service: no snapshotters loaded")
}
return &service{
snapshotters: snapshotters, // 包含对各个文件系统overlay/btrfs的metadata
publisher: ic.Events,
}, nil
}
            metadata/snapshot.go:
type snapshotter struct {
snapshot.Snapshotter
name string
db *bolt.DB
}
// NewSnapshotter returns a new Snapshotter which namespaces the given snapshot
// using the provided name and metadata store.
func NewSnapshotter(db *bolt.DB, name string, sn snapshot.Snapshotter) snapshot.Snapshotter {
return &snapshotter{
Snapshotter: sn,
name: name,
db: db,
}
}
// NewSnapshotter是对overlay/btrfs snapshotter的进一步封装
        metrics/cgroups/cgroups.go
func init() {
plugin.Register(&plugin.Registration{
Type: plugin.TaskMonitorPlugin, // "io.containerd.monitor.v1"
ID: "cgroups",
Init: New,
})
}
          func New(ic *plugin.InitContext) (interface{}, error) {
var (
ns = metrics.NewNamespace("container", "", nil)
collector = NewCollector(ns)
)
oom, err := NewOOMCollector(ns)
if err != nil {
return nil, err
}
metrics.Register(ns)
return &cgroupsMonitor{
collector: collector,
oom: oom,
context: ic.Context,
publisher: ic.Events,
}, nil
}
            metrics/cgroups/metrics.go:
func NewCollector(ns *metrics.Namespace) *Collector {
// add machine cpus and memory info
c := &Collector{
ns: ns,
cgroups: make(map[string]*task),
}
c.metrics = append(c.metrics, pidMetrics...)
c.metrics = append(c.metrics, cpuMetrics...)
c.metrics = append(c.metrics, memoryMetrics...)
c.metrics = append(c.metrics, hugetlbMetrics...)
c.metrics = append(c.metrics, blkioMetrics...)
ns.Add(c)
return c
}
        linux/runtime.go:
func init() {
plugin.Register(&plugin.Registration{
Type: plugin.RuntimePlugin, // "io.containerd.runtime.v1"
ID: "linux",
Init: New,
Requires: []plugin.PluginType{
plugin.TaskMonitorPlugin,
plugin.MetadataPlugin,
},
Config: &Config{
Shim: defaultShim, // "containerd-shim"
Runtime: defaultRuntime, // "runc"
},
})
}
          linux/runtime.go: 值得深入看一下
func New(ic *plugin.InitContext) (interface{}, error) {
if err := os.MkdirAll(ic.Root, 0711); err != nil {
return nil, err
}
if err := os.MkdirAll(ic.State, 0711); err != nil {
return nil, err
}
monitor, err := ic.Get(plugin.TaskMonitorPlugin)
if err != nil {
return nil, err
}
m, err := ic.Get(plugin.MetadataPlugin)
if err != nil {
return nil, err
}
cfg := ic.Config.(*Config)
r := &Runtime{
root: ic.Root,
state: ic.State,
remote: !cfg.NoShim,
shim: cfg.Shim,
shimDebug: cfg.ShimDebug,
runtime: cfg.Runtime,
monitor: monitor.(runtime.TaskMonitor),
tasks: runtime.NewTaskList(),
db: m.(*bolt.DB),
address: ic.Address,
events: ic.Events,
}
tasks, err := r.restoreTasks(ic.Context)
if err != nil {
return nil, err
}
for _, t := range tasks {
if err := r.tasks.AddWithNamespace(t.namespace, t); err != nil {
return nil, err
}
}
return r, nil
}
            func (r *Runtime) restoreTasks(ctx context.Context) ([]*Task, error) {
dir, err := ioutil.ReadDir(r.state)
if err != nil {
return nil, err
}
var o []*Task
for _, namespace := range dir {
if !namespace.IsDir() {
continue
}
name := namespace.Name()
log.G(ctx).WithField("namespace", name).Debug("loading tasks in namespace")
tasks, err := r.loadTasks(ctx, name)
if err != nil {
return nil, err
}
o = append(o, tasks...)
}
return o, nil
}
        services/tasks/service.go:
func init() {
plugin.Register(&plugin.Registration{
Type: plugin.GRPCPlugin,
ID: "tasks",
Requires: []plugin.PluginType{
plugin.RuntimePlugin,
plugin.MetadataPlugin,
plugin.ContentPlugin,
},
Init: New,
})
}
          func New(ic *plugin.InitContext) (interface{}, error) {
rt, err := ic.GetAll(plugin.RuntimePlugin)
if err != nil {
return nil, err
}
m, err := ic.Get(plugin.MetadataPlugin)
if err != nil {
return nil, err
}
ct, err := ic.Get(plugin.ContentPlugin)
if err != nil {
return nil, err
}
cs := metadata.NewContentStore(m.(*bolt.DB), ct.(content.Store))
runtimes := make(map[string]runtime.Runtime)
for _, rr := range rt {
r := rr.(runtime.Runtime)
runtimes[r.ID()] = r
}
return &Service{
runtimes: runtimes,
db: m.(*bolt.DB),
store: cs,
publisher: ic.Events,
}, nil
}
      // register services after all plugins have been initialized
for _, service := range services {
if err := service.Register(rpc); err != nil {
return nil, err
}
}
第一个被初始化的，1号依赖进程。
2号