zoreide/orchestrator.go

package main

import (
	"fmt"
	"log"
	"regexp"
	"slices"
	"strconv"
	"time"
)

func (mip *AbstractMulticast) AddUniqueAndSort(hier int64) {

	if slices.Contains(mip.HierarchyArray, hier) {
		log.Println("Element already in the array: ", hier)
	} else {
		mip.HierarchyArray = append(mip.HierarchyArray, hier)
		log.Println("Here the numbers known:", mip.HierarchyArray)
	}

}

func (mip *AbstractMulticast) IsAlpha(hier int64) bool {

	alpha := slices.Max(mip.HierarchyArray)
	log.Println("Maximum element is :", alpha)

	return alpha == hier

}

func (mip *AbstractMulticast) WriteNumberToMulticast(br AbstractBridge) {

	defer func() {
		if r := recover(); r != nil {
			fmt.Println("An error happened in <WriteNumberToMulticast()>, but Zoreide recovered. ")
			fmt.Println("Error was: ", r)
		}
	}()

	log.Println("Initiating ticker")

	bstNumber := fmt.Sprintf("%d", br.hIerarchyNumber)

	for range time.Tick(1 * time.Second) {

		_, err := mip.Wconn.Write([]byte(bstNumber))
		if err != nil {
			log.Println("Cannot write to multicast:" + err.Error())
		}
	}

}

func (mip *AbstractMulticast) ReadNumberFromMulticast() {

	defer func() {
		if r := recover(); r != nil {
			fmt.Println("An error happened in <ReadNumberToMulticast()>, but Zoreide recovered. ")
			fmt.Println("Error was: ", r)
		}
	}()

	log.Println("Initiating reader")
	buffer := make([]byte, mip.MaxDatagramSize)

	// Loop forever reading from the socket
	for {

		_, _, err := mip.Rconn.ReadFromUDP(buffer)
		if err != nil {
			log.Println("ReadFromUDP failed:", err)
		}

		BstChannel <- string(buffer)

	}

}

func (b *AbstractBridge) HierarchyReLocator(entity AbstractMulticast) {

	defer func() {
		if r := recover(); r != nil {
			fmt.Println("An error happened in <HierarchyLocator()>, but Zoreide recovered. ")
			fmt.Println("Error was: ", r)
		}
	}()

	log.Println("Inizializing HierarchyManager")

	entity.AddUniqueAndSort(b.hIerarchyNumber)

	re := regexp.MustCompile("[0-9]+")

	for bstNumber := range BstChannel {

		cleanStr := re.FindAllString(bstNumber, -1)
		if cleanStr == nil {
			log.Println("No numbers in multicast: ", bstNumber)
			continue
		} else {
			log.Println("Extracted numbers in multicast: ", cleanStr)
			bstNumber = cleanStr[0]
		}

		brdNumber, err := strconv.ParseInt(bstNumber, 10, 64)
		if err != nil {
			log.Println("Garbage received on multicast: ", bstNumber)
			log.Println("Cannot convert to int64:", err.Error())
			continue
		} else {
			log.Println("Adding received:", brdNumber)
			entity.AddUniqueAndSort(brdNumber)
		}

		// finished feeding the new number
		// if Alpha:
		if entity.IsAlpha(b.hIerarchyNumber) {
			if b.IsAssigned() {
				log.Println("Still ALPHA. This is ok.")
			} else {
				log.Println("I'm the new ALPHA! Get out my path, losers!")
				b.configureIpAndBridgeUp()
				log.Println("Ip is active: ", b.IsActive())
			}
			// here we manage the case when we are not alpha
		} else {
			log.Println("GULP! There is a bigger one, better descalating")
			if b.IsAssigned() {
				log.Println("Start removing the IP from the interface:")
				b.removeIPandBridgeInt()
				log.Println("Removed.Ip is reachable: ", b.IsActive())
			} else {
				log.Println("Nothing to do, since IP is not assigned to the interface: ", b.ExistingInterface)
				log.Println("Ip is reachable: ", b.IsActive())
			}
		}

	}

}

func (b *AbstractBridge) WaitAndClean(entity AbstractMulticast) {

	defer func() {
		if r := recover(); r != nil {
			fmt.Println("An error happened in <WaitAndClean()>, but Zoreide recovered. ")
			fmt.Println("Error was: ", r)
		}
	}()

	log.Println("Inizializing HA-Manager")

	for {

		pollTime := len(entity.HierarchyArray) + 1
		time.Sleep(time.Duration(pollTime) * time.Second)
		// Evitiamo di avere l'IP senza essere alpha
		if b.IsAssigned() {
			if entity.IsAlpha(b.hIerarchyNumber) {
				log.Println("We are alpha and IP is assigned. All ok")
			} else {
				log.Println("Cannot have IP assigned without being alpha")
				log.Println("Inconsistent situation: start removing the IP from the interface:")
				b.removeIPandBridgeInt()
				log.Println("Removed.Ip is still reachable: ", b.IsActive())
			}
		}

		if b.IsActive() {
			log.Println("Ip reachable, cluster OK")
			continue
		} else {
			log.Println("Situation non consistent, restart from green field.")
			entity.HierarchyArray = entity.HierarchyArray[:0]
			entity.HierarchyArray = slices.Clip(entity.HierarchyArray)
			entity.AddUniqueAndSort(b.hIerarchyNumber)
			log.Println("Restarted from green field: ", entity.HierarchyArray)
		}

	}
}