-
Notifications
You must be signed in to change notification settings - Fork 4.7k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #19706 from smarterclayton/log_route_uid
Make router conflict detection work even during initial informer sync
- Loading branch information
Showing
8 changed files
with
423 additions
and
270 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,281 @@ | ||
| package controller | ||
|
|
||
| import ( | ||
| "sync" | ||
| "time" | ||
|
|
||
| "github.com/golang/glog" | ||
|
|
||
| metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" | ||
| "k8s.io/client-go/tools/cache" | ||
|
|
||
| routeapi "github.com/openshift/origin/pkg/route/apis/route" | ||
| ) | ||
|
|
||
| // ContentionTracker records modifications to a particular entry to prevent endless | ||
| // loops when multiple routers are configured with conflicting info. A given router | ||
| // process tracks whether the ingress status is change from a correct value to any | ||
| // other value (by invoking IsContended when the state has diverged). | ||
| type ContentionTracker interface { | ||
| // IsContended should be invoked when the state of the object in storage differs | ||
| // from the desired state. It will return true if the provided id was recently | ||
| // reset from the correct state to an incorrect state. The current ingress is the | ||
| // expected state of the object at this time and may be used by the tracker to | ||
| // determine if the most recent update was a contention. This method does not | ||
| // update the state of the tracker. | ||
| IsChangeContended(id string, now time.Time, current *routeapi.RouteIngress) bool | ||
| // Clear informs the tracker that the provided ingress state was confirmed to | ||
| // match the current state of this process. If a subsequent call to IsChangeContended | ||
| // is made within the expiration window, the object will be considered as contended. | ||
| Clear(id string, current *routeapi.RouteIngress) | ||
| } | ||
|
|
||
| type elementState int | ||
|
|
||
| const ( | ||
| stateCandidate elementState = iota | ||
| stateContended | ||
| ) | ||
|
|
||
| type trackerElement struct { | ||
| at time.Time | ||
| state elementState | ||
| last *routeapi.RouteIngress | ||
| } | ||
|
|
||
| // SimpleContentionTracker tracks whether a given identifier is changed from a correct | ||
| // state (set by Clear) to an incorrect state (inferred by calling IsContended). | ||
| type SimpleContentionTracker struct { | ||
| informer cache.SharedInformer | ||
| routerName string | ||
|
|
||
| expires time.Duration | ||
| // maxContentions is the number of contentions detected before the entire | ||
| maxContentions int | ||
| message string | ||
|
|
||
| lock sync.Mutex | ||
| contentions int | ||
| ids map[string]trackerElement | ||
| } | ||
|
|
||
| // NewSimpleContentionTracker creates a ContentionTracker that will prevent writing | ||
| // to the same route more often than once per interval. A background process will | ||
| // periodically flush old entries (at twice interval) in order to prevent the list | ||
| // growing unbounded if routes are created and deleted frequently. The informer | ||
| // detects changes to ingress records for routerName and will advance the tracker | ||
| // state from candidate to contended if the host, wildcardPolicy, or canonical host | ||
| // name fields are repeatedly updated. | ||
| func NewSimpleContentionTracker(informer cache.SharedInformer, routerName string, interval time.Duration) *SimpleContentionTracker { | ||
| return &SimpleContentionTracker{ | ||
| informer: informer, | ||
| routerName: routerName, | ||
| expires: interval, | ||
| maxContentions: 5, | ||
|
|
||
| ids: make(map[string]trackerElement), | ||
| } | ||
| } | ||
|
|
||
| // SetConflictMessage will print message whenever contention with another writer | ||
| // is detected. | ||
| func (t *SimpleContentionTracker) SetConflictMessage(message string) { | ||
| t.lock.Lock() | ||
| defer t.lock.Unlock() | ||
| t.message = message | ||
| } | ||
|
|
||
| // Run starts the background cleanup process for expired items. | ||
| func (t *SimpleContentionTracker) Run(stopCh <-chan struct{}) { | ||
| // Watch the informer for changes to the route ingress we care about (identified | ||
| // by router name) and if we see it change remember it. This loop can process | ||
| // changes to routes faster than the router, which means it has more up-to-date | ||
| // contention info and can detect contention while the main controller is still | ||
| // syncing. | ||
| t.informer.AddEventHandler(cache.ResourceEventHandlerFuncs{ | ||
| UpdateFunc: func(oldObj, obj interface{}) { | ||
| oldRoute, ok := oldObj.(*routeapi.Route) | ||
| if !ok { | ||
| return | ||
| } | ||
| route, ok := obj.(*routeapi.Route) | ||
| if !ok { | ||
| return | ||
| } | ||
| if ingress := ingressChanged(oldRoute, route, t.routerName); ingress != nil { | ||
| t.Changed(string(route.UID), ingress) | ||
| } | ||
| }, | ||
| }) | ||
|
|
||
| // periodically clean up expired changes | ||
| ticker := time.NewTicker(t.expires * 2) | ||
| defer ticker.Stop() | ||
| for { | ||
| select { | ||
| case <-stopCh: | ||
| return | ||
| case <-ticker.C: | ||
| t.flush() | ||
| } | ||
| } | ||
| } | ||
|
|
||
| func (t *SimpleContentionTracker) flush() { | ||
| t.lock.Lock() | ||
| defer t.lock.Unlock() | ||
|
|
||
| // reset conflicts every expiration interval, but remove tracking info less often | ||
| contentionExpiration := nowFn().Add(-t.expires) | ||
| trackerExpiration := contentionExpiration.Add(-2 * t.expires) | ||
|
|
||
| removed := 0 | ||
| contentions := 0 | ||
| for id, last := range t.ids { | ||
| switch last.state { | ||
| case stateContended: | ||
| if last.at.Before(contentionExpiration) { | ||
| delete(t.ids, id) | ||
| removed++ | ||
| continue | ||
| } | ||
| contentions++ | ||
| default: | ||
| if last.at.Before(trackerExpiration) { | ||
| delete(t.ids, id) | ||
| removed++ | ||
| continue | ||
| } | ||
| } | ||
| } | ||
| if t.contentions > 0 && len(t.message) > 0 { | ||
| glog.Warning(t.message) | ||
| } | ||
| glog.V(5).Infof("Flushed contention tracker (%s): %d out of %d removed, %d total contentions", t.expires*2, removed, removed+len(t.ids), t.contentions) | ||
| t.contentions = contentions | ||
| } | ||
|
|
||
| // Changed records that a change to an ingress value was detected. This is called from | ||
| // a separate goroutine and may have seen newer events than the current route controller | ||
| // plugins, so we don't do direct time comparisons. Instead we count edge transitions on | ||
| // a given id. | ||
| func (t *SimpleContentionTracker) Changed(id string, current *routeapi.RouteIngress) { | ||
| t.lock.Lock() | ||
| defer t.lock.Unlock() | ||
|
|
||
| // we have detected a sufficient number of conflicts to skip all updates for this interval | ||
| if t.contentions > t.maxContentions { | ||
| glog.V(4).Infof("Reached max contentions, stop tracking changes") | ||
| return | ||
| } | ||
|
|
||
| // if we have never recorded this object | ||
| last, ok := t.ids[id] | ||
| if !ok { | ||
| t.ids[id] = trackerElement{ | ||
| at: nowFn().Time, | ||
| state: stateCandidate, | ||
| last: current, | ||
| } | ||
| glog.V(4).Infof("Object %s is a candidate for contention", id) | ||
| return | ||
| } | ||
|
|
||
| // the previous state matches the current state, nothing to do | ||
| if ingressEqual(last.last, current) { | ||
| glog.V(4).Infof("Object %s is unchanged", id) | ||
| return | ||
| } | ||
|
|
||
| if last.state == stateContended { | ||
| t.contentions++ | ||
| glog.V(4).Infof("Object %s is contended and has been modified by another writer", id) | ||
| return | ||
| } | ||
|
|
||
| // if it appears that the state is being changed by another party, mark it as contended | ||
| if last.state == stateCandidate { | ||
| t.ids[id] = trackerElement{ | ||
| at: nowFn().Time, | ||
| state: stateContended, | ||
| last: current, | ||
| } | ||
| t.contentions++ | ||
| glog.V(4).Infof("Object %s has been modified by another writer", id) | ||
| return | ||
| } | ||
| } | ||
|
|
||
| func (t *SimpleContentionTracker) IsChangeContended(id string, now time.Time, current *routeapi.RouteIngress) bool { | ||
| t.lock.Lock() | ||
| defer t.lock.Unlock() | ||
|
|
||
| // we have detected a sufficient number of conflicts to skip all updates for this interval | ||
| if t.contentions > t.maxContentions { | ||
| glog.V(4).Infof("Reached max contentions, rejecting all update attempts until the next interval") | ||
| return true | ||
| } | ||
|
|
||
| // if we have expired or never recorded this object | ||
| last, ok := t.ids[id] | ||
| if !ok || last.at.Add(t.expires).Before(now) { | ||
| return false | ||
| } | ||
|
|
||
| // if the object is contended, exit early | ||
| if last.state == stateContended { | ||
| glog.V(4).Infof("Object %s is being contended by another writer", id) | ||
| return true | ||
| } | ||
|
|
||
| return false | ||
| } | ||
|
|
||
| func (t *SimpleContentionTracker) Clear(id string, current *routeapi.RouteIngress) { | ||
| t.lock.Lock() | ||
| defer t.lock.Unlock() | ||
|
|
||
| last, ok := t.ids[id] | ||
| if !ok { | ||
| return | ||
| } | ||
| last.last = current | ||
| last.state = stateCandidate | ||
| t.ids[id] = last | ||
| } | ||
|
|
||
| func ingressEqual(a, b *routeapi.RouteIngress) bool { | ||
| return a.Host == b.Host && a.RouterCanonicalHostname == b.RouterCanonicalHostname && a.WildcardPolicy == b.WildcardPolicy && a.RouterName == b.RouterName | ||
| } | ||
|
|
||
| func ingressConditionTouched(ingress *routeapi.RouteIngress) *metav1.Time { | ||
| var lastTouch *metav1.Time | ||
| for _, condition := range ingress.Conditions { | ||
| if t := condition.LastTransitionTime; t != nil { | ||
| switch { | ||
| case lastTouch == nil, t.After(lastTouch.Time): | ||
| lastTouch = t | ||
| } | ||
| } | ||
| } | ||
| return lastTouch | ||
| } | ||
|
|
||
| func ingressChanged(oldRoute, route *routeapi.Route, routerName string) *routeapi.RouteIngress { | ||
| var ingress *routeapi.RouteIngress | ||
| for i := range route.Status.Ingress { | ||
| if route.Status.Ingress[i].RouterName == routerName { | ||
| ingress = &route.Status.Ingress[i] | ||
| for _, old := range oldRoute.Status.Ingress { | ||
| if old.RouterName == routerName { | ||
| if !ingressEqual(ingress, &old) { | ||
| return ingress | ||
| } | ||
| return nil | ||
| } | ||
| } | ||
| return nil | ||
| } | ||
| } | ||
| return nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.