Resolving race condition with Global configs in KDD

- Added a check to see if the error we see if of the type
`errors.ErrorResourceUpdateConflict` when we set global config
variables, as this means another Node set it while the current Node was
trying to.  As all the Nodes are trying to set them as the same value we can safely ignore
this and carry on.

- Added test that spins up API server and creates 10 Nodes in goroutines
  to simulate a bunch of Nodes coming up at once.

Author: heschlie

calico_node/Makefile

@@ -236,7 +236,9 @@ dist/allocate-ipip-addr: $(ALLOCATE_IPIP_FILES) vendor
 # Default value: directory with Makefile
 SOURCE_DIR?=$(dir $(lastword $(MAKEFILE_LIST)))
 SOURCE_DIR:=$(abspath $(SOURCE_DIR))
+CRD_PATH=$(CURDIR)/vendor/github.com/projectcalico/libcalico-go/test/
 LOCAL_IP_ENV?=$(shell ip route get 8.8.8.8 | head -1 | awk '{print $$7}')
+K8S_VERSION=v1.7.4
 ST_TO_RUN?=tests/st/
 
 # Can exclude the slower tests with "-a '!slow'"
@@ -396,6 +398,39 @@ run-etcd-host:
 	--advertise-client-urls "http://$(LOCAL_IP_ENV):2379,http://127.0.0.1:2379" \
 	--listen-client-urls "http://0.0.0.0:2379"
 
+## Kubernetes apiserver used for tests
+run-k8s-apiserver: stop-k8s-apiserver run-etcd vendor
+	docker run \
+		--net=host --name st-apiserver \
+		--detach \
+		gcr.io/google_containers/hyperkube-amd64:${K8S_VERSION} \
+		/hyperkube apiserver \
+			--bind-address=0.0.0.0 \
+			--insecure-bind-address=0.0.0.0 \
+				--etcd-servers=http://127.0.0.1:2379 \
+			--admission-control=NamespaceLifecycle,LimitRanger,DefaultStorageClass,ResourceQuota \
+			--authorization-mode=RBAC \
+			--service-cluster-ip-range=10.101.0.0/16 \
+			--v=10 \
+			--logtostderr=true
+
+	# Wait until we can configure a cluster role binding which allows anonymous auth.
+	while ! docker exec st-apiserver kubectl create clusterrolebinding anonymous-admin --clusterrole=cluster-admin --user=system:anonymous; do echo "Trying to create ClusterRoleBinding"; sleep 2; done
+
+	# Create CustomResourceDefinition (CRD) for Calico resources
+	# from the manifest crds.yaml
+	docker run \
+		--net=host \
+		--rm \
+		-v  $(CRD_PATH):/manifests \
+		lachlanevenson/k8s-kubectl:${K8S_VERSION} \
+		--server=http://localhost:8080 \
+		apply -f /manifests/crds.yaml
+
+## Stop Kubernetes apiserver
+stop-k8s-apiserver:
+	@-docker rm -f st-apiserver
+
 ###############################################################################
 # calico_node FVs
 ###############################################################################
@@ -421,7 +456,7 @@ node-fv:
 
 PHONY: node-test-containerized
 ## Run the tests in a container. Useful for CI, Mac dev.
-node-test-containerized: vendor run-etcd-host
+node-test-containerized: vendor run-etcd-host run-k8s-apiserver
 	docker run --rm \
 	-v $(CURDIR):/go/src/$(PACKAGE_NAME):rw \
 	-v $(VERSIONS_FILE):/versions.yaml:ro \

calico_node/startup/startup.go

@@ -788,12 +788,12 @@ func ensureDefaultConfig(cfg *api.CalicoAPIConfig, c *client.Client, node *api.N
 	}
 
 	// Store the Calico Version as a global felix config setting.
