// checkAndDecrement checks if the provided PodDisruptionBudget allows any disruption.
func (r *EvictionREST) checkAndDecrement(namespace string, podName string, pdb policy.PodDisruptionBudget) (ok bool, err error) {
	if pdb.Status.ObservedGeneration < pdb.Generation {
		return false, nil
	}
	if pdb.Status.PodDisruptionsAllowed < 0 {
		return false, errors.NewForbidden(policy.Resource("poddisruptionbudget"), pdb.Name, fmt.Errorf("pdb disruptions allowed is negative"))
	}
	if len(pdb.Status.DisruptedPods) > MaxDisruptedPodSize {
		return false, errors.NewForbidden(policy.Resource("poddisruptionbudget"), pdb.Name, fmt.Errorf("DisrputedPods map too big - too many evictions not confirmed by PDB controller"))
	}
	if pdb.Status.PodDisruptionsAllowed == 0 {
		return false, nil
	}
	pdb.Status.PodDisruptionsAllowed--
	if pdb.Status.DisruptedPods == nil {
		pdb.Status.DisruptedPods = make(map[string]metav1.Time)
	}
	// Eviction handler needs to inform the PDB controller that it is about to delete a pod
	// so it should not consider it as available in calculations when updating PodDisruptions allowed.
	// If the pod is not deleted within a reasonable time limit PDB controller will assume that it won't
	// be deleted at all and remove it from DisruptedPod map.
	pdb.Status.DisruptedPods[podName] = metav1.Time{Time: time.Now()}
	if _, err := r.podDisruptionBudgetClient.PodDisruptionBudgets(namespace).UpdateStatus(&pdb); err != nil {
		return false, err
	}

	return true, nil
}

// Get retrieves the PodDisruptionBudget from the indexer for a given namespace and name.
func (s podDisruptionBudgetNamespaceLister) Get(name string) (*v1beta1.PodDisruptionBudget, error) {
	obj, exists, err := s.indexer.GetByKey(s.namespace + "/" + name)
	if err != nil {
		return nil, err
	}
	if !exists {
		return nil, errors.NewNotFound(policy.Resource("poddisruptionbudget"), name)
	}
	return obj.(*v1beta1.PodDisruptionBudget), nil
}

// Get retrieves the Eviction from the indexer for a given namespace and name.
func (s evictionNamespaceLister) Get(name string) (*v1beta1.Eviction, error) {
	obj, exists, err := s.indexer.GetByKey(s.namespace + "/" + name)
	if err != nil {
		return nil, err
	}
	if !exists {
		return nil, errors.NewNotFound(policy.Resource("eviction"), name)
	}
	return obj.(*v1beta1.Eviction), nil
}

// NewREST returns a RESTStorage object that will work against pod disruption budgets.
func NewREST(opts generic.RESTOptions) (*REST, *StatusREST) {
	prefix := "/" + opts.ResourcePrefix

	newListFunc := func() runtime.Object { return &policyapi.PodDisruptionBudgetList{} }
	storageInterface, dFunc := opts.Decorator(
		opts.StorageConfig,
		cachesize.GetWatchCacheSizeByResource(cachesize.PodDisruptionBudget),
		&policyapi.PodDisruptionBudget{},
		prefix,
		poddisruptionbudget.Strategy,
		newListFunc,
		poddisruptionbudget.GetAttrs,
		storage.NoTriggerPublisher,
	)

	store := &genericregistry.Store{
		NewFunc: func() runtime.Object { return &policyapi.PodDisruptionBudget{} },

		// NewListFunc returns an object capable of storing results of an etcd list.
		NewListFunc: newListFunc,
		// Produces a podDisruptionBudget that etcd understands, to the root of the resource
		// by combining the namespace in the context with the given prefix
		KeyRootFunc: func(ctx api.Context) string {
			return genericregistry.NamespaceKeyRootFunc(ctx, prefix)
		},
		// Produces a podDisruptionBudget that etcd understands, to the resource by combining
		// the namespace in the context with the given prefix
		KeyFunc: func(ctx api.Context, name string) (string, error) {
			return genericregistry.NamespaceKeyFunc(ctx, prefix, name)
		},
		// Retrieve the name field of a pod disruption budget
		ObjectNameFunc: func(obj runtime.Object) (string, error) {
			return obj.(*policyapi.PodDisruptionBudget).Name, nil
		},
		// Used to match objects based on labels/fields for list and watch
		PredicateFunc:           poddisruptionbudget.MatchPodDisruptionBudget,
		QualifiedResource:       policyapi.Resource("poddisruptionbudgets"),
		EnableGarbageCollection: opts.EnableGarbageCollection,
		DeleteCollectionWorkers: opts.DeleteCollectionWorkers,

		// Used to validate controller creation
		CreateStrategy: poddisruptionbudget.Strategy,

		// Used to validate controller updates
		UpdateStrategy: poddisruptionbudget.Strategy,
		DeleteStrategy: poddisruptionbudget.Strategy,

		Storage:     storageInterface,
		DestroyFunc: dFunc,
	}
	statusStore := *store
	statusStore.UpdateStrategy = poddisruptionbudget.StatusStrategy
	return &REST{store}, &StatusREST{store: &statusStore}
}

