// Add a master node
func addMaster(netplugin *plugin.NetPlugin, srvInfo core.ServiceInfo) error {
	// save it in db
	masterDB[masterKey(srvInfo)] = &srvInfo

	// tell the plugin about the master
	return netplugin.AddMaster(srvInfo)
}

// delete master node
func deleteMaster(netplugin *plugin.NetPlugin, srvInfo core.ServiceInfo) error {
	// delete from the db
	delete(masterDB, masterKey(srvInfo))

	// tel plugin about it
	return netplugin.DeleteMaster(srvInfo)
}

//processSvcProviderUpdEvent updates service provider events
func processSvcProviderUpdEvent(netPlugin *plugin.NetPlugin, svcProvider *mastercfg.SvcProvider, isDelete bool) error {
	if isDelete {
		//ignore delete event since servicelb delete will take care of this.
		return nil
	}
	netPlugin.SvcProviderUpdate(svcProvider.ServiceName, svcProvider.Providers)
	return nil
}

// delete master node
func deleteMaster(netplugin *plugin.NetPlugin, srvInfo objdb.ServiceInfo) error {
	// delete from the db
	delete(MasterDB, masterKey(srvInfo))

	// tel plugin about it
	return netplugin.DeleteMaster(core.ServiceInfo{
		HostAddr: srvInfo.HostAddr,
		Port:     netmasterRPCPort,
	})
}

// Add a master node
func addMaster(netplugin *plugin.NetPlugin, srvInfo objdb.ServiceInfo) error {
	// save it in db
	masterDB[masterKey(srvInfo)] = &srvInfo

	// tell the plugin about the master
	return netplugin.AddMaster(core.ServiceInfo{
		HostAddr: srvInfo.HostAddr,
		Port:     ofnet.OFNET_MASTER_PORT,
	})
}

// Add a master node
func addMaster(netplugin *plugin.NetPlugin, srvInfo objdb.ServiceInfo) error {
	// save it in db
	MasterDB[masterKey(srvInfo)] = &srvInfo

	// tell the plugin about the master
	return netplugin.AddMaster(core.ServiceInfo{
		HostAddr: srvInfo.HostAddr,
		Port:     netmasterRPCPort,
	})
}

func processNetEvent(netPlugin *plugin.NetPlugin, nwCfg *mastercfg.CfgNetworkState,
	isDelete bool) (err error) {
	// take a lock to ensure we are programming one event at a time.
	// Also network create events need to be processed before endpoint creates
	// and reverse shall happen for deletes. That order is ensured by netmaster,
	// so we don't need to worry about that here
	netPlugin.Lock()
	defer func() { netPlugin.Unlock() }()

	operStr := ""
	if isDelete {
		err = netPlugin.DeleteNetwork(nwCfg.ID, nwCfg.NwType, nwCfg.PktTagType, nwCfg.PktTag, nwCfg.ExtPktTag,
			nwCfg.Gateway, nwCfg.Tenant)
		operStr = "delete"
	} else {
		err = netPlugin.CreateNetwork(nwCfg.ID)
		operStr = "create"
	}
	if err != nil {
		log.Errorf("Network operation %s failed. Error: %s", operStr, err)
	} else {
		log.Infof("Network operation %s succeeded", operStr)
	}

	return
}

//processBgpEvent processes Bgp neighbor add/delete events
func processBgpEvent(netPlugin *plugin.NetPlugin, opts cliOpts, hostID string, isDelete bool) error {
	var err error

	if opts.hostLabel != hostID {
		log.Errorf("Ignoring Bgp Event on this host")
		return err
	}
	netPlugin.Lock()
	defer func() { netPlugin.Unlock() }()

	operStr := ""
	if isDelete {
		err = netPlugin.DeleteBgp(hostID)
		operStr = "delete"
	} else {
		err = netPlugin.AddBgp(hostID)
		operStr = "create"
	}
	if err != nil {
		log.Errorf("Bgp operation %s failed. Error: %s", operStr, err)
	} else {
		log.Infof("Bgp operation %s succeeded", operStr)
	}

	return err
}

