WIP Implement resource constraints on nodes

pkg/apis/config/v1alpha4/types.go

@@ -16,6 +16,10 @@ limitations under the License.
 
 package v1alpha4
 
+import (
+	"k8s.io/apimachinery/pkg/api/resource"
+)
+
 // Cluster contains kind cluster configuration
 type Cluster struct {
 	TypeMeta `yaml:",inline"`
@@ -106,6 +110,9 @@ type Node struct {
 	// binded to a host Port
 	ExtraPortMappings []PortMapping `yaml:"extraPortMappings,omitempty"`
 
+	// Constraints describes the node resources constraints
+	Constraints *NodeResources `yaml:"constraints,omitempty"`
+
 	// KubeadmConfigPatches are applied to the generated kubeadm config as
 	// merge patches. The `kind` field must match the target object, and
 	// if `apiVersion` is specified it will only be applied to matching objects.
@@ -283,3 +290,11 @@ const (
 	// PortMappingProtocolSCTP specifies SCTP protocol
 	PortMappingProtocolSCTP PortMappingProtocol = "SCTP"
 )
+
+// NodeResources represents the node resources (CPU/Memory)
+type NodeResources struct {
+	// The maximum amount of memory the node can use.
+	Memory resource.Quantity `json:"memory,omitempty"`
+	// Specify how much of the available CPU resources a node can use
+	Cpus resource.Quantity `json:"cpus,omitempty"`
+}

pkg/apis/config/v1alpha4/yaml.go

@@ -19,6 +19,7 @@ package v1alpha4
 import (
 	"strings"
 
+	"k8s.io/apimachinery/pkg/api/resource"
 	"sigs.k8s.io/kind/pkg/errors"
 )
 
@@ -72,3 +73,23 @@ func (p *PortMapping) UnmarshalYAML(unmarshal func(interface{}) error) error {
 	*p = PortMapping(a)
 	return nil
 }
+
+// UnmarshalYAML implements custom decoding YAML
+// https://godoc.org/gopkg.in/yaml.v3
+func (n *NodeResources) UnmarshalYAML(unmarshal func(interface{}) error) error {
+	// first unmarshal in the alias type (to avoid a recursion loop on unmarshal)
+	type NodeResourcesAlias struct {
+		Cpus   string
+		Memory string
+	}
+	var a NodeResourcesAlias
+	if err := unmarshal(&a); err != nil {
+		return err
+	}
+	// and copy over the fields
+	*n = NodeResources{
+		Cpus:   resource.MustParse(a.Cpus),
+		Memory: resource.MustParse(a.Memory),
+	}
+	return nil
+}

pkg/cluster/internal/create/actions/config/config.go

@@ -31,6 +31,8 @@ import (
 	"sigs.k8s.io/kind/pkg/cluster/internal/providers/provider/common"
 	"sigs.k8s.io/kind/pkg/cluster/nodeutils"
 	"sigs.k8s.io/kind/pkg/internal/apis/config"
+
+	"k8s.io/apimachinery/pkg/api/resource"
 )
 
 // Action implements action for creating the node config files
@@ -201,6 +203,28 @@ func getKubeadmConfig(cfg *config.Cluster, data kubeadm.ConfigData, node nodes.N
 		data.NodeAddress = nodeAddressIPv6
 	}
 
+	// configure the resources constraints per node by reserving system resources on the kubelet
+	// node allocatable = total - node-system-reserverd
+	// https://kubernetes.io/docs/tasks/administer-cluster/reserve-compute-resources/
+
+	if configNode.Constraints != nil {
+		if !configNode.Constraints.Cpus.IsZero() {
+			nodeCPUs := resource.MustParse(string(common.GetSystemCPUs()))
+			nodeCPUs.Sub(configNode.Constraints.Cpus)
+			if nodeCPUs.Sign() == 1 {
+				data.NodeCPU = nodeCPUs.String()
+			}
+		}
+
+		if !configNode.Constraints.Memory.IsZero() {
+			nodeMemory := resource.MustParse(string(common.GetSystemMemTotal()))
+			nodeMemory.Sub(configNode.Constraints.Memory)
+			if nodeMemory.Sign() == 1 {
+				data.NodeMemory = nodeMemory.String()
+			}
+		}
+	}
+
 	// generate the config contents
 	cf, err := kubeadm.Config(data)
 	if err != nil {

pkg/cluster/internal/kubeadm/config.go

@@ -43,6 +43,10 @@ type ConfigData struct {
 	ControlPlane bool
 	// The main IP address of the node
 	NodeAddress string
+	// The amount of CPU reserved for the system components
+	NodeCPU string
+	// The amount of memory reserved for the system components
+	NodeMemory string
 	// The Token for TLS bootstrap
 	Token string
 	// The subnet used for pods
@@ -507,6 +511,11 @@ evictionHard:
   nodefs.available: "0%"
   nodefs.inodesFree: "0%"
   imagefs.available: "0%"
+# reserve system resources in order to limit the node allocatable resources
+# https://kubernetes.io/docs/tasks/administer-cluster/reserve-compute-resources/
+systemReserved:
+  cpu: "{{ .NodeCPU }}"
+  memory: "{{ .NodeMemory }}"
 {{if .FeatureGates}}featureGates:
 {{ range $key := .SortedFeatureGateKeys }}
   "{{ $key }}": {{ index $.FeatureGates $key }}

pkg/cluster/internal/providers/docker/provision.go

@@ -231,13 +231,16 @@ func runArgsForNode(node *config.Node, clusterIPFamily config.ClusterIPFamily, n
 		args...,
 	)
 
-	// convert mounts and port mappings to container run args
+	// convert mounts, port mappings and resource constraints to container run args
 	args = append(args, generateMountBindings(node.ExtraMounts...)...)
 	mappingArgs, err := generatePortMappings(clusterIPFamily, node.ExtraPortMappings...)
 	if err != nil {
 		return nil, err
 	}
 	args = append(args, mappingArgs...)
+	if node.Constraints != nil {
+		args = append(args, generateNodeConstraints(node.Constraints)...)
+	}
 
 	// finally, specify the image to run
 	return append(args, node.Image), nil
@@ -376,3 +379,21 @@ func generatePortMappings(clusterIPFamily config.ClusterIPFamily, portMappings .
 	}
 	return args, nil
 }
+
+// generateNodeConstraints converts the nodesConstraints to a list of args for docker
+// https://docs.docker.com/config/containers/resource_constraints/
+func generateNodeConstraints(resources *config.NodeResources) []string {
+	var args []string
+	if resources.Cpus.Sign() > 0 {
+		args = append(args, fmt.Sprintf("--cpus=%s", resources.Cpus.String()))
+	}
+
+	if resources.Memory.Sign() > 0 {
+		args = append(args, fmt.Sprintf("--memory=%s", resources.Memory.String()))
+		// prevent a container from using swap because we want to emulate a real node
+		// https://docs.docker.com/config/containers/resource_constraints/#prevent-a-container-from-using-swap
+		args = append(args, fmt.Sprintf("--memory-swap=%s", resources.Memory.String()))
+	}
+
+	return args
+}

pkg/cluster/internal/providers/provider/common/host.go

@@ -0,0 +1,64 @@
+/*
+Copyright 2019 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package common
+
+import (
+	"bufio"
+	"log"
+	"os"
+	"runtime"
+	"strconv"
+	"strings"
+)
+
+// /proc/meminfo is used by to report the amount of free and used memory
+// on the system as well as the shared memory and buffers used by the kernel.
+const memFile = "/proc/meminfo"
+
+// GetSystemMemTotal returns the total number of memory available on the system in bytes
+// It returns 0 if it can not obtain the memory
+func GetSystemMemTotal() uint64 {
+	file, err := os.Open(memFile)
+	if err != nil {
+		log.Fatal(err)
+	}
+	defer file.Close()
+
+	scanner := bufio.NewScanner(file)
+	for scanner.Scan() {
+		fields := strings.Split(scanner.Text(), ":")
+		if len(fields) != 2 {
+			continue
+		}
+		key := strings.TrimSpace(fields[0])
+		if key == "MemTotal" {
+			value := strings.TrimSpace(fields[1])
+			value = strings.Replace(value, " kB", "", -1)
+			t, err := strconv.ParseUint(value, 10, 64)
+			if err != nil {
+				return 0
+			}
+			return t * 1024
+		}
+	}
+	return 0
+}
+
+// GetSystemCPUs return the number of CPUs in the system
+func GetSystemCPUs() int {
+	return runtime.NumCPU()
+}

pkg/internal/apis/config/types.go

@@ -16,6 +16,10 @@ limitations under the License.
 
 package config
 
+import (
+	"k8s.io/apimachinery/pkg/api/resource"
+)
+
 // Cluster contains kind cluster configuration
 type Cluster struct {
 	// Nodes contains the list of nodes defined in the `kind` Cluster
@@ -80,6 +84,9 @@ type Node struct {
 	// binded to a host Port
 	ExtraPortMappings []PortMapping
 
+	// Constraints describes the node resources constraints
+	Constraints *NodeResources
+
 	// KubeadmConfigPatches are applied to the generated kubeadm config as
 	// strategic merge patches to `kustomize build` internally
 	// https://github.com/kubernetes/community/blob/a9cf5c8f3380bb52ebe57b1e2dbdec136d8dd484/contributors/devel/sig-api-machinery/strategic-merge-patch.md
@@ -234,3 +241,11 @@ const (
 	// PortMappingProtocolSCTP specifies SCTP protocol
 	PortMappingProtocolSCTP PortMappingProtocol = "SCTP"
 )
+
+// NodeResources represents the node resources (CPU/Memory)
+type NodeResources struct {
+	// The maximum amount of memory the node can use.
+	Memory resource.Quantity
+	// Specify how much of the available CPU resources a node can use.
+	Cpus resource.Quantity
+}

pkg/internal/apis/config/validate.go

@@ -19,6 +19,8 @@ package config
 import (
 	"net"
 
+	"k8s.io/apimachinery/pkg/api/resource"
+
 	"sigs.k8s.io/kind/pkg/errors"
 )
 
@@ -101,6 +103,19 @@ func (n *Node) Validate() error {
 		}
 	}
 
+	if n.Constraints != nil {
+		// validate node resource constraints
+		if n.Constraints.Cpus.Sign() != 1 {
+			errs = append(errs, errors.Errorf("invalid number of Cpus: %s", n.Constraints.Cpus.String()))
+		}
+
+		// minimum memory size is 4m
+		minMemory := resource.MustParse("4m")
+		if n.Constraints.Memory.Sign() != 1 && n.Constraints.Memory.Cmp(minMemory) < 0 {
+			errs = append(errs, errors.Errorf("invalid Memory Size (minimum 4m): %v", n.Constraints.Memory.String()))
+		}
+	}
+
 	if len(errs) > 0 {
 		return errors.NewAggregate(errs)
 	}

pkg/internal/apis/config/validate_test.go

@@ -19,6 +19,8 @@ package config
 import (
 	"testing"
 
+	"k8s.io/apimachinery/pkg/api/resource"
+
 	"sigs.k8s.io/kind/pkg/errors"
 )
 
@@ -107,6 +109,24 @@ func TestClusterValidate(t *testing.T) {
 			}(),
 			ExpectErrors: 1,
 		},
+		{
+			Name: "wrong resources constraints node",
+			Cluster: func() Cluster {
+				c := Cluster{}
+				SetDefaultsCluster(&c)
+				n, n2 := Node{}, Node{}
+				SetDefaultsNode(&n)
+				SetDefaultsNode(&n2)
+				r := NodeResources{
+					Cpus:   resource.MustParse("-12"),
+					Memory: resource.MustParse("1m"),
+				}
+				n.Constraints = &r
+				c.Nodes = []Node{n, n2}
+				return c
+			}(),
+			ExpectErrors: 1,
+		},
 	}
 
 	for _, tc := range cases {
@@ -202,6 +222,24 @@ func TestNodeValidate(t *testing.T) {
 			}(),
 			ExpectErrors: 1,
 		},
+		{
+			TestName: "Negative CPU constraint",
+			Node: func() Node {
+				cfg := newDefaultedNode(ControlPlaneRole)
+				cfg.Constraints.Cpus = resource.MustParse("-12")
+				return cfg
+			}(),
+			ExpectErrors: 1,
+		},
+		{
+			TestName: "Minimum value for memory constraint",
+			Node: func() Node {
+				cfg := newDefaultedNode(ControlPlaneRole)
+				cfg.Constraints.Memory = resource.MustParse("2m")
+				return cfg
+			}(),
+			ExpectErrors: 1,
+		},
 		{
 			TestName: "Invalid HostPort",
 			Node: func() Node {