WIP router: add basic support for config via ingress

Author: marun

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/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/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,37 @@ 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
+
+	IngressTranslator *IngressTranslator
 }
 
 // Run begins watching and syncing.
@@ -63,6 +76,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 +89,29 @@ func (c *RouterController) handleFirstSync() bool {
 
 	synced := c.RoutesListSuccessfulAtLeastOnce() &&
 		c.EndpointsListSuccessfulAtLeastOnce() &&
-		(c.Namespaces == nil || c.filteredByNamespace)
+		(c.Namespaces == nil || c.filteredByNamespace) &&
+		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.IngressesListCount() == 0 {
+		c.ingressesListConsumed = true
+	}
+	if c.SecretsListCount() == 0 {
+		c.secretsListConsumed = true
+	}
 	c.commit()
 
 	return true
@@ -161,13 +183,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 +212,60 @@ 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.ingressesListConsumed && c.secretsListConsumed)
 	c.logSyncState(syncing)
 	if syncing {
 		return
@@ -219,3 +285,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)
+		}
+	}
+}