From 0591db121386ab1b740388fff29d6c0d01a665f2 Mon Sep 17 00:00:00 2001 From: Neil Jerram Date: Thu, 6 Sep 2018 17:26:03 +0100 Subject: [PATCH 1/4] Restore the BGP syncer It needs to be here after all, because used by both confd and typha. --- lib/backend/syncersv1/bgpsyncer/bgpsyncer.go | 51 +++ .../syncersv1/bgpsyncer/bgpsyncer_e2e_test.go | 294 ++++++++++++++++++ .../bgpsyncer/bgpsyncer_suite_test.go | 30 ++ lib/backend/syncersv1/bgpsyncer/doc.go | 23 ++ lib/testutils/syncertester.go | 9 + 5 files changed, 407 insertions(+) create mode 100644 lib/backend/syncersv1/bgpsyncer/bgpsyncer.go create mode 100644 lib/backend/syncersv1/bgpsyncer/bgpsyncer_e2e_test.go create mode 100644 lib/backend/syncersv1/bgpsyncer/bgpsyncer_suite_test.go create mode 100644 lib/backend/syncersv1/bgpsyncer/doc.go diff --git a/lib/backend/syncersv1/bgpsyncer/bgpsyncer.go b/lib/backend/syncersv1/bgpsyncer/bgpsyncer.go new file mode 100644 index 000000000..ea911217a --- /dev/null +++ b/lib/backend/syncersv1/bgpsyncer/bgpsyncer.go @@ -0,0 +1,51 @@ +// Copyright (c) 2017-2018 Tigera, Inc. All rights reserved. + +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package bgpsyncer + +import ( + apiv3 "github.com/projectcalico/libcalico-go/lib/apis/v3" + "github.com/projectcalico/libcalico-go/lib/backend/api" + "github.com/projectcalico/libcalico-go/lib/backend/model" + "github.com/projectcalico/libcalico-go/lib/backend/syncersv1/updateprocessors" + "github.com/projectcalico/libcalico-go/lib/backend/watchersyncer" +) + +// New creates a new BGP v1 Syncer. Since only etcdv3 supports Watchers for all of +// the required resource types, the WatcherSyncer will go into a polling loop for +// KDD. An optional node name may be supplied. If set, the syncer only watches +// the specified node rather than all nodes. +func New(client api.Client, callbacks api.SyncerCallbacks, node string) api.Syncer { + // Create ResourceTypes required for BGP. + resourceTypes := []watchersyncer.ResourceType{ + { + ListInterface: model.ResourceListOptions{Kind: apiv3.KindIPPool}, + UpdateProcessor: updateprocessors.NewIPPoolUpdateProcessor(), + }, + { + ListInterface: model.ResourceListOptions{Kind: apiv3.KindBGPConfiguration}, + UpdateProcessor: updateprocessors.NewBGPConfigUpdateProcessor(), + }, + { + ListInterface: model.ResourceListOptions{Kind: apiv3.KindNode}, + }, + { + ListInterface: model.ResourceListOptions{Kind: apiv3.KindBGPPeer}, + }, + { + ListInterface: model.BlockAffinityListOptions{Host: node}, + }, + } + return watchersyncer.New(client, resourceTypes, callbacks) +} diff --git a/lib/backend/syncersv1/bgpsyncer/bgpsyncer_e2e_test.go b/lib/backend/syncersv1/bgpsyncer/bgpsyncer_e2e_test.go new file mode 100644 index 000000000..280a38a0c --- /dev/null +++ b/lib/backend/syncersv1/bgpsyncer/bgpsyncer_e2e_test.go @@ -0,0 +1,294 @@ +// Copyright (c) 2017 Tigera, Inc. All rights reserved. + +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package bgpsyncer_test + +import ( + "context" + + . "github.com/onsi/ginkgo" + . "github.com/onsi/gomega" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + "github.com/projectcalico/libcalico-go/lib/apiconfig" + apiv3 "github.com/projectcalico/libcalico-go/lib/apis/v3" + "github.com/projectcalico/libcalico-go/lib/backend" + "github.com/projectcalico/libcalico-go/lib/backend/api" + "github.com/projectcalico/libcalico-go/lib/backend/model" + "github.com/projectcalico/libcalico-go/lib/backend/syncersv1/bgpsyncer" + "github.com/projectcalico/libcalico-go/lib/clientv3" + "github.com/projectcalico/libcalico-go/lib/ipam" + "github.com/projectcalico/libcalico-go/lib/ipip" + "github.com/projectcalico/libcalico-go/lib/net" + "github.com/projectcalico/libcalico-go/lib/numorstring" + "github.com/projectcalico/libcalico-go/lib/options" + "github.com/projectcalico/libcalico-go/lib/testutils" +) + +// These tests validate that the various resources that the BGP watches are +// handled correctly by the syncer. We don't validate in detail the behavior of +// each of udpate handlers that are invoked, since these are tested more thoroughly +// elsewhere. +var _ = testutils.E2eDatastoreDescribe("BGP syncer tests", testutils.DatastoreAll, func(config apiconfig.CalicoAPIConfig) { + + ctx := context.Background() + + Describe("BGP syncer functionality", func() { + It("should receive the synced after return all current data", func() { + // Create a v3 client to drive data changes (luckily because this is the _test module, + // we don't get circular imports. + c, err := clientv3.New(config) + Expect(err).NotTo(HaveOccurred()) + + // Create the backend client to obtain a syncer interface. + be, err := backend.NewClient(config) + Expect(err).NotTo(HaveOccurred()) + be.Clean() + + // Create a SyncerTester to receive the BGP syncer callback events and to allow us + // to assert state. + syncTester := testutils.NewSyncerTester() + syncer := bgpsyncer.New(be, syncTester, "127.0.0.1") + syncer.Start() + expectedCacheSize := 0 + + By("Checking status is updated to sync'd at start of day") + syncTester.ExpectStatusUpdate(api.WaitForDatastore) + syncTester.ExpectCacheSize(expectedCacheSize) + syncTester.ExpectStatusUpdate(api.ResyncInProgress) + if config.Spec.DatastoreType == apiconfig.Kubernetes { + expectedCacheSize += 2 + } + syncTester.ExpectCacheSize(expectedCacheSize) + syncTester.ExpectStatusUpdate(api.InSync) + syncTester.ExpectCacheSize(expectedCacheSize) + + // For Kubernetes test the two entries already in the cache - one + // affinity block and one node. + if config.Spec.DatastoreType == apiconfig.Kubernetes { + syncTester.ExpectPath("/calico/resources/v3/projectcalico.org/nodes/127.0.0.1") + syncTester.ExpectData(model.KVPair{ + Key: model.BlockAffinityKey{Host: "127.0.0.1", CIDR: net.MustParseCIDR("10.10.10.0/24")}, + Value: &model.BlockAffinity{State: model.StateConfirmed}, + }) + } + + By("Disabling node to node mesh and adding a default ASNumber") + n2n := false + asn := numorstring.ASNumber(12345) + bgpCfg, err := c.BGPConfigurations().Create( + ctx, + &apiv3.BGPConfiguration{ + ObjectMeta: metav1.ObjectMeta{Name: "default"}, + Spec: apiv3.BGPConfigurationSpec{ + NodeToNodeMeshEnabled: &n2n, + ASNumber: &asn, + }, + }, + options.SetOptions{}, + ) + Expect(err).NotTo(HaveOccurred()) + expectedCacheSize += 2 + + // We should have entries for each config option (i.e. 2) + syncTester.ExpectCacheSize(expectedCacheSize) + syncTester.ExpectData(model.KVPair{ + Key: model.GlobalBGPConfigKey{"as_num"}, + Value: "12345", + Revision: bgpCfg.ResourceVersion, + }) + syncTester.ExpectData(model.KVPair{ + Key: model.GlobalBGPConfigKey{"node_mesh"}, + Value: "{\"enabled\":false}", + Revision: bgpCfg.ResourceVersion, + }) + + var node *apiv3.Node + if config.Spec.DatastoreType == apiconfig.Kubernetes { + // For Kubernetes, update the existing node config to have some BGP configuration. + By("Configuring a node with BGP configuration") + node, err = c.Nodes().Get(ctx, "127.0.0.1", options.GetOptions{}) + Expect(err).NotTo(HaveOccurred()) + node.Spec.BGP = &apiv3.NodeBGPSpec{ + IPv4Address: "1.2.3.4/24", + IPv6Address: "aa:bb::cc/120", + } + node, err = c.Nodes().Update(ctx, node, options.SetOptions{}) + Expect(err).NotTo(HaveOccurred()) + + // The existing Node resource is updated; no change in cache size. + } else { + // For non-Kubernetes, add a new node with valid BGP configuration. + By("Creating a node with BGP configuration") + node, err = c.Nodes().Create( + ctx, + &apiv3.Node{ + ObjectMeta: metav1.ObjectMeta{Name: "127.0.0.1"}, + Spec: apiv3.NodeSpec{ + BGP: &apiv3.NodeBGPSpec{ + IPv4Address: "1.2.3.4/24", + IPv6Address: "aa:bb::cc/120", + }, + }, + }, + options.SetOptions{}, + ) + Expect(err).NotTo(HaveOccurred()) + expectedCacheSize += 1 + } + + syncTester.ExpectCacheSize(expectedCacheSize) + syncTester.ExpectPath("/calico/resources/v3/projectcalico.org/nodes/127.0.0.1") + + By("Updating the BGPConfiguration to remove the default ASNumber") + bgpCfg.Spec.ASNumber = nil + _, err = c.BGPConfigurations().Update(ctx, bgpCfg, options.SetOptions{}) + Expect(err).NotTo(HaveOccurred()) + // Removing one config option ( -1 ) + expectedCacheSize -= 1 + syncTester.ExpectCacheSize(expectedCacheSize) + syncTester.ExpectNoData(model.GlobalBGPConfigKey{"as_num"}) + + By("Creating an IPPool") + poolCIDR := "192.124.0.0/21" + poolCIDRNet := net.MustParseCIDR(poolCIDR) + pool, err := c.IPPools().Create( + ctx, + &apiv3.IPPool{ + ObjectMeta: metav1.ObjectMeta{Name: "mypool"}, + Spec: apiv3.IPPoolSpec{ + CIDR: poolCIDR, + IPIPMode: apiv3.IPIPModeCrossSubnet, + NATOutgoing: true, + }, + }, + options.SetOptions{}, + ) + Expect(err).NotTo(HaveOccurred()) + // The pool will add as single entry ( +1 ) + poolKeyV1 := model.IPPoolKey{CIDR: net.MustParseCIDR("192.124.0.0/21")} + expectedCacheSize += 1 + syncTester.ExpectCacheSize(expectedCacheSize) + syncTester.ExpectData(model.KVPair{ + Key: poolKeyV1, + Value: &model.IPPool{ + CIDR: poolCIDRNet, + IPIPInterface: "tunl0", + IPIPMode: ipip.CrossSubnet, + Masquerade: true, + IPAM: true, + Disabled: false, + }, + Revision: pool.ResourceVersion, + }) + + By("Creating a BGPPeer") + _, err = c.BGPPeers().Create( + ctx, + &apiv3.BGPPeer{ + ObjectMeta: metav1.ObjectMeta{Name: "peer1"}, + Spec: apiv3.BGPPeerSpec{ + PeerIP: "192.124.10.20", + ASNumber: numorstring.ASNumber(75758), + }, + }, + options.SetOptions{}, + ) + Expect(err).NotTo(HaveOccurred()) + + // The peer will add as single entry ( +1 ) + expectedCacheSize += 1 + syncTester.ExpectCacheSize(expectedCacheSize) + syncTester.ExpectPath("/calico/resources/v3/projectcalico.org/bgppeers/peer1") + + // For non-kubernetes, check that we can allocate an IP address and get a syncer update + // for the allocation block. + var blockAffinityKeyV1 model.BlockAffinityKey + if config.Spec.DatastoreType != apiconfig.Kubernetes { + By("Allocating an IP address and checking that we get an allocation block") + ips1, _, err := c.IPAM().AutoAssign(ctx, ipam.AutoAssignArgs{ + Num4: 1, + Hostname: "127.0.0.1", + }) + Expect(err).NotTo(HaveOccurred()) + + // Allocating an IP will create an affinity block that we should be notified of. Not sure + // what CIDR will be chosen, so search the cached entries. + expectedCacheSize += 1 + syncTester.ExpectCacheSize(expectedCacheSize) + current := syncTester.GetCacheEntries() + for _, kvp := range current { + if kab, ok := kvp.Key.(model.BlockAffinityKey); ok { + if kab.Host == "127.0.0.1" && poolCIDRNet.Contains(kab.CIDR.IP) { + blockAffinityKeyV1 = kab + break + } + } + } + Expect(blockAffinityKeyV1).NotTo(BeNil(), "Did not find affinity block in sync data") + + By("Allocating an IP address on a different host and checking for no updates") + // The syncer only monitors affine blocks for one host, so IP allocations for a different + // host should not result in updates. + ips2, _, err := c.IPAM().AutoAssign(ctx, ipam.AutoAssignArgs{ + Num4: 1, + Hostname: "not-this-host", + }) + Expect(err).NotTo(HaveOccurred()) + syncTester.ExpectCacheSize(expectedCacheSize) + + By("Releasing the IP addresses and checking for no updates") + // Releasing IPs should leave the affine blocks assigned, so releasing the IPs + // should result in no updates. + _, err = c.IPAM().ReleaseIPs(ctx, ips1) + Expect(err).NotTo(HaveOccurred()) + _, err = c.IPAM().ReleaseIPs(ctx, ips2) + Expect(err).NotTo(HaveOccurred()) + syncTester.ExpectCacheSize(expectedCacheSize) + + By("Deleting the IPPool and checking for pool and affine block deletion") + // Deleting the pool will also release all affine blocks associated with the pool. + _, err = c.IPPools().Delete(ctx, "mypool", options.DeleteOptions{}) + Expect(err).NotTo(HaveOccurred()) + + // The pool and the affine block for 127.0.0.1 should have deletion events. + expectedCacheSize -= 2 + syncTester.ExpectCacheSize(expectedCacheSize) + syncTester.ExpectNoData(blockAffinityKeyV1) + syncTester.ExpectNoData(poolKeyV1) + } + + By("Starting a new syncer and verifying that all current entries are returned before sync status") + // We need to create a new syncTester and syncer. + current := syncTester.GetCacheEntries() + syncTester = testutils.NewSyncerTester() + syncer = bgpsyncer.New(be, syncTester, "127.0.0.1") + syncer.Start() + + // Verify the data is the same as the data from the previous cache. We got the cache in the previous + // step. + syncTester.ExpectStatusUpdate(api.WaitForDatastore) + syncTester.ExpectStatusUpdate(api.ResyncInProgress) + syncTester.ExpectCacheSize(expectedCacheSize) + for _, e := range current { + if config.Spec.DatastoreType == apiconfig.Kubernetes { + // Don't check revisions for K8s since the node data gets updated constantly. + e.Revision = "" + } + syncTester.ExpectData(e) + } + syncTester.ExpectStatusUpdate(api.InSync) + }) + }) +}) diff --git a/lib/backend/syncersv1/bgpsyncer/bgpsyncer_suite_test.go b/lib/backend/syncersv1/bgpsyncer/bgpsyncer_suite_test.go new file mode 100644 index 000000000..7996d57bd --- /dev/null +++ b/lib/backend/syncersv1/bgpsyncer/bgpsyncer_suite_test.go @@ -0,0 +1,30 @@ +// Copyright (c) 2017 Tigera, Inc. All rights reserved. + +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package bgpsyncer + +import ( + "testing" + + . "github.com/onsi/ginkgo" + . "github.com/onsi/gomega" + + "github.com/projectcalico/libcalico-go/lib/testutils" +) + +func TestClient(t *testing.T) { + testutils.HookLogrusForGinkgo() + RegisterFailHandler(Fail) + RunSpecs(t, "BGP syncer test suite") +} diff --git a/lib/backend/syncersv1/bgpsyncer/doc.go b/lib/backend/syncersv1/bgpsyncer/doc.go new file mode 100644 index 000000000..450c0ae1d --- /dev/null +++ b/lib/backend/syncersv1/bgpsyncer/doc.go @@ -0,0 +1,23 @@ +// Copyright (c) 2017 Tigera, Inc. All rights reserved. + +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package bgpsyncer + +/* +bgpsyncer implements an api.Syncer for consumers of BGP configuration. + +The primary use case here is for a generic Calico backend for confd. + +This implementation uses the watchersyncer. +*/ diff --git a/lib/testutils/syncertester.go b/lib/testutils/syncertester.go index dd29d74a6..03f6ff2f5 100644 --- a/lib/testutils/syncertester.go +++ b/lib/testutils/syncertester.go @@ -222,6 +222,15 @@ func (st *SyncerTester) ExpectData(kvp model.KVPair) { } } +// ExpectPath verifies that a KVPair with a specified path is in the cache. +func (st *SyncerTester) ExpectPath(path string) { + kv := func() interface{} { + return st.GetCacheKVPair(path) + } + Eventually(kv, 6*time.Second, time.Millisecond).ShouldNot(BeNil()) + Consistently(kv).ShouldNot(BeNil()) +} + // ExpectDataMatch verifies that the KV in the cache exists and matches the GomegaMatcher. func (st *SyncerTester) ExpectValueMatches(k model.Key, match gomegatypes.GomegaMatcher) { key, err := model.KeyToDefaultPath(k) From c28c84c4ba14ebac0b9b576c2febbcecd4896d01 Mon Sep 17 00:00:00 2001 From: Shaun Crampton Date: Wed, 8 Aug 2018 16:08:49 +0100 Subject: [PATCH 2/4] Add parsing logic for BGP syncer keys --- lib/backend/model/bgppeer.go | 22 ++++++++++++---------- lib/backend/model/keys.go | 10 ++++++++++ 2 files changed, 22 insertions(+), 10 deletions(-) diff --git a/lib/backend/model/bgppeer.go b/lib/backend/model/bgppeer.go index e0a012edc..1d098c717 100644 --- a/lib/backend/model/bgppeer.go +++ b/lib/backend/model/bgppeer.go @@ -1,4 +1,4 @@ -// Copyright (c) 2016 Tigera, Inc. All rights reserved. +// Copyright (c) 2016-2018 Tigera, Inc. All rights reserved. // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -86,14 +86,15 @@ func (options NodeBGPPeerListOptions) KeyFromDefaultPath(path string) Key { log.Debugf("Get BGPPeer key from %s", path) nodename := "" peerIP := net.IP{} - ekeyb := []byte(path) - if r := matchHostBGPPeer.FindAllSubmatch(ekeyb, -1); len(r) == 1 { - nodename = string(r[0][1]) - if err := peerIP.UnmarshalText(r[0][2]); err != nil { - log.WithError(err).WithField("PeerIP", r[0][2]).Error("Error unmarshalling GlobalBGPPeer IP address") + if r := matchHostBGPPeer.FindAllStringSubmatch(path, -1); len(r) == 1 { + nodename = r[0][1] + peerIPOrNil := net.ParseIP(r[0][2]) + if peerIPOrNil == nil { + log.WithField("PeerIP", r[0][1]).Error("Error unmarshalling NodeBGPPeer IP address") return nil } + peerIP = *peerIPOrNil } else { log.Debugf("%s didn't match regex", path) return nil @@ -155,13 +156,14 @@ func (options GlobalBGPPeerListOptions) defaultPathRoot() string { func (options GlobalBGPPeerListOptions) KeyFromDefaultPath(path string) Key { log.Debugf("Get BGPPeer key from %s", path) peerIP := net.IP{} - ekeyb := []byte(path) - if r := matchGlobalBGPPeer.FindAllSubmatch(ekeyb, -1); len(r) == 1 { - if err := peerIP.UnmarshalText(r[0][1]); err != nil { - log.WithError(err).WithField("PeerIP", r[0][1]).Error("Error unmarshalling GlobalBGPPeer IP address") + if r := matchGlobalBGPPeer.FindAllStringSubmatch(path, -1); len(r) == 1 { + peerIPOrNil := net.ParseIP(r[0][1]) + if peerIPOrNil == nil { + log.WithField("PeerIP", r[0][1]).Error("Error unmarshalling GlobalBGPPeer IP address") return nil } + peerIP = *peerIPOrNil } else { log.Debugf("%s didn't match regex", path) return nil diff --git a/lib/backend/model/keys.go b/lib/backend/model/keys.go index 61fac368f..328c24f85 100644 --- a/lib/backend/model/keys.go +++ b/lib/backend/model/keys.go @@ -250,6 +250,16 @@ func KeyFromDefaultPath(path string) Key { } else if matchReadyFlag.MatchString(path) { log.Debugf("Path is a ready flag: %v", path) return ReadyFlagKey{} + } else if k := (NodeBGPPeerListOptions{}).KeyFromDefaultPath(path); k != nil { + return k + } else if k := (GlobalBGPPeerListOptions{}).KeyFromDefaultPath(path); k != nil { + return k + } else if k := (NodeBGPConfigListOptions{}).KeyFromDefaultPath(path); k != nil { + return k + } else if k := (GlobalBGPConfigListOptions{}).KeyFromDefaultPath(path); k != nil { + return k + } else if k := (BlockAffinityListOptions{}).KeyFromDefaultPath(path); k != nil { + return k } else { log.Debugf("Path is unknown: %v", path) } From 6452ece34e3989694e0e00a1965de638f82856bf Mon Sep 17 00:00:00 2001 From: Neil Jerram Date: Thu, 6 Sep 2018 19:15:08 +0100 Subject: [PATCH 3/4] Revert v1 BGP peer key changes, add v3 Node and Peer key parsing --- lib/backend/model/bgppeer.go | 22 ++++++++++------------ lib/backend/model/keys.go | 9 +++++---- 2 files changed, 15 insertions(+), 16 deletions(-) diff --git a/lib/backend/model/bgppeer.go b/lib/backend/model/bgppeer.go index 1d098c717..e0a012edc 100644 --- a/lib/backend/model/bgppeer.go +++ b/lib/backend/model/bgppeer.go @@ -1,4 +1,4 @@ -// Copyright (c) 2016-2018 Tigera, Inc. All rights reserved. +// Copyright (c) 2016 Tigera, Inc. All rights reserved. // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -86,15 +86,14 @@ func (options NodeBGPPeerListOptions) KeyFromDefaultPath(path string) Key { log.Debugf("Get BGPPeer key from %s", path) nodename := "" peerIP := net.IP{} + ekeyb := []byte(path) - if r := matchHostBGPPeer.FindAllStringSubmatch(path, -1); len(r) == 1 { - nodename = r[0][1] - peerIPOrNil := net.ParseIP(r[0][2]) - if peerIPOrNil == nil { - log.WithField("PeerIP", r[0][1]).Error("Error unmarshalling NodeBGPPeer IP address") + if r := matchHostBGPPeer.FindAllSubmatch(ekeyb, -1); len(r) == 1 { + nodename = string(r[0][1]) + if err := peerIP.UnmarshalText(r[0][2]); err != nil { + log.WithError(err).WithField("PeerIP", r[0][2]).Error("Error unmarshalling GlobalBGPPeer IP address") return nil } - peerIP = *peerIPOrNil } else { log.Debugf("%s didn't match regex", path) return nil @@ -156,14 +155,13 @@ func (options GlobalBGPPeerListOptions) defaultPathRoot() string { func (options GlobalBGPPeerListOptions) KeyFromDefaultPath(path string) Key { log.Debugf("Get BGPPeer key from %s", path) peerIP := net.IP{} + ekeyb := []byte(path) - if r := matchGlobalBGPPeer.FindAllStringSubmatch(path, -1); len(r) == 1 { - peerIPOrNil := net.ParseIP(r[0][1]) - if peerIPOrNil == nil { - log.WithField("PeerIP", r[0][1]).Error("Error unmarshalling GlobalBGPPeer IP address") + if r := matchGlobalBGPPeer.FindAllSubmatch(ekeyb, -1); len(r) == 1 { + if err := peerIP.UnmarshalText(r[0][1]); err != nil { + log.WithError(err).WithField("PeerIP", r[0][1]).Error("Error unmarshalling GlobalBGPPeer IP address") return nil } - peerIP = *peerIPOrNil } else { log.Debugf("%s didn't match regex", path) return nil diff --git a/lib/backend/model/keys.go b/lib/backend/model/keys.go index 328c24f85..27156f7e6 100644 --- a/lib/backend/model/keys.go +++ b/lib/backend/model/keys.go @@ -23,6 +23,7 @@ import ( "strings" "time" + "github.com/projectcalico/libcalico-go/lib/apis/v3" "github.com/projectcalico/libcalico-go/lib/net" log "github.com/sirupsen/logrus" ) @@ -250,16 +251,16 @@ func KeyFromDefaultPath(path string) Key { } else if matchReadyFlag.MatchString(path) { log.Debugf("Path is a ready flag: %v", path) return ReadyFlagKey{} - } else if k := (NodeBGPPeerListOptions{}).KeyFromDefaultPath(path); k != nil { - return k - } else if k := (GlobalBGPPeerListOptions{}).KeyFromDefaultPath(path); k != nil { - return k } else if k := (NodeBGPConfigListOptions{}).KeyFromDefaultPath(path); k != nil { return k } else if k := (GlobalBGPConfigListOptions{}).KeyFromDefaultPath(path); k != nil { return k } else if k := (BlockAffinityListOptions{}).KeyFromDefaultPath(path); k != nil { return k + } else if k := (ResourceListOptions{Kind: v3.KindNode}).KeyFromDefaultPath(path); k != nil { + return k + } else if k := (ResourceListOptions{Kind: v3.KindBGPPeer}).KeyFromDefaultPath(path); k != nil { + return k } else { log.Debugf("Path is unknown: %v", path) } From 4dd0e61bf0bf282ce87b8534add8569b928efe8d Mon Sep 17 00:00:00 2001 From: Neil Jerram Date: Fri, 7 Sep 2018 09:26:38 +0100 Subject: [PATCH 4/4] Implement AffinityBlock list for all nodes, on KDD --- lib/backend/k8s/k8s_test.go | 11 +++---- lib/backend/k8s/resources/affinityblock.go | 36 ++++++++++++++++++++++ 2 files changed, 41 insertions(+), 6 deletions(-) diff --git a/lib/backend/k8s/k8s_test.go b/lib/backend/k8s/k8s_test.go index b2177f6ec..d0dc1c649 100644 --- a/lib/backend/k8s/k8s_test.go +++ b/lib/backend/k8s/k8s_test.go @@ -1422,12 +1422,6 @@ var _ = Describe("Test Syncer API for Kubernetes backend", func() { }) }) - It("should error on unsupported List() calls", func() { - objs, err := c.List(ctx, model.BlockAffinityListOptions{}, "") - Expect(err).To(HaveOccurred()) - Expect(objs).To(BeNil()) - }) - It("should not error on unsupported List() calls", func() { var nodename string By("Listing all Nodes to find a suitable Node name", func() { @@ -1441,6 +1435,11 @@ var _ = Describe("Test Syncer API for Kubernetes backend", func() { Expect(err).NotTo(HaveOccurred()) Expect(len(objs.KVPairs)).To(Equal(1)) }) + By("Listing all BlockAffinity for all Nodes", func() { + objs, err := c.List(ctx, model.BlockAffinityListOptions{}, "") + Expect(err).NotTo(HaveOccurred()) + Expect(len(objs.KVPairs)).To(Equal(1)) + }) }) It("should support setting and getting FelixConfig", func() { diff --git a/lib/backend/k8s/resources/affinityblock.go b/lib/backend/k8s/resources/affinityblock.go index 58031edfd..3ada52500 100644 --- a/lib/backend/k8s/resources/affinityblock.go +++ b/lib/backend/k8s/resources/affinityblock.go @@ -116,6 +116,42 @@ func (c *AffinityBlockClient) List(ctx context.Context, list model.ListInterface return kvpl, nil } + // When host is not specified... + if bl.IPVersion == 0 { + // Get the node settings, we use the nodes PodCIDR as the only node affinity block. + nodeList, err := c.clientSet.CoreV1().Nodes().List(metav1.ListOptions{ResourceVersion: revision}) + if err != nil { + err = K8sErrorToCalico(err, list) + if _, ok := err.(cerrors.ErrorResourceDoesNotExist); !ok { + return nil, err + } + return kvpl, nil + } + + kvpl.Revision = nodeList.ResourceVersion + for _, node := range nodeList.Items { + // Return no results if the pod CIDR is not assigned. + podcidr := node.Spec.PodCIDR + if len(podcidr) == 0 { + continue + } + + _, cidr, err := cnet.ParseCIDR(podcidr) + if err != nil { + return nil, err + } + kvpl.KVPairs = append(kvpl.KVPairs, &model.KVPair{ + Key: model.BlockAffinityKey{ + CIDR: *cidr, + Host: node.Name, + }, + Value: &model.BlockAffinity{State: model.StateConfirmed}, + Revision: node.ResourceVersion, + }) + } + return kvpl, nil + } + // Currently querying the affinity block is only used by the BGP syncer *and* we always // query for a specific Node, so for now fail List requests for all nodes. log.Warn("Operation List (all nodes or all IP versions) is not supported on AffinityBlock type")