Merge pull request #1070 from heschlie/config-race

Resolving race condition with Global configs in KDD

Author: caseydavenport

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{})
+		}
+	})
+})