kubernetes源码分析

yadianna

   Kube-proxy需要在每一个minion结点上运行。他的作用是service的代理，负责将业务连接到service后面具体执行结点（endpoints）。

　　我们列一下体现kube-proxy主要设计的代码部分。
　　总的来说kubernetes的代码可以从cmd看进去，看每个组件的启动过程，以及提供的业务。Cmd部分组织启动进程的过程以及代码调用，具体的代码实现在pkg里面。
　　Kube-proxy的重点代码是pkg中的serviceConfig、endpointsConfig（在pkg/proxy/config中），每个config中的mux，以及注册到config的处理函数proxier和endpointsHandler（这两个在pkg/proxy/userspace中），还有监听服务信息的API（代码也在pkg/proxy/config中）。

　　在执行kube-proxy命令的时候，NewProxyServerDefault函数依次创建serviceConfig，注册proxier；创建endpointConfig，注册endpointsHandler； 创建NewSourceAPI，在这里面会启动两个channel监听service和endpoint的变化，具体代码见下段：
　　serviceConfig := proxyconfig.NewServiceConfig()
　　serviceConfig.RegisterHandler(proxier)
　　endpointsConfig := proxyconfig.NewEndpointsConfig()
　　endpointsConfig.RegisterHandler(endpointsHandler)
　　proxyconfig.NewSourceAPI(
　　client,
　　config.ConfigSyncPeriod,
　　serviceConfig.Channel("api"),
　　endpointsConfig.Channel("api"),
　　)
　　下面是代码调用依赖。我的画图工具太差，有点难看~~

　　在
　　proxyconfig.NewSourceAPI(
　　client,
　　config.ConfigSyncPeriod,
　　serviceConfig.Channel("api"),
　　endpointsConfig.Channel("api"),
　　这段代码中，serviceConfig和endpointsConfig会分别创建一个channel，并监听这个channel的消息收到后推送到config内的mux。
　　serviceConfig的创建在
　　func NewServiceConfig() *ServiceConfig {
　　updates := make(chan struct{})
　　store := &serviceStore{updates: updates, services: make(map[string]map[types.NamespacedName]api.Service)}
　　mux := config.NewMux(store)
　　bcaster := config.NewBroadcaster()
　　go watchForUpdates(bcaster, store, updates)
　　return &ServiceConfig{mux, bcaster, store}
　　}
　　他创建了mux，注入进mux的serviceStore会提供Merge方法：
　　func (s *serviceStore) Merge(source string, change interface{}) error {
　　这个方法加工组织收到的数据，并通知proxier，调用他的OnServiceUpdate方法。具体是怎么触发到的呢？Merge方法在加工好数据后，构造一个空的数据{}推给叫updates的 channel，他触发了前面注册进去的proxier。 代码是serviceConfig.RegisterHandler(proxier)。
　　这样，当service有变化就会触发到OnServiceUpdate。
　　以新增一个service为例，前端api将消息发过来最后触发OnServiceUpdate后，这个方法会启动一个随机端口并监听他；再通过修改iptables的方式，将发给service的portal（集群内虚地址）的报文全部截收，发给自己刚才监听的那个随机端口。随机端口收到消息后，会选择这个service的一个endpoint，并建立到这个endpoint的链接，同时转发client和endpoints直接的报文。
　　这里可以看到，kubernetes这种解决服务发现的方法，多转发了一次报文。
　　OnServiceUpdate调用了两个很重要的函数addServiceOnPort和openPortal，通过他们完成了上述工作。
　　func (proxier *Proxier) addServiceOnPort(service proxy.ServicePortName, protocol api.Protocol, proxyPort int, timeout time.Duration) (*serviceInfo, error) {
　　sock, err := newProxySocket(protocol, proxier.listenIP, proxyPort)
　　if err != nil {
　　return nil, err
　　}
　　_, portStr, err := net.SplitHostPort(sock.Addr().String())
　　if err != nil {
　　sock.Close()
　　return nil, err
　　}
　　portNum, err := strconv.Atoi(portStr)
　　if err != nil {
　　sock.Close()
　　return nil, err
　　}
　　si := &serviceInfo{
　　isAliveAtomic:       1,
　　proxyPort:           portNum,
　　protocol:            protocol,
　　socket:              sock,
　　timeout:             timeout,
　　activeClients:       newClientCache(),
　　sessionAffinityType: api.ServiceAffinityNone, // default
　　stickyMaxAgeMinutes: 180,                     // TODO: parameterize this in the API.
　　}
　　proxier.setServiceInfo(service, si)
　　glog.V(2).Infof("Proxying for service %q on %s port %d", service, protocol, portNum)
　　go func(service proxy.ServicePortName, proxier *Proxier) {
　　defer util.HandleCrash()
　　atomic.AddInt32(&proxier.numProxyLoops, 1)
　　sock.ProxyLoop(service, si, proxier)
　　atomic.AddInt32(&proxier.numProxyLoops, -1)
　　}(service, proxier)
　　return si, nil
　　}
　　func (proxier *Proxier) openPortal(service proxy.ServicePortName, info *serviceInfo) error {
　　err := proxier.openOnePortal(info.portal, info.protocol, proxier.listenIP, info.proxyPort, service)
　　if err != nil {
　　return err
　　}
　　for _, publicIP := range info.externalIPs {
　　err = proxier.openOnePortal(portal{net.ParseIP(publicIP), info.portal.port, true}, info.protocol, proxier.listenIP, info.proxyPort, service)
　　if err != nil {
　　return err
　　}
　　}
　　for _, ingress := range info.loadBalancerStatus.Ingress {
　　if ingress.IP != "" {
　　err = proxier.openOnePortal(portal{net.ParseIP(ingress.IP), info.portal.port, false}, info.protocol, proxier.listenIP, info.proxyPort, service)
　　if err != nil {
　　return err
　　}
　　}
　　}
　　if info.nodePort != 0 {
　　err = proxier.openNodePort(info.nodePort, info.protocol, proxier.listenIP, info.proxyPort, service)
　　if err != nil {
　　return err
　　}
　　}
　　return nil
　　}
　　在addServiceOnPort中调用的重要函数是
　　func (udp *udpProxySocket) getBackendConn(activeClients *clientCache, cliAddr net.Addr, proxier *Proxier, service proxy.ServicePortName, timeout time.Duration) (net.Conn, error) {
　　activeClients.mu.Lock()
　　defer activeClients.mu.Unlock()
　　svrConn, found := activeClients.clients[cliAddr.String()]
　　if !found {
　　// TODO: This could spin up a new goroutine to make the outbound connection,
　　// and keep accepting inbound traffic.
　　glog.V(3).Infof("New UDP connection from %s", cliAddr)
　　var err error
　　svrConn, err = tryConnect(service, cliAddr, "udp", proxier)
　　if err != nil {
　　return nil, err
　　}
　　if err = svrConn.SetDeadline(time.Now().Add(timeout)); err != nil {
　　glog.Errorf("SetDeadline failed: %v", err)
　　return nil, err
　　}
　　activeClients.clients[cliAddr.String()] = svrConn
　　go func(cliAddr net.Addr, svrConn net.Conn, activeClients *clientCache, timeout time.Duration) {
　　defer util.HandleCrash()
　　udp.proxyClient(cliAddr, svrConn, activeClients, timeout)
　　}(cliAddr, svrConn, activeClients, timeout)
　　}
　　return svrConn, nil
　　}
　　他选择service的一个endpoint，并建立clientSock和endpointSock之间的连接
　　http://blog.csdn.net/zhaoguoguang/article/details/51225426