docker_util.go

// Unless explicitly stated otherwise all files in this repository are licensed
// under the Apache License Version 2.0.
// This product includes software developed at Datadog (https://www.datadoghq.com/).
// Copyright 2016-present Datadog, Inc.

//go:build docker

package docker

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"

	workloadfilter "github.com/DataDog/datadog-agent/comp/core/workloadfilter/def"
	workloadmetafilter "github.com/DataDog/datadog-agent/comp/core/workloadfilter/util/workloadmeta"
	workloadmeta "github.com/DataDog/datadog-agent/comp/core/workloadmeta/def"
	pkgconfigsetup "github.com/DataDog/datadog-agent/pkg/config/setup"
	dderrors "github.com/DataDog/datadog-agent/pkg/errors"
	"github.com/DataDog/datadog-agent/pkg/util/cache"
	"github.com/DataDog/datadog-agent/pkg/util/log"
	"github.com/DataDog/datadog-agent/pkg/util/retry"
	cerrdefs "github.com/containerd/errdefs"
	dcontainer "github.com/moby/moby/api/types/container"
	"github.com/moby/moby/api/types/image"
	"github.com/moby/moby/client"
)

// DockerUtil wraps interactions with a local docker API.
type DockerUtil struct {
	// used to setup the DockerUtil
	initRetry retry.Retrier

	sync.Mutex
	cfg          *Config
	cli          *client.Client
	queryTimeout time.Duration
	// tracks the last time we invalidate our internal caches
	lastInvalidate time.Time
	// image sha mapping cache
	imageNameBySha map[string]string
	// event subscribers and state
	eventState *eventStreamState
}

// init makes an empty DockerUtil bootstrap itself.
// This is not exposed as public API but is called by the retrier embed.
func (d *DockerUtil) init() error {
	d.queryTimeout = pkgconfigsetup.Datadog().GetDuration("docker_query_timeout") * time.Second

	// Major failure risk is here, do that first
	ctx, cancel := context.WithTimeout(context.Background(), d.queryTimeout)
	defer cancel()
	cli, err := ConnectToDocker(ctx)
	if err != nil {
		return err
	}

	cfg := &Config{
		// TODO: bind them to config entries if relevant
		CollectNetwork: true,
		CacheDuration:  10 * time.Second,
	}

	d.cfg = cfg
	d.cli = cli
	d.imageNameBySha = make(map[string]string)
	d.lastInvalidate = time.Now()
	d.eventState = newEventStreamState()

	return nil
}

// ConnectToDocker connects to docker and negotiates the API version
func ConnectToDocker(ctx context.Context) (*client.Client, error) {
	cli, err := client.New(client.FromEnv)
	if err != nil {
		return nil, err
	}
	// client.New does not actually contact the daemon. Force a round-trip
	// to verify availability. safeInfo tolerates daemons that emit invalid
	// CIDRs in /info's DefaultAddressPools, which would otherwise fail the
	// strict netip.Prefix decoding introduced by the moby v29 client.
	if _, err := safeInfo(ctx, cli); err != nil {
		return nil, err
	}

	log.Debugf("Successfully connected to Docker server")

	return cli, nil
}

// Images returns a slice of all images.
func (d *DockerUtil) Images(ctx context.Context, includeIntermediate bool) ([]image.Summary, error) {
	ctx, cancel := context.WithTimeout(ctx, d.queryTimeout)
	defer cancel()
	result, err := d.cli.ImageList(ctx, client.ImageListOptions{All: includeIntermediate})
	if err != nil {
		return nil, fmt.Errorf("unable to list docker images: %s", err)
	}
	return result.Items, nil
}

