Add support for CFS quota in kubelet

cmd/kubelet/app/server.go

@@ -124,7 +124,7 @@ type KubeletServer struct {
 	MaxPods                        int
 	DockerExecHandlerName          string
 	ResolverConfig                 string
-
+	CPUCFSQuota                    bool
 	// Flags intended for testing
 
 	// Crash immediately, rather than eating panics.
@@ -189,6 +189,7 @@ func NewKubeletServer() *KubeletServer {
 		SystemContainer:             "",
 		ConfigureCBR0:               false,
 		DockerExecHandlerName:       "native",
+		CPUCFSQuota:                 false,
 	}
 }
 
@@ -255,6 +256,7 @@ func (s *KubeletServer) AddFlags(fs *pflag.FlagSet) {
 	fs.StringVar(&s.DockerExecHandlerName, "docker-exec-handler", s.DockerExecHandlerName, "Handler to use when executing a command in a container. Valid values are 'native' and 'nsenter'. Defaults to 'native'.")
 	fs.StringVar(&s.PodCIDR, "pod-cidr", "", "The CIDR to use for pod IP addresses, only used in standalone mode.  In cluster mode, this is obtained from the master.")
 	fs.StringVar(&s.ResolverConfig, "resolv-conf", kubelet.ResolvConfDefault, "Resolver configuration file used as the basis for the container DNS resolution configuration.")
+	fs.BoolVar(&s.CPUCFSQuota, "cpu-cfs-quota", s.CPUCFSQuota, "Enable CPU CFS quota enforcement for containers that specify CPU limits")
 	// Flags intended for testing, not recommended used in production environments.
 	fs.BoolVar(&s.ReallyCrashForTesting, "really-crash-for-testing", s.ReallyCrashForTesting, "If true, when panics occur crash. Intended for testing.")
 	fs.Float64Var(&s.ChaosChance, "chaos-chance", s.ChaosChance, "If > 0.0, introduce random client errors and latency. Intended for testing. [default=0.0]")
@@ -362,6 +364,7 @@ func (s *KubeletServer) KubeletConfig() (*KubeletConfig, error) {
 		MaxPods:                   s.MaxPods,
 		DockerExecHandler:         dockerExecHandler,
 		ResolverConfig:            s.ResolverConfig,
+		CPUCFSQuota:               s.CPUCFSQuota,
 	}, nil
 }
 
@@ -604,6 +607,7 @@ func SimpleKubelet(client *client.Client,
 		MaxPods:                   32,
 		DockerExecHandler:         &dockertools.NativeExecHandler{},
 		ResolverConfig:            kubelet.ResolvConfDefault,
+		CPUCFSQuota:               false,
 	}
 	return &kcfg
 }
