Add docker and containerd meta labels (#3102)

This adds docker and containerd container labels as meta labels so users
can relabel them onto samples the same way as it is possible with eg.
kubernetes pod labels.

Author: brancz

README.md

@@ -301,9 +301,13 @@ Using relabeling the following labels can be attached to profiles:
 * `__meta_kubernetes_node_annotationpresent_*`: Whether the annotation `*` of the node the process is running on is present.
 * `__meta_docker_container_id`: The ID of the container the process is running in.
 * `__meta_docker_container_name`: The name of the container the process is running in.
+* `__meta_docker_container_label_*`: The value of the label `*` of the container the process is running in.
+* `__meta_docker_container_labelpresent_*`: Whether the label `*` of the container the process is running in is present.
 * `__meta_docker_build_kit_container_id`: The ID of the container the process is running in.
 * `__meta_containerd_container_id`: The ID of the container the process is running in.
 * `__meta_containerd_container_name`: The name of the container the process is running in.
+* `__meta_containerd_container_label_*`: The value of the label `*` of the container the process is running in.
+* `__meta_containerd_container_labelpresent_*`: Whether the label `*` of the container the process is running in is present.
 * `__meta_containerd_pod_name`: The name of the pod the process is running in.
 * `__meta_lxc_container_id`: The ID of the container the process is running in.
 * `__meta_cpu`: The CPU the sample was taken on.

reporter/metadata/agent.go

@@ -1,8 +1,10 @@
 package metadata
 
 import (
-	"go.opentelemetry.io/ebpf-profiler/libpf"
+	"context"
+
 	"github.com/prometheus/prometheus/model/labels"
+	"go.opentelemetry.io/ebpf-profiler/libpf"
 )
 
 type agentMetadataProvider struct {
@@ -13,7 +15,7 @@ func NewAgentMetadataProvider(revision string) MetadataProvider {
 	return &agentMetadataProvider{revision: revision}
 }
 
-func (p *agentMetadataProvider) AddMetadata(_ libpf.PID, lb *labels.Builder) bool {
+func (p *agentMetadataProvider) AddMetadata(_ context.Context, _ libpf.PID, lb *labels.Builder) bool {
 	lb.Set("__meta_agent_revision", p.revision)
 	return true
 }

reporter/metadata/containermetadata.go

@@ -99,7 +99,7 @@ var (
 type MetadataProvider interface {
 	// AddMetadata adds metadata to the provided labels.Builder for the given PID.
 	// It returns whether the metadata can be safely cached.
-	AddMetadata(pid libpf.PID, lb *labels.Builder) bool
+	AddMetadata(ctx context.Context, pid libpf.PID, lb *labels.Builder) bool
 }
 
 // containerMetadataProvider does the retrieval of container metadata for a particular pid.
@@ -521,7 +521,7 @@ func matchContainerID(containerIDStr string) (string, error) {
 }
 
 // AddMetadata adds metadata to the provided labels.Builder for the given PID.
-func (p *containerMetadataProvider) AddMetadata(pid libpf.PID, lb *labels.Builder) bool {
+func (p *containerMetadataProvider) AddMetadata(ctx context.Context, pid libpf.PID, lb *labels.Builder) bool {
 	// Fast path, check container metadata has been cached
 	// For kubernetes pods, the shared informer may have updated
 	// the container id to container metadata cache, so retrieve the container ID for this pid.
@@ -552,7 +552,7 @@ func (p *containerMetadataProvider) AddMetadata(pid libpf.PID, lb *labels.Builde
 	// client.
 	switch {
 	case isContainerEnvironment(env, envKubernetes) && p.kubeClientSet != nil:
-		metadata, err := p.getKubernetesPodMetadata(pidContainerID)
+		metadata, err := p.getKubernetesPodMetadata(ctx, pidContainerID)
 		if err != nil {
 			log.Debugf("Failed to get kubernetes pod metadata for container id %v: %v",
 				pidContainerID, err)
@@ -563,7 +563,7 @@ func (p *containerMetadataProvider) AddMetadata(pid libpf.PID, lb *labels.Builde
 		}
 		return true
 	case isContainerEnvironment(env, envDocker) && p.dockerClient != nil:
-		metadata, err := p.getDockerContainerMetadata(pidContainerID)
+		metadata, err := p.getDockerContainerMetadata(ctx, pidContainerID)
 		if err != nil {
 			log.Warnf("Failed to get docker container metadata for container id %v: %v",
 				pidContainerID, err)
@@ -574,7 +574,7 @@ func (p *containerMetadataProvider) AddMetadata(pid libpf.PID, lb *labels.Builde
 		}
 		return true
 	case isContainerEnvironment(env, envContainerd) && p.containerdClient != nil:
-		metadata, err := p.getContainerdContainerMetadata(pidContainerID)
+		metadata, err := p.getContainerdContainerMetadata(ctx, pidContainerID)
 		if err != nil {
 			log.Debugf("Failed to get containerd container metadata for container id %v: %v",
 				pidContainerID, err)
@@ -596,12 +596,12 @@ func (p *containerMetadataProvider) AddMetadata(pid libpf.PID, lb *labels.Builde
 	}
 }
 
-func (p *containerMetadataProvider) getKubernetesPodMetadata(pidContainerID string) (
+func (p *containerMetadataProvider) getKubernetesPodMetadata(ctx context.Context, pidContainerID string) (
 	model.LabelSet, error) {
 	log.Debugf("Get kubernetes pod metadata for container id %v", pidContainerID)
 
 	p.kubernetesClientQueryCount.Add(1)
-	pods, err := p.kubeClientSet.CoreV1().Pods("").List(context.TODO(), v1.ListOptions{
+	pods, err := p.kubeClientSet.CoreV1().Pods("").List(ctx, v1.ListOptions{
 		FieldSelector: "spec.nodeName=" + p.nodeName,
 	})
 	if err != nil {
@@ -657,12 +657,12 @@ func (p *containerMetadataProvider) getKubernetesPodMetadata(pidContainerID stri
 			"containerID '%v' in %d pods", pidContainerID, len(pods.Items))
 }
 
-func (p *containerMetadataProvider) getDockerContainerMetadata(pidContainerID string) (
+func (p *containerMetadataProvider) getDockerContainerMetadata(ctx context.Context, pidContainerID string) (
 	model.LabelSet, error) {
 	log.Debugf("Get docker container metadata for container id %v", pidContainerID)
 
 	p.dockerClientQueryCount.Add(1)
-	containers, err := p.dockerClient.ContainerList(context.Background(),
+	containers, err := p.dockerClient.ContainerList(ctx,
 		container.ListOptions{})
 	if err != nil {
 		return nil, fmt.Errorf("failed to list docker containers, %v", err)
@@ -676,6 +676,13 @@ func (p *containerMetadataProvider) getDockerContainerMetadata(pidContainerID st
 				"__meta_docker_container_id":   lv(containers[i].ID),
 				"__meta_docker_container_name": lv(containerName),
 			}
+
+			for k, v := range containers[i].Labels {
+				ln := strutil.SanitizeLabelName(k)
+				metadata[model.LabelName("__meta_docker_container_label_"+ln)] = lv(v)
+				metadata[model.LabelName("__meta_docker_container_labelpresent_"+ln)] = presentValue
+			}
+
 			p.containerMetadataCache.Add(pidContainerID, metadata)
 			return metadata, nil
 		}
@@ -686,7 +693,7 @@ func (p *containerMetadataProvider) getDockerContainerMetadata(pidContainerID st
 			pidContainerID)
 }
 
-func (p *containerMetadataProvider) getContainerdContainerMetadata(pidContainerID string) (
+func (p *containerMetadataProvider) getContainerdContainerMetadata(ctx context.Context, pidContainerID string) (
 	model.LabelSet, error) {
 	log.Debugf("Get containerd container metadata for container id %v", pidContainerID)
 
@@ -701,7 +708,7 @@ func (p *containerMetadataProvider) getContainerdContainerMetadata(pidContainerI
 	}
 
 	p.containerdClientQueryCount.Add(1)
-	ctx := namespaces.WithNamespace(context.Background(), fields[1])
+	ctx = namespaces.WithNamespace(ctx, fields[1])
 	containers, err := p.containerdClient.Containers(ctx)
 	if err != nil {
 		return nil,
@@ -718,6 +725,19 @@ func (p *containerMetadataProvider) getContainerdContainerMetadata(pidContainerI
 				"__meta_containerd_container_name": lv(fields[2]),
 				"__meta_containerd_pod_name":       lv(fields[1]),
 			}
+
+			lbls, err := container.Labels(ctx)
+			if err != nil {
+				return nil,
+					fmt.Errorf("failed to get containerd container labels for container id '%s': %v",
+						fields[2], err)
+			}
+			for k, v := range lbls {
+				ln := strutil.SanitizeLabelName(k)
+				metadata[model.LabelName("__meta_containerd_container_label_"+ln)] = lv(v)
+				metadata[model.LabelName("__meta_containerd_container_labelpresent_"+ln)] = presentValue
+			}
+
 			p.containerMetadataCache.Add(pidContainerID, metadata)
 			return metadata, nil
 		}

reporter/metadata/process.go

@@ -3,6 +3,7 @@ package metadata
 import (
 	"bufio"
 	"bytes"
+	"context"
 	"errors"
 	"fmt"
 	"io"
@@ -12,9 +13,9 @@ import (
 	"strings"
 
 	lru "github.com/elastic/go-freelru"
-	"go.opentelemetry.io/ebpf-profiler/libpf"
 	"github.com/prometheus/prometheus/model/labels"
 	log "github.com/sirupsen/logrus"
+	"go.opentelemetry.io/ebpf-profiler/libpf"
 )
 
 var ErrFileParse = errors.New("Error Parsing File")
@@ -167,6 +168,7 @@ func NewMainExecutableMetadataProvider(
 
 // AddMetadata adds metadata labels for the main executable of a process to the given labels.Builder.
 func (p *mainExecutableMetadataProvider) AddMetadata(
+	_ context.Context,
 	pid libpf.PID,
 	lb *labels.Builder,
 ) bool {
@@ -202,7 +204,7 @@ func NewProcessMetadataProvider() MetadataProvider {
 }
 
 // AddMetadata adds metadata labels for a process to the given labels.Builder.
-func (pmp *processMetadataProvider) AddMetadata(pid libpf.PID, lb *labels.Builder) bool {
+func (pmp *processMetadataProvider) AddMetadata(_ context.Context, pid libpf.PID, lb *labels.Builder) bool {
 	cache := true
 	lb.Set("__meta_process_pid", strconv.Itoa(int(pid)))
 

reporter/metadata/system.go

@@ -1,11 +1,12 @@
 package metadata
 
 import (
+	"context"
 	"strings"
 	"syscall"
 
-	"go.opentelemetry.io/ebpf-profiler/libpf"
 	"github.com/prometheus/prometheus/model/labels"
+	"go.opentelemetry.io/ebpf-profiler/libpf"
 )
 
 type systemMetadataProvider struct {
@@ -37,7 +38,7 @@ func NewSystemMetadataProvider() (MetadataProvider, error) {
 	}, nil
 }
 
-func (p *systemMetadataProvider) AddMetadata(_ libpf.PID, lb *labels.Builder) bool {
+func (p *systemMetadataProvider) AddMetadata(_ context.Context, _ libpf.PID, lb *labels.Builder) bool {
 	lb.Set("__meta_system_kernel_machine", p.kernelMachine)
 	lb.Set("__meta_system_kernel_release", p.kernelRelease)
 	return true

reporter/parca_reporter.go

@@ -285,11 +285,11 @@ func (r *ParcaReporter) ReportTraceEvent(trace *libpf.Trace,
 	return nil
 }
 
-func (r *ParcaReporter) addMetadataForPID(pid libpf.PID, lb *labels.Builder) bool {
+func (r *ParcaReporter) addMetadataForPID(ctx context.Context, pid libpf.PID, lb *labels.Builder) bool {
 	cache := true
 
 	for _, p := range r.metadataProviders {
-		cacheable := p.AddMetadata(pid, lb)
+		cacheable := p.AddMetadata(ctx, pid, lb)
 		cache = cache && cacheable
 	}
 
@@ -315,7 +315,7 @@ func (r *ParcaReporter) labelsForTID(tid, pid libpf.PID, comm string, cpu int, e
 		lb.Set("job", "oomprof")
 	}
 
-	cacheable := r.addMetadataForPID(pid, lb)
+	cacheable := r.addMetadataForPID(context.TODO(), pid, lb)
 
 	keep := relabel.ProcessBuilder(lb, r.relabelConfigs...)