[goclidote] Add AddDevicesToAntidote() function

The AddDevicesToAntidote89 function gracefully adds to AntidoteDB a
[]cnml.DeviceIpv4sGraphserver list of devices reusing a single
connection, hence reducing the operation time.
Signed-off-by: Roger Pueyo Centelles <rpueyo@ac.upc.edu>

goclidote/goclidote.go

@@ -502,6 +502,133 @@ func RemoveDevicesFromAntidote(oldDevices []cnml.DeviceIpv4sGraphserver, Databas
 	return remDevices
 }
 
+// AddDevicesToAntidote adds an array of DeviceIpv4sGraphservers from AntidoteDB
+func AddDevicesToAntidote(newDevices []cnml.DeviceIpv4sGraphserver, DatabaseHost string, DatabasePort int) []cnml.DeviceIpv4sGraphserver {
+	var addedDevices []cnml.DeviceIpv4sGraphserver
+
+	// Start an AntidoteDB client
+	client, err := antidoteclient.NewClient(antidoteclient.Host{Name: DatabaseHost, Port: DatabasePort})
+	errCheck(err, ERROR, "[AddDevicesToAntidote]: Unable to create a new AntidoteDB go client")
+	defer client.Close()
+
+	// Iterate all the new devices
+	for _, v := range newDevices {
+
+		// Start a transaction
+		tx, err := client.StartTransaction()
+		errCheck(err, ERROR, "[AddDevicesToAntidote]: Unable to start a transaction using the go client")
+
+		devError := false
+
+		// Add the device ID to the devices set in the guifi bucket
+		bucketName := "guifi"
+		setName := "devices"
+		itemValue := fmt.Sprintf("%d", v.ID)
+
+		bucket := antidoteclient.Bucket{Bucket: []byte(bucketName)}
+		key := antidoteclient.Key(setName)
+
+		err = bucket.Update(tx, antidoteclient.SetAdd(key, []byte(itemValue)))
+		errCheck(err, ERROR, fmt.Sprintf("[AddDevicesToAntidote]: Unable to add item %s from set %s in bucket %s", itemValue, setName, bucketName))
+
+		setVal, err := bucket.ReadSet(tx, key)
+		errCheck(err, ERROR, fmt.Sprintf("[AddDevicesToAntidote]: Unable to read set %s in bucket %s", setName, bucketName))
+
+		itemInSet := false
+
+		// Iterate the updated set to see if the item was added
+		for _, w := range setVal {
+			if string(w[:]) == itemValue {
+				itemInSet = true
+			}
+		}
+
+		if !itemInSet {
+			devError = true
+		}
+
+		// Add the graphserver register to the device-id bucket
+		bucketName = fmt.Sprintf("device-%d", v.ID)
+		registerName := "graphserver"
+		itemValue = fmt.Sprintf("%d", v.GraphServer)
+
+		bucket = antidoteclient.Bucket{Bucket: []byte(bucketName)}
+		key = antidoteclient.Key(registerName)
+
+		err = bucket.Update(tx, antidoteclient.RegPut(key, []byte(itemValue)))
+		errCheck(err, ERROR, fmt.Sprintf("[AddDevicesToAntidote]: Unable to set a value %s to register %s in bucket %s", itemValue, registerName, bucketName))
+
+		regVal, err := bucket.ReadReg(tx, key)
+		errCheck(err, ERROR, fmt.Sprintf("[AddDevicesToAntidote]: Unable to read register %s in bucket %s", registerName, bucketName))
+
+		if string(regVal[:]) != itemValue {
+			devError = true
+		}
+
+		// Add the IPv4 addresses to the ipv4s set in the device-id bucket
+		for _, w := range v.Addresses {
+			setName := "ipv4s"
+			itemValue = w
+
+			key = antidoteclient.Key(setName)
+
+			err = bucket.Update(tx, antidoteclient.SetAdd(key, []byte(itemValue)))
+			errCheck(err, ERROR, fmt.Sprintf("[AddDevicesToAntidote]: Unable to add item %s from set %s in bucket %s", itemValue, setName, bucketName))
+
+			setVal, err := bucket.ReadSet(tx, key)
+			errCheck(err, ERROR, fmt.Sprintf("[AddDevicesToAntidote]: Unable to read set %s in bucket %s", setName, bucketName))
+
+			itemInSet := false
+
+			for _, x := range setVal {
+				if string(x[:]) == itemValue {
+					itemInSet = true
+				}
+			}
+
+			if !itemInSet {
+				devError = true
+			}
+		}
+
+		// Add the SNMP names to the snmpnames set in the device-id bucket
+		for _, w := range v.SNMPNames {
+			setName := "snmpnames"
+			itemValue := w
+
+			key = antidoteclient.Key(setName)
+
+			err = bucket.Update(tx, antidoteclient.SetAdd(key, []byte(itemValue)))
+			errCheck(err, ERROR, fmt.Sprintf("[AddDevicesToAntidote]: Unable to add item %s from set %s in bucket %s", itemValue, setName, bucketName))
+
+			setVal, err := bucket.ReadSet(tx, key)
+			errCheck(err, ERROR, fmt.Sprintf("[AddDevicesToAntidote]: Unable to read set %s in bucket %s", setName, bucketName))
+
+			itemInSet := false
+
+			for _, x := range setVal {
+				if string(x[:]) == itemValue {
+					itemInSet = true
+				}
+			}
+
+			if !itemInSet {
+				devError = true
+			}
+		}
+
+		if devError == false {
+			// Device successfully removed
+			addedDevices = append(addedDevices, v)
+		}
+
+		err = tx.Commit()
+		errCheck(err, ERROR, "[AddDevicesToAntidote]: Unable to commit transaction")
+	}
+
+	return addedDevices
+}
+
 // The errCheck function gracefully manages errors, warnings, etc.
 func errCheck(e error, level int, message string) {
 

monitor-fetch-and-upload/monitor-fetch-and-upload.go

@@ -237,6 +237,7 @@ func main() {
 
 			if v.ID == w.ID {
 				devFound = true
+				break
 			}
 		}
 
@@ -293,45 +294,72 @@ func main() {
 	fmt.Println("")
 
 	// Add or update those devices that are in the CNML file
-	for k, v := range cnmlDevsIpv4sGraphs {
+	var newDevs []cnml.DeviceIpv4sGraphserver
+	var curDevs []cnml.DeviceIpv4sGraphserver
+
+	for _, v := range cnmlDevsIpv4sGraphs {
 		// Check if the device is in Antidote already and add it if missing
 		if !isDeviceIpv4sGraphserverInArray(v, keepAntidoteDevices) && len(v.Addresses) > 0 {
-			saveDevFail := false
-			// Add the device to the devices set
-			if saveDeviceIpv4sGraphserverToAntidote(v) {
-				antidoteAddDeviceSuccess++
+			newDevs = append(newDevs, v)
 		} else {
-				antidoteAddDeviceFail++
-				saveDevFail = true
+			curDevs = append(curDevs, v)
+		}
 	}
 
-			if !saveDevFail {
-				// Save the default graph server
-				if saveGraphserverToDeviceIpv4sGraphserverToAntidote(v) {
-					antidoteAddGraphserverSuccess++
-				} else {
-					antidoteAddGraphserverFail++
+	addDevs := goclidote.AddDevicesToAntidote(newDevs, dbHost, dbPort)
+
+	for k, v := range newDevs {
+		devFound := false
+		for _, w := range addDevs {
+
+			if v.ID == w.ID {
+				devFound = true
+				break
+			}
 		}
 
-				// Iterate through all the device's addresses and add them if new
-				for _, w := range v.Addresses {
-					if saveIpv4addressToDeviceIpv4sGraphserverToAntidote(v, w) {
+		if devFound {
+			// Added the device to AntidoteDB
+			antidoteAddDeviceSuccess++
+			// Added the device's graphserver to AntidoteDB
+			antidoteAddGraphserverSuccess++
+			// Added the device's IPv4 addresses from AntidoteDB
+			for _ = range v.Addresses {
 				antidoteAddAddressSuccess++
+			}
+			// Removed the device's SNMP names from AntidoteDB
+			for _ = range v.SNMPNames {
+				antidoteAddSNMPNameSuccess++
+			}
 		} else {
+			// Failed to add the device to AntidoteDB
+			antidoteAddDeviceFail++
+			// Failed to add the device's graphserver to AntidoteDB
+			antidoteAddGraphserverFail++
+			// Failed to add the device's IPv4 addresses to AntidoteDB
+			for _ = range v.Addresses {
 				antidoteAddAddressFail++
 			}
+			// Failed to add the device's SNMP names to AntidoteDB
+			for _ = range v.SNMPNames {
+				antidoteAddSNMPNameFail++
+			}
 		}
 
-				// Iterate through all the device's SNMP names and add them if new
-				for _, w := range v.SNMPNames {
-					if saveSNMPNameToDeviceIpv4sGraphserverToAntidote(v, w) {
-						antidoteAddSNMPNameSuccess++
+		if k == 0 {
+			fmt.Printf("%d devices added to AntidoteDB (%d success, %d fail)\n", antidoteAddDeviceSuccess+antidoteAddDeviceFail, antidoteAddDeviceSuccess, antidoteAddDeviceFail)
+			fmt.Printf("%d graphservers added or updated to AntidoteDB (%d success, %d fail)\n", antidoteAddGraphserverSuccess+antidoteAddGraphserverFail, antidoteAddGraphserverSuccess, antidoteAddGraphserverFail)
+			fmt.Printf("%d IPv4 addresses added or updated to AntidoteDB (%d success, %d fail)\n", antidoteAddAddressSuccess+antidoteAddAddressFail, antidoteAddAddressSuccess, antidoteAddAddressFail)
+			fmt.Printf("%d SNMP names added or updated to AntidoteDB (%d success, %d fail)\n", antidoteAddSNMPNameSuccess+antidoteAddSNMPNameFail, antidoteAddSNMPNameSuccess, antidoteAddSNMPNameFail)
 		} else {
-						antidoteAddSNMPNameFail++
-					}
+			fmt.Printf("\033[4A\r%d devices added to AntidoteDB (%d success, %d fail)\n", antidoteAddDeviceSuccess+antidoteAddDeviceFail, antidoteAddDeviceSuccess, antidoteAddDeviceFail)
+			fmt.Printf("\r%d graphservers added or updated to AntidoteDB (%d success, %d fail)\n", antidoteAddGraphserverSuccess+antidoteAddGraphserverFail, antidoteAddGraphserverSuccess, antidoteAddGraphserverFail)
+			fmt.Printf("\r%d IPv4 addresses added or updated to AntidoteDB (%d success, %d fail)\n", antidoteAddAddressSuccess+antidoteAddAddressFail, antidoteAddAddressSuccess, antidoteAddAddressFail)
+			fmt.Printf("\r%d SNMP names added or updated to AntidoteDB (%d success, %d fail)\n", antidoteAddSNMPNameSuccess+antidoteAddSNMPNameFail, antidoteAddSNMPNameSuccess, antidoteAddSNMPNameFail)
 		}
 	}
-		} else {
+
+	for k, v := range curDevs {
 		// The device is in AntidoteDB; check if it has changed
 		if v.GraphServer != keepAntidoteDevices[posDeviceInSlice(v, keepAntidoteDevices)].GraphServer {
 			// Save the new default graph server
@@ -365,7 +393,6 @@ func main() {
 				}
 			}
 		}
-		}
 
 		if k == 0 {
 			fmt.Printf("%d devices added to AntidoteDB (%d success, %d fail)\n", antidoteAddDeviceSuccess+antidoteAddDeviceFail, antidoteAddDeviceSuccess, antidoteAddDeviceFail)