「连载」边缘计算（十八）02-21：边缘部分源码（源码分析篇）

（接上篇）

k8s.io/Kubernetes/pkg/Kubelet/dockershim/docker_service.go

// NewdockerService creates a new `dockerService` struct.

// NOTE: Anything passed to dockerService should be eventually handled in another way when we switch to running the shim as a different process.

func NewdockerService(config *ClientConfig, podSandboxImage string, streamingConfig *streaming.Config, pluginSettings *NetworkPluginSettings,

cgroupsName string, kubeCgroupDriver string, dockershimRootDir string, startLocalStreamingServer bool) (dockerService, error) {

...

ds := &dockerService{

client:          c,

os:              kubecontainer.RealOS{},

podSandboxImage: podSandboxImage,

streamingRuntime: &streamingRuntime{

client:      client,

execHandler: &NativeExecHandler{},

},

containerManager:          cm.NewContainerManager(cgroupsName, client),

checkpointManager:         checkpointManager,

startLocalStreamingServer: startLocalStreamingServer,

networkReady:              make(map[string]bool),

containerCleanupInfos:     make(map[string]*containerCleanupInfo),

}

...

}

从 NewdockerService()函数可以看出，dockershim的真身是dockerService struct。dockerService struct定义具体如下所示。

k8s.io/Kubernetes/pkg/Kubelet/dockershim/docker_service.go

type dockerService struct {

client           libdocker.Interface

os               kubecontainer.OSInterface

podSandboxImage  string

streamingRuntime *streamingRuntime

streamingServer  streaming.Server

network *network.PluginManager

// Map of podSandboxID :: network-is-ready

networkReady     map[string]bool

networkReadyLock sync.Mutex

containerManager cm.ContainerManager

// cgroup driver used by docker runtime.

cgroupDriver      string

checkpointManager checkpointmanager.CheckpointManager

// caches the version of the runtime.

// To be compatible with multiple docker versions, we need to perform

// version checking for some operations. Use this cache to avoid querying

// the docker daemon every time we need to do such checks.

versionCache *cache.ObjectCache

// startLocalStreamingServer indicates whether dockershim should start a

// streaming server on localhost.

startLocalStreamingServer bool

// containerCleanupInfos maps container IDs to the `containerCleanupInfo` structs

// needed to clean up after containers have been started or removed.

// (see `applyPlatformSpecificdockerConfig` and `performPlatformSpecificContainerCleanup`

// methods for more info).

containerCleanupInfos map[string]*containerCleanupInfo

}

从dockerService的定义可以看出，大部分属性都是与container、pod、image相关的。我们从这些属性可以推测dockerService是与container runtime交互的一个组件。为了进一步验证猜想，可以在同一个文件中（k8s.io/Kubernetes/pkg/Kubelet/dockershim/docker_service.go）查看dockerService实现的方法。

以上只是剖析了edged的初始化过程。下面剖析edged的启动过程，具体如下所示。

KubeEdge/edge/pkg/edged/edged.go

func (e *edged) Start(c *context.Context) {

e.context = c

e.metaClient = client.New(c)

// use self defined client to replace fake kube client

e.kubeClient = fakekube.NewSimpleClientset(e.metaClient)

e.statusManager = status.NewManager(e.kubeClient, e.podManager, utilpod.NewpodDeleteSafety(), e.metaClient)

if err := e.initializeModules(); err != nil {

klog.Errorf("initialize module error: %v", err)

os.Exit(1)

}

e.volumeManager = volumemanager.NewVolumeManager(

...

)

go e.volumeManager.Run(edgedutil.NewSourcesReady(), utilwait.NeverStop)

go utilwait.Until(e.syncNodeStatus, e.nodeStatusUpdateFrequency, utilwait.NeverStop)

e.probeManager = prober.NewManager(e.statusManager, e.livenessManager, containers.NewContainerRunner(), kubecontainer.NewRefManager(), record.NewEventRecorder())

e.pleg = edgepleg.NewGenericLifecycleRemote(e.containerRuntime, e.probeManager, plegChannelCapacity, plegRelistPeriod, e.podManager, e.statusManager, e.podCache, clock.RealClock{}, e.interfaceName)

e.statusManager.Start()

e.pleg.Start()

e.podAddWorkerRun(concurrentConsumers)

e.podRemoveWorkerRun(concurrentConsumers)

housekeepingTicker := time.NewTicker(housekeepingPeriod)

syncWorkQueueCh := time.NewTicker(syncWorkQueuePeriod)

e.probeManager.Start()

go e.syncLoopIteration(e.pleg.Watch(), housekeepingTicker.C, syncWorkQueueCh.C)

go e.server.ListenAndServe()

e.imageGCManager.Start()

e.StartGarbageCollection()

e.pluginManager = pluginmanager.NewPluginManager(

...

)

// Adding Registration Callback function for CSI Driver

e.pluginManager.AddHandler(pluginwatcherapi.CSIPlugin, plugincache.PluginHandler(csiplugin.PluginHandler))

// Start the plugin manager

go e.pluginManager.Run(edgedutil.NewSourcesReady(), utilwait.NeverStop)

e.syncpod()

}

 「未完待续……」