Merge pull request #228 from yunkon-kim/track-agent-status

Track the state of cb-network agent

poc-cb-net/cmd/admin-web/admin-web.go

@@ -232,17 +232,17 @@ func handleTestSpecification(etcdClient *clientv3.Client, responseText []byte) {
 
 	for _, kv := range resp.Kvs {
 
-		var hostRule model.HostRule
+		var peer model.Peer
 		CBLogger.Tracef("Key : %v", kv.Key)
-		CBLogger.Tracef("The host rule: %v", kv.Value)
+		CBLogger.Tracef("The peer: %v", kv.Value)
 
-		err := json.Unmarshal(kv.Value, &hostRule)
+		err := json.Unmarshal(kv.Value, &peer)
 		if err != nil {
 			CBLogger.Error(err)
 		}
 
 		// Put the evaluation specification of the CLADNet to the etcd
-		keyControlCommand := fmt.Sprint(etcdkey.ControlCommand + "/" + hostRule.CLADNetID + "/" + hostRule.HostID)
+		keyControlCommand := fmt.Sprint(etcdkey.ControlCommand + "/" + peer.CLADNetID + "/" + peer.HostID)
 		CBLogger.Tracef("keyControlCommand: \"%s\"", keyControlCommand)
 
 		strStatusTestSpecification, _ := json.Marshal(testSpecification)
@@ -280,18 +280,18 @@ func handleControlCommand(etcdClient *clientv3.Client, responseText string) {
 	}
 
 	for _, kv := range resp.Kvs {
+		key := string(kv.Key)
+		CBLogger.Tracef("Key : %v", key)
+		CBLogger.Tracef("The peer: %v", kv.Value)
 
-		var hostRule model.HostRule
-		CBLogger.Tracef("Key : %v", kv.Key)
-		CBLogger.Tracef("The host rule: %v", kv.Value)
-
-		err := json.Unmarshal(kv.Value, &hostRule)
+		var peer model.Peer
+		err := json.Unmarshal(kv.Value, &peer)
 		if err != nil {
 			CBLogger.Error(err)
 		}
 
 		// Put the evaluation specification of the CLADNet to the etcd
-		keyControlCommand := fmt.Sprint(etcdkey.ControlCommand + "/" + hostRule.CLADNetID + "/" + hostRule.HostID)
+		keyControlCommand := fmt.Sprint(etcdkey.ControlCommand + "/" + peer.CLADNetID + "/" + peer.HostID)
 
 		cmdMessageBody := cmd.BuildCommandMessage(controlCommand, controlCommandOption)
 		CBLogger.Tracef("%#v", cmdMessageBody)
@@ -322,7 +322,7 @@ func getExistingNetworkInfo(etcdClient *clientv3.Client) error {
 		CBLogger.Debug("Send a networking rule of CLADNet to AdminWeb frontend")
 
 		// Build the response bytes of a networking rule
-		responseBytes := buildResponseBytes("HostRule", string(kv.Value))
+		responseBytes := buildResponseBytes("peer", string(kv.Value))
 
 		// Send the networking rule to the front-end
 		CBLogger.Debug("Send the networking rule to AdminWeb frontend")
@@ -480,11 +480,11 @@ func watchNetworkingRule(wg *sync.WaitGroup, etcdClient *clientv3.Client) {
 			case mvccpb.PUT: // The watched value has changed.
 				CBLogger.Tracef("Watch - %s %q : %q", event.Type, event.Kv.Key, event.Kv.Value)
 
-				hostRule := event.Kv.Value
-				CBLogger.Tracef("A host rule of CLADNet: %v", hostRule)
+				peer := event.Kv.Value
+				CBLogger.Tracef("A peer of CLADNet: %v", peer)
 
 				// Build the response bytes of the networking rule
-				responseBytes := buildResponseBytes("HostRule", string(hostRule))
+				responseBytes := buildResponseBytes("peer", string(peer))
 
 				// Send the networking rule to the front-end
 				CBLogger.Debug("Send the networking rule to AdminWeb frontend")

poc-cb-net/cmd/agent/agent.go

@@ -103,6 +103,7 @@ func handleCommand(command string, commandOption string, etcdClient *clientv3.Cl
 	CBLogger.Tracef("CommandOption: %+v", commandOption)
 	switch command {
 	case "suspend":
+		updatePeerState(model.Suspended, etcdClient)
 		CBNet.Shutdown()
 
 	case "resume":
@@ -151,8 +152,26 @@ func watchNetworkingRule(etcdClient *clientv3.Client) {
 		for _, event := range watchResponse.Events {
 			CBLogger.Tracef("Watch - %s %q : %q", event.Type, event.Kv.Key, event.Kv.Value)
 
+			key := string(event.Kv.Key)
+			peerBytes := event.Kv.Value
+
 			// Update a host's configuration in the networking rule
-			CBNet.UpdateHostRule(event.Kv.Value)
+			isThisPeerInitialized := CBNet.UpdatePeer(peerBytes)
+
+			if isThisPeerInitialized {
+				var peer model.Peer
+				if err := json.Unmarshal(event.Kv.Value, &peer); err != nil {
+					CBLogger.Error(err)
+				}
+				peer.State = model.Running
+
+				CBLogger.Debugf("Put \"%v\"", key)
+				doc, _ := json.Marshal(peer)
+
+				if _, err := etcdClient.Put(context.TODO(), key, string(doc)); err != nil {
+					CBLogger.Error(err)
+				}
+			}
 		}
 	}
 	CBLogger.Debug("End.........")
@@ -192,23 +211,59 @@ func compareAndSwapHostNetworkInformation(etcdClient *clientv3.Client) {
 	if txnResp.Succeeded {
 		// Set the networking rule to the host
 		for _, kv := range txnResp.Responses[0].GetResponseRange().Kvs {
-			respKey := kv.Key
-			slicedKeys := strings.Split(string(respKey), "/")
-			parsedHostID := slicedKeys[len(slicedKeys)-1]
-			CBLogger.Tracef("ParsedHostID: %v", parsedHostID)
+			key := string(kv.Key)
+			CBLogger.Tracef("Key: %v", key)
 
-			hostRule := kv.Value
-			CBLogger.Tracef("A host's configuration: %v", hostRule)
+			peerBytes := kv.Value
+			CBLogger.Tracef("A host's configuration: %v", peerBytes)
 			CBLogger.Debug("Update a host's configuration")
 
 			// Update a host's configuration in the networking rule
-			CBNet.UpdateHostRule(hostRule)
+			isThisPeerInitialized := CBNet.UpdatePeer(peerBytes)
+
+			if isThisPeerInitialized {
+				var peer model.Peer
+				if err := json.Unmarshal(peerBytes, &peer); err != nil {
+					CBLogger.Error(err)
+				}
+				peer.State = model.Running
+
+				CBLogger.Debugf("Put \"%v\"", key)
+				doc, _ := json.Marshal(peer)
+
+				if _, err := etcdClient.Put(context.TODO(), key, string(doc)); err != nil {
+					CBLogger.Error(err)
+				}
+			}
 		}
 	}
 
 	CBLogger.Debug("End.........")
 }
 
+func updatePeerState(state string, etcdClient *clientv3.Client) {
+
+	idx := CBNet.NetworkingRule.GetIndexOfHostID(CBNet.HostID)
+
+	peer := &model.Peer{
+		CLADNetID:          CBNet.NetworkingRule.CLADNetID,
+		HostID:             CBNet.NetworkingRule.HostID[idx],
+		PrivateIPv4Network: CBNet.NetworkingRule.HostIPv4Network[idx],
+		PrivateIPv4Address: CBNet.NetworkingRule.HostIPAddress[idx],
+		PublicIPv4Address:  CBNet.NetworkingRule.PublicIPAddress[idx],
+		State:              state,
+	}
+
+	keyNetworkingRuleOfPeer := fmt.Sprint(etcdkey.NetworkingRule + "/" + CBNet.ID + "/" + CBNet.HostID)
+
+	CBLogger.Debugf("Put \"%v\"", keyNetworkingRuleOfPeer)
+	doc, _ := json.Marshal(peer)
+
+	if _, err := etcdClient.Put(context.TODO(), keyNetworkingRuleOfPeer, string(doc)); err != nil {
+		CBLogger.Error(err)
+	}
+}
+
 func watchSecret(etcdClient *clientv3.Client) {
 	CBLogger.Debug("Start.........")
 

poc-cb-net/cmd/controller/controller.go

@@ -166,7 +166,7 @@ func watchHostNetworkInformation(wg *sync.WaitGroup, etcdClient *clientv3.Client
 					}
 
 					// Create a key of host in the specific CLADNet's networking rule
-					keyNetworkingRuleOfHost := fmt.Sprint(etcdkey.NetworkingRule + "/" + parsedCLADNetID + "/" + parsedHostID)
+					keyNetworkingRuleOfPeer := fmt.Sprint(etcdkey.NetworkingRule + "/" + parsedCLADNetID + "/" + parsedHostID)
 
 					// // Needed?? (not sure yet)
 					// lock := concurrency.NewMutex(session, keyNetworkingRuleOfHost)
@@ -179,13 +179,13 @@ func watchHostNetworkInformation(wg *sync.WaitGroup, etcdClient *clientv3.Client
 					// CBLogger.Debugf("Acquired lock for '%v'", keyNetworkingRuleOfHost)
 
 					// Get a host's networking rule
-					CBLogger.Tracef("Key: %v", keyNetworkingRuleOfHost)
-					respRule, respRuleErr := etcdClient.Get(context.TODO(), keyNetworkingRuleOfHost)
+					CBLogger.Tracef("Key: %v", keyNetworkingRuleOfPeer)
+					respRule, respRuleErr := etcdClient.Get(context.TODO(), keyNetworkingRuleOfPeer)
 					if respRuleErr != nil {
 						CBLogger.Error(respRuleErr)
 					}
 
-					var hostRule model.HostRule
+					var peer model.Peer
 
 					// Newly allocate the host's configuration
 					if respRule.Count == 0 {
@@ -200,27 +200,28 @@ func watchHostNetworkInformation(wg *sync.WaitGroup, etcdClient *clientv3.Client
 						}
 
 						hostIPv4Network, hostIPAddress := assignIPAddressToHost(cladnetIpv4AddressSpace, uint32(resp.Count+2))
-						hostRule = model.HostRule{
+						peer = model.Peer{
 							CLADNetID:          parsedCLADNetID,
 							HostID:             parsedHostID,
 							PrivateIPv4Network: hostIPv4Network,
 							PrivateIPv4Address: hostIPAddress,
 							PublicIPv4Address:  hostNetworkInformation.PublicIP,
+							State:              model.Suspended,
 						}
 
 					} else { // Update the host's configuration
 
-						if err := json.Unmarshal(respRule.Kvs[0].Value, &hostRule); err != nil {
+						if err := json.Unmarshal(respRule.Kvs[0].Value, &peer); err != nil {
 							CBLogger.Error(err)
 						}
 
-						hostRule.PublicIPv4Address = hostNetworkInformation.PublicIP
+						peer.PublicIPv4Address = hostNetworkInformation.PublicIP
 					}
 
-					CBLogger.Debugf("Put \"%v\"", keyNetworkingRuleOfHost)
-					doc, _ := json.Marshal(hostRule)
+					CBLogger.Debugf("Put \"%v\"", keyNetworkingRuleOfPeer)
+					doc, _ := json.Marshal(peer)
 
-					if _, err := etcdClient.Put(context.TODO(), keyNetworkingRuleOfHost, string(doc)); err != nil {
+					if _, err := etcdClient.Put(context.TODO(), keyNetworkingRuleOfPeer, string(doc)); err != nil {
 						CBLogger.Error(err)
 					}
 

poc-cb-net/internal/cb-network/cb-network.go

@@ -265,47 +265,49 @@ func (cbnetwork CBNetwork) GetHostNetworkInformation() model.HostNetworkInformat
 	return temp
 }
 
-func (cbnetwork *CBNetwork) updateNetworkingRule(hostRule model.HostRule) {
+func (cbnetwork *CBNetwork) updateNetworkingRule(peer model.Peer) {
 	CBLogger.Debug("Start.........")
 
 	CBLogger.Debug("Lock to update the networking rule")
 	mutex.Lock()
-	cbnetwork.NetworkingRule.UpdateRule(hostRule.HostID, hostRule.PrivateIPv4Network, hostRule.PrivateIPv4Address, hostRule.PublicIPv4Address)
+	cbnetwork.NetworkingRule.CLADNetID = peer.CLADNetID
+	cbnetwork.NetworkingRule.UpdateRule(peer.HostID, peer.PrivateIPv4Network, peer.PrivateIPv4Address, peer.PublicIPv4Address)
 	CBLogger.Debug("Unlock to update the networking rule")
 	mutex.Unlock()
 
 	CBLogger.Debug("End.........")
 }
 
-// UpdateHostRule represents a function to decode binary of networking rule and set it.
-func (cbnetwork *CBNetwork) UpdateHostRule(value []byte) {
+// UpdatePeer represents a function to decode binary of networking rule and set it.
+func (cbnetwork *CBNetwork) UpdatePeer(value []byte) (isThisPeerInitialized bool) {
 	CBLogger.Debug("Start.........")
 
-	var hostRule model.HostRule
-
-	err := json.Unmarshal(value, &hostRule)
+	var peer model.Peer
+	err := json.Unmarshal(value, &peer)
 	if err != nil {
 		CBLogger.Error(err)
 	}
 
-	prettyJSON, _ := json.MarshalIndent(hostRule, "", "\t")
+	prettyJSON, _ := json.MarshalIndent(peer, "", "\t")
 	CBLogger.Trace("Pretty JSON")
 	CBLogger.Trace(string(prettyJSON))
 
-	cbnetwork.updateNetworkingRule(hostRule)
+	cbnetwork.updateNetworkingRule(peer)
 
-	if hostRule.HostID == cbnetwork.HostID {
+	if peer.HostID == cbnetwork.HostID {
 		if !cbnetwork.isInterfaceConfigured {
 			err := cbnetwork.configureCBNetworkInterface()
 			if err != nil {
 				CBLogger.Error(err)
-				return
+				return false
 			}
 			cbnetwork.isInterfaceConfigured = true
 			cbnetwork.notificationChannel <- cbnetwork.isInterfaceConfigured
+			return true
 		}
 	}
 	CBLogger.Debug("End.........")
+	return false
 }
 
 func (cbnetwork *CBNetwork) configureCBNetworkInterface() error {
@@ -555,6 +557,7 @@ func (cbnetwork *CBNetwork) Shutdown() {
 	// [To be improved] Stop tunneling routines
 	// Currently just return func() when an error occur
 
+	cbnetwork.runIP("link", "set", "dev", cbnetwork.name, "down")
 	cbnetwork.Interface.Close()
 	cbnetwork.isInterfaceConfigured = false
 

poc-cb-net/internal/cb-network/model/networking-rule.go

@@ -66,7 +66,7 @@ func (netrule *NetworkingRule) AppendRule(ID string, privateIPv4Network string,
 func (netrule *NetworkingRule) UpdateRule(id string, privateIPv4Network string, privateIPv4Address string, publicIPv4Address string) {
 	CBLogger.Infof("A rule: {%s, %s, %s, %s}", id, privateIPv4Network, privateIPv4Address, publicIPv4Address)
 	if netrule.Contain(id) { // If HostID exists, update rule
-		index := netrule.GetIndexOfID(id)
+		index := netrule.GetIndexOfHostID(id)
 		if privateIPv4Network != "" {
 			netrule.HostIPv4Network[index] = privateIPv4Network
 		}
@@ -79,8 +79,8 @@ func (netrule *NetworkingRule) UpdateRule(id string, privateIPv4Network string,
 	}
 }
 
-// GetIndexOfID represents a function to find and return an index of HostID from NetworkingRule
-func (netrule NetworkingRule) GetIndexOfID(id string) int {
+// GetIndexOfHostID represents a function to find and return an index of HostID from NetworkingRule
+func (netrule NetworkingRule) GetIndexOfHostID(id string) int {
 	return netrule.find(netrule.HostID, id)
 }
 

poc-cb-net/internal/cb-network/model/peer.go

@@ -0,0 +1,19 @@
+package cbnet
+
+const (
+	// Running is const for the running state
+	Running = "running"
+
+	// Suspended is const for the suspended state
+	Suspended = "suspended"
+)
+
+// Peer represents a host's rule of the cloud adaptive network.
+type Peer struct {
+	CLADNetID          string `json:"cladNetID"`
+	HostID             string `json:"hostID"`
+	PrivateIPv4Network string `json:"privateIPv4Network"`
+	PrivateIPv4Address string `json:"privateIPv4Address"`
+	PublicIPv4Address  string `json:"publicIPv4Address"`
+	State              string `json:"state"`
+}