switches: added mqtt listening of events

src/switch/cmd/l2sm-add-port/main.go

+package main
+
+import (
+	"encoding/json"
+	"errors"
+	"flag"
+	"fmt"
+	"io/ioutil"
+	"ovs-switch/pkg/ovs"
+)
+
+type Node struct {
+	Name          string `json:"name"`
+	NodeIP        string `json:"nodeIP"`
+	NeighborNodes []Node `json:"neighborNodes"`
+}
+
+// Script that takes two required arguments:
+// the first one is the name in the cluster of the node where the script is running
+// the second one is the path to the configuration file, in reference to the code.
+func main() {
+
+	portName, err := takeArguments()
+
+	bridge := ovs.FromName("brtun")
+
+	if err != nil {
+		fmt.Println("Error with the arguments. Error:", err)
+		return
+	}
+
+	bridge.AddPort(portName)
+
+	if err != nil {
+		fmt.Println("Port not added: ", err)
+		return
+	}
+}
+
+func takeArguments() (string, error) {
+
+	portName := flag.String("port_name", "", "port you want to add. Required.")
+
+	flag.Parse()
+
+	if *portName == "" {
+		return "", errors.New("port name is not defined")
+
+	}
+
+	return *portName, nil
+}
+
+func createTopology(bridge ovs.Bridge, nodes []Node, nodeName string) error {
+
+	// Search for the corresponding node in the configuration, according to the first passed parameter.
+	// Once the node is found, create a bridge for every neighbour node defined.
+	// The bridge is created with the nodeIp and neighborNodeIP and VNI. The VNI is generated in the l2sm-controller thats why its set to 'flow'.
+	for _, node := range nodes {
+		if node.Name == nodeName {
+			//nodeIP := strings.TrimSpace(node.NodeIP)
+			connectToNeighbors(bridge, node)
+		}
+	}
+	return nil
+}
+
+func readFile(configDir string) ([]Node, error) {
+	/// Read file and save in memory the JSON info
+	data, err := ioutil.ReadFile(configDir)
+	if err != nil {
+		fmt.Println("No input file was found.", err)
+		return nil, err
+	}
+
+	var nodes []Node
+	err = json.Unmarshal(data, &nodes)
+	if err != nil {
+		return nil, err
+	}
+
+	return nodes, nil
+
+}
+
+func connectToNeighbors(bridge ovs.Bridge, node Node) error {
+	for vxlanNumber, neighbor := range node.NeighborNodes {
+		vxlanId := fmt.Sprintf("vxlan%d", vxlanNumber)
+		err := bridge.CreateVxlan(ovs.Vxlan{VxlanId: vxlanId, LocalIp: node.NodeIP, RemoteIp: neighbor.NodeIP, UdpPort: "7000"})
+
+		if err != nil {
+			return fmt.Errorf("could not create vxlan between node %s and node %s", node.Name, neighbor)
+		} else {
+			fmt.Printf("Created vxlan between node %s and node %s.\n", node.Name, neighbor)
+		}
+	}
+	return nil
+}

src/switch/cmd/l2sm-init/main.go

@@ -5,6 +5,7 @@ import (
 	"flag"
 	"fmt"
 	"os/exec"
+	"ovs-switch/pkg/ovs"
 	"regexp"
 )
 
@@ -21,9 +22,11 @@ func main() {
 	}
 
 	fmt.Println("Initializing switch, connected to controller: ", controllerIP)
-	err = initializeSwitch(controllerIP)
+
+	bridge, err := initializeSwitch(controllerIP)
 
 	if err != nil {
+
 		fmt.Println("Could not initialize switch. Error:", err)
 		return
 	}
