[k8sattributes processor] Add container metadata

processor/k8sattributesprocessor/doc.go

@@ -40,7 +40,29 @@
 //
 // If Pod association rules are not configured resources are associated with metadata only by connection's IP Address.
 //
-//
+// Which metadata to collect is determined by `metadata` configuration that defines list of resource attributes
+// to be added. Items in the list called exactly the same as the resource attributes that will be added.
+// All the available attributes are enabled by default, you can reduce the list with `metadata` configuration.
+// The following attributes will be added if pod identified:
+//   - k8s.namespace.name
+//   - k8s.pod.name
+//   - k8s.pod.uid
+//   - k8s.pod.start_time
+//   - k8s.deployment.name
+//   - k8s.cluster.name
+//   - k8s.node.name
+// Not all the attributes are guaranteed to be added. For example `k8s.cluster.name` usually is not provided by k8s API,
+// so likely it won't be set as an attribute.
+
+// The following container level attributes require additional attributes to identify a particular container in a pod:
+//   1. Container spec attributes - will be set only if container identifying attribute `container.name` is set
+//      as a resource attribute (similar to all other attributes, pod has to be identified as well):
+//     - container.image.name
+//     - container.image.tag
+//   2. Container status attributes - in addition to pod identifier and `container.name` attribute, these attributes
+//     require identifier of a particular container run set as `run_id` in resource attributes:
+//     - container.id
+
 //The k8sattributesprocessor can be used for automatic tagging of spans, metrics and logs with k8s labels and annotations from pods and namespaces.
 //The config for associating the data passing through the processor (spans, metrics and logs) with specific Pod/Namespace annotations/labels is configured via "annotations"  and "labels" keys.
 //This config represents a list of annotations/labels that are extracted from pods/namespaces and added to spans, metrics and logs.

processor/k8sattributesprocessor/kube/client.go

@@ -330,6 +330,48 @@ func (c *WatchClient) extractPodAttributes(pod *api_v1.Pod) map[string]string {
 	return tags
 }
 
+func (c *WatchClient) extractPodContainersAttributes(pod *api_v1.Pod) map[string]*Container {
+	containers := map[string]*Container{}
+
+	if c.Rules.ContainerImageName || c.Rules.ContainerImageTag {
+		for _, spec := range append(pod.Spec.Containers, pod.Spec.InitContainers...) {
+			container := &Container{}
+			imageParts := strings.Split(spec.Image, ":")
+			if c.Rules.ContainerImageName {
+				container.ImageName = imageParts[0]
+			}
+			if c.Rules.ContainerImageTag && len(imageParts) > 1 {
+				container.ImageTag = imageParts[1]
+			}
+			containers[spec.Name] = container
+		}
+	}
+
+	if c.Rules.ContainerID {
+		for _, apiStatus := range append(pod.Status.ContainerStatuses, pod.Status.InitContainerStatuses...) {
+			container, ok := containers[apiStatus.Name]
+			if !ok {
+				container = &Container{}
+				containers[apiStatus.Name] = container
+			}
+			if container.Statuses == nil {
+				container.Statuses = map[int]ContainerStatus{}
+			}
+
+			containerID := apiStatus.ContainerID
+
+			// Remove container runtime prefix
+			idParts := strings.Split(containerID, "://")
+			if len(idParts) == 2 {
+				containerID = idParts[1]
+			}
+
+			container.Statuses[int(apiStatus.RestartCount)] = ContainerStatus{containerID}
+		}
+	}
+	return containers
+}
+
 func (c *WatchClient) extractNamespaceAttributes(namespace *api_v1.Namespace) map[string]string {
 	tags := map[string]string{}
 
@@ -378,6 +420,9 @@ func (c *WatchClient) addOrUpdatePod(pod *api_v1.Pod) {
 		newPod.Ignore = true
 	} else {
 		newPod.Attributes = c.extractPodAttributes(pod)
+		if needContainerAttributes(c.Rules) {
+			newPod.Containers = c.extractPodContainersAttributes(pod)
+		}
 	}
 
 	c.m.Lock()
@@ -513,3 +558,7 @@ func (c *WatchClient) extractNamespaceLabelsAnnotations() bool {
 
 	return false
 }
+
+func needContainerAttributes(rules ExtractionRules) bool {
+	return rules.ContainerImageName || rules.ContainerImageTag || rules.ContainerID
+}

processor/k8sattributesprocessor/kube/client_test.go

@@ -747,6 +747,166 @@ func TestPodIgnorePatterns(t *testing.T) {
 	}
 }
 
