status.go

package controller

import (
	"fmt"
	"time"

	"github.com/golang/glog"

	"k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
	"k8s.io/apimachinery/pkg/util/sets"
	"k8s.io/apimachinery/pkg/watch"
	kapi "k8s.io/kubernetes/pkg/apis/core"

	routeapi "github.com/openshift/origin/pkg/route/apis/route"
	client "github.com/openshift/origin/pkg/route/generated/internalclientset/typed/route/internalversion"
	"github.com/openshift/origin/pkg/router"
	"github.com/openshift/origin/pkg/util/writerlease"
)

// RejectionRecorder is an object capable of recording why a route was rejected
type RejectionRecorder interface {
	RecordRouteRejection(route *routeapi.Route, reason, message string)
}

// StatusAdmitter ensures routes added to the plugin have status set.
type StatusAdmitter struct {
	plugin                  router.Plugin
	client                  client.RoutesGetter
	routerName              string
	routerCanonicalHostname string
	lease                   writerlease.Lease
}

// NewStatusAdmitter creates a plugin wrapper that ensures every accepted
// route has a status field set that matches this router. The admitter manages
// an LRU of recently seen conflicting updates to handle when two router processes
// with differing configurations are writing updates at the same time.
func NewStatusAdmitter(plugin router.Plugin, client client.RoutesGetter, name, hostName string, lease writerlease.Lease) *StatusAdmitter {
	return &StatusAdmitter{
		plugin:                  plugin,
		client:                  client,
		routerName:              name,
		routerCanonicalHostname: hostName,
		lease: lease,
	}
}

// Return a time truncated to the second to ensure that in-memory and
// serialized timestamps can be safely compared.
func getRfc3339Timestamp() metav1.Time {
	return metav1.Now().Rfc3339Copy()
}

// nowFn allows the package to be tested
var nowFn = getRfc3339Timestamp

// findOrCreateIngress loops through the router status ingress array looking for an entry
// that matches name. If there is no entry in the array, it creates one and appends it
// to the array. If there are multiple entries with that name, the first one is
// returned and later ones are removed. Changed is returned as true if any part of the
// array is altered.
func findOrCreateIngress(route *routeapi.Route, name, hostName string) (_ *routeapi.RouteIngress, changed bool) {
	position := -1
	updated := make([]routeapi.RouteIngress, 0, len(route.Status.Ingress))
	for i := range route.Status.Ingress {
		existing := &route.Status.Ingress[i]
		if existing.RouterName != name {
			updated = append(updated, *existing)
			continue
		}
		if position != -1 {
			changed = true
			continue
		}
		updated = append(updated, *existing)
		position = len(updated) - 1
	}
	switch {
	case position == -1:
		position = len(route.Status.Ingress)
		route.Status.Ingress = append(route.Status.Ingress, routeapi.RouteIngress{
			RouterName:              name,
			Host:                    route.Spec.Host,
			WildcardPolicy:          route.Spec.WildcardPolicy,
			RouterCanonicalHostname: hostName,
		})
		changed = true
	case changed:
		route.Status.Ingress = updated
	}
	ingress := &route.Status.Ingress[position]
	if ingress.Host != route.Spec.Host {
		ingress.Host = route.Spec.Host
		changed = true
	}
	if ingress.WildcardPolicy != route.Spec.WildcardPolicy {
		ingress.WildcardPolicy = route.Spec.WildcardPolicy
		changed = true
	}
	if ingress.RouterCanonicalHostname != hostName {
		ingress.RouterCanonicalHostname = hostName
		changed = true
	}
	return ingress, changed
}

// setIngressCondition records the condition on the ingress, returning true if the ingress was changed and
// false if no modification was made (or the only modification would have been to update a time).
func setIngressCondition(ingress *routeapi.RouteIngress, condition routeapi.RouteIngressCondition) bool {
	for i, existing := range ingress.Conditions {
		// ensures that the comparison is based on the actual value, not the time
		existing.LastTransitionTime = condition.LastTransitionTime
		if existing == condition {
			// This will always be the case if we're receiving an update on the host
			// value (or the like), since findOrCreateIngress sets that for us.  We
			// still need to set the last-touched time so that others can tell we've
			// modified this Ingress value
			now := nowFn()
			ingress.Conditions[i].LastTransitionTime = &now
			return false
		}
	}
	now := nowFn()
	condition.LastTransitionTime = &now
	ingress.Conditions = []routeapi.RouteIngressCondition{condition}
	return true
}

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
}

