feat: Add OpenPBS collector

.golangci.yml

@@ -83,6 +83,7 @@ linters:
       - third_party$
       - builtin$
       - examples$
+      - website$
 formatters:
   enable:
     - gci
@@ -95,3 +96,4 @@ formatters:
       - third_party$
       - builtin$
       - examples$
+      - website$

CITATION.cff

@@ -22,7 +22,6 @@ preferred-citation:
       orcid: "https://orcid.org/0000-0002-1300-0216"
   booktitle: "SC24-W: Workshops of the International Conference for High Performance Computing, Networking, Storage and Analysis"
   series: "Proceedings of SC24-W: Workshops of the International Conference for High Performance Computing, Networking, Storage and Analysis"
-  volume:
   pages: "1862-1866"
   year: 2024
   url: "https://ieeexplore.ieee.org/document/10820816"

Makefile

@@ -170,6 +170,7 @@ test-e2e: $(PROMTOOL) build pkg/collector/testdata/sys/.unpacked pkg/collector/t
 	./scripts/e2e-test.sh -s exporter-cgroups-v2-all-metrics
 	./scripts/e2e-test.sh -s exporter-cgroups-v1-lsf
 	./scripts/e2e-test.sh -s exporter-cgroups-v2-lsf
+	./scripts/e2e-test.sh -s exporter-cgroups-v1-openpbs
 	./scripts/e2e-test.sh -s exporter-cgroups-v1-libvirt
 	./scripts/e2e-test.sh -s exporter-cgroups-v2-libvirt
 	./scripts/e2e-test.sh -s exporter-cgroups-v2-libvirt-nonsystemd-layout
@@ -266,6 +267,7 @@ test-e2e-update: build pkg/collector/testdata/sys/.unpacked pkg/collector/testda
 	./scripts/e2e-test.sh -s exporter-cgroups-v2-all-metrics -u || true
 	./scripts/e2e-test.sh -s exporter-cgroups-v1-lsf -u || true
 	./scripts/e2e-test.sh -s exporter-cgroups-v2-lsf -u || true
+	./scripts/e2e-test.sh -s exporter-cgroups-v1-openpbs -u || true
 	./scripts/e2e-test.sh -s exporter-cgroups-v1-libvirt -u || true
 	./scripts/e2e-test.sh -s exporter-cgroups-v2-libvirt -u || true
 	./scripts/e2e-test.sh -s exporter-cgroups-v2-libvirt-nonsystemd-layout -u || true

pkg/collector/cgroup.go

