From 832fe2f0b06e7d4fd8a8fd4b336acf40b9b4998b Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Thu, 30 Nov 2017 19:47:31 +0000 Subject: [PATCH 1/3] Update local copy of annotations even after error e.g. if we get a clash because someone else has updated the data, it returns the new data so we should work off that next time. Also re-fetch the Annotations every time round the main loop for safety. --- prog/kube-peers/annotations.go | 38 ++++++++++++++-------------------- prog/kube-peers/main.go | 3 +++ 2 files changed, 19 insertions(+), 22 deletions(-) diff --git a/prog/kube-peers/annotations.go b/prog/kube-peers/annotations.go index dd8b2db8..86915618 100644 --- a/prog/kube-peers/annotations.go +++ b/prog/kube-peers/annotations.go @@ -130,46 +130,40 @@ func (cml *configMapAnnotations) UpdatePeerList(list peerList) error { return cml.UpdateAnnotation(KubePeersAnnotationKey, string(recordBytes)) } -func (cml *configMapAnnotations) UpdateAnnotation(key, value string) error { +func (cml *configMapAnnotations) UpdateAnnotation(key, value string) (err error) { if cml.cm == nil { return errors.New("endpoint not initialized, call Init first") } - cm := cml.cm - cm.Annotations[key] = value - cm, err := cml.Client.ConfigMaps(cml.Namespace).Update(cml.cm) - if err == nil { - cml.cm = cm - } + // speculatively change the state, then replace with whatever comes back + // from Update(), which will be the latest on the server whatever happened + cml.cm.Annotations[key] = value + cml.cm, err = cml.Client.ConfigMaps(cml.Namespace).Update(cml.cm) return err } -func (cml *configMapAnnotations) RemoveAnnotation(key string) error { +func (cml *configMapAnnotations) RemoveAnnotation(key string) (err error) { if cml.cm == nil { return errors.New("endpoint not initialized, call Init first") } - cm := cml.cm - delete(cm.Annotations, key) - cm, err := cml.Client.ConfigMaps(cml.Namespace).Update(cml.cm) - if err == nil { - cml.cm = cm - } + // speculatively change the state, then replace with whatever comes back + // from Update(), which will be the latest on the server whatever happened + delete(cml.cm.Annotations, key) + cml.cm, err = cml.Client.ConfigMaps(cml.Namespace).Update(cml.cm) return err } -func (cml *configMapAnnotations) RemoveAnnotationsWithValue(valueToRemove string) error { +func (cml *configMapAnnotations) RemoveAnnotationsWithValue(valueToRemove string) (err error) { if cml.cm == nil { return errors.New("endpoint not initialized, call Init first") } - cm := cml.cm - for key, value := range cm.Annotations { + // speculatively change the state, then replace with whatever comes back + // from Update(), which will be the latest on the server whatever happened + for key, value := range cml.cm.Annotations { if value == valueToRemove { - delete(cm.Annotations, key) + delete(cml.cm.Annotations, key) } } - cm, err := cml.Client.ConfigMaps(cml.Namespace).Update(cml.cm) - if err == nil { - cml.cm = cm - } + cml.cm, err = cml.Client.ConfigMaps(cml.Namespace).Update(cml.cm) return err } diff --git a/prog/kube-peers/main.go b/prog/kube-peers/main.go index 03a15225..1961f70c 100644 --- a/prog/kube-peers/main.go +++ b/prog/kube-peers/main.go @@ -98,6 +98,9 @@ func checkIamInPeerList(cml *configMapAnnotations, c *kubernetes.Clientset, peer // Kubernetes, remove it from Weave IPAM func reclaimRemovedPeers(weave *weaveapi.Client, cml *configMapAnnotations, nodes []nodeInfo, myPeerName string) error { for { + if err := cml.Init(); err != nil { + return err + } // 1. Compare peers stored in the peerList against all peers reported by k8s now. storedPeerList, err := cml.GetPeerList() if err != nil { From 333f4a1d3cba68c9b556b443a08da88142e3c65e Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Thu, 30 Nov 2017 19:50:42 +0000 Subject: [PATCH 2/3] Make sure Annotation names conform to Kubernetes requirements. E.g. replace "26:c6:0b:ba:fd:31" with "26_c6_0b_ba_fd_31" --- prog/kube-peers/annotations.go | 29 +++++++++++++++++++++++++---- prog/kube-peers/main.go | 6 +++--- 2 files changed, 28 insertions(+), 7 deletions(-) diff --git a/prog/kube-peers/annotations.go b/prog/kube-peers/annotations.go index 86915618..20b45cdd 100644 --- a/prog/kube-peers/annotations.go +++ b/prog/kube-peers/annotations.go @@ -73,8 +73,10 @@ const ( retryPeriod = time.Second * 2 jitterFactor = 1.0 + // Prefix all our annotation keys with this string so they don't clash with anyone else's + KubePeersPrefix = "kube-peers.weave.works/" // KubePeersAnnotationKey is the default annotation key - KubePeersAnnotationKey = "kube-peers.weave.works/peers" + KubePeersAnnotationKey = KubePeersPrefix + "peers" ) func (cml *configMapAnnotations) Init() error { @@ -130,13 +132,32 @@ func (cml *configMapAnnotations) UpdatePeerList(list peerList) error { return cml.UpdateAnnotation(KubePeersAnnotationKey, string(recordBytes)) } +// Clean up a string so it meets the Kubernetes requiremements for Annotation keys: +// name part must consist of alphanumeric characters, '-', '_' or '.', and must +// start and end with an alphanumeric character (e.g. 'MyName', or 'my.name', or '123-abc') +func cleanKey(key string) string { + buf := []byte(key) + for i, c := range buf { + if (c >= 'A' && c <= 'Z') || (c >= 'a' && c <= 'z') || (c >= '0' && c <= '9') || c == '-' || c == '_' || c == '.' || c == '/' { + continue + } + buf[i] = '_' + } + return string(buf) +} + +func (cml *configMapAnnotations) GetAnnotation(key string) (string, bool) { + value, ok := cml.cm.Annotations[cleanKey(key)] + return value, ok +} + func (cml *configMapAnnotations) UpdateAnnotation(key, value string) (err error) { if cml.cm == nil { return errors.New("endpoint not initialized, call Init first") } // speculatively change the state, then replace with whatever comes back // from Update(), which will be the latest on the server whatever happened - cml.cm.Annotations[key] = value + cml.cm.Annotations[cleanKey(key)] = value cml.cm, err = cml.Client.ConfigMaps(cml.Namespace).Update(cml.cm) return err } @@ -147,7 +168,7 @@ func (cml *configMapAnnotations) RemoveAnnotation(key string) (err error) { } // speculatively change the state, then replace with whatever comes back // from Update(), which will be the latest on the server whatever happened - delete(cml.cm.Annotations, key) + delete(cml.cm.Annotations, cleanKey(key)) cml.cm, err = cml.Client.ConfigMaps(cml.Namespace).Update(cml.cm) return err } @@ -160,7 +181,7 @@ func (cml *configMapAnnotations) RemoveAnnotationsWithValue(valueToRemove string // from Update(), which will be the latest on the server whatever happened for key, value := range cml.cm.Annotations { if value == valueToRemove { - delete(cml.cm.Annotations, key) + delete(cml.cm.Annotations, key) // don't need to clean this key as it came from the map } } cml.cm, err = cml.Client.ConfigMaps(cml.Namespace).Update(cml.cm) diff --git a/prog/kube-peers/main.go b/prog/kube-peers/main.go index 1961f70c..8872a4a6 100644 --- a/prog/kube-peers/main.go +++ b/prog/kube-peers/main.go @@ -122,7 +122,7 @@ func reclaimRemovedPeers(weave *weaveapi.Client, cml *configMapAnnotations, node common.Log.Debugln("[kube-peers] Preparing to remove disappeared peer", peer) okToRemove := false // 3. Check if there is an existing annotation with key X - if existingAnnotation, found := cml.cm.Annotations[peer.PeerName]; found { + if existingAnnotation, found := cml.GetAnnotation(KubePeersPrefix + peer.PeerName); found { common.Log.Debugln("[kube-peers] Existing annotation", existingAnnotation) // 4. If annotation already contains my identity, ok; if existingAnnotation == myPeerName { @@ -131,7 +131,7 @@ func reclaimRemovedPeers(weave *weaveapi.Client, cml *configMapAnnotations, node } else { // 5. If non-existent, write an annotation with key X and contents "my identity" common.Log.Debugln("[kube-peers] Noting I plan to remove ", peer.PeerName) - if err := cml.UpdateAnnotation(peer.PeerName, myPeerName); err == nil { + if err := cml.UpdateAnnotation(KubePeersPrefix+peer.PeerName, myPeerName); err == nil { okToRemove = true } } @@ -155,7 +155,7 @@ func reclaimRemovedPeers(weave *weaveapi.Client, cml *configMapAnnotations, node } common.Log.Infoln("[kube-peers] Removing annotation ", peer.PeerName) // 7b. Remove annotation with key X - return cml.RemoveAnnotation(peer.PeerName) + return cml.RemoveAnnotation(KubePeersPrefix + peer.PeerName) }) common.Log.Debugln("[kube-peers] Finished removal of ", peer.PeerName) } From d2ed91ccc8a9a4259b609711962681a53b2509f0 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Fri, 1 Dec 2017 02:39:13 +0000 Subject: [PATCH 3/3] Clean up log messages a bit --- prog/kube-peers/main.go | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/prog/kube-peers/main.go b/prog/kube-peers/main.go index 8872a4a6..53708d03 100644 --- a/prog/kube-peers/main.go +++ b/prog/kube-peers/main.go @@ -133,12 +133,14 @@ func reclaimRemovedPeers(weave *weaveapi.Client, cml *configMapAnnotations, node common.Log.Debugln("[kube-peers] Noting I plan to remove ", peer.PeerName) if err := cml.UpdateAnnotation(KubePeersPrefix+peer.PeerName, myPeerName); err == nil { okToRemove = true + } else { + common.Log.Debugln("[kube-peers] error from UpdateAnnotation: ", err) } } if okToRemove { // 6. If step 4 or 5 succeeded, rmpeer X result, err := weave.RmPeer(peer.PeerName) - common.Log.Infoln("[kube-peers] rmpeer of", peer.PeerName, ":", result) + common.Log.Infof("[kube-peers] rmpeer of %s: %s", peer.PeerName, result) if err != nil { return err } @@ -149,15 +151,12 @@ func reclaimRemovedPeers(weave *weaveapi.Client, cml *configMapAnnotations, node } // 7a. Remove X from peerList storedPeerList.remove(peer.PeerName) - common.Log.Infoln("[kube-peers] Removing peer ", peer.PeerName, ". Expecting to remove linked annotation next.") if err := cml.UpdatePeerList(*storedPeerList); err != nil { return err } - common.Log.Infoln("[kube-peers] Removing annotation ", peer.PeerName) // 7b. Remove annotation with key X return cml.RemoveAnnotation(KubePeersPrefix + peer.PeerName) }) - common.Log.Debugln("[kube-peers] Finished removal of ", peer.PeerName) } // 8. If step 5 failed due to optimistic lock conflict, stop: someone else is handling X