discovery.process and pyroscope.java

component/all/all.go

@@ -24,6 +24,7 @@ import (
 	_ "github.com/grafana/agent/component/discovery/nerve"                          // Import discovery.nerve
 	_ "github.com/grafana/agent/component/discovery/nomad"                          // Import discovery.nomad
 	_ "github.com/grafana/agent/component/discovery/openstack"                      // Import discovery.openstack
+	_ "github.com/grafana/agent/component/discovery/process"                        // Import discovery.process
 	_ "github.com/grafana/agent/component/discovery/puppetdb"                       // Import discovery.puppetdb
 	_ "github.com/grafana/agent/component/discovery/relabel"                        // Import discovery.relabel
 	_ "github.com/grafana/agent/component/discovery/scaleway"                       // Import discovery.scaleway

component/discovery/process/container.go

@@ -0,0 +1,56 @@
+package process
+
+import (
+	"bufio"
+	"io"
+	"regexp"
+	"strings"
+
+	"github.com/grafana/agent/component/discovery"
+)
+
+var (
+	// cgroupContainerIDRe matches a container ID from a /proc/{pid}}/cgroup
+	cgroupContainerIDRe = regexp.MustCompile(`^.*/(?:.*-)?([0-9a-f]{64})(?:\.|\s*$)`)
+)
+
+func getContainerIDFromCGroup(cgroup io.Reader) string {
+	scanner := bufio.NewScanner(cgroup)
+	for scanner.Scan() {
+		line := scanner.Bytes()
+		matches := cgroupContainerIDRe.FindSubmatch(line)
+		if len(matches) <= 1 {
+			continue
+		}
+		return string(matches[1])
+	}
+	return ""
+}
+
+var knownContainerIDPrefixes = []string{"docker://", "containerd://", "cri-o://"}
+
+// get container id from __meta_kubernetes_pod_container_id label
+func getContainerIDFromK8S(k8sContainerID string) string {
+	for _, p := range knownContainerIDPrefixes {
+		if strings.HasPrefix(k8sContainerID, p) {
+			return strings.TrimPrefix(k8sContainerID, p)
+		}
+	}
+	return ""
+}
+
+func getContainerIDFromTarget(target discovery.Target) string {
+	cid, ok := target[labelProcessContainerID]
+	if ok && cid != "" {
+		return cid
+	}
+	cid, ok = target["__meta_kubernetes_pod_container_id"]
+	if ok && cid != "" {
+		return getContainerIDFromK8S(cid)
+	}
+	cid, ok = target["__meta_docker_container_id"]
+	if ok && cid != "" {
+		return cid
+	}
+	return ""
+}

component/discovery/process/container_test.go

@@ -0,0 +1,64 @@
+package process
+
+import (
+	"bytes"
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestCGroupMatching(t *testing.T) {
+	type testcase = struct {
+		containerID, cgroup, expectedID string
+	}
+	testcases := []testcase{
+		{
+			containerID: "containerd://a534eb629135e43beb13213976e37bb2ab95cba4c0d1d0b4e27c6bc4d8091b83",
+			cgroup: "12:cpuset:/kubepods.slice/kubepods-burstable.slice/" +
+				"kubepods-burstable-pod471203d1_984f_477e_9c35_db96487ffe5e.slice/" +
+				"cri-containerd-a534eb629135e43beb13213976e37bb2ab95cba4c0d1d0b4e27c6bc4d8091b83.scope",
+			expectedID: "a534eb629135e43beb13213976e37bb2ab95cba4c0d1d0b4e27c6bc4d8091b83",
+		},
+		{
+			containerID: "cri-o://0ecc7949cbaf17e883264ea1055f60b184a7cb264fd759c4a692e1155086fe2d",
+			cgroup: "0::/kubepods.slice/kubepods-burstable.slice/kubepods-burstable-podb57320a0_e7eb_4ac8_a791_4c4472796867.slice/" +
+				"crio-0ecc7949cbaf17e883264ea1055f60b184a7cb264fd759c4a692e1155086fe2d.scope",
+			expectedID: "0ecc7949cbaf17e883264ea1055f60b184a7cb264fd759c4a692e1155086fe2d",
+		},
+		{
+
+			containerID: "docker://656959d9ee87a0b131c601ce9d9f8f76b1dda60e8608c503b5979d849cbdc714",
+			cgroup: "0::/../../kubepods-besteffort-pod88f6f4e3_59c0_4ce8_9ecf_391c8b5a60ad.slice/" +
+				"docker-656959d9ee87a0b131c601ce9d9f8f76b1dda60e8608c503b5979d849cbdc714.scope",
+			expectedID: "656959d9ee87a0b131c601ce9d9f8f76b1dda60e8608c503b5979d849cbdc714",
+		},
+		{
+			containerID: "containerd://47e320f795efcec1ecf2001c3a09c95e3701ed87de8256837b70b10e23818251",
+			cgroup: "0::/kubepods.slice/kubepods-burstable.slice/" +
+				"kubepods-burstable-podf9a04ecc_1875_491b_926c_d2f64757704e.slice/" +
+				"cri-containerd-47e320f795efcec1ecf2001c3a09c95e3701ed87de8256837b70b10e23818251.scope",
+			expectedID: "47e320f795efcec1ecf2001c3a09c95e3701ed87de8256837b70b10e23818251",
+		},
+		{
+			containerID: "docker://7edda1de1e0d1d366351e478359cf5fa16bb8ab53063a99bb119e56971bfb7e2",
+			cgroup: "11:devices:/kubepods/besteffort/pod85adbef3-622f-4ef2-8f60-a8bdf3eb6c72/" +
+				"7edda1de1e0d1d366351e478359cf5fa16bb8ab53063a99bb119e56971bfb7e2",
+			expectedID: "7edda1de1e0d1d366351e478359cf5fa16bb8ab53063a99bb119e56971bfb7e2",
+		},
+		{
+			containerID: "",
+			cgroup:      "0::/../../user.slice/user-501.slice/session-3.scope",
+			expectedID:  "",
+		},
+	}
+	for i, tc := range testcases {
+		t.Run(fmt.Sprintf("testcase %d %s", i, tc.cgroup), func(t *testing.T) {
+			cid := getContainerIDFromCGroup(bytes.NewReader([]byte(tc.cgroup)))
+			expected := tc.expectedID
+			require.Equal(t, expected, cid)
+			cid = getContainerIDFromK8S(tc.containerID)
+			require.Equal(t, expected, cid)
+		})
+	}
+}

component/discovery/process/discover.go

@@ -0,0 +1,105 @@
+package process
+
+import (
+	"errors"
+	"fmt"
+	"os"
+	"path"
+	"strconv"
+
+	"github.com/go-kit/log"
+	"github.com/go-kit/log/level"
+	"github.com/grafana/agent/component/discovery"
+)
+
+const (
+	labelProcessID          = "__process_pid__"
+	labelProcessExe         = "__meta_process_exe"
+	labelProcessCwd         = "__meta_process_cwd"
+	labelProcessContainerID = "__container_id__"
+)
+
+type process struct {
+	pid         string
+	exe         string
+	cwd         string
+	containerID string
+}
+
+func discover(l log.Logger, procFS string) ([]process, error) {
+	var (
+		err error
+		ps  []process
+	)
+	pids, err := os.ReadDir(procFS)
+	if err != nil {
+		return nil, fmt.Errorf("discovery.process: failed to read /proc: %w", err)
+	}
+	for _, entry := range pids {
+		var (
+			exe    string
+			cwd    string
+			cgroup *os.File
+		)
+		if !entry.IsDir() {
+			continue
+		}
+		pidDir := entry
+		pid := pidDir.Name()
+		_, err = strconv.Atoi(pid)
+		if err != nil {
+			continue
+		}
+		exe, err = os.Readlink(path.Join(procFS, pid, "exe"))
+		if err != nil {
+			if !errors.Is(err, os.ErrNotExist) {
+				_ = level.Error(l).Log("msg", "failed to read /proc/{pid}/exe", "err", err)
+			}
+			continue
+		}
+		cwd, err = os.Readlink(path.Join(procFS, pid, "cwd"))
+		if err != nil {
+			if !errors.Is(err, os.ErrNotExist) {
+				_ = level.Error(l).Log("msg", "failed to read /proc/{pid}/cwd", "err", err)
+			}
+			continue
+		}
+		cgroup, err = os.Open(path.Join(procFS, pid, "cgroup"))
+		if err != nil {
+			if !errors.Is(err, os.ErrNotExist) {
+				_ = level.Error(l).Log("msg", "failed to read /proc/{pid}/cgroup", "err", err)
+			}
+			continue
+		}
+		cid := getContainerIDFromCGroup(cgroup)
+		ps = append(ps, process{
+			pid:         pid,
+			exe:         exe,
+			cwd:         cwd,
+			containerID: cid,
+		})
+		_ = cgroup.Close()
+	}
+	return ps, nil
+}
+
+func convertProcesses(ps []process) []discovery.Target {
+	var res []discovery.Target
+	for _, p := range ps {
+		t := convertProcess(p)
+		res = append(res, t)
+	}
+	return res
+}
+
+func convertProcess(p process) discovery.Target {
+	t := discovery.Target{
+		labelProcessID:  p.pid,
+		labelProcessExe: p.exe,
+		labelProcessCwd: p.cwd,
+	}
+	if p.containerID != "" {
+		t[labelProcessContainerID] = p.containerID
+	}
+	return t
+}

component/discovery/process/join.go

@@ -0,0 +1,40 @@
+package process
+
+import "github.com/grafana/agent/component/discovery"
+
+func join(processes, containers []discovery.Target) []discovery.Target {
+	res := make([]discovery.Target, 0, len(processes)+len(containers))
+
+	cid2container := make(map[string]discovery.Target, len(containers))
+	for _, container := range containers {
+		cid := getContainerIDFromTarget(container)
+		if cid != "" {
+			cid2container[cid] = container
+		} else {
+			res = append(res, container)
+		}
+	}
+	for _, p := range processes {
+		cid := getContainerIDFromTarget(p)
+		if cid == "" {
+			res = append(res, p)
+			continue
+		}
+		container, ok := cid2container[cid]
+		if !ok {
+			continue
+		}
+		mergedTarget := make(discovery.Target, len(p)+len(container))
+		for k, v := range p {
+			mergedTarget[k] = v
+		}
+		for k, v := range container {
+			mergedTarget[k] = v
+		}
+		res = append(res, mergedTarget)
+	}
+	for _, target := range cid2container {
+		res = append(res, target)
+	}
+	return res
+}

component/discovery/process/join_test.go

@@ -0,0 +1,94 @@
+package process
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/grafana/agent/component/discovery"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestJoin(t *testing.T) {
+	testdata := []struct {
+		processes  []discovery.Target
+		containers []discovery.Target
+		res        []discovery.Target
+	}{
+		{
+			[]discovery.Target{
+				convertProcess(process{
+					pid:         "239",
+					exe:         "/bin/foo",
+					cwd:         "/",
+					containerID: "7edda1de1e0d1d366351e478359cf5fa16bb8ab53063a99bb119e56971bfb7e2",
+				}),
+				convertProcess(process{
+					pid:         "240",
+					exe:         "/bin/bar",
+					cwd:         "/tmp",
+					containerID: "7edda1de1e0d1d366351e478359cf5fa16bb8ab53063a99bb119e56971bfb7e2",
+				}),
+				convertProcess(process{
+					pid:         "241",
+					exe:         "/bin/bash",
+					cwd:         "/opt",
+					containerID: "",
+				}),
+			}, []discovery.Target{
+				{
+					"__meta_docker_container_id": "7edda1de1e0d1d366351e478359cf5fa16bb8ab53063a99bb119e56971bfb7e2",
+					"foo":                        "bar",
+				},
+				{
+					"__meta_kubernetes_pod_container_id": "docker://47e320f795efcec1ecf2001c3a09c95e3701ed87de8256837b70b10e23818251",
+					"qwe":                                "asd",
+				},
+				{
+					"lol": "lol",
+				},
+			}, []discovery.Target{
+				{
+					"__process_pid__":            "239",
+					"__meta_process_exe":         "/bin/foo",
+					"__meta_process_cwd":         "/",
+					"__container_id__":           "7edda1de1e0d1d366351e478359cf5fa16bb8ab53063a99bb119e56971bfb7e2",
+					"__meta_docker_container_id": "7edda1de1e0d1d366351e478359cf5fa16bb8ab53063a99bb119e56971bfb7e2",
+					"foo":                        "bar",
+				},
+				{
+					"__process_pid__":            "240",
+					"__meta_process_exe":         "/bin/bar",
+					"__meta_process_cwd":         "/tmp",
+					"__container_id__":           "7edda1de1e0d1d366351e478359cf5fa16bb8ab53063a99bb119e56971bfb7e2",
+					"__meta_docker_container_id": "7edda1de1e0d1d366351e478359cf5fa16bb8ab53063a99bb119e56971bfb7e2",
+					"foo":                        "bar",
+				},
+				{
+					"__meta_docker_container_id": "7edda1de1e0d1d366351e478359cf5fa16bb8ab53063a99bb119e56971bfb7e2",
+					"foo":                        "bar",
+				},
+				{
+					"__process_pid__":    "241",
+					"__meta_process_exe": "/bin/bash",
+					"__meta_process_cwd": "/opt",
+				},
+				{
+					"__meta_kubernetes_pod_container_id": "docker://47e320f795efcec1ecf2001c3a09c95e3701ed87de8256837b70b10e23818251",
+					"qwe":                                "asd",
+				},
+				{
+					"lol": "lol",
+				},
+			},
+		},
+	}
+	for i, testdatum := range testdata {
+		t.Run(fmt.Sprintf("test %d", i), func(t *testing.T) {
+			res := join(testdatum.processes, testdatum.containers)
+			assert.Len(t, res, len(testdatum.res))
+			for _, re := range testdatum.res {
+				assert.Contains(t, res, re)
+			}
+		})
+	}
+}