Various improvements and bugfixes for kube-apiserver policy matching

[christarazi]

Please see individual commit msgs.

• ipcache: Update IDMD diagram
• ipcache: Fix incorrect ipcache source used inside IDMD
• ipcache: Reduce scope for propagating updates to ipcache
• ipcache: Remove incorrect identity release upon label injection
• ipcache: Fix incorrect variable used in RemoveLabelsFromIPs
• testutils/identity: Implement proper refcounting for mock
• ipcache: Generate new identity if label is removed
• ipcache, k8s/watchers: Fix non-atomic window when removing labels

Related: #17962

Stablize kube-apiserver policy matching feature, namely by fixing unncessary identity churn when kube-apiserver is running outside of the cluster

[christarazi]

/test

[christarazi]

/test

Job 'Cilium-PR-Runtime-net-next' failed and has not been observed before, so may be related to your PR:

Click to show.

Test Name

RuntimePrivilegedUnitTests Run Tests

Failure Output

FAIL: Failed to run privileged unit tests

If it is a flake, comment /mlh new-flake Cilium-PR-Runtime-net-next so I can create a new GitHub issue to track it.

[christarazi]

/test-runtime

[christarazi]

/test-gke

[pchaigno]

Only nits. My codeowners look good.

[nbusseneau]

LGTM, thanks :)

[christarazi]

/test

[christarazi]

/test

Job 'Cilium-PR-K8s-1.23-kernel-net-next' failed and has not been observed before, so may be related to your PR:

Click to show.

Test Name

K8sBandwidthTest Checks Bandwidth Rate-Limiting Checks Pod to Pod bandwidth, vxlan tunneling

Failure Output

FAIL: failed to ensure kubectl version: failed to run kubectl version

If it is a flake, comment /mlh new-flake Cilium-PR-K8s-1.23-kernel-net-next so I can create a new GitHub issue to track it.

[christarazi]

/test-1.23-net-next

[joestringer]

Nice work. I've posted a few questions mainly around the second-to-last commit where it seems like we can potentially simplify the code a little further and also eliminate one source of policy drops when labels are injected or removed.

[christarazi]

Relevant diff addressing Joe's comments above:

Diff

