router: add support for configuration by ingress

Author: marun

contrib/completions/bash/openshift

@@ -20646,6 +20646,8 @@ _openshift_infra_router()
     local_nonpersistent_flags+=("--default-certificate-path=")
     flags+=("--denied-domains=")
     local_nonpersistent_flags+=("--denied-domains=")
+    flags+=("--enable-ingress")
+    local_nonpersistent_flags+=("--enable-ingress")
     flags+=("--extended-validation")
     local_nonpersistent_flags+=("--extended-validation")
     flags+=("--fields=")

contrib/completions/zsh/openshift

@@ -20794,6 +20794,8 @@ _openshift_infra_router()
     local_nonpersistent_flags+=("--default-certificate-path=")
     flags+=("--denied-domains=")
     local_nonpersistent_flags+=("--denied-domains=")
+    flags+=("--enable-ingress")
+    local_nonpersistent_flags+=("--enable-ingress")
     flags+=("--extended-validation")
     local_nonpersistent_flags+=("--extended-validation")
     flags+=("--fields=")

docs/man/man1/openshift-infra-router.1

@@ -89,6 +89,10 @@ You may restrict the set of routes exposed to a single project (with \-\-namespa
 \fB\-\-denied\-domains\fP=[]
     List of comma separated domains to deny in routes
 
+.PP
+\fB\-\-enable\-ingress\fP=false
+    Enable ingress resource configuration of the router
+
 .PP
 \fB\-\-extended\-validation\fP=true
     If set, then an additional extended validation step is performed on all routes admitted in by this router. Defaults to true and enables the extended validation checks.

pkg/api/helpers.go

@@ -3,6 +3,7 @@ package api
 import (
 	"fmt"
 
+	kapi "k8s.io/kubernetes/pkg/api"
 	"k8s.io/kubernetes/pkg/api/validation"
 	"k8s.io/kubernetes/pkg/api/validation/path"
 )
@@ -33,3 +34,10 @@ func GetFieldLabelConversionFunc(supportedLabels map[string]string, overrideLabe
 		return "", "", fmt.Errorf("field label not supported: %s", label)
 	}
 }
+
+// GetResourceKey returns a string of the form [namespace]/[name] for
+// the given resource.  This is a common way of ensuring a key for a
+// resource that is unique across the cluster.
+func GetResourceKey(obj kapi.ObjectMeta) string {
+	return fmt.Sprintf("%s/%s", obj.Namespace, obj.Name)
+}

pkg/cmd/infra/router/f5.go