@@ -42,7 +42,8 @@ const (
 
 // Custom errors.
 var (
-	errUnknownManager = errors.New("unknown resource manager")
+	errUnknownManager     = errors.New("unknown resource manager")
+	errUnsupportedCgroups = errors.New("cgroups version not supported")
 )
 
 type manager int
@@ -52,6 +53,7 @@ const (
 	_ manager = iota
 	slurm
 	lsf
+	openpbs
 	libvirt
 	k8s
 )
@@ -60,6 +62,7 @@ const (
 var rmNames = map[manager]string{
 	slurm:   "slurm",
 	lsf:     "lsf",
+	openpbs: "openpbs",
 	libvirt: "libvirt",
 	k8s:     "k8s",
 }
@@ -119,6 +122,27 @@ var (
 	lsfIgnoreProcsRegex  = regexp.MustCompile("(.*)/.lsbatch/(.*)")
 )
 
+// Regular expressions of cgroup paths for Open PBS.
+// ^.*/(?:(.*?)_)?slurm(?:_(.*?)/)?(?:.*?)/job_([0-9]+)(?:.*$)
+// ^.*/slurm(?:_(.*?))?/(?:.*?)/job_([0-9]+)(?:.*$)
+/*
+	For v1 possibilities are /cpuacct/pbs_jobs.service/jobid/2025.ceems-open-pbs/
+							 /memory/pbs_jobs.service/jobid/2025.ceems-open-pbs/
+							 /memory/pbs_jobs.service/jobid/2025\[0\].ceems-open-pbs/ for job arrays
+
+	Currently Open PBS supports only cgroups v1.
+
+	Ignored processes are of form:
+
+	/var/spool/pbs/mom_priv/jobs/2027[0].ceems-open-pbs.SC
+	/bin/bash /home/usr1/.lsbatch/1775296755.13
+	/bin/bash /home/usr1/.lsbatch/1775296755.13.shell
+*/
+var (
+	openpbsCgroupV1PathRegex = regexp.MustCompile(`^.*/(?:(?P<cgroups_prefix>.*?))/jobid/(?P<id>[^/]+)(?:.*)$`)
+	openpbsIgnoreProcsRegex  = regexp.MustCompile("(.*)/mom_priv/(.*)")
+)
+
 // Ref: https://libvirt.org/cgroups.html#legacy-cgroups-layout
 // Take escaped unicode characters in regex
 /*
@@ -387,6 +411,51 @@ func NewCgroupManager(name manager, logger *slog.Logger) (*cgroupManager, error)
 
 		return manager, nil
 
+	case openpbs:
+		if (*forceCgroupsVersion == "" && cgroups.Mode() == cgroups.Unified) || *forceCgroupsVersion == "v2" {
+			logger.Error("OpenPBS does not support cgroups v2", "err", errUnsupportedCgroups)
+
+			return nil, fmt.Errorf("openpbs does not support cgroups v2: %w", errUnsupportedCgroups)
+		}
+
+		// cgroups v1 or hybrid mode
+		var mode cgroups.CGMode
+		if *forceCgroupsVersion == "v1" {
+			mode = cgroups.Legacy
+		} else {
+			mode = cgroups.Mode()
+		}
+
+		// Resolve subsystem
+		activeSubsystem := resolveSubsystem(*activeController)
+
+		manager = &cgroupManager{
+			logger:           logger,
+			fs:               fs,
+			mode:             mode,
+			root:             *cgroupfsPath,
+			idRegex:          openpbsCgroupV1PathRegex,
+			activeController: activeSubsystem,
+			slices:           []string{"pbs_jobs.service"},
+		}
+
+		// Add manager field
+		manager.id = name
+		manager.name = rmNames[name]
+
+		// For the moment, we are not sure if there are any child cgroups created
+		manager.isChild = func(_ string) bool {
+			return false
+		}
+		manager.ignoreProc = func(p string) bool {
+			return openpbsIgnoreProcsRegex.MatchString(p)
+		}
+
+		// Set mountpoints
+		manager.setMountPoints()
+
+		return manager, nil
+
 	case libvirt:
 		if (*forceCgroupsVersion == "" && cgroups.Mode() == cgroups.Unified) || *forceCgroupsVersion == "v2" {
 			manager = &cgroupManager{
@@ -642,6 +711,14 @@ func (c *cgroupManager) setMountPoints() {
 			// For cgroups v1 we need to shift root to /sys/fs/cgroup/cpuacct
 			c.root = filepath.Join(c.root, c.activeController)
 		}
+	case openpbs:
+		// /sys/fs/cgroup/cpuacct/pbs_jobs.service
+		for _, slice := range c.slices {
+			c.mountPoints = append(c.mountPoints, filepath.Join(c.root, c.activeController, slice))
+		}
+
+		// For cgroups v1 we need to shift root to /sys/fs/cgroup/cpuacct
+		c.root = filepath.Join(c.root, c.activeController)
 	case libvirt:
 		switch c.mode { //nolint: exhaustive
 		case cgroups.Unified:
@@ -1401,7 +1478,7 @@ func (c *cgroupCollector) cpusFromShares(path string) (int, error) {
 // getCPUs returns number of milli CPUs in the cgroup.
 func (c *cgroupCollector) getCPUs(path string, ncpusDefault int) (int, error) {
 	switch c.cgroupManager.id {
-	case slurm:
+	case slurm, openpbs:
 		return c.cpusFromCPUSet(path)
 	case lsf:
 		return c.cpusFromCPUSetUsingDefault(path, ncpusDefault)
@@ -1739,5 +1816,5 @@ func parseCgroupSubSysIds() ([]cgroupController, error) {
 		idx++
 	}
 
-	return cgroupControllers, nil
+	return cgroupControllers, fscanner.Err()
 }

pkg/collector/cgroup_test.go

@@ -107,6 +107,14 @@ func TestNewCgroupManagerV1(t *testing.T) {
 			mountPoints: []string{"testdata/sys/fs/cgroup/cpu,cpuacct/lsf"},
 			numCgroups:  3,
 		},
+		{
+			name:    "openpbs",
+			manager: openpbs,
+			mountPoints: []string{
+				"testdata/sys/fs/cgroup/cpu,cpuacct/pbs_jobs.service",
+			},
+			numCgroups: 3,
+		},
 		{
 			name:        "libvirt",
 			manager:     libvirt,
@@ -391,7 +399,7 @@ func TestCgroupsV1Metrics(t *testing.T) {
 			name:    "slurm",
 			manager: slurm,
 			expected: cgMetric{
-				cgroup:          cgroup{uuid: "1009249", path: cgroupPath{rel: "/slurm/uid_1000/job_1009248"}},
+				cgroup:          cgroup{uuid: "1009248", path: cgroupPath{rel: "/slurm/uid_1000/job_1009248"}},
 				cpuUser:         0.39,
 				cpuSystem:       0.45,
 				cpuTotal:        1.012410966,
@@ -433,6 +441,28 @@ func TestCgroupsV1Metrics(t *testing.T) {
 				err:             false,
 			},
 		},
+		{
+			name:    "openpbs",
+			manager: openpbs,
+			expected: cgMetric{
+				cgroup:          cgroup{uuid: "1009249[1].testhost-1", path: cgroupPath{rel: "/pbs_jobs.service/jobid/1009249[1].testhost-1"}},
+				cpuUser:         2483.49,
+				cpuSystem:       0.1,
+				cpuTotal:        2483.616433633,
+				cpus:            5000,
+				cpuPressure:     0,
+				memoryRSS:       1.0420224e+07,
+				memoryCache:     0,
+				memoryUsed:      1.5839232e+07,
+				memoryTotal:     9.223372036854772e+18,
+				memoryFailCount: 0,
+				memswUsed:       0,
+				memswTotal:      0,
+				memswFailCount:  0,
+				memoryPressure:  0,
+				err:             false,
+			},
+		},
 		{
 			name:    "libvirt",
 			manager: libvirt,

pkg/collector/gpu.go

@@ -1708,7 +1708,7 @@ func parseTopologyProperties(path string, re *regexp.Regexp) (uint64, error) {
 		return strconv.ParseUint(match[1], 10, 64)
 	}
 
-	return 0, fmt.Errorf("no property found in the file match %s", re.String())
+	return 0, fmt.Errorf("no property found in the file match %s: %w", re.String(), scanner.Err())
 }
 
 // parseBusID parses PCIe bus ID string to BusID struct.

pkg/collector/helper_test.go

@@ -262,10 +262,10 @@ func TestReadProcEnvirons(t *testing.T) {
 			name: "Existing env vars without proc filter",
 			dataPtr: &readProcSecurityCtxData{
 				procs:         procs,
-				targetEnvVars: []string{"CUDA_VISIBLE_DEVICES", "CUDA_VISIBLE_DEVICES1"},
+				targetEnvVars: []string{"GOLANG_VERSION", "CUDA_VISIBLE_DEVICES2"},
 			},
 			expected: map[string]string{
-				"CUDA_VISIBLE_DEVICES1": "MIG-GPU-956348bc-d43d-23ed-53d4-857749fa2b67/1/0,MIG-GPU-feba7e40-d724-01ff-b00f-3a439a28a6c7/1/0",
+				"GOLANG_VERSION": "1.12.5",
 			},
 		},
 		{

pkg/collector/lsf.go

@@ -446,7 +446,8 @@ func (c *lsfCollector) jobResources(cgroups []cgroup) {
 		}
 	}
 
-	// If no MIG devices found, nothing to do here
+	// If MIG is enabled, we need privileges to read proc environ files to get MIG
+	// device IDs
 	if nvidiaMIGEnabled && len(c.securityContexts) == 0 {
 		c.logger.Warn("GPU Instances enabled but collector does not have enough permissions to fetch job MIG devices. Please restart the exporter")
 	}