[CONTINT-3554] Send the Datadog-Entity-ID header, containing either…

… the container-id or the cgroup inode if available (#2402)

Co-authored-by: Ahmed Mezghani <38987709+ahmed-mez@users.noreply.github.com>

ddtrace/tracer/transport.go

@@ -99,6 +99,9 @@ func newHTTPTransport(url string, client *http.Client) *httpTransport {
 	if cid := internal.ContainerID(); cid != "" {
 		defaultHeaders["Datadog-Container-ID"] = cid
 	}
+	if eid := internal.EntityID(); eid != "" {
+		defaultHeaders["Datadog-Entity-ID"] = eid
+	}
 	return &httpTransport{
 		traceURL: fmt.Sprintf("%s/v0.4/traces", url),
 		statsURL: fmt.Sprintf("%s/v0.6/stats", url),

internal/container.go

@@ -10,12 +10,24 @@ import (
 	"fmt"
 	"io"
 	"os"
+	"path"
 	"regexp"
+	"strings"
+	"syscall"
 )
 
 const (
 	// cgroupPath is the path to the cgroup file where we can find the container id if one exists.
 	cgroupPath = "/proc/self/cgroup"
+
+	// cgroupV2Key is the key used to store the cgroup v2 identify the cgroupv2 mount point in the cgroupMounts map.
+	cgroupV2Key = "cgroupv2"
+
+	// cgroupV1BaseController is the base controller used to identify the cgroup v1 mount point in the cgroupMounts map.
+	cgroupV1BaseController = "memory"
+
+	// defaultCgroupMountPath is the path to the cgroup mount point.
+	defaultCgroupMountPath = "/sys/fs/cgroup"
 )
 
 const (
@@ -33,10 +45,17 @@ var (
 
 	// containerID is the containerID read at init from /proc/self/cgroup
 	containerID string
+
+	// entityID is the entityID to use for the container. It is the `cid-<containerID>` if the container id available,
+	// otherwise the cgroup v2 node inode prefixed with `in-` or an empty string on cgroup v1 or incompatible OS.
+	// It is retrieved by finding cgroup2 mounts in /proc/mounts, finding the cgroup v2 node path in /proc/self/cgroup and
+	// calling stat on mountPath+nodePath to get the inode.
+	entityID string
 )
 
 func init() {
 	containerID = readContainerID(cgroupPath)
+	entityID = readEntityID(defaultCgroupMountPath, cgroupPath)
 }
 
 // parseContainerID finds the first container ID reading from r and returns it.
@@ -69,3 +88,73 @@ func readContainerID(fpath string) string {
 func ContainerID() string {
 	return containerID
 }
+
+// parseCgroupNodePath parses /proc/self/cgroup and returns a map of controller to its associated cgroup node path.
+func parseCgroupNodePath(r io.Reader) map[string]string {
+	res := make(map[string]string)
+	scn := bufio.NewScanner(r)
+	for scn.Scan() {
+		line := scn.Text()
+		tokens := strings.Split(line, ":")
+		if len(tokens) != 3 {
+			continue
+		}
+		if tokens[1] == cgroupV1BaseController || tokens[1] == "" {
+			res[tokens[1]] = tokens[2]
+		}
+	}
+	return res
+}
+
+// getCgroupInode returns the cgroup controller inode if it exists otherwise an empty string.
+// The inode is prefixed by "in-" and is used by the agent to retrieve the container ID.
+// We first try to retrieve the cgroupv1 memory controller inode, if it fails we try to retrieve the cgroupv2 inode.
+func getCgroupInode(cgroupMountPath, procSelfCgroupPath string) string {
+	// Parse /proc/self/cgroup to retrieve the paths to the memory controller (cgroupv1) and the cgroup node (cgroupv2)
+	f, err := os.Open(procSelfCgroupPath)
+	if err != nil {
+		return ""
+	}
+	defer f.Close()
+	cgroupControllersPaths := parseCgroupNodePath(f)
+
+	// Retrieve the cgroup inode from /sys/fs/cgroup + controller + cgroupNodePath
+	for _, controller := range []string{cgroupV1BaseController, ""} {
+		cgroupNodePath, ok := cgroupControllersPaths[controller]
+		if !ok {
+			continue
+		}
+		inode := inodeForPath(path.Join(cgroupMountPath, controller, cgroupNodePath))
+		if inode != "" {
+			return inode
+		}
+	}
+	return ""
+}
+
+// inodeForPath returns the inode for the provided path or empty on failure.
+func inodeForPath(path string) string {
+	fi, err := os.Stat(path)
+	if err != nil {
+		return ""
+	}
+	stats, ok := fi.Sys().(*syscall.Stat_t)
+	if !ok {
+		return ""
+	}
+	return fmt.Sprintf("in-%d", stats.Ino)
+}
+
+// readEntityID attempts to return the cgroup v2 node inode or empty on failure.
+func readEntityID(mountPath, cgroupPath string) string {
+	if containerID != "" {
+		return "cid-" + containerID
+	}
+	return getCgroupInode(mountPath, cgroupPath)
+}
+
+// EntityID attempts to return the container ID or the cgroup v2 node inode if the container ID is not available.
+// The cid is prefixed with `cid-` and the inode with `in-`.
+func EntityID() string {
+	return entityID
+}

internal/container_test.go

@@ -6,12 +6,17 @@
 package internal
 
 import (
+	"fmt"
 	"io"
+	"io/ioutil"
 	"os"
+	"path"
 	"strings"
+	"syscall"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestReadContainerID(t *testing.T) {
@@ -79,3 +84,176 @@ func TestReadContainerIDFromCgroup(t *testing.T) {
 	actualCID := readContainerID(tmpFile.Name())
 	assert.Equal(t, cid, actualCID)
 }
+
+func TestReadEntityIDPrioritizeCID(t *testing.T) {
+	// reset cid after test
+	defer func(cid string) { containerID = cid }(containerID)
+
+	containerID = "fakeContainerID"
+	eid := readEntityID("", "")
+	assert.Equal(t, "cid-fakeContainerID", eid)
+}
+
+func TestReadEntityIDFallbackOnInode(t *testing.T) {
+	// reset cid after test
+	defer func(cid string) { containerID = cid }(containerID)
+	containerID = ""
+
+	sysFsCgroupPath := path.Join(os.TempDir(), "sysfscgroup")
+	groupControllerPath := path.Join(sysFsCgroupPath, "mynode")
+	err := os.MkdirAll(groupControllerPath, 0755)
+	require.NoError(t, err)
+	defer os.RemoveAll(groupControllerPath)
+
+	stat, err := os.Stat(groupControllerPath)
+	require.NoError(t, err)
+	expectedInode := fmt.Sprintf("in-%d", stat.Sys().(*syscall.Stat_t).Ino)
+
+	procSelfCgroup, err := ioutil.TempFile("", "procselfcgroup")
+	require.NoError(t, err)
+	defer os.Remove(procSelfCgroup.Name())
+	_, err = procSelfCgroup.WriteString("0::/mynode")
+	require.NoError(t, err)
+	err = procSelfCgroup.Close()
+	require.NoError(t, err)
+
+	eid := readEntityID(sysFsCgroupPath, procSelfCgroup.Name())
+	assert.Equal(t, expectedInode, eid)
+}
+
+func TestParsegroupControllerPath(t *testing.T) {
+	// Test cases
+	cases := []struct {
+		name     string
+		content  string
+		expected map[string]string
+	}{
+		{
+			name:     "cgroup2 normal case",
+			content:  `0::/`,
+			expected: map[string]string{"": "/"},
+		},
+		{
+			name: "hybrid",
+			content: `other_line
+0::/
+1:memory:/docker/abc123`,
+			expected: map[string]string{
+				"":       "/",
+				"memory": "/docker/abc123",
+			},
+		},
+		{
+			name: "with other controllers",
+			content: `other_line
+12:pids:/docker/abc123
+11:hugetlb:/docker/abc123
+10:net_cls,net_prio:/docker/abc123
+0::/docker/abc123
+`,
+			expected: map[string]string{
+				"": "/docker/abc123",
+			},
+		},
+		{
+			name:     "no controller",
+			content:  "empty",
+			expected: map[string]string{},
+		},
+	}
+
+	// Run test cases
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			reader := strings.NewReader(c.content)
+			result := parseCgroupNodePath(reader)
+			require.Equal(t, c.expected, result)
+		})
+	}
+}
+
+func TestGetCgroupInode(t *testing.T) {
+	tests := []struct {
+		description           string
+		cgroupNodeDir         string
+		procSelfCgroupContent string
+		expectedResult        string
+		controller            string
+	}{
+		{
+			description:           "matching entry in /proc/self/cgroup and /proc/mounts - cgroup2 only",
+			cgroupNodeDir:         "system.slice/docker-abcdef0123456789abcdef0123456789.scope",
+			procSelfCgroupContent: "0::/system.slice/docker-abcdef0123456789abcdef0123456789.scope\n",
+			expectedResult:        "in-%d", // Will be formatted with inode number
+		},
+		{
+			description:   "matching entry in /proc/self/cgroup and /proc/mounts - cgroup/hybrid only",
+			cgroupNodeDir: "system.slice/docker-abcdef0123456789abcdef0123456789.scope",
+			procSelfCgroupContent: `
+3:memory:/system.slice/docker-abcdef0123456789abcdef0123456789.scope
+2:net_cls,net_prio:c
+1:name=systemd:b
+0::a
+`,
+			expectedResult: "in-%d",
+			controller:     cgroupV1BaseController,
+		},
+		{
+			description:   "non memory or empty controller",
+			cgroupNodeDir: "system.slice/docker-abcdef0123456789abcdef0123456789.scope",
+			procSelfCgroupContent: `
+3:cpu:/system.slice/docker-abcdef0123456789abcdef0123456789.scope
+2:net_cls,net_prio:c
+1:name=systemd:b
+0::a
+`,
+			expectedResult: "",
+			controller:     "cpu",
+		},
+		{
+			description:   "path does not exist",
+			cgroupNodeDir: "dummy.scope",
+			procSelfCgroupContent: `
+3:memory:/system.slice/docker-abcdef0123456789abcdef0123456789.scope
+2:net_cls,net_prio:c
+1:name=systemd:b
+0::a
+`,
+			expectedResult: "",
+		},
+		{
+			description:           "no entry in /proc/self/cgroup",
+			cgroupNodeDir:         "system.slice/docker-abcdef0123456789abcdef0123456789.scope",
+			procSelfCgroupContent: "nothing",
+			expectedResult:        "",
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.description, func(t *testing.T) {
+			sysFsCgroupPath := path.Join(os.TempDir(), "sysfscgroup")
+			groupControllerPath := path.Join(sysFsCgroupPath, tc.controller, tc.cgroupNodeDir)
+			err := os.MkdirAll(groupControllerPath, 0755)
+			require.NoError(t, err)
+			defer os.RemoveAll(groupControllerPath)
+
+			stat, err := os.Stat(groupControllerPath)
+			require.NoError(t, err)
+			expectedInode := ""
+			if tc.expectedResult != "" {
+				expectedInode = fmt.Sprintf(tc.expectedResult, stat.Sys().(*syscall.Stat_t).Ino)
+			}
+
+			procSelfCgroup, err := ioutil.TempFile("", "procselfcgroup")
+			require.NoError(t, err)
+			defer os.Remove(procSelfCgroup.Name())
+			_, err = procSelfCgroup.WriteString(tc.procSelfCgroupContent)
+			require.NoError(t, err)
+			err = procSelfCgroup.Close()
+			require.NoError(t, err)
+
+			result := getCgroupInode(sysFsCgroupPath, procSelfCgroup.Name())
+			require.Equal(t, expectedInode, result)
+		})
+	}
+}

internal/datastreams/transport.go

@@ -68,6 +68,9 @@ func newHTTPTransport(agentURL *url.URL, client *http.Client) *httpTransport {
 	if cid := internal.ContainerID(); cid != "" {
 		defaultHeaders["Datadog-Container-ID"] = cid
 	}
+	if entityID := internal.ContainerID(); entityID != "" {
+		defaultHeaders["Datadog-Entity-ID"] = entityID
+	}
 	url := fmt.Sprintf("%s/v0.1/pipeline_stats", agentURL.String())
 	return &httpTransport{
 		url:     url,

internal/telemetry/client.go

@@ -472,6 +472,7 @@ func (c *client) newRequest(t RequestType) *Request {
 		"DD-Agent-Env":               {c.Env},
 		"DD-Agent-Hostname":          {hostname},
 		"Datadog-Container-ID":       {internal.ContainerID()},
+		"Datadog-Entity-ID":          {internal.EntityID()},
 	}
 	if c.URL == getAgentlessURL() {
 		header.Set("DD-API-KEY", c.APIKey)

profiler/profiler.go

@@ -31,6 +31,7 @@ var (
 	mu             sync.Mutex
 	activeProfiler *profiler
 	containerID    = internal.ContainerID() // replaced in tests
+	entityID       = internal.EntityID()    // replaced in tests
 )
 
 // Start starts the profiler. If the profiler is already running, it will be

profiler/upload.go

@@ -110,6 +110,9 @@ func (p *profiler) doRequest(bat batch) error {
 	if containerID != "" {
 		req.Header.Set("Datadog-Container-ID", containerID)
 	}
+	if entityID != "" {
+		req.Header.Set("Datadog-Entity-ID", entityID)
+	}
 	req.Header.Set("Content-Type", contentType)
 
 	resp, err := p.cfg.httpClient.Do(req)