Adding a custom collector which can collect metrics from do-agent sel…

cmd/do-agent/config.go

@@ -372,6 +372,9 @@ func initCollectors() []prometheus.Collector {
 			log.Debug("node_exporter collector registered %q", name)
 		}
 		cols = append(cols, node)
+		// Fills in sidecar volume mounts (e.g. /pgdata) that node_exporter
+		// misses because it reads /proc/1/mounts (pause) instead of self.
+		cols = append(cols, collector.NewSelfFilesystemCollector())
 	}
 
 	return cols

pkg/collector/filesystem_self.go

@@ -0,0 +1,150 @@
+package collector
+
+import (
+	"bufio"
+	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+	"regexp"
+	"strings"
+
+	"github.com/digitalocean/do-agent/internal/log"
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+// node_exporter reads /proc/1/mounts. In a shared-PID-namespace sidecar that
+// is the pause container, which does not have volume mounts like /pgdata.
+// This collector emits the same node_filesystem_* series for mounts visible
+// in /proc/self/mounts that PID 1 does not have.
+
+var (
+	selfFSSizeDesc = prometheus.NewDesc(
+		"node_filesystem_size_bytes",
+		"Filesystem size in bytes.",
+		[]string{"device", "fstype", "mountpoint"},
+		nil,
+	)
+	selfFSFreeDesc = prometheus.NewDesc(
+		"node_filesystem_free_bytes",
+		"Filesystem free space in bytes.",
+		[]string{"device", "fstype", "mountpoint"},
+		nil,
+	)
+
+	// Keep in sync with cmd/do-agent/config_filesystem.go so we do not emit
+	// tmpfs/overlay/proc junk the node collector already ignores.
+	selfIgnoredMountPoints = regexp.MustCompile(`^/(rootfs/)?(boot|sys|proc|dev|host|etc|tmp|usr/(home|ports|src)|var/(audit|crash|log|mail|tmp)|var/(lib|run)/docker/[^$]+|run/docker/[^$]+)($$|/)`)
+	selfIgnoredFSTypes     = regexp.MustCompile(`^(aufs|autofs|binfmt_misc|cd9660|cifs|cgroup|debugfs|devpts|devtmpfs|ecryptfs|efivarfs|fuse|hugetlbfs|mqueue|nfs|overlay|overlayfs|proc|pstore|rpc_pipefs|securityfs|smb|sysfs|tmpfs|tracefs|squashfs|nsfs)$`)
+)
+
+type fsMount struct {
+	device     string
+	mountPoint string
+	fsType     string
+}
+
+type statfsFunc func(path string) (size, free float64, err error)
+
+// SelfFilesystemCollector reports filesystem metrics from this process's mount
+// table when they are missing from PID 1's mount table.
+type SelfFilesystemCollector struct {
+	procfs string
+	statfs statfsFunc
+}
+
+// NewSelfFilesystemCollector creates a collector that fills in sidecar volume
+// mounts node_exporter misses (for example /pgdata on Advanced PG).
+func NewSelfFilesystemCollector() *SelfFilesystemCollector {
+	return &SelfFilesystemCollector{
+		procfs: "/proc",
+		statfs: defaultStatfs,
+	}
+}
+
+// Describe implements prometheus.Collector.
+func (c *SelfFilesystemCollector) Describe(ch chan<- *prometheus.Desc) {
+	ch <- selfFSSizeDesc
+	ch <- selfFSFreeDesc
+}
+
+// Collect implements prometheus.Collector.
+func (c *SelfFilesystemCollector) Collect(ch chan<- prometheus.Metric) {
+	pid1, err := readMounts(filepath.Join(c.procfs, "1", "mounts"))
+	if err != nil {
+		// Missing /proc/1/mounts (hidepid) is the case node_exporter already
+		// falls back from. Emitting self mounts would duplicate those series.
+		log.Debug("self filesystem: skip, cannot read pid1 mounts: %v", err)
+		return
+	}
+	self, err := readMounts(filepath.Join(c.procfs, "self", "mounts"))
+	if err != nil {
+		log.Error("self filesystem: cannot read self mounts: %v", err)
+		return
+	}
+
+	for _, m := range extraMounts(self, pid1) {
+		if selfIgnoredMountPoints.MatchString(m.mountPoint) || selfIgnoredFSTypes.MatchString(m.fsType) {
+			continue
+		}
+		size, free, err := c.statfs(m.mountPoint)
+		if err != nil {
+			log.Debug("self filesystem: statfs %s: %v", m.mountPoint, err)
+			continue
+		}
+		ch <- prometheus.MustNewConstMetric(selfFSSizeDesc, prometheus.GaugeValue, size, m.device, m.fsType, m.mountPoint)
+		ch <- prometheus.MustNewConstMetric(selfFSFreeDesc, prometheus.GaugeValue, free, m.device, m.fsType, m.mountPoint)
+	}
+}
+
+func readMounts(path string) ([]fsMount, error) {
+	f, err := os.Open(path)
+	if err != nil {
+		return nil, err
+	}
+	defer f.Close()
+	return parseMounts(f)
+}
+
+func parseMounts(r io.Reader) ([]fsMount, error) {
+	var mounts []fsMount
+	sc := bufio.NewScanner(r)
+	for sc.Scan() {
+		parts := strings.Fields(sc.Text())
+		if len(parts) < 3 {
+			return nil, fmt.Errorf("malformed mount line: %q", sc.Text())
+		}
+		mounts = append(mounts, fsMount{
+			device:     parts[0],
+			mountPoint: unescapeMount(parts[1]),
+			fsType:     parts[2],
+		})
+	}
+	return mounts, sc.Err()
+}
+
+func extraMounts(self, pid1 []fsMount) []fsMount {
+	seenPID1 := make(map[string]struct{}, len(pid1))
+	for _, m := range pid1 {
+		seenPID1[m.mountPoint] = struct{}{}
+	}
+	seenSelf := make(map[string]struct{})
+	extra := make([]fsMount, 0, len(self))
+	for _, m := range self {
+		if _, ok := seenPID1[m.mountPoint]; ok {
+			continue
+		}
+		if _, ok := seenSelf[m.mountPoint]; ok {
+			continue
+		}
+		seenSelf[m.mountPoint] = struct{}{}
+		extra = append(extra, m)
+	}
+	return extra
+}
+
+func unescapeMount(p string) string {
+	p = strings.ReplaceAll(p, `\040`, " ")
+	p = strings.ReplaceAll(p, `\011`, "\t")
+	return p
+}

pkg/collector/filesystem_self_linux.go

@@ -0,0 +1,14 @@
+//go:build linux
+
+package collector
+
+import "golang.org/x/sys/unix"
+
+func defaultStatfs(path string) (size, free float64, err error) {
+	var buf unix.Statfs_t
+	if err := unix.Statfs(path, &buf); err != nil {
+		return 0, 0, err
+	}
+	bsize := float64(buf.Bsize)
+	return float64(buf.Blocks) * bsize, float64(buf.Bfree) * bsize, nil
+}

pkg/collector/filesystem_self_other.go

@@ -0,0 +1,9 @@
+//go:build !linux
+
+package collector
+
+import "fmt"
+
+func defaultStatfs(path string) (size, free float64, err error) {
+	return 0, 0, fmt.Errorf("statfs not supported on this platform: %s", path)
+}

pkg/collector/filesystem_self_test.go

@@ -0,0 +1,107 @@
+package collector
+
+import (
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestParseMounts(t *testing.T) {
+	in := strings.NewReader(strings.Join([]string{
+		`/dev/vda1 / ext4 rw,relatime 0 0`,
+		`/dev/disk/by-id/scsi-0DO_Volume_pvc-abc /pgdata ext4 ro,relatime 0 0`,
+		`tmpfs /tmp tmpfs rw 0 0`,
+	}, "\n") + "\n")
+
+	got, err := parseMounts(in)
+	require.NoError(t, err)
+	assert.Equal(t, []fsMount{
+		{device: "/dev/vda1", mountPoint: "/", fsType: "ext4"},
+		{device: "/dev/disk/by-id/scsi-0DO_Volume_pvc-abc", mountPoint: "/pgdata", fsType: "ext4"},
+		{device: "tmpfs", mountPoint: "/tmp", fsType: "tmpfs"},
+	}, got)
+}
+
+func TestExtraMountsSkipsPID1(t *testing.T) {
+	pid1 := []fsMount{{device: "/dev/vda1", mountPoint: "/", fsType: "ext4"}}
+	self := []fsMount{
+		{device: "/dev/vda1", mountPoint: "/", fsType: "ext4"},
+		{device: "/dev/disk/by-id/scsi-0DO_Volume_pvc-abc", mountPoint: "/pgdata", fsType: "ext4"},
+		{device: "tmpfs", mountPoint: "/tmp", fsType: "tmpfs"},
+	}
+
+	assert.Equal(t, []fsMount{
+		{device: "/dev/disk/by-id/scsi-0DO_Volume_pvc-abc", mountPoint: "/pgdata", fsType: "ext4"},
+		{device: "tmpfs", mountPoint: "/tmp", fsType: "tmpfs"},
+	}, extraMounts(self, pid1))
+}
+
+func TestExtraMountsEmptyWhenSameAsPID1(t *testing.T) {
+	same := []fsMount{
+		{device: "/dev/vda1", mountPoint: "/", fsType: "ext4"},
+		{device: "/dev/sda", mountPoint: "/mnt/volume", fsType: "ext4"},
+	}
+	assert.Empty(t, extraMounts(same, same))
+}
+
+func TestSelfFilesystemCollectEmitsPgdata(t *testing.T) {
+	dir := t.TempDir()
+	require.NoError(t, writeMounts(dir, "1", "/dev/vda1 / ext4 rw 0 0\n"))
+	require.NoError(t, writeMounts(dir, "self", strings.Join([]string{
+		`/dev/vda1 / ext4 rw 0 0`,
+		`/dev/disk/by-id/scsi-0DO_Volume_pvc-abc /pgdata ext4 ro 0 0`,
+		`tmpfs /tmp tmpfs rw 0 0`,
+		`overlay /run overlay rw 0 0`,
+	}, "\n")+"\n"))
+
+	c := &SelfFilesystemCollector{
+		procfs: dir,
+		statfs: func(path string) (size, free float64, err error) {
+			assert.Equal(t, "/pgdata", path)
+			return 100, 40, nil
+		},
+	}
+
+	ch := make(chan prometheus.Metric, 8)
+	c.Collect(ch)
+	close(ch)
+
+	var descs []string
+	for m := range ch {
+		descs = append(descs, m.Desc().String())
+	}
+	require.Len(t, descs, 2)
+	assert.Contains(t, descs[0], `fqName: "node_filesystem_size_bytes"`)
+	assert.Contains(t, descs[1], `fqName: "node_filesystem_free_bytes"`)
+}
+
+func TestSelfFilesystemCollectSkipsWhenPID1Missing(t *testing.T) {
+	dir := t.TempDir()
+	require.NoError(t, writeMounts(dir, "self", "/dev/vda1 / ext4 rw 0 0\n"))
+
+	c := &SelfFilesystemCollector{
+		procfs: dir,
+		statfs: func(string) (float64, float64, error) {
+			t.Fatal("statfs should not run when pid1 mounts are missing")
+			return 0, 0, nil
+		},
+	}
+
+	ch := make(chan prometheus.Metric, 8)
+	c.Collect(ch)
+	close(ch)
+	assert.Empty(t, ch)
+}
+
+func writeMounts(procfs, name, body string) error {
+	dir := filepath.Join(procfs, name)
+	if err := os.MkdirAll(dir, 0o755); err != nil {
+		return err
+	}
+	return os.WriteFile(filepath.Join(dir, "mounts"), []byte(body), 0o644)
+}