func processServiceLBEvent(netPlugin *plugin.NetPlugin, opts cliOpts, svcLBCfg *mastercfg.CfgServiceLBState,
	isDelete bool) error {
	var err error
	portSpecList := []core.PortSpec{}
	portSpec := core.PortSpec{}

	netPlugin.Lock()
	defer func() { netPlugin.Unlock() }()
	serviceID := svcLBCfg.ID

	log.Infof("Recevied Process Service load balancer event {%v}", svcLBCfg)

	//create portspect list from state.
	//Ports format: servicePort:ProviderPort:Protocol
	for _, port := range svcLBCfg.Ports {

		portInfo := strings.Split(port, ":")
		if len(portInfo) != 3 {
			return errors.New("Invalid Port Format")
		}
		svcPort := portInfo[0]
		provPort := portInfo[1]
		portSpec.Protocol = portInfo[2]

		sPort, _ := strconv.ParseUint(svcPort, 10, 16)
		portSpec.SvcPort = uint16(sPort)

		pPort, _ := strconv.ParseUint(provPort, 10, 16)
		portSpec.ProvPort = uint16(pPort)

		portSpecList = append(portSpecList, portSpec)
	}

	spec := &core.ServiceSpec{
		IPAddress: svcLBCfg.IPAddress,
		Ports:     portSpecList,
	}
	operStr := ""
	if isDelete {
		err = netPlugin.DeleteServiceLB(serviceID, spec)
		operStr = "delete"
	} else {
		err = netPlugin.AddServiceLB(serviceID, spec)
		operStr = "create"
	}
	if err != nil {
		log.Errorf("Service Load Balancer %s failed.Error:%s", operStr, err)
		return err
	}
	log.Infof("Service Load Balancer %s succeeded", operStr)

	return nil
}

func processGlobalConfigUpdEvent(netPlugin *plugin.NetPlugin, opts core.InstanceInfo, cfg *mastercfg.GlobConfig) {

	// parse store URL
	parts := strings.Split(opts.DbURL, "://")
	if len(parts) < 2 {
		log.Fatalf("Invalid cluster-store-url %s", opts.DbURL)
	}
	stateStore := parts[0]
	// initialize the config
	pluginConfig := plugin.Config{
		Drivers: plugin.Drivers{
			Network: "ovs",
			State:   stateStore,
		},
		Instance: opts,
	}
	pluginConfig.Instance.ArpMode = cfg.ArpMode
	netPlugin.GlobalConfigUpdate(pluginConfig)

	log.Infof("Global Config updated")
}

// processEpState restores endpoint state
func processEpState(netPlugin *plugin.NetPlugin, opts cliOpts, epID string) error {
	// take a lock to ensure we are programming one event at a time.
	// Also network create events need to be processed before endpoint creates
	// and reverse shall happen for deletes. That order is ensured by netmaster,
	// so we don't need to worry about that here
	netPlugin.Lock()
	defer func() { netPlugin.Unlock() }()

	// read endpoint config
	epCfg := &mastercfg.CfgEndpointState{}
	epCfg.StateDriver = netPlugin.StateDriver
	err := epCfg.Read(epID)

	if err != nil {
		log.Errorf("Failed to read config for ep '%s' \n", epID)
		return err
	}
	// if the endpoint is not for this host, ignore it
	if skipHost(epCfg.VtepIP, epCfg.HomingHost, opts.hostLabel) {
		log.Infof("skipping mismatching host for ep %s. EP's host %s (my host: %s)",
			epID, epCfg.HomingHost, opts.hostLabel)
		return nil
	}

	// Create the endpoint
	err = netPlugin.CreateEndpoint(epID)
	if err != nil {
		log.Errorf("Endpoint operation create failed. Error: %s", err)
		return err
	}

	log.Infof("Endpoint operation create succeeded")

	return err
}

// Process Infra Nw Create
// Auto allocate an endpoint for this node
func processInfraNwCreate(netPlugin *plugin.NetPlugin, nwCfg *mastercfg.CfgNetworkState, opts cliOpts) (err error) {
	pluginHost := opts.hostLabel

	// Build endpoint request
	mreq := master.CreateEndpointRequest{
		TenantName:  nwCfg.Tenant,
		NetworkName: nwCfg.NetworkName,
		EndpointID:  pluginHost,
		ConfigEP: intent.ConfigEP{
			Container: pluginHost,
			Host:      pluginHost,
		},
	}

	var mresp master.CreateEndpointResponse
	err = cluster.MasterPostReq("/plugin/createEndpoint", &mreq, &mresp)
	if err != nil {
		log.Errorf("master failed to create endpoint %s", err)
		return err
	}

	log.Infof("Got endpoint create resp from master: %+v", mresp)

	// Take lock to ensure netPlugin processes only one cmd at a time
	netPlugin.Lock()
	defer func() { netPlugin.Unlock() }()

	// Ask netplugin to create the endpoint
	netID := nwCfg.NetworkName + "." + nwCfg.Tenant
	err = netPlugin.CreateEndpoint(netID + "-" + pluginHost)
	if err != nil {
		log.Errorf("Endpoint creation failed. Error: %s", err)
		return err
	}

	// Assign IP to interface
	ipCIDR := fmt.Sprintf("%s/%d", mresp.EndpointConfig.IPAddress, nwCfg.SubnetLen)
	err = netutils.SetInterfaceIP(nwCfg.NetworkName, ipCIDR)
	if err != nil {
		log.Errorf("Could not assign ip: %s", err)
		return err
	}

	return nil
}

