remove useless noderpc in vGPUmonitor

Author: zhitianli

cmd/vGPUmonitor/cudevshr.go

@@ -17,8 +17,6 @@ limitations under the License.
 package main
 
 import (
-	"bytes"
-	"encoding/binary"
 	"errors"
 	"fmt"
 	"os"
@@ -28,8 +26,6 @@ import (
 	"golang.org/x/exp/mmap"
 )
 
-const maxDevices = 16
-
 type deviceMemory struct {
 	contextSize uint64
 	moduleSize  uint64
@@ -72,57 +68,13 @@ type sharedRegionT struct {
 	priority          int32
 }
 
-type SharedRegionInfoT struct {
-	pid          int32
-	fd           int32
-	initStatus   int16
-	sharedRegion sharedRegionT
-}
-
 type nvidiaCollector struct {
 	// Exposed for testing
 	cudevshrPath string
 	at           *mmap.ReaderAt
 	cudaCache    *sharedRegionT
 }
 
-func setProcSlot(offset int64, at *mmap.ReaderAt) (shrregProcSlotT, error) {
-	temp := shrregProcSlotT{}
-	buff := make([]byte, 4)
-	at.ReadAt(buff, offset)
-	bytesbuffer := bytes.NewBuffer(buff)
-	binary.Read(bytesbuffer, binary.LittleEndian, &temp.pid)
-	var monitorused uint64
-	//fmt.Println("pid==", temp.pid, "buff=", buff)
-	buff = make([]byte, 8)
-	for i := 0; i < maxDevices; i++ {
-		at.ReadAt(buff, offset+8+8*int64(i))
-		bytesbuffer = bytes.NewBuffer(buff)
-		binary.Read(bytesbuffer, binary.LittleEndian, &temp.used[i])
-	}
-	for i := 0; i < maxDevices; i++ {
-		at.ReadAt(buff, offset+8+8*16+8*int64(i))
-		bytesbuffer = bytes.NewBuffer(buff)
-		binary.Read(bytesbuffer, binary.LittleEndian, &monitorused)
-		if monitorused > temp.used[i].total {
-			temp.used[i].total = monitorused
-		}
-	}
-	return temp, nil
-}
-
-func getDeviceUsedMemory(idx int, sharedregion sharedRegionT) (uint64, error) {
-	var sum uint64
-	sum = 0
-	if idx < 0 || idx > 16 {
-		return 0, errors.New("out of device idx")
-	}
-	for _, val := range sharedregion.procs {
-		sum += val.used[idx].total
-	}
-	return sum, nil
-}
-
 func mmapcachefile(filename string, nc *nvidiaCollector) error {
 	var m = &sharedRegionT{}
 	f, err := os.OpenFile(filename, os.O_RDWR, 0666)

cmd/vGPUmonitor/feedback.go

@@ -17,166 +17,20 @@ limitations under the License.
 package main
 
 import (
-	"errors"
-	"fmt"
-	"os"
-	"sort"
-	"strconv"
-	"strings"
-	"sync"
 	"time"
 
 	"github.com/NVIDIA/go-nvml/pkg/nvml"
-	corev1 "k8s.io/api/core/v1"
 	"k8s.io/klog/v2"
 )
 
-var cgroupDriver int
-
-type hostGPUPid struct {
-	hostGPUPid int
-	mtime      uint64
-}
-
 type UtilizationPerDevice []int
 
-var mutex sync.Mutex
 var srPodList map[string]podusage
 
 func init() {
 	srPodList = make(map[string]podusage)
 }
 
-func setcGgroupDriver() int {
-	// 1 for cgroupfs 2 for systemd
-	kubeletconfig, err := os.ReadFile("/hostvar/lib/kubelet/config.yaml")
-	if err != nil {
-		return 0
-	}
-	content := string(kubeletconfig)
-	pos := strings.LastIndex(content, "cgroupDriver:")
-	if pos < 0 {
-		return 0
-	}
-	if strings.Contains(content, "systemd") {
-		return 2
-	}
-	if strings.Contains(content, "cgroupfs") {
-		return 1
-	}
-	return 0
-}
-
-func getUsedGPUPid() ([]uint, nvml.Return) {
-	tmp := []nvml.ProcessInfo{}
-	count, err := nvml.DeviceGetCount()
-	if err != nvml.SUCCESS {
-		return []uint{}, err
-	}
-	for i := 0; i < count; i++ {
-		device, err := nvml.DeviceGetHandleByIndex(i)
-		if err != nvml.SUCCESS {
-			return []uint{}, err
-		}
-		ids, err := device.GetComputeRunningProcesses()
-		if err != nvml.SUCCESS {
-			return []uint{}, err
-		}
-		tmp = append(tmp, ids...)
-	}
-	result := make([]uint, 0)
-	m := make(map[uint]bool)
-	for _, v := range tmp {
-		if _, ok := m[uint(v.Pid)]; !ok {
-			result = append(result, uint(v.Pid))
-			m[uint(v.Pid)] = true
-		}
-	}
-	sort.Slice(tmp, func(i, j int) bool { return tmp[i].Pid > tmp[j].Pid })
-	return result, nvml.SUCCESS
-}
-
-func setHostPid(pod corev1.Pod, ctr corev1.ContainerStatus, sr *podusage) error {
-	var pids []string
-	mutex.Lock()
-	defer mutex.Unlock()
-
-	if cgroupDriver == 0 {
-		cgroupDriver = setcGgroupDriver()
-	}
-	if cgroupDriver == 0 {
-		return errors.New("can not identify cgroup driver")
-	}
-	usedGPUArray, err := getUsedGPUPid()
-	if err != nvml.SUCCESS {
-		return errors.New("get usedGPUID failed, ret:" + nvml.ErrorString(err))
-	}
-	if len(usedGPUArray) == 0 {
-		return nil
-	}
-	qos := strings.ToLower(string(pod.Status.QOSClass))
-	var filename string
-	if cgroupDriver == 1 {
-		/* Cgroupfs */
-		filename = fmt.Sprintf("/sysinfo/fs/cgroup/memory/kubepods/%s/pod%s/%s/tasks", qos, pod.UID, strings.TrimPrefix(ctr.ContainerID, "docker://"))
-	}
-	if cgroupDriver == 2 {
-		/* Systemd */
-		cgroupuid := strings.ReplaceAll(string(pod.UID), "-", "_")
-		filename = fmt.Sprintf("/sysinfo/fs/cgroup/systemd/kubepods.slice/kubepods-%s.slice/kubepods-%s-pod%s.slice/docker-%s.scope/tasks", qos, qos, cgroupuid, strings.TrimPrefix(ctr.ContainerID, "docker://"))
-	}
-	fmt.Println("filename=", filename)
-	content, ferr := os.ReadFile(filename)
-	if ferr != nil {
-		return ferr
-	}
-	pids = strings.Split(string(content), "\n")
-	hostPidArray := []hostGPUPid{}
-	for _, val := range pids {
-		tmp, _ := strconv.Atoi(val)
-		if tmp != 0 {
-			var stat os.FileInfo
-			var err error
-			if stat, err = os.Lstat(fmt.Sprintf("/proc/%v", tmp)); err != nil {
-				return err
-			}
-			mtime := stat.ModTime().Unix()
-			hostPidArray = append(hostPidArray, hostGPUPid{
-				hostGPUPid: tmp,
-				mtime:      uint64(mtime),
-			})
-		}
-	}
-	usedGPUHostArray := []hostGPUPid{}
-	for _, val := range usedGPUArray {
-		for _, hostpid := range hostPidArray {
-			if uint(hostpid.hostGPUPid) == val {
-				usedGPUHostArray = append(usedGPUHostArray, hostpid)
-			}
-		}
-	}
-	//fmt.Println("usedHostGPUArray=", usedGPUHostArray)
-	sort.Slice(usedGPUHostArray, func(i, j int) bool { return usedGPUHostArray[i].mtime > usedGPUHostArray[j].mtime })
-	if sr == nil || sr.sr == nil {
-		return nil
-	}
-	for idx, val := range sr.sr.procs {
-		//fmt.Println("pid=", val.pid)
-		if val.pid == 0 {
-			break
-		}
-		if idx < len(usedGPUHostArray) {
-			if val.hostpid == 0 || val.hostpid != int32(usedGPUHostArray[idx].hostGPUPid) {
-				fmt.Println("Assign host pid to pid instead", usedGPUHostArray[idx].hostGPUPid, val.pid, val.hostpid)
-				sr.sr.procs[idx].hostpid = int32(usedGPUHostArray[idx].hostGPUPid)
-				fmt.Println("val=", val.hostpid, sr.sr.procs[idx].hostpid)
-			}
-		}
-	}
-	return nil
-
-}
-
 func CheckBlocking(utSwitchOn map[string]UtilizationPerDevice, p int, pu podusage) bool {
 	for _, devuuid := range pu.sr.uuids {
 		_, ok := utSwitchOn[string(devuuid.uuid[:])]

cmd/vGPUmonitor/main.go

@@ -24,9 +24,7 @@ func main() {
 	if err := ValidateEnvVars(); err != nil {
 		klog.Fatalf("Failed to validate environment variables: %v", err)
 	}
-	cgroupDriver = 0
 	errchannel := make(chan error)
-	go serveInfo(errchannel)
 	go initMetrics()
 	go watchAndFeedback()
 	for {

cmd/vGPUmonitor/metrics.go

@@ -17,7 +17,6 @@ limitations under the License.
 package main
 
 import (
-	"errors"
 	"fmt"
 	"log"
 	"net/http"
@@ -121,15 +120,6 @@ func (cc ClusterManagerCollector) Describe(ch chan<- *prometheus.Desc) {
 	//prometheus.DescribeByCollect(cc, ch)
 }
 
-func parseidstr(podusage string) (string, string, error) {
-	tmp := strings.Split(podusage, "_")
-	if len(tmp) > 1 {
-		return tmp[0], tmp[1], nil
-	} else {
-		return "", "", errors.New("parse error")
-	}
-}
-
 func gettotalusage(usage podusage, vidx int) (deviceMemory, error) {
 	added := deviceMemory{
 		bufferSize:  0,
@@ -148,10 +138,6 @@ func gettotalusage(usage podusage, vidx int) (deviceMemory, error) {
 	return added, nil
 }
 
-func getsrlist() map[string]podusage {
-	return srPodList
-}
-
 // Collect first triggers the ReallyExpensiveAssessmentOfTheSystemState. Then it
 // creates constant metrics for each host on the fly based on the returned data.
 //