channels: move from protokube to a static pod

Makefile

@@ -78,6 +78,8 @@ KOPS_CONTROLLER_TAG=$(IMAGE_TAG)
 KUBE_APISERVER_HEALTHCHECK_TAG=$(IMAGE_TAG)
 #   discovery/cmd/discovery-server/
 DISCOVERY_SERVER_TAG=$(IMAGE_TAG)
+#   nodeup/pkg/model/channels.go
+KOPS_CHANNELS_TAG=$(IMAGE_TAG)
 
 CGO_ENABLED=0
 export CGO_ENABLED
@@ -111,7 +113,7 @@ nodeup-install: nodeup
 all-install: all kops-install channels-install nodeup-install
 
 .PHONY: all
-all: kops protokube nodeup channels ko-kops-controller-export ko-dns-controller-export ko-kops-utils-cp-export ko-kube-apiserver-healthcheck-export ko-discovery-server-export
+all: kops protokube nodeup channels ko-kops-controller-export ko-kops-channels-export ko-dns-controller-export ko-kops-utils-cp-export ko-kube-apiserver-healthcheck-export ko-discovery-server-export
 
 include tests/e2e/e2e.mk
 
@@ -528,6 +530,17 @@ ko-kops-controller-export-linux-amd64 ko-kops-controller-export-linux-arm64: ko-
 ko-kops-controller-export: ko-kops-controller-export-linux-amd64 ko-kops-controller-export-linux-arm64
 	echo "Done exporting kops-controller images"
 
+.PHONY: ko-kops-channels-export-linux-amd64 ko-kops-channels-export-linux-arm64
+ko-kops-channels-export-linux-amd64 ko-kops-channels-export-linux-arm64: ko-kops-channels-export-linux-%:
+	mkdir -p ${IMAGES}
+	KO_DOCKER_REPO="registry.k8s.io/kops" ${KO} build --tags ${KOPS_CHANNELS_TAG} --platform=linux/$* -B --push=false --tarball=${IMAGES}/kops-channels-$*.tar ./channels/cmd/channels/
+	gzip -f ${IMAGES}/kops-channels-$*.tar
+	tools/sha256 ${IMAGES}/kops-channels-$*.tar.gz ${IMAGES}/kops-channels-$*.tar.gz.sha256
+
+.PHONY: ko-kops-channels-export
+ko-kops-channels-export: ko-kops-channels-export-linux-amd64 ko-kops-channels-export-linux-arm64
+	echo "Done exporting kops-channels images"
+
 .PHONY: ko-kube-apiserver-healthcheck-export-linux-amd64 ko-kube-apiserver-healthcheck-export-linux-arm64
 ko-kube-apiserver-healthcheck-export-linux-amd64 ko-kube-apiserver-healthcheck-export-linux-arm64: ko-kube-apiserver-healthcheck-export-linux-%:
 	mkdir -p ${IMAGES}
@@ -693,6 +706,23 @@ dev-upload-kops-controller: version-dist-kops-controller
 dev-upload-kops-controller-amd64 dev-upload-kops-controller-arm64: dev-upload-kops-controller-%: version-dist-kops-controller-%
 	${UPLOAD_CMD} ${UPLOAD}/ ${UPLOAD_DEST}
 
+# dev-upload-kops-channels uploads kops-channels
+.PHONY: version-dist-kops-channels version-dist-kops-channels-amd64 version-dist-kops-channels-arm64
+version-dist-kops-channels: version-dist-kops-channels-amd64 version-dist-kops-channels-arm64
+
+version-dist-kops-channels-amd64 version-dist-kops-channels-arm64: version-dist-kops-channels-%: ko-kops-channels-export-linux-%
+	mkdir -p ${UPLOAD}/kops/${VERSION}/images/
+	cp -fp ${IMAGES}/kops-channels-$*.tar.gz ${UPLOAD}/kops/${VERSION}/images/kops-channels-$*.tar.gz
+	cp -fp ${IMAGES}/kops-channels-$*.tar.gz.sha256 ${UPLOAD}/kops/${VERSION}/images/kops-channels-$*.tar.gz.sha256
+
+.PHONY: dev-upload-kops-channels
+dev-upload-kops-channels: version-dist-kops-channels
+	${UPLOAD_CMD} ${UPLOAD}/ ${UPLOAD_DEST}
+
+.PHONY: dev-upload-kops-channels-amd64 dev-upload-kops-channels-arm64
+dev-upload-kops-channels-amd64 dev-upload-kops-channels-arm64: dev-upload-kops-channels-%: version-dist-kops-channels-%
+	${UPLOAD_CMD} ${UPLOAD}/ ${UPLOAD_DEST}
+
 # dev-upload-kube-apiserver-healthcheck uploads kube-apiserver-healthcheck
 .PHONY: version-dist-kube-apiserver-healthcheck version-dist-kube-apiserver-healthcheck-amd64 version-dist-kube-apiserver-healthcheck-arm64
 version-dist-kube-apiserver-healthcheck: version-dist-kube-apiserver-healthcheck-amd64 version-dist-kube-apiserver-healthcheck-arm64