@@ -225,7 +225,8 @@ func (o *F5RouterOptions) Run() error {
 
 	factory := o.RouterSelection.NewFactory(oc, kc)
 	watchNodes := (len(o.InternalAddress) != 0 && len(o.VxlanGateway) != 0)
-	controller := factory.Create(plugin, watchNodes)
+	// TODO should ingress be configurable for f5?
+	controller := factory.Create(plugin, watchNodes, false)
 	controller.Run()
 
 	select {}

pkg/cmd/infra/router/router.go

@@ -79,10 +79,22 @@ func (o *RouterSelection) RouteSelectionFunc() controller.RouteHostFunc {
 		if !o.OverrideHostname && len(route.Spec.Host) > 0 {
 			return route.Spec.Host
 		}
+		nameForHost := route.Name
+		if controller.IsGeneratedRoute(route) {
+			// Use the resource name embedded in the route name to
+			// construct the host.  The name of routes generated from
+			// ingress rules will be 'ingress/[name]/[host]/[path]'.
+			//
+			// When a route and ingress in the same namespace share a
+			// name, the route and the ingress' rules should receive
+			// the same generated host.
+			nameParts := strings.Split(nameForHost, "/")
+			nameForHost = nameParts[1]
+		}
 		s, err := variable.ExpandStrict(o.HostnameTemplate, func(key string) (string, bool) {
 			switch key {
 			case "name":
-				return route.Name, true
+				return nameForHost, true
 			case "namespace":
 				return route.Namespace, true
 			default:

pkg/cmd/infra/router/template.go

@@ -63,6 +63,7 @@ type TemplateRouter struct {
 	ExtendedValidation     bool
 	RouterService          *ktypes.NamespacedName
 	BindPortsAfterSync     bool
+	EnableIngress          bool
 }
 
 // reloadInterval returns how often to run the router reloads. The interval
@@ -88,6 +89,7 @@ func (o *TemplateRouter) Bind(flag *pflag.FlagSet) {
 	flag.DurationVar(&o.ReloadInterval, "interval", reloadInterval(), "Controls how often router reloads are invoked. Mutiple router reload requests are coalesced for the duration of this interval since the last reload time.")
 	flag.BoolVar(&o.ExtendedValidation, "extended-validation", util.Env("EXTENDED_VALIDATION", "true") == "true", "If set, then an additional extended validation step is performed on all routes admitted in by this router. Defaults to true and enables the extended validation checks.")
 	flag.BoolVar(&o.BindPortsAfterSync, "bind-ports-after-sync", util.Env("ROUTER_BIND_PORTS_AFTER_SYNC", "") == "true", "Bind ports only after route state has been synchronized")
+	flag.BoolVar(&o.EnableIngress, "enable-ingress", util.Env("ROUTER_ENABLE_INGRESS", "") == "true", "Enable ingress resource configuration of the router")
 }
 
 type RouterStats struct {
@@ -215,7 +217,7 @@ func (o *TemplateRouterOptions) Run() error {
 	plugin := controller.NewHostAdmitter(uniqueHostPlugin, o.RouteAdmissionFunc(), o.RestrictSubdomainOwnership, controller.RejectionRecorder(statusPlugin))
 
 	factory := o.RouterSelection.NewFactory(oc, kc)
-	controller := factory.Create(plugin, false)
+	controller := factory.Create(plugin, false, o.EnableIngress)
 	controller.Run()
 
 	proc.StartReaper()

pkg/route/api/v1/defaults.go

@@ -2,6 +2,10 @@ package v1
 
 import "k8s.io/kubernetes/pkg/runtime"
 
+// If adding or changing route defaults, updates may be required to
+// pkg/router/controller/controller.go to ensure the routes generated from
+// ingress resources will match routes created via the api.
+
 func SetDefaults_RouteSpec(obj *RouteSpec) {
 	if len(obj.WildcardPolicy) == 0 {
 		obj.WildcardPolicy = WildcardPolicyNone

pkg/router/controller/controller.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/golang/glog"
 	kapi "k8s.io/kubernetes/pkg/api"
+	"k8s.io/kubernetes/pkg/apis/extensions"
 	utilruntime "k8s.io/kubernetes/pkg/util/runtime"
 	"k8s.io/kubernetes/pkg/util/sets"
 	utilwait "k8s.io/kubernetes/pkg/util/wait"
@@ -30,25 +31,38 @@ type RouterController struct {
 	NextRoute     func() (watch.EventType, *routeapi.Route, error)
 	NextNode      func() (watch.EventType, *kapi.Node, error)
 	NextEndpoints func() (watch.EventType, *kapi.Endpoints, error)
+	NextIngress   func() (watch.EventType, *extensions.Ingress, error)
+	NextSecret    func() (watch.EventType, *kapi.Secret, error)
 
 	RoutesListConsumed    func() bool
 	EndpointsListConsumed func() bool
+	IngressesListConsumed func() bool
+	SecretsListConsumed   func() bool
 	routesListConsumed    bool
 	endpointsListConsumed bool
+	ingressesListConsumed bool
+	secretsListConsumed   bool
 	filteredByNamespace   bool
 	syncing               bool
 
 	RoutesListSuccessfulAtLeastOnce    func() bool
 	EndpointsListSuccessfulAtLeastOnce func() bool
+	IngressesListSuccessfulAtLeastOnce func() bool
+	SecretsListSuccessfulAtLeastOnce   func() bool
 	RoutesListCount                    func() int
 	EndpointsListCount                 func() int
+	IngressesListCount                 func() int
+	SecretsListCount                   func() int
 
 	WatchNodes bool
 
 	Namespaces            NamespaceLister
 	NamespaceSyncInterval time.Duration
 	NamespaceWaitInterval time.Duration
 	NamespaceRetries      int
+
+	EnableIngress     bool
+	IngressTranslator *IngressTranslator
 }
 
 // Run begins watching and syncing.
@@ -63,6 +77,8 @@ func (c *RouterController) Run() {
 	if c.WatchNodes {
 		go utilwait.Forever(c.HandleNode, 0)
 	}
+	go utilwait.Forever(c.HandleIngress, 0)
+	go utilwait.Forever(c.HandleSecret, 0)
 	go c.watchForFirstSync()
 }
 
@@ -74,22 +90,31 @@ func (c *RouterController) handleFirstSync() bool {
 
 	synced := c.RoutesListSuccessfulAtLeastOnce() &&
 		c.EndpointsListSuccessfulAtLeastOnce() &&
-		(c.Namespaces == nil || c.filteredByNamespace)
+		(c.Namespaces == nil || c.filteredByNamespace) &&
+		(!c.EnableIngress ||
+			(c.IngressesListSuccessfulAtLeastOnce() && c.SecretsListSuccessfulAtLeastOnce()))
 	if !synced {
 		return false
 	}
 
-	// If either of the event queues were empty after the initial
-	// List, the tracking listConsumed variable's default value of
-	// 'false' may prevent the router from committing the readiness
-	// status.  Set the value to 'true' to ensure that state will be
-	// committed if necessary.
+	// If any of the event queues were empty after the initial List,
+	// the tracking listConsumed variable's default value of 'false'
+	// may prevent the router from committing.  Set the value to
+	// 'true' to ensure that state can be committed if necessary.
 	if c.RoutesListCount() == 0 {
 		c.routesListConsumed = true
 	}
 	if c.EndpointsListCount() == 0 {
 		c.endpointsListConsumed = true
 	}
+	if c.EnableIngress {
+		if c.IngressesListCount() == 0 {
+			c.ingressesListConsumed = true
+		}
+		if c.SecretsListCount() == 0 {
+			c.secretsListConsumed = true
+		}
+	}
 	c.commit()
 
 	return true
@@ -109,6 +134,14 @@ func (c *RouterController) HandleNamespaces() {
 	for i := 0; i < c.NamespaceRetries; i++ {
 		namespaces, err := c.Namespaces.NamespaceNames()
 		if err == nil {
+
+			// The ingress translator synchronizes access to its cache with a
+			// lock, so calls to it are made outside of the controller lock to
+			// avoid unintended interaction.
+			if c.EnableIngress {
+				c.IngressTranslator.UpdateNamespaces(namespaces)
+			}
+
 			c.lock.Lock()
 			defer c.lock.Unlock()
 
@@ -161,13 +194,7 @@ func (c *RouterController) HandleRoute() {
 	c.lock.Lock()
 	defer c.lock.Unlock()
 
-	glog.V(4).Infof("Processing Route: %s -> %s", route.Name, route.Spec.To.Name)
-	glog.V(4).Infof("           Alias: %s", route.Spec.Host)
-	glog.V(4).Infof("           Event: %s", eventType)
-
-	if err := c.Plugin.HandleRoute(eventType, route); err != nil {
-		utilruntime.HandleError(err)
-	}
+	c.processRoute(eventType, route)
 
 	// Change the local sync state within the lock to ensure that all
 	// event handlers have the same view of sync state.
@@ -196,10 +223,61 @@ func (c *RouterController) HandleEndpoints() {
 	c.commit()
 }
 
+// HandleIngress handles a single Ingress event and synchronizes the router backend.
+func (c *RouterController) HandleIngress() {
+	eventType, ingress, err := c.NextIngress()
+	if err != nil {
+		utilruntime.HandleError(fmt.Errorf("unable to read ingress: %v", err))
+		return
+	}
+
+	// The ingress translator synchronizes access to its cache with a
+	// lock, so calls to it are made outside of the controller lock to
+	// avoid unintended interaction.
+	events := c.IngressTranslator.TranslateIngressEvent(eventType, ingress)
+
+	c.lock.Lock()
+	defer c.lock.Unlock()
+
+	c.processIngressEvents(events)
+
+	// Change the local sync state within the lock to ensure that all
+	// event handlers have the same view of sync state.
+	c.ingressesListConsumed = c.IngressesListConsumed()
+	c.commit()
+}
+
+// HandleSecret handles a single Secret event and synchronizes the router backend.
+func (c *RouterController) HandleSecret() {
+	eventType, secret, err := c.NextSecret()
+	if err != nil {
+		utilruntime.HandleError(fmt.Errorf("unable to read secret: %v", err))
+		return
+
+	}
+
+	// The ingress translator synchronizes access to its cache with a
+	// lock, so calls to it are made outside of the controller lock to
+	// avoid unintended interaction.
+	events := c.IngressTranslator.TranslateSecretEvent(eventType, secret)
+
+	c.lock.Lock()
+	defer c.lock.Unlock()
+
+	c.processIngressEvents(events)
+
+	// Change the local sync state within the lock to ensure that all
+	// event handlers have the same view of sync state.
+	c.secretsListConsumed = c.SecretsListConsumed()
+	c.commit()
+}
+
 // commit notifies the plugin that it is safe to commit state.
 func (c *RouterController) commit() {
 	syncing := !(c.endpointsListConsumed && c.routesListConsumed &&
-		(c.Namespaces == nil || c.filteredByNamespace))
+		(c.Namespaces == nil || c.filteredByNamespace) &&
+		(!c.EnableIngress ||
+			(c.ingressesListConsumed && c.secretsListConsumed)))
 	c.logSyncState(syncing)
 	if syncing {
 		return
@@ -219,3 +297,24 @@ func (c *RouterController) logSyncState(syncing bool) {
 		}
 	}
 }
+
+// processRoute logs and propagates a route event to the plugin
+func (c *RouterController) processRoute(eventType watch.EventType, route *routeapi.Route) {
+	glog.V(4).Infof("Processing Route: %s/%s -> %s", route.Namespace, route.Name, route.Spec.To.Name)
+	glog.V(4).Infof("           Alias: %s", route.Spec.Host)
+	glog.V(4).Infof("           Event: %s", eventType)
+
+	if err := c.Plugin.HandleRoute(eventType, route); err != nil {
+		utilruntime.HandleError(err)
+	}
+}
+
+// processIngressEvents logs and propagates the route events resulting from an ingress event
+func (c *RouterController) processIngressEvents(events []ingressRouteEvents) {
+	for _, ingressEvent := range events {
+		glog.V(4).Infof("Processing Ingress %s", ingressEvent.ingressKey)
+		for _, routeEvent := range ingressEvent.routeEvents {
+			c.processRoute(routeEvent.eventType, routeEvent.route)
+		}
+	}
+}