// CountVolumes returns the number of attached and dangling volumes.
func (d *DockerUtil) CountVolumes(ctx context.Context) (int, int, error) {
	attachedFilter, _ := buildDockerFilter("dangling", "false")
	danglingFilter, _ := buildDockerFilter("dangling", "true")
	ctx, cancel := context.WithTimeout(ctx, d.queryTimeout)
	defer cancel()

	attachedVolumes, err := d.cli.VolumeList(ctx, attachedFilter)
	if err != nil {
		return 0, 0, fmt.Errorf("unable to list attached docker volumes: %s", err)
	}
	danglingVolumes, err := d.cli.VolumeList(ctx, danglingFilter)
	if err != nil {
		return 0, 0, fmt.Errorf("unable to list dangling docker volumes: %s", err)
	}

	return len(attachedVolumes.Items), len(danglingVolumes.Items), nil
}

// RawClient returns the underlying docker client being used by this object.
func (d *DockerUtil) RawClient() *client.Client {
	return d.cli
}

// RawContainerList wraps around the docker client's ContainerList method.
// Value validation and error handling are the caller's responsibility.
func (d *DockerUtil) RawContainerList(ctx context.Context, options client.ContainerListOptions) ([]dcontainer.Summary, error) {
	ctx, cancel := context.WithTimeout(ctx, d.queryTimeout)
	defer cancel()
	result, err := d.cli.ContainerList(ctx, options)
	if err != nil {
		return nil, err
	}
	return result.Items, nil
}

// RawContainerListWithFilter is like RawContainerList but with a container filter.
func (d *DockerUtil) RawContainerListWithFilter(ctx context.Context, options client.ContainerListOptions, filter workloadfilter.FilterBundle, wmeta workloadmeta.Component) ([]dcontainer.Summary, error) {
	containers, err := d.RawContainerList(ctx, options)
	if err != nil {
		return nil, err
	}

	if filter == nil {
		return containers, nil
	}

	isExcluded := func(container dcontainer.Summary) bool {
		pod, _ := wmeta.GetKubernetesPodForContainer(container.ID)
		filterablePod := workloadmetafilter.CreatePod(pod)
		for _, name := range container.Names {
			filterableContainer := workloadfilter.CreateContainer(container.ID, name, container.Image, filterablePod)
			if filter.IsExcluded(filterableContainer) {
				log.Tracef("Container with ID %q and image %q is filtered-out", container.ID, container.Image)
				return true
			}
		}

		return false
	}

	filtered := []dcontainer.Summary{}
	for _, container := range containers {
		if !isExcluded(container) {
			filtered = append(filtered, container)
		}
	}

	return filtered, nil
}

// GetHostname returns the hostname from the docker api
func (d *DockerUtil) GetHostname(ctx context.Context) (string, error) {
	ctx, cancel := context.WithTimeout(ctx, d.queryTimeout)
	defer cancel()
	info, err := safeInfo(ctx, d.cli)
	if err != nil {
		return "", fmt.Errorf("unable to get Docker info: %s", err)
	}
	return info.Name, nil
}

// GetStorageStats returns the docker global storage stats if available
// or ErrStorageStatsNotAvailable
func (d *DockerUtil) GetStorageStats(ctx context.Context) ([]*StorageStats, error) {
	ctx, cancel := context.WithTimeout(ctx, d.queryTimeout)
	defer cancel()
	info, err := safeInfo(ctx, d.cli)
	if err != nil {
		return []*StorageStats{}, fmt.Errorf("unable to get Docker info: %s", err)
	}
	return parseStorageStatsFromInfo(info)
}

func isImageShaOrRepoDigest(image string) bool {
	return strings.HasPrefix(image, "sha256:") || strings.Contains(image, "@sha256:")
}