// NewREST returns a RESTStorage object that will work against pod disruption budgets.
func NewREST(optsGetter generic.RESTOptionsGetter) (*REST, *StatusREST) {
	store := &genericregistry.Store{
		NewFunc:     func() runtime.Object { return &policyapi.PodDisruptionBudget{} },
		NewListFunc: func() runtime.Object { return &policyapi.PodDisruptionBudgetList{} },
		ObjectNameFunc: func(obj runtime.Object) (string, error) {
			return obj.(*policyapi.PodDisruptionBudget).Name, nil
		},
		PredicateFunc:     poddisruptionbudget.MatchPodDisruptionBudget,
		QualifiedResource: policyapi.Resource("poddisruptionbudgets"),

		CreateStrategy: poddisruptionbudget.Strategy,
		UpdateStrategy: poddisruptionbudget.Strategy,
		DeleteStrategy: poddisruptionbudget.Strategy,
	}
	options := &generic.StoreOptions{RESTOptions: optsGetter, AttrFunc: poddisruptionbudget.GetAttrs}
	if err := store.CompleteWithOptions(options); err != nil {
		panic(err) // TODO: Propagate error up
	}

	statusStore := *store
	statusStore.UpdateStrategy = poddisruptionbudget.StatusStrategy
	return &REST{store}, &StatusREST{store: &statusStore}
}

// getPolicyResources returns the resources for policy api
func (m *Master) getPolicyResources(c *Config) map[string]rest.Storage {
	// TODO update when we support more than one version of this group
	version := policyapiv1alpha1.SchemeGroupVersion

	storage := map[string]rest.Storage{}
	if c.APIResourceConfigSource.ResourceEnabled(version.WithResource("poddisruptionbudgets")) {
		poddisruptionbudgetStorage, poddisruptionbudgetStatusStorage := poddisruptionbudgetetcd.NewREST(m.GetRESTOptionsOrDie(c, policy.Resource("poddisruptionbudgets")))
		storage["poddisruptionbudgets"] = poddisruptionbudgetStorage
		storage["poddisruptionbudgets/status"] = poddisruptionbudgetStatusStorage
	}
	return storage
}

func buildPolicyResources(apiResourceConfigSource genericapiserver.APIResourceConfigSource, restOptionsGetter RESTOptionsGetter) (genericapiserver.APIGroupInfo, bool) {
	apiGroupInfo := NewDefaultAPIGroupInfo(policy.GroupName)

	storageForVersion := func(version unversioned.GroupVersion) map[string]rest.Storage {
		storage := map[string]rest.Storage{}
		if apiResourceConfigSource.ResourceEnabled(version.WithResource("poddisruptionbudgets")) {
			poddisruptionbudgetStorage, poddisruptionbudgetStatusStorage := poddisruptionbudgetetcd.NewREST(restOptionsGetter(policy.Resource("poddisruptionbudgets")))
			storage["poddisruptionbudgets"] = poddisruptionbudgetStorage
			storage["poddisruptionbudgets/status"] = poddisruptionbudgetStatusStorage
		}
		return storage
	}

	if apiResourceConfigSource.AnyResourcesForVersionEnabled(policyapiv1alpha1.SchemeGroupVersion) {
		apiGroupInfo.VersionedResourcesStorageMap[policyapiv1alpha1.SchemeGroupVersion.Version] = storageForVersion(policyapiv1alpha1.SchemeGroupVersion)
		apiGroupInfo.GroupMeta.GroupVersion = policyapiv1alpha1.SchemeGroupVersion
	}

	return apiGroupInfo, true
}

func (p RESTStorageProvider) v1alpha1Storage(apiResourceConfigSource genericapiserver.APIResourceConfigSource, restOptionsGetter genericapiserver.RESTOptionsGetter) map[string]rest.Storage {
	version := policyapiv1alpha1.SchemeGroupVersion

	storage := map[string]rest.Storage{}
	if apiResourceConfigSource.ResourceEnabled(version.WithResource("poddisruptionbudgets")) {
		poddisruptionbudgetStorage, poddisruptionbudgetStatusStorage := poddisruptionbudgetetcd.NewREST(restOptionsGetter(policy.Resource("poddisruptionbudgets")))
		storage["poddisruptionbudgets"] = poddisruptionbudgetStorage
		storage["poddisruptionbudgets/status"] = poddisruptionbudgetStatusStorage
	}
	return storage
}