Contiv源码分析：Netplugin

func main() {
	// 读取并解析命令行参数
	...
	
	// initialize the config
	pluginConfig := plugin.Config{
		Drivers: plugin.Drivers{
			Network: "ovs",
			State:   stateStore,
		},
		Instance: core.InstanceInfo{
			HostLabel:  opts.hostLabel,
			CtrlIP:     opts.ctrlIP,
			VtepIP:     opts.vtepIP,
			UplinkIntf: opts.vlanIntf,
			DbURL:      opts.dbURL,
			PluginMode: opts.pluginMode,
		},
	}

	// Create a new agent
	ag := agent.NewAgent(&pluginConfig)

	// Process all current state
	ag.ProcessCurrentState()

	// post initialization processing
	ag.PostInit()

	// handle events
	if err := ag.HandleEvents(); err != nil {
		log.Infof("Netplugin exiting due to error: %v", err)
		os.Exit(1)
	}
}

// NewAgent creates a new netplugin agent
func NewAgent(pluginConfig *plugin.Config) *Agent {
	opts := pluginConfig.Instance
	netPlugin := &plugin.NetPlugin{}

	// Initialize service registry plugin
	svcPlugin, quitCh, err := svcplugin.NewSvcregPlugin(opts.DbURL, nil)
	if err != nil {
		log.Fatalf("Error initializing service registry plugin")
	}

	// init cluster state
	err = cluster.Init(opts.DbURL)
	if err != nil {
		log.Fatalf("Error initializing cluster. Err: %v", err)
	}

	// Init the driver plugins..
	err = netPlugin.Init(*pluginConfig)
	if err != nil {
		log.Fatalf("Failed to initialize the plugin. Error: %s", err)
	}

	// Initialize appropriate plugin
	switch opts.PluginMode {
	case "docker":
		dockplugin.InitDockPlugin(netPlugin, svcPlugin)

	case "kubernetes":
		k8splugin.InitCNIServer(netPlugin)

	case "test":
		// nothing to do. internal mode for testing
	default:
		log.Fatalf("Unknown plugin mode -- should be docker | kubernetes")
	}
	// init mesos plugin
	mesosplugin.InitPlugin(netPlugin)

	// create a new agent
	agent := &Agent{
		netPlugin:    netPlugin,
		pluginConfig: pluginConfig,
		svcPlugin:    svcPlugin,
		svcQuitCh:    quitCh,
	}

	return agent
}

// PostInit post initialization
func (ag *Agent) PostInit() error {
	opts := ag.pluginConfig.Instance

	// Initialize clustering
	err := cluster.RunLoop(ag.netPlugin, opts.CtrlIP, opts.VtepIP, opts.HostLabel)
	if err != nil {
		log.Errorf("Error starting cluster run loop")
	}

	// start service REST requests
	ag.serveRequests()

	return nil
}

// implement utilities for instantiating the supported core.Driver
// (state, network and endpoint) instances

type driverConfigTypes struct {
	DriverType reflect.Type
	ConfigType reflect.Type
}

var networkDriverRegistry = map[string]driverConfigTypes{
	OvsNameStr: {
		DriverType: reflect.TypeOf(drivers.OvsDriver{}),
		ConfigType: reflect.TypeOf(drivers.OvsDriver{}),
	},
	// fakedriver is used for tests, so not exposing a public name for it.
	"fakedriver": {
		DriverType: reflect.TypeOf(drivers.FakeNetEpDriver{}),
		ConfigType: reflect.TypeOf(drivers.FakeNetEpDriverConfig{}),
	},
}

var stateDriverRegistry = map[string]driverConfigTypes{
	EtcdNameStr: {
		DriverType: reflect.TypeOf(state.EtcdStateDriver{}),
		ConfigType: reflect.TypeOf(state.EtcdStateDriverConfig{}),
	},
	ConsulNameStr: {
		DriverType: reflect.TypeOf(state.ConsulStateDriver{}),
		ConfigType: reflect.TypeOf(state.ConsulStateDriverConfig{}),
	},
	// fakestate-driver is used for tests, so not exposing a public name for it.
	"fakedriver": {
		DriverType: reflect.TypeOf(state.FakeStateDriver{}),
		ConfigType: reflect.TypeOf(state.FakeStateDriverConfig{}),
	},
}

