Merge pull request #19857 from zexi/container-system-cpu-simulate

feat(region,host,climc): simulating /sys/devices/system/cpu directory

cmd/climc/shell/compute/containers.go

@@ -26,6 +26,7 @@ import (
 	"yunion.io/x/pkg/errors"
 
 	"yunion.io/x/onecloud/cmd/climc/shell"
+	computeapi "yunion.io/x/onecloud/pkg/apis/compute"
 	"yunion.io/x/onecloud/pkg/mcclient"
 	modules "yunion.io/x/onecloud/pkg/mcclient/modules/compute"
 	options "yunion.io/x/onecloud/pkg/mcclient/options/compute"
@@ -85,4 +86,26 @@ func init() {
 
 		return nil
 	})
+
+	type SetSpecOptions struct {
+		ID                string `help:"ID or name of server" json:"-"`
+		EnableSimulateCpu bool   `help:"Enable simulating /sys/devices/system/cpu directory"`
+	}
+	R(&SetSpecOptions{}, "container-set-spec", "Set spec of a container", func(s *mcclient.ClientSession, opts *SetSpecOptions) error {
+		result, err := modules.Containers.Get(s, opts.ID, nil)
+		if err != nil {
+			return errors.Wrap(err, "get container id")
+		}
+		spec := new(computeapi.ContainerSpec)
+		if err := result.Unmarshal(spec, "spec"); err != nil {
+			return errors.Wrap(err, "unmarshal to spec")
+		}
+		spec.SimulateCpu = opts.EnableSimulateCpu
+		result.(*jsonutils.JSONDict).Set("spec", jsonutils.Marshal(spec))
+
+		if _, err := modules.Containers.Update(s, opts.ID, result); err != nil {
+			return errors.Wrap(err, "update spec")
+		}
+		return nil
+	})
 }

pkg/apis/container.go