@@ -33,12 +36,11 @@ func main() {
 	// Set all virtual interfaces up, and connect them to the tunnel bridge:
 	for i := 1; i <= vethNumber; i++ {
 		veth := fmt.Sprintf("net%d", i)
-		cmd := exec.Command("ip", "link", "set", veth, "up") // i.e: ip link set veth1 up
-		if err := cmd.Run(); err != nil {
+		if err := bridge.AddPort(veth); err != nil {
 			fmt.Println("Error:", err)
 		}
-		exec.Command("ovs-vsctl", "add-port", "brtun", veth).Run() // i.e: ovs-vsctl add-port brtun veth1
 	}
+	fmt.Printf("Switch initialized, current state: ", bridge)
 }
 
 func takeArguments() (int, string, error) {
@@ -56,7 +58,7 @@ func takeArguments() (int, string, error) {
 	return *vethNumber, *controllerIP, nil
 }
 
-func initializeSwitch(controllerIP string) error {
+func initializeSwitch(controllerIP string) (ovs.Bridge, error) {
 
 	re := regexp.MustCompile(`\b(?:[0-9]{1,3}\.){3}[0-9]{1,3}\b`)
 	if !re.MatchString(controllerIP) {
@@ -64,32 +66,9 @@ func initializeSwitch(controllerIP string) error {
 		controllerIP = re.FindString(string(out))
 	}
 
-	var err error
-
-	err = exec.Command("ovs-vsctl", "add-br", "brtun").Run()
+	controller := fmt.Sprintf("tcp:%s:6633", controllerIP)
 
-	if err != nil {
-		return errors.New("could not create brtun interface")
-	}
-
-	err = exec.Command("ip", "link", "set", "brtun", "up").Run()
-
-	if err != nil {
-		return errors.New("could not set brtun interface up")
-	}
+	bridge, err := ovs.NewBridge(ovs.Bridge{Name: "brtun", Controller: controller, Protocol: "OpenFlow13"})
 
-	err = exec.Command("ovs-vsctl", "set", "bridge", "brtun", "protocols=OpenFlow13").Run()
-
-	if err != nil {
-		return errors.New("could not set brtun messaing protocol to OpenFlow13")
-	}
-
-	target := fmt.Sprintf("tcp:%s:6633", controllerIP)
-
-	err = exec.Command("ovs-vsctl", "set-controller", "brtun", target).Run()
-
-	if err != nil {
-		return errors.New("could not connect to controller")
-	}
-	return nil
+	return bridge, err
 }

src/switch/cmd/l2sm-vxlans/main.go

@@ -7,14 +7,13 @@ import (
 	"fmt"
 	"io/ioutil"
 	"os"
-	"os/exec"
-	"strings"
+	"ovs-switch/pkg/ovs"
 )
 
 type Node struct {
-	Name          string   `json:"name"`
-	NodeIP        string   `json:"nodeIP"`
-	NeighborNodes []string `json:"neighborNodes"`
+	Name          string `json:"name"`
+	NodeIP        string `json:"nodeIP"`
+	NeighborNodes []Node `json:"neighborNodes"`
 }
 
 // Script that takes two required arguments:
@@ -22,14 +21,24 @@ type Node struct {
 // the second one is the path to the configuration file, in reference to the code.
 func main() {
 
-	configDir, nodeName, err := takeArguments()
+	configDir, nodeName, fileType, err := takeArguments()
+
+	bridge := ovs.FromName("brtun")
 
 	if err != nil {
 		fmt.Println("Error with the arguments. Error:", err)
 		return
 	}
 
-	err = createVxlans(configDir, nodeName)
+	nodes, err := readFile(configDir)
+
+	switch fileType {
+	case "topology":
+		err = createTopology(bridge, nodes, nodeName)
+
+	case "neighbors":
+		err = connectToNeighbors(bridge, nodes[0])
+	}
 
 	if err != nil {
 		fmt.Println("Vxlans not created: ", err)
@@ -37,73 +46,68 @@ func main() {
 	}
 }
 
-func takeArguments() (string, string, error) {
+func takeArguments() (string, string, string, error) {
 	configDir := os.Args[len(os.Args)-1]
 
 	nodeName := flag.String("node_name", "", "name of the node the script is executed in. Required.")
 
+	fileType := flag.String("file_type", "topology", "type of filed passed as an argument. Can either be topology or neighbors. Default: topology.")
+
 	flag.Parse()
 
 	switch {
 	case *nodeName == "":
-		return "", "", errors.New("node name is not defined")
+		return "", "", "", errors.New("node name is not defined")
+	case *fileType != "topology" || *fileType != "neighbors":
+		return "", "", "", errors.New("file type not supported. Available types: 'topology' and 'neighbors'")
 	case configDir == "":
-		return "", "", errors.New("config directory is not defined")
+		return "", "", "", errors.New("config directory is not defined")
 	}
 
-	return configDir, *nodeName, nil
+	return configDir, *nodeName, *fileType, nil
 }
 
-func createVxlans(configDir, nodeName string) error {
+func createTopology(bridge ovs.Bridge, nodes []Node, nodeName string) error {
+
+	// Search for the corresponding node in the configuration, according to the first passed parameter.
+	// Once the node is found, create a bridge for every neighbour node defined.
+	// The bridge is created with the nodeIp and neighborNodeIP and VNI. The VNI is generated in the l2sm-controller thats why its set to 'flow'.
+	for _, node := range nodes {
+		if node.Name == nodeName {
+			//nodeIP := strings.TrimSpace(node.NodeIP)
+			connectToNeighbors(bridge, node)
+		}
+	}
+	return nil
+}
 
+func readFile(configDir string) ([]Node, error) {
 	/// Read file and save in memory the JSON info
 	data, err := ioutil.ReadFile(configDir)
 	if err != nil {
 		fmt.Println("No input file was found.", err)
-		return err
+		return nil, err
 	}
 
 	var nodes []Node
 	err = json.Unmarshal(data, &nodes)
 	if err != nil {
-		return err
+		return nil, err
 	}
 
-	// Search for the corresponding node in the configuration, according to the first passed parameter.
-	// Once the node is found, create a bridge for every neighbour node defined.
-	// The bridge is created with the nodeIp and neighborNodeIP and VNI. The VNI is generated in the l2sm-controller thats why its set to 'flow'.
-	for _, node := range nodes {
-		if node.Name == nodeName {
-			nodeIP := strings.TrimSpace(node.NodeIP)
-			for _, neighbor := range node.NeighborNodes {
-				vxlanNumber := 1
-				for _, n := range nodes {
-					if n.Name == neighbor {
-						neighborIP := strings.TrimSpace(n.NodeIP)
-						commandArgs := []string{
-							"add-port",
-							"brtun",
-							fmt.Sprintf("vxlan%d", vxlanNumber),
-							"--",
-							"set", "interface",
-							fmt.Sprintf("vxlan%d", vxlanNumber),
-							"type=vxlan",
-							"options:key=flow",
-							fmt.Sprintf("options:remote_ip=%s", neighborIP),
-							fmt.Sprintf("options:local_ip=%s", nodeIP),
-							"options:dst_port=7000",
-						}
-						_, err := exec.Command("ovs-vsctl", commandArgs...).Output()
-						if err != nil {
-							return fmt.Errorf("could not create vxlan between node %s and node %s", node.Name, neighbor)
-						} else {
-							fmt.Printf("Created vxlan between node %s and node %s.\n", node.Name, neighbor)
-						}
-					}
-					vxlanNumber++
-				}
-
-			}
+	return nodes, nil
+
+}
+
+func connectToNeighbors(bridge ovs.Bridge, node Node) error {
+	for vxlanNumber, neighbor := range node.NeighborNodes {
+		vxlanId := fmt.Sprintf("vxlan%d", vxlanNumber)
+		err := bridge.CreateVxlan(ovs.Vxlan{VxlanId: vxlanId, LocalIp: node.NodeIP, RemoteIp: neighbor.NodeIP, UdpPort: "7000"})
+
+		if err != nil {
+			return fmt.Errorf("could not create vxlan between node %s and node %s", node.Name, neighbor)
+		} else {
+			fmt.Printf("Created vxlan between node %s and node %s.\n", node.Name, neighbor)
 		}
 	}
 	return nil

src/switch/go.mod

-module app
+module ovs-switch
 
-go 1.18
+go 1.21.7
+
+require github.com/eclipse/paho.mqtt.golang v1.4.3
+
+require (
+	github.com/gorilla/websocket v1.5.0 // indirect
+	golang.org/x/net v0.8.0 // indirect
+	golang.org/x/sync v0.1.0 // indirect
+)

src/switch/go.sum

+github.com/eclipse/paho.mqtt.golang v1.4.3 h1:2kwcUGn8seMUfWndX0hGbvH8r7crgcJguQNCyp70xik=
+github.com/eclipse/paho.mqtt.golang v1.4.3/go.mod h1:CSYvoAlsMkhYOXh/oKyxa8EcBci6dVkLCbo5tTC1RIE=
+github.com/gorilla/websocket v1.5.0 h1:PPwGk2jz7EePpoHN/+ClbZu8SPxiqlu12wZP/3sWmnc=
+github.com/gorilla/websocket v1.5.0/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
+golang.org/x/net v0.8.0 h1:Zrh2ngAOFYneWTAIAPethzeaQLuHwhuBkuV6ZiRnUaQ=
+golang.org/x/net v0.8.0/go.mod h1:QVkue5JL9kW//ek3r6jTKnTFis1tRmNAW2P1shuFdJc=
+golang.org/x/sync v0.1.0 h1:wsuoTGHzEhffawBOhz5CYhcrV4IdKZbEyZjBMuTp12o=
+golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=

src/switch/pkg/mqtt_sub/mqtt_sub.go

+package main
+
+import (
+	"encoding/json"
+	"log"
+
+	ovs "ovs-switch/pkg/ovs"
+
+	mqtt "github.com/eclipse/paho.mqtt.golang"
+)
+
+type Command struct {
+	Action string     `json:"action"`
+	Bridge ovs.Bridge `json:"bridge,omitempty"`
+	Port   string     `json:"port,omitempty"`
+	Vxlan  ovs.Vxlan  `json:"vxlan,omitempty"`
+}
+
+var broker = "tcp://mqtt_broker_ip:1883"
+var topic = "ovs/commands"
+
+func LaunchSubscriber() {
+	opts := mqtt.NewClientOptions().AddBroker(broker).SetClientID("go_mqtt_client")
+	opts.SetDefaultPublishHandler(func(client mqtt.Client, msg mqtt.Message) {
+		handleMessage(msg.Payload())
+	})
+
+	client := mqtt.NewClient(opts)
+	if token := client.Connect(); token.Wait() && token.Error() != nil {
+		log.Fatal(token.Error())
+	}
+
+	if token := client.Subscribe(topic, 0, nil); token.Wait() && token.Error() != nil {
+		log.Fatal(token.Error())
+	}
+
+	// Block main routine forever
+	select {}
+}
+
+func handleMessage(payload []byte) {
+	var cmd Command
+	err := json.Unmarshal(payload, &cmd)
+	if err != nil {
+		log.Printf("Error decoding command: %v", err)
+		return
+	}
+
+	switch cmd.Action {
+	case "create_bridge":
+		bridge, err := ovs.NewBridge(cmd.Bridge)
+		if err != nil {
+			log.Printf("Error creating bridge: %v", err)
+		} else {
+			log.Printf("Bridge created: %+v", bridge)
+		}
+	case "add_port":
+		bridge := ovs.FromName(cmd.Bridge.Name)
+		err := bridge.AddPort(cmd.Port)
+		if err != nil {
+			log.Printf("Error adding port: %v", err)
+		} else {
+			log.Printf("Port added: %s", cmd.Port)
+		}
+	case "create_vxlan":
+		bridge := ovs.FromName(cmd.Bridge.Name)
+		err := bridge.CreateVxlan(cmd.Vxlan)
+		if err != nil {
+			log.Printf("Error creating VXLAN: %v", err)
+		} else {
+			log.Printf("VXLAN created: %+v", cmd.Vxlan)
+		}
+	default:
+		log.Printf("Unknown action: %s", cmd.Action)
+	}
+}

src/switch/pkg/ovs/vsctl.go

+package ovs
+
+import (
+	"bytes"
+	"errors"
+	"fmt"
+	"os/exec"
+	"strings"
+)
+
+// TODO: Abstract exec client as a separate entity, that doesnt use CLI. The following presented hardcoded way is not clean.
+
+type Port struct {
+	Name   string
+	Status string
+}
+
+type Bridge struct {
+	Controller string
+	Name       string
+	Protocol   string
+	Ports      []Port
+}
+
+type Vxlan struct {
+	VxlanId  string
+	LocalIp  string
+	RemoteIp string
+	UdpPort  string
+}
+
+func FromName(bridgeName string) Bridge {
+
+	bridge := Bridge{Name: bridgeName}
+
+	bridge.getPorts()
+
+	return bridge
+}
+
+func NewBridge(bridgeConf Bridge) (Bridge, error) {
+
+	var err error
+
+	bridge := Bridge{}
+
+	err = exec.Command("ovs-vsctl", "add-br", bridgeConf.Name).Run()
+
+	if err != nil {
+		return bridge, errors.New("could not create brtun interface")
+	}
+
+	bridge.Name = bridgeConf.Name
+
+	err = exec.Command("ip", "link", "set", bridge.Name, "up").Run()
+
+	if err != nil {
+		return bridge, errors.New("could not set brtun interface up")
+	}
+
+	protocolString := fmt.Sprintf("protocols=%s", bridgeConf.Protocol)
+	err = exec.Command("ovs-vsctl", "set", "bridge", "brtun", protocolString).Run()
+
+	if err != nil {
+		return bridge, errors.New("could not set brtun messaing protocol to OpenFlow13")
+	}
+
+	bridge.Protocol = bridgeConf.Protocol
+
+	err = exec.Command("ovs-vsctl", "set-controller", bridge.Name, bridgeConf.Controller).Run()
+
+	if err != nil {
+		return bridge, errors.New("could not connect to controller")
+	}
+
+	bridge.Controller = bridgeConf.Name
+
+	return bridge, nil
+}
+
+func (bridge *Bridge) CreateVxlan(vxlan Vxlan) error {
+
+	commandArgs := []string{
+		"add-port",
+		bridge.Name,
+		vxlan.VxlanId,
+		"--",
+		"set", "interface",
+		vxlan.VxlanId,
+		"type=vxlan",
+		"options:key=flow",
+		fmt.Sprintf("options:remote_ip=%s", vxlan.RemoteIp),
+		fmt.Sprintf("options:local_ip=%s", vxlan.LocalIp),
+		fmt.Sprintf("options:dst_port=%s", vxlan.UdpPort),
+	}
+	_, err := exec.Command("ovs-vsctl", commandArgs...).Output()
+
+	return err
+
+}
+
+func (bridge *Bridge) AddPort(portName string) error {
+
+	cmd := exec.Command("ip", "link", "set", portName, "up") // i.e: ip link set veth1 up
+	if err := cmd.Run(); err != nil {
+		return err
+	}
+	exec.Command("ovs-vsctl", "add-port", bridge.Name, portName).Run() // i.e: ovs-vsctl add-port brtun veth1
+	bridge.Ports = append(bridge.Ports, Port{Name: portName, Status: "UP"})
+	return nil
+}
+
+func (bridge *Bridge) getPorts() error {
+	// Executes the ovs-vsctl command to list ports on the bridge
+	cmd := exec.Command("ovs-vsctl", "list-ports", bridge.Name)
+	var out bytes.Buffer
+	cmd.Stdout = &out
+	err := cmd.Run()
+	if err != nil {
+		return err
+	}
+
+	// Split the output by lines for each port name
+	portNames := strings.Split(out.String(), "\n")
+	for _, portName := range portNames {
+		if portName == "" {
+			continue
+		}
+		// TODO:, retrieve more details for each port; here we just set the name
+		port := Port{Name: portName}
+
+		// Retrieve status
+		// cmd = exec.Command("ovs-vsctl", "get", "Interface", portName, "status")
+
+		// Add the port to the Ports slice
+		bridge.Ports = append(bridge.Ports, port)
+	}
+
+	return nil
+}