// NewStateDriver instantiates a 'named' state-driver with specified configuration
func NewStateDriver(name string, instInfo *core.InstanceInfo) (core.StateDriver, error) {
	if name == "" || instInfo == nil {
		return nil, core.Errorf("invalid driver name or configuration passed.")
	}

	if gStateDriver != nil {
		return nil, core.Errorf("statedriver instance already exists.")
	}

	driver, err := initHelper(stateDriverRegistry, name)
	if err != nil {
		return nil, err
	}

	d := driver.(core.StateDriver)
	err = d.Init(instInfo)
	if err != nil {
		return nil, err
	}

	gStateDriver = d
	return d, nil
}

// GetStateDriver returns the singleton instance of the state-driver
func GetStateDriver() (core.StateDriver, error) {
	if gStateDriver == nil {
		return nil, core.Errorf("statedriver has not been not created.")
	}

	return gStateDriver, nil
}

// NewNetworkDriver instantiates a 'named' network-driver with specified configuration
func NewNetworkDriver(name string, instInfo *core.InstanceInfo) (core.NetworkDriver, error) {
	if name == "" || instInfo == nil {
		return nil, core.Errorf("invalid driver name or configuration passed.")
	}

	driver, err := initHelper(networkDriverRegistry, name)
	if err != nil {
		return nil, err
	}

	d := driver.(core.NetworkDriver)
	err = d.Init(instInfo)
	if err != nil {
		return nil, err
	}

	return d, nil
}

// InitCNIServer initializes the k8s cni server
func InitCNIServer(netplugin *plugin.NetPlugin) error {

	netPlugin = netplugin
	hostname, err := os.Hostname()
	if err != nil {
		log.Fatalf("Could not retrieve hostname: %v", err)
	}

	pluginHost = hostname

	// Set up the api client instance
	kubeAPIClient = setUpAPIClient()
	if kubeAPIClient == nil {
		log.Fatalf("Could not init kubernetes API client")
	}

	log.Debugf("Configuring router")

	router := mux.NewRouter()

	// register handlers for cni
	t := router.Headers("Content-Type", "application/json").Methods("POST").Subrouter()
	t.HandleFunc(cniapi.EPAddURL, makeHTTPHandler(addPod))
	t.HandleFunc(cniapi.EPDelURL, makeHTTPHandler(deletePod))
	t.HandleFunc("/ContivCNI.{*}", unknownAction)

	driverPath := cniapi.ContivCniSocket
	os.Remove(driverPath)
	os.MkdirAll(cniapi.PluginPath, 0700)

	go func() {
		l, err := net.ListenUnix("unix", &net.UnixAddr{Name: driverPath, Net: "unix"})
		if err != nil {
			panic(err)
		}

		log.Infof("k8s plugin listening on %s", driverPath)
		http.Serve(l, router)
		l.Close()
		log.Infof("k8s plugin closing %s", driverPath)
	}()

	//InitKubServiceWatch(netplugin)
	return nil
}

