k8s-apiserver源码解析

贮备知识

flag用法

一文彻底搞懂golang工厂模式设计方式

必须懂得上述知识，看源码才能有头绪，尤其是理解工厂模式。整个apiserver的代码设计都是用的工厂模式

源码解析

apiserver主要负责将来自用户的需求（前端/命令行）记录到etcd中，并掉用其他组件去实际操作。

入口文件

kubernetes/cmd/apiserver/apiserver.go apiserver

package main

import (
	"flag"
	"fmt"
	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
	"log"
	"net/http"
	"time"

	"github.com/coreos/go-etcd/etcd"

	"github.com/GoogleCloudPlatform/kubernetes/pkg/apiserver"
	kube_client "github.com/GoogleCloudPlatform/kubernetes/pkg/client"
	"github.com/GoogleCloudPlatform/kubernetes/pkg/registry"
)

// 一些命令行参数
var (
	port                        = flag.Uint("port", 8080, "The port to listen on.  Default 8080.")
	address                     = flag.String("address", "127.0.0.1", "The address on the local server to listen to. Default 127.0.0.1")
	apiPrefix                   = flag.String("api_prefix", "/api/v1beta1", "The prefix for API requests on the server. Default '/api/v1beta1'")
	etcdServerList, machineList util.StringList
)

// init函数  自动执行的函数 命令行参数赋值
func init() {
	flag.Var(&etcdServerList, "etcd_servers", "Servers for the etcd (http://ip:port), comma separated")
	flag.Var(&machineList, "machines", "List of machines to schedule onto, comma separated.")
}

func main() {
	flag.Parse()

	if len(machineList) == 0 {
		log.Fatal("No machines specified!")
	}

	var (
		// registry 是针对dao层的封装
		// 不同资源到dao层操作的封装

		// registry.TaskRegistry是个接口 规定了task这个资源dao层的接口都要实现那些方法
		taskRegistry       registry.TaskRegistry 
		// registry.ControllerRegistry是个接口 规定了controller这个资源dao层的接口都要实现那些方式
		controllerRegistry registry.ControllerRegistry
		serviceRegistry    registry.ServiceRegistry
	)
	
	// 根据判断选择dao层的操作主要使用etcd还是memory
	if len(etcdServerList) > 0 {
		log.Printf("Creating etcd client pointing to %v", etcdServerList)
		etcdClient := etcd.NewClient(etcdServerList)
		// 以下是对接口的具体实现
		
		// 以下的返回值都是EtcdRegistry，EtcdRegistry具体实现了registry.ServiceRegistry，
		// registry.ControllerRegistry，registry.TaskRegistry 这三个接口中的方法，也就是
		// 实现了这些接口
		// 返回的是EtcdRegistry 实现了TaskRegistry  从某种意义上说EtcdRegistry就是TaskRegistry
		taskRegistry = registry.MakeEtcdRegistry(etcdClient, machineList) 
		controllerRegistry = registry.MakeEtcdRegistry(etcdClient, machineList)
		serviceRegistry = registry.MakeEtcdRegistry(etcdClient, machineList)
	} else {
		taskRegistry = registry.MakeMemoryRegistry()
		controllerRegistry = registry.MakeMemoryRegistry()
		serviceRegistry = registry.MakeMemoryRegistry()
	}

	containerInfo := &kube_client.HTTPContainerInfo{
		Client: http.DefaultClient,
		Port:   10250,
	}

	// RESTStorage 对于http请求的那一层封装
	// 这里就用到了工厂模式  使用map
	// 根据不同的资源 执行资源的方法
	// tasks，replicationControllers，services这三个资源分别都实现了RESTStorage这个接口
	storage := map[string]apiserver.RESTStorage{
		"tasks":                  registry.MakeTaskRegistryStorage(taskRegistry, containerInfo, registry.MakeFirstFitScheduler(machineList, taskRegistry)),
		"replicationControllers": registry.MakeControllerRegistryStorage(controllerRegistry),
		"services":               registry.MakeServiceRegistryStorage(serviceRegistry),
	}

	endpoints := registry.MakeEndpointController(serviceRegistry, taskRegistry)
	go util.Forever(func() { endpoints.SyncServiceEndpoints() }, time.Second*10)

	s := &http.Server{
		Addr:           fmt.Sprintf("%s:%d", *address, *port),
		Handler:        apiserver.New(storage, *apiPrefix),
		ReadTimeout:    10 * time.Second,
		WriteTimeout:   10 * time.Second,
		MaxHeaderBytes: 1 << 20,
	}
	log.Fatal(s.ListenAndServe())
}

dao层 数据库操作封装

kubernetes/pkg/registry/interfaces.go 抽象dao层接口方法规定

package registry

import (
	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
)

// 抽象task这个资源关于dao层封装都要实现的接口方法
// dao层的封装 数据库操作 公共接口 所有针对dao层的封装都要实现这个接口
// 这样可以统一封装 比如 etcd操作 内存操作 mysql操作等等
// TaskRegistry is an interface implemented by things that know how to store Task objects
type TaskRegistry interface {
	// ListTasks obtains a list of tasks that match query.
	// Query may be nil in which case all tasks are returned.
	ListTasks(query *map[string]string) ([]api.Task, error)
	// Get a specific task
	GetTask(taskId string) (*api.Task, error)
	// Create a task based on a specification, schedule it onto a specific machine.
	CreateTask(machine string, task api.Task) error
	// Update an existing task
	UpdateTask(task api.Task) error
	// Delete an existing task
	DeleteTask(taskId string) error
}

// 抽象Controller这个资源操作dao层需要实现的接口方法
// ControllerRegistry is an interface for things that know how to store Controllers
type ControllerRegistry interface {
	ListControllers() ([]api.ReplicationController, error)
	GetController(controllerId string) (*api.ReplicationController, error)
	CreateController(controller api.ReplicationController) error
	UpdateController(controller api.ReplicationController) error
	DeleteController(controllerId string) error
}

kubernetes/pkg/registry/etcd_registry.go 对dao层接口方法具体的实现，主要是针对不同的资源对etcd的不同的操作

http层接口封装

/Users/yangqiqi/GolandProjects/kubernetes/pkg/apiserver/apiserver.go http层接口抽象

// http层接口需要实现的方法 和前端（人工命令行）交互
// RESTStorage is a generic interface for RESTful storage services
type RESTStorage interface {
	List(*url.URL) (interface{}, error)
	Get(id string) (interface{}, error)
	Delete(id string) error
	Extract(body string) (interface{}, error)
	Create(interface{}) error
	Update(interface{}) error
}

kubernetes/pkg/registry/controller_registry.go

kubernetes/pkg/registry/service_registry.go

kubernetes/pkg/registry/task_registry.go

对RESTStorage接口的具体实现