// ResolveImageName will resolve sha image name to their user-friendly name.
// For non-sha/non-repodigest names we will just return the name as-is.
func (d *DockerUtil) ResolveImageName(ctx context.Context, image string) (string, error) {
	if !isImageShaOrRepoDigest(image) {
		return image, nil
	}

	d.Lock()
	if preferredName, found := d.imageNameBySha[image]; found {
		d.Unlock()
		return preferredName, nil
	}

	ctx, cancel := context.WithTimeout(ctx, d.queryTimeout)
	defer cancel()
	r, err := d.cli.ImageInspect(ctx, image)
	if err != nil {
		// Only log errors that aren't "not found" because some images may
		// just not be available in docker inspect.
		if !cerrdefs.IsNotFound(err) {
			d.Unlock()
			return image, err
		}
		d.imageNameBySha[image] = image
	}

	d.Unlock()
	return d.GetPreferredImageName(r.InspectResponse.ID, r.InspectResponse.RepoTags, r.InspectResponse.RepoDigests), nil
}

// GetPreferredImageName returns preferred image name based on RepoTags and RepoDigests
func (d *DockerUtil) GetPreferredImageName(imageID string, repoTags []string, repoDigests []string) string {
	d.Lock()
	defer d.Unlock()

	if preferredName, found := d.imageNameBySha[imageID]; found {
		return preferredName
	}

	var preferredName string
	// Try RepoTags first and fall back to RepoDigest otherwise.
	if len(repoTags) > 0 {
		sort.Strings(repoTags)
		preferredName = repoTags[0]
	} else if len(repoDigests) > 0 {
		// Digests formatted like quay.io/foo/bar@sha256:hash
		sort.Strings(repoDigests)
		sp := strings.SplitN(repoDigests[0], "@", 2)
		preferredName = sp[0]
	} else {
		preferredName = imageID
	}

	d.imageNameBySha[imageID] = preferredName
	return preferredName
}

// ImageInspect returns an image inspect object for a given image ID
func (d *DockerUtil) ImageInspect(ctx context.Context, imageID string) (image.InspectResponse, error) {
	ctx, cancel := context.WithTimeout(ctx, d.queryTimeout)
	defer cancel()

	result, err := d.cli.ImageInspect(ctx, imageID)
	if err != nil {
		return result.InspectResponse, fmt.Errorf("error inspecting image: %w", err)
	}

	return result.InspectResponse, nil
}

// ImageHistory returns the history for a given image ID
func (d *DockerUtil) ImageHistory(ctx context.Context, imageID string) ([]image.HistoryResponseItem, error) {
	ctx, cancel := context.WithTimeout(ctx, d.queryTimeout)
	defer cancel()

	result, err := d.cli.ImageHistory(ctx, imageID)
	if err != nil {
		return result.Items, fmt.Errorf("error getting image history: %w", err)
	}

	return result.Items, nil
}

// ResolveImageNameFromContainer will resolve the container sha image name to their user-friendly name.
// It is similar to ResolveImageName except it tries to match the image to the container Config.Image.
// For non-sha names we will just return the name as-is.
func (d *DockerUtil) ResolveImageNameFromContainer(ctx context.Context, co dcontainer.InspectResponse) (string, error) {
	if co.Config.Image != "" && !isImageShaOrRepoDigest(co.Config.Image) {
		return co.Config.Image, nil
	}

	return d.ResolveImageName(ctx, co.Image)
}

// Inspect returns a docker inspect object for a given container ID.
// It tries to locate the container in the inspect cache before making the docker inspect call
func (d *DockerUtil) Inspect(ctx context.Context, id string, withSize bool) (dcontainer.InspectResponse, error) {
	cacheKey := GetInspectCacheKey(id, withSize)
	var container dcontainer.InspectResponse

	cached, hit := cache.Cache.Get(cacheKey)
	// Try to get sized hit if we got a miss and withSize=false
	if !hit && !withSize {
		cached, hit = cache.Cache.Get(GetInspectCacheKey(id, true))
	}

	if hit {
		container, ok := cached.(dcontainer.InspectResponse)
		if !ok {
			log.Errorf("Invalid inspect cache format, forcing a cache miss")
		} else {
			return container, nil
		}
	}

	container, err := d.InspectNoCache(ctx, id, withSize)
	if err != nil {
		return container, err
	}

	// cache the inspect for 10 seconds to reduce pressure on the daemon
	cache.Cache.Set(cacheKey, container, 10*time.Second)

	return container, nil
}