// addPod is the handler for pod additions
func addPod(r *http.Request) (interface{}, error) {

	resp := cniapi.RspAddPod{}

	logEvent("add pod")

	content, err := ioutil.ReadAll(r.Body)
	if err != nil {
		log.Errorf("Failed to read request: %v", err)
		return resp, err
	}

	pInfo := cniapi.CNIPodAttr{}
	if err := json.Unmarshal(content, &pInfo); err != nil {
		return resp, err
	}

	// Get labels from the kube api server
	epReq, err := getEPSpec(&pInfo)
	if err != nil {
		log.Errorf("Error getting labels. Err: %v", err)
		setErrorResp(&resp, "Error getting labels", err)
		return resp, err
	}

	ep, err := createEP(epReq)
	if err != nil {
		log.Errorf("Error creating ep. Err: %v", err)
		setErrorResp(&resp, "Error creating EP", err)
		return resp, err
	}

	// convert netns to pid that netlink needs
	pid, err := nsToPID(pInfo.NwNameSpace)
	if err != nil {
		log.Errorf("Error moving to netns. Err: %v", err)
		setErrorResp(&resp, "Error moving to netns", err)
		return resp, err
	}

	// Set interface attributes for the new port
	err = setIfAttrs(pid, ep.PortName, ep.IPAddress, pInfo.IntfName)
	if err != nil {
		log.Errorf("Error setting interface attributes. Err: %v", err)
		setErrorResp(&resp, "Error setting interface attributes", err)
		return resp, err
	}

	// if Gateway is not specified on the nw, use the host gateway
	gwIntf := pInfo.IntfName
	gw := ep.Gateway
	if gw == "" {
		hostIf := netutils.GetHostIntfName(ep.PortName)
		hostIP, _ := netutils.HostIfToIP(hostIf)
		err = netPlugin.CreateHostAccPort(hostIf, ep.IPAddress, hostIP)
		if err != nil {
			log.Errorf("Error setting host access. Err: %v", err)
		} else {
			err = setIfAttrs(pid, hostIf, hostIP, "host1")
			if err != nil {
				log.Errorf("Move to pid %d failed", pid)
			} else {
				gw = hostGWIP
				gwIntf = "host1"
				// make sure service subnet points to eth0
				svcSubnet := contivK8Config.SvcSubnet
				addStaticRoute(pid, svcSubnet, pInfo.IntfName)
			}
		}

	}

	// Set default gateway
	err = setDefGw(pid, gw, gwIntf)
	if err != nil {
		log.Errorf("Error setting default gateway. Err: %v", err)
		setErrorResp(&resp, "Error setting default gateway", err)
		return resp, err
	}

	resp.Result = 0
	resp.IPAddress = ep.IPAddress
	resp.EndpointID = pInfo.InfraContainerID
	return resp, nil
}

// deletePod is the handler for pod deletes
func deletePod(r *http.Request) (interface{}, error) {

	resp := cniapi.RspAddPod{}

	logEvent("del pod")

	content, err := ioutil.ReadAll(r.Body)
	if err != nil {
		log.Errorf("Failed to read request: %v", err)
		return resp, err
	}

	pInfo := cniapi.CNIPodAttr{}
	if err := json.Unmarshal(content, &pInfo); err != nil {
		return resp, err
	}

	// Get labels from the kube api server
	epReq, err := getEPSpec(&pInfo)
	if err != nil {
		log.Errorf("Error getting labels. Err: %v", err)
		setErrorResp(&resp, "Error getting labels", err)
		return resp, err
	}

	netPlugin.DeleteHostAccPort(epReq.EndpointID)
	err = epCleanUp(epReq)
	resp.Result = 0
	resp.EndpointID = pInfo.InfraContainerID
	return resp, err
}

func mainfunc() {
	pInfo := cniapi.CNIPodAttr{}
	cniCmd := os.Getenv("CNI_COMMAND")

	// Open a logfile
	f, err := os.OpenFile("/var/log/contivk8s.log", os.O_RDWR|os.O_CREATE|os.O_APPEND, 0666)
	if err != nil {
		logger.Fatalf("error opening file: %v", err)
	}
	defer f.Close()

	logger.SetOutput(f)
	log = getPrefixedLogger()

	log.Infof("==> Start New Log <==\n")
	log.Infof("command: %s, cni_args: %s", cniCmd, os.Getenv("CNI_ARGS"))

	// Collect information passed by CNI
	err = getPodInfo(&pInfo)
	if err != nil {
		log.Fatalf("Error parsing environment. Err: %v", err)
	}

	nc := clients.NewNWClient()
	if cniCmd == "ADD" {
		addPodToContiv(nc, &pInfo)
	} else if cniCmd == "DEL" {
		deletePodFromContiv(nc, &pInfo)
	}

}