77
88 "github.com/golang/glog"
99 kapi "k8s.io/kubernetes/pkg/api"
10+ "k8s.io/kubernetes/pkg/apis/extensions"
1011 utilruntime "k8s.io/kubernetes/pkg/util/runtime"
1112 "k8s.io/kubernetes/pkg/util/sets"
1213 utilwait "k8s.io/kubernetes/pkg/util/wait"
@@ -30,25 +31,38 @@ type RouterController struct {
3031 NextRoute func () (watch.EventType , * routeapi.Route , error )
3132 NextNode func () (watch.EventType , * kapi.Node , error )
3233 NextEndpoints func () (watch.EventType , * kapi.Endpoints , error )
34+ NextIngress func () (watch.EventType , * extensions.Ingress , error )
35+ NextSecret func () (watch.EventType , * kapi.Secret , error )
3336
3437 RoutesListConsumed func () bool
3538 EndpointsListConsumed func () bool
39+ IngressesListConsumed func () bool
40+ SecretsListConsumed func () bool
3641 routesListConsumed bool
3742 endpointsListConsumed bool
43+ ingressesListConsumed bool
44+ secretsListConsumed bool
3845 filteredByNamespace bool
3946 syncing bool
4047
4148 RoutesListSuccessfulAtLeastOnce func () bool
4249 EndpointsListSuccessfulAtLeastOnce func () bool
50+ IngressesListSuccessfulAtLeastOnce func () bool
51+ SecretsListSuccessfulAtLeastOnce func () bool
4352 RoutesListCount func () int
4453 EndpointsListCount func () int
54+ IngressesListCount func () int
55+ SecretsListCount func () int
4556
4657 WatchNodes bool
4758
4859 Namespaces NamespaceLister
4960 NamespaceSyncInterval time.Duration
5061 NamespaceWaitInterval time.Duration
5162 NamespaceRetries int
63+
64+ EnableIngress bool
65+ IngressTranslator * IngressTranslator
5266}
5367
5468// Run begins watching and syncing.
@@ -63,6 +77,10 @@ func (c *RouterController) Run() {
6377 if c .WatchNodes {
6478 go utilwait .Forever (c .HandleNode , 0 )
6579 }
80+ if c .EnableIngress {
81+ go utilwait .Forever (c .HandleIngress , 0 )
82+ go utilwait .Forever (c .HandleSecret , 0 )
83+ }
6684 go c .watchForFirstSync ()
6785}
6886
@@ -74,22 +92,31 @@ func (c *RouterController) handleFirstSync() bool {
7492
7593 synced := c .RoutesListSuccessfulAtLeastOnce () &&
7694 c .EndpointsListSuccessfulAtLeastOnce () &&
77- (c .Namespaces == nil || c .filteredByNamespace )
95+ (c .Namespaces == nil || c .filteredByNamespace ) &&
96+ (! c .EnableIngress ||
97+ (c .IngressesListSuccessfulAtLeastOnce () && c .SecretsListSuccessfulAtLeastOnce ()))
7898 if ! synced {
7999 return false
80100 }
81101
82- // If either of the event queues were empty after the initial
83- // List, the tracking listConsumed variable's default value of
84- // 'false' may prevent the router from committing the readiness
85- // status. Set the value to 'true' to ensure that state will be
86- // committed if necessary.
102+ // If any of the event queues were empty after the initial List,
103+ // the tracking listConsumed variable's default value of 'false'
104+ // may prevent the router from committing. Set the value to
105+ // 'true' to ensure that state can be committed if necessary.
87106 if c .RoutesListCount () == 0 {
88107 c .routesListConsumed = true
89108 }
90109 if c .EndpointsListCount () == 0 {
91110 c .endpointsListConsumed = true
92111 }
112+ if c .EnableIngress {
113+ if c .IngressesListCount () == 0 {
114+ c .ingressesListConsumed = true
115+ }
116+ if c .SecretsListCount () == 0 {
117+ c .secretsListConsumed = true
118+ }
119+ }
93120 c .commit ()
94121
95122 return true
@@ -109,6 +136,14 @@ func (c *RouterController) HandleNamespaces() {
109136 for i := 0 ; i < c .NamespaceRetries ; i ++ {
110137 namespaces , err := c .Namespaces .NamespaceNames ()
111138 if err == nil {
139+
140+ // The ingress translator synchronizes access to its cache with a
141+ // lock, so calls to it are made outside of the controller lock to
142+ // avoid unintended interaction.
143+ if c .EnableIngress {
144+ c .IngressTranslator .UpdateNamespaces (namespaces )
145+ }
146+
112147 c .lock .Lock ()
113148 defer c .lock .Unlock ()
114149
@@ -161,13 +196,7 @@ func (c *RouterController) HandleRoute() {
161196 c .lock .Lock ()
162197 defer c .lock .Unlock ()
163198
164- glog .V (4 ).Infof ("Processing Route: %s -> %s" , route .Name , route .Spec .To .Name )
165- glog .V (4 ).Infof (" Alias: %s" , route .Spec .Host )
166- glog .V (4 ).Infof (" Event: %s" , eventType )
167-
168- if err := c .Plugin .HandleRoute (eventType , route ); err != nil {
169- utilruntime .HandleError (err )
170- }
199+ c .processRoute (eventType , route )
171200
172201 // Change the local sync state within the lock to ensure that all
173202 // event handlers have the same view of sync state.
@@ -196,10 +225,61 @@ func (c *RouterController) HandleEndpoints() {
196225 c .commit ()
197226}
198227
228+ // HandleIngress handles a single Ingress event and synchronizes the router backend.
229+ func (c * RouterController ) HandleIngress () {
230+ eventType , ingress , err := c .NextIngress ()
231+ if err != nil {
232+ utilruntime .HandleError (fmt .Errorf ("unable to read ingress: %v" , err ))
233+ return
234+ }
235+
236+ // The ingress translator synchronizes access to its cache with a
237+ // lock, so calls to it are made outside of the controller lock to
238+ // avoid unintended interaction.
239+ events := c .IngressTranslator .TranslateIngressEvent (eventType , ingress )
240+
241+ c .lock .Lock ()
242+ defer c .lock .Unlock ()
243+
244+ c .processIngressEvents (events )
245+
246+ // Change the local sync state within the lock to ensure that all
247+ // event handlers have the same view of sync state.
248+ c .ingressesListConsumed = c .IngressesListConsumed ()
249+ c .commit ()
250+ }
251+
252+ // HandleSecret handles a single Secret event and synchronizes the router backend.
253+ func (c * RouterController ) HandleSecret () {
254+ eventType , secret , err := c .NextSecret ()
255+ if err != nil {
256+ utilruntime .HandleError (fmt .Errorf ("unable to read secret: %v" , err ))
257+ return
258+
259+ }
260+
261+ // The ingress translator synchronizes access to its cache with a
262+ // lock, so calls to it are made outside of the controller lock to
263+ // avoid unintended interaction.
264+ events := c .IngressTranslator .TranslateSecretEvent (eventType , secret )
265+
266+ c .lock .Lock ()
267+ defer c .lock .Unlock ()
268+
269+ c .processIngressEvents (events )
270+
271+ // Change the local sync state within the lock to ensure that all
272+ // event handlers have the same view of sync state.
273+ c .secretsListConsumed = c .SecretsListConsumed ()
274+ c .commit ()
275+ }
276+
199277// commit notifies the plugin that it is safe to commit state.
200278func (c * RouterController ) commit () {
201279 syncing := ! (c .endpointsListConsumed && c .routesListConsumed &&
202- (c .Namespaces == nil || c .filteredByNamespace ))
280+ (c .Namespaces == nil || c .filteredByNamespace ) &&
281+ (! c .EnableIngress ||
282+ (c .ingressesListConsumed && c .secretsListConsumed )))
203283 c .logSyncState (syncing )
204284 if syncing {
205285 return
@@ -219,3 +299,24 @@ func (c *RouterController) logSyncState(syncing bool) {
219299 }
220300 }
221301}
302+
303+ // processRoute logs and propagates a route event to the plugin
304+ func (c * RouterController ) processRoute (eventType watch.EventType , route * routeapi.Route ) {
305+ glog .V (4 ).Infof ("Processing Route: %s/%s -> %s" , route .Namespace , route .Name , route .Spec .To .Name )
306+ glog .V (4 ).Infof (" Alias: %s" , route .Spec .Host )
307+ glog .V (4 ).Infof (" Event: %s" , eventType )
308+
309+ if err := c .Plugin .HandleRoute (eventType , route ); err != nil {
310+ utilruntime .HandleError (err )
311+ }
312+ }
313+
314+ // processIngressEvents logs and propagates the route events resulting from an ingress or secret event
315+ func (c * RouterController ) processIngressEvents (events []ingressRouteEvents ) {
316+ for _ , ingressEvent := range events {
317+ glog .V (4 ).Infof ("Processing Ingress %s" , ingressEvent .ingressKey )
318+ for _ , routeEvent := range ingressEvent .routeEvents {
319+ c .processRoute (routeEvent .eventType , routeEvent .route )
320+ }
321+ }
322+ }
0 commit comments