Delete route controller

cmd/cloud-controller-manager/app/controllermanager.go

@@ -19,9 +19,7 @@ package app
 import (
 	"context"
 	"fmt"
-	"net"
 	"os"
-	"strings"
 	"time"
 
 	"github.com/golang/glog"
@@ -39,7 +37,6 @@ import (
 	genericcontrollermanager "k8s.io/kubernetes/cmd/controller-manager/app"
 	"k8s.io/kubernetes/pkg/cloudprovider"
 	cloudcontrollers "k8s.io/kubernetes/pkg/controller/cloud"
-	routecontroller "k8s.io/kubernetes/pkg/controller/route"
 	servicecontroller "k8s.io/kubernetes/pkg/controller/service"
 	"k8s.io/kubernetes/pkg/util/configz"
 	utilflag "k8s.io/kubernetes/pkg/util/flag"
@@ -230,27 +227,6 @@ func startControllers(c *cloudcontrollerconfig.CompletedConfig, stop <-chan stru
 		time.Sleep(wait.Jitter(c.ComponentConfig.Generic.ControllerStartInterval.Duration, ControllerStartJitter))
 	}
 
-	// If CIDRs should be allocated for pods and set on the CloudProvider, then start the route controller
-	if c.ComponentConfig.KubeCloudShared.AllocateNodeCIDRs && c.ComponentConfig.KubeCloudShared.ConfigureCloudRoutes {
-		if routes, ok := cloud.Routes(); !ok {
-			glog.Warning("configure-cloud-routes is set, but cloud provider does not support routes. Will not configure cloud provider routes.")
-		} else {
-			var clusterCIDR *net.IPNet
-			if len(strings.TrimSpace(c.ComponentConfig.KubeCloudShared.ClusterCIDR)) != 0 {
-				_, clusterCIDR, err = net.ParseCIDR(c.ComponentConfig.KubeCloudShared.ClusterCIDR)
-				if err != nil {
-					glog.Warningf("Unsuccessful parsing of cluster CIDR %v: %v", c.ComponentConfig.KubeCloudShared.ClusterCIDR, err)
-				}
-			}
-
-			routeController := routecontroller.New(routes, client("route-controller"), c.SharedInformers.Core().V1().Nodes(), c.ComponentConfig.KubeCloudShared.ClusterName, clusterCIDR)
-			go routeController.Run(stop, c.ComponentConfig.KubeCloudShared.RouteReconciliationPeriod.Duration)
-			time.Sleep(wait.Jitter(c.ComponentConfig.Generic.ControllerStartInterval.Duration, ControllerStartJitter))
-		}
-	} else {
-		glog.Infof("Will not configure cloud provider routes for allocate-node-cidrs: %v, configure-cloud-routes: %v.", c.ComponentConfig.KubeCloudShared.AllocateNodeCIDRs, c.ComponentConfig.KubeCloudShared.ConfigureCloudRoutes)
-	}
-
 	// If apiserver is not running we should wait for some time and fail only then. This is particularly
 	// important when we start apiserver and controller manager at the same time.
 	err = genericcontrollermanager.WaitForAPIServer(c.VersionedClient, 10*time.Second)

cmd/kube-controller-manager/app/controllermanager.go

@@ -364,7 +364,6 @@ func NewControllerInitializers(loopMode ControllerLoopMode) map[string]InitFunc
 	controllers["nodeipam"] = startNodeIpamController
 	if loopMode == IncludeCloudLoops {
 		controllers["service"] = startServiceController
-		controllers["route"] = startRouteController
 		// TODO: volume controller into the IncludeCloudLoops only set.
 		// TODO: Separate cluster in cloud check from node lifecycle controller.
 	}

cmd/kube-controller-manager/app/core.go

@@ -47,7 +47,6 @@ import (
 	"k8s.io/kubernetes/pkg/controller/podgc"
 	replicationcontroller "k8s.io/kubernetes/pkg/controller/replication"
 	resourcequotacontroller "k8s.io/kubernetes/pkg/controller/resourcequota"
-	routecontroller "k8s.io/kubernetes/pkg/controller/route"
 	servicecontroller "k8s.io/kubernetes/pkg/controller/service"
 	serviceaccountcontroller "k8s.io/kubernetes/pkg/controller/serviceaccount"
 	ttlcontroller "k8s.io/kubernetes/pkg/controller/ttl"
@@ -145,29 +144,6 @@ func startNodeLifecycleController(ctx ControllerContext) (http.Handler, bool, er
 	return nil, true, nil
 }
 
-func startRouteController(ctx ControllerContext) (http.Handler, bool, error) {
-	if !ctx.ComponentConfig.KubeCloudShared.AllocateNodeCIDRs || !ctx.ComponentConfig.KubeCloudShared.ConfigureCloudRoutes {
-		glog.Infof("Will not configure cloud provider routes for allocate-node-cidrs: %v, configure-cloud-routes: %v.", ctx.ComponentConfig.KubeCloudShared.AllocateNodeCIDRs, ctx.ComponentConfig.KubeCloudShared.ConfigureCloudRoutes)
-		return nil, false, nil
-	}
-	if ctx.Cloud == nil {
-		glog.Warning("configure-cloud-routes is set, but no cloud provider specified. Will not configure cloud provider routes.")
-		return nil, false, nil
-	}
-	routes, ok := ctx.Cloud.Routes()
-	if !ok {
-		glog.Warning("configure-cloud-routes is set, but cloud provider does not support routes. Will not configure cloud provider routes.")
-		return nil, false, nil
-	}
-	_, clusterCIDR, err := net.ParseCIDR(ctx.ComponentConfig.KubeCloudShared.ClusterCIDR)
-	if err != nil {
-		glog.Warningf("Unsuccessful parsing of cluster CIDR %v: %v", ctx.ComponentConfig.KubeCloudShared.ClusterCIDR, err)
-	}
-	routeController := routecontroller.New(routes, ctx.ClientBuilder.ClientOrDie("route-controller"), ctx.InformerFactory.Core().V1().Nodes(), ctx.ComponentConfig.KubeCloudShared.ClusterName, clusterCIDR)
-	go routeController.Run(ctx.Stop, ctx.ComponentConfig.KubeCloudShared.RouteReconciliationPeriod.Duration)
-	return nil, true, nil
-}
-
 func startPersistentVolumeBinderController(ctx ControllerContext) (http.Handler, bool, error) {
 	params := persistentvolumecontroller.ControllerParameters{
 		KubeClient:                ctx.ClientBuilder.ClientOrDie("persistent-volume-binder"),

pkg/controller/route/BUILD

-package(default_visibility = ["//visibility:public"])
-
-load(
-    "@io_bazel_rules_go//go:def.bzl",
-    "go_library",
-    "go_test",
-)
-
-go_library(
-    name = "go_default_library",
-    srcs = [
-        "doc.go",
-        "route_controller.go",
-    ],
-    importpath = "k8s.io/kubernetes/pkg/controller/route",
-    deps = [
-        "//pkg/api/v1/node:go_default_library",
-        "//pkg/cloudprovider:go_default_library",
-        "//pkg/controller:go_default_library",
-        "//pkg/util/metrics:go_default_library",
-        "//pkg/util/node:go_default_library",
-        "//staging/src/k8s.io/api/core/v1:go_default_library",
-        "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
-        "//staging/src/k8s.io/apimachinery/pkg/labels:go_default_library",
-        "//staging/src/k8s.io/apimachinery/pkg/types:go_default_library",
-        "//staging/src/k8s.io/apimachinery/pkg/util/runtime:go_default_library",
-        "//staging/src/k8s.io/apimachinery/pkg/util/wait:go_default_library",
-        "//staging/src/k8s.io/client-go/informers/core/v1:go_default_library",
-        "//staging/src/k8s.io/client-go/kubernetes:go_default_library",
-        "//staging/src/k8s.io/client-go/kubernetes/scheme:go_default_library",
-        "//staging/src/k8s.io/client-go/kubernetes/typed/core/v1:go_default_library",
-        "//staging/src/k8s.io/client-go/listers/core/v1:go_default_library",
-        "//staging/src/k8s.io/client-go/tools/cache:go_default_library",
-        "//staging/src/k8s.io/client-go/tools/record:go_default_library",
-        "//staging/src/k8s.io/client-go/util/retry:go_default_library",
-        "//vendor/github.com/golang/glog:go_default_library",
-    ],
-)
-
-go_test(
-    name = "go_default_test",
-    srcs = ["route_controller_test.go"],
-    embed = [":go_default_library"],
-    deps = [
-        "//pkg/api/v1/node:go_default_library",
-        "//pkg/cloudprovider:go_default_library",
-        "//pkg/cloudprovider/providers/fake:go_default_library",
-        "//pkg/controller:go_default_library",
-        "//staging/src/k8s.io/api/core/v1:go_default_library",
-        "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
-        "//staging/src/k8s.io/apimachinery/pkg/types:go_default_library",
-        "//staging/src/k8s.io/client-go/informers:go_default_library",
-        "//staging/src/k8s.io/client-go/kubernetes/fake:go_default_library",
-        "//staging/src/k8s.io/client-go/testing:go_default_library",
-    ],
-)
-
-filegroup(
-    name = "package-srcs",
-    srcs = glob(["**"]),
-    tags = ["automanaged"],
-    visibility = ["//visibility:private"],
-)
-
-filegroup(
-    name = "all-srcs",
-    srcs = [":package-srcs"],
-    tags = ["automanaged"],
-)

pkg/controller/route/OWNERS

-approvers:
-- gmarek
-- wojtek-t
-- bowei
-reviewers:
-- gmarek
-- wojtek-t

pkg/controller/route/doc.go

-/*
-Copyright 2015 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 route contains code for syncing cloud routing rules with
-// the list of registered nodes.
-package route // import "k8s.io/kubernetes/pkg/controller/route"

pkg/controller/route/route_controller.go

-/*
-Copyright 2015 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 route
-
-import (
-	"context"
-	"fmt"
-	"net"
-	"sync"
-	"time"
-
-	"github.com/golang/glog"
-
-	"k8s.io/api/core/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/labels"
-	"k8s.io/apimachinery/pkg/types"
-	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
-	"k8s.io/apimachinery/pkg/util/wait"
-	coreinformers "k8s.io/client-go/informers/core/v1"
-	clientset "k8s.io/client-go/kubernetes"
-	"k8s.io/client-go/kubernetes/scheme"
-	v1core "k8s.io/client-go/kubernetes/typed/core/v1"
-	corelisters "k8s.io/client-go/listers/core/v1"
-	"k8s.io/client-go/tools/cache"
-	"k8s.io/client-go/tools/record"
-	clientretry "k8s.io/client-go/util/retry"
-	v1node "k8s.io/kubernetes/pkg/api/v1/node"
-	"k8s.io/kubernetes/pkg/cloudprovider"
-	"k8s.io/kubernetes/pkg/controller"
-	"k8s.io/kubernetes/pkg/util/metrics"
-	nodeutil "k8s.io/kubernetes/pkg/util/node"
-)
-
-const (
-	// Maximal number of concurrent CreateRoute API calls.
-	// TODO: This should be per-provider.
-	maxConcurrentRouteCreations int = 200
-)
-
-var updateNetworkConditionBackoff = wait.Backoff{
-	Steps:    5, // Maximum number of retries.
-	Duration: 100 * time.Millisecond,
-	Jitter:   1.0,
-}
-
-type RouteController struct {
-	routes           cloudprovider.Routes
-	kubeClient       clientset.Interface
-	clusterName      string
-	clusterCIDR      *net.IPNet
-	nodeLister       corelisters.NodeLister
-	nodeListerSynced cache.InformerSynced
-	broadcaster      record.EventBroadcaster
-	recorder         record.EventRecorder
-}
-
-func New(routes cloudprovider.Routes, kubeClient clientset.Interface, nodeInformer coreinformers.NodeInformer, clusterName string, clusterCIDR *net.IPNet) *RouteController {
-	if kubeClient != nil && kubeClient.CoreV1().RESTClient().GetRateLimiter() != nil {
-		metrics.RegisterMetricAndTrackRateLimiterUsage("route_controller", kubeClient.CoreV1().RESTClient().GetRateLimiter())
-	}
-
-	if clusterCIDR == nil {
-		glog.Fatal("RouteController: Must specify clusterCIDR.")
-	}
-
-	eventBroadcaster := record.NewBroadcaster()
-	eventBroadcaster.StartLogging(glog.Infof)
-	recorder := eventBroadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: "route_controller"})
-
-	rc := &RouteController{
-		routes:           routes,
-		kubeClient:       kubeClient,
-		clusterName:      clusterName,
-		clusterCIDR:      clusterCIDR,
-		nodeLister:       nodeInformer.Lister(),
-		nodeListerSynced: nodeInformer.Informer().HasSynced,
-		broadcaster:      eventBroadcaster,
-		recorder:         recorder,
-	}
-
-	return rc
-}
-
-func (rc *RouteController) Run(stopCh <-chan struct{}, syncPeriod time.Duration) {
-	defer utilruntime.HandleCrash()
-
-	glog.Info("Starting route controller")
-	defer glog.Info("Shutting down route controller")
-
-	if !controller.WaitForCacheSync("route", stopCh, rc.nodeListerSynced) {
-		return
-	}
-
-	if rc.broadcaster != nil {
-		rc.broadcaster.StartRecordingToSink(&v1core.EventSinkImpl{Interface: rc.kubeClient.CoreV1().Events("")})
-	}
-
-	// TODO: If we do just the full Resync every 5 minutes (default value)
-	// that means that we may wait up to 5 minutes before even starting
-	// creating a route for it. This is bad.
-	// We should have a watch on node and if we observe a new node (with CIDR?)
-	// trigger reconciliation for that node.
-	go wait.NonSlidingUntil(func() {
-		if err := rc.reconcileNodeRoutes(); err != nil {
-			glog.Errorf("Couldn't reconcile node routes: %v", err)
-		}
-	}, syncPeriod, stopCh)
-
-	<-stopCh
-}
-
-func (rc *RouteController) reconcileNodeRoutes() error {
-	routeList, err := rc.routes.ListRoutes(context.TODO(), rc.clusterName)
-	if err != nil {
-		return fmt.Errorf("error listing routes: %v", err)
-	}
-	nodes, err := rc.nodeLister.List(labels.Everything())
-	if err != nil {
-		return fmt.Errorf("error listing nodes: %v", err)
-	}
-	return rc.reconcile(nodes, routeList)
-}
-
-func (rc *RouteController) reconcile(nodes []*v1.Node, routes []*cloudprovider.Route) error {
-	// nodeCIDRs maps nodeName->nodeCIDR
-	nodeCIDRs := make(map[types.NodeName]string)
-	// routeMap maps routeTargetNode->route
-	routeMap := make(map[types.NodeName]*cloudprovider.Route)
-	for _, route := range routes {
-		if route.TargetNode != "" {
-			routeMap[route.TargetNode] = route
-		}
-	}
-
-	wg := sync.WaitGroup{}
-	rateLimiter := make(chan struct{}, maxConcurrentRouteCreations)
-
-	for _, node := range nodes {
-		// Skip if the node hasn't been assigned a CIDR yet.
-		if node.Spec.PodCIDR == "" {
-			continue
-		}
-		nodeName := types.NodeName(node.Name)
-		// Check if we have a route for this node w/ the correct CIDR.
-		r := routeMap[nodeName]
-		if r == nil || r.DestinationCIDR != node.Spec.PodCIDR {
-			// If not, create the route.
-			route := &cloudprovider.Route{
-				TargetNode:      nodeName,
-				DestinationCIDR: node.Spec.PodCIDR,
-			}
-			nameHint := string(node.UID)
-			wg.Add(1)
-			go func(nodeName types.NodeName, nameHint string, route *cloudprovider.Route) {
-				defer wg.Done()
-				err := clientretry.RetryOnConflict(updateNetworkConditionBackoff, func() error {
-					startTime := time.Now()
-					// Ensure that we don't have more than maxConcurrentRouteCreations
-					// CreateRoute calls in flight.
-					rateLimiter <- struct{}{}
-					glog.Infof("Creating route for node %s %s with hint %s, throttled %v", nodeName, route.DestinationCIDR, nameHint, time.Since(startTime))
-					err := rc.routes.CreateRoute(context.TODO(), rc.clusterName, nameHint, route)
-					<-rateLimiter
-
-					rc.updateNetworkingCondition(nodeName, err == nil)
-					if err != nil {
-						msg := fmt.Sprintf("Could not create route %s %s for node %s after %v: %v", nameHint, route.DestinationCIDR, nodeName, time.Since(startTime), err)
-						if rc.recorder != nil {
-							rc.recorder.Eventf(
-								&v1.ObjectReference{
-									Kind:      "Node",
-									Name:      string(nodeName),
-									UID:       types.UID(nodeName),
-									Namespace: "",
-								}, v1.EventTypeWarning, "FailedToCreateRoute", msg)
-						}
-						glog.V(4).Infof(msg)
-						return err
-					}
-					glog.Infof("Created route for node %s %s with hint %s after %v", nodeName, route.DestinationCIDR, nameHint, time.Now().Sub(startTime))
-					return nil
-				})
-				if err != nil {
-					glog.Errorf("Could not create route %s %s for node %s: %v", nameHint, route.DestinationCIDR, nodeName, err)
-				}
-			}(nodeName, nameHint, route)
-		} else {
-			// Update condition only if it doesn't reflect the current state.
-			_, condition := v1node.GetNodeCondition(&node.Status, v1.NodeNetworkUnavailable)
-			if condition == nil || condition.Status != v1.ConditionFalse {
-				rc.updateNetworkingCondition(types.NodeName(node.Name), true)
-			}
-		}
-		nodeCIDRs[nodeName] = node.Spec.PodCIDR
-	}
-	for _, route := range routes {
-		if rc.isResponsibleForRoute(route) {
-			// Check if this route is a blackhole, or applies to a node we know about & has an incorrect CIDR.
-			if route.Blackhole || (nodeCIDRs[route.TargetNode] != route.DestinationCIDR) {
-				wg.Add(1)
-				// Delete the route.
-				go func(route *cloudprovider.Route, startTime time.Time) {
-					defer wg.Done()
-					glog.Infof("Deleting route %s %s", route.Name, route.DestinationCIDR)
-					if err := rc.routes.DeleteRoute(context.TODO(), rc.clusterName, route); err != nil {
-						glog.Errorf("Could not delete route %s %s after %v: %v", route.Name, route.DestinationCIDR, time.Since(startTime), err)
-					} else {
-						glog.Infof("Deleted route %s %s after %v", route.Name, route.DestinationCIDR, time.Since(startTime))
-					}
-				}(route, time.Now())
-			}
-		}
-	}
-	wg.Wait()
-	return nil
-}
-
-func (rc *RouteController) updateNetworkingCondition(nodeName types.NodeName, routeCreated bool) error {
-	err := clientretry.RetryOnConflict(updateNetworkConditionBackoff, func() error {
-		var err error
-		// Patch could also fail, even though the chance is very slim. So we still do
-		// patch in the retry loop.
-		currentTime := metav1.Now()
-		if routeCreated {
-			err = nodeutil.SetNodeCondition(rc.kubeClient, nodeName, v1.NodeCondition{
-				Type:               v1.NodeNetworkUnavailable,
-				Status:             v1.ConditionFalse,
-				Reason:             "RouteCreated",
-				Message:            "RouteController created a route",
-				LastTransitionTime: currentTime,
-			})
-		} else {
-			err = nodeutil.SetNodeCondition(rc.kubeClient, nodeName, v1.NodeCondition{
-				Type:               v1.NodeNetworkUnavailable,
-				Status:             v1.ConditionTrue,
-				Reason:             "NoRouteCreated",
-				Message:            "RouteController failed to create a route",
-				LastTransitionTime: currentTime,
-			})
-		}
-		if err != nil {
-			glog.V(4).Infof("Error updating node %s, retrying: %v", nodeName, err)
-		}
-		return err
-	})
-
-	if err != nil {
-		glog.Errorf("Error updating node %s: %v", nodeName, err)
-	}
-
-	return err
-}
-
-func (rc *RouteController) isResponsibleForRoute(route *cloudprovider.Route) bool {
-	_, cidr, err := net.ParseCIDR(route.DestinationCIDR)
-	if err != nil {
-		glog.Errorf("Ignoring route %s, unparsable CIDR: %v", route.Name, err)
-		return false
-	}
-	// Not responsible if this route's CIDR is not within our clusterCIDR
-	lastIP := make([]byte, len(cidr.IP))
-	for i := range lastIP {
-		lastIP[i] = cidr.IP[i] | ^cidr.Mask[i]
-	}
-	if !rc.clusterCIDR.Contains(cidr.IP) || !rc.clusterCIDR.Contains(lastIP) {
-		return false
-	}
-	return true
-}