@@ -765,7 +795,7 @@ dev-upload-discovery-server-amd64 dev-upload-discovery-server-arm64: dev-upload-
 .PHONY: dev-version-dist dev-version-dist-amd64 dev-version-dist-arm64
 dev-version-dist: dev-version-dist-amd64 dev-version-dist-arm64
 
-dev-version-dist-amd64 dev-version-dist-arm64: dev-version-dist-%: version-dist-nodeup-% version-dist-channels-% version-dist-protokube-% version-dist-kops-controller-% version-dist-kube-apiserver-healthcheck-% version-dist-dns-controller-% version-dist-kops-utils-cp-% version-dist-discovery-server-%
+dev-version-dist-amd64 dev-version-dist-arm64: dev-version-dist-%: version-dist-nodeup-% version-dist-channels-% version-dist-protokube-% version-dist-kops-controller-% version-dist-kops-channels-% version-dist-kube-apiserver-healthcheck-% version-dist-dns-controller-% version-dist-kops-utils-cp-% version-dist-discovery-server-%
 
 .PHONY: dev-upload-linux-amd64 dev-upload-linux-arm64
 dev-upload-linux-amd64 dev-upload-linux-arm64: dev-upload-linux-%: dev-version-dist-%
@@ -791,6 +821,16 @@ kops-controller-push: ko-kops-controller-push
 ko-kops-controller-push:
 	KO_DOCKER_REPO="${DOCKER_REGISTRY}/${DOCKER_IMAGE_PREFIX}kops-controller" ${KO} build --tags ${KOPS_CONTROLLER_TAG} --platform=linux/amd64,linux/arm64 --bare ./cmd/kops-controller/
 
+#------------------------------------------------------
+# kops-channels
+
+.PHONY: kops-channels-push
+kops-channels-push: ko-kops-channels-push
+
+.PHONY: ko-kops-channels-push
+ko-kops-channels-push:
+	KO_DOCKER_REPO="${DOCKER_REGISTRY}/${DOCKER_IMAGE_PREFIX}channels" ${KO} build --tags ${KOPS_CHANNELS_TAG} --platform=linux/amd64,linux/arm64 --bare ./channels/cmd/channels/
+
 #------------------------------------------------------
 # kube-apiserver-healthcheck
 

channels/pkg/cmd/apply_channel.go