@@ -61,6 +61,7 @@ type ContainerSpec struct {
 	Privileged         bool                    `json:"privileged"`
 	Lifecyle           *ContainerLifecyle      `json:"lifecyle"`
 	CgroupDevicesAllow []string                `json:"cgroup_devices_allow"`
+	SimulateCpu        bool                    `json:"simulate_cpu"`
 }
 
 type ContainerCapability struct {

pkg/hostman/guestman/guestman.go

@@ -206,6 +206,10 @@ func (m *SGuestManager) GetCRI() pod.CRI {
 	return m.host.GetCRI()
 }
 
+func (m *SGuestManager) GetContainerCPUMap() *pod.HostContainerCPUMap {
+	return m.host.GetContainerCPUMap()
+}
+
 func (m *SGuestManager) getPythonPath() string {
 	return m.pythonPath
 }

pkg/hostman/guestman/pod.go

@@ -120,6 +120,10 @@ func (s *sPodGuestInstance) getCRI() pod.CRI {
 	return s.manager.GetCRI()
 }
 
+func (s *sPodGuestInstance) getHostCPUMap() *pod.HostContainerCPUMap {
+	return s.manager.GetContainerCPUMap()
+}
+
 func (s *sPodGuestInstance) getPod(ctx context.Context) (*runtimeapi.PodSandbox, error) {
 	pods, err := s.getCRI().ListPods(ctx, pod.ListPodOptions{})
 	if err != nil {
@@ -816,13 +820,21 @@ func (s *sPodGuestInstance) createContainer(ctx context.Context, userCred mcclie
 	if err != nil {
 		return "", errors.Wrap(err, "getPodSandboxConfig")
 	}
+	spec := input.Spec
 	mounts, err := s.getContainerMounts(ctrId, input)
 	if err != nil {
 		return "", errors.Wrap(err, "get container mounts")
 	}
+	if spec.SimulateCpu {
+		systemCpuMounts, err := s.simulateContainerSystemCpu(ctx, ctrId)
+		if err != nil {
+			return "", errors.Wrapf(err, "simulate container system cpu")
+		}
+		newMounts := systemCpuMounts
+		newMounts = append(newMounts, mounts...)
+		mounts = newMounts
+	}
 
-	// REF: https://docs.docker.com/config/containers/resource_constraints/#configure-the-default-cfs-scheduler
-	spec := input.Spec
 	ctrCfg := &runtimeapi.ContainerConfig{
 		Metadata: &runtimeapi.ContainerMetadata{
 			Name: input.Name,
@@ -832,6 +844,7 @@ func (s *sPodGuestInstance) createContainer(ctx context.Context, userCred mcclie
 		},
 		Linux: &runtimeapi.LinuxContainerConfig{
 			Resources: &runtimeapi.LinuxContainerResources{
+				// REF: https://docs.docker.com/config/containers/resource_constraints/#configure-the-default-cfs-scheduler
 				CpuPeriod: s.getDefaultCPUPeriod(),
 				CpuQuota:  s.GetDesc().Cpu * s.getDefaultCPUPeriod(),
 				//CpuShares:              defaultCPUPeriod,
@@ -917,6 +930,69 @@ func (s *sPodGuestInstance) createContainer(ctx context.Context, userCred mcclie
 	return criId, nil
 }
 
+func (s *sPodGuestInstance) getContainerSystemCpusDir(ctrId string) string {
+	return filepath.Join(s.HomeDir(), "cpus", ctrId)
+}
+
+func (s *sPodGuestInstance) ensureContainerSystemCpuDir(cpuDir string, cpuCnt int64) error {
+	// create cpu dir like /var/lib/docker/cpus/$ctr_name
+	if err := pod.EnsureContainerSystemCpuDir(cpuDir, cpuCnt); err != nil {
+		return errors.Wrap(err, "ensure container system cpu dir")
+	}
+
+	return nil
+}
+
+func (s *sPodGuestInstance) findHostCpuPath(ctrId string, cpuIndex int) (int, error) {
+	return s.getHostCPUMap().Get(ctrId, cpuIndex)
+}
+
+func (s *sPodGuestInstance) simulateContainerSystemCpu(ctx context.Context, ctrId string) ([]*runtimeapi.Mount, error) {
+	cpuDir := s.getContainerSystemCpusDir(ctrId)
+	cpuCnt := s.GetDesc().Cpu
+	if err := s.ensureContainerSystemCpuDir(cpuDir, cpuCnt); err != nil {
+		return nil, err
+	}
+	sysCpuPath := "/sys/devices/system/cpu"
+	ret := []*runtimeapi.Mount{
+		{
+			ContainerPath: sysCpuPath,
+			HostPath:      cpuDir,
+		},
+	}
+	for i := 0; i < int(cpuCnt); i++ {
+		hostCpuIdx, err := s.findHostCpuPath(ctrId, i)
+		if err != nil {
+			return nil, errors.Wrapf(err, "find host cpu by container %s with index %d", ctrId, i)
+		}
+		hostCpuPath := filepath.Join(sysCpuPath, fmt.Sprintf("cpu%d", hostCpuIdx))
+		ret = append(ret, &runtimeapi.Mount{
+			ContainerPath: filepath.Join(sysCpuPath, fmt.Sprintf("cpu%d", i)),
+			HostPath:      hostCpuPath,
+		})
+	}
+	pathMap := func(baseName string) *runtimeapi.Mount {
+		p := filepath.Join(sysCpuPath, baseName)
+		return &runtimeapi.Mount{
+			ContainerPath: p,
+			HostPath:      p,
+			Readonly:      true,
+		}
+	}
+	for _, baseName := range []string{
+		"modalias",
+		"power",
+		"cpuidle",
+		"hotplug",
+		"isolated",
+		"cpufreq",
+		"uevent",
+	} {
+		ret = append(ret, pathMap(baseName))
+	}
+	return ret, nil
+}
+
 func (s *sPodGuestInstance) DeleteContainer(ctx context.Context, userCred mcclient.TokenCredential, ctrId string) (jsonutils.JSONObject, error) {
 	criId, err := s.getContainerCRIId(ctrId)
 	if err != nil && errors.Cause(err) != errors.ErrNotFound {
@@ -932,6 +1008,9 @@ func (s *sPodGuestInstance) DeleteContainer(ctx context.Context, userCred mcclie
 	if err := s.saveContainersFile(s.containers); err != nil {
 		return nil, errors.Wrap(err, "saveContainersFile")
 	}
+	if err := s.getHostCPUMap().Delete(ctrId); err != nil {
+		log.Warningf("delete container %s cpu map: %v", ctrId, err)
+	}
 	return nil, nil
 }
 

pkg/hostman/hostinfo/hostinfo.go

@@ -118,7 +118,8 @@ type SHostInfo struct {
 
 	IoScheduler string
 
-	cri pod.CRI
+	cri             pod.CRI
+	containerCPUMap *pod.HostContainerCPUMap
 }
 
 func (h *SHostInfo) GetContainerDeviceConfigurationFilePath() string {
@@ -225,6 +226,9 @@ func (h *SHostInfo) Init() error {
 		if err := h.initCRI(); err != nil {
 			return errors.Wrap(err, "init container runtime interface")
 		}
+		if err := h.initContainerCPUMap(h.sysinfo.Topology); err != nil {
+			return errors.Wrap(err, "init container cpu map")
+		}
 	}
 
 	return nil
@@ -244,10 +248,24 @@ func (h *SHostInfo) initCRI() error {
 	return nil
 }
 
+func (h *SHostInfo) initContainerCPUMap(topo *hostapi.HostTopology) error {
+	statefile := path.Join(options.HostOptions.ServersPath, "container_cpu_map")
+	cm, err := pod.NewHostContainerCPUMap(topo, statefile)
+	if err != nil {
+		return errors.Wrap(err, "NewHostContainerCPUMap")
+	}
+	h.containerCPUMap = cm
+	return nil
+}
+
 func (h *SHostInfo) GetCRI() pod.CRI {
 	return h.cri
 }
 
+func (h *SHostInfo) GetContainerCPUMap() *pod.HostContainerCPUMap {
+	return h.containerCPUMap
+}
+
 func (h *SHostInfo) setupOvnChassis() error {
 	opts := &options.HostOptions
 	if opts.BridgeDriver != hostbridge.DRV_OPEN_VSWITCH {

pkg/hostman/hostutils/hostutils.go

@@ -77,6 +77,7 @@ type IHost interface {
 	IsContainerHost() bool
 	GetContainerRuntimeEndpoint() string
 	GetCRI() pod.CRI
+	GetContainerCPUMap() *pod.HostContainerCPUMap
 }
 
 func GetComputeSession(ctx context.Context) *mcclient.ClientSession {

pkg/mcclient/options/compute/containers.go

@@ -57,6 +57,7 @@ type ContainerCreateCommonOptions struct {
 	EnableLxcfs       bool     `help:"Enable lxcfs"`
 	PostStartExec     string   `help:"Post started execution command"`
 	CgroupDeviceAllow []string `help:"Cgroup devices.allow, e.g.: 'c 13:* rwm'"`
+	SimulateCpu       bool     `help:"Simulating /sys/devices/system/cpu files"`
 }
 
 func (o ContainerCreateCommonOptions) getCreateSpec() (*computeapi.ContainerSpec, error) {
@@ -70,6 +71,7 @@ func (o ContainerCreateCommonOptions) getCreateSpec() (*computeapi.ContainerSpec
 			Privileged:         o.Privileged,
 			Capabilities:       &apis.ContainerCapability{},
 			CgroupDevicesAllow: o.CgroupDeviceAllow,
+			SimulateCpu:        o.SimulateCpu,
 		},
 	}
 	if len(o.PostStartExec) != 0 {

pkg/util/pod/cpus.go

@@ -0,0 +1,253 @@
+package pod
+
+import (
+	"fmt"
+	"path/filepath"
+	"strconv"
+	"sync"
+
+	"yunion.io/x/jsonutils"
+	"yunion.io/x/pkg/errors"
+
+	hostapi "yunion.io/x/onecloud/pkg/apis/host"
+	"yunion.io/x/onecloud/pkg/util/fileutils2"
+	"yunion.io/x/onecloud/pkg/util/procutils"
+)
+
+func EnsureDir(dir string) error {
+	out, err := procutils.NewRemoteCommandAsFarAsPossible("mkdir", "-p", dir).Output()
+	if err != nil {
+		return errors.Wrapf(err, "mkdir %s: %s", dir, out)
+	}
+	return nil
+}
+
+func EnsureFile(filename string, content string, mod string) error {
+	cmds := []string{
+		fmt.Sprintf("echo '%s' > %s", content, filename),
+	}
+	if mod != "" {
+		cmds = append(cmds, fmt.Sprintf("chmod %s %s", mod, filename))
+	}
+	for _, cmd := range cmds {
+		out, err := procutils.NewRemoteCommandAsFarAsPossible("sh", "-c", cmd).Output()
+		if err != nil {
+			return errors.Wrapf(err, "cmd %s: %s", cmd, out)
+		}
+	}
+	return nil
+}
+
+func EnsureContainerSystemCpuDir(cpuDir string, cpuCount int64) error {
+	if err := EnsureDir(cpuDir); err != nil {
+		return err
+	}
+	for i := 0; i < int(cpuCount); i++ {
+		singleCpuDir := filepath.Join(cpuDir, fmt.Sprintf("cpu%d", i))
+		if err := EnsureDir(singleCpuDir); err != nil {
+			return errors.Wrapf(err, "create cpu%d", i)
+		}
+	}
+	// create other dirs
+	freqDir := filepath.Join(cpuDir, "cpufreq")
+	idleDir := filepath.Join(cpuDir, "cpuidle")
+	hotplugDir := filepath.Join(cpuDir, "hotplug")
+	powerDir := filepath.Join(cpuDir, "power")
+	for _, dir := range []string{freqDir, idleDir, hotplugDir, powerDir} {
+		if err := EnsureDir(dir); err != nil {
+			return err
+		}
+	}
+	// create files
+	if err := EnsureFile(filepath.Join(cpuDir, "isolated"), "", "755"); err != nil {
+		return err
+	}
+	if err := EnsureFile(filepath.Join(cpuDir, "kernel_max"), fmt.Sprintf("%d", cpuCount), "644"); err != nil {
+		return err
+	}
+	if err := EnsureFile(filepath.Join(cpuDir, "modalias"), "", "755"); err != nil {
+		return err
+	}
+	if err := EnsureFile(filepath.Join(cpuDir, "offline"), "", "755"); err != nil {
+		return err
+	}
+	ensureCpuRangeFile := func(baseName string, mode string) error {
+		if mode == "" {
+			mode = "644"
+		}
+		if cpuCount > 1 {
+			if err := EnsureFile(filepath.Join(cpuDir, baseName), fmt.Sprintf("0-%d", cpuCount-1), mode); err != nil {
+				return err
+			}
+			return nil
+		}
+		if err := EnsureFile(filepath.Join(cpuDir, baseName), "0", mode); err != nil {
+			return err
+		}
+		return nil
+	}
+
+	if err := ensureCpuRangeFile("online", "755"); err != nil {
+		return err
+	}
+	if err := ensureCpuRangeFile("possible", ""); err != nil {
+		return err
+	}
+	if err := ensureCpuRangeFile("present", ""); err != nil {
+		return err
+	}
+	if err := EnsureFile(filepath.Join(cpuDir, "uevent"), "", "755"); err != nil {
+		return err
+	}
+	return nil
+}
+
+type ContainerCPU struct {
+	ContainerId string `json:"container_id"`
+	Index       int    `json:"index"`
+}
+
+func NewContainerCPU(ctrId string, ctrIdx int) *ContainerCPU {
+	return &ContainerCPU{
+		ContainerId: ctrId,
+		Index:       ctrIdx,
+	}
+}
+
+type HostContainerCPU struct {
+	Index      int                        `json:"index"`
+	Containers map[string][]*ContainerCPU `json:"containers"`
+}
+
+func NewHostContainerCPU(hostIndex int) *HostContainerCPU {
+	return &HostContainerCPU{
+		Index:      hostIndex,
+		Containers: make(map[string][]*ContainerCPU),
+	}
+}
+
+func (h *HostContainerCPU) HasContainer(ctrId string) bool {
+	_, ok := h.Containers[ctrId]
+	return ok
+}
+
+func (h *HostContainerCPU) DeleteContainer(ctrId string) {
+	delete(h.Containers, ctrId)
+}
+
+func (h *HostContainerCPU) InsertContainer(ctrId string, ctrIdx int) {
+	if h.HasContainer(ctrId) {
+		h.Containers[ctrId] = append(h.Containers[ctrId], NewContainerCPU(ctrId, ctrIdx))
+	} else {
+		h.Containers[ctrId] = []*ContainerCPU{NewContainerCPU(ctrId, ctrIdx)}
+	}
+}
+
+var (
+	hostContainerCPUMapLock = sync.Mutex{}
+)
+
+type HostContainerCPUMap struct {
+	Map       map[string]*HostContainerCPU `json:"map"`
+	stateFile string
+}
+
+func NewHostContainerCPUMap(topo *hostapi.HostTopology, stateFile string) (*HostContainerCPUMap, error) {
+	ret := make(map[string]*HostContainerCPU)
+	if fileutils2.Exists(stateFile) {
+		content, err := fileutils2.FileGetContents(stateFile)
+		if err != nil {
+			return nil, errors.Wrapf(err, "get file contents: %s", stateFile)
+		}
+		obj, err := jsonutils.ParseString(content)
+		if err != nil {
+			return nil, errors.Wrapf(err, "parse to json: %s", content)
+		}
+		hm := new(HostContainerCPUMap)
+		if err := obj.Unmarshal(hm); err != nil {
+			return nil, errors.Wrap(err, "unmarshal to HostContainerCPUMap")
+		}
+		hm.stateFile = stateFile
+		return hm, nil
+	}
+	nodes := topo.Nodes
+	for _, node := range nodes {
+		for _, core := range node.Cores {
+			for _, processor := range core.LogicalProcessors {
+				ret[fmt.Sprintf("%d", processor)] = NewHostContainerCPU(processor)
+			}
+		}
+	}
+	return &HostContainerCPUMap{Map: ret, stateFile: stateFile}, nil
+}
+
+func (hm *HostContainerCPUMap) dumpToFile() error {
+	return fileutils2.FilePutContents(hm.stateFile, jsonutils.Marshal(hm).PrettyString(), false)
+}
+
+func (hm *HostContainerCPUMap) Delete(ctrId string) error {
+	hostContainerCPUMapLock.Lock()
+	defer hostContainerCPUMapLock.Unlock()
+
+	for _, cm := range hm.Map {
+		if cm.HasContainer(ctrId) {
+			cm.DeleteContainer(ctrId)
+		}
+	}
+	return hm.dumpToFile()
+}
+
+func (hm *HostContainerCPUMap) Get(ctrId string, ctrCpuIndex int) (int, error) {
+	hostContainerCPUMapLock.Lock()
+	defer hostContainerCPUMapLock.Unlock()
+
+	hostIndex := hm.findLeastUsedIndex(ctrId, ctrCpuIndex)
+	if err := hm.markUsed(hostIndex, ctrId, ctrCpuIndex); err != nil {
+		return 0, errors.Wrapf(err, "mark container %s %d used", ctrId, ctrCpuIndex)
+	}
+	return hostIndex, nil
+}
+
+func (hm *HostContainerCPUMap) findLeastUsedIndex(ctrId string, ctrCpuIndex int) int {
+	unusedMap := make(map[string]*HostContainerCPU)
+	usedMap := make(map[string]*HostContainerCPU)
+	for idx, hc := range hm.Map {
+		tmpHc := hc
+		if tmpHc.HasContainer(ctrId) {
+			usedMap[idx] = tmpHc
+		} else {
+			unusedMap[idx] = tmpHc
+		}
+	}
+	if len(unusedMap) != 0 {
+		for idx, _ := range unusedMap {
+			idxNum, _ := strconv.Atoi(idx)
+			return idxNum
+		}
+	}
+	// find the least used index from usedMap
+	isStart := true
+	leastIndex := 0
+	leastUsed := 0
+	for idx, hc := range usedMap {
+		cm := hc.Containers[ctrId]
+		idxNum, _ := strconv.Atoi(idx)
+		if isStart {
+			leastIndex = idxNum
+			leastUsed = len(cm)
+			isStart = false
+			continue
+		}
+		if len(cm) < leastUsed {
+			leastUsed = len(cm)
+			leastIndex = idxNum
+		}
+	}
+	return leastIndex
+}
+
+func (hm *HostContainerCPUMap) markUsed(hostIdx int, ctrId string, ctrIdx int) error {
+	hc := hm.Map[fmt.Sprintf("%d", hostIdx)]
+	hc.InsertContainer(ctrId, ctrIdx)
+	return hm.dumpToFile()
+}