Merge pull request #7496 from justinsb/label_controller

kops-controller

Makefile

@@ -56,6 +56,7 @@ unexport SKIP_REGION_CHECK S3_ACCESS_KEY_ID S3_ENDPOINT S3_REGION S3_SECRET_ACCE
 
 # Keep in sync with upup/models/cloudup/resources/addons/dns-controller/
 DNS_CONTROLLER_TAG=1.15.0-alpha.1
+KOPS_CONTROLLER_TAG=1.15.0-alpha.1
 
 # Keep in sync with logic in get_workspace_status
 # TODO: just invoke tools/get_workspace_status.sh?
@@ -902,3 +903,10 @@ dev-upload: dev-upload-nodeup dev-upload-protokube dev-copy-utils
 .PHONY: crds
 crds:
 	go run vendor/sigs.k8s.io/controller-tools/cmd/controller-gen/main.go crd --apis-path pkg/apis/kops/v1alpha2 --domain k8s.io --output-dir k8s/crds/
+
+#------------------------------------------------------
+# kops-controller
+
+.PHONY: kops-controller-push
+kops-controller-push:
+	DOCKER_REGISTRY=${DOCKER_REGISTRY} DOCKER_IMAGE_PREFIX=${DOCKER_IMAGE_PREFIX} KOPS_CONTROLLER_TAG=${KOPS_CONTROLLER_TAG} bazel run //cmd/kops-controller:push-image

WORKSPACE

@@ -137,3 +137,11 @@ go_repository(
     sum = "h1:PTrxTL8TNRbZts4KqdJMsqRlrdjoiKFDq6MVitj8mPk=",
     version = "v0.0.0-20190829181151-21b2e01cec04",
 )
+
+# Start using distroless base
+container_pull(
+    name = "distroless_base",
+    digest = "sha256:a4624843fb1d7d43d9e3d62f6d76b51b6e02b3d03221e29fef4e223d81ef3378",
+    registry = "gcr.io",
+    repository = "distroless/base",
+)

cmd/kops-controller/BUILD.bazel

@@ -0,0 +1,52 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_binary", "go_library")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["main.go"],
+    importpath = "k8s.io/kops/cmd/kops-controller",
+    visibility = ["//visibility:private"],
+    deps = [
+        "//cmd/kops-controller/controllers:go_default_library",
+        "//pkg/nodeidentity:go_default_library",
+        "//pkg/nodeidentity/aws:go_default_library",
+        "//pkg/nodeidentity/gce:go_default_library",
+        "//vendor/k8s.io/api/core/v1:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/runtime:go_default_library",
+        "//vendor/k8s.io/client-go/plugin/pkg/client/auth/gcp:go_default_library",
+        "//vendor/k8s.io/klog:go_default_library",
+        "//vendor/k8s.io/klog/klogr:go_default_library",
+        "//vendor/sigs.k8s.io/controller-runtime:go_default_library",
+        "//vendor/sigs.k8s.io/controller-runtime/pkg/manager:go_default_library",
+    ],
+)
+
+go_binary(
+    name = "kops-controller",
+    embed = [":go_default_library"],
+    visibility = ["//visibility:public"],
+)
+
+load(
+    "@io_bazel_rules_docker//container:container.bzl",
+    "container_image",
+    "container_push",
+)
+
+container_image(
+    name = "image",
+    base = "@distroless_base//image",
+    cmd = ["/usr/bin/kops-controller"],
+    directory = "/usr/bin/",
+    files = [
+        "//cmd/kops-controller",
+    ],
+)
+
+container_push(
+    name = "push-image",
+    format = "Docker",
+    image = ":image",
+    registry = "{STABLE_DOCKER_REGISTRY}",
+    repository = "{STABLE_DOCKER_IMAGE_PREFIX}kops-controller",
+    tag = "{STABLE_KOPS_CONTROLLER_TAG}",
+)

cmd/kops-controller/PROJECT

@@ -0,0 +1,3 @@
+version: "2"
+domain: kops.k8s.io
+repo: k8s.io/kops/cmd/kops-controller

cmd/kops-controller/controllers/BUILD.bazel

@@ -0,0 +1,25 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_library")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["node_controller.go"],
+    importpath = "k8s.io/kops/cmd/kops-controller/controllers",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//pkg/apis/kops:go_default_library",
+        "//pkg/apis/kops/registry:go_default_library",
+        "//pkg/nodeidentity:go_default_library",
+        "//pkg/nodelabels:go_default_library",
+        "//upup/pkg/fi/utils:go_default_library",
+        "//util/pkg/vfs:go_default_library",
+        "//vendor/github.com/go-logr/logr:go_default_library",
+        "//vendor/k8s.io/api/core/v1:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/api/errors:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/types:go_default_library",
+        "//vendor/k8s.io/client-go/kubernetes/typed/core/v1:go_default_library",
+        "//vendor/k8s.io/klog:go_default_library",
+        "//vendor/sigs.k8s.io/controller-runtime:go_default_library",
+        "//vendor/sigs.k8s.io/controller-runtime/pkg/client:go_default_library",
+        "//vendor/sigs.k8s.io/controller-runtime/pkg/manager:go_default_library",
+    ],
+)