@@ -22,6 +22,9 @@ import (
 	"io"
 	"net/url"
 	"os"
+	"os/signal"
+	"syscall"
+	"time"
 
 	"github.com/blang/semver/v4"
 	certmanager "github.com/cert-manager/cert-manager/pkg/client/clientset/versioned"
@@ -31,33 +34,89 @@ import (
 	"k8s.io/client-go/dynamic"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/restmapper"
+	"k8s.io/klog/v2"
 
 	"k8s.io/kops/channels/pkg/channels"
+	"k8s.io/kops/channels/pkg/nodelabeler"
 	"k8s.io/kops/util/pkg/tables"
 	"k8s.io/kops/util/pkg/vfs"
 )
 
 type ApplyChannelOptions struct {
-	Yes bool
+	Yes      bool
+	Interval time.Duration
+	NodeName string
 }
 
 func NewCmdApplyChannel(f *ChannelsFactory, out io.Writer) *cobra.Command {
 	var options ApplyChannelOptions
 
 	cmd := &cobra.Command{
-		Use:   "channel CHANNEL",
-		Short: "Applies updates from the given channel",
+		Use:   "channel CHANNEL...",
+		Short: "Applies updates from the given channel(s)",
 		RunE: func(cmd *cobra.Command, args []string) error {
-			ctx := context.TODO()
-			return RunApplyChannel(ctx, f, out, &options, args)
+			if options.Interval > 0 {
+				ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
+				defer cancel()
+				return runApplyChannelLoop(ctx, out, &options, args)
+			}
+			return runApplyChannelIteration(context.TODO(), f, out, &options, args)
 		},
 	}
 
 	cmd.Flags().BoolVar(&options.Yes, "yes", false, "Apply update")
+	cmd.Flags().DurationVar(&options.Interval, "interval", 0, "If non-zero, re-apply the channel on this interval until interrupted (e.g. 60s)")
+	cmd.Flags().StringVar(&options.NodeName, "node-name", "", "If set, patch the named node with the mandatory control-plane labels each iteration; typically supplied via the downward API.")
 
 	return cmd
 }
 
+// runApplyChannelIteration patches node labels (when --node-name is set) then
+// applies the channel. Labels go first so addons targeting the control-plane
+// label can schedule on the local node as soon as their manifests land.
+func runApplyChannelIteration(ctx context.Context, f *ChannelsFactory, out io.Writer, options *ApplyChannelOptions, args []string) error {
+	var merr error
+	if options.NodeName != "" {
+		labelerClient, err := f.KubernetesClient()
+		if err != nil {
+			merr = multierr.Append(merr, fmt.Errorf("building kubernetes client for node labeler: %w", err))
+		} else if err := nodelabeler.BootstrapControlPlaneNodeLabels(ctx, labelerClient, options.NodeName); err != nil {
+			merr = multierr.Append(merr, fmt.Errorf("bootstrapping node labels: %w", err))
+		}
+	}
+	if err := RunApplyChannel(ctx, f, out, options, args); err != nil {
+		merr = multierr.Append(merr, err)
+	}
+	return merr
+}
+
+// runApplyChannelLoop reconciles repeatedly until ctx is cancelled. A fresh
+// ChannelsFactory per iteration drops cached REST configs and the discovery
+// cache, picking up cert rotation and new CRDs without a restart.
+func runApplyChannelLoop(ctx context.Context, out io.Writer, options *ApplyChannelOptions, args []string) error {
+	// Retry quickly until the first success: the apiserver is usually
+	// unreachable while the control plane is still coming up.
+	const startupRetryInterval = 5 * time.Second
+
+	settled := false
+	for {
+		interval := options.Interval
+		if err := runApplyChannelIteration(ctx, NewChannelsFactory(), out, options, args); err != nil {
+			if !settled {
+				interval = min(startupRetryInterval, options.Interval)
+			}
+			klog.Warningf("error in apply iteration (will retry in %s): %v", interval, err)
+		} else {
+			settled = true
+		}
+		select {
+		case <-ctx.Done():
+			return nil
+		case <-time.After(interval):
+		}
+	}
+}
+
 func RunApplyChannel(ctx context.Context, f *ChannelsFactory, out io.Writer, options *ApplyChannelOptions, args []string) error {
 	restConfig, err := f.RESTConfig()
 	if err != nil {
@@ -68,7 +127,7 @@ func RunApplyChannel(ctx context.Context, f *ChannelsFactory, out io.Writer, opt
 		return err
 	}
 
-	k8sClient, err := kubernetes.NewForConfigAndClient(restConfig, httpClient)
+	k8sClient, err := f.KubernetesClient()
 	if err != nil {
 		return fmt.Errorf("building kube client: %w", err)
 	}
@@ -101,19 +160,22 @@ func RunApplyChannel(ctx context.Context, f *ChannelsFactory, out io.Writer, opt
 	// Remove Pre and Patch, as they make semver comparisons impractical
 	kubernetesVersion.Pre = nil
 
-	if len(args) != 1 {
-		return fmt.Errorf("unexpected number of arguments. Only one channel may be processed at the same time")
+	if len(args) == 0 {
+		return fmt.Errorf("at least one channel URL is required")
 	}
 
-	channelLocation := args[0]
-
-	// menu is the expected list of addons in the cluster and their configurations.
-	menu, err := buildMenu(f.VFSContext(), kubernetesVersion, channelLocation)
-	if err != nil {
-		return fmt.Errorf("cannot build the addon menu from args: %w", err)
+	var merr error
+	for _, channelLocation := range args {
+		menu, err := buildMenu(f.VFSContext(), kubernetesVersion, channelLocation)
+		if err != nil {
+			merr = multierr.Append(merr, fmt.Errorf("building menu for %q: %w", channelLocation, err))
+			continue
+		}
+		if err := applyMenu(ctx, menu, f.VFSContext(), k8sClient, cmClient, dynamicClient, restMapper, options.Yes); err != nil {
+			merr = multierr.Append(merr, fmt.Errorf("applying %q: %w", channelLocation, err))
+		}
 	}
-
-	return applyMenu(ctx, menu, f.VFSContext(), k8sClient, cmClient, dynamicClient, restMapper, options.Yes)
+	return merr
 }
 
 func applyMenu(ctx context.Context, menu *channels.AddonMenu, vfsContext *vfs.VFSContext, k8sClient kubernetes.Interface, cmClient certmanager.Interface, dynamicClient dynamic.Interface, restMapper *restmapper.DeferredDiscoveryRESTMapper, apply bool) error {

channels/pkg/cmd/factory.go

@@ -22,6 +22,7 @@ import (
 
 	"k8s.io/cli-runtime/pkg/genericclioptions"
 	"k8s.io/client-go/dynamic"
+	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/restmapper"
 	"k8s.io/kops/util/pkg/vfs"
@@ -36,6 +37,7 @@ type ChannelsFactory struct {
 	vfsContext       *vfs.VFSContext
 	restMapper       *restmapper.DeferredDiscoveryRESTMapper
 	dynamicClient    dynamic.Interface
+	kubernetesClient kubernetes.Interface
 }
 
 func NewChannelsFactory() *ChannelsFactory {
@@ -108,6 +110,25 @@ func (f *ChannelsFactory) DynamicClient() (dynamic.Interface, error) {
 	return f.dynamicClient, nil
 }
 
+func (f *ChannelsFactory) KubernetesClient() (kubernetes.Interface, error) {
+	if f.kubernetesClient == nil {
+		restConfig, err := f.RESTConfig()
+		if err != nil {
+			return nil, err
+		}
+		httpClient, err := f.HTTPClient()
+		if err != nil {
+			return nil, err
+		}
+		kubernetesClient, err := kubernetes.NewForConfigAndClient(restConfig, httpClient)
+		if err != nil {
+			return nil, err
+		}
+		f.kubernetesClient = kubernetesClient
+	}
+	return f.kubernetesClient, nil
+}
+
 func (f *ChannelsFactory) VFSContext() *vfs.VFSContext {
 	if f.vfsContext == nil {
 		// TODO vfs.NewVFSContext()

channels/pkg/nodelabeler/labeler.go

@@ -0,0 +1,78 @@
+/*
+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 nodelabeler
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/klog/v2"
+	"k8s.io/kops/pkg/nodelabels"
+)
+
+type nodePatch struct {
+	Metadata *nodePatchMetadata `json:"metadata,omitempty"`
+}
+
+type nodePatchMetadata struct {
+	Labels map[string]string `json:"labels,omitempty"`
+}
+
+// BootstrapControlPlaneNodeLabels applies labels to the current node so that it acts as a control-plane.
+// Safe to call repeatedly: the patch is skipped when the labels already match.
+func BootstrapControlPlaneNodeLabels(ctx context.Context, client kubernetes.Interface, nodeName string) error {
+	if nodeName == "" {
+		return fmt.Errorf("node name is required")
+	}
+
+	klog.V(2).Infof("querying k8s for node %q", nodeName)
+	node, err := client.CoreV1().Nodes().Get(ctx, nodeName, metav1.GetOptions{})
+	if err != nil {
+		return fmt.Errorf("querying node %q: %w", nodeName, err)
+	}
+
+	labels := nodelabels.BuildMandatoryControlPlaneLabels()
+
+	shouldPatch := false
+	for k, v := range labels {
+		if actual, found := node.Labels[k]; !found || actual != v {
+			shouldPatch = true
+			break
+		}
+	}
+	if !shouldPatch {
+		return nil
+	}
+
+	klog.V(2).Infof("patching node %q to add labels %v", nodeName, labels)
+	patch, err := json.Marshal(&nodePatch{
+		Metadata: &nodePatchMetadata{Labels: labels},
+	})
+	if err != nil {
+		return fmt.Errorf("building node patch: %w", err)
+	}
+
+	klog.V(2).Infof("sending patch for node %q: %s", node.Name, patch)
+	if _, err := client.CoreV1().Nodes().Patch(ctx, node.Name, types.StrategicMergePatchType, patch, metav1.PatchOptions{FieldManager: "kops-channels"}); err != nil {
+		return fmt.Errorf("patching node %q: %w", node.Name, err)
+	}
+	return nil
+}

cloudbuild.yaml

@@ -18,6 +18,7 @@ steps:
     args:
       - kops-utils-cp-push
       - kops-controller-push
+      - kops-channels-push
       - dns-controller-push
       - kube-apiserver-healthcheck-push
       - discovery-server-push