@@ -774,6 +778,7 @@ type KubeletConfig struct {
 	MaxPods                        int
 	DockerExecHandler              dockertools.ExecHandler
 	ResolverConfig                 string
+	CPUCFSQuota                    bool
 }
 
 func createAndInitKubelet(kc *KubeletConfig) (k KubeletBootstrap, pc *config.PodConfig, err error) {
@@ -833,7 +838,8 @@ func createAndInitKubelet(kc *KubeletConfig) (k KubeletBootstrap, pc *config.Pod
 		kc.PodCIDR,
 		kc.MaxPods,
 		kc.DockerExecHandler,
-		kc.ResolverConfig)
+		kc.ResolverConfig,
+		kc.CPUCFSQuota)
 
 	if err != nil {
 		return nil, nil, err

contrib/mesos/pkg/executor/service/service.go

@@ -262,6 +262,7 @@ func (s *KubeletExecutorServer) Run(hks hyperkube.Interface, _ []string) error {
 		MaxPods:                   s.MaxPods,
 		DockerExecHandler:         dockerExecHandler,
 		ResolverConfig:            s.ResolverConfig,
+		CPUCFSQuota:               s.CPUCFSQuota,
 	}
 
 	kcfg.NodeName = kcfg.Hostname
@@ -364,6 +365,7 @@ func (ks *KubeletExecutorServer) createAndInitKubelet(
 		kc.MaxPods,
 		kc.DockerExecHandler,
 		kc.ResolverConfig,
+		kc.CPUCFSQuota,
 	)
 	if err != nil {
 		return nil, nil, err

hack/verify-flags/known-flags.txt

@@ -264,3 +264,4 @@ whitelist-override-label
 www-prefix
 retry_time
 file_content_in_loop
+cpu-cfs-quota

pkg/kubelet/dockertools/docker.go

@@ -50,6 +50,9 @@ const (
 	minShares     = 2
 	sharesPerCPU  = 1024
 	milliCPUToCPU = 1000
+
+	// 100000 is equivalent to 100ms
+	quotaPeriod = 100000
 )
 
 // DockerInterface is an abstract interface for testability.  It abstracts the interface of docker.Client.
@@ -306,6 +309,28 @@ func ConnectToDockerOrDie(dockerEndpoint string) DockerInterface {
 	return client
 }
 
+// milliCPUToQuota converts milliCPU to CFS quota and period values
+func milliCPUToQuota(milliCPU int64) (quota int64, period int64) {
+	// CFS quota is measured in two values:
+	//  - cfs_period_us=100ms (the amount of time to measure usage across)
+	//  - cfs_quota=20ms (the amount of cpu time allowed to be used across a period)
+	// so in the above example, you are limited to 20% of a single CPU
+	// for multi-cpu environments, you just scale equivalent amounts
+
+	if milliCPU == 0 {
+		// take the default behavior from docker
+		return
+	}
+
+	// we set the period to 100ms by default
+	period = quotaPeriod
+
+	// we then convert your milliCPU to a value normalized over a period
+	quota = (milliCPU * quotaPeriod) / milliCPUToCPU
+
+	return
+}
+
 func milliCPUToShares(milliCPU int64) int64 {
 	if milliCPU == 0 {
 		// Docker converts zero milliCPU to unset, which maps to kernel default

pkg/kubelet/dockertools/docker_test.go

@@ -737,3 +737,43 @@ func TestMakePortsAndBindings(t *testing.T) {
 		}
 	}
 }
+
+func TestMilliCPUToQuota(t *testing.T) {
+	testCases := []struct {
+		input  int64
+		quota  int64
+		period int64
+	}{
+		{
+			input:  int64(0),
+			quota:  int64(0),
+			period: int64(0),
+		},
+		{
+			input:  int64(200),
+			quota:  int64(20000),
+			period: int64(100000),
+		},
+		{
+			input:  int64(500),
+			quota:  int64(50000),
+			period: int64(100000),
+		},
+		{
+			input:  int64(1000),
+			quota:  int64(100000),
+			period: int64(100000),
+		},
+		{
+			input:  int64(1500),
+			quota:  int64(150000),
+			period: int64(100000),
+		},
+	}
+	for _, testCase := range testCases {
+		quota, period := milliCPUToQuota(testCase.input)
+		if quota != testCase.quota || period != testCase.period {
+			t.Errorf("Input %v, expected quota %v period %v, but got quota %v period %v", testCase.input, testCase.quota, testCase.period, quota, period)
+		}
+	}
+}

pkg/kubelet/dockertools/fake_manager.go

@@ -46,7 +46,7 @@ func NewFakeDockerManager(
 	fakeProcFs := procfs.NewFakeProcFs()
 	dm := NewDockerManager(client, recorder, readinessManager, containerRefManager, machineInfo, podInfraContainerImage, qps,
 		burst, containerLogsDir, osInterface, networkPlugin, generator, httpClient, &NativeExecHandler{},
-		fakeOomAdjuster, fakeProcFs)
+		fakeOomAdjuster, fakeProcFs, false)
 	dm.dockerPuller = &FakeDockerPuller{}
 	dm.prober = prober.New(nil, readinessManager, containerRefManager, recorder)
 	return dm

pkg/kubelet/dockertools/manager.go

@@ -132,6 +132,9 @@ type DockerManager struct {
 
 	// Get information from /proc mount.
 	procFs procfs.ProcFsInterface
+
+	// If true, enforce container cpu limits with CFS quota support
+	cpuCFSQuota bool
 }
 
 func NewDockerManager(
@@ -150,7 +153,8 @@ func NewDockerManager(
 	httpClient kubeletTypes.HttpGetter,
 	execHandler ExecHandler,
 	oomAdjuster *oom.OomAdjuster,
-	procFs procfs.ProcFsInterface) *DockerManager {
+	procFs procfs.ProcFsInterface,
+	cpuCFSQuota bool) *DockerManager {
 	// Work out the location of the Docker runtime, defaulting to /var/lib/docker
 	// if there are any problems.
 	dockerRoot := "/var/lib/docker"
@@ -201,6 +205,7 @@ func NewDockerManager(
 		execHandler:            execHandler,
 		oomAdjuster:            oomAdjuster,
 		procFs:                 procFs,
+		cpuCFSQuota:            cpuCFSQuota,
 	}
 	dm.runner = lifecycle.NewHandlerRunner(httpClient, dm, dm)
 	dm.prober = prober.New(dm, readinessManager, containerRefManager, recorder)
@@ -673,6 +678,7 @@ func (dm *DockerManager) runContainer(
 		// of CPU shares.
 		cpuShares = milliCPUToShares(cpuRequest.MilliValue())
 	}
+
 	_, containerName := BuildDockerName(dockerName, container)
 	dockerOpts := docker.CreateContainerOptions{
 		Name: containerName,
@@ -742,6 +748,15 @@ func (dm *DockerManager) runContainer(
 		MemorySwap: -1,
 		CPUShares:  cpuShares,
 	}
+
+	if dm.cpuCFSQuota {
+		// if cpuLimit.Amount is nil, then the appropriate default value is returned to allow full usage of cpu resource.
+		cpuQuota, cpuPeriod := milliCPUToQuota(cpuLimit.MilliValue())
+
+		hc.CPUQuota = cpuQuota
+		hc.CPUPeriod = cpuPeriod
+	}
+
 	if len(opts.DNS) > 0 {
 		hc.DNS = opts.DNS
 	}

pkg/kubelet/kubelet.go

@@ -164,7 +164,8 @@ func NewMainKubelet(
 	podCIDR string,
 	pods int,
 	dockerExecHandler dockertools.ExecHandler,
-	resolverConfig string) (*Kubelet, error) {
+	resolverConfig string,
+	cpuCFSQuota bool) (*Kubelet, error) {
 	if rootDirectory == "" {
 		return nil, fmt.Errorf("invalid root directory %q", rootDirectory)
 	}
@@ -281,6 +282,7 @@ func NewMainKubelet(
 		pods:                           pods,
 		syncLoopMonitor:                util.AtomicValue{},
 		resolverConfig:                 resolverConfig,
+		cpuCFSQuota:                    cpuCFSQuota,
 	}
 
 	if plug, err := network.InitNetworkPlugin(networkPlugins, networkPluginName, &networkHost{klet}); err != nil {
@@ -317,7 +319,8 @@ func NewMainKubelet(
 			klet.httpClient,
 			dockerExecHandler,
 			oomAdjuster,
-			procFs)
+			procFs,
+			klet.cpuCFSQuota)
 	case "rkt":
 		conf := &rkt.Config{
 			Path:               rktPath,
@@ -556,6 +559,9 @@ type Kubelet struct {
 
 	// Optionally shape the bandwidth of a pod
 	shaper bandwidth.BandwidthShaper
+
+	// True if container cpu limits should be enforced via cgroup CFS quota
+	cpuCFSQuota bool
 }
 
 // getRootDir returns the full path to the directory under which kubelet can