// InspectNoCache returns a docker inspect object for a given container ID. It
// ignores the inspect cache, always collecting fresh data from the docker
// daemon.
func (d *DockerUtil) InspectNoCache(ctx context.Context, id string, withSize bool) (dcontainer.InspectResponse, error) {
	ctx, cancel := context.WithTimeout(ctx, d.queryTimeout)
	defer cancel()

	result, err := d.cli.ContainerInspect(ctx, id, client.ContainerInspectOptions{Size: withSize})
	container := result.Container
	if cerrdefs.IsNotFound(err) {
		return container, dderrors.NewNotFound("docker container " + id)
	}
	if err != nil {
		return container, err
	}

	// Check for empty inspect data
	if container.ID == "" {
		return container, errors.New("invalid inspect data")
	}

	return container, nil
}

// AllContainerLabels retrieves all running containers (`docker ps`) and returns
// a map mapping containerID to container labels as a map[string]string
func (d *DockerUtil) AllContainerLabels(ctx context.Context) (map[string]map[string]string, error) {
	ctx, cancel := context.WithTimeout(ctx, d.queryTimeout)
	defer cancel()
	result, err := d.cli.ContainerList(ctx, client.ContainerListOptions{})
	if err != nil {
		return nil, fmt.Errorf("error listing containers: %s", err)
	}
	containers := result.Items

	labelMap := make(map[string]map[string]string)

	for _, container := range containers {
		if len(container.ID) == 0 {
			continue
		}
		labelMap[container.ID] = container.Labels
	}

	return labelMap, nil
}

// GetContainerStats returns docker container stats
func (d *DockerUtil) GetContainerStats(ctx context.Context, containerID string) (*dcontainer.StatsResponse, error) {
	ctx, cancel := context.WithTimeout(ctx, d.queryTimeout)
	defer cancel()
	stats, err := d.cli.ContainerStats(ctx, containerID, client.ContainerStatsOptions{})
	if err != nil {
		return nil, fmt.Errorf("unable to get Docker stats: %s", err)
	}
	containerStats := &dcontainer.StatsResponse{}
	err = json.NewDecoder(stats.Body).Decode(&containerStats)
	if err != nil {
		return nil, fmt.Errorf("error listing containers: %s", err)
	}
	return containerStats, nil
}

// ContainerLogs returns a container logs reader
func (d *DockerUtil) ContainerLogs(ctx context.Context, container string, options client.ContainerLogsOptions) (io.ReadCloser, error) {
	return d.cli.ContainerLogs(ctx, container, options)
}

// GetContainerPIDs returns a list of containerID's running PIDs
func (d *DockerUtil) GetContainerPIDs(ctx context.Context, containerID string) ([]int, error) {

	// Index into the returned [][]string slice for process IDs
	pidIdx := -1

	// Docker API to collect PIDs associated with containerID
	procs, err := d.cli.ContainerTop(ctx, containerID, client.ContainerTopOptions{})
	if err != nil {
		return nil, fmt.Errorf("unable to get PIDs for container %s: %s", containerID, err)
	}

	// get the offset into the string[][] slice for the process ID index
	for idx, val := range procs.Titles {
		if val == "PID" {
			pidIdx = idx
			break
		}
	}
	if pidIdx == -1 {
		return nil, errors.New("unable to locate PID index into returned process slice")
	}

	// Create slice large enough to hold each PID
	pids := make([]int, len(procs.Processes))

	// Iterate returned Processes and pull out their PIDs
	for idx, entry := range procs.Processes {
		// Convert to ints
		pid, sterr := strconv.Atoi(entry[pidIdx])
		if sterr != nil {
			log.Debugf("unable to convert PID to int: %s", sterr)
			continue
		}
		pids[idx] = pid
	}
	return pids, nil
}