+func Test_extractPodContainersAttributes(t *testing.T) {
+	pod := api_v1.Pod{
+		Spec: api_v1.PodSpec{
+			Containers: []api_v1.Container{
+				{
+					Name:  "container1",
+					Image: "test/image1:0.1.0",
+				},
+				{
+					Name:  "container2",
+					Image: "test/image2:0.2.0",
+				},
+			},
+			InitContainers: []api_v1.Container{
+				{
+					Name:  "init_container",
+					Image: "test/init-image:1.0.2",
+				},
+			},
+		},
+		Status: api_v1.PodStatus{
+			ContainerStatuses: []api_v1.ContainerStatus{
+				{
+					Name:         "container1",
+					ContainerID:  "docker://container1-id-123",
+					RestartCount: 0,
+				},
+				{
+					Name:         "container2",
+					ContainerID:  "docker://container2-id-456",
+					RestartCount: 2,
+				},
+			},
+			InitContainerStatuses: []api_v1.ContainerStatus{
+				{
+					Name:         "init_container",
+					ContainerID:  "containerd://init-container-id-123",
+					RestartCount: 0,
+				},
+			},
+		},
+	}
+	tests := []struct {
+		name  string
+		rules ExtractionRules
+		pod   api_v1.Pod
+		want  map[string]*Container
+	}{
+		{
+			name: "no-data",
+			rules: ExtractionRules{
+				ContainerImageName: true,
+				ContainerImageTag:  true,
+				ContainerID:        true,
+			},
+			pod:  api_v1.Pod{},
+			want: map[string]*Container{},
+		},
+		{
+			name:  "no-rules",
+			rules: ExtractionRules{},
+			pod:   pod,
+			want:  map[string]*Container{},
+		},
+		{
+			name: "image-name-only",
+			rules: ExtractionRules{
+				ContainerImageName: true,
+			},
+			pod: pod,
+			want: map[string]*Container{
+				"container1":     {ImageName: "test/image1"},
+				"container2":     {ImageName: "test/image2"},
+				"init_container": {ImageName: "test/init-image"},
+			},
+		},
+		{
+			name: "no-image-tag-available",
+			rules: ExtractionRules{
+				ContainerImageName: true,
+			},
+			pod: api_v1.Pod{
+				Spec: api_v1.PodSpec{
+					Containers: []api_v1.Container{
+						{
+							Name:  "test-container",
+							Image: "test/image",
+						},
+					},
+				},
+			},
+			want: map[string]*Container{
+				"test-container": {ImageName: "test/image"},
+			},
+		},
+		{
+			name: "container-id-only",
+			rules: ExtractionRules{
+				ContainerID: true,
+			},
+			pod: pod,
+			want: map[string]*Container{
+				"container1": {
+					Statuses: map[int]ContainerStatus{
+						0: {ContainerID: "container1-id-123"},
+					},
+				},
+				"container2": {
+					Statuses: map[int]ContainerStatus{
+						2: {ContainerID: "container2-id-456"},
+					},
+				},
+				"init_container": {
+					Statuses: map[int]ContainerStatus{
+						0: {ContainerID: "init-container-id-123"},
+					},
+				},
+			},
+		},
+		{
+			name: "all-container-attributes",
+			rules: ExtractionRules{
+				ContainerImageName: true,
+				ContainerImageTag:  true,
+				ContainerID:        true,
+			},
+			pod: pod,
+			want: map[string]*Container{
+				"container1": {
+					ImageName: "test/image1",
+					ImageTag:  "0.1.0",
+					Statuses: map[int]ContainerStatus{
+						0: {ContainerID: "container1-id-123"},
+					},
+				},
+				"container2": {
+					ImageName: "test/image2",
+					ImageTag:  "0.2.0",
+					Statuses: map[int]ContainerStatus{
+						2: {ContainerID: "container2-id-456"},
+					},
+				},
+				"init_container": {
+					ImageName: "test/init-image",
+					ImageTag:  "1.0.2",
+					Statuses: map[int]ContainerStatus{
+						0: {ContainerID: "init-container-id-123"},
+					},
+				},
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			c := WatchClient{Rules: tt.rules}
+			assert.Equal(t, tt.want, c.extractPodContainersAttributes(&tt.pod))
+		})
+	}
+}
+
 func Test_extractField(t *testing.T) {
 	c := WatchClient{}
 	type args struct {

processor/k8sattributesprocessor/kube/kube.go

@@ -71,9 +71,26 @@ type Pod struct {
 	Ignore     bool
 	Namespace  string
 
+	// Containers is a map of container name to Container struct.
+	Containers map[string]*Container
+
 	DeletedAt time.Time
 }
 
+// Container stores resource attributes for a specific container defined by k8s pod spec.
+type Container struct {
+	ImageName string
+	ImageTag  string
+
+	// Statuses is a map of container run_id (restart count) attribute to ContainerStatus struct.
+	Statuses map[int]ContainerStatus
+}
+
+// ContainerStatus stores resource attributes for a particular container run defined by k8s pod status.
+type ContainerStatus struct {
+	ContainerID string
+}
+
 // Namespace represents a kubernetes namespace.
 type Namespace struct {
 	Name         string
@@ -118,13 +135,16 @@ type FieldFilter struct {
 // ExtractionRules is used to specify the information that needs to be extracted
 // from pods and added to the spans as tags.
 type ExtractionRules struct {
-	Deployment bool
-	Namespace  bool
-	PodName    bool
-	PodUID     bool
-	Node       bool
-	Cluster    bool
-	StartTime  bool
+	Deployment         bool
+	Namespace          bool
+	PodName            bool
+	PodUID             bool
+	Node               bool
+	Cluster            bool
+	StartTime          bool
+	ContainerID        bool
+	ContainerImageName bool
+	ContainerImageTag  bool
 
 	Annotations []FieldExtractionRule
 	Labels      []FieldExtractionRule

processor/k8sattributesprocessor/options.go

@@ -78,6 +78,9 @@ func WithExtractMetadata(fields ...string) Option {
 				conventions.AttributeK8SDeploymentName,
 				conventions.AttributeK8SClusterName,
 				conventions.AttributeK8SNodeName,
+				conventions.AttributeContainerID,
+				conventions.AttributeContainerImageName,
+				conventions.AttributeContainerImageTag,
 			}
 		}
 		for _, field := range fields {
@@ -99,6 +102,12 @@ func WithExtractMetadata(fields ...string) Option {
 				p.rules.Cluster = true
 			case metadataNode, conventions.AttributeK8SNodeName:
 				p.rules.Node = true
+			case conventions.AttributeContainerID:
+				p.rules.ContainerID = true
+			case conventions.AttributeContainerImageName:
+				p.rules.ContainerImageName = true
+			case conventions.AttributeContainerImageTag:
+				p.rules.ContainerImageTag = true
 			default:
 				return fmt.Errorf("\"%s\" is not a supported metadata field", field)
 			}