Merge pull request #355 from robbrockbank/1.1.1-merge

Commits required for v1.1.1

Author: robbrockbank

lib/api/ippool.go

@@ -63,7 +63,7 @@ type IPIPConfiguration struct {
 	// addresses within this pool.  A mode of "cross-subnet" will only use IPIP
 	// tunneling when the destination node is on a different subnet to the
 	// originating node.  The default value (if not specified) is "always".
-	Mode ipip.Mode `json:"mode,omitempty"`
+	Mode ipip.Mode `json:"mode,omitempty" validate:"ipipmode"`
 }
 
 // NewIPPool creates a new (zeroed) Pool struct with the TypeMetadata initialised to the current

lib/api/unversioned/types.go

@@ -30,8 +30,8 @@ var (
 //
 // All resource and resource lists embed a TypeMetadata as an anonymous field.
 type TypeMetadata struct {
-	Kind       string `json:"kind" validate:"required"`
-	APIVersion string `json:"apiVersion" validate:"required"`
+	Kind       string `json:"kind"`
+	APIVersion string `json:"apiVersion"`
 }
 
 func (md TypeMetadata) GetTypeMetadata() TypeMetadata {

lib/backend/etcd/syncer.go

@@ -1,4 +1,4 @@
-// Copyright (c) 2016 Tigera, Inc. All rights reserved.
+// Copyright (c) 2016-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.
@@ -245,6 +245,7 @@ func sendSnapshotNode(node *client.Node, snapshotUpdates chan<- interface{}, res
 // so that the merge goroutine can trigger a new resync via snapshot.
 func (syn *etcdSyncer) watchEtcd(watcherUpdateC chan<- interface{}) {
 	log.Info("etcd watch thread started.")
+	var timeOfLastError time.Time
 	// Each trip around the outer loop establishes the current etcd index
 	// of the cluster, triggers a new snapshot read from that index (via
 	// message to the merge goroutine) and starts watching from that index.
@@ -278,6 +279,13 @@ func (syn *etcdSyncer) watchEtcd(watcherUpdateC chan<- interface{}) {
 			// Wait for the next event from the watcher.
 			resp, err := watcher.Next(context.Background())
 			if err != nil {
+				// Prevent a tight loop if etcd is repeatedly failing.
+				if time.Since(timeOfLastError) < 1*time.Second {
+					log.Warning("May be tight looping, throttling retries.")
+					time.Sleep(1 * time.Second)
+				}
+				timeOfLastError = time.Now()
+
 				if !retryableWatcherError(err) {
 					// Break out of the loop to trigger a new resync.
 					log.WithError(err).Warning("Lost sync with etcd, restarting watcher.")

lib/backend/k8s/conversion.go

@@ -219,6 +219,12 @@ func (c converter) podToWorkloadEndpoint(pod *kapiv1.Pod) (*model.KVPair, error)
 		ipNets = []cnet.IPNet{*ipNet}
 	}
 
+	// Check to see if the Pod has a Node.
+	if len(pod.Spec.NodeName) == 0 {
+		log.Warnf("%s - NodeName is empty. Spec: %+v Status: %+v", pod.Name, pod.Spec, pod.Status)
+		return nil, fmt.Errorf("Pod '%s' is missing a NodeName", pod.Name)
+	}
+
 	// Generate the interface name and MAC based on workload.  This must match
 	// the host-side veth configured by the CNI plugin.
 	interfaceName := VethNameForWorkload(workload)

lib/backend/k8s/conversion_test.go

@@ -191,6 +191,23 @@ var _ = Describe("Test Pod conversion", func() {
 		Expect(wep.Value.(*model.WorkloadEndpoint).State).To(Equal("active"))
 		Expect(wep.Value.(*model.WorkloadEndpoint).Labels).To(Equal(map[string]string{"calico/k8s_ns": "default"}))
 	})
+
+	It("should not Parse a Pod with no NodeName", func() {
+		pod := k8sapi.Pod{
+			ObjectMeta: k8sapi.ObjectMeta{
+				Name:      "podA",
+				Namespace: "default",
+			},
+			Spec: k8sapi.PodSpec{},
+			Status: k8sapi.PodStatus{
+				PodIP: "192.168.0.1",
+			},
+		}
+
+		_, err := c.podToWorkloadEndpoint(&pod)
+		Expect(err).To(HaveOccurred())
+	})
+
 })
 
 var _ = Describe("Test NetworkPolicy conversion", func() {

lib/backend/k8s/k8s.go

@@ -113,7 +113,7 @@ func NewKubeClient(kc *KubeConfig) (*KubeClient, error) {
 
 	tprClient, err := buildTPRClient(config)
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("Failed to build TPR client: %s", err)
 	}
 	kubeClient := &KubeClient{
 		clientSet: cs,
@@ -131,15 +131,15 @@ func (c *KubeClient) EnsureInitialized() error {
 	log.Info("Ensuring ThirdPartyResources exist")
 	err := c.ensureThirdPartyResources()
 	if err != nil {
-		return err
+		return fmt.Errorf("Failed to ensure ThirdPartyResources exist: %s", err)
 	}
 	log.Info("ThirdPartyResources exist")
 
 	// Ensure ClusterType is set.
 	log.Info("Ensuring ClusterType is set")
 	err = c.waitForClusterType()
 	if err != nil {
-		return err
+		return fmt.Errorf("Failed to ensure ClusterType is set: %s", err)
 	}
 	log.Info("ClusterType is set")
 	return nil
@@ -221,6 +221,7 @@ func (c *KubeClient) ensureClusterType() (bool, error) {
 		}
 		value = existingValue
 	}
+	log.WithField("value", value).Debug("Setting ClusterType")
 	_, err = c.Apply(&model.KVPair{
 		Key:   k,
 		Value: value,
@@ -276,6 +277,7 @@ func (c *KubeClient) Syncer(callbacks api.SyncerCallbacks) api.Syncer {
 
 // Create an entry in the datastore.  This errors if the entry already exists.
 func (c *KubeClient) Create(d *model.KVPair) (*model.KVPair, error) {
+	log.Debugf("Performing 'Create' for %+v", d)
 	switch d.Key.(type) {
 	case model.GlobalConfigKey:
 		return c.createGlobalConfig(d)
@@ -293,21 +295,23 @@ func (c *KubeClient) Create(d *model.KVPair) (*model.KVPair, error) {
 // Update an existing entry in the datastore.  This errors if the entry does
 // not exist.
 func (c *KubeClient) Update(d *model.KVPair) (*model.KVPair, error) {
+	log.Debugf("Performing 'Update' for %+v", d)
 	switch d.Key.(type) {
 	case model.GlobalConfigKey:
 		return c.updateGlobalConfig(d)
 	case model.IPPoolKey:
 		return c.ipPoolClient.Update(d)
 	default:
 		// If the resource isn't supported, then this is a no-op.
-		log.Infof("'Update' for %+v is no-op", d.Key)
+		log.Debugf("'Update' for %+v is no-op", d.Key)
 		return d, nil
 	}
 }
 
 // Set an existing entry in the datastore.  This ignores whether an entry already
 // exists.
 func (c *KubeClient) Apply(d *model.KVPair) (*model.KVPair, error) {
+	log.Debugf("Performing 'Apply' for %+v", d)
 	switch d.Key.(type) {
 	case model.WorkloadEndpointKey:
 		return c.applyWorkloadEndpoint(d)
@@ -316,20 +320,21 @@ func (c *KubeClient) Apply(d *model.KVPair) (*model.KVPair, error) {
 	case model.IPPoolKey:
 		return c.ipPoolClient.Apply(d)
 	default:
-		log.Infof("'Apply' for %s is no-op", d.Key)
+		log.Debugf("'Apply' for %s is no-op", d.Key)
 		return d, nil
 	}
 }
 
 // Delete an entry in the datastore. This is a no-op when using the k8s backend.
 func (c *KubeClient) Delete(d *model.KVPair) error {
+	log.Debugf("Performing 'Delete' for %+v", d)
 	switch d.Key.(type) {
 	case model.GlobalConfigKey:
 		return c.deleteGlobalConfig(d)
 	case model.IPPoolKey:
 		return c.ipPoolClient.Delete(d)
 	default:
-		log.Warn("Attempt to 'Delete' using kubernetes backend is not supported.")
+		log.Warn("Attempt to 'Delete' using kubernetes datastore driver is not supported.")
 		return nil
 	}
 }
@@ -414,11 +419,11 @@ func (c *KubeClient) listProfiles(l model.ProfileListOptions) ([]*model.KVPair,
 // getProfile gets the Profile from the k8s API based on existing Namespaces.
 func (c *KubeClient) getProfile(k model.ProfileKey) (*model.KVPair, error) {
 	if k.Name == "" {
-		return nil, goerrors.New("Missing profile name")
+		return nil, fmt.Errorf("Profile key missing name: %+v", k)
 	}
 	namespaceName, err := c.converter.parseProfileName(k.Name)
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("Failed to parse Profile name: %s", err)
 	}
 	namespace, err := c.clientSet.Namespaces().Get(namespaceName, metav1.GetOptions{})
 	if err != nil {
@@ -430,6 +435,8 @@ func (c *KubeClient) getProfile(k model.ProfileKey) (*model.KVPair, error) {
 
 // applyWorkloadEndpoint patches the existing Pod to include an IP address, if
 // one has been set on the workload endpoint.
+// TODO: This is only required as a workaround for an upstream k8s issue.  Once fixed,
+// this should be a no-op. See https://github.com/kubernetes/kubernetes/issues/39113
 func (c *KubeClient) applyWorkloadEndpoint(k *model.KVPair) (*model.KVPair, error) {
 	ips := k.Value.(*model.WorkloadEndpoint).IPv4Nets
 	if len(ips) > 0 {

lib/backend/k8s/resources/ippools.go

@@ -114,18 +114,32 @@ func (c *client) Get(key model.Key) (*model.KVPair, error) {
 
 func (c *client) List(list model.ListInterface) ([]*model.KVPair, error) {
 	kvps := []*model.KVPair{}
-	tprs := thirdparty.IpPoolList{}
 	l := list.(model.IPPoolListOptions)
 
-	// Build the request.
-	req := c.tprClient.Get().Resource("ippools").Namespace("kube-system")
+	// If the CIDR is specified, k8s will return a single resource
+	// rather than a list, so handle this case separately, using our
+	// Get method to return the single result.
 	if l.CIDR.IP != nil {
-		k := model.IPPoolKey{CIDR: l.CIDR}
-		req.Name(tprName(k))
+		log.Info("Performing IP pool List with name")
+		if kvp, err := c.Get(model.IPPoolKey{CIDR: l.CIDR}); err == nil {
+			kvps = append(kvps, kvp)
+		} else {
+			if !kerrors.IsNotFound(err) {
+				return nil, K8sErrorToCalico(err, l)
+			}
+		}
+		return kvps, nil
 	}
 
+	// Since are not performing an exact Get, Kubernetes will return a list
+	// of resources.
+	tprs := thirdparty.IpPoolList{}
+
 	// Perform the request.
-	err := req.Do().Into(&tprs)
+	err := c.tprClient.Get().
+		Resource("ippools").
+		Namespace("kube-system").
+		Do().Into(&tprs)
 	if err != nil {
 		// Don't return errors for "not found".  This just
 		// means there are no IPPools, and we should return

lib/backend/k8s/syncer.go

@@ -142,7 +142,7 @@ func (syn *kubeSyncer) readFromKubernetesAPI() {
 		// If we need to resync, do so.
 		if needsResync {
 			// Set status to ResyncInProgress.
-			log.Warnf("Resync required - latest versions: %+v", latestVersions)
+			log.Debugf("Resync required - latest versions: %+v", latestVersions)
 			syn.callbacks.OnStatusUpdated(api.ResyncInProgress)
 
 			// Get snapshot from datastore.
@@ -155,7 +155,7 @@ func (syn *kubeSyncer) readFromKubernetesAPI() {
 			// Send the snapshot through.
 			syn.sendUpdates(snap)
 
-			log.Warnf("Snapshot complete - start watch from %+v", latestVersions)
+			log.Debugf("Snapshot complete - start watch from %+v", latestVersions)
 			syn.callbacks.OnStatusUpdated(api.InSync)
 
 			// Create the Kubernetes API watchers.
@@ -226,6 +226,7 @@ func (syn *kubeSyncer) readFromKubernetesAPI() {
 
 		// Don't start watches if we're in oneshot mode.
 		if syn.OneShot {
+			log.Info("OneShot mode, do not start watches")
 			return
 		}
 
@@ -235,7 +236,7 @@ func (syn *kubeSyncer) readFromKubernetesAPI() {
 			log.Debugf("Incoming Namespace watch event. Type=%s", event.Type)
 			if needsResync = syn.eventTriggersResync(event); needsResync {
 				// We need to resync.  Break out into the sync loop.
-				log.Warn("Event triggered resync: %+v", event)
+				log.Warnf("Event triggered resync: %+v", event)
 				continue
 			}
 
@@ -248,7 +249,7 @@ func (syn *kubeSyncer) readFromKubernetesAPI() {
 			log.Debugf("Incoming Pod watch event. Type=%s", event.Type)
 			if needsResync = syn.eventTriggersResync(event); needsResync {
 				// We need to resync.  Break out into the sync loop.
-				log.Warn("Event triggered resync: %+v", event)
+				log.Warnf("Event triggered resync: %+v", event)
 				continue
 			}
 
@@ -263,7 +264,7 @@ func (syn *kubeSyncer) readFromKubernetesAPI() {
 			log.Debugf("Incoming NetworkPolicy watch event. Type=%s", event.Type)
 			if needsResync = syn.eventTriggersResync(event); needsResync {
 				// We need to resync.  Break out into the sync loop.
-				log.Warn("Event triggered resync: %+v", event)
+				log.Warnf("Event triggered resync: %+v", event)
 				continue
 			}
 
@@ -275,7 +276,7 @@ func (syn *kubeSyncer) readFromKubernetesAPI() {
 			log.Debugf("Incoming GlobalConfig watch event. Type=%s", event.Type)
 			if needsResync = syn.eventTriggersResync(event); needsResync {
 				// We need to resync.  Break out into the sync loop.
-				log.Warn("Event triggered resync: %+v", event)
+				log.Warnf("Event triggered resync: %+v", event)
 				continue
 			}
 
@@ -287,7 +288,7 @@ func (syn *kubeSyncer) readFromKubernetesAPI() {
 			log.Debugf("Incoming IPPool watch event. Type=%s", event.Type)
 			if needsResync = syn.eventTriggersResync(event); needsResync {
 				// We need to resync.  Break out into the sync loop.
-				log.Warn("Event triggered resync: %+v", event)
+				log.Warnf("Event triggered resync: %+v", event)
 				continue
 			}
 
@@ -339,6 +340,8 @@ func (syn *kubeSyncer) performSnapshot() ([]model.KVPair, map[string]bool, resou
 			time.Sleep(1 * time.Second)
 			continue
 		}
+		log.Info("Received Namespace List() response")
+
 		versions.namespaceVersion = nsList.ListMeta.ResourceVersion
 		for _, ns := range nsList.Items {
 			// The Syncer API expects a profile to be broken into its underlying
@@ -378,6 +381,7 @@ func (syn *kubeSyncer) performSnapshot() ([]model.KVPair, map[string]bool, resou
 			time.Sleep(1 * time.Second)
 			continue
 		}
+		log.Info("Received NetworkPolicy List() response")
 
 		versions.networkPolicyVersion = npList.ListMeta.ResourceVersion
 		for _, np := range npList.Items {
@@ -394,9 +398,17 @@ func (syn *kubeSyncer) performSnapshot() ([]model.KVPair, map[string]bool, resou
 			time.Sleep(1 * time.Second)
 			continue
 		}
+		log.Info("Received Pod List() response")
 
 		versions.podVersion = poList.ListMeta.ResourceVersion
 		for _, po := range poList.Items {
+			// Ignore any updates for host networked pods.
+			if syn.kc.converter.isHostNetworked(&po) {
+				log.Debugf("Skipping host networked pod %s/%s", po.ObjectMeta.Namespace, po.ObjectMeta.Name)
+				continue
+			}
+
+			// Convert to a workload endpoint.
 			wep, _ := syn.kc.converter.podToWorkloadEndpoint(&po)
 			if wep != nil {
 				snap = append(snap, *wep)
@@ -405,25 +417,29 @@ func (syn *kubeSyncer) performSnapshot() ([]model.KVPair, map[string]bool, resou
 		}
 
 		// Sync GlobalConfig.
+		log.Info("Syncing GlobalConfig")
 		confList, err := syn.kc.listGlobalConfig(model.GlobalConfigListOptions{})
 		if err != nil {
 			log.Warnf("Error querying GlobalConfig during snapshot, retrying: %s", err)
 			time.Sleep(1 * time.Second)
 			continue
 		}
+		log.Info("Received GlobalConfig List() response")
 
 		for _, c := range confList {
 			snap = append(snap, *c)
 			keys[c.Key.String()] = true
 		}
 
 		// Sync IP Pools.
+		log.Info("Syncing IP Pools")
 		poolList, err := syn.kc.List(model.IPPoolListOptions{})
 		if err != nil {
 			log.Warnf("Error querying IP Pools during snapshot, retrying: %s", err)
 			time.Sleep(1 * time.Second)
 			continue
 		}
+		log.Info("Received IP Pools List() response")
 
 		for _, p := range poolList {
 			snap = append(snap, *p)
@@ -506,10 +522,12 @@ func (syn *kubeSyncer) parsePodEvent(e watch.Event) *model.KVPair {
 		return nil
 	}
 
-	// Convert the received Namespace into a KVPair.
+	// Convert the received Pod into a KVPair.
 	kvp, err := syn.kc.converter.podToWorkloadEndpoint(pod)
 	if err != nil {
-		log.Panicf("%s", err)
+		// If we fail to parse, then ignore this update and emit a log.
+		log.WithField("error", err).Error("Failed to parse Pod event")
+		return nil
 	}
 
 	// We behave differently based on the event type.
@@ -519,7 +537,7 @@ func (syn *kubeSyncer) parsePodEvent(e watch.Event) *model.KVPair {
 		log.Debugf("Delete for pod %s/%s", pod.ObjectMeta.Namespace, pod.ObjectMeta.Name)
 		kvp.Value = nil
 
-		// Remove it from the cache, if it is there.
+		// Remove it from the label cache, if it is there.
 		workload := kvp.Key.(model.WorkloadEndpointKey).WorkloadID
 		delete(syn.labelCache, workload)
 	default: