hubble/parser: Always preserve datapath numeric identity

This introduces a check that we do not overwrite the numeric security
identity provided by the datapath trace point. Only if the datapath did
not provide an identity (i.e. in `FROM_LXC` trace points) do we want to
fall back on the identity from the user-space ip cache or endpoint
manager.

The numeric identity from the datapath can differ from the one we obtain
from user-space (e.g. the endpoint manager or the IP cache), because the
identity could have changed between the time the datapath event was
created and the time the event reaches the Hubble parser. To aid in
troubleshooting, we want to preserve what the datapath observed when it
made the policy decision.

Signed-off-by: Sebastian Wicki <sebastian@isovalent.com>

pkg/hubble/parser/threefour/parser.go

@@ -29,6 +29,7 @@ import (
 	"github.com/cilium/cilium/pkg/hubble/parser/getters"
 	"github.com/cilium/cilium/pkg/identity"
 	"github.com/cilium/cilium/pkg/lock"
+	"github.com/cilium/cilium/pkg/logging/logfields"
 	"github.com/cilium/cilium/pkg/monitor"
 	monitorAPI "github.com/cilium/cilium/pkg/monitor/api"
 
@@ -253,42 +254,71 @@ func sortAndFilterLabels(log logrus.FieldLogger, labels []string, securityIdenti
 	return labels
 }
 
-func (p *Parser) resolveEndpoint(ip net.IP, securityIdentity uint32) *pb.Endpoint {
+func (p *Parser) resolveEndpoint(ip net.IP, datapathSecurityIdentity uint32) *pb.Endpoint {
+	// The datapathSecurityIdentity parameter is the numeric security identity
+	// obtained from the datapath.
+	// The numeric identity from the datapath can differ from the one we obtain
+	// from user-space (e.g. the endpoint manager or the IP cache), because
+	// the identity could have changed between the time the datapath event was
+	// created and the time the event reaches the Hubble parser.
+	// To aid in troubleshooting, we want to preserve what the datapath observed
+	// when it made the policy decision.
+	resolveIdentityConflict := func(identity identity.NumericIdentity) uint32 {
+		// if the datapath did not provide an identity (e.g. FROM_LXC trace
+		// points), use what we have in the user-space cache
+		userspaceSecurityIdentity := uint32(identity)
+		if datapathSecurityIdentity == 0 {
+			return userspaceSecurityIdentity
+		}
+
+		if datapathSecurityIdentity != userspaceSecurityIdentity {
+			p.log.WithFields(logrus.Fields{
+				logfields.Identity:    datapathSecurityIdentity,
+				logfields.OldIdentity: userspaceSecurityIdentity,
+				logfields.IPAddr:      ip,
+			}).Debugf("stale identity observed")
+		}
+
+		return datapathSecurityIdentity
+	}
+
 	// for local endpoints, use the available endpoint information
 	if p.endpointGetter != nil {
 		if ep, ok := p.endpointGetter.GetEndpointInfo(ip); ok {
+			epIdentity := resolveIdentityConflict(ep.GetIdentity())
 			return &pb.Endpoint{
 				ID:        uint32(ep.GetID()),
-				Identity:  uint32(ep.GetIdentity()),
+				Identity:  epIdentity,
 				Namespace: ep.GetK8sNamespace(),
-				Labels:    sortAndFilterLabels(p.log, ep.GetLabels(), securityIdentity),
+				Labels:    sortAndFilterLabels(p.log, ep.GetLabels(), epIdentity),
 				PodName:   ep.GetK8sPodName(),
 			}
 		}
 	}
 
 	// for remote endpoints, assemble the information via ip and identity
+	numericIdentity := datapathSecurityIdentity
 	var namespace, podName string
 	if p.ipGetter != nil {
 		if ipIdentity, ok := p.ipGetter.LookupSecIDByIP(ip); ok {
-			securityIdentity = uint32(ipIdentity.ID)
+			numericIdentity = resolveIdentityConflict(ipIdentity.ID)
 		}
 		if meta := p.ipGetter.GetK8sMetadata(ip); meta != nil {
 			namespace, podName = meta.Namespace, meta.PodName
 		}
 	}
 	var labels []string
 	if p.identityGetter != nil {
-		if id, err := p.identityGetter.GetIdentity(securityIdentity); err != nil {
-			p.log.WithError(err).WithField("identity", securityIdentity).
+		if id, err := p.identityGetter.GetIdentity(numericIdentity); err != nil {
+			p.log.WithError(err).WithField("identity", numericIdentity).
 				Warn("failed to resolve identity")
 		} else {
-			labels = sortAndFilterLabels(p.log, id.Labels, securityIdentity)
+			labels = sortAndFilterLabels(p.log, id.Labels, numericIdentity)
 		}
 	}
 
 	return &pb.Endpoint{
-		Identity:  securityIdentity,
+		Identity:  numericIdentity,
 		Namespace: namespace,
 		Labels:    labels,
 		PodName:   podName,

pkg/hubble/parser/threefour/parser_test.go

@@ -66,6 +66,7 @@ func TestL34Decode(t *testing.T) {
 			if ip.Equal(net.ParseIP("10.16.236.178")) {
 				return &testutils.FakeEndpointInfo{
 					ID:           1234,
+					Identity:     5678,
 					PodName:      "pod-10.16.236.178",
 					PodNamespace: "default",
 				}, true
@@ -97,6 +98,8 @@ func TestL34Decode(t *testing.T) {
 		OnLookupSecIDByIP: func(ip net.IP) (ipcache.Identity, bool) {
 			// pretend IP belongs to a pod on a remote node
 			if ip.String() == "192.168.33.11" {
+				// This numeric identity will be ignored because the above
+				// TraceNotify event already contains the source identity
 				return ipcache.Identity{
 					ID:     1234,
 					Source: source.Unspec,
@@ -137,6 +140,7 @@ func TestL34Decode(t *testing.T) {
 	assert.Equal(t, "remote", f.GetSource().GetNamespace())
 	assert.Equal(t, "service-1234", f.GetSourceService().GetName())
 	assert.Equal(t, "remote", f.GetSourceService().GetNamespace())
+	assert.Equal(t, uint32(1), f.GetSource().GetIdentity())
 
 	assert.Equal(t, []string(nil), f.GetDestinationNames())
 	assert.Equal(t, "10.16.236.178", f.GetIP().GetDestination())
@@ -145,6 +149,7 @@ func TestL34Decode(t *testing.T) {
 	assert.Equal(t, "default", f.GetDestination().GetNamespace())
 	assert.Equal(t, "service-4321", f.GetDestinationService().GetName())
 	assert.Equal(t, "default", f.GetDestinationService().GetNamespace())
+	assert.Equal(t, uint32(5678), f.GetDestination().GetIdentity())
 
 	assert.Equal(t, int32(api.MessageTypeTrace), f.GetEventType().GetType())
 	assert.Equal(t, int32(api.TraceFromHost), f.GetEventType().GetSubType())
@@ -846,7 +851,7 @@ func TestTraceNotifyLocalEndpoint(t *testing.T) {
 
 	v0 := monitor.TraceNotifyV0{
 		Type:     byte(api.MessageTypeTrace),
-		SrcLabel: 456, // overwritten by ep.Identity
+		SrcLabel: 456, // takes precedence over ep.Identity
 		Version:  monitor.TraceNotifyVersion0,
 	}
 
@@ -867,7 +872,7 @@ func TestTraceNotifyLocalEndpoint(t *testing.T) {
 	require.NoError(t, err)
 
 	assert.Equal(t, uint32(ep.ID), f.Source.ID)
-	assert.Equal(t, uint32(ep.Identity), f.Source.Identity)
+	assert.Equal(t, uint32(v0.SrcLabel), f.Source.Identity)
 	assert.Equal(t, ep.PodNamespace, f.Source.Namespace)
 	assert.Equal(t, ep.Labels, f.Source.Labels)
 	assert.Equal(t, ep.PodName, f.Source.PodName)