Repository: cilium/cilium
Pull request #14368 (2020-12-10 22:16:25)

policy: Track selectors that contribute to MapStateEntries
[ upstream commit 04840b96530031a84bc359c476a59d320617d2db ]

Track which selectors in policy require a specific bpf policy map key to
be present, and keep policy entries in the map as long as any selector
requires it's presence. Without this it is possible for a timed-out
DNS cache entry to clear a policy cache key that is still required by
another selector (FQDN or CIDR).

To implement this, each MapStateEntry is now equipped with a set of
(cached) selectors through which the policy map key/value was
added. 'nil' has the special significance that it is used as the
CachedSelector in cases where the policy map entry is added due to
some administrative or configuration reason. Currently incremental
updates will never remove such entries.

Incremental policy updates now simply collect the requested map
changes. When the endpoint then pulls the changes they are first
applied the desired policy map (MapState), while tallying which
selectors still need the map entries to be present. The actual bpf map
diffs are recorded based on the total count of selectors on each map
entry.

Signed-off-by: Jarno Rajahalme <jarno@covalent.io>


jrajahalme: test-backport-1.7
jrajahalme: test-backport-1.7

Patch:
diff --git a/pkg/endpoint/bpf.go b/pkg/endpoint/bpf.go
index 665bc3a152..c4c491ab21 100644
--- a/pkg/endpoint/bpf.go
+++ b/pkg/endpoint/bpf.go
@@ -250,7 +250,7 @@ func (e *Endpoint) addNewRedirectsFromDesiredPolicy(ingress bool, desiredRedirec
 				} else {
 					insertedDesiredMapState[keyFromFilter] = struct{}{}
 				}
