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

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

	storage := map[string]rest.Storage{}
	if apiResourceConfigSource.ResourceEnabled(version.WithResource("petsets")) {
		petsetStorage, petsetStatusStorage := petsetetcd.NewREST(restOptionsGetter(apps.Resource("petsets")))
		storage["petsets"] = petsetStorage
		storage["petsets/status"] = petsetStatusStorage
	}
	return storage
}

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

	newListFunc := func() runtime.Object { return &appsapi.StatefulSetList{} }
	storageInterface, dFunc := opts.Decorator(
		opts.StorageConfig,
		cachesize.GetWatchCacheSizeByResource(cachesize.StatefulSet),
		&appsapi.StatefulSet{},
		prefix,
		petset.Strategy,
		newListFunc,
		petset.GetAttrs,
		storage.NoTriggerPublisher,
	)

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

		// NewListFunc returns an object capable of storing results of an etcd list.
		NewListFunc: newListFunc,
		// Produces a statefulSet 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 statefulSet 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 replication controller
		ObjectNameFunc: func(obj runtime.Object) (string, error) {
			return obj.(*appsapi.StatefulSet).Name, nil
		},
		// Used to match objects based on labels/fields for list and watch
		PredicateFunc:           petset.MatchStatefulSet,
		QualifiedResource:       appsapi.Resource("statefulsets"),
		EnableGarbageCollection: opts.EnableGarbageCollection,
		DeleteCollectionWorkers: opts.DeleteCollectionWorkers,

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

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

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

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

	storage := map[string]rest.Storage{}
	if c.APIResourceConfigSource.ResourceEnabled(version.WithResource("petsets")) {
		petsetStorage, petsetStatusStorage := petsetetcd.NewREST(m.GetRESTOptionsOrDie(c, apps.Resource("petsets")))
		storage["petsets"] = petsetStorage
		storage["petsets/status"] = petsetStatusStorage
	}
	return storage
}

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

	storageForVersion := func(version unversioned.GroupVersion) map[string]rest.Storage {
		storage := map[string]rest.Storage{}
		if apiResourceConfigSource.ResourceEnabled(version.WithResource("petsets")) {
			petsetStorage, petsetStatusStorage := petsetetcd.NewREST(restOptionsGetter(apps.Resource("petsets")))
			storage["petsets"] = petsetStorage
			storage["petsets/status"] = petsetStatusStorage
		}
		return storage
	}

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

	return apiGroupInfo, true
}

// NewREST returns a RESTStorage object that will work against replication controllers.
func NewREST(optsGetter generic.RESTOptionsGetter) (*REST, *StatusREST) {
	store := &genericregistry.Store{
		NewFunc:     func() runtime.Object { return &appsapi.StatefulSet{} },
		NewListFunc: func() runtime.Object { return &appsapi.StatefulSetList{} },
		ObjectNameFunc: func(obj runtime.Object) (string, error) {
			return obj.(*appsapi.StatefulSet).Name, nil
		},
		PredicateFunc:     petset.MatchStatefulSet,
		QualifiedResource: appsapi.Resource("statefulsets"),

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

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

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

	storage := map[string]rest.Storage{}
	if apiResourceConfigSource.ResourceEnabled(version.WithResource("statefulsets")) {
		statefulsetStorage, statefulsetStatusStorage := statefulsetetcd.NewREST(restOptionsGetter(apps.Resource("statefulsets")))
		storage["statefulsets"] = statefulsetStorage
		storage["statefulsets/status"] = statefulsetStatusStorage
	}
	return storage
}

// resourcesToCheck is a map of resources and corresponding kinds of things that
// we want handled in this plugin
// TODO: Include a function that will extract the PodSpec from the resource for
// each type added here.
var resourcesToCheck = map[unversioned.GroupResource]unversioned.GroupKind{
	kapi.Resource("pods"):                                       kapi.Kind("Pod"),
	kapi.Resource("podtemplates"):                               kapi.Kind("PodTemplate"),
	kapi.Resource("replicationcontrollers"):                     kapi.Kind("ReplicationController"),
	batch.Resource("jobs"):                                      batch.Kind("Job"),
	batch.Resource("jobtemplates"):                              batch.Kind("JobTemplate"),
	batch.Resource("scheduledjobs"):                             batch.Kind("ScheduledJob"),
	extensions.Resource("deployments"):                          extensions.Kind("Deployment"),
	extensions.Resource("replicasets"):                          extensions.Kind("ReplicaSet"),
	extensions.Resource("jobs"):                                 extensions.Kind("Job"),
	extensions.Resource("jobtemplates"):                         extensions.Kind("JobTemplate"),
	apps.Resource("petsets"):                                    apps.Kind("PetSet"),
	deployapi.Resource("deploymentconfigs"):                     deployapi.Kind("DeploymentConfig"),
	securityapi.Resource("podsecuritypolicysubjectreviews"):     securityapi.Kind("PodSecurityPolicySubjectReview"),
	securityapi.Resource("podsecuritypolicyselfsubjectreviews"): securityapi.Kind("PodSecurityPolicySelfSubjectReview"),
	securityapi.Resource("podsecuritypolicyreviews"):            securityapi.Kind("PodSecurityPolicyReview"),
}

// resourcesToIgnore is a list of resource kinds that contain a PodSpec that
// we choose not to handle in this plugin
var resourcesToIgnore = []unversioned.GroupKind{
	extensions.Kind("DaemonSet"),
}

func shouldCheckResource(resource unversioned.GroupResource, kind unversioned.GroupKind) (bool, error) {
	expectedKind, shouldCheck := resourcesToCheck[resource]
	if !shouldCheck {