-	if err := c.Config().SetFelixConfig("CalicoVersion", "", VERSION); err != nil {
+	if err := checkConflictError(c.Config().SetFelixConfig("CalicoVersion", "", VERSION)); err != nil {
 		return err
 	}
 
 	// Update the ClusterType with the type in CLUSTER_TYPE
-	if err := updateClusterType(c, "ClusterType", os.Getenv("CLUSTER_TYPE")); err != nil {
+	if err := checkConflictError(updateClusterType(c, "ClusterType", os.Getenv("CLUSTER_TYPE"))); err != nil {
 		return err
 	}
 
@@ -818,7 +818,7 @@ func ensureGlobalFelixConfig(c *client.Client, key, def string) error {
 	if val, assigned, err := c.Config().GetFelixConfig(key, ""); err != nil {
 		return err
 	} else if !assigned {
-		return c.Config().SetFelixConfig(key, "", def)
+		return checkConflictError(c.Config().SetFelixConfig(key, "", def))
 	} else {
 		log.WithField(key, val).Debug("Global Felix value already assigned")
 		return nil
@@ -889,13 +889,24 @@ func ensureGlobalBGPConfig(c *client.Client, key, def string) error {
 	if val, assigned, err := c.Config().GetBGPConfig(key, ""); err != nil {
 		return err
 	} else if !assigned {
-		return c.Config().SetBGPConfig(key, "", def)
+		return checkConflictError(c.Config().SetBGPConfig(key, "", def))
 	} else {
 		log.WithField(key, val).Debug("Global BGP value already assigned")
 		return nil
 	}
 }
 
+// checkConflictError checks to see if the given error is of the type ErrorResourceUpdateConflict
+// and ignore it if so. This is to allow our global configs to ignore conflict from multiple Nodes
+// trying to set the same value at the same time.
+func checkConflictError(err error) error {
+	if conflict, ok := err.(errors.ErrorResourceUpdateConflict); ok {
+		log.Infof("Ignoring conflict when setting value %s", conflict.Identifier)
+		return nil
+	}
+	return err
+}
+
 // message prints a message to screen and to log.  A newline terminator is
 // not required in the format string.
 func message(format string, args ...interface{}) {

calico_node/startup/startup_test.go

@@ -15,9 +15,11 @@
 package main
 
 import (
+	"fmt"
 	"log"
 	"os"
 	"strings"
+	"sync"
 	"time"
 
 	. "github.com/onsi/ginkgo"
@@ -29,6 +31,12 @@ import (
 	"github.com/projectcalico/libcalico-go/lib/ipip"
 	"github.com/projectcalico/libcalico-go/lib/net"
 	"github.com/projectcalico/libcalico-go/lib/testutils"
+	"k8s.io/client-go/tools/clientcmd"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/pkg/api/v1"
+	clientcmdapi "k8s.io/client-go/tools/clientcmd/api"
 )
 
 var exitCode int
@@ -454,3 +462,82 @@ var _ = Describe("UT for Node IP assignment and conflict checking.", func() {
 		Entry("Test with \"IP6\" env var set to IP and BGP spec populated with different IP", makeNode("192.168.1.10/24", "2001:db8:85a3:8d3:1319:8a2e:370:7348/32"), []EnvItem{{"IP", "192.168.1.10/24"}, {"IP6", "2001:db8:85a3:8d3:1319:8a2e:370:7349/32"}}, true),
 	)
 })
+
+var _ = Describe("FV tests against K8s API server.", func() {
+	It("should not throw an error when multiple Nodes configure the same global CRD value.", func() {
+		// How many Nodes we want to "create".
+		numNodes := 10
+
+		// Create a K8s client.
+		configOverrides := &clientcmd.ConfigOverrides{
+			ClusterInfo: clientcmdapi.Cluster{
+				Server:                "http://127.0.0.1:8080",
+				InsecureSkipTLSVerify: true,
+			},
+		}
+
+		kcfg, err := clientcmd.NewNonInteractiveDeferredLoadingClientConfig(&clientcmd.ClientConfigLoadingRules{}, configOverrides).ClientConfig()
+		if err != nil {
+			Fail(fmt.Sprintf("Failed to create K8s config: %v", err))
+		}
+
+		cs, err := kubernetes.NewForConfig(kcfg)
+		if err != nil {
+			Fail(fmt.Sprintf("Could not create K8s client: %v", err))
+		}
+
+		// Create Calico client with k8s backend.
+		cfg := api.NewCalicoAPIConfig()
+		cfg.Spec = api.CalicoAPIConfigSpec{
+			DatastoreType: api.Kubernetes,
+			KubeConfig: api.KubeConfig{
+				K8sAPIEndpoint:           "http://127.0.0.1:8080",
+				K8sInsecureSkipTLSVerify: true,
+			},
+		}
+
+		c := testutils.CreateClient(*cfg)
+
+		// Create some Nodes using K8s client, Calico client does not support Node creation for KDD.
+		kNodes := []*v1.Node{}
+		for i := 0; i < numNodes; i++ {
+			n := &v1.Node{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: fmt.Sprintf("raceNode%02d", i+1),
+				},
+			}
+			kNodes = append(kNodes, n)
+			cs.Nodes().Create(n)
+		}
+
+		// Pull above Nodes using Calico client.
+		nodes, err := c.Nodes().List(api.NodeMetadata{})
+		if err != nil {
+			Fail(fmt.Sprintf("Could not retrieve Nodes %v", err))
+		}
+
+		// Run ensureDefaultConfig against each of the Nodes using goroutines to simulate multiple Nodes coming online.
+		var wg sync.WaitGroup
+		errors := []error{}
+		for _, node := range nodes.Items {
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				err = ensureDefaultConfig(cfg, c, &node)
+				if err != nil {
+					errors = append(errors, err)
+				}
+			}()
+		}
+
+		wg.Wait()
+
+		// Verify all runs complete without error.
+		Expect(len(errors)).To(Equal(0))
+
+		// Clean up our Nodes.
+		for _, node := range nodes.Items {
+			cs.Nodes().Delete(node.Metadata.Name, &metav1.DeleteOptions{})
+		}
+	})
+})