func processEpgEvent(netPlugin *plugin.NetPlugin, opts core.InstanceInfo, ID string, isDelete bool) error {
	log.Infof("Received processEpgEvent")
	var err error

	netPlugin.Lock()
	defer func() { netPlugin.Unlock() }()

	operStr := ""
	if isDelete {
		operStr = "delete"
	} else {
		err = netPlugin.UpdateEndpointGroup(ID)
		operStr = "update"
	}
	if err != nil {
		log.Errorf("Epg %s failed. Error: %s", operStr, err)
	} else {
		log.Infof("Epg %s succeeded", operStr)
	}

	return err
}

func processGlobalFwdModeUpdEvent(netPlugin *plugin.NetPlugin, opts core.InstanceInfo, fwdMode string) {

	// parse store URL
	parts := strings.Split(opts.DbURL, "://")
	if len(parts) < 2 {
		log.Fatalf("Invalid cluster-store-url %s", opts.DbURL)
	}
	stateStore := parts[0]
	// initialize the config
	pluginConfig := plugin.Config{
		Drivers: plugin.Drivers{
			Network: "ovs",
			State:   stateStore,
		},
		Instance: opts,
	}
	pluginConfig.Instance.FwdMode = fwdMode
	netPlugin.GlobalFwdModeUpdate(pluginConfig)

	for _, master := range cluster.MasterDB {
		netPlugin.AddMaster(core.ServiceInfo{
			HostAddr: master.HostAddr,
			Port:     9001, //netmasterRPCPort
		})
	}

	serviceList, _ := cluster.ObjdbClient.GetService("netplugin")
	for _, serviceInfo := range serviceList {
		if serviceInfo.HostAddr != opts.VtepIP {
			netPlugin.AddPeerHost(core.ServiceInfo{
				HostAddr: serviceInfo.HostAddr,
				Port:     4789, //vxlanUDPPort
			})
		}
	}

}