diff --git a/pkg/ipcache/ipcache.go b/pkg/ipcache/ipcache.go
index 7e1107c6fe..6ebc0466e0 100644
--- a/pkg/ipcache/ipcache.go
+++ b/pkg/ipcache/ipcache.go
@@ -223,10 +223,23 @@ func (ipc *IPCache) updateNamedPorts() (namedPortsChanged bool) {
 func (ipc *IPCache) Upsert(ip string, hostIP net.IP, hostKey uint8, k8sMeta *K8sMetadata, newIdentity Identity) (namedPortsChanged bool, err error) {
 	ipc.mutex.Lock()
 	defer ipc.mutex.Unlock()
-	return ipc.upsertLocked(ip, hostIP, hostKey, k8sMeta, newIdentity)
-}
-
-func (ipc *IPCache) upsertLocked(ip string, hostIP net.IP, hostKey uint8, k8sMeta *K8sMetadata, newIdentity Identity) (namedPortsChanged bool, err error) {
+	return ipc.upsertLocked(ip, hostIP, hostKey, k8sMeta, newIdentity, false /* !force */)
+}
+
+// upsertLocked adds / updates the provided IP and identity into the IPCache,
+// assuming that the IPCache lock has been taken. Warning, do not use force
+// unless you know exactly what you're doing. Forcing adding / updating the
+// IPCache will not take into account the source of the identity and bypasses
+// the overwrite logic! Once GH-18301 is addressed, there will be no need for
+// any force logic.
+func (ipc *IPCache) upsertLocked(
+	ip string,
+	hostIP net.IP,
+	hostKey uint8,
+	k8sMeta *K8sMetadata,
+	newIdentity Identity,
+	force bool,
+) (namedPortsChanged bool, err error) {
 	var newNamedPorts policy.NamedPortMap
 	if k8sMeta != nil {
 		newNamedPorts = k8sMeta.NamedPorts
@@ -258,7 +271,7 @@ func (ipc *IPCache) upsertLocked(ip string, hostIP net.IP, hostKey uint8, k8sMet
 
 	cachedIdentity, found := ipc.ipToIdentityCache[ip]
 	if found {
-		if !source.AllowOverwrite(cachedIdentity.Source, newIdentity.Source) {
+		if !force && !source.AllowOverwrite(cachedIdentity.Source, newIdentity.Source) {
 			return false, NewErrOverwrite(cachedIdentity.Source, newIdentity.Source)
 		}
 
diff --git a/pkg/ipcache/metadata.go b/pkg/ipcache/metadata.go
index 22d2415423..ecdd4a80ed 100644
--- a/pkg/ipcache/metadata.go
+++ b/pkg/ipcache/metadata.go
@@ -73,12 +73,15 @@ func GetIDMetadataByIP(prefix string) labels.Labels {
 // InjectLabels injects labels from the identityMetadata (IDMD) map into the
 // identities used for the prefixes in the IPCache. The given source is the
 // source of the caller, as inserting into the IPCache requires knowing where
-// this updated information comes from.
+// this updated information comes from. Conversely, RemoveLabelsExcluded()
+// performs the inverse: removes labels from the IDMD map and releases
+// identities allocated by this function.
 //
 // Note that as this function iterates through the IDMD, if it detects a change
 // in labels for a given prefix, then this might allocate a new identity. If a
 // prefix was previously associated with an identity, it will get deallocated,
-// so a balance is kept.
+// so a balance is kept, ensuring a one-to-one mapping between prefix and
+// identity.
 func InjectLabels(src source.Source, updater identityUpdater, triggerer policyTriggerer) error {
 	if IdentityAllocator == nil || !IdentityAllocator.IsLocalIdentityAllocatorInitialized() {
 		return ErrLocalIdentityAllocatorUninitialized
@@ -131,8 +134,8 @@ func InjectLabels(src source.Source, updater identityUpdater, triggerer policyTr
 				newLbls := id.Labels
 				tmpSrc = source.KubeAPIServer
 				trigger = true
-				// If any reserved ID has changed, update its labels.
-				if id.IsReserved() {
+				// If host identity has changed, update its labels.
+				if id.ID == identity.ReservedIdentityHost {
 					identity.AddReservedIdentityWithLabels(id.ID, newLbls)
 				}
 				idsToPropagate[id.ID] = newLbls.LabelArray()
@@ -189,7 +192,7 @@ func InjectLabels(src source.Source, updater identityUpdater, triggerer policyTr
 		if _, err := IPIdentityCache.upsertLocked(ip, hIP, key, meta, Identity{
 			ID:     id.ID,
 			Source: id.Source,
-		}); err != nil {
+		}, false /* !force */); err != nil {
 			return fmt.Errorf("failed to upsert %s into ipcache with identity %d: %w", ip, id.ID, err)
 		}
 	}
@@ -305,12 +308,9 @@ func filterMetadataByLabels(filter labels.Labels) []string {
 	return matching
 }
 
-// removeLabelsFromIPs provides a convenient method for the caller to remove
-// all given prefixes at once. This function should be the inverse of
-// InjectLabels(). This function will trigger policy update and recalculation
-// if necessary on behalf of the caller.
-//
-// Identities allocated by InjectLabels() may be released by removeLabels().
+// removeLabelsFromIPs removes all given prefixes at once. This function will
+// trigger policy update and recalculation if necessary on behalf of the
+// caller.
 //
 // A prefix will only be removed from the IDMD if the set of labels becomes
 // empty.
@@ -325,8 +325,8 @@ func removeLabelsFromIPs(
 	var (
 		idsToAdd    = make(map[identity.NumericIdentity]labels.LabelArray)
 		idsToDelete = make(map[identity.NumericIdentity]labels.LabelArray)
-		// toReplace stores identity pairs to replace in the ipcache.
-		toReplace = make(map[string]idPair)
+		// toReplace stores the identity to replace in the ipcache.
+		toReplace = make(map[string]Identity)
 	)
 
 	IPIdentityCache.Lock()
@@ -338,48 +338,25 @@ func removeLabelsFromIPs(
 			continue
 		}
 
+		idsToDelete[id.ID] = nil // labels for deletion don't matter to UpdateIdentities()
+
 		// Insert to propagate the updated set of labels after removal.
-		//
-		// There are two concurring cases here and they are annotated in the
-		// code below:
-		//   1) When labels are disassociated with an IP, the remaining set
-		//      either contains some labels or it's empty. In the former, the
-		//      identity is both marked for deletion and addition, because
-		//      there are labels still associated with the IP. In the latter,
-		//      the identity is marked for deletion only.
-		//   2) If removeLabels() indicates that a new identity needs to be
-		//      generated (newIdentity), a new identity is allocated (with the
-		//      appropriate labels removed), marked for addition, and lastly,
-		//      upserted into the ipcache (replacing the existing identity
-		//      entry). This can happen when an existing identity was *not*
-		//      released during removeLabels() due to the identity's refcount
-		//      being >1.
-		var (
-			newIdentity bool
-			la          labels.LabelArray
-		)
-		l, newIdentity := removeLabels(prefix, lbls, src)
-		if l != nil {
-			la = l.LabelArray()
-		}
-		idsToDelete[id.ID] = la
-		if len(la) > 0 { // (1)
-			// If for example kube-apiserver label is removed from
-			// a remote-node, then removeLabels() will return a
-			// non-empty set representing the new full set of
-			// labels to associate with the node. In order to
-			// propagate the new identity, we must emit a delete
-			// event for the old identity and then an add event for
-			// the new identity.
-			idsToAdd[id.ID] = la
-		}
-		if newIdentity { // (2)
-			// If for example kube-apiserver label is removed from a CIDR
-			// identity (contains CIDR labels), then removeLabels() will return
-			// a non-empty set representing just the CIDR labels. We do the
-			// same as (1) from above, but additionally, we must also mark the
-			// new identity for addition and upsert the new identity into the
-			// IPCache.
+		l := removeLabels(prefix, lbls, src)
+		if len(l) > 0 {
+			// If for example kube-apiserver label is removed from the local
+			// host identity (when the kube-apiserver is running within the
+			// cluster and it is no longer running on the current local host),
+			// then removeLabels() will return a non-empty set representing the
+			// new full set of labels to associate with the node (in this
+			// example, just the local host label). In order to propagate the
+			// new identity, we must emit a delete event for the old identity
+			// and then an add event for the new identity.
+
+			// If host identity has changed, update its labels.
+			if id.ID == identity.ReservedIdentityHost {
+				identity.AddReservedIdentityWithLabels(id.ID, l)
+			}
+
 			newID, _, err := IdentityAllocator.AllocateIdentity(context.TODO(), l, false)
 			if err != nil {
 				log.WithError(err).WithFields(logrus.Fields{
@@ -392,16 +369,10 @@ func removeLabelsFromIPs(
 				)
 				continue
 			}
-			idsToAdd[newID.ID] = la
-			toReplace[prefix] = idPair{
-				old: Identity{
-					ID:     id.ID,
-					Source: sourceByLabels(src, lbls),
-				},
-				new: Identity{
-					ID:     newID.ID,
-					Source: sourceByLabels(src, l),
-				},
+			idsToAdd[newID.ID] = l.LabelArray()
+			toReplace[prefix] = Identity{
+				ID:     newID.ID,
+				Source: sourceByLabels(src, l),
 			}
 		}
 	}
@@ -417,64 +388,89 @@ func removeLabelsFromIPs(
 
 		triggerer.TriggerPolicyUpdates(false, "kube-apiserver identity updated by removal")
 	}
-	for ip, pair := range toReplace {
+	for ip, id := range toReplace {
 		hIP, key := IPIdentityCache.getHostIPCache(ip)
 		meta := IPIdentityCache.getK8sMetadata(ip)
-		IPIdentityCache.deleteLocked(ip, pair.old.Source) // remove old entry and replace
-		if _, err := IPIdentityCache.upsertLocked(ip, hIP, key, meta, Identity{
-			ID:     pair.new.ID,
-			Source: pair.new.Source,
-		}); err != nil {
+		if _, err := IPIdentityCache.upsertLocked(
+			ip,
+			hIP,
+			key,
+			meta,
+			id,
+			true, /* force upsert */
+		); err != nil {
 			log.WithError(err).WithFields(logrus.Fields{
 				logfields.IPAddr:   ip,
-				logfields.Identity: pair.new,
+				logfields.Identity: id,
 			}).Error("Failed to replace ipcache entry with new identity after label removal. Traffic may be disrupted.")
 		}
 	}
 }
 
-type idPair struct {
-	old, new Identity
-}
-
 // removeLabels removes the given labels association with the given prefix. The
-// leftover labels are returned, if any, and a boolean indicating whether the
-// caller must generate a new identity with the leftover labels. If the caller
-// must do so, then the following must occur *in order* to avoid drops:
-//   1) policy recalculation must be triggered
+// leftover labels are returned, if any. If there are leftover labels, the
+// caller must allocate a new identity and do the following *in order* to avoid
+// drops:
+//   1) policy recalculation must be implemented into the datapath and
 //   2) new identity must have a new entry upserted into the IPCache
+// Note: GH-17962, triggering policy recalculation doesn't actually *implement*
+// the changes into datapath (because it's an async call), this is a known
+// issue. There's a very small window for drops when two policies select the
+// same traffic and the identity changes. For example, this is possible if an
+// IP is associated with the kube-apiserver and referenced inside a ToCIDR
+// policy, and then the IP is no longer associated with the kube-apiserver.
 //
 // Identities are deallocated and their subequent entry in the IPCache is
 // removed if the prefix is no longer associated with any labels.
 //
 // This function assumes that the IDMD and the IPIdentityCache locks are taken!
-//
-// It is the responsibility of the caller to trigger policy recalculation after
-// calling this function.
-func removeLabels(prefix string, lbls labels.Labels, src source.Source) (labels.Labels, bool) {
-	var needNewIdentity bool
-
+func removeLabels(prefix string, lbls labels.Labels, src source.Source) labels.Labels {
 	l, ok := identityMetadata[prefix]
 	if !ok {
-		return nil, needNewIdentity
+		return nil
 	}
 
 	l = l.Remove(lbls)
-	if len(l) != 0 { // Labels left over, do not deallocate
+	if len(l) == 0 { // Labels empty, delete
+		// No labels left. Example case: when the kube-apiserver is running
+		// outside of the cluster, meaning that the IDMD only ever had the
+		// kube-apiserver label (CIDR labels are not added) and it's now being
+		// removed.
+		delete(identityMetadata, prefix)
+	} else {
+		// This case is only hit if the prefix for the kube-apiserver is
+		// running within the cluster. Therefore, the IDMD can only ever has
+		// the following labels:
+		//   * kube-apiserver + remote-node
+		//   * kube-apiserver + host
 		identityMetadata[prefix] = l
-		return l, needNewIdentity
 	}
 
-	// No labels left, perform deallocation
-
-	delete(identityMetadata, prefix)
 	id, exists := IPIdentityCache.LookupByIPRLocked(prefix)
 	if !exists {
-		return nil, needNewIdentity
+		log.WithFields(logrus.Fields{
+			logfields.CIDR:   prefix,
+			logfields.Labels: lbls,
+		}).Warn(
+			"Identity for prefix was unexpectedly not found in ipcache, unable " +
+				"to remove labels from prefix. If a network policy is applied, check " +
+				"for any drops. It's possible that insertion or removal from " +
+				"the ipcache failed.",
+		)
+		return nil
 	}
 	realID := IdentityAllocator.LookupIdentityByID(context.TODO(), id.ID)
 	if realID == nil {
-		return nil, needNewIdentity
+		log.WithFields(logrus.Fields{
+			logfields.CIDR:     prefix,
+			logfields.Labels:   lbls,
+			logfields.Identity: id,
+		}).Warn(
+			"Identity unexpectedly not found within the identity allocator, " +
+				"unable to remove labels from prefix. It's possible that insertion " +
+				"or removal from the ipcache failed.",
+		)
+		return nil
 	}
 	released, err := IdentityAllocator.Release(context.TODO(), realID, false)
 	if err != nil {
@@ -486,21 +482,21 @@ func removeLabels(prefix string, lbls labels.Labels, src source.Source) (labels.
 		}).Error(
 			"Failed to release assigned identity to IP while removing label association, this might be a leak.",
 		)
+		return nil
 	}
 	if released {
 		IPIdentityCache.deleteLocked(prefix, sourceByLabels(src, lbls))
-		return nil, needNewIdentity
+		return nil
 	}
 
 	// Generate new identity with the label removed. This should be the case
 	// where the existing identity had >1 refcount, meaning that something was
 	// referring to it.
-	needNewIdentity = true
 	allLbls := labels.NewLabelsFromModel(nil)
 	allLbls.MergeLabels(realID.Labels)
 	allLbls = allLbls.Remove(lbls)
 
-	return allLbls, needNewIdentity
+	return allLbls
 }
 
 func sourceByLabels(d source.Source, lbls labels.Labels) source.Source {

[christarazi]

/test

[joestringer]

LGTM.