Kubernetes state_* metricsets are ignoring namespace #6281

CHANGELOG.asciidoc

@@ -119,6 +119,7 @@ https://github.com/elastic/beats/compare/v6.0.0-beta2...master[Check the HEAD di
 - Fix system process metricset for kernel processes. {issue}5700[5700]
 - Fix panic in http dependent modules when invalid config was used.
 - Fix system.filesystem.used.pct value to match what df reports. {issue}5494[5494]
+- Fix namespace disambiguation in Kubernetes state_* metricsets. {issue}6281[6281]
 
 *Packetbeat*
 

metricbeat/module/kubernetes/state_container/data.go

@@ -21,11 +21,15 @@ func eventMapping(families []*dto.MetricFamily) ([]common.MapStr, error) {
 			if container == "" {
 				continue
 			}
-			event, ok := eventsMap[container]
+
+			namespace := util.GetLabel(metric, "namespace")
+			containerKey := namespace + "::" + container
+			event, ok := eventsMap[containerKey]
 			if !ok {
 				event = common.MapStr{}
-				eventsMap[container] = event
+				eventsMap[containerKey] = event
 			}
+
 			switch family.GetName() {
 			case "kube_pod_container_info":
 				event.Put(mb.ModuleDataKey+".pod.name", util.GetLabel(metric, "pod"))
@@ -84,7 +88,8 @@ func eventMapping(families []*dto.MetricFamily) ([]common.MapStr, error) {
 		}
 	}
 
-	var events []common.MapStr
+	// initialize, populate events array from values in eventsMap
+	events := make([]common.MapStr, 0, len(eventsMap))
 	for _, event := range eventsMap {
 		events = append(events, event)
 	}

metricbeat/module/kubernetes/state_container/state_container_test.go

@@ -44,38 +44,91 @@ func TestEventMapping(t *testing.T) {
 	events, err := f.Fetch()
 	assert.NoError(t, err)
 
-	assert.Equal(t, 9, len(events), "Wrong number of returned events")
-
-	testCases := map[string]interface{}{
-		"_module.namespace": "kube-system",
-		"_module.node.name": "minikube",
-		"_module.pod.name":  "kube-dns-v20-5g5cb",
-
-		"image":           "gcr.io/google_containers/exechealthz-amd64:1.2",
-		"status.phase":    "running",
-		"status.ready":    true,
-		"status.restarts": 2,
-
-		"memory.limit.bytes":    52428800,
-		"memory.request.bytes":  52428800,
-		"cpu.request.nanocores": 10000000,
-	}
+	assert.Equal(t, 10, len(events), "Wrong number of returned events")
 
+	testCases := testCases()
 	for _, event := range events {
 		name, err := event.GetValue("name")
-		if err == nil && name == "healthz" {
-			for k, v := range testCases {
-				testValue(t, event, k, v)
+		if err == nil {
+			namespace, err := event.GetValue("_module.namespace")
+			if err == nil {
+				eventKey := namespace.(string) + "@" + name.(string)
+				oneTestCase, oneTestCaseFound := testCases[eventKey]
+				if oneTestCaseFound {
+					for k, v := range oneTestCase {
+						testValue(eventKey, t, event, k, v)
+					}
+					delete(testCases, eventKey)
+				}
 			}
-			return
 		}
 	}
 
-	t.Error("Test reference event not found")
+	if len(testCases) > 0 {
+		t.Errorf("Test reference events not found: %v, \n\ngot: %v", testCases, events)
+	}
 }
 
-func testValue(t *testing.T, event common.MapStr, field string, expected interface{}) {
+func testValue(eventKey string, t *testing.T, event common.MapStr, field string, expected interface{}) {
 	data, err := event.GetValue(field)
-	assert.NoError(t, err, "Could not read field "+field)
-	assert.EqualValues(t, expected, data, "Wrong value for field "+field)
+	assert.NoError(t, err, eventKey+": Could not read field "+field)
+	assert.EqualValues(t, expected, data, eventKey+": Wrong value for field "+field)
+}
+
+// Test cases built to match 3 examples in 'module/kubernetes/_meta/test/kube-state-metrics'.
+// In particular, test same named containers  in different namespaces
+func testCases() map[string]map[string]interface{} {
+	return map[string]map[string]interface{}{
+		"kube-system@kubedns": {
+			"_namespace":        "container",
+			"_module.namespace": "kube-system",
+			"_module.node.name": "minikube",
+			"_module.pod.name":  "kube-dns-v20-5g5cb",
+			"name":              "kubedns",
+			"id":                "docker://fa3d83f648de42492b38fa3e8501d109376f391c50f2bd210c895c8477ae4b62",
+
+			"image":           "gcr.io/google_containers/kubedns-amd64:1.9",
+			"status.phase":    "running",
+			"status.ready":    true,
+			"status.restarts": 2,
+
+			"memory.limit.bytes":    178257920,
+			"memory.request.bytes":  73400320,
+			"cpu.request.nanocores": float64(1e+08),
+		},
+		"test@kubedns": {
+			"_namespace":        "container",
+			"_module.namespace": "test",
+			"_module.node.name": "minikube-test",
+			"_module.pod.name":  "kube-dns-v20-5g5cb-test",
+			"name":              "kubedns",
+			"id":                "docker://fa3d83f648de42492b38fa3e8501d109376f391c50f2bd210c895c8477ae4b62-test",
+
+			"image":           "gcr.io/google_containers/kubedns-amd64:1.9-test",
+			"status.phase":    "terminate",
+			"status.ready":    false,
+			"status.restarts": 3,
+
+			"memory.limit.bytes":    278257920,
+			"memory.request.bytes":  83400320,
+			"cpu.request.nanocores": float64(2e+08),
+		},
+		"kube-system@healthz": {
+			"_namespace":        "container",
+			"_module.namespace": "kube-system",
+			"_module.node.name": "minikube",
+			"_module.pod.name":  "kube-dns-v20-5g5cb",
+			"name":              "healthz",
+			"id":                "docker://52fa55e051dc5b68e44c027588685b7edd85aaa03b07f7216d399249ff4fc821",
+
+			"image":           "gcr.io/google_containers/exechealthz-amd64:1.2",
+			"status.phase":    "running",
+			"status.ready":    true,
+			"status.restarts": 2,
+
+			"memory.limit.bytes":    52428800,
+			"memory.request.bytes":  52428800,
+			"cpu.request.nanocores": float64(1e+07),
+		},
+	}
 }

metricbeat/module/kubernetes/state_deployment/data.go

@@ -21,11 +21,14 @@ func eventMapping(families []*dto.MetricFamily) ([]common.MapStr, error) {
 			if deployment == "" {
 				continue
 			}
-			event, ok := eventsMap[deployment]
+			namespace := util.GetLabel(metric, "namespace")
+			deploymentKey := namespace + "::" + deployment
+			event, ok := eventsMap[deploymentKey]
 			if !ok {
 				event = common.MapStr{}
-				eventsMap[deployment] = event
+				eventsMap[deploymentKey] = event
 			}
+
 			switch family.GetName() {
 			case "kube_deployment_metadata_generation":
 				event.Put(mb.ModuleDataKey+".namespace", util.GetLabel(metric, "namespace"))
@@ -54,7 +57,8 @@ func eventMapping(families []*dto.MetricFamily) ([]common.MapStr, error) {
 		}
 	}
 
-	var events []common.MapStr
+	// initialize, populate events array from values in eventsMap
+	events := make([]common.MapStr, 0, len(eventsMap))
 	for _, event := range eventsMap {
 		events = append(events, event)
 	}

metricbeat/module/kubernetes/state_deployment/state_deployment_test.go

@@ -44,35 +44,76 @@ func TestEventMapping(t *testing.T) {
 	events, err := f.Fetch()
 	assert.NoError(t, err)
 
-	assert.Equal(t, 4, len(events), "Wrong number of returned events")
-
-	testCases := map[string]interface{}{
-		"_module.namespace": "default",
-
-		"name":   "jumpy-owl-redis",
-		"paused": false,
-
-		"replicas.available":   0,
-		"replicas.desired":     1,
-		"replicas.unavailable": 1,
-		"replicas.updated":     1,
-	}
+	assert.Equal(t, 5, len(events), "Wrong number of returned events")
 
+	testCases := testCases()
 	for _, event := range events {
 		name, err := event.GetValue("name")
-		if err == nil && name == "jumpy-owl-redis" {
-			for k, v := range testCases {
-				testValue(t, event, k, v)
+		if err == nil {
+			namespace, err := event.GetValue("_module.namespace")
+			if err == nil {
+				eventKey := namespace.(string) + "@" + name.(string)
+				oneTestCase, oneTestCaseFound := testCases[eventKey]
+				if oneTestCaseFound {
+					for k, v := range oneTestCase {
+						testValue(eventKey, t, event, k, v)
+					}
+					delete(testCases, eventKey)
+				}
 			}
-			return
 		}
 	}
 
-	t.Error("Test reference event not found")
+	if len(testCases) > 0 {
+		t.Errorf("Test reference events not found: %v, \n\ngot: %v", testCases, events)
+	}
 }
 
-func testValue(t *testing.T, event common.MapStr, field string, expected interface{}) {
+func testValue(eventKey string, t *testing.T, event common.MapStr, field string, expected interface{}) {
 	data, err := event.GetValue(field)
-	assert.NoError(t, err, "Could not read field "+field)
-	assert.EqualValues(t, expected, data, "Wrong value for field "+field)
+	assert.NoError(t, err, eventKey+": Could not read field "+field)
+	assert.EqualValues(t, expected, data, eventKey+": Wrong value for field "+field)
+}
+
+// Test cases built to match 3 examples in 'module/kubernetes/_meta/test/kube-state-metrics'.
+// In particular, test same named deployments in different namespaces
+func testCases() map[string]map[string]interface{} {
+	return map[string]map[string]interface{}{
+		"default@jumpy-owl-redis": {
+			"_namespace":        "deployment",
+			"_module.namespace": "default",
+
+			"name":   "jumpy-owl-redis",
+			"paused": false,
+
+			"replicas.available":   0,
+			"replicas.desired":     1,
+			"replicas.unavailable": 1,
+			"replicas.updated":     1,
+		},
+		"test@jumpy-owl-redis": {
+			"_namespace":        "deployment",
+			"_module.namespace": "test",
+
+			"name":   "jumpy-owl-redis",
+			"paused": true,
+
+			"replicas.available":   6,
+			"replicas.desired":     2,
+			"replicas.unavailable": 7,
+			"replicas.updated":     8,
+		},
+		"kube-system@tiller-deploy": {
+			"_namespace":        "deployment",
+			"_module.namespace": "kube-system",
+
+			"name":   "tiller-deploy",
+			"paused": false,
+
+			"replicas.available":   1,
+			"replicas.desired":     1,
+			"replicas.unavailable": 0,
+			"replicas.updated":     1,
+		},
+	}
 }

metricbeat/module/kubernetes/state_node/state_node_test.go

@@ -44,32 +44,67 @@ func TestEventMapping(t *testing.T) {
 	events, err := f.Fetch()
 	assert.NoError(t, err)
 
-	assert.Equal(t, 1, len(events), "Wrong number of returned events")
+	assert.Equal(t, 2, len(events), "Wrong number of returned events")
+
+	testCases := testCases()
+	for _, event := range events {
+		name, err := event.GetValue("name")
+		if err == nil {
+			eventKey := name.(string)
+			oneTestCase, oneTestCaseFound := testCases[eventKey]
+			if oneTestCaseFound {
+				for k, v := range oneTestCase {
+					testValue(eventKey, t, event, k, v)
+				}
+				delete(testCases, eventKey)
+			}
+		}
+	}
 
-	testCases := map[string]interface{}{
-		"_namespace": "node",
-		"name":       "minikube",
+	if len(testCases) > 0 {
+		t.Errorf("Test reference events not found: %v, \n\ngot: %v", testCases, events)
+	}
+}
 
-		"status.ready":         "true",
-		"status.unschedulable": false,
+func testValue(eventKey string, t *testing.T, event common.MapStr, field string, expected interface{}) {
+	data, err := event.GetValue(field)
+	assert.NoError(t, err, eventKey+": Could not read field "+field)
+	assert.EqualValues(t, expected, data, eventKey+": Wrong value for field "+field)
+}
 
-		"cpu.allocatable.cores": 2,
-		"cpu.capacity.cores":    2,
+func testCases() map[string]map[string]interface{} {
+	return map[string]map[string]interface{}{
+		"minikube": {
+			"_namespace": "node",
+			"name":       "minikube",
 
-		"memory.allocatable.bytes": 2097786880,
-		"memory.capacity.bytes":    2097786880,
+			"status.ready":         "true",
+			"status.unschedulable": false,
 
-		"pod.allocatable.total": 110,
-		"pod.capacity.total":    110,
-	}
+			"cpu.allocatable.cores": 2,
+			"cpu.capacity.cores":    2,
 
-	for k, v := range testCases {
-		testValue(t, events[0], k, v)
-	}
-}
+			"memory.allocatable.bytes": 2097786880,
+			"memory.capacity.bytes":    2097786880,
 
-func testValue(t *testing.T, event common.MapStr, field string, expected interface{}) {
-	data, err := event.GetValue(field)
-	assert.NoError(t, err, "Could not read field "+field)
-	assert.EqualValues(t, expected, data, "Wrong value for field "+field)
+			"pod.allocatable.total": 110,
+			"pod.capacity.total":    110,
+		},
+		"minikube-test": {
+			"_namespace": "node",
+			"name":       "minikube-test",
+
+			"status.ready":         "true",
+			"status.unschedulable": true,
+
+			"cpu.allocatable.cores": 3,
+			"cpu.capacity.cores":    4,
+
+			"memory.allocatable.bytes": 3097786880,
+			"memory.capacity.bytes":    4097786880,
+
+			"pod.allocatable.total": 210,
+			"pod.capacity.total":    310,
+		},
+	}
 }

metricbeat/module/kubernetes/state_pod/data.go

@@ -18,11 +18,14 @@ func eventMapping(families []*dto.MetricFamily) ([]common.MapStr, error) {
 			if pod == "" {
 				continue
 			}
-			event, ok := eventsMap[pod]
+			namespace := util.GetLabel(metric, "namespace")
+			podKey := namespace + "::" + pod
+			event, ok := eventsMap[podKey]
 			if !ok {
 				event = common.MapStr{}
-				eventsMap[pod] = event
+				eventsMap[podKey] = event
 			}
+
 			switch family.GetName() {
 			case "kube_pod_info":
 				event.Put(mb.ModuleDataKey+".node.name", util.GetLabel(metric, "node"))
@@ -62,7 +65,8 @@ func eventMapping(families []*dto.MetricFamily) ([]common.MapStr, error) {
 		}
 	}
 
-	var events []common.MapStr
+	// initialize, populate events array from values in eventsMap
+	events := make([]common.MapStr, 0, len(eventsMap))
 	for _, event := range eventsMap {
 		events = append(events, event)
 	}