[k8sattributes processor] Add optional container metadata

This change provides an option to fetch container metadata from k8s API in addition to k8s pod metadata. The following attributes now can be automatically added by the k8sattributes processor:
- container.image.name
- container.image.tag
- container.id

`container.image.name` and `container.image.tag` require additional container identifier present in resource attributes: `container.name`.

`container.id` requires additional container run identifiers present in resource attributes: `container.name` and `run_id`.

`run_id` identified is a subject to change, see open-telemetry/opentelemetry-specification#1945

processor/k8sattributesprocessor/doc.go

@@ -40,7 +40,28 @@
 //
 // 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.
+// The following list of attributes is enabled by default if `metadata` configuration is not specified:
+//   - 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 attributes are not included by default, but can be enabled with `metadata` config option.
+//   1. Container spec attributes - will be set only if container identifying attribute `container.name` is set
+//      as a resource attributes (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

@@ -99,6 +99,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)
 			}