// recordIngressConditionFailure updates the matching ingress on the route (or adds a new one) with the specified
// condition, returning true if the object was modified.
func recordIngressConditionFailure(route *routeapi.Route, name, hostName string, condition routeapi.RouteIngressCondition) bool {
	for i := range route.Status.Ingress {
		existing := &route.Status.Ingress[i]
		if existing.RouterName != name {
			continue
		}

		// we've changed things if we either replaced the host value...
		changed := false
		if existing.Host != route.Spec.Host {
			existing.Host = route.Spec.Host
			changed = true
		}
		// ...or replaced the entire condition list
		// (NB: order matters in this OR -- short circuiting)
		changed = setIngressCondition(existing, condition) || changed
		return changed
	}
	route.Status.Ingress = append(route.Status.Ingress, routeapi.RouteIngress{RouterName: name, RouterCanonicalHostname: hostName, Host: route.Spec.Host})
	ingress := &route.Status.Ingress[len(route.Status.Ingress)-1]
	setIngressCondition(ingress, condition)
	return true
}

// HandleRoute attempts to admit the provided route on watch add / modifications.
func (a *StatusAdmitter) HandleRoute(eventType watch.EventType, route *routeapi.Route) error {
	if IsGeneratedRouteName(route.Name) {
		// Can't record status for ingress resources
	} else {
		switch eventType {
		case watch.Added, watch.Modified:
			if ok := setRouteAdmitted(a.lease, a.client, route, a.routerName, a.routerCanonicalHostname); !ok {
				glog.V(4).Infof("skipping route: %s", route.Name)
				return nil
			}
		}
	}
	return a.plugin.HandleRoute(eventType, route)
}

func (a *StatusAdmitter) HandleNode(eventType watch.EventType, node *kapi.Node) error {
	return a.plugin.HandleNode(eventType, node)
}

func (a *StatusAdmitter) HandleEndpoints(eventType watch.EventType, route *kapi.Endpoints) error {
	return a.plugin.HandleEndpoints(eventType, route)
}

func (a *StatusAdmitter) HandleNamespaces(namespaces sets.String) error {
	return a.plugin.HandleNamespaces(namespaces)
}

func (a *StatusAdmitter) Commit() error {
	return a.plugin.Commit()
}

// RecordRouteRejection attempts to update the route status with a reason for a route being rejected.
func (a *StatusAdmitter) RecordRouteRejection(route *routeapi.Route, reason, message string) {
	if IsGeneratedRouteName(route.Name) {
		// Can't record status for ingress resources
		return
	}

	changed := recordIngressConditionFailure(route, a.routerName, a.routerCanonicalHostname, routeapi.RouteIngressCondition{
		Type:    routeapi.RouteAdmitted,
		Status:  kapi.ConditionFalse,
		Reason:  reason,
		Message: message,
	})
	if !changed {
		glog.V(4).Infof("reject: no changes to route needed: %s/%s", route.Namespace, route.Name)
		return
	}

	a.lease.Try(leaseWork(a.client, route))
}

// setRouteAdmitted updates the route status with a condition indicating the current router
// has admitted the route.
func setRouteAdmitted(lease writerlease.Lease, oc client.RoutesGetter, route *routeapi.Route, name, hostName string) bool {
	ingress, updated := findOrCreateIngress(route, name, hostName)

	// check to see if we already have admitted the route
	if !updated {
		for i := range ingress.Conditions {
			cond := &ingress.Conditions[i]
			if cond.Type == routeapi.RouteAdmitted && cond.Status == kapi.ConditionTrue {
				lease.Extend(fmt.Sprintf("%s/%s", route.Namespace, route.Name))
				glog.V(4).Infof("admit: route status already admitted")
				return true
			}
		}
	}
	setIngressCondition(ingress, routeapi.RouteIngressCondition{
		Type:   routeapi.RouteAdmitted,
		Status: kapi.ConditionTrue,
	})

	lease.Try(leaseWork(oc, route))
	leader, ok := lease.WaitUntil(10 * time.Second)
	if !ok {
		glog.V(4).Infof("admit: did not update route status within interval, rejecting route until next resync")
	}
	return leader
}

func leaseWork(oc client.RoutesGetter, route *routeapi.Route) (string, writerlease.WorkFunc) {
	key := fmt.Sprintf("%s/%s", route.Namespace, route.Name)
	return key, writerlease.LimitRetries(3, func() (bool, bool) {
		glog.V(4).Infof("admit: updating status on route: %s: %s", route.Name, route.Spec.Host)
		switch _, err := oc.Routes(route.Namespace).UpdateStatus(route); {
		case err == nil:
			return true, false
		case errors.IsForbidden(err):
			// if the router can't write status updates, allow the route to go through
			utilruntime.HandleError(fmt.Errorf("Unable to write router status - please ensure you reconcile your system policy or grant this router access to update route status: %v", err))
			return true, false
		case errors.IsConflict(err):
			// just follow the normal process, and retry when we receive the update notification due to
			// the other entity updating the route.
			return false, false
		default:
			utilruntime.HandleError(fmt.Errorf("Unable to write router status: %v", err))
			return false, true
		}
	})
}