Allow docker and containerd samplers running together (NR-148791)

cmd/newrelic-infra-ctl/newrelic-infra-ctl.go

@@ -21,9 +21,11 @@ import (
 )
 
 var (
-	agentPID    int
-	containerID string
-	apiVersion  string
+	agentPID            int
+	containerID         string
+	apiVersion          string
+	containerdNamespace string
+	containerRuntime    string
 )
 
 func init() {
@@ -47,6 +49,20 @@ func init() {
 		config.DefaultDockerApiVersion,
 		"Docker API version [Optional] (Containerised agent)",
 	)
+
+	flag.StringVar(
+		&containerdNamespace,
+		"containerd-namespace",
+		"default",
+		"Namespace in containerd where container is runing [Optional] (Containerised agent)",
+	)
+
+	flag.StringVar(
+		&containerRuntime,
+		"container-runtime",
+		sender.RuntimeDocker,
+		"Container runtime [Optional] ('docker' or 'containerd') (Containerised agent)",
+	)
 }
 
 func main() {
@@ -82,7 +98,7 @@ func getClient() (sender.Client, error) {
 		return sender.NewClient(agentPID)
 	}
 	if containerID != "" {
-		return sender.NewContainerisedClient(apiVersion, containerID)
+		return sender.NewContainerClient(apiVersion, containerdNamespace, containerID, containerRuntime)
 	}
-	return sender.NewAutoDetectedClient(apiVersion)
+	return sender.NewAutoDetectedClient(apiVersion, containerdNamespace, containerRuntime)
 }

pkg/config/config.go

