Add pod_owner label to metrics when -spire-agent-address is set

cmd/nginx-ingress/main.go

@@ -500,7 +500,7 @@ func main() {
 		StubStatusOverUnixSocketForOSS: *enablePrometheusMetrics,
 		TLSPassthrough:                 *enableTLSPassthrough,
 		EnableSnippets:                 *enableSnippets,
-		SpiffeCerts:                    *spireAgentAddress != "",
+		NginxServiceMesh:               *spireAgentAddress != "",
 		MainAppProtectLoadModule:       *appProtect,
 		EnableLatencyMetrics:           *enableLatencyMetrics,
 	}
@@ -548,6 +548,9 @@ func main() {
 	if *enablePrometheusMetrics {
 		upstreamServerVariableLabels := []string{"service", "resource_type", "resource_name", "resource_namespace"}
 		upstreamServerPeerVariableLabelNames := []string{"pod_name"}
+		if staticCfgParams.NginxServiceMesh {
+			upstreamServerPeerVariableLabelNames = append(upstreamServerPeerVariableLabelNames, "pod_owner")
+		}
 		if *nginxPlus {
 			serverZoneVariableLabels := []string{"resource_type", "resource_name", "resource_namespace"}
 			variableLabelNames := nginxCollector.NewVariableLabelNames(upstreamServerVariableLabels, serverZoneVariableLabels, upstreamServerPeerVariableLabelNames)

internal/configs/config_params.go

@@ -109,7 +109,7 @@ type StaticConfigParams struct {
 	StubStatusOverUnixSocketForOSS bool
 	TLSPassthrough                 bool
 	EnableSnippets                 bool
-	SpiffeCerts                    bool
+	NginxServiceMesh               bool
 	EnableInternalRoutes           bool
 	MainAppProtectLoadModule       bool
 	PodName                        string

internal/configs/configurator.go

@@ -164,9 +164,13 @@ func (cnf *Configurator) updateIngressMetricsLabels(ingEx *IngressEx, upstreams
 		newUpstreamsNames = append(newUpstreamsNames, u.Name)
 		for _, server := range u.UpstreamServers {
 			s := fmt.Sprintf("%v:%v", server.Address, server.Port)
-			podName := ingEx.PodsByIP[s]
+			podInfo := ingEx.PodsByIP[s]
 			labelKey := fmt.Sprintf("%v/%v", u.Name, s)
-			upstreamServerPeerLabels[labelKey] = []string{podName}
+			upstreamServerPeerLabels[labelKey] = []string{podInfo.Name}
+			if cnf.staticCfgParams.NginxServiceMesh {
+				ownerLabelVal := fmt.Sprintf("%s/%s", podInfo.OwnerType, podInfo.OwnerName)
+				upstreamServerPeerLabels[labelKey] = append(upstreamServerPeerLabels[labelKey], ownerLabelVal)
+			}
 			newPeers[labelKey] = true
 			newPeersIPs = append(newPeersIPs, labelKey)
 		}
@@ -314,9 +318,13 @@ func (cnf *Configurator) updateVirtualServerMetricsLabels(virtualServerEx *Virtu
 		newUpstreams[u.Name] = true
 		newUpstreamsNames = append(newUpstreamsNames, u.Name)
 		for _, server := range u.Servers {
-			podName := virtualServerEx.PodsByIP[server.Address]
+			podInfo := virtualServerEx.PodsByIP[server.Address]
 			labelKey := fmt.Sprintf("%v/%v", u.Name, server.Address)
-			upstreamServerPeerLabels[labelKey] = []string{podName}
+			upstreamServerPeerLabels[labelKey] = []string{podInfo.Name}
+			if cnf.staticCfgParams.NginxServiceMesh {
+				ownerLabelVal := fmt.Sprintf("%s/%s", podInfo.OwnerType, podInfo.OwnerName)
+				upstreamServerPeerLabels[labelKey] = append(upstreamServerPeerLabels[labelKey], ownerLabelVal)
+			}
 			newPeers[labelKey] = true
 			newPeersIPs = append(newPeersIPs, labelKey)
 		}

internal/configs/configurator_test.go

@@ -640,9 +640,9 @@ func TestUpdateIngressMetricsLabels(t *testing.T) {
 				},
 			},
 		},
-		PodsByIP: map[string]string{
-			"10.0.0.1:80": "pod-1",
-			"10.0.0.2:80": "pod-2",
+		PodsByIP: map[string]PodInfo{
+			"10.0.0.1:80": {Name: "pod-1"},
+			"10.0.0.2:80": {Name: "pod-2"},
 		},
 	}
 
@@ -800,9 +800,9 @@ func TestUpdateVirtualServerMetricsLabels(t *testing.T) {
 				Host: "example.com",
 			},
 		},
-		PodsByIP: map[string]string{
-			"10.0.0.1:80": "pod-1",
-			"10.0.0.2:80": "pod-2",
+		PodsByIP: map[string]PodInfo{
+			"10.0.0.1:80": {Name: "pod-1"},
+			"10.0.0.2:80": {Name: "pod-2"},
 		},
 	}
 

internal/configs/ingress.go

@@ -26,7 +26,7 @@ type IngressEx struct {
 	Endpoints         map[string][]string
 	HealthChecks      map[string]*api_v1.Probe
 	ExternalNameSvcs  map[string]bool
-	PodsByIP          map[string]string
+	PodsByIP          map[string]PodInfo
 	AppProtectPolicy  *unstructured.Unstructured
 	AppProtectLogConf *unstructured.Unstructured
 	AppProtectLogDst  string
@@ -250,7 +250,7 @@ func generateNginxCfg(ingEx *IngressEx, pems map[string]string, apResources map[
 			Namespace:   ingEx.Ingress.Namespace,
 			Annotations: ingEx.Ingress.Annotations,
 		},
-		SpiffeClientCerts: staticParams.SpiffeCerts && !cfgParams.SpiffeServerCerts,
+		SpiffeClientCerts: staticParams.NginxServiceMesh && !cfgParams.SpiffeServerCerts,
 	}
 }
 
@@ -478,10 +478,10 @@ func generateNginxCfgForMergeableIngresses(mergeableIngs *MergeableIngresses, ma
 		Upstreams:         upstreams,
 		Keepalive:         keepalive,
 		Ingress:           masterNginxCfg.Ingress,
-		SpiffeClientCerts: staticParams.SpiffeCerts && !baseCfgParams.SpiffeServerCerts,
+		SpiffeClientCerts: staticParams.NginxServiceMesh && !baseCfgParams.SpiffeServerCerts,
 	}
 }
 
 func isSSLEnabled(isSSLService bool, cfgParams ConfigParams, staticCfgParams *StaticConfigParams) bool {
-	return isSSLService || staticCfgParams.SpiffeCerts && !cfgParams.SpiffeServerCerts
+	return isSSLService || staticCfgParams.NginxServiceMesh && !cfgParams.SpiffeServerCerts
 }

internal/configs/ingress_test.go

@@ -766,7 +766,7 @@ func TestGenerateNginxCfgForSpiffe(t *testing.T) {
 	}
 
 	apResources := make(map[string]string)
-	result := generateNginxCfg(&cafeIngressEx, pems, apResources, false, configParams, false, false, "", &StaticConfigParams{SpiffeCerts: true})
+	result := generateNginxCfg(&cafeIngressEx, pems, apResources, false, configParams, false, false, "", &StaticConfigParams{NginxServiceMesh: true})
 
 	if !reflect.DeepEqual(result, expected) {
 		t.Errorf("generateNginxCfg returned \n%v,  but expected \n%v", result, expected)
@@ -788,7 +788,7 @@ func TestGenerateNginxCfgForInternalRoute(t *testing.T) {
 	}
 
 	apResources := make(map[string]string)
-	result := generateNginxCfg(&cafeIngressEx, pems, apResources, false, configParams, false, false, "", &StaticConfigParams{SpiffeCerts: true, EnableInternalRoutes: true})
+	result := generateNginxCfg(&cafeIngressEx, pems, apResources, false, configParams, false, false, "", &StaticConfigParams{NginxServiceMesh: true, EnableInternalRoutes: true})
 
 	if !reflect.DeepEqual(result, expected) {
 		t.Errorf("generateNginxCfg returned \n%+v,  but expected \n%+v", result, expected)
@@ -799,61 +799,61 @@ func TestIsSSLEnabled(t *testing.T) {
 	type testCase struct {
 		IsSSLService,
 		SpiffeServerCerts,
-		SpiffeCerts,
+		NginxServiceMesh,
 		Expected bool
 	}
 	var testCases = []testCase{
 		{
 			IsSSLService:      false,
 			SpiffeServerCerts: false,
-			SpiffeCerts:       false,
+			NginxServiceMesh:  false,
 			Expected:          false,
 		},
 		{
 			IsSSLService:      false,
 			SpiffeServerCerts: true,
-			SpiffeCerts:       true,
+			NginxServiceMesh:  true,
 			Expected:          false,
 		},
 		{
 			IsSSLService:      false,
 			SpiffeServerCerts: false,
-			SpiffeCerts:       true,
+			NginxServiceMesh:  true,
 			Expected:          true,
 		},
 		{
 			IsSSLService:      false,
 			SpiffeServerCerts: true,
-			SpiffeCerts:       false,
+			NginxServiceMesh:  false,
 			Expected:          false,
 		},
 		{
 			IsSSLService:      true,
 			SpiffeServerCerts: true,
-			SpiffeCerts:       true,
+			NginxServiceMesh:  true,
 			Expected:          true,
 		},
 		{
 			IsSSLService:      true,
 			SpiffeServerCerts: false,
-			SpiffeCerts:       true,
+			NginxServiceMesh:  true,
 			Expected:          true,
 		},
 		{
 			IsSSLService:      true,
 			SpiffeServerCerts: true,
-			SpiffeCerts:       false,
+			NginxServiceMesh:  false,
 			Expected:          true,
 		},
 		{
 			IsSSLService:      true,
 			SpiffeServerCerts: false,
-			SpiffeCerts:       false,
+			NginxServiceMesh:  false,
 			Expected:          true,
 		},
 	}
 	for i, tc := range testCases {
-		actual := isSSLEnabled(tc.IsSSLService, ConfigParams{SpiffeServerCerts: tc.SpiffeServerCerts}, &StaticConfigParams{SpiffeCerts: tc.SpiffeCerts})
+		actual := isSSLEnabled(tc.IsSSLService, ConfigParams{SpiffeServerCerts: tc.SpiffeServerCerts}, &StaticConfigParams{NginxServiceMesh: tc.NginxServiceMesh})
 		if actual != tc.Expected {
 			t.Errorf("isSSLEnabled returned %v but expected %v for the case %v", actual, tc.Expected, i)
 		}

internal/configs/virtualserver.go

@@ -29,6 +29,22 @@ var incompatibleLBMethodsForSlowStart = map[string]bool{
 	"random two least_time=last_byte": true,
 }
 
+// MeshPodOwner contains the type and name of the K8s resource that owns the pod.
+// This owner information is needed for NGINX Service Mesh metrics.
+type MeshPodOwner struct {
+	// OwnerType is one of the following: statefulset, daemonset, deployment.
+	OwnerType string
+	// OwnerName is the name of the statefulset, daemonset, or deployment.
+	OwnerName string
+}
+
+// PodInfo contains the name of the Pod and the MeshPodOwner information
+// which is used for NGINX Service Mesh metrics.
+type PodInfo struct {
+	Name string
+	MeshPodOwner
+}
+
 // VirtualServerEx holds a VirtualServer along with the resources that are referenced in this VirtualServer.
 type VirtualServerEx struct {
 	VirtualServer       *conf_v1.VirtualServer
@@ -37,7 +53,7 @@ type VirtualServerEx struct {
 	VirtualServerRoutes []*conf_v1.VirtualServerRoute
 	ExternalNameSvcs    map[string]bool
 	Policies            map[string]*conf_v1alpha1.Policy
-	PodsByIP            map[string]string
+	PodsByIP            map[string]PodInfo
 }
 
 func (vsx *VirtualServerEx) String() string {
@@ -168,7 +184,7 @@ func newVirtualServerConfigurator(cfgParams *ConfigParams, isPlus bool, isResolv
 		isTLSPassthrough:     staticParams.TLSPassthrough,
 		enableSnippets:       staticParams.EnableSnippets,
 		warnings:             make(map[runtime.Object][]string),
-		spiffeCerts:          staticParams.SpiffeCerts,
+		spiffeCerts:          staticParams.NginxServiceMesh,
 	}
 }
 

internal/configs/virtualserver_test.go

@@ -699,7 +699,7 @@ func TestGenerateVirtualServerConfigWithSpiffeCerts(t *testing.T) {
 	isPlus := false
 	isResolverConfigured := false
 	tlsPemFileName := ""
-	staticConfigParams := &StaticConfigParams{TLSPassthrough: true, SpiffeCerts: true}
+	staticConfigParams := &StaticConfigParams{TLSPassthrough: true, NginxServiceMesh: true}
 	vsc := newVirtualServerConfigurator(&baseCfgParams, isPlus, isResolverConfigured, staticConfigParams)
 	result, warnings := vsc.GenerateVirtualServerConfig(&virtualServerEx, tlsPemFileName)
 	if !reflect.DeepEqual(result, expected) {

internal/k8s/controller.go

@@ -89,6 +89,8 @@ var (
 type podEndpoint struct {
 	Address string
 	PodName string
+	// MeshPodOwner is used for NGINX Service Mesh metrics
+	configs.MeshPodOwner
 }
 
 // LoadBalancerController watches Kubernetes API and
@@ -2360,7 +2362,7 @@ func (lbc *LoadBalancerController) createIngress(ing *networking.Ingress) (*conf
 	ingEx.Endpoints = make(map[string][]string)
 	ingEx.HealthChecks = make(map[string]*api_v1.Probe)
 	ingEx.ExternalNameSvcs = make(map[string]bool)
-	ingEx.PodsByIP = make(map[string]string)
+	ingEx.PodsByIP = make(map[string]configs.PodInfo)
 
 	if ing.Spec.Backend != nil {
 		podEndps := []podEndpoint{}
@@ -2393,7 +2395,10 @@ func (lbc *LoadBalancerController) createIngress(ing *networking.Ingress) (*conf
 
 		if lbc.isNginxPlus || lbc.isLatencyMetricsEnabled {
 			for _, endpoint := range podEndps {
-				ingEx.PodsByIP[endpoint.Address] = endpoint.PodName
+				ingEx.PodsByIP[endpoint.Address] = configs.PodInfo{
+					Name:         endpoint.PodName,
+					MeshPodOwner: endpoint.MeshPodOwner,
+				}
 			}
 		}
 	}
@@ -2440,7 +2445,10 @@ func (lbc *LoadBalancerController) createIngress(ing *networking.Ingress) (*conf
 
 			if lbc.isNginxPlus || lbc.isLatencyMetricsEnabled {
 				for _, endpoint := range podEndps {
-					ingEx.PodsByIP[endpoint.Address] = endpoint.PodName
+					ingEx.PodsByIP[endpoint.Address] = configs.PodInfo{
+						Name:         endpoint.PodName,
+						MeshPodOwner: endpoint.MeshPodOwner,
+					}
 				}
 			}
 		}
@@ -2551,7 +2559,7 @@ func (lbc *LoadBalancerController) createVirtualServer(virtualServer *conf_v1.Vi
 
 	endpoints := make(map[string][]string)
 	externalNameSvcs := make(map[string]bool)
-	podsByIP := make(map[string]string)
+	podsByIP := make(map[string]configs.PodInfo)
 
 	for _, u := range virtualServer.Spec.Upstreams {
 		endpointsKey := configs.GenerateEndpointsKey(virtualServer.Namespace, u.Service, u.Subselector, u.Port)
@@ -2579,7 +2587,10 @@ func (lbc *LoadBalancerController) createVirtualServer(virtualServer *conf_v1.Vi
 
 		if lbc.isNginxPlus || lbc.isLatencyMetricsEnabled {
 			for _, endpoint := range podEndps {
-				podsByIP[endpoint.Address] = endpoint.PodName
+				podsByIP[endpoint.Address] = configs.PodInfo{
+					Name:         endpoint.PodName,
+					MeshPodOwner: endpoint.MeshPodOwner,
+				}
 			}
 		}
 	}
@@ -2667,7 +2678,10 @@ func (lbc *LoadBalancerController) createVirtualServer(virtualServer *conf_v1.Vi
 
 			if lbc.isNginxPlus || lbc.isLatencyMetricsEnabled {
 				for _, endpoint := range podEndps {
-					podsByIP[endpoint.Address] = endpoint.PodName
+					podsByIP[endpoint.Address] = configs.PodInfo{
+						Name: endpoint.PodName,
+						MeshPodOwner: endpoint.MeshPodOwner,
+					}
 				}
 			}
 		}
@@ -2831,10 +2845,14 @@ func getEndpointsBySubselectedPods(targetPort int32, pods []*api_v1.Pod, svcEps
 				for _, address := range subset.Addresses {
 					if address.IP == pod.Status.PodIP {
 						addr := fmt.Sprintf("%v:%v", pod.Status.PodIP, targetPort)
-
+						ownerType, ownerName := getPodOwnerTypeAndName(pod)
 						podEnd := podEndpoint{
 							Address: addr,
 							PodName: getPodName(address.TargetRef),
+							MeshPodOwner: configs.MeshPodOwner{
+								OwnerType: ownerType,
+								OwnerName: ownerName,
+							},
 						}
 						endps = append(endps, podEnd)
 					}
@@ -2960,7 +2978,12 @@ func (lbc *LoadBalancerController) getEndpointsForPort(endps api_v1.Endpoints, i
 					addr := fmt.Sprintf("%v:%v", address.IP, port.Port)
 					podEnd := podEndpoint{
 						Address: addr,
-						PodName: address.TargetRef.Name,
+					}
+					if address.TargetRef != nil {
+						parentType, parentName := lbc.getPodOwnerTypeAndNameFromAddress(address.TargetRef.Namespace, address.TargetRef.Name)
+						podEnd.OwnerType = parentType
+						podEnd.OwnerName = parentName
+						podEnd.PodName = address.TargetRef.Name
 					}
 					endpoints = append(endpoints, podEnd)
 				}
@@ -2972,6 +2995,35 @@ func (lbc *LoadBalancerController) getEndpointsForPort(endps api_v1.Endpoints, i
 	return nil, fmt.Errorf("No endpoints for target port %v in service %s", targetPort, svc.Name)
 }
 
+func (lbc *LoadBalancerController) getPodOwnerTypeAndNameFromAddress(ns, name string) (parentType, parentName string) {
+	obj, exists, err := lbc.podLister.GetByKey(fmt.Sprintf("%s/%s", ns, name))
+	if err != nil {
+		glog.Warningf("could not get pod by key %s/%s: %v", ns, name, err)
+		return "", ""
+	}
+	if exists {
+		pod := obj.(*api_v1.Pod)
+		return getPodOwnerTypeAndName(pod)
+	}
+	return "", ""
+}
+
+func getPodOwnerTypeAndName(pod *api_v1.Pod) (parentType, parentName string) {
+	parentType = "deployment"
+	for _, owner := range pod.GetOwnerReferences() {
+		parentName = owner.Name
+		if owner.Controller != nil && *owner.Controller {
+			if owner.Kind == "StatefulSet" || owner.Kind == "DaemonSet" {
+				parentType = strings.ToLower(owner.Kind)
+			}
+			if owner.Kind == "ReplicaSet" && strings.HasSuffix(owner.Name, pod.Labels["pod-template-hash"]) {
+				parentName = strings.TrimSuffix(owner.Name, "-"+pod.Labels["pod-template-hash"])
+			}
+		}
+	}
+	return parentType, parentName
+}
+
 func (lbc *LoadBalancerController) getServicePortForIngressPort(ingSvcPort intstr.IntOrString, svc *api_v1.Service) *api_v1.ServicePort {
 	for _, port := range svc.Spec.Ports {
 		if (ingSvcPort.Type == intstr.Int && port.Port == int32(ingSvcPort.IntValue())) || (ingSvcPort.Type == intstr.String && port.Name == ingSvcPort.String()) {