cmd/kops-controller/controllers/node_controller.go

@@ -0,0 +1,249 @@
+/*
+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 controllers
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"time"
+
+	"github.com/go-logr/logr"
+	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/types"
+	corev1client "k8s.io/client-go/kubernetes/typed/core/v1"
+	"k8s.io/klog"
+	"k8s.io/kops/pkg/apis/kops"
+	"k8s.io/kops/pkg/apis/kops/registry"
+	"k8s.io/kops/pkg/nodeidentity"
+	"k8s.io/kops/pkg/nodelabels"
+	"k8s.io/kops/upup/pkg/fi/utils"
+	"k8s.io/kops/util/pkg/vfs"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
+)
+
+// NewNodeReconciler is the constructor for a NodeReconciler
+func NewNodeReconciler(mgr manager.Manager, configPath string, identifier nodeidentity.Identifier) (*NodeReconciler, error) {
+	r := &NodeReconciler{
+		client:     mgr.GetClient(),
+		log:        ctrl.Log.WithName("controllers").WithName("Node"),
+		identifier: identifier,
+		cache:      vfs.NewCache(),
+	}
+
+	coreClient, err := corev1client.NewForConfig(mgr.GetConfig())
+	if err != nil {
+		return nil, fmt.Errorf("error building corev1 client: %v", err)
+	}
+	r.coreV1Client = coreClient
+
+	configBase, err := vfs.Context.BuildVfsPath(configPath)
+	if err != nil {
+		return nil, fmt.Errorf("cannot parse ConfigBase %q: %v", configPath, err)
+	}
+	r.configBase = configBase
+
+	return r, nil
+}
+
+// NodeReconciler observes Node objects, and labels them with the correct labels for the instancegroup
+// This used to be done by the kubelet, but is moving to a central controller for greater security in 1.16
+type NodeReconciler struct {
+	// client is the controller-runtime client
+	client client.Client
+
+	// log is a logr
+	log logr.Logger
+
+	// coreV1Client is a client-go client for patching nodes
+	coreV1Client *corev1client.CoreV1Client
+
+	// identifier is a provider that can securely map node ProviderIDs to InstanceGroups
+	identifier nodeidentity.Identifier
+
+	// configBase is the parsed path to the base location of our configuration files
+	configBase vfs.Path
+
+	// cache caches the instancegroup and cluster values, to avoid repeated GCS/S3 calls
+	cache *vfs.Cache
+}
+
+// +kubebuilder:rbac:groups=,resources=nodes,verbs=get;list;watch;patch
+// Reconciler is the main reconciler function that observes node changes
+func (r *NodeReconciler) Reconcile(req ctrl.Request) (ctrl.Result, error) {
+	ctx := context.Background()
+	_ = r.log.WithValues("nodecontroller", req.NamespacedName)
+
+	node := &corev1.Node{}
+	if err := r.client.Get(ctx, req.NamespacedName, node); err != nil {
+		klog.Warningf("unable to fetch node %s: %v", node.Name, err)
+		if apierrors.IsNotFound(err) {
+			// we'll ignore not-found errors, since they can't be fixed by an immediate
+			// requeue (we'll need to wait for a new notification), and we can get them
+			// on deleted requests.
+			return ctrl.Result{}, nil
+		}
+		return ctrl.Result{}, err
+	}
+
+	cluster, err := r.getClusterForNode(node)
+	if err != nil {
+		return ctrl.Result{}, fmt.Errorf("unable to load cluster object for node %s: %v", node.Name, err)
+	}
+
+	ig, err := r.getInstanceGroupForNode(ctx, node)
+	if err != nil {
+		return ctrl.Result{}, fmt.Errorf("unable to load instance group object for node %s: %v", node.Name, err)
+	}
+
+	labels, err := nodelabels.BuildNodeLabels(cluster, ig)
+	if err != nil {
+		return ctrl.Result{}, fmt.Errorf("unable to build config for node %s: %v", node.Name, err)
+	}
+
+	updateLabels := make(map[string]string)
+	for k, v := range labels {
+		actual, found := node.Labels[k]
+		if !found || actual != v {
+			updateLabels[k] = v
+		}
+	}
+
+	if len(updateLabels) == 0 {
+		klog.V(4).Infof("no label changes needed for %s", node.Name)
+		return ctrl.Result{}, nil
+	}
+
+	if err := r.patchNodeLabels(node, updateLabels); err != nil {
+		klog.Warningf("failed to patch node labels on %s: %v", node.Name, err)
+		return ctrl.Result{}, err
+	}
+
+	return ctrl.Result{}, nil
+}
+
+func (r *NodeReconciler) SetupWithManager(mgr ctrl.Manager) error {
+	return ctrl.NewControllerManagedBy(mgr).
+		For(&corev1.Node{}).
+		Complete(r)
+}
+
+type nodePatch struct {
+	Metadata *nodePatchMetadata `json:"metadata,omitempty"`
+}
+
+type nodePatchMetadata struct {
+	Labels map[string]string `json:"labels,omitempty"`
+}
+
+// patchNodeLabels patches the node labels to set the specified labels
+func (r *NodeReconciler) patchNodeLabels(node *corev1.Node, setLabels map[string]string) error {
+	nodePatchMetadata := &nodePatchMetadata{
+		Labels: setLabels,
+	}
+	nodePatch := &nodePatch{
+		Metadata: nodePatchMetadata,
+	}
+	nodePatchJson, err := json.Marshal(nodePatch)
+	if err != nil {
+		return fmt.Errorf("error building node patch: %v", err)
+	}
+
+	klog.V(2).Infof("sending patch for node %q: %q", node.Name, string(nodePatchJson))
+
+	_, err = r.coreV1Client.Nodes().Patch(node.Name, types.StrategicMergePatchType, nodePatchJson)
+	if err != nil {
+		return fmt.Errorf("error applying patch to node: %v", err)
+	}
+
+	return nil
+}
+
+// getClusterForNode returns the kops.Cluster object for the node
+// The cluster is actually loaded when we first start
+func (r *NodeReconciler) getClusterForNode(node *corev1.Node) (*kops.Cluster, error) {
+	clusterPath := r.configBase.Join(registry.PathClusterCompleted)
+	cluster, err := r.loadCluster(clusterPath)
+	if err != nil {
+		return nil, err
+	}
+	return cluster, nil
+}
+
+// getInstanceGroupForNode returns the kops.InstanceGroup object for the node
+func (r *NodeReconciler) getInstanceGroupForNode(ctx context.Context, node *corev1.Node) (*kops.InstanceGroup, error) {
+	// We assume that if the instancegroup label is set, that it is correct
+	// TODO: Should we be paranoid?
+	instanceGroupName := node.Labels["kops.k8s.io/instancegroup"]
+
+	if instanceGroupName == "" {
+		providerID := node.Spec.ProviderID
+		if providerID == "" {
+			return nil, fmt.Errorf("node providerID not set for node %q", node.Name)
+		}
+
+		identity, err := r.identifier.IdentifyNode(ctx, node)
+		if err != nil {
+			return nil, fmt.Errorf("error identifying node %q: %v", node.Name, err)
+		}
+
+		if identity.InstanceGroup == "" {
+			return nil, fmt.Errorf("node %q did not have an associate instance group", node.Name)
+		}
+		instanceGroupName = identity.InstanceGroup
+	}
+
+	return r.loadNamedInstanceGroup(instanceGroupName)
+}
+
+// loadCluster loads a kops.Cluster object from a vfs.Path
+func (r *NodeReconciler) loadCluster(p vfs.Path) (*kops.Cluster, error) {
+	ttl := time.Hour
+
+	b, err := r.cache.Read(p, ttl)
+	if err != nil {
+		return nil, fmt.Errorf("error loading Cluster %q: %v", p, err)
+	}
+
+	cluster := &kops.Cluster{}
+	if err := utils.YamlUnmarshal(b, cluster); err != nil {
+		return nil, fmt.Errorf("error parsing Cluster %q: %v", p, err)
+	}
+
+	return cluster, nil
+}
+
+// loadInstanceGroup loads a kops.InstanceGroup object from the vfs backing store
+func (r *NodeReconciler) loadNamedInstanceGroup(name string) (*kops.InstanceGroup, error) {
+	p := r.configBase.Join("instancegroup", name)
+
+	ttl := time.Hour
+	b, err := r.cache.Read(p, ttl)
+	if err != nil {
+		return nil, fmt.Errorf("error loading InstanceGroup %q: %v", p, err)
+	}
+
+	instanceGroup := &kops.InstanceGroup{}
+	if err := utils.YamlUnmarshal(b, instanceGroup); err != nil {
+		return nil, fmt.Errorf("error parsing InstanceGroup %q: %v", p, err)
+	}
+
+	return instanceGroup, nil
+}

cmd/kops-controller/hack/boilerplate.go.txt

@@ -0,0 +1,15 @@
+/*
+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.
+*/