@@ -185,7 +185,7 @@ type Config struct {
 	CollectorURL string `yaml:"collector_url" envconfig:"collector_url" public:"false"`
 
 	// IdentityURL defines the base URL for the identity connect.
-	// Default: https://infra-api.newrelic.com
+	// Default: https://identity-api.newrelic.com
 	// Public: No
 	IdentityURL string `yaml:"identity_url" envconfig:"identity_url" public:"false"`
 
@@ -280,6 +280,11 @@ type Config struct {
 	// Public: Yes
 	DockerApiVersion string `yaml:"docker_api_version" envconfig:"docker_api_version"`
 
+	// DockerContainerdNamespace specifies the Containerd namespace used by docker.
+	// Default: moby
+	// Public: Yes
+	DockerContainerdNamespace string `yaml:"docker_containerd_namespace" envconfig:"docker_containerd_namespace"`
+
 	// CustomAttributes is a list of custom attributes to annotate the data from this agent instance. Separate keys and
 	// values with colons :, as in KEY: VALUE, and separate each key-value pair with a line break. Keys can be any
 	// valid YAML except slashes /. Values can be any YAML string, including spaces.
@@ -1819,6 +1824,7 @@ func NewConfig() *Config {
 		TCPServerPort:                 defaultTCPServerPort,
 		StatusServerPort:              defaultStatusServerPort,
 		DockerApiVersion:              DefaultDockerApiVersion,
+		DockerContainerdNamespace:     DefaultDockerContainerdNamespace,
 		FingerprintUpdateFreqSec:      defaultFingerprintUpdateFreqSec,
 		CloudMetadataExpiryInSec:      defaultCloudMetadataExpiryInSec,
 		RegisterConcurrency:           defaultRegisterConcurrency,
@@ -2301,6 +2307,8 @@ func NormalizeConfig(cfg *Config, cfgMetadata config_loader.YAMLMetadata) (err e
 
 	// DockerApiVersion default value defined in NewConfig
 	nlog.WithField("DockerApiVersion", cfg.DockerApiVersion).Debug("Docker client API version.")
+	// DockerContainerdNamespace default value defined in NewConfig
+	nlog.WithField("DockerContainerdNamespace", cfg.DockerContainerdNamespace).Debug("Docker containerd namespace.")
 	// FingerprintUpdateFreqSec default value defined in NewConfig
 	nlog.WithField("FingerprintUpdateFreqSec", cfg.FingerprintUpdateFreqSec).Debug("Fingerprint update freq.")
 	// DnsHostnameResolution value defined in NewConfig

pkg/config/config_test.go

@@ -356,6 +356,7 @@ agent_dir: /my/overriden/agent/dir
 	c.Assert(cfg.SendInterval, Equals, defaultSendInterval)
 
 	c.Assert(cfg.DockerApiVersion, Equals, DefaultDockerApiVersion)
+	c.Assert(cfg.DockerContainerdNamespace, Equals, DefaultDockerContainerdNamespace)
 	c.Assert(cfg.SelinuxEnableSemodule, Equals, defaultSelinuxEnableSemodule)
 	c.Assert(cfg.DnsHostnameResolution, Equals, defaultDnsHostnameResolution)
 	c.Assert(cfg.IgnoreReclaimable, Equals, defaultIgnoreReclaimable)

pkg/config/defaults.go

@@ -40,6 +40,7 @@ var (
 	// public
 	DefaultContainerCacheMetadataLimit = 60
 	DefaultDockerApiVersion            = "1.24" // minimum supported API by Docker 18.09.0
+	DefaultDockerContainerdNamespace   = "moby"
 	DefaultHeartBeatFrequencySecs      = 60
 	DefaultDMPeriodSecs                = 5           // default telemetry SDK value
 	DefaultMaxMetricsBatchSizeBytes    = 1000 * 1000 // Size limit from Vortex collector service (1MB)

pkg/ctl/sender/client.go

@@ -5,39 +5,44 @@ package sender
 import (
 	"context"
 
-	"github.com/newrelic/infrastructure-agent/pkg/ipc"
-
+	"github.com/newrelic/infrastructure-agent/pkg/config"
 	"github.com/newrelic/infrastructure-agent/pkg/helpers/detection"
+	"github.com/newrelic/infrastructure-agent/pkg/ipc"
 	"github.com/newrelic/infrastructure-agent/pkg/log"
 )
 
+const (
+	RuntimeDocker     = "docker"
+	RuntimeContainerd = "containerd"
+)
+
 var clog = log.WithComponent("NotificationClient")
 
 // Client is used to notify a running agent process.
 type Client interface {
 	// Notify will send a notification that will be captured by the agent which will run a handler.
 	Notify(ctx context.Context, message ipc.Message) error
 
-	// Return the identification for the notified agent.
+	// GetID returns the identification for the notified agent.
 	GetID() string
 }
 
 // NewAutoDetectedClient will try to detect the NRIA instance type and return a notification client for it.
-func NewAutoDetectedClient(dockerAPIVersion string) (Client, error) {
+func NewAutoDetectedClient(dockerAPIVersion, containerdNamespace, containerRuntime string) (Client, error) {
 	pid, err := detection.GetInfraAgentProcess()
 	if err != nil {
 		return nil, err
 	}
 
 	clog.WithField("pid", pid).Info("found agent")
 
-	inContainer, containerID, err := detection.IsContainerized(pid, dockerAPIVersion)
+	inContainer, containerID, err := detection.IsContainerized(pid, dockerAPIVersion, config.DefaultDockerContainerdNamespace)
 	if err != nil {
 		clog.WithError(err).Info("Container ID not identified")
 	}
 
 	if inContainer {
-		return NewContainerisedClient(dockerAPIVersion, containerID)
+		return NewContainerClient(dockerAPIVersion, containerdNamespace, containerID, containerRuntime)
 	}
 	return NewClient(int(pid))
 }

pkg/ctl/sender/client_container_unix.go

@@ -0,0 +1,12 @@
+// Copyright 2023 New Relic Corporation. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+//go:build linux || darwin
+
+package sender
+
+func NewContainerClient(dockerAPIVersion, containerdNamespace, containerID, runtime string) (Client, error) {
+	if runtime == RuntimeContainerd {
+		return NewContainerdClient(containerdNamespace, containerID)
+	}
+	return NewDockerClient(dockerAPIVersion, containerID)
+}

pkg/ctl/sender/client_container_windows.go

@@ -0,0 +1,10 @@
+// Copyright 2020 New Relic Corporation. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+//go:build windows
+// +build windows
+
+package sender
+
+func NewContainerClient(dockerAPIVersion, containerdNamespace, containerID, runtime string) (Client, error) {
+	return NewDockerClient(dockerAPIVersion, containerID)
+}

pkg/ctl/sender/client_containerd.go

@@ -0,0 +1,67 @@
+// Copyright 2023 New Relic Corporation. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+//go:build linux || darwin
+
+package sender
+
+import (
+	"context"
+	"syscall"
+
+	"github.com/containerd/containerd"
+	"github.com/containerd/containerd/api/services/tasks/v1"
+	"github.com/containerd/containerd/namespaces"
+	"github.com/newrelic/infrastructure-agent/pkg/helpers"
+	"github.com/newrelic/infrastructure-agent/pkg/ipc"
+	"github.com/pkg/errors"
+)
+
+type ContainerdClient struct {
+	client      *containerd.Client
+	containerID string
+	namespace   string
+}
+
+// NewContainerdClient creates a containerised agent control client for Containerd containers.
+func NewContainerdClient(namespace, containerID string) (c Client, err error) {
+	if containerID == "" {
+		err = RequiredContainerIDErr
+		return
+	}
+
+	if !helpers.IsContainerdRunning() {
+		err = helpers.ErrNoContainerd
+		return
+	}
+
+	cl, err := containerd.New(helpers.UnixContainerdSocket)
+	if err != nil {
+		err = errors.Wrap(err, "failed to initialize containerd client")
+		return
+	}
+
+	c = &ContainerdClient{
+		client:      cl,
+		containerID: containerID,
+		namespace:   namespace,
+	}
+
+	return
+}
+
+// Notify will notify a running agent process inside a docker container.
+func (c *ContainerdClient) Notify(ctx context.Context, _ ipc.Message) error {
+	ctx = namespaces.WithNamespace(ctx, c.namespace)
+
+	killRequest := tasks.KillRequest{
+		ContainerID: c.containerID,
+		Signal:      uint32(syscall.SIGUSR1),
+	}
+	_, err := c.client.TaskService().Kill(ctx, &killRequest)
+	return err
+}
+
+// GetID returns the identification for the notified agent.
+func (c *ContainerdClient) GetID() string {
+	return c.containerID
+}

pkg/ctl/sender/client_docker.go

@@ -22,9 +22,9 @@ type dockerClient struct {
 // RequiredContainerIDErr is the error to return when container ID is not provided.
 var RequiredContainerIDErr = errors.New("container ID of the agent is required")
 
-// NewContainerisedClient creates a containerised agent control client for Docker containers.
+// NewDockerClient creates a containerised agent control client for Docker containers.
 // Didn't use helpers.DockerClient because it'll broad the interface breaking SRP.
-func NewContainerisedClient(apiVersion string, containerID string) (c Client, err error) {
+func NewDockerClient(apiVersion string, containerID string) (c Client, err error) {
 	if containerID == "" {
 		err = RequiredContainerIDErr
 		return

pkg/helpers/containerd_utils.go

@@ -15,8 +15,7 @@ import (
 
 // ContainerdSocket is the default socket for containerd.
 const (
-	unixContainerdSocket  = "/run/containerd/containerd.sock"
-	windowsContainerdPipe = `\\.\pipe\containerd-containerd`
+	UnixContainerdSocket = "/run/containerd/containerd.sock"
 )
 
 // ErrNoContainerd is returned when containerd is not running.
@@ -60,7 +59,7 @@ func (cc *ContainerdClient) Initialize() error {
 		return containerdError(ErrNoContainerd)
 	}
 
-	client, err := containerd.New(unixContainerdSocket)
+	client, err := containerd.New(UnixContainerdSocket)
 	if err != nil {
 		return containerdError(err)
 	}
@@ -115,12 +114,10 @@ func (cc *ContainerdClient) containersFromNamespaces(nss []string) (map[namespac
 
 func IsContainerdRunning() bool {
 	if runtime.GOOS == "windows" {
-		_, err := os.Stat(windowsContainerdPipe)
-
-		return err == nil
+		return false
 	}
 
-	sock, err := os.Stat(unixContainerdSocket)
+	sock, err := os.Stat(UnixContainerdSocket)
 	if err != nil {
 		return false
 	}

pkg/helpers/detection/detection.go

@@ -35,11 +35,11 @@ func GetProcessID(processName string) (int32, error) {
 }
 
 // IsContainerized is checking if a pid is running inside a docker container.
-func IsContainerized(pid int32, dockerAPIVersion string) (isContainerized bool, containerID string, err error) {
+func IsContainerized(pid int32, dockerAPIVersion, dockerContainerdNamespace string) (isContainerized bool, containerID string, err error) {
 	p := &types.ProcessSample{
 		ProcessID: pid,
 	}
-	containerSampler := metrics.GetContainerSampler(60, dockerAPIVersion) //nolint:gomnd
+	containerSamplers := metrics.GetContainerSamplers(60, dockerAPIVersion, dockerContainerdNamespace) //nolint:gomnd
 
 	logger := log.WithFieldsF(func() logrus.Fields {
 		return logrus.Fields{
@@ -49,14 +49,16 @@ func IsContainerized(pid int32, dockerAPIVersion string) (isContainerized bool,
 		}
 	})
 
-	if containerSampler.Enabled() {
-		logger.Info("A container runtime is enabled, checking for containerized agent")
-		var dc metrics.ProcessDecorator
-		dc, err = containerSampler.NewDecorator()
-		if err != nil {
-			return
+	for _, containerSampler := range containerSamplers {
+		if containerSampler.Enabled() {
+			logger.Info("A container runtime is enabled, checking for containerized agent")
+			var dc metrics.ProcessDecorator
+			dc, err = containerSampler.NewDecorator()
+			if err != nil {
+				return
+			}
+			dc.Decorate(p)
 		}
-		dc.Decorate(p)
 	}
 
 	logger.WithField("containerID", p.ContainerID).Info("Containerized agent found in container")