【K8s】controller-manager 源码分析 01-01

cmd/controller-manager/controller-manager.go

var (
	etcd_servers = flag.String("etcd_servers", "", "Servers for the etcd (http://ip:port).")
	master       = flag.String("master", "", "The address of the Kubernetes API server")
)

controllerManager := registry.MakeReplicationManager()

// ReplicationManager is responsible for synchronizing ReplicationController objects stored in etcd
// with actual running tasks.
// TODO: Remove the etcd dependency and re-factor in terms of a generic watch interface
type ReplicationManager struct {
	etcdClient  *etcd.Client
	kubeClient  client.ClientInterface
	taskControl TaskControlInterface
	updateLock  sync.Mutex
}

pkg/client/client.go

// ClientInterface holds the methods for clients of Kubenetes, an interface to allow mock testing
type ClientInterface interface {
	ListTasks(labelQuery map[string]string) (api.TaskList, error)
	GetTask(name string) (api.Task, error)
	DeleteTask(name string) error
	CreateTask(api.Task) (api.Task, error)
	UpdateTask(api.Task) (api.Task, error)

	GetReplicationController(name string) (api.ReplicationController, error)
	CreateReplicationController(api.ReplicationController) (api.ReplicationController, error)
	UpdateReplicationController(api.ReplicationController) (api.ReplicationController, error)
	DeleteReplicationController(string) error

	GetService(name string) (api.Service, error)
	CreateService(api.Service) (api.Service, error)
	UpdateService(api.Service) (api.Service, error)
	DeleteService(string) error
}

go util.Forever(func() { controllerManager.Synchronize() }, 20*time.Second)
go util.Forever(func() { controllerManager.WatchControllers() }, 20*time.Second)

func (rm *ReplicationManager) Synchronize() {}

var controllerSpec ReplicationController
err = rm.syncReplicationController(controllerSpec)

func (rm *ReplicationManager) syncReplicationController(controllerSpec ReplicationController) error {}

func (rm *ReplicationManager) syncReplicationController(controllerSpec ReplicationController) error {
	// 加锁，更新任务信息
	rm.updateLock.Lock()
	// 根据 label 标签，获取任务列表
	taskList, err := rm.kubeClient.ListTasks(controllerSpec.DesiredState.ReplicasInSet)
	if err != nil {
		return err
	}
	// 过滤所有退出状态的任务
	filteredList := rm.filterActiveTasks(taskList.Items)
	// 看期望的副本数和当前任务数的差值
	diff := len(filteredList) - controllerSpec.DesiredState.Replicas
	log.Printf("%#v", filteredList)
	if diff < 0 {
		// 将 负数 转换成 正数
		diff *= -1
		log.Printf("Too few replicas, creating %d\n", diff)
		// 根据期望数，使用期望状态创建任务
		for i := 0; i < diff; i++ {
			rm.taskControl.createReplica(controllerSpec)
		}
	} else if diff > 0 {
		// 任务数超过期望数，删除多余的任务
		log.Print("Too many replicas, deleting")
		for i := 0; i < diff; i++ {
			rm.taskControl.deleteTask(filteredList[i].ID)
		}
	}
	// 释放锁
	rm.updateLock.Unlock()
	return nil
}

func (rm *ReplicationManager) WatchControllers() {}

func (rm *ReplicationManager) WatchControllers() {
	// 创建无缓冲的管道
	watchChannel := make(chan *etcd.Response)
	// 通过起协程来监控 etcd 中 key 的状态变化，将变化的信息同步给 controller
	go util.Forever(func() { rm.etcdClient.Watch("/registry/controllers", 0, true, watchChannel, nil) }, 0)
	for {
		// 持续从管道中读取消息，如果没有变化，这里会阻塞
		watchResponse := <-watchChannel
		// 读出空值，直接休眠 10s，重新读取
		if watchResponse == nil {
			time.Sleep(time.Second * 10)
			continue
		}
		log.Printf("Got watch: %#v", watchResponse)
		// 监控 etcd 中 key 的变化， 对 set 的动作进行响应
		controller, err := rm.handleWatchResponse(watchResponse)
		if err != nil {
			log.Printf("Error handling data: %#v, %#v", err, watchResponse)
			continue
		}
		// 将变动后的信息，同步给副本控制器
		rm.syncReplicationController(*controller)
	}
}