-				if entry != policy.NoRedirectEntry {
+				if entry.IsRedirectEntry() {
 					entry.ProxyPort = redirectPort
 				}
 				e.desiredPolicy.PolicyMapState[keyFromFilter] = entry
@@ -360,9 +360,10 @@ func (e *Endpoint) addVisibilityRedirects(ingress bool, desiredRedirects map[str
 			TrafficDirection: direction.Uint8(),
 		}
 
-		e.desiredPolicy.PolicyMapState[newKey] = policy.MapStateEntry{
-			ProxyPort: redirectPort,
-		}
+		entry := policy.NewMapStateEntry(nil, true)
+		entry.ProxyPort = redirectPort
+
+		e.desiredPolicy.PolicyMapState[newKey] = entry
 
 		insertedDesiredMapState[newKey] = struct{}{}
 	}
@@ -1027,11 +1028,6 @@ func (e *Endpoint) deletePolicyKey(keyToDelete policy.Key, incremental bool, had
 	// Operation was successful, remove from realized state.
 	delete(e.realizedPolicy.PolicyMapState, keyToDelete)
 
-	// Incremental updates need to update the desired state as well.
-	if incremental && e.desiredPolicy != e.realizedPolicy {
-		delete(e.desiredPolicy.PolicyMapState, keyToDelete)
-	}
-
 	e.policyDebug(logrus.Fields{
 		logfields.BPFMapKey:   keyToDelete,
 		logfields.BPFMapValue: entry,
@@ -1062,11 +1058,6 @@ func (e *Endpoint) addPolicyKey(keyToAdd policy.Key, entry policy.MapStateEntry,
 	// Operation was successful, add to realized state.
 	e.realizedPolicy.PolicyMapState[keyToAdd] = entry
 
-	// Incremental updates need to update the desired state as well.
-	if incremental && e.desiredPolicy != e.realizedPolicy {
-		e.desiredPolicy.PolicyMapState[keyToAdd] = entry
-	}
-
 	e.policyDebug(logrus.Fields{
 		logfields.BPFMapKey:   keyToAdd,
 		logfields.BPFMapValue: entry,
@@ -1116,12 +1107,15 @@ func (e *Endpoint) applyPolicyMapChanges() (proxyChanges bool, err error) {
 	//  collected on the newly computed desired policy, which is
 	//  not fully realized yet. This is why we get the map changes
 	//  from the desired policy here.
+	//  ConsumeMapChanges() applies the incremental updates to the
+	//  desired policy and only returns changes that need to be
+	//  applied to the Endpoint's bpf policy map.
 	adds, deletes := e.desiredPolicy.ConsumeMapChanges()
 
 	// Add policy map entries before deleting to avoid transient drops
 	for keyToAdd, entry := range adds {
 		// Keep the existing proxy port, if any
-		if entry != policy.NoRedirectEntry {
+		if entry.IsRedirectEntry() {
 			entry.ProxyPort = e.realizedRedirects[policy.ProxyIDFromKey(e.ID, keyToAdd)]
 			if entry.ProxyPort != 0 {
 				proxyChanges = true
@@ -1201,7 +1195,7 @@ func (e *Endpoint) addPolicyMapDelta() error {
 	errors := 0
 
 	for keyToAdd, entry := range e.desiredPolicy.PolicyMapState {
-		if oldEntry, ok := e.realizedPolicy.PolicyMapState[keyToAdd]; !ok || oldEntry != entry {
+		if oldEntry, ok := e.realizedPolicy.PolicyMapState[keyToAdd]; !ok || !oldEntry.Equal(&entry) {
 			if !e.addPolicyKey(keyToAdd, entry, false) {
 				errors++
 			}
@@ -1230,10 +1224,6 @@ func (e *Endpoint) syncPolicyMapWithDump() error {
 		e.realizedPolicy.PolicyMapState = make(policy.MapState)
 	}
 
-	if e.desiredPolicy.PolicyMapState == nil {
-		e.desiredPolicy.PolicyMapState = make(policy.MapState)
-	}
-
 	if e.policyMap == nil {
 		return fmt.Errorf("not syncing PolicyMap state for endpoint because PolicyMap is nil")
 	}
diff --git a/pkg/policy/l4.go b/pkg/policy/l4.go
index f71e0570e8..256dc2c498 100644
--- a/pkg/policy/l4.go
+++ b/pkg/policy/l4.go
@@ -241,7 +241,7 @@ func (l4 *L4Filter) GetPort() uint16 {
 }
 
 // ToMapState converts filter into a MapState with two possible values:
-// - NoRedirectEntry (ProxyPort = 0): No proxy redirection is needed for this key
+// - Entry with ProxyPort = 0: No proxy redirection is needed for this key
 // - Entry with any other port #: Proxy redirection is required for this key,
 //                                caller must replace the ProxyPort with the actual
 //                                listening port number.
@@ -291,10 +291,7 @@ func (l4 *L4Filter) ToMapState(direction trafficdirection.TrafficDirection) MapS
 			}).Debug("ToMapState: Skipping L3/L4 key due to existing L4-only key")
 			continue
 		}
-		entry := NoRedirectEntry
-		if l7 != nil {
-			entry = redirectEntry
-		}
+		entry := NewMapStateEntry(cs, l7 != nil)
 
 		if cs.IsWildcard() {
 			keyToAdd.Identity = 0
@@ -355,15 +352,15 @@ func (l4 *L4Filter) IdentitySelectionUpdated(selector CachedSelector, selections
 
 	// Push endpoint policy changes.
 	//
-	// `l4.policy` is set to nil when the filter is detached so
-	// that we could not push updates on a stale policy.
+	// `l4.policy` is nil when the filter is detached so
+	// that we could not push updates on an unstable policy.
 	l4Policy := (*L4Policy)(atomic.LoadPointer(&l4.policy))
 	if l4Policy != nil {
 		direction := trafficdirection.Egress
 		if l4.Ingress {
 			direction = trafficdirection.Ingress
 		}
-		l4Policy.AccumulateMapChanges(added, deleted, uint16(l4.Port), uint8(l4.U8Proto), direction,
+		l4Policy.AccumulateMapChanges(selector, added, deleted, uint16(l4.Port), uint8(l4.U8Proto), direction,
 			l4.L7RulesPerSelector[selector] != nil)
 	}
 }
@@ -540,11 +537,15 @@ func (l4 *L4Filter) removeSelectors(selectorCache *SelectorCache) {
 
 // detach releases the references held in the L4Filter and must be called before
 // the filter is left to be garbage collected.
+// L4Filter may still be accessed concurrently after it has been detached.
 func (l4 *L4Filter) detach(selectorCache *SelectorCache) {
 	l4.removeSelectors(selectorCache)
 	l4.attach(nil, nil)
 }
 
+// attach signifies that the L4Filter is ready and reacheable for updates
+// from SelectorCache. L4Filter is read-only after this is called,
+// multiple goroutines will be reading the fields from that point on.
 func (l4 *L4Filter) attach(ctx PolicyContext, l4Policy *L4Policy) {
 	// Compute Envoy policies when a policy is ready to be used
 	if ctx != nil {
@@ -664,6 +665,7 @@ func (l4 L4PolicyMap) Detach(selectorCache *SelectorCache) {
 }
 
 // Attach makes all the L4Filters to point back to the L4Policy that contains them.
+// This is done before the L4PolicyMap is exposed to concurrent access.
 func (l4 L4PolicyMap) Attach(ctx PolicyContext, l4Policy *L4Policy) {
 	for _, f := range l4 {
 		f.attach(ctx, l4Policy)
@@ -803,11 +805,11 @@ func (l4 *L4Policy) insertUser(user *EndpointPolicy) {
 //
 // The caller is responsible for making sure the same identity is not
 // present in both 'adds' and 'deletes'.
-func (l4 *L4Policy) AccumulateMapChanges(adds, deletes []identity.NumericIdentity,
+func (l4 *L4Policy) AccumulateMapChanges(cs CachedSelector, adds, deletes []identity.NumericIdentity,
 	port uint16, proto uint8, direction trafficdirection.TrafficDirection, redirect bool) {
 	l4.mutex.RLock()
 	for epPolicy := range l4.users {
-		epPolicy.policyMapChanges.AccumulateMapChanges(adds, deletes, port, proto, direction, redirect)
+		epPolicy.policyMapChanges.AccumulateMapChanges(cs, adds, deletes, port, proto, direction, redirect)
 	}
 	l4.mutex.RUnlock()
 }
diff --git a/pkg/policy/mapstate.go b/pkg/policy/mapstate.go
index 03cb29ed8b..36e647bd95 100644
--- a/pkg/policy/mapstate.go
+++ b/pkg/policy/mapstate.go
@@ -80,6 +80,50 @@ type MapStateEntry struct {
 	// If 0 (default), there is no proxy redirection for the corresponding
 	// Key. Any other value signifies proxy redirection.
 	ProxyPort uint16
+
+	// Selectors collects the selectors in the policy that require this key to be present.
+	// TODO: keep track which selector needed the entry to be a redirect, or just allow.
+	selectors map[CachedSelector]struct{}
+}
+
+// NewMapStateEntry creates a map state entry. If redirect is true, the
+// caller is expected to replace the ProxyPort field before it is added to
+// the actual BPF map.
+// 'cs' is used to keep track of which policy selectors need this entry. If it is 'nil' this entry
+// will become sticky and cannot be completely removed via incremental updates.
+func NewMapStateEntry(cs CachedSelector, redirect bool) MapStateEntry {
+	var proxyPort uint16
+	if redirect {
+		// Any non-zero value will do, as the callers replace this with the
+		// actual proxy listening port number before the entry is added to the
+		// actual bpf map.
+		proxyPort = 1
+	}
+
+	return MapStateEntry{
+		ProxyPort: proxyPort,
+		selectors: map[CachedSelector]struct{}{cs: {}},
+	}
+}
+
+// MergeSelectors adds selectors from entry 'b' to 'e'. 'b' is not modified.
+func (e *MapStateEntry) MergeSelectors(b *MapStateEntry) {
+	for cs, v := range b.selectors {
+		e.selectors[cs] = v
+	}
+}
+
+// IsRedirectEntry returns true if e contains a redirect
+func (e *MapStateEntry) IsRedirectEntry() bool {
+	return e.ProxyPort != 0
+}
+
+// Equal returns true of two entries are equal
+func (e *MapStateEntry) Equal(o *MapStateEntry) bool {
+	if e == nil || o == nil {
+		return e == o
+	}
+	return e.ProxyPort == o.ProxyPort
 }
 
 // String returns a string representation of the MapStateEntry
@@ -87,28 +131,95 @@ func (e MapStateEntry) String() string {
 	return fmt.Sprintf("ProxyPort=%d", e.ProxyPort)
 }
 
-// NoRedirectEntry is a special MapStateEntry used to signify that the
-// entry is not redirecting to a proxy.
-var NoRedirectEntry = MapStateEntry{ProxyPort: 0}
-
-// redirectEntry is a MapStateEntry used to signify that the entry
-// must redirect to a proxy. Any non-zero value would do, as the
-// callers replace this with the actual proxy listening port number
-// before the entry is added to the actual bpf map.
-var redirectEntry = MapStateEntry{ProxyPort: 1}
-
 // RedirectPreferredInsert inserts a new entry giving priority to L7-redirects by
 // not overwriting a L7-redirect entry with a non-redirect entry
+// This form may be used when a full policy is computed and we are not yet interested
+// in accumulating incremental changes.
 func (keys MapState) RedirectPreferredInsert(key Key, entry MapStateEntry) {
-	if entry == NoRedirectEntry {
-		if _, ok := keys[key]; ok {
-			// Key already exist, keep the existing entry so that
-			// a redirect entry is never overwritten by a non-redirect
-			// entry
-			return
+	keys.redirectPreferredInsertWithChanges(key, entry, nil, nil)
+}
+
+// addKeyWithChanges adds a 'key' with value 'entry' to 'keys' keeping track of incremental changes in 'adds' and 'deletes'
+func (keys MapState) addKeyWithChanges(key Key, entry MapStateEntry, adds, deletes MapState) {
+	// Keep all selectors that need this entry so that it is deleted only if all the selectors delete their contribution
+	updatedEntry := entry
+	oldEntry, exists := keys[key]
+	if exists {
+		// keep the existing selectors map of the old entry
+		updatedEntry.selectors = oldEntry.selectors
+	} else if len(entry.selectors) > 0 {
+		// create a new selectors map
+		updatedEntry.selectors = make(map[CachedSelector]struct{}, len(entry.selectors))
+	} else {
+		// Keep the map as nil when empty. This makes a difference for unit testing only.
+		// MergeSelectors below becomes a no-op as 'entry' is empty.
+		updatedEntry.selectors = nil
+	}
+
+	// TODO: Do we need to merge labels as well?
+	// Merge new selectors to the updated entry without modifying 'entry' as it is being reused by the caller
+	updatedEntry.MergeSelectors(&entry)
+	// Update (or insert) the entry
+	keys[key] = updatedEntry
+
+	// Record an incremental Add if desired and entry is new or changed
+	if adds != nil && (!exists || oldEntry.ProxyPort != entry.ProxyPort) {
+		updatedEntry.selectors = nil
+		adds[key] = updatedEntry // copy w/o selectors
+		// Key add overrides any previous delete of the same key
+		if deletes != nil {
+			delete(deletes, key)
+		}
+	}
+}
+
+// deleteKeyWithChanges deletes a 'key' from 'keys' keeping track of incremental changes in 'adds' and 'deletes'.
+// The key is unconditionally deleted if 'cs' is nil, otherwise only the contribution of this 'cs' is removed.
+func (keys MapState) deleteKeyWithChanges(key Key, cs CachedSelector, adds, deletes MapState) {
+	if entry, exists := keys[key]; exists {
+		if cs != nil {
+			// remove the contribution of the given selector only
+			if _, exists = entry.selectors[cs]; exists {
+				// Remove the contribution of this selector from the entry
+				delete(entry.selectors, cs)
+				// key is not deleted if other selectors still need it
+				if len(entry.selectors) > 0 {
+					return
+				}
+			}
 		}
+		if deletes != nil {
+			entry.selectors = nil
+			deletes[key] = entry // copy w/o selectors
+			// Remove a potential previously added key
+			if adds != nil {
+				delete(adds, key)
+			}
+		}
+		delete(keys, key)
 	}
-	keys[key] = entry
+}
+
+// redirectPreferredInsertWithChanges inserts a new entry giving priority to L7-redirects by
+// not overwriting a L7-redirect entry with a non-redirect entry.
+func (keys MapState) redirectPreferredInsertWithChanges(key Key, entry MapStateEntry, adds, deletes MapState) {
+	// Do not overwrite the entry, but only merge selectors if the old entry is a redirect.
+	// This prevents an existing redirect being overridden by a non-redirect.
+	if oldEntry, exists := keys[key]; exists && oldEntry.IsRedirectEntry() {
+		oldEntry.MergeSelectors(&entry)
+		keys[key] = oldEntry
+		// For compatibility with old redirect management code we'll have to pass on
+		// redirect entry if the oldEntry is also a redirect, even if they are equal.
+		// We store the new entry here, the proxy port of it will be fixed up before
+		// insertion to the bpf map.
+		if adds != nil && entry.IsRedirectEntry() {
+			entry.selectors = nil
+			adds[key] = entry // copy w/o selectors
+		}
+		return
+	}
+	// Otherwise write the entry to the map
+	keys.addKeyWithChanges(key, entry, adds, deletes)
 }
 
 // DetermineAllowLocalhostIngress determines whether communication should be allowed
@@ -117,9 +228,9 @@ func (keys MapState) RedirectPreferredInsert(key Key, entry MapStateEntry) {
 // endpoint.
 func (keys MapState) DetermineAllowLocalhostIngress(l4Policy *L4Policy) {
 	if option.Config.AlwaysAllowLocalhost() {
-		keys[localHostKey] = MapStateEntry{}
+		keys[localHostKey] = NewMapStateEntry(nil, false)
 		if !option.Config.EnableRemoteNodeIdentity {
-			keys[localRemoteNodeKey] = MapStateEntry{}
+			keys[localRemoteNodeKey] = NewMapStateEntry(nil, false)
 		}
 	}
 }
@@ -135,7 +246,7 @@ func (keys MapState) AllowAllIdentities(ingress, egress bool) {
 			Nexthdr:          0,
 			TrafficDirection: trafficdirection.Ingress.Uint8(),
 		}
-		keys[keyToAdd] = MapStateEntry{}
+		keys[keyToAdd] = NewMapStateEntry(nil, false)
 	}
 	if egress {
 		keyToAdd := Key{
@@ -144,7 +255,7 @@ func (keys MapState) AllowAllIdentities(ingress, egress bool) {
 			Nexthdr:          0,
 			TrafficDirection: trafficdirection.Egress.Uint8(),
 		}
-		keys[keyToAdd] = MapStateEntry{}
+		keys[keyToAdd] = NewMapStateEntry(nil, false)
 	}
 }
 
@@ -153,20 +264,21 @@ func (keys MapState) AllowAllIdentities(ingress, egress bool) {
 // and deletes. 'mutex' must be held for any access.
 type MapChanges struct {
 	mutex   lock.Mutex
-	adds    MapState
-	deletes MapState
+	changes []MapChange
+}
+
+type MapChange struct {
+	add   bool // false deletes
+	key   Key
+	value MapStateEntry
 }
 
 // AccumulateMapChanges accumulates the given changes to the
-// MapChanges, updating both maps for each add and delete, as
-// applicable.
+// MapChanges.
 //
 // The caller is responsible for making sure the same identity is not
-// present in both 'adds' and 'deletes'.  Accross multiple calls we
-// maintain the adds and deletes within the MapChanges are disjoint in
-// cases where an identity is first added and then deleted, or first
-// deleted and then added.
-func (mc *MapChanges) AccumulateMapChanges(adds, deletes []identity.NumericIdentity,
+// present in both 'adds' and 'deletes'.
+func (mc *MapChanges) AccumulateMapChanges(cs CachedSelector, adds, deletes []identity.NumericIdentity,
 	port uint16, proto uint8, direction trafficdirection.TrafficDirection, redirect bool) {
 	key := Key{
 		// The actual identity is set in the loops below
@@ -176,13 +288,12 @@ func (mc *MapChanges) AccumulateMapChanges(adds, deletes []identity.NumericIdent
 		Nexthdr:          proto,
 		TrafficDirection: direction.Uint8(),
 	}
-	var value MapStateEntry // defaults to ProxyPort 0 == no redirect
-	if redirect {
-		value = redirectEntry // Special entry to mark the need for redirection
-	}
+
+	value := NewMapStateEntry(cs, redirect)
 
 	if option.Config.Debug {
 		log.WithFields(logrus.Fields{
+			logfields.EndpointSelector: cs,
 			logfields.AddedPolicyID:    adds,
 			logfields.DeletedPolicyID:  deletes,
 			logfields.Port:             port,
@@ -193,45 +304,37 @@ func (mc *MapChanges) AccumulateMapChanges(adds, deletes []identity.NumericIdent
 	}
 
 	mc.mutex.Lock()
-	if len(adds) > 0 {
-		if mc.adds == nil {
-			mc.adds = make(MapState)
-		}
-		for _, id := range adds {
-			key.Identity = id.Uint32()
-			// insert but do not allow NoRedirectEntry to overwrite a redirect entry
-			mc.adds.RedirectPreferredInsert(key, value)
-
-			// Remove a potential previously deleted key
-			if mc.deletes != nil {
-				delete(mc.deletes, key)
-			}
-		}
+	for _, id := range adds {
+		key.Identity = id.Uint32()
+		mc.changes = append(mc.changes, MapChange{true, key, value})
 	}
-	if len(deletes) > 0 {
-		if mc.deletes == nil {
-			mc.deletes = make(MapState)
-		}
-		for _, id := range deletes {
-			key.Identity = id.Uint32()
-			mc.deletes[key] = value
-			// Remove a potential previously added key
-			if mc.adds != nil {
-				delete(mc.adds, key)
-			}
-		}
+	for _, id := range deletes {
+		key.Identity = id.Uint32()
+		mc.changes = append(mc.changes, MapChange{false, key, value})
 	}
 	mc.mutex.Unlock()
 }
 
-// consumeMapChanges transfers the changes from MapChanges to the caller.
-// May return nil maps.
-func (mc *MapChanges) consumeMapChanges() (adds, deletes MapState) {
+// consumeMapChanges transfers the incremental changes from MapChanges to the caller,
+// while applying the changes to PolicyMapState.
+func (mc *MapChanges) consumeMapChanges(policyMapState MapState) (adds, deletes MapState) {
 	mc.mutex.Lock()
-	adds = mc.adds
-	mc.adds = nil
-	deletes = mc.deletes
-	mc.deletes = nil
+	adds = make(MapState, len(mc.changes))
+	deletes = make(MapState, len(mc.changes))
+
+	for i := range mc.changes {
+		if mc.changes[i].add {
+			// Insert but do not allow non-redirect entries to overwrite a redirect entry.
+			// Collect the incremental changes to the overall state in 'adds' and 'deletes'.
+			policyMapState.redirectPreferredInsertWithChanges(mc.changes[i].key, mc.changes[i].value, adds, deletes)
+		} else {
+			// Delete the contribution of this cs to the key and collect incremental changes
+			for cs := range mc.changes[i].value.selectors { // get the sole selector
+				policyMapState.deleteKeyWithChanges(mc.changes[i].key, cs, adds, deletes)
+			}
+		}
+	}
+	mc.changes = nil
 	mc.mutex.Unlock()
 	return adds, deletes
 }
diff --git a/pkg/policy/resolve.go b/pkg/policy/resolve.go
index f6d33a59da..5eb828767f 100644
--- a/pkg/policy/resolve.go
+++ b/pkg/policy/resolve.go
@@ -64,6 +64,7 @@ type EndpointPolicy struct {
 	// It maps each Key to the proxy port if proxy redirection is needed.
 	// Proxy port 0 indicates no proxy redirection.
 	// All fields within the Key and the proxy port must be in host byte-order.
+	// Must only be accessed with PolicyOwner (aka Endpoint) lock taken.
 	PolicyMapState MapState
 
 	// policyMapChanges collects pending changes to the PolicyMapState
@@ -157,13 +158,13 @@ func (p *EndpointPolicy) computeDirectionL4PolicyMapEntries(l4PolicyMap L4Policy
 		keysFromFilter := filter.ToMapState(direction)
 		for keyFromFilter, entry := range keysFromFilter {
 			// Fix up the proxy port for entries that need proxy redirection
-			if entry != NoRedirectEntry {
+			if entry.IsRedirectEntry() {
 				entry.ProxyPort = p.PolicyOwner.LookupRedirectPort(filter)
 				// If the currently allocated proxy port is 0, this is a new
 				// redirect, for which no port has been allocated yet. Ignore
 				// it for now. This will be configured by
 				// e.addNewRedirectsFromDesiredPolicy() once the port has been allocated.
-				if entry == NoRedirectEntry {
+				if !entry.IsRedirectEntry() {
 					continue
 				}
 			}
@@ -175,10 +176,11 @@ func (p *EndpointPolicy) computeDirectionL4PolicyMapEntries(l4PolicyMap L4Policy
 // ConsumeMapChanges transfers the changes from MapChanges to the caller,
 // locking the selector cache to make sure concurrent identity updates
 // have completed.
+// PolicyOwner (aka Endpoint) is also locked during this call.
 func (p *EndpointPolicy) ConsumeMapChanges() (adds, deletes MapState) {
 	p.selectorPolicy.SelectorCache.mutex.Lock()
 	defer p.selectorPolicy.SelectorCache.mutex.Unlock()
-	return p.policyMapChanges.consumeMapChanges()
+	return p.policyMapChanges.consumeMapChanges(p.PolicyMapState)
 }
 
 // NewEndpointPolicy returns an empty EndpointPolicy stub.


Test patch:
diff --git a/pkg/policy/distillery_test.go b/pkg/policy/distillery_test.go
index 54965a6b38..fdd8573cd5 100644
--- a/pkg/policy/distillery_test.go
+++ b/pkg/policy/distillery_test.go
@@ -246,8 +246,8 @@ var (
 	mapKeyAllow___L4 = Key{0, 80, 6, dirIngress}
 	mapKeyAllowAll__ = Key{0, 0, 0, dirIngress}
 	// Desired map entries for no L7 redirect / redirect to Proxy
-	mapEntryL7None_ = NoRedirectEntry
-	mapEntryL7Proxy = redirectEntry
+	mapEntryL7None_ = NewMapStateEntry(nil, false)
+	mapEntryL7Proxy = NewMapStateEntry(nil, true)
 )
 
 // combineL4L7 returns a new PortRule that refers to the specified l4 ports and
@@ -296,6 +296,7 @@ func (d *policyDistillery) distillPolicy(epLabels labels.LabelArray) (MapState,
 		allowAllIngress := true
 		allowAllEgress := false // Skip egress
 		result.AllowAllIdentities(allowAllIngress, allowAllEgress)
+		result.clearCachedSelectors()
 		return result, nil
 	}
 
@@ -316,14 +317,29 @@ func (d *policyDistillery) distillPolicy(epLabels labels.LabelArray) (MapState,
 	for _, l4 := range l4IngressPolicy {
 		io.WriteString(d.log, fmt.Sprintf("[distill] Processing L4Filter (l4: %d/%s), (l3/7: %+v)\n", l4.Port, l4.Protocol, l4.L7RulesPerSelector))
 		for key, entry := range l4.ToMapState(0) {
-			io.WriteString(d.log, fmt.Sprintf("[distill] L4 ingress allow %+v (parser=%s, redirect=%t)\n", key, l4.L7Parser, entry != NoRedirectEntry))
+			io.WriteString(d.log, fmt.Sprintf("[distill] L4 ingress allow %+v (parser=%s, redirect=%t)\n", key, l4.L7Parser, entry.IsRedirectEntry()))
 			result[key] = entry
 		}
 	}
 	l4IngressPolicy.Detach(d.Repository.GetSelectorCache())
+	result.clearCachedSelectors()
 	return result, nil
 }
 
+// clearCachedSelectors removes CachedSelectors from MapStateEntries
+// for testing purposes.  Table-driven testing pattern used for these
+// tests does not allow expected MapStateEntries to contain actual
+// CachedSelectors as those have not been inserted to the selector
+// cache at the time when the expectations are created.
+func (m MapState) clearCachedSelectors() {
+	for k, v := range m {
+		v.selectors = nilSelectors
+		m[k] = v
+	}
+}
+
+var nilSelectors = map[CachedSelector]struct{}{nil: {}}
+
 func Test_MergeL3(t *testing.T) {
 	identityCache := cache.IdentityCache{
 		identity.NumericIdentity(identityFoo): labelsFoo,
diff --git a/pkg/policy/mapstate_test.go b/pkg/policy/mapstate_test.go
index 7af78d10d0..98eda14c1d 100644
--- a/pkg/policy/mapstate_test.go
+++ b/pkg/policy/mapstate_test.go
@@ -17,11 +17,33 @@
 package policy
 
 import (
+	"github.com/cilium/cilium/pkg/checker"
 	"github.com/cilium/cilium/pkg/policy/trafficdirection"
 
 	"gopkg.in/check.v1"
 )
 
+// WithSelectors returns a copy of 'e', but selectors replaced with 'selectors'. 'e' is not modified.
+// No selectors is represented with a 'nil' map.
+func (e MapStateEntry) WithSelectors(selectors ...CachedSelector) MapStateEntry {
+	mse := e
+	if len(selectors) > 0 {
+		mse.selectors = make(map[CachedSelector]struct{}, len(selectors))
+		for _, cs := range selectors {
+			mse.selectors[cs] = struct{}{}
+		}
+	} else {
+		mse.selectors = nil
+	}
+	return mse
+}
+
+// WithoutSelectors returns a copy of 'e', but selectors replaced with 'nil'. 'e' is not modified.
+// Note: This is used only in unit tests and helps test readability.
+func (e MapStateEntry) WithoutSelectors() MapStateEntry {
+	return e.WithSelectors()
+}
+
 func (ds *PolicyTestSuite) TestPolicyKeyTrafficDirection(c *check.C) {
 	k := Key{TrafficDirection: trafficdirection.Ingress.Uint8()}
 	c.Assert(k.IsIngress(), check.Equals, true)
@@ -31,3 +53,247 @@ func (ds *PolicyTestSuite) TestPolicyKeyTrafficDirection(c *check.C) {
 	c.Assert(k.IsIngress(), check.Equals, false)
 	c.Assert(k.IsEgress(), check.Equals, true)
 }
+
+func (ds *PolicyTestSuite) TestMapState_AccumulateMapChanges(c *check.C) {
+	csFoo := newTestCachedSelector("Foo", false)
+	csBar := newTestCachedSelector("Bar", false)
+
+	TestKey := func(id int, port uint16, proto uint8, direction trafficdirection.TrafficDirection) Key {
+		return Key{
+			Identity:         uint32(id),
+			DestPort:         port,
+			Nexthdr:          proto,
+			TrafficDirection: direction.Uint8(),
+		}
+	}
+	TestIngressKey := func(id int, port uint16, proto uint8) Key {
+		return TestKey(id, port, proto, trafficdirection.Ingress)
+	}
+	TestEgressKey := func(id int, port uint16, proto uint8) Key {
+		return TestKey(id, port, proto, trafficdirection.Egress)
+	}
+	DNSUDPEgressKey := func(id int) Key {
+		return TestEgressKey(id, 53, 17)
+	}
+	DNSTCPEgressKey := func(id int) Key {
+		return TestEgressKey(id, 53, 6)
+	}
+	HostIngressKey := func() Key {
+		return TestIngressKey(1, 0, 0)
+	}
+	AnyIngressKey := func() Key {
+		return TestIngressKey(0, 0, 0)
+	}
+	//AnyEgressKey := func() Key {
+	//	return TestEgressKey(0, 0, 0)
+	//}
+	HttpIngressKey := func(id int) Key {
+		return TestIngressKey(id, 80, 6)
+	}
+	HttpEgressKey := func(id int) Key {
+		return TestEgressKey(id, 80, 6)
+	}
+
+	TestEntry := func(proxyPort uint16, selectors ...CachedSelector) MapStateEntry {
+		entry := MapStateEntry{
+			ProxyPort: proxyPort,
+		}
+		if len(selectors) > 0 {
+			entry.selectors = make(map[CachedSelector]struct{}, len(selectors))
+			for _, cs := range selectors {
+				entry.selectors[cs] = struct{}{}
+			}
+		}
+		return entry
+	}
+
+	type args struct {
+		cs       *testCachedSelector
+		adds     []int
+		deletes  []int
+		port     uint16
+		proto    uint8
+		ingress  bool
+		redirect bool
+	}
+	tests := []struct {
+		continued bool // Start from the end state of the previous test
+		name      string
+		args      []args // changes applied, in order
+		state     MapState
+		adds      MapState
+		deletes   MapState
+	}{{
+		name: "test-1 - Adding identity to an empty state",
+		args: []args{
+			{cs: csFoo, adds: []int{42}, deletes: []int{}, port: 80, proto: 6, ingress: true, redirect: false},
+		},
+		state: MapState{
+			HttpIngressKey(42): TestEntry(0, csFoo),
+		},
+		adds: MapState{
+			HttpIngressKey(42): TestEntry(0),
+		},
+		deletes: MapState{},
+	}, {
+		continued: true,
+		name:      "test-2 - Removing the sole key",
+		args: []args{
+			{cs: csFoo, adds: nil, deletes: []int{42}, port: 80, proto: 6, ingress: true, redirect: false},
+		},
+		state: MapState{},
+		adds:  MapState{},
+		deletes: MapState{
+			HttpIngressKey(42): TestEntry(0),
+		},
+	}, {
+		name: "test-3 - Adding 2 identities, and deleting a nonexisting key on an empty state",
+		args: []args{
+			{cs: csFoo, adds: []int{42, 43}, deletes: []int{50}, port: 80, proto: 6, ingress: true, redirect: false},
+		},
+		state: MapState{
+			HttpIngressKey(42): TestEntry(0, csFoo),
+			HttpIngressKey(43): TestEntry(0, csFoo),
+		},
+		adds: MapState{
+			HttpIngressKey(42): TestEntry(0),
+			HttpIngressKey(43): TestEntry(0),
+		},
+		deletes: MapState{},
+	}, {
+		continued: true,
+		name:      "test-4 - Adding Bar also selecting 42",
+		args: []args{
+			{cs: csBar, adds: []int{42, 44}, deletes: []int{50}, port: 80, proto: 6, ingress: true, redirect: false},
+		},
+		state: MapState{
+			HttpIngressKey(42): TestEntry(0, csFoo, csBar),
+			HttpIngressKey(43): TestEntry(0, csFoo),
+			HttpIngressKey(44): TestEntry(0, csBar),
+		},
+		adds: MapState{
+			HttpIngressKey(44): TestEntry(0),
+		},
+		deletes: MapState{},
+	}, {
+		continued: true,
+		name:      "test-5 - Deleting 42 from Foo, remains on Bar and no deletes",
+		args: []args{
+			{cs: csFoo, adds: []int{}, deletes: []int{42}, port: 80, proto: 6, ingress: true, redirect: false},
+		},
+		state: MapState{
+			HttpIngressKey(42): TestEntry(0, csBar),
+			HttpIngressKey(43): TestEntry(0, csFoo),
+			HttpIngressKey(44): TestEntry(0, csBar),
+		},
+		adds:    MapState{},
+		deletes: MapState{},
+	}, {
+		continued: true,
+		name:      "test-6 - Deleting 42 from Bar, deleted",
+		args: []args{
+			{cs: csFoo, adds: []int{}, deletes: []int{42}, port: 80, proto: 6, ingress: true, redirect: false},
+		},
+		state: MapState{
+			HttpIngressKey(43): TestEntry(0, csFoo),
+			HttpIngressKey(44): TestEntry(0, csBar),
+		},
+		adds: MapState{},
+		deletes: MapState{
+			HttpIngressKey(42): TestEntry(0),
+		},
+	}, {
+		continued: true,
+		name:      "test-6b - Adding an entry that already exists, no adds",
+		args: []args{
+			{cs: csBar, adds: []int{44}, deletes: []int{}, port: 80, proto: 6, ingress: true, redirect: false},
+		},
+		state: MapState{
+			HttpIngressKey(43): TestEntry(0, csFoo),
+			HttpIngressKey(44): TestEntry(0, csBar),
+		},
+		adds:    MapState{},
+		deletes: MapState{},
+	}, {
+		continued: false,
+		name:      "test-7a - egress HTTP proxy (setup)",
+		args: []args{
+			{cs: nil, adds: []int{0}, deletes: []int{}, port: 0, proto: 0, ingress: true, redirect: false},
+			{cs: nil, adds: []int{1}, deletes: []int{}, port: 0, proto: 0, ingress: true, redirect: false},
+			{cs: csBar, adds: []int{42}, deletes: []int{}, port: 53, proto: 17, ingress: false, redirect: false},
+			{cs: csBar, adds: []int{42}, deletes: []int{}, port: 53, proto: 6, ingress: false, redirect: false},
+		},
+		state: MapState{
+			AnyIngressKey():     TestEntry(0, nil),
+			HostIngressKey():    TestEntry(0, nil),
+			DNSUDPEgressKey(42): TestEntry(0, csBar),
+			DNSTCPEgressKey(42): TestEntry(0, csBar),
+		},
+		adds: MapState{
+			AnyIngressKey():     TestEntry(0),
+			HostIngressKey():    TestEntry(0),
+			DNSUDPEgressKey(42): TestEntry(0),
+			DNSTCPEgressKey(42): TestEntry(0),
+		},
+		deletes: MapState{},
+	}, {
+		continued: true,
+		name:      "test-7b - egress HTTP proxy (incremental update)",
+		args: []args{
+			{cs: csFoo, adds: []int{43}, deletes: []int{}, port: 80, proto: 6, ingress: false, redirect: true},
+		},
+		state: MapState{
+			AnyIngressKey():     TestEntry(0, nil),
+			HostIngressKey():    TestEntry(0, nil),
+			DNSUDPEgressKey(42): TestEntry(0, csBar),
+			DNSTCPEgressKey(42): TestEntry(0, csBar),
+			HttpEgressKey(43):   TestEntry(1, csFoo),
+		},
+		adds: MapState{
+			HttpEgressKey(43): TestEntry(1),
+		},
+		deletes: MapState{},
+	}, {
+		continued: false,
+		name:      "test-n - title",
+		args:      []args{
+			//{cs: csFoo, adds: []int{42, 43}, deletes: []int{50}, port: 80, proto: 6, ingress: true, redirect: false},
+		},
+		state: MapState{
+			//HttpIngressKey(42): TestEntry(0, csFoo),
+		},
+		adds: MapState{
+			//HttpIngressKey(42): TestEntry(0),
+		},
+		deletes: MapState{
+			//HttpIngressKey(43): TestEntry(0),
+		},
+	},
+	}
+
+	policyMapState := MapState{}
+
+	for _, tt := range tests {
+		policyMaps := MapChanges{}
+		if !tt.continued {
+			policyMapState = MapState{}
+		}
+		for _, x := range tt.args {
+			dir := trafficdirection.Egress
+			if x.ingress {
+				dir = trafficdirection.Ingress
+			}
+			adds := x.cs.addSelections(x.adds...)
+			deletes := x.cs.deleteSelections(x.deletes...)
+			var cs CachedSelector
+			if x.cs != nil {
+				cs = x.cs
+			}
+			policyMaps.AccumulateMapChanges(cs, adds, deletes, x.port, x.proto, dir, x.redirect)
+		}
+		adds, deletes := policyMaps.consumeMapChanges(policyMapState)
+		c.Assert(policyMapState, checker.DeepEquals, tt.state, check.Commentf(tt.name+" (MapState)"))
+		c.Assert(adds, checker.DeepEquals, tt.adds, check.Commentf(tt.name+" (adds)"))
+		c.Assert(deletes, checker.DeepEquals, tt.deletes, check.Commentf(tt.name+" (deletes)"))
+	}
+}
diff --git a/pkg/policy/resolve_test.go b/pkg/policy/resolve_test.go
index ad6a4cf95b..5e07ea58bf 100644
--- a/pkg/policy/resolve_test.go
+++ b/pkg/policy/resolve_test.go
@@ -447,6 +447,9 @@ func (ds *PolicyTestSuite) TestMapStateWithIngressWildcard(c *C) {
 	c.Assert(err, IsNil)
 	policy := selPolicy.DistillPolicy(DummyOwner{})
 
+	rule1MapStateEntry := NewMapStateEntry(wildcardCachedSelector, false)
+	allowEgressMapStateEntry := NewMapStateEntry(nil, false)
+
 	expectedEndpointPolicy := EndpointPolicy{
 		selectorPolicy: &selectorPolicy{
 			Revision:      repo.GetRevision(),
@@ -475,18 +478,17 @@ func (ds *PolicyTestSuite) TestMapStateWithIngressWildcard(c *C) {
 		},
 		PolicyOwner: DummyOwner{},
 		PolicyMapState: MapState{
-			{TrafficDirection: trafficdirection.Egress.Uint8()}: {},
-			{DestPort: 80, Nexthdr: 6}:                          {},
+			{TrafficDirection: trafficdirection.Egress.Uint8()}: allowEgressMapStateEntry,
+			{DestPort: 80, Nexthdr: 6}:                          rule1MapStateEntry,
 		},
 	}
 
-	// Add new identity to test accumulation of PolicyMapChanges
+	// Add new identity to test accumulation of MapChanges
 	added1 := cache.IdentityCache{
 		identity.NumericIdentity(192): labels.ParseSelectLabelArray("id=resolve_test_1"),
 	}
 	testSelectorCache.UpdateIdentities(added1, nil)
-	c.Assert(policy.policyMapChanges.adds, HasLen, 0)
-	c.Assert(policy.policyMapChanges.deletes, HasLen, 0)
+	c.Assert(policy.policyMapChanges.changes, HasLen, 0)
 
 	// Have to remove circular reference before testing to avoid an infinite loop
 	policy.selectorPolicy.Detach()
@@ -543,22 +545,22 @@ func (ds *PolicyTestSuite) TestMapStateWithIngress(c *C) {
 	c.Assert(err, IsNil)
 	policy := selPolicy.DistillPolicy(DummyOwner{})
 
-	// Add new identity to test accumulation of PolicyMapChanges
+	// Add new identity to test accumulation of MapChanges
 	added1 := cache.IdentityCache{
 		identity.NumericIdentity(192): labels.ParseSelectLabelArray("id=resolve_test_1", "num=1"),
 		identity.NumericIdentity(193): labels.ParseSelectLabelArray("id=resolve_test_1", "num=2"),
 		identity.NumericIdentity(194): labels.ParseSelectLabelArray("id=resolve_test_1", "num=3"),
 	}
 	testSelectorCache.UpdateIdentities(added1, nil)
-	c.Assert(policy.policyMapChanges.adds, HasLen, 3)
-	c.Assert(policy.policyMapChanges.deletes, HasLen, 0)
+	// Cleanup the identities from the testSelectorCache
+	defer testSelectorCache.UpdateIdentities(nil, added1)
+	c.Assert(policy.policyMapChanges.changes, HasLen, 3)
 
 	deleted1 := cache.IdentityCache{
 		identity.NumericIdentity(193): labels.ParseSelectLabelArray("id=resolve_test_1", "num=2"),
 	}
 	testSelectorCache.UpdateIdentities(nil, deleted1)
-	c.Assert(policy.policyMapChanges.adds, HasLen, 2)
-	c.Assert(policy.policyMapChanges.deletes, HasLen, 1)
+	c.Assert(policy.policyMapChanges.changes, HasLen, 4)
 
 	cachedSelectorWorld := testSelectorCache.FindCachedIdentitySelector(api.ReservedEndpointSelectors[labels.IDNameWorld])
 	c.Assert(cachedSelectorWorld, Not(IsNil))
@@ -566,6 +568,9 @@ func (ds *PolicyTestSuite) TestMapStateWithIngress(c *C) {
 	cachedSelectorTest := testSelectorCache.FindCachedIdentitySelector(api.NewESFromLabels(lblTest))
 	c.Assert(cachedSelectorTest, Not(IsNil))
 
+	rule1MapStateEntry := NewMapStateEntry(cachedSelectorTest, false)
+	allowEgressMapStateEntry := NewMapStateEntry(nil, false)
+
 	expectedEndpointPolicy := EndpointPolicy{
 		selectorPolicy: &selectorPolicy{
 			Revision:      repo.GetRevision(),
@@ -594,17 +599,10 @@ func (ds *PolicyTestSuite) TestMapStateWithIngress(c *C) {
 		},
 		PolicyOwner: DummyOwner{},
 		PolicyMapState: MapState{
-			{TrafficDirection: trafficdirection.Egress.Uint8()}:                          {},
-			{Identity: uint32(identity.ReservedIdentityWorld), DestPort: 80, Nexthdr: 6}: {},
-		},
-		policyMapChanges: MapChanges{
-			adds: MapState{
-				{Identity: 192, DestPort: 80, Nexthdr: 6}: {},
-				{Identity: 194, DestPort: 80, Nexthdr: 6}: {},
-			},
-			deletes: MapState{
-				{Identity: 193, DestPort: 80, Nexthdr: 6}: {},
-			},
+			{TrafficDirection: trafficdirection.Egress.Uint8()}:                          allowEgressMapStateEntry,
+			{Identity: uint32(identity.ReservedIdentityWorld), DestPort: 80, Nexthdr: 6}: rule1MapStateEntry.WithSelectors(cachedSelectorWorld),
+			{Identity: 192, DestPort: 80, Nexthdr: 6}:                                    rule1MapStateEntry,
+			{Identity: 194, DestPort: 80, Nexthdr: 6}:                                    rule1MapStateEntry,
 		},
 	}
 
@@ -615,18 +613,17 @@ func (ds *PolicyTestSuite) TestMapStateWithIngress(c *C) {
 	cachedSelectorTest = testSelectorCache.FindCachedIdentitySelector(api.NewESFromLabels(lblTest))
 	c.Assert(cachedSelectorTest, IsNil)
 
-	c.Assert(policy, checker.Equals, &expectedEndpointPolicy)
-
 	adds, deletes := policy.ConsumeMapChanges()
 	// maps on the policy got cleared
-	c.Assert(policy.policyMapChanges.adds, IsNil)
-	c.Assert(policy.policyMapChanges.deletes, IsNil)
+	c.Assert(policy.policyMapChanges.changes, IsNil)
 
 	c.Assert(adds, checker.Equals, MapState{
-		{Identity: 192, DestPort: 80, Nexthdr: 6}: {},
-		{Identity: 194, DestPort: 80, Nexthdr: 6}: {},
+		{Identity: 192, DestPort: 80, Nexthdr: 6}: rule1MapStateEntry.WithoutSelectors(),
+		{Identity: 194, DestPort: 80, Nexthdr: 6}: rule1MapStateEntry.WithoutSelectors(),
 	})
 	c.Assert(deletes, checker.Equals, MapState{
-		{Identity: 193, DestPort: 80, Nexthdr: 6}: {},
+		{Identity: 193, DestPort: 80, Nexthdr: 6}: rule1MapStateEntry.WithoutSelectors(),
 	})
+
+	c.Assert(policy, checker.Equals, &expectedEndpointPolicy)
 }
diff --git a/pkg/policy/selectorcache_test.go b/pkg/policy/selectorcache_test.go
index 7ed6c40a29..4530120cf8 100644
--- a/pkg/policy/selectorcache_test.go
+++ b/pkg/policy/selectorcache_test.go
@@ -123,6 +123,83 @@ func (csu *cachedSelectionUser) IdentitySelectionUpdated(selector CachedSelector
 	csu.selections[selector] = selections
 }
 
+// Mock CachedSelector for unit testing.
+
+type testCachedSelector struct {
+	name       string
+	wildcard   bool
+	selections []identity.NumericIdentity
+}
+
+func newTestCachedSelector(name string, wildcard bool, selections ...int) *testCachedSelector {
+	cs := &testCachedSelector{
+		name:       name,
+		wildcard:   wildcard,
+		selections: make([]identity.NumericIdentity, 0, len(selections)),
+	}
+	cs.addSelections(selections...)
+	return cs
+}
+
+// returns selections as []identity.NumericIdentity
+func (cs *testCachedSelector) addSelections(selections ...int) (adds []identity.NumericIdentity) {
+	for _, id := range selections {
+		nid := identity.NumericIdentity(id)
+		adds = append(adds, nid)
+		if cs == nil {
+			continue
+		}
+		if !cs.Selects(nid) {
+			cs.selections = append(cs.selections, nid)
+		}
+	}
+	return adds
+}
+
+// returns selections as []identity.NumericIdentity
+func (cs *testCachedSelector) deleteSelections(selections ...int) (deletes []identity.NumericIdentity) {
+	for _, id := range selections {
+		nid := identity.NumericIdentity(id)
+		deletes = append(deletes, nid)
+		if cs == nil {
+			continue
+		}
+		for i := 0; i < len(cs.selections); i++ {
+			if nid == cs.selections[i] {
+				cs.selections = append(cs.selections[:i], cs.selections[i+1:]...)
+				i--
+			}
+		}
+	}
+	return deletes
+}
+
+// CachedSelector interface
+
+func (cs *testCachedSelector) GetSelections() []identity.NumericIdentity {
+	return cs.selections
+}
+func (cs *testCachedSelector) Selects(nid identity.NumericIdentity) bool {
+	for _, id := range cs.selections {
+		if id == nid {
+			return true
+		}
+	}
+	return false
+}
+
+func (cs *testCachedSelector) IsWildcard() bool {
+	return cs.wildcard
+}
+
+func (cs *testCachedSelector) IsNone() bool {
+	return false
+}
+
+func (cs *testCachedSelector) String() string {
+	return cs.name
+}
+
 func (ds *SelectorCacheTestSuite) SetUpTest(c *C) {
 }