Implement DisruptionController.

Part of kubernetes#12611
lavalamp · Aug 16, 2016 · d60ba3c · d60ba3c
1 parent 1b0bc94
commit d60ba3c
Show file tree

Hide file tree

Showing 6 changed files with 1,259 additions and 0 deletions.
diff --git a/cmd/kube-controller-manager/app/controllermanager.go b/cmd/kube-controller-manager/app/controllermanager.go
@@ -49,6 +49,7 @@ import (
 	certcontroller "k8s.io/kubernetes/pkg/controller/certificates"
 	"k8s.io/kubernetes/pkg/controller/daemon"
 	"k8s.io/kubernetes/pkg/controller/deployment"
+	"k8s.io/kubernetes/pkg/controller/disruption"
 	endpointcontroller "k8s.io/kubernetes/pkg/controller/endpoint"
 	"k8s.io/kubernetes/pkg/controller/framework/informers"
 	"k8s.io/kubernetes/pkg/controller/garbagecollector"
@@ -367,6 +368,18 @@ func StartControllers(s *options.CMServer, kubeClient *client.Client, kubeconfig
 		}
 	}
 
+	groupVersion = "policy/v1alpha1"
+	resources, found = resourceMap[groupVersion]
+	glog.Infof("Attempting to start disruption controller, full resource map %+v", resourceMap)
+	if containsVersion(versions, groupVersion) && found {
+		glog.Infof("Starting %s apis", groupVersion)
+		if containsResource(resources, "poddisruptionbudgets") {
+			glog.Infof("Starting disruption controller")
+			go disruption.NewDisruptionController(sharedInformers.Pods().Informer(), kubeClient).Run(wait.NeverStop)
+			time.Sleep(wait.Jitter(s.ControllerStartInterval.Duration, ControllerStartJitter))
+		}
+	}
+
 	groupVersion = "apps/v1alpha1"
 	resources, found = resourceMap[groupVersion]
 	glog.Infof("Attempting to start petset, full resource map %+v", resourceMap)

diff --git a/pkg/client/cache/listers.go b/pkg/client/cache/listers.go
@@ -26,6 +26,7 @@ import (
 	"k8s.io/kubernetes/pkg/apis/batch"
 	"k8s.io/kubernetes/pkg/apis/certificates"
 	"k8s.io/kubernetes/pkg/apis/extensions"
+	"k8s.io/kubernetes/pkg/apis/policy"
 	"k8s.io/kubernetes/pkg/labels"
 )
 
@@ -710,3 +711,44 @@ func (i *IndexerToNamespaceLister) List(selector labels.Selector) (namespaces []
 
 	return namespaces, nil
 }
+
+type StoreToPodDisruptionBudgetLister struct {
+	Store
+}
+
+// GetPodPodDisruptionBudgets returns a list of PodDisruptionBudgets matching a pod.  Returns an error only if no matching PodDisruptionBudgets are found.
+func (s *StoreToPodDisruptionBudgetLister) GetPodPodDisruptionBudgets(pod *api.Pod) (pdbList []policy.PodDisruptionBudget, err error) {
+	var selector labels.Selector
+
+	if len(pod.Labels) == 0 {
+		err = fmt.Errorf("no PodDisruptionBudgets found for pod %v because it has no labels", pod.Name)
+		return
+	}
+
+	for _, m := range s.Store.List() {
+		pdb, ok := m.(*policy.PodDisruptionBudget)
+		if !ok {
+			glog.Errorf("Unexpected: %v is not a PodDisruptionBudget", m)
+			continue
+		}
+		if pdb.Namespace != pod.Namespace {
+			continue
+		}
+		selector, err = unversioned.LabelSelectorAsSelector(pdb.Spec.Selector)
+		if err != nil {
+			glog.Warningf("invalid selector: %v", err)
+			// TODO(mml): add an event to the PDB
+			continue
+		}
+
+		// If a PDB with a nil or empty selector creeps in, it should match nothing, not everything.
+		if selector.Empty() || !selector.Matches(labels.Set(pod.Labels)) {
+			continue
+		}
+		pdbList = append(pdbList, *pdb)
+	}
+	if len(pdbList) == 0 {
+		err = fmt.Errorf("could not find PodDisruptionBudget for pod %s in namespace %s with labels: %v", pod.Name, pod.Namespace, pod.Labels)
+	}
+	return
+}
diff --git a/pkg/client/unversioned/client.go b/pkg/client/unversioned/client.go
@@ -183,6 +183,10 @@ func (c *Client) Rbac() RbacInterface {
 	return c.RbacClient
 }
 
+func (c *Client) Policy() PolicyInterface {
+	return c.PolicyClient
+}
+
 func (c *Client) Discovery() discovery.DiscoveryInterface {
 	return c.DiscoveryClient
 }