// Main loop to discover peer hosts and masters
func peerDiscoveryLoop(netplugin *plugin.NetPlugin, objdbClient objdb.API, localIP string) {
	// Create channels for watch thread
	nodeEventCh := make(chan objdb.WatchServiceEvent, 1)
	watchStopCh := make(chan bool, 1)
	masterEventCh := make(chan objdb.WatchServiceEvent, 1)
	masterWatchStopCh := make(chan bool, 1)

	// Start a watch on netmaster
	err := objdbClient.WatchService("netmaster", masterEventCh, masterWatchStopCh)
	if err != nil {
		log.Fatalf("Could not start a watch on netmaster service. Err: %v", err)
	}

	// Start a watch on netplugin service
	err = objdbClient.WatchService("netplugin", nodeEventCh, watchStopCh)
	if err != nil {
		log.Fatalf("Could not start a watch on netplugin service. Err: %v", err)
	}

	for {
		select {
		case srvEvent := <-nodeEventCh:
			log.Infof("Received netplugin service watch event: %+v", srvEvent)

			// collect the info about the node
			nodeInfo := srvEvent.ServiceInfo

			// check if its our own info coming back to us
			if nodeInfo.HostAddr == localIP {
				break
			}

			// Handle based on event type
			if srvEvent.EventType == objdb.WatchServiceEventAdd {
				log.Infof("Node add event for {%+v}", nodeInfo)

				// add the node
				err := netplugin.AddPeerHost(core.ServiceInfo{
					HostAddr: nodeInfo.HostAddr,
					Port:     ofnet.OFNET_AGENT_VXLAN_PORT,
				})
				if err != nil {
					log.Errorf("Error adding node {%+v}. Err: %v", nodeInfo, err)
				}
				// add the node
				err = netplugin.AddPeerHost(core.ServiceInfo{
					HostAddr: nodeInfo.HostAddr,
					Port:     ofnet.OFNET_AGENT_VLAN_PORT,
				})
				if err != nil {
					log.Errorf("Error adding node {%+v}. Err: %v", nodeInfo, err)
				}

			} else if srvEvent.EventType == objdb.WatchServiceEventDel {
				log.Infof("Node delete event for {%+v}", nodeInfo)

				// remove the node
				err := netplugin.DeletePeerHost(core.ServiceInfo{
					HostAddr: nodeInfo.HostAddr,
					Port:     ofnet.OFNET_AGENT_VXLAN_PORT,
				})
				if err != nil {
					log.Errorf("Error adding node {%+v}. Err: %v", nodeInfo, err)
				}
				// remove the node
				err = netplugin.DeletePeerHost(core.ServiceInfo{
					HostAddr: nodeInfo.HostAddr,
					Port:     ofnet.OFNET_AGENT_VLAN_PORT,
				})
				if err != nil {
					log.Errorf("Error adding node {%+v}. Err: %v", nodeInfo, err)
				}
			}
		case srvEvent := <-masterEventCh:
			log.Infof("Received netmaster service watch event: %+v", srvEvent)

			// collect the info about the node
			nodeInfo := srvEvent.ServiceInfo

			// Handle based on event type
			if srvEvent.EventType == objdb.WatchServiceEventAdd {
				log.Infof("Master add event for {%+v}", nodeInfo)

				// Add the master
				err := addMaster(netplugin, core.ServiceInfo{
					HostAddr: nodeInfo.HostAddr,
					Port:     ofnet.OFNET_MASTER_PORT,
				})
				if err != nil {
					log.Errorf("Error adding master {%+v}. Err: %v", nodeInfo, err)
				}
			} else if srvEvent.EventType == objdb.WatchServiceEventDel {
				log.Infof("Master delete event for {%+v}", nodeInfo)

				// Delete the master
				err := deleteMaster(netplugin, core.ServiceInfo{
					HostAddr: nodeInfo.HostAddr,
					Port:     nodeInfo.Port,
				})
//.........这里部分代码省略.........

// Main loop to discover peer hosts and masters
func peerDiscoveryLoop(netplugin *plugin.NetPlugin, objdbClient objdb.ObjdbApi, localIP string) {
	// Create channels for watch thread
	nodeEventCh := make(chan objdb.WatchServiceEvent, 1)
	watchStopCh := make(chan bool, 1)
	masterEventCh := make(chan objdb.WatchServiceEvent, 1)
	masterWatchStopCh := make(chan bool, 1)

	// Start a watch on netplugin service so that we dont miss any
	err := objdbClient.WatchService("netplugin", nodeEventCh, watchStopCh)
	if err != nil {
		log.Fatalf("Could not start a watch on netplugin service. Err: %v", err)
	}

	// Start a watch on netmaster too
	err = objdbClient.WatchService("netmaster", masterEventCh, masterWatchStopCh)
	if err != nil {
		log.Fatalf("Could not start a watch on netmaster service. Err: %v", err)
	}

	// Get a list of all existing netplugin nodes
	nodeList, err := objdbClient.GetService("netplugin")
	if err != nil {
		log.Errorf("Error getting node list from objdb. Err: %v", err)
	}

	log.Infof("Got netplugin service list: %+v", nodeList)

	// walk each node and add it as a PeerHost
	for _, node := range nodeList {
		// Ignore if its our own info
		if node.HostAddr == localIP {
			continue
		}
		// add the node
		err := netplugin.AddPeerHost(core.ServiceInfo{
			HostAddr: node.HostAddr,
			Port:     ofnet.OFNET_AGENT_PORT,
		})
		if err != nil {
			log.Errorf("Error adding node {%+v}. Err: %v", node, err)
		}
	}

	// Get a list of all existing netmasters
	masterList, err := objdbClient.GetService("netmaster")
	if err != nil {
		log.Errorf("Error getting master list from objdb. Err: %v", err)
	}

	log.Infof("Got netmaster service list: %+v", masterList)

	// Walk each master and add it
	for _, master := range masterList {
		// Add the master
		err := addMaster(netplugin, core.ServiceInfo{
			HostAddr: master.HostAddr,
			Port:     ofnet.OFNET_MASTER_PORT,
		})
		if err != nil {
			log.Errorf("Error adding master {%+v}. Err: %v", master, err)
		}
	}

	for {
		select {
		case srvEvent := <-nodeEventCh:
			log.Infof("Received netplugin service watch event: %+v", srvEvent)

			// collect the info about the node
			nodeInfo := srvEvent.ServiceInfo

			// check if its our own info coming back to us
			if nodeInfo.HostAddr == localIP {
				break
			}

			// Handle based on event type
			if srvEvent.EventType == objdb.WatchServiceEventAdd {
				log.Infof("Node add event for {%+v}", nodeInfo)

				// add the node
				err := netplugin.AddPeerHost(core.ServiceInfo{
					HostAddr: nodeInfo.HostAddr,
					Port:     ofnet.OFNET_AGENT_PORT,
				})
				if err != nil {
					log.Errorf("Error adding node {%+v}. Err: %v", nodeInfo, err)
				}
			} else if srvEvent.EventType == objdb.WatchServiceEventDel {
				log.Infof("Node delete event for {%+v}", nodeInfo)

				// remove the node
				err := netplugin.DeletePeerHost(core.ServiceInfo{
					HostAddr: nodeInfo.HostAddr,
					Port:     ofnet.OFNET_AGENT_PORT,
				})
				if err != nil {
					log.Errorf("Error adding node {%+v}. Err: %v", nodeInfo, err)
				}
//.........这里部分代码省略.........

// Main loop to discover peer hosts and masters
func peerDiscoveryLoop(netplugin *plugin.NetPlugin, objClient objdb.API, ctrlIP, vtepIP string) {
	// Create channels for watch thread
	nodeEventCh := make(chan objdb.WatchServiceEvent, 1)
	watchStopCh := make(chan bool, 1)
	masterEventCh := make(chan objdb.WatchServiceEvent, 1)
	masterWatchStopCh := make(chan bool, 1)

	// Start a watch on netmaster
	err := objClient.WatchService("netmaster.rpc", masterEventCh, masterWatchStopCh)
	if err != nil {
		log.Fatalf("Could not start a watch on netmaster service. Err: %v", err)
	}

	// Start a watch on netplugin service
	err = objClient.WatchService("netplugin.vtep", nodeEventCh, watchStopCh)
	if err != nil {
		log.Fatalf("Could not start a watch on netplugin service. Err: %v", err)
	}

	for {
		select {
		case srvEvent := <-nodeEventCh:
			log.Debugf("Received netplugin service watch event: %+v", srvEvent)

			// collect the info about the node
			nodeInfo := srvEvent.ServiceInfo

			// check if its our own info coming back to us
			if nodeInfo.HostAddr == vtepIP {
				break
			}

			// Handle based on event type
			if srvEvent.EventType == objdb.WatchServiceEventAdd {
				log.Infof("Node add event for {%+v}", nodeInfo)

				// add the node
				err := netplugin.AddPeerHost(core.ServiceInfo{
					HostAddr: nodeInfo.HostAddr,
					Port:     vxlanUDPPort,
				})
				if err != nil {
					log.Errorf("Error adding node {%+v}. Err: %v", nodeInfo, err)
				}
			} else if srvEvent.EventType == objdb.WatchServiceEventDel {
				log.Infof("Node delete event for {%+v}", nodeInfo)

				// remove the node
				err := netplugin.DeletePeerHost(core.ServiceInfo{
					HostAddr: nodeInfo.HostAddr,
					Port:     vxlanUDPPort,
				})
				if err != nil {
					log.Errorf("Error adding node {%+v}. Err: %v", nodeInfo, err)
				}
			}
		case srvEvent := <-masterEventCh:
			log.Infof("Received netmaster service watch event: %+v", srvEvent)

			// collect the info about the node
			nodeInfo := srvEvent.ServiceInfo

			// Handle based on event type
			if srvEvent.EventType == objdb.WatchServiceEventAdd {
				log.Infof("Master add event for {%+v}", nodeInfo)

				// Add the master
				err := addMaster(netplugin, nodeInfo)
				if err != nil {
					log.Errorf("Error adding master {%+v}. Err: %v", nodeInfo, err)
				}
			} else if srvEvent.EventType == objdb.WatchServiceEventDel {
				log.Infof("Master delete event for {%+v}", nodeInfo)

				// Delete the master
				err := deleteMaster(netplugin, nodeInfo)
				if err != nil {
					log.Errorf("Error deleting master {%+v}. Err: %v", nodeInfo, err)
				}
			}
		}

		// Dont process next peer event for another 100ms
		time.Sleep(100 * time.Millisecond)
	}
}