Merge pull request contiv#680 from DivyaVavili/upd-mbr-links

upgrade uplink changes

Author: Joji Mekkatt

drivers/ovsSwitch.go

@@ -19,6 +19,8 @@ import (
 	"errors"
 	"fmt"
 	"net"
+	"reflect"
+	"sort"
 	"strings"
 	"time"
 
@@ -572,6 +574,55 @@ func (sw *OvsSwitch) DeleteVtep(vtepIP string) error {
 	return sw.ovsdbDriver.DeleteVtep(intfName)
 }
 
+func (sw *OvsSwitch) cleanupOldUplinkState(portName string, intfList []string) (bool, error) {
+	var err error
+	var oldUplinkIntf []string
+	portCreateReq := true
+
+	// Check if uplink is already created
+	// Case 1: Bonded ports - port name is the uplinkName
+	// Case 2: Indiviual port - port name is the interface name
+	portPresent := sw.ovsdbDriver.IsPortNamePresent(portName)
+	if portPresent {
+		/* If port already exists, make sure it has the same member links
+		   If not, a cleanup is required */
+		sort.Strings(intfList)
+		oldUplinkIntf = sw.ovsdbDriver.GetInterfacesInPort(portName)
+		if reflect.DeepEqual(intfList, oldUplinkIntf) {
+			log.Warnf("Uplink already part of %s", sw.bridgeName)
+			portCreateReq = false
+		} else {
+			log.Warnf("Deleting old uplink bond with intfs: %+v", oldUplinkIntf)
+			err = sw.ovsdbDriver.DeletePortBond(portName, oldUplinkIntf)
+			if err == nil {
+				portCreateReq = true
+			}
+		}
+		return portCreateReq, err
+	}
+
+	if len(intfList) == 1 && sw.ovsdbDriver.IsPortNamePresent(intfList[0]) {
+		// Uplink port already part of switch. No change required.
+		log.Debugf("Uplink intf %s already part of %s", intfList[0], sw.bridgeName)
+		return false, nil
+	}
+
+	/* Cleanup any other individual ports that may exist */
+	for _, intf := range intfList {
+		if sw.ovsdbDriver.IsIntfNamePresent(intf) {
+			log.Infof("Deleting old uplink port: %+v", intf)
+			err = sw.ovsdbDriver.DeletePort(intf)
+			if err != nil {
+				break
+			}
+			portCreateReq = true
+		}
+	}
+
+	// No cleanup done and new port creation required
+	return portCreateReq, err
+}
+
 // AddUplink adds uplink port(s) to the OVS
 func (sw *OvsSwitch) AddUplink(uplinkName string, intfList []string) error {
 	var err error
@@ -583,16 +634,23 @@ func (sw *OvsSwitch) AddUplink(uplinkName string, intfList []string) error {
 		log.Fatalf("Can not add uplink to OVS type %s.", sw.netType)
 	}
 
-	// Check if port is already part of the OVS and add it
-	if !sw.ovsdbDriver.IsPortNamePresent(uplinkName) {
+	createUplink, err := sw.cleanupOldUplinkState(uplinkName, intfList)
+	if err != nil {
+		log.Errorf("Could not cleanup previous uplink state")
+		return err
+	}
+
+	if createUplink {
 		if len(intfList) > 1 {
+			log.Debugf("Creating uplink port bond: %s with intf: %+v", uplinkName, intfList)
 			err = sw.ovsdbDriver.CreatePortBond(intfList, uplinkName)
 			if err != nil {
 				log.Errorf("Error adding uplink %s to OVS. Err: %v", intfList, err)
 				return err
 			}
 			uplinkType = ofnet.BondType
 		} else {
+			log.Debugf("Creating uplink port: %s", intfList[0])
 			// Ask OVSDB driver to add the port as a trunk port
 			err = sw.ovsdbDriver.CreatePort(intfList[0], "", uplinkName, 0, 0, 0)
 			if err != nil {
@@ -601,14 +659,14 @@ func (sw *OvsSwitch) AddUplink(uplinkName string, intfList []string) error {
 			}
 			uplinkType = ofnet.PortType
 		}
-	}
 
-	// HACK: When an uplink is added to OVS, it disconnects the controller connection.
-	//       This is a hack to workaround this issue. We wait for the OVS to reconnect
-	//       to the controller.
-	// Wait for a while for OVS switch to disconnect/connect to ofnet agent
-	time.Sleep(time.Second)
-	sw.ofnetAgent.WaitForSwitchConnection()
+		// HACK: When an uplink is added to OVS, it disconnects the controller connection.
+		//       This is a hack to workaround this issue. We wait for the OVS to reconnect
+		//       to the controller.
+		// Wait for a while for OVS switch to disconnect/connect to ofnet agent
+		time.Sleep(time.Second)
+		sw.ofnetAgent.WaitForSwitchConnection()
+	}
 
 	uplinkInfo := ofnet.PortInfo{
 		Name: uplinkName,
@@ -686,16 +744,31 @@ func (sw *OvsSwitch) HandleLinkUpdates(linkUpd ofnet.LinkUpdateInfo) {
 func (sw *OvsSwitch) RemoveUplinks() error {
 
 	var err error
+
 	// some error checking
 	if sw.netType != "vlan" {
 		log.Fatalf("Can not remove uplink from OVS type %s.", sw.netType)
 	}
+
 	for intfListObj := range sw.uplinkDb.IterBuffered() {
 		intfList := intfListObj.Val.([]string)
 		portName := intfListObj.Key
-		if !sw.ovsdbDriver.IsPortNamePresent(portName) {
+
+		// Remove uplink from agent
+		err = sw.ofnetAgent.RemoveUplink(portName)
+		if err != nil {
+			log.Errorf("Error removing uplink %s. Err: %v", portName, err)
+			return err
+		}
+		log.Infof("Removed uplink %s from ofnet", portName)
+
+		isPortPresent := sw.ovsdbDriver.IsPortNamePresent(portName)
+		if len(intfList) == 1 {
+			isPortPresent = sw.ovsdbDriver.IsPortNamePresent(intfList[0])
+		}
+		if isPortPresent {
 			if len(intfList) == 1 {
-				err = sw.ovsdbDriver.DeletePort(portName)
+				err = sw.ovsdbDriver.DeletePort(intfList[0])
 			} else {
 				err = sw.ovsdbDriver.DeletePortBond(portName, intfList)
 			}
@@ -705,17 +778,11 @@ func (sw *OvsSwitch) RemoveUplinks() error {
 			}
 		}
 		time.Sleep(time.Second)
-
-		// Remove uplink from agent
-		err = sw.ofnetAgent.RemoveUplink(portName)
-		if err != nil {
-			log.Errorf("Error removing uplink %s. Err: %v", portName, err)
-			return err
-		}
 		sw.uplinkDb.Remove(portName)
 
-		log.Infof("Removed uplink %s from OVS switch %s.", portName, sw.bridgeName)
+		log.Infof("Removed uplink %s(%+v) from OVS switch %s.", portName, intfList, sw.bridgeName)
 	}
+
 	return nil
 }
 

drivers/ovsdbDriver.go

@@ -19,6 +19,7 @@ import (
 	"errors"
 	"fmt"
 	"reflect"
+	"sort"
 	"sync"
 	"time"
 
@@ -370,6 +371,57 @@ func (d *OvsdbDriver) CreatePort(intfName, intfType, id string, tag, burst int,
 	return d.performOvsdbOps(operations)
 }
 
+// GetInterfacesInPort gets list of interfaces in a port in sorted order
+func (d *OvsdbDriver) GetInterfacesInPort(portName string) []string {
+	var intfList []string
+	d.cacheLock.RLock()
+	defer d.cacheLock.RUnlock()
+
+	for _, row := range d.cache["Port"] {
+		name := row.Fields["name"].(string)
+		if name == portName {
+			// Port found
+			// Iterate over the list of interfaces
+			switch (row.Fields["interfaces"]).(type) {
+			case libovsdb.UUID: // Individual interface case
+				intfUUID := row.Fields["interfaces"].(libovsdb.UUID)
+				intfInfo := d.GetIntfInfo(intfUUID)
+				if reflect.DeepEqual(intfInfo, libovsdb.Row{}) {
+					log.Errorf("could not find interface with UUID: %+v", intfUUID)
+					break
+				}
+				intfList = append(intfList, intfInfo.Fields["name"].(string))
+			case libovsdb.OvsSet: // Port bond case
+				intfUUIDList := row.Fields["interfaces"].(libovsdb.OvsSet)
+				for _, intfUUID := range intfUUIDList.GoSet {
+					intfInfo := d.GetIntfInfo(intfUUID.(libovsdb.UUID))
+					if reflect.DeepEqual(intfInfo, libovsdb.Row{}) {
+						continue
+					}
+					intfList = append(intfList, intfInfo.Fields["name"].(string))
+				}
+			}
+			sort.Strings(intfList)
+			break
+		}
+	}
+	return intfList
+}
+
+// GetIntfInfo gets interface information from "Interface" table
+func (d *OvsdbDriver) GetIntfInfo(uuid libovsdb.UUID) libovsdb.Row {
+	d.cacheLock.RLock()
+	defer d.cacheLock.RUnlock()
+
+	for intfUUID, row := range d.cache["Interface"] {
+		if intfUUID == uuid {
+			return row
+		}
+	}
+
+	return libovsdb.Row{}
+}
+
 //CreatePortBond creates port bond in OVS
 func (d *OvsdbDriver) CreatePortBond(intfList []string, bondName string) error {
 
@@ -720,13 +772,38 @@ func (d *OvsdbDriver) IsControllerPresent(target string) bool {
 }
 
 // IsPortNamePresent checks if port already exists in OVS bridge
-func (d *OvsdbDriver) IsPortNamePresent(intfName string) bool {
+func (d *OvsdbDriver) IsPortNamePresent(portName string) bool {
 	d.cacheLock.RLock()
 	defer d.cacheLock.RUnlock()
 
 	// walk the local cache
 	for tName, table := range d.cache {
 		if tName == "Port" {
+			for _, row := range table {
+				for fieldName, value := range row.Fields {
+					if fieldName == "name" {
+						if value == portName {
+							// Port name exists.
+							return true
+						}
+					}
+				}
+			}
+		}
+	}
+
+	// We could not find the port name
+	return false
+}
+
+// IsIntfNamePresent checks if intf already exists in OVS bridge
+func (d *OvsdbDriver) IsIntfNamePresent(intfName string) bool {
+	d.cacheLock.RLock()
+	defer d.cacheLock.RUnlock()
+
+	// walk the local cache
+	for tName, table := range d.cache {
+		if tName == "Interface" {
 			for _, row := range table {
 				for fieldName, value := range row.Fields {
 					if fieldName == "name" {

drivers/ovsdriver.go

@@ -190,7 +190,7 @@ func (d *OvsDriver) Init(info *core.InstanceInfo) error {
 
 	// Add uplink to VLAN switch
 	if len(info.UplinkIntf) != 0 {
-		err = d.switchDb["vlan"].AddUplink("uplinkBond", info.UplinkIntf)
+		err = d.switchDb["vlan"].AddUplink("uplinkPort", info.UplinkIntf)
 		if err != nil {
 			log.Errorf("Could not add uplink %v to vlan OVS. Err: %v", info.UplinkIntf, err)
 		}

drivers/ovsdriver_test.go

@@ -518,7 +518,7 @@ func TestOvsDriverUplinkBridgeMode(t *testing.T) {
 		t.Fatalf("Could not add uplink %+v to vlan OVS. Err: %v", uplinkPorts, err)
 	}
 
-	time.Sleep(300 * time.Millisecond)
+	time.Sleep(time.Second)
 
 	// verify uplink port
 	output, err := exec.Command("ovs-vsctl", "list", "Port").CombinedOutput()

test/systemtests/docker_test.go

@@ -468,6 +468,7 @@ func (d *docker) cleanupSlave() {
 	vNode := d.node.tbnode
 	vNode.RunCommand("sudo ovs-vsctl del-br contivVxlanBridge")
 	vNode.RunCommand("sudo ovs-vsctl del-br contivVlanBridge")
+	vNode.RunCommand("sudo ovs-vsctl del-br contivHostBridge")
 	vNode.RunCommand("for p in `ifconfig  | grep vport | awk '{print $1}'`; do sudo ip link delete $p type veth; done")
 	vNode.RunCommand("sudo rm /var/run/docker/plugins/netplugin.sock")
 	vNode.RunCommand("sudo service docker restart")

test/systemtests/k8setup_test.go

@@ -486,6 +486,7 @@ func (k *kubernetes) cleanupSlave() {
 	vNode := k.node.tbnode
 	vNode.RunCommand("sudo ovs-vsctl del-br contivVxlanBridge")
 	vNode.RunCommand("sudo ovs-vsctl del-br contivVlanBridge")
+	vNode.RunCommand("sudo ovs-vsctl del-br contivHostBridge")
 	vNode.RunCommand("for p in `ifconfig  | grep vport | awk '{print $1}'`; do sudo ip link delete $p type veth; done")
 	vNode.RunCommand("sudo rm /var/run/docker/plugins/netplugin.sock")
 	vNode.RunCommand("sudo service docker restart")

test/systemtests/node_test.go

@@ -110,6 +110,38 @@ func (n *node) cleanupMaster() {
 	n.exec.cleanupMaster()
 }
 
+func (n *node) verifyUplinkState(uplinks []string) error {
+	var err error
+	var portName string
+	var cmd, output string
+
+	if len(uplinks) > 1 {
+		portName = "uplinkPort"
+	} else {
+		portName = uplinks[0]
+	}
+
+	// Verify port state
+	cmd = fmt.Sprintf("sudo ovs-vsctl find Port name=%s", portName)
+	output, err = n.runCommand(cmd)
+	if err != nil || !(strings.Contains(string(output), portName)) {
+		err = fmt.Errorf("Lookup failed for uplink Port %s. Err: %+v", portName, err)
+		return err
+	}
+
+	// Verify Interface state
+	for _, uplink := range uplinks {
+		cmd = fmt.Sprintf("sudo ovs-vsctl find Interface name=%s", uplink)
+		output, err = n.runCommand(cmd)
+		if err != nil || !(strings.Contains(string(output), uplink)) {
+			err = fmt.Errorf("Lookup failed for uplink interface %s for uplink cfg:%+v. Err: %+v", uplink, uplinks, err)
+			return err
+		}
+	}
+
+	return err
+}
+
 func (n *node) runCommand(cmd string) (string, error) {
 	var (
 		str string

test/systemtests/swarm_test.go

@@ -430,6 +430,7 @@ func (w *swarm) cleanupSlave() {
 	vNode := w.node.tbnode
 	vNode.RunCommand("sudo ovs-vsctl del-br contivVxlanBridge")
 	vNode.RunCommand("sudo ovs-vsctl del-br contivVlanBridge")
+	vNode.RunCommand("sudo ovs-vsctl del-br contivHostBridge")
 	vNode.RunCommand("for p in `ifconfig  | grep vport | awk '{print $1}'`; do sudo ip link delete $p type veth; done")
 }