Merge pull request #31190 from hongchaodeng/r2
Automatic merge from submit-queue Pass SelectionPredicate instead of Filter to storage layer Depends on #31189 (first commit). ref: #29888 What? This PR removes the filtering logic and passes SelectionPredicate to storage layer. Why? Filter doesn't provide enough information of and isn't the right abstraction for indexing. We need to pass in SelectionPredicate instead.
This commit is contained in:
commit
987aef1f64
@ -61,8 +61,8 @@ func NewREST(config *storagebackend.Config, storageDecorator generic.StorageDeco
|
|||||||
return obj.(*testgroup.TestType).Name, nil
|
return obj.(*testgroup.TestType).Name, nil
|
||||||
},
|
},
|
||||||
// Used to match objects based on labels/fields for list.
|
// Used to match objects based on labels/fields for list.
|
||||||
PredicateFunc: func(label labels.Selector, field fields.Selector) *generic.SelectionPredicate {
|
PredicateFunc: func(label labels.Selector, field fields.Selector) storage.SelectionPredicate {
|
||||||
return &generic.SelectionPredicate{
|
return storage.SelectionPredicate{
|
||||||
Label: label,
|
Label: label,
|
||||||
Field: field,
|
Field: field,
|
||||||
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
||||||
|
@ -26,6 +26,7 @@ import (
|
|||||||
"k8s.io/kubernetes/pkg/labels"
|
"k8s.io/kubernetes/pkg/labels"
|
||||||
"k8s.io/kubernetes/pkg/registry/generic"
|
"k8s.io/kubernetes/pkg/registry/generic"
|
||||||
"k8s.io/kubernetes/pkg/runtime"
|
"k8s.io/kubernetes/pkg/runtime"
|
||||||
|
apistorage "k8s.io/kubernetes/pkg/storage"
|
||||||
"k8s.io/kubernetes/pkg/util/validation/field"
|
"k8s.io/kubernetes/pkg/util/validation/field"
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -44,8 +45,8 @@ func ClusterToSelectableFields(cluster *federation.Cluster) fields.Set {
|
|||||||
return generic.ObjectMetaFieldsSet(&cluster.ObjectMeta, false)
|
return generic.ObjectMetaFieldsSet(&cluster.ObjectMeta, false)
|
||||||
}
|
}
|
||||||
|
|
||||||
func MatchCluster(label labels.Selector, field fields.Selector) *generic.SelectionPredicate {
|
func MatchCluster(label labels.Selector, field fields.Selector) apistorage.SelectionPredicate {
|
||||||
return &generic.SelectionPredicate{
|
return apistorage.SelectionPredicate{
|
||||||
Label: label,
|
Label: label,
|
||||||
Field: field,
|
Field: field,
|
||||||
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
||||||
|
@ -277,7 +277,7 @@ func ListResource(r rest.Lister, rw rest.Watcher, scope RequestScope, forceWatch
|
|||||||
|
|
||||||
if hasName {
|
if hasName {
|
||||||
// metadata.name is the canonical internal name.
|
// metadata.name is the canonical internal name.
|
||||||
// generic.SelectionPredicate will notice that this is
|
// SelectionPredicate will notice that this is
|
||||||
// a request for a single object and optimize the
|
// a request for a single object and optimize the
|
||||||
// storage query accordingly.
|
// storage query accordingly.
|
||||||
nameSelector := fields.OneTermEqualSelector("metadata.name", name)
|
nameSelector := fields.OneTermEqualSelector("metadata.name", name)
|
||||||
|
@ -27,6 +27,7 @@ import (
|
|||||||
"k8s.io/kubernetes/pkg/labels"
|
"k8s.io/kubernetes/pkg/labels"
|
||||||
"k8s.io/kubernetes/pkg/registry/generic"
|
"k8s.io/kubernetes/pkg/registry/generic"
|
||||||
"k8s.io/kubernetes/pkg/runtime"
|
"k8s.io/kubernetes/pkg/runtime"
|
||||||
|
"k8s.io/kubernetes/pkg/storage"
|
||||||
"k8s.io/kubernetes/pkg/util/validation/field"
|
"k8s.io/kubernetes/pkg/util/validation/field"
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -103,8 +104,8 @@ func PetSetToSelectableFields(petSet *apps.PetSet) fields.Set {
|
|||||||
|
|
||||||
// MatchPetSet is the filter used by the generic etcd backend to watch events
|
// MatchPetSet is the filter used by the generic etcd backend to watch events
|
||||||
// from etcd to clients of the apiserver only interested in specific labels/fields.
|
// from etcd to clients of the apiserver only interested in specific labels/fields.
|
||||||
func MatchPetSet(label labels.Selector, field fields.Selector) *generic.SelectionPredicate {
|
func MatchPetSet(label labels.Selector, field fields.Selector) storage.SelectionPredicate {
|
||||||
return &generic.SelectionPredicate{
|
return storage.SelectionPredicate{
|
||||||
Label: label,
|
Label: label,
|
||||||
Field: field,
|
Field: field,
|
||||||
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
||||||
|
@ -24,8 +24,8 @@ import (
|
|||||||
"k8s.io/kubernetes/pkg/apis/autoscaling/validation"
|
"k8s.io/kubernetes/pkg/apis/autoscaling/validation"
|
||||||
"k8s.io/kubernetes/pkg/fields"
|
"k8s.io/kubernetes/pkg/fields"
|
||||||
"k8s.io/kubernetes/pkg/labels"
|
"k8s.io/kubernetes/pkg/labels"
|
||||||
"k8s.io/kubernetes/pkg/registry/generic"
|
|
||||||
"k8s.io/kubernetes/pkg/runtime"
|
"k8s.io/kubernetes/pkg/runtime"
|
||||||
|
"k8s.io/kubernetes/pkg/storage"
|
||||||
"k8s.io/kubernetes/pkg/util/validation/field"
|
"k8s.io/kubernetes/pkg/util/validation/field"
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -88,8 +88,8 @@ func AutoscalerToSelectableFields(hpa *autoscaling.HorizontalPodAutoscaler) fiel
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func MatchAutoscaler(label labels.Selector, field fields.Selector) *generic.SelectionPredicate {
|
func MatchAutoscaler(label labels.Selector, field fields.Selector) storage.SelectionPredicate {
|
||||||
return &generic.SelectionPredicate{
|
return storage.SelectionPredicate{
|
||||||
Label: label,
|
Label: label,
|
||||||
Field: field,
|
Field: field,
|
||||||
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
||||||
|
@ -28,6 +28,7 @@ import (
|
|||||||
"k8s.io/kubernetes/pkg/labels"
|
"k8s.io/kubernetes/pkg/labels"
|
||||||
"k8s.io/kubernetes/pkg/registry/generic"
|
"k8s.io/kubernetes/pkg/registry/generic"
|
||||||
"k8s.io/kubernetes/pkg/runtime"
|
"k8s.io/kubernetes/pkg/runtime"
|
||||||
|
"k8s.io/kubernetes/pkg/storage"
|
||||||
"k8s.io/kubernetes/pkg/util/validation/field"
|
"k8s.io/kubernetes/pkg/util/validation/field"
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -166,8 +167,8 @@ func JobToSelectableFields(job *batch.Job) fields.Set {
|
|||||||
// MatchJob is the filter used by the generic etcd backend to route
|
// MatchJob is the filter used by the generic etcd backend to route
|
||||||
// watch events from etcd to clients of the apiserver only interested in specific
|
// watch events from etcd to clients of the apiserver only interested in specific
|
||||||
// labels/fields.
|
// labels/fields.
|
||||||
func MatchJob(label labels.Selector, field fields.Selector) *generic.SelectionPredicate {
|
func MatchJob(label labels.Selector, field fields.Selector) storage.SelectionPredicate {
|
||||||
return &generic.SelectionPredicate{
|
return storage.SelectionPredicate{
|
||||||
Label: label,
|
Label: label,
|
||||||
Field: field,
|
Field: field,
|
||||||
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
||||||
|
@ -26,6 +26,7 @@ import (
|
|||||||
"k8s.io/kubernetes/pkg/labels"
|
"k8s.io/kubernetes/pkg/labels"
|
||||||
"k8s.io/kubernetes/pkg/registry/generic"
|
"k8s.io/kubernetes/pkg/registry/generic"
|
||||||
"k8s.io/kubernetes/pkg/runtime"
|
"k8s.io/kubernetes/pkg/runtime"
|
||||||
|
"k8s.io/kubernetes/pkg/storage"
|
||||||
"k8s.io/kubernetes/pkg/util/validation/field"
|
"k8s.io/kubernetes/pkg/util/validation/field"
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -104,8 +105,8 @@ func ScheduledJobToSelectableFields(scheduledJob *batch.ScheduledJob) fields.Set
|
|||||||
// MatchScheduledJob is the filter used by the generic etcd backend to route
|
// MatchScheduledJob is the filter used by the generic etcd backend to route
|
||||||
// watch events from etcd to clients of the apiserver only interested in specific
|
// watch events from etcd to clients of the apiserver only interested in specific
|
||||||
// labels/fields.
|
// labels/fields.
|
||||||
func MatchScheduledJob(label labels.Selector, field fields.Selector) *generic.SelectionPredicate {
|
func MatchScheduledJob(label labels.Selector, field fields.Selector) storage.SelectionPredicate {
|
||||||
return &generic.SelectionPredicate{
|
return storage.SelectionPredicate{
|
||||||
Label: label,
|
Label: label,
|
||||||
Field: field,
|
Field: field,
|
||||||
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
||||||
|
@ -62,7 +62,7 @@ func NewREST(opts generic.RESTOptions) (*REST, *StatusREST, *ApprovalREST) {
|
|||||||
ObjectNameFunc: func(obj runtime.Object) (string, error) {
|
ObjectNameFunc: func(obj runtime.Object) (string, error) {
|
||||||
return obj.(*certificates.CertificateSigningRequest).Name, nil
|
return obj.(*certificates.CertificateSigningRequest).Name, nil
|
||||||
},
|
},
|
||||||
PredicateFunc: func(label labels.Selector, field fields.Selector) *generic.SelectionPredicate {
|
PredicateFunc: func(label labels.Selector, field fields.Selector) storage.SelectionPredicate {
|
||||||
return csrregistry.Matcher(label, field)
|
return csrregistry.Matcher(label, field)
|
||||||
},
|
},
|
||||||
QualifiedResource: certificates.Resource("certificatesigningrequests"),
|
QualifiedResource: certificates.Resource("certificatesigningrequests"),
|
||||||
|
@ -26,6 +26,7 @@ import (
|
|||||||
"k8s.io/kubernetes/pkg/labels"
|
"k8s.io/kubernetes/pkg/labels"
|
||||||
"k8s.io/kubernetes/pkg/registry/generic"
|
"k8s.io/kubernetes/pkg/registry/generic"
|
||||||
"k8s.io/kubernetes/pkg/runtime"
|
"k8s.io/kubernetes/pkg/runtime"
|
||||||
|
apistorage "k8s.io/kubernetes/pkg/storage"
|
||||||
"k8s.io/kubernetes/pkg/util/validation/field"
|
"k8s.io/kubernetes/pkg/util/validation/field"
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -168,8 +169,8 @@ func (csrApprovalStrategy) ValidateUpdate(ctx api.Context, obj, old runtime.Obje
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Matcher returns a generic matcher for a given label and field selector.
|
// Matcher returns a generic matcher for a given label and field selector.
|
||||||
func Matcher(label labels.Selector, field fields.Selector) *generic.SelectionPredicate {
|
func Matcher(label labels.Selector, field fields.Selector) apistorage.SelectionPredicate {
|
||||||
return &generic.SelectionPredicate{
|
return apistorage.SelectionPredicate{
|
||||||
Label: label,
|
Label: label,
|
||||||
Field: field,
|
Field: field,
|
||||||
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
||||||
|
@ -26,6 +26,7 @@ import (
|
|||||||
"k8s.io/kubernetes/pkg/labels"
|
"k8s.io/kubernetes/pkg/labels"
|
||||||
"k8s.io/kubernetes/pkg/registry/generic"
|
"k8s.io/kubernetes/pkg/registry/generic"
|
||||||
"k8s.io/kubernetes/pkg/runtime"
|
"k8s.io/kubernetes/pkg/runtime"
|
||||||
|
apistorage "k8s.io/kubernetes/pkg/storage"
|
||||||
"k8s.io/kubernetes/pkg/util/validation/field"
|
"k8s.io/kubernetes/pkg/util/validation/field"
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -88,8 +89,8 @@ func ConfigMapToSelectableFields(cfg *api.ConfigMap) fields.Set {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// MatchConfigMap returns a generic matcher for a given label and field selector.
|
// MatchConfigMap returns a generic matcher for a given label and field selector.
|
||||||
func MatchConfigMap(label labels.Selector, field fields.Selector) *generic.SelectionPredicate {
|
func MatchConfigMap(label labels.Selector, field fields.Selector) apistorage.SelectionPredicate {
|
||||||
return &generic.SelectionPredicate{
|
return apistorage.SelectionPredicate{
|
||||||
Label: label,
|
Label: label,
|
||||||
Field: field,
|
Field: field,
|
||||||
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
||||||
|
@ -30,6 +30,7 @@ import (
|
|||||||
"k8s.io/kubernetes/pkg/labels"
|
"k8s.io/kubernetes/pkg/labels"
|
||||||
"k8s.io/kubernetes/pkg/registry/generic"
|
"k8s.io/kubernetes/pkg/registry/generic"
|
||||||
"k8s.io/kubernetes/pkg/runtime"
|
"k8s.io/kubernetes/pkg/runtime"
|
||||||
|
apistorage "k8s.io/kubernetes/pkg/storage"
|
||||||
"k8s.io/kubernetes/pkg/util/validation/field"
|
"k8s.io/kubernetes/pkg/util/validation/field"
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -120,8 +121,8 @@ func ControllerToSelectableFields(controller *api.ReplicationController) fields.
|
|||||||
// MatchController is the filter used by the generic etcd backend to route
|
// MatchController is the filter used by the generic etcd backend to route
|
||||||
// watch events from etcd to clients of the apiserver only interested in specific
|
// watch events from etcd to clients of the apiserver only interested in specific
|
||||||
// labels/fields.
|
// labels/fields.
|
||||||
func MatchController(label labels.Selector, field fields.Selector) *generic.SelectionPredicate {
|
func MatchController(label labels.Selector, field fields.Selector) apistorage.SelectionPredicate {
|
||||||
return &generic.SelectionPredicate{
|
return apistorage.SelectionPredicate{
|
||||||
Label: label,
|
Label: label,
|
||||||
Field: field,
|
Field: field,
|
||||||
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
||||||
|
@ -26,6 +26,7 @@ import (
|
|||||||
"k8s.io/kubernetes/pkg/labels"
|
"k8s.io/kubernetes/pkg/labels"
|
||||||
"k8s.io/kubernetes/pkg/registry/generic"
|
"k8s.io/kubernetes/pkg/registry/generic"
|
||||||
"k8s.io/kubernetes/pkg/runtime"
|
"k8s.io/kubernetes/pkg/runtime"
|
||||||
|
apistorage "k8s.io/kubernetes/pkg/storage"
|
||||||
"k8s.io/kubernetes/pkg/util/validation/field"
|
"k8s.io/kubernetes/pkg/util/validation/field"
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -79,12 +80,12 @@ func (endpointsStrategy) AllowUnconditionalUpdate() bool {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// MatchEndpoints returns a generic matcher for a given label and field selector.
|
// MatchEndpoints returns a generic matcher for a given label and field selector.
|
||||||
func MatchEndpoints(label labels.Selector, field fields.Selector) *generic.SelectionPredicate {
|
func MatchEndpoints(label labels.Selector, field fields.Selector) apistorage.SelectionPredicate {
|
||||||
return &generic.SelectionPredicate{Label: label, Field: field, GetAttrs: EndpointsAttributes}
|
return apistorage.SelectionPredicate{Label: label, Field: field, GetAttrs: EndpointsAttributes}
|
||||||
}
|
}
|
||||||
|
|
||||||
// EndpointsAttributes returns the attributes of an endpoint such that a
|
// EndpointsAttributes returns the attributes of an endpoint such that a
|
||||||
// generic.SelectionPredicate can match appropriately.
|
// SelectionPredicate can match appropriately.
|
||||||
func EndpointsAttributes(obj runtime.Object) (objLabels labels.Set, objFields fields.Set, err error) {
|
func EndpointsAttributes(obj runtime.Object) (objLabels labels.Set, objFields fields.Set, err error) {
|
||||||
endpoints, ok := obj.(*api.Endpoints)
|
endpoints, ok := obj.(*api.Endpoints)
|
||||||
if !ok {
|
if !ok {
|
||||||
|
@ -25,6 +25,7 @@ import (
|
|||||||
"k8s.io/kubernetes/pkg/labels"
|
"k8s.io/kubernetes/pkg/labels"
|
||||||
"k8s.io/kubernetes/pkg/registry/generic"
|
"k8s.io/kubernetes/pkg/registry/generic"
|
||||||
"k8s.io/kubernetes/pkg/runtime"
|
"k8s.io/kubernetes/pkg/runtime"
|
||||||
|
"k8s.io/kubernetes/pkg/storage"
|
||||||
"k8s.io/kubernetes/pkg/util/validation/field"
|
"k8s.io/kubernetes/pkg/util/validation/field"
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -69,8 +70,8 @@ func (eventStrategy) AllowUnconditionalUpdate() bool {
|
|||||||
return true
|
return true
|
||||||
}
|
}
|
||||||
|
|
||||||
func MatchEvent(label labels.Selector, field fields.Selector) *generic.SelectionPredicate {
|
func MatchEvent(label labels.Selector, field fields.Selector) storage.SelectionPredicate {
|
||||||
return &generic.SelectionPredicate{
|
return storage.SelectionPredicate{
|
||||||
Label: label,
|
Label: label,
|
||||||
Field: field,
|
Field: field,
|
||||||
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
||||||
|
@ -23,8 +23,8 @@ import (
|
|||||||
"k8s.io/kubernetes/pkg/api/validation"
|
"k8s.io/kubernetes/pkg/api/validation"
|
||||||
"k8s.io/kubernetes/pkg/fields"
|
"k8s.io/kubernetes/pkg/fields"
|
||||||
"k8s.io/kubernetes/pkg/labels"
|
"k8s.io/kubernetes/pkg/labels"
|
||||||
"k8s.io/kubernetes/pkg/registry/generic"
|
|
||||||
"k8s.io/kubernetes/pkg/runtime"
|
"k8s.io/kubernetes/pkg/runtime"
|
||||||
|
"k8s.io/kubernetes/pkg/storage"
|
||||||
"k8s.io/kubernetes/pkg/util/uuid"
|
"k8s.io/kubernetes/pkg/util/uuid"
|
||||||
"k8s.io/kubernetes/pkg/util/validation/field"
|
"k8s.io/kubernetes/pkg/util/validation/field"
|
||||||
)
|
)
|
||||||
@ -85,8 +85,8 @@ func (limitrangeStrategy) Export(api.Context, runtime.Object, bool) error {
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func MatchLimitRange(label labels.Selector, field fields.Selector) *generic.SelectionPredicate {
|
func MatchLimitRange(label labels.Selector, field fields.Selector) storage.SelectionPredicate {
|
||||||
return &generic.SelectionPredicate{
|
return storage.SelectionPredicate{
|
||||||
Label: label,
|
Label: label,
|
||||||
Field: field,
|
Field: field,
|
||||||
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
||||||
|
@ -25,6 +25,7 @@ import (
|
|||||||
"k8s.io/kubernetes/pkg/labels"
|
"k8s.io/kubernetes/pkg/labels"
|
||||||
"k8s.io/kubernetes/pkg/registry/generic"
|
"k8s.io/kubernetes/pkg/registry/generic"
|
||||||
"k8s.io/kubernetes/pkg/runtime"
|
"k8s.io/kubernetes/pkg/runtime"
|
||||||
|
apistorage "k8s.io/kubernetes/pkg/storage"
|
||||||
"k8s.io/kubernetes/pkg/util/validation/field"
|
"k8s.io/kubernetes/pkg/util/validation/field"
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -135,8 +136,8 @@ func (namespaceFinalizeStrategy) PrepareForUpdate(ctx api.Context, obj, old runt
|
|||||||
}
|
}
|
||||||
|
|
||||||
// MatchNamespace returns a generic matcher for a given label and field selector.
|
// MatchNamespace returns a generic matcher for a given label and field selector.
|
||||||
func MatchNamespace(label labels.Selector, field fields.Selector) *generic.SelectionPredicate {
|
func MatchNamespace(label labels.Selector, field fields.Selector) apistorage.SelectionPredicate {
|
||||||
return &generic.SelectionPredicate{
|
return apistorage.SelectionPredicate{
|
||||||
Label: label,
|
Label: label,
|
||||||
Field: field,
|
Field: field,
|
||||||
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
||||||
|
@ -147,8 +147,8 @@ func NodeToSelectableFields(node *api.Node) fields.Set {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// MatchNode returns a generic matcher for a given label and field selector.
|
// MatchNode returns a generic matcher for a given label and field selector.
|
||||||
func MatchNode(label labels.Selector, field fields.Selector) *generic.SelectionPredicate {
|
func MatchNode(label labels.Selector, field fields.Selector) pkgstorage.SelectionPredicate {
|
||||||
return &generic.SelectionPredicate{
|
return pkgstorage.SelectionPredicate{
|
||||||
Label: label,
|
Label: label,
|
||||||
Field: field,
|
Field: field,
|
||||||
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
||||||
|
@ -25,6 +25,7 @@ import (
|
|||||||
"k8s.io/kubernetes/pkg/labels"
|
"k8s.io/kubernetes/pkg/labels"
|
||||||
"k8s.io/kubernetes/pkg/registry/generic"
|
"k8s.io/kubernetes/pkg/registry/generic"
|
||||||
"k8s.io/kubernetes/pkg/runtime"
|
"k8s.io/kubernetes/pkg/runtime"
|
||||||
|
"k8s.io/kubernetes/pkg/storage"
|
||||||
"k8s.io/kubernetes/pkg/util/validation/field"
|
"k8s.io/kubernetes/pkg/util/validation/field"
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -95,8 +96,8 @@ func (persistentvolumeStatusStrategy) ValidateUpdate(ctx api.Context, obj, old r
|
|||||||
}
|
}
|
||||||
|
|
||||||
// MatchPersistentVolume returns a generic matcher for a given label and field selector.
|
// MatchPersistentVolume returns a generic matcher for a given label and field selector.
|
||||||
func MatchPersistentVolumes(label labels.Selector, field fields.Selector) *generic.SelectionPredicate {
|
func MatchPersistentVolumes(label labels.Selector, field fields.Selector) storage.SelectionPredicate {
|
||||||
return &generic.SelectionPredicate{
|
return storage.SelectionPredicate{
|
||||||
Label: label,
|
Label: label,
|
||||||
Field: field,
|
Field: field,
|
||||||
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
||||||
|
@ -25,6 +25,7 @@ import (
|
|||||||
"k8s.io/kubernetes/pkg/labels"
|
"k8s.io/kubernetes/pkg/labels"
|
||||||
"k8s.io/kubernetes/pkg/registry/generic"
|
"k8s.io/kubernetes/pkg/registry/generic"
|
||||||
"k8s.io/kubernetes/pkg/runtime"
|
"k8s.io/kubernetes/pkg/runtime"
|
||||||
|
"k8s.io/kubernetes/pkg/storage"
|
||||||
"k8s.io/kubernetes/pkg/util/validation/field"
|
"k8s.io/kubernetes/pkg/util/validation/field"
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -95,8 +96,8 @@ func (persistentvolumeclaimStatusStrategy) ValidateUpdate(ctx api.Context, obj,
|
|||||||
}
|
}
|
||||||
|
|
||||||
// MatchPersistentVolumeClaim returns a generic matcher for a given label and field selector.
|
// MatchPersistentVolumeClaim returns a generic matcher for a given label and field selector.
|
||||||
func MatchPersistentVolumeClaim(label labels.Selector, field fields.Selector) *generic.SelectionPredicate {
|
func MatchPersistentVolumeClaim(label labels.Selector, field fields.Selector) storage.SelectionPredicate {
|
||||||
return &generic.SelectionPredicate{
|
return storage.SelectionPredicate{
|
||||||
Label: label,
|
Label: label,
|
||||||
Field: field,
|
Field: field,
|
||||||
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
||||||
|
@ -155,8 +155,8 @@ func (podStatusStrategy) ValidateUpdate(ctx api.Context, obj, old runtime.Object
|
|||||||
}
|
}
|
||||||
|
|
||||||
// MatchPod returns a generic matcher for a given label and field selector.
|
// MatchPod returns a generic matcher for a given label and field selector.
|
||||||
func MatchPod(label labels.Selector, field fields.Selector) *generic.SelectionPredicate {
|
func MatchPod(label labels.Selector, field fields.Selector) storage.SelectionPredicate {
|
||||||
return &generic.SelectionPredicate{
|
return storage.SelectionPredicate{
|
||||||
Label: label,
|
Label: label,
|
||||||
Field: field,
|
Field: field,
|
||||||
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
||||||
|
@ -79,7 +79,8 @@ func TestMatchPod(t *testing.T) {
|
|||||||
},
|
},
|
||||||
}
|
}
|
||||||
for _, testCase := range testCases {
|
for _, testCase := range testCases {
|
||||||
result, err := MatchPod(labels.Everything(), testCase.fieldSelector).Matches(testCase.in)
|
m := MatchPod(labels.Everything(), testCase.fieldSelector)
|
||||||
|
result, err := m.Matches(testCase.in)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Errorf("Unexpected error %v", err)
|
t.Errorf("Unexpected error %v", err)
|
||||||
}
|
}
|
||||||
|
@ -23,8 +23,8 @@ import (
|
|||||||
"k8s.io/kubernetes/pkg/api/validation"
|
"k8s.io/kubernetes/pkg/api/validation"
|
||||||
"k8s.io/kubernetes/pkg/fields"
|
"k8s.io/kubernetes/pkg/fields"
|
||||||
"k8s.io/kubernetes/pkg/labels"
|
"k8s.io/kubernetes/pkg/labels"
|
||||||
"k8s.io/kubernetes/pkg/registry/generic"
|
|
||||||
"k8s.io/kubernetes/pkg/runtime"
|
"k8s.io/kubernetes/pkg/runtime"
|
||||||
|
"k8s.io/kubernetes/pkg/storage"
|
||||||
"k8s.io/kubernetes/pkg/util/validation/field"
|
"k8s.io/kubernetes/pkg/util/validation/field"
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -86,8 +86,8 @@ func PodTemplateToSelectableFields(podTemplate *api.PodTemplate) fields.Set {
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func MatchPodTemplate(label labels.Selector, field fields.Selector) *generic.SelectionPredicate {
|
func MatchPodTemplate(label labels.Selector, field fields.Selector) storage.SelectionPredicate {
|
||||||
return &generic.SelectionPredicate{
|
return storage.SelectionPredicate{
|
||||||
Label: label,
|
Label: label,
|
||||||
Field: field,
|
Field: field,
|
||||||
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
||||||
|
@ -25,6 +25,7 @@ import (
|
|||||||
"k8s.io/kubernetes/pkg/labels"
|
"k8s.io/kubernetes/pkg/labels"
|
||||||
"k8s.io/kubernetes/pkg/registry/generic"
|
"k8s.io/kubernetes/pkg/registry/generic"
|
||||||
"k8s.io/kubernetes/pkg/runtime"
|
"k8s.io/kubernetes/pkg/runtime"
|
||||||
|
"k8s.io/kubernetes/pkg/storage"
|
||||||
"k8s.io/kubernetes/pkg/util/validation/field"
|
"k8s.io/kubernetes/pkg/util/validation/field"
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -98,8 +99,8 @@ func (resourcequotaStatusStrategy) ValidateUpdate(ctx api.Context, obj, old runt
|
|||||||
}
|
}
|
||||||
|
|
||||||
// MatchResourceQuota returns a generic matcher for a given label and field selector.
|
// MatchResourceQuota returns a generic matcher for a given label and field selector.
|
||||||
func MatchResourceQuota(label labels.Selector, field fields.Selector) *generic.SelectionPredicate {
|
func MatchResourceQuota(label labels.Selector, field fields.Selector) storage.SelectionPredicate {
|
||||||
return &generic.SelectionPredicate{
|
return storage.SelectionPredicate{
|
||||||
Label: label,
|
Label: label,
|
||||||
Field: field,
|
Field: field,
|
||||||
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
||||||
|
@ -27,6 +27,7 @@ import (
|
|||||||
"k8s.io/kubernetes/pkg/labels"
|
"k8s.io/kubernetes/pkg/labels"
|
||||||
"k8s.io/kubernetes/pkg/registry/generic"
|
"k8s.io/kubernetes/pkg/registry/generic"
|
||||||
"k8s.io/kubernetes/pkg/runtime"
|
"k8s.io/kubernetes/pkg/runtime"
|
||||||
|
apistorage "k8s.io/kubernetes/pkg/storage"
|
||||||
"k8s.io/kubernetes/pkg/util/validation/field"
|
"k8s.io/kubernetes/pkg/util/validation/field"
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -94,8 +95,8 @@ func (s strategy) Export(ctx api.Context, obj runtime.Object, exact bool) error
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Matcher returns a generic matcher for a given label and field selector.
|
// Matcher returns a generic matcher for a given label and field selector.
|
||||||
func Matcher(label labels.Selector, field fields.Selector) *generic.SelectionPredicate {
|
func Matcher(label labels.Selector, field fields.Selector) apistorage.SelectionPredicate {
|
||||||
return &generic.SelectionPredicate{
|
return apistorage.SelectionPredicate{
|
||||||
Label: label,
|
Label: label,
|
||||||
Field: field,
|
Field: field,
|
||||||
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
||||||
|
@ -25,6 +25,7 @@ import (
|
|||||||
"k8s.io/kubernetes/pkg/labels"
|
"k8s.io/kubernetes/pkg/labels"
|
||||||
"k8s.io/kubernetes/pkg/registry/generic"
|
"k8s.io/kubernetes/pkg/registry/generic"
|
||||||
"k8s.io/kubernetes/pkg/runtime"
|
"k8s.io/kubernetes/pkg/runtime"
|
||||||
|
apistorage "k8s.io/kubernetes/pkg/storage"
|
||||||
"k8s.io/kubernetes/pkg/util/validation/field"
|
"k8s.io/kubernetes/pkg/util/validation/field"
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -100,8 +101,8 @@ func (svcStrategy) Export(ctx api.Context, obj runtime.Object, exact bool) error
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func MatchServices(label labels.Selector, field fields.Selector) *generic.SelectionPredicate {
|
func MatchServices(label labels.Selector, field fields.Selector) apistorage.SelectionPredicate {
|
||||||
return &generic.SelectionPredicate{
|
return apistorage.SelectionPredicate{
|
||||||
Label: label,
|
Label: label,
|
||||||
Field: field,
|
Field: field,
|
||||||
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
||||||
|
@ -25,6 +25,7 @@ import (
|
|||||||
"k8s.io/kubernetes/pkg/labels"
|
"k8s.io/kubernetes/pkg/labels"
|
||||||
"k8s.io/kubernetes/pkg/registry/generic"
|
"k8s.io/kubernetes/pkg/registry/generic"
|
||||||
"k8s.io/kubernetes/pkg/runtime"
|
"k8s.io/kubernetes/pkg/runtime"
|
||||||
|
apistorage "k8s.io/kubernetes/pkg/storage"
|
||||||
"k8s.io/kubernetes/pkg/util/validation/field"
|
"k8s.io/kubernetes/pkg/util/validation/field"
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -77,8 +78,8 @@ func (strategy) AllowUnconditionalUpdate() bool {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Matcher returns a generic matcher for a given label and field selector.
|
// Matcher returns a generic matcher for a given label and field selector.
|
||||||
func Matcher(label labels.Selector, field fields.Selector) *generic.SelectionPredicate {
|
func Matcher(label labels.Selector, field fields.Selector) apistorage.SelectionPredicate {
|
||||||
return &generic.SelectionPredicate{
|
return apistorage.SelectionPredicate{
|
||||||
Label: label,
|
Label: label,
|
||||||
Field: field,
|
Field: field,
|
||||||
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
||||||
|
@ -27,6 +27,7 @@ import (
|
|||||||
"k8s.io/kubernetes/pkg/labels"
|
"k8s.io/kubernetes/pkg/labels"
|
||||||
"k8s.io/kubernetes/pkg/registry/generic"
|
"k8s.io/kubernetes/pkg/registry/generic"
|
||||||
"k8s.io/kubernetes/pkg/runtime"
|
"k8s.io/kubernetes/pkg/runtime"
|
||||||
|
"k8s.io/kubernetes/pkg/storage"
|
||||||
"k8s.io/kubernetes/pkg/util/validation/field"
|
"k8s.io/kubernetes/pkg/util/validation/field"
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -112,8 +113,8 @@ func DaemonSetToSelectableFields(daemon *extensions.DaemonSet) fields.Set {
|
|||||||
// MatchSetDaemon is the filter used by the generic etcd backend to route
|
// MatchSetDaemon is the filter used by the generic etcd backend to route
|
||||||
// watch events from etcd to clients of the apiserver only interested in specific
|
// watch events from etcd to clients of the apiserver only interested in specific
|
||||||
// labels/fields.
|
// labels/fields.
|
||||||
func MatchDaemonSet(label labels.Selector, field fields.Selector) *generic.SelectionPredicate {
|
func MatchDaemonSet(label labels.Selector, field fields.Selector) storage.SelectionPredicate {
|
||||||
return &generic.SelectionPredicate{
|
return storage.SelectionPredicate{
|
||||||
Label: label,
|
Label: label,
|
||||||
Field: field,
|
Field: field,
|
||||||
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
||||||
|
@ -30,6 +30,7 @@ import (
|
|||||||
"k8s.io/kubernetes/pkg/labels"
|
"k8s.io/kubernetes/pkg/labels"
|
||||||
"k8s.io/kubernetes/pkg/registry/generic"
|
"k8s.io/kubernetes/pkg/registry/generic"
|
||||||
"k8s.io/kubernetes/pkg/runtime"
|
"k8s.io/kubernetes/pkg/runtime"
|
||||||
|
apistorage "k8s.io/kubernetes/pkg/storage"
|
||||||
"k8s.io/kubernetes/pkg/util/validation/field"
|
"k8s.io/kubernetes/pkg/util/validation/field"
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -130,8 +131,8 @@ func DeploymentToSelectableFields(deployment *extensions.Deployment) fields.Set
|
|||||||
// MatchDeployment is the filter used by the generic etcd backend to route
|
// MatchDeployment is the filter used by the generic etcd backend to route
|
||||||
// watch events from etcd to clients of the apiserver only interested in specific
|
// watch events from etcd to clients of the apiserver only interested in specific
|
||||||
// labels/fields.
|
// labels/fields.
|
||||||
func MatchDeployment(label labels.Selector, field fields.Selector) *generic.SelectionPredicate {
|
func MatchDeployment(label labels.Selector, field fields.Selector) apistorage.SelectionPredicate {
|
||||||
return &generic.SelectionPredicate{
|
return apistorage.SelectionPredicate{
|
||||||
Label: label,
|
Label: label,
|
||||||
Field: field,
|
Field: field,
|
||||||
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
||||||
|
@ -27,6 +27,7 @@ import (
|
|||||||
"k8s.io/kubernetes/pkg/labels"
|
"k8s.io/kubernetes/pkg/labels"
|
||||||
"k8s.io/kubernetes/pkg/registry/generic"
|
"k8s.io/kubernetes/pkg/registry/generic"
|
||||||
"k8s.io/kubernetes/pkg/runtime"
|
"k8s.io/kubernetes/pkg/runtime"
|
||||||
|
"k8s.io/kubernetes/pkg/storage"
|
||||||
"k8s.io/kubernetes/pkg/util/validation/field"
|
"k8s.io/kubernetes/pkg/util/validation/field"
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -105,8 +106,8 @@ func IngressToSelectableFields(ingress *extensions.Ingress) fields.Set {
|
|||||||
// MatchIngress is the filter used by the generic etcd backend to ingress
|
// MatchIngress is the filter used by the generic etcd backend to ingress
|
||||||
// watch events from etcd to clients of the apiserver only interested in specific
|
// watch events from etcd to clients of the apiserver only interested in specific
|
||||||
// labels/fields.
|
// labels/fields.
|
||||||
func MatchIngress(label labels.Selector, field fields.Selector) *generic.SelectionPredicate {
|
func MatchIngress(label labels.Selector, field fields.Selector) storage.SelectionPredicate {
|
||||||
return &generic.SelectionPredicate{
|
return storage.SelectionPredicate{
|
||||||
Label: label,
|
Label: label,
|
||||||
Field: field,
|
Field: field,
|
||||||
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
||||||
|
@ -27,6 +27,7 @@ import (
|
|||||||
"k8s.io/kubernetes/pkg/labels"
|
"k8s.io/kubernetes/pkg/labels"
|
||||||
"k8s.io/kubernetes/pkg/registry/generic"
|
"k8s.io/kubernetes/pkg/registry/generic"
|
||||||
"k8s.io/kubernetes/pkg/runtime"
|
"k8s.io/kubernetes/pkg/runtime"
|
||||||
|
"k8s.io/kubernetes/pkg/storage"
|
||||||
"k8s.io/kubernetes/pkg/util/validation/field"
|
"k8s.io/kubernetes/pkg/util/validation/field"
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -97,8 +98,8 @@ func NetworkPolicyToSelectableFields(networkPolicy *extensions.NetworkPolicy) fi
|
|||||||
|
|
||||||
// MatchNetworkPolicy is the filter used by the generic etcd backend to watch events
|
// MatchNetworkPolicy is the filter used by the generic etcd backend to watch events
|
||||||
// from etcd to clients of the apiserver only interested in specific labels/fields.
|
// from etcd to clients of the apiserver only interested in specific labels/fields.
|
||||||
func MatchNetworkPolicy(label labels.Selector, field fields.Selector) *generic.SelectionPredicate {
|
func MatchNetworkPolicy(label labels.Selector, field fields.Selector) storage.SelectionPredicate {
|
||||||
return &generic.SelectionPredicate{
|
return storage.SelectionPredicate{
|
||||||
Label: label,
|
Label: label,
|
||||||
Field: field,
|
Field: field,
|
||||||
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
||||||
|
@ -27,6 +27,7 @@ import (
|
|||||||
"k8s.io/kubernetes/pkg/labels"
|
"k8s.io/kubernetes/pkg/labels"
|
||||||
"k8s.io/kubernetes/pkg/registry/generic"
|
"k8s.io/kubernetes/pkg/registry/generic"
|
||||||
"k8s.io/kubernetes/pkg/runtime"
|
"k8s.io/kubernetes/pkg/runtime"
|
||||||
|
"k8s.io/kubernetes/pkg/storage"
|
||||||
"k8s.io/kubernetes/pkg/util/validation/field"
|
"k8s.io/kubernetes/pkg/util/validation/field"
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -74,8 +75,8 @@ func (strategy) ValidateUpdate(ctx api.Context, obj, old runtime.Object) field.E
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Matcher returns a generic matcher for a given label and field selector.
|
// Matcher returns a generic matcher for a given label and field selector.
|
||||||
func MatchPodSecurityPolicy(label labels.Selector, field fields.Selector) *generic.SelectionPredicate {
|
func MatchPodSecurityPolicy(label labels.Selector, field fields.Selector) storage.SelectionPredicate {
|
||||||
return &generic.SelectionPredicate{
|
return storage.SelectionPredicate{
|
||||||
Label: label,
|
Label: label,
|
||||||
Field: field,
|
Field: field,
|
||||||
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
||||||
|
@ -31,6 +31,7 @@ import (
|
|||||||
"k8s.io/kubernetes/pkg/labels"
|
"k8s.io/kubernetes/pkg/labels"
|
||||||
"k8s.io/kubernetes/pkg/registry/generic"
|
"k8s.io/kubernetes/pkg/registry/generic"
|
||||||
"k8s.io/kubernetes/pkg/runtime"
|
"k8s.io/kubernetes/pkg/runtime"
|
||||||
|
apistorage "k8s.io/kubernetes/pkg/storage"
|
||||||
"k8s.io/kubernetes/pkg/util/validation/field"
|
"k8s.io/kubernetes/pkg/util/validation/field"
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -121,8 +122,8 @@ func ReplicaSetToSelectableFields(rs *extensions.ReplicaSet) fields.Set {
|
|||||||
// MatchReplicaSet is the filter used by the generic etcd backend to route
|
// MatchReplicaSet is the filter used by the generic etcd backend to route
|
||||||
// watch events from etcd to clients of the apiserver only interested in specific
|
// watch events from etcd to clients of the apiserver only interested in specific
|
||||||
// labels/fields.
|
// labels/fields.
|
||||||
func MatchReplicaSet(label labels.Selector, field fields.Selector) *generic.SelectionPredicate {
|
func MatchReplicaSet(label labels.Selector, field fields.Selector) apistorage.SelectionPredicate {
|
||||||
return &generic.SelectionPredicate{
|
return apistorage.SelectionPredicate{
|
||||||
Label: label,
|
Label: label,
|
||||||
Field: field,
|
Field: field,
|
||||||
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
||||||
|
@ -25,8 +25,8 @@ import (
|
|||||||
"k8s.io/kubernetes/pkg/apis/extensions/validation"
|
"k8s.io/kubernetes/pkg/apis/extensions/validation"
|
||||||
"k8s.io/kubernetes/pkg/fields"
|
"k8s.io/kubernetes/pkg/fields"
|
||||||
"k8s.io/kubernetes/pkg/labels"
|
"k8s.io/kubernetes/pkg/labels"
|
||||||
"k8s.io/kubernetes/pkg/registry/generic"
|
|
||||||
"k8s.io/kubernetes/pkg/runtime"
|
"k8s.io/kubernetes/pkg/runtime"
|
||||||
|
"k8s.io/kubernetes/pkg/storage"
|
||||||
"k8s.io/kubernetes/pkg/util/validation/field"
|
"k8s.io/kubernetes/pkg/util/validation/field"
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -78,8 +78,8 @@ func (strategy) AllowUnconditionalUpdate() bool {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Matcher returns a generic matcher for a given label and field selector.
|
// Matcher returns a generic matcher for a given label and field selector.
|
||||||
func Matcher(label labels.Selector, field fields.Selector) *generic.SelectionPredicate {
|
func Matcher(label labels.Selector, field fields.Selector) storage.SelectionPredicate {
|
||||||
return &generic.SelectionPredicate{
|
return storage.SelectionPredicate{
|
||||||
Label: label,
|
Label: label,
|
||||||
Field: field,
|
Field: field,
|
||||||
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
||||||
|
@ -25,8 +25,8 @@ import (
|
|||||||
"k8s.io/kubernetes/pkg/apis/extensions/validation"
|
"k8s.io/kubernetes/pkg/apis/extensions/validation"
|
||||||
"k8s.io/kubernetes/pkg/fields"
|
"k8s.io/kubernetes/pkg/fields"
|
||||||
"k8s.io/kubernetes/pkg/labels"
|
"k8s.io/kubernetes/pkg/labels"
|
||||||
"k8s.io/kubernetes/pkg/registry/generic"
|
|
||||||
"k8s.io/kubernetes/pkg/runtime"
|
"k8s.io/kubernetes/pkg/runtime"
|
||||||
|
apistorage "k8s.io/kubernetes/pkg/storage"
|
||||||
"k8s.io/kubernetes/pkg/util/validation/field"
|
"k8s.io/kubernetes/pkg/util/validation/field"
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -75,8 +75,8 @@ func (strategy) AllowUnconditionalUpdate() bool {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Matcher returns a generic matcher for a given label and field selector.
|
// Matcher returns a generic matcher for a given label and field selector.
|
||||||
func Matcher(label labels.Selector, field fields.Selector) *generic.SelectionPredicate {
|
func Matcher(label labels.Selector, field fields.Selector) apistorage.SelectionPredicate {
|
||||||
return &generic.SelectionPredicate{
|
return apistorage.SelectionPredicate{
|
||||||
Label: label,
|
Label: label,
|
||||||
Field: field,
|
Field: field,
|
||||||
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
||||||
|
@ -19,14 +19,8 @@ package generic
|
|||||||
import (
|
import (
|
||||||
"k8s.io/kubernetes/pkg/api"
|
"k8s.io/kubernetes/pkg/api"
|
||||||
"k8s.io/kubernetes/pkg/fields"
|
"k8s.io/kubernetes/pkg/fields"
|
||||||
"k8s.io/kubernetes/pkg/labels"
|
|
||||||
"k8s.io/kubernetes/pkg/runtime"
|
|
||||||
"k8s.io/kubernetes/pkg/storage"
|
|
||||||
)
|
)
|
||||||
|
|
||||||
// AttrFunc returns label and field sets for List or Watch to compare against, or an error.
|
|
||||||
type AttrFunc func(obj runtime.Object) (label labels.Set, field fields.Set, err error)
|
|
||||||
|
|
||||||
// ObjectMetaFieldsSet returns a fields that represents the ObjectMeta.
|
// ObjectMetaFieldsSet returns a fields that represents the ObjectMeta.
|
||||||
func ObjectMetaFieldsSet(objectMeta *api.ObjectMeta, hasNamespaceField bool) fields.Set {
|
func ObjectMetaFieldsSet(objectMeta *api.ObjectMeta, hasNamespaceField bool) fields.Set {
|
||||||
if !hasNamespaceField {
|
if !hasNamespaceField {
|
||||||
@ -47,79 +41,3 @@ func MergeFieldsSets(source fields.Set, fragment fields.Set) fields.Set {
|
|||||||
}
|
}
|
||||||
return source
|
return source
|
||||||
}
|
}
|
||||||
|
|
||||||
// SelectionPredicate implements a generic predicate that can be passed to
|
|
||||||
// GenericRegistry's List or Watch methods. Implements the Matcher interface.
|
|
||||||
type SelectionPredicate struct {
|
|
||||||
Label labels.Selector
|
|
||||||
Field fields.Selector
|
|
||||||
GetAttrs AttrFunc
|
|
||||||
IndexFields []string
|
|
||||||
}
|
|
||||||
|
|
||||||
// Matches returns true if the given object's labels and fields (as
|
|
||||||
// returned by s.GetAttrs) match s.Label and s.Field. An error is
|
|
||||||
// returned if s.GetAttrs fails.
|
|
||||||
func (s *SelectionPredicate) Matches(obj runtime.Object) (bool, error) {
|
|
||||||
if s.Label.Empty() && s.Field.Empty() {
|
|
||||||
return true, nil
|
|
||||||
}
|
|
||||||
labels, fields, err := s.GetAttrs(obj)
|
|
||||||
if err != nil {
|
|
||||||
return false, err
|
|
||||||
}
|
|
||||||
matched := s.Label.Matches(labels)
|
|
||||||
if s.Field != nil {
|
|
||||||
matched = (matched && s.Field.Matches(fields))
|
|
||||||
}
|
|
||||||
return matched, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// MatchesSingle will return (name, true) if and only if s.Field matches on the object's
|
|
||||||
// name.
|
|
||||||
func (s *SelectionPredicate) MatchesSingle() (string, bool) {
|
|
||||||
// TODO: should be namespace.name
|
|
||||||
if name, ok := s.Field.RequiresExactMatch("metadata.name"); ok {
|
|
||||||
return name, true
|
|
||||||
}
|
|
||||||
return "", false
|
|
||||||
}
|
|
||||||
|
|
||||||
// For any index defined by IndexFields, if a matcher can match only (a subset)
|
|
||||||
// of objects that return <value> for a given index, a pair (<index name>, <value>)
|
|
||||||
// wil be returned.
|
|
||||||
// TODO: Consider supporting also labels.
|
|
||||||
func (s *SelectionPredicate) MatcherIndex() []storage.MatchValue {
|
|
||||||
var result []storage.MatchValue
|
|
||||||
for _, field := range s.IndexFields {
|
|
||||||
if value, ok := s.Field.RequiresExactMatch(field); ok {
|
|
||||||
result = append(result, storage.MatchValue{IndexName: field, Value: value})
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return result
|
|
||||||
}
|
|
||||||
|
|
||||||
// Matcher can return true if an object matches the Matcher's selection
|
|
||||||
// criteria. If it is known that the matcher will match only a single object
|
|
||||||
// then MatchesSingle should return the key of that object and true. This is an
|
|
||||||
// optimization only--Matches() should continue to work.
|
|
||||||
type Matcher interface {
|
|
||||||
// Matches should return true if obj matches this matcher's requirements.
|
|
||||||
Matches(obj runtime.Object) (matchesThisObject bool, err error)
|
|
||||||
|
|
||||||
// If this matcher matches a single object, return the key for that
|
|
||||||
// object and true here. This will greatly increase efficiency. You
|
|
||||||
// must still implement Matches(). Note that key does NOT need to
|
|
||||||
// include the object's namespace.
|
|
||||||
MatchesSingle() (key string, matchesSingleObject bool)
|
|
||||||
|
|
||||||
// For any known index, if a matcher can match only (a subset) of objects
|
|
||||||
// that return <value> for a given index, a pair (<index name>, <value>)
|
|
||||||
// will be returned.
|
|
||||||
MatcherIndex() []storage.MatchValue
|
|
||||||
}
|
|
||||||
|
|
||||||
var (
|
|
||||||
// Assert implementations match the interface.
|
|
||||||
_ = Matcher(&SelectionPredicate{})
|
|
||||||
)
|
|
||||||
|
@ -32,7 +32,6 @@ import (
|
|||||||
"k8s.io/kubernetes/pkg/api/validation/path"
|
"k8s.io/kubernetes/pkg/api/validation/path"
|
||||||
"k8s.io/kubernetes/pkg/fields"
|
"k8s.io/kubernetes/pkg/fields"
|
||||||
"k8s.io/kubernetes/pkg/labels"
|
"k8s.io/kubernetes/pkg/labels"
|
||||||
"k8s.io/kubernetes/pkg/registry/generic"
|
|
||||||
"k8s.io/kubernetes/pkg/runtime"
|
"k8s.io/kubernetes/pkg/runtime"
|
||||||
"k8s.io/kubernetes/pkg/storage"
|
"k8s.io/kubernetes/pkg/storage"
|
||||||
utilruntime "k8s.io/kubernetes/pkg/util/runtime"
|
utilruntime "k8s.io/kubernetes/pkg/util/runtime"
|
||||||
@ -85,7 +84,7 @@ type Store struct {
|
|||||||
TTLFunc func(obj runtime.Object, existing uint64, update bool) (uint64, error)
|
TTLFunc func(obj runtime.Object, existing uint64, update bool) (uint64, error)
|
||||||
|
|
||||||
// Returns a matcher corresponding to the provided labels and fields.
|
// Returns a matcher corresponding to the provided labels and fields.
|
||||||
PredicateFunc func(label labels.Selector, field fields.Selector) *generic.SelectionPredicate
|
PredicateFunc func(label labels.Selector, field fields.Selector) storage.SelectionPredicate
|
||||||
|
|
||||||
// Called to cleanup storage clients.
|
// Called to cleanup storage clients.
|
||||||
DestroyFunc func()
|
DestroyFunc func()
|
||||||
@ -201,12 +200,11 @@ func (e *Store) List(ctx api.Context, options *api.ListOptions) (runtime.Object,
|
|||||||
}
|
}
|
||||||
|
|
||||||
// ListPredicate returns a list of all the items matching m.
|
// ListPredicate returns a list of all the items matching m.
|
||||||
func (e *Store) ListPredicate(ctx api.Context, m *generic.SelectionPredicate, options *api.ListOptions) (runtime.Object, error) {
|
func (e *Store) ListPredicate(ctx api.Context, p storage.SelectionPredicate, options *api.ListOptions) (runtime.Object, error) {
|
||||||
list := e.NewListFunc()
|
list := e.NewListFunc()
|
||||||
filter := e.createFilter(m)
|
if name, ok := p.MatchesSingle(); ok {
|
||||||
if name, ok := m.MatchesSingle(); ok {
|
|
||||||
if key, err := e.KeyFunc(ctx, name); err == nil {
|
if key, err := e.KeyFunc(ctx, name); err == nil {
|
||||||
err := e.Storage.GetToList(ctx, key, filter, list)
|
err := e.Storage.GetToList(ctx, key, p, list)
|
||||||
return list, storeerr.InterpretListError(err, e.QualifiedResource)
|
return list, storeerr.InterpretListError(err, e.QualifiedResource)
|
||||||
}
|
}
|
||||||
// if we cannot extract a key based on the current context, the optimization is skipped
|
// if we cannot extract a key based on the current context, the optimization is skipped
|
||||||
@ -215,7 +213,7 @@ func (e *Store) ListPredicate(ctx api.Context, m *generic.SelectionPredicate, op
|
|||||||
if options == nil {
|
if options == nil {
|
||||||
options = &api.ListOptions{ResourceVersion: "0"}
|
options = &api.ListOptions{ResourceVersion: "0"}
|
||||||
}
|
}
|
||||||
err := e.Storage.List(ctx, e.KeyRootFunc(ctx), options.ResourceVersion, filter, list)
|
err := e.Storage.List(ctx, e.KeyRootFunc(ctx), options.ResourceVersion, p, list)
|
||||||
return list, storeerr.InterpretListError(err, e.QualifiedResource)
|
return list, storeerr.InterpretListError(err, e.QualifiedResource)
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -853,7 +851,7 @@ func (e *Store) finalizeDelete(obj runtime.Object, runHooks bool) (runtime.Objec
|
|||||||
|
|
||||||
// Watch makes a matcher for the given label and field, and calls
|
// Watch makes a matcher for the given label and field, and calls
|
||||||
// WatchPredicate. If possible, you should customize PredicateFunc to produre a
|
// WatchPredicate. If possible, you should customize PredicateFunc to produre a
|
||||||
// matcher that matches by key. generic.SelectionPredicate does this for you
|
// matcher that matches by key. SelectionPredicate does this for you
|
||||||
// automatically.
|
// automatically.
|
||||||
func (e *Store) Watch(ctx api.Context, options *api.ListOptions) (watch.Interface, error) {
|
func (e *Store) Watch(ctx api.Context, options *api.ListOptions) (watch.Interface, error) {
|
||||||
label := labels.Everything()
|
label := labels.Everything()
|
||||||
@ -872,15 +870,13 @@ func (e *Store) Watch(ctx api.Context, options *api.ListOptions) (watch.Interfac
|
|||||||
}
|
}
|
||||||
|
|
||||||
// WatchPredicate starts a watch for the items that m matches.
|
// WatchPredicate starts a watch for the items that m matches.
|
||||||
func (e *Store) WatchPredicate(ctx api.Context, m *generic.SelectionPredicate, resourceVersion string) (watch.Interface, error) {
|
func (e *Store) WatchPredicate(ctx api.Context, p storage.SelectionPredicate, resourceVersion string) (watch.Interface, error) {
|
||||||
filter := e.createFilter(m)
|
if name, ok := p.MatchesSingle(); ok {
|
||||||
|
|
||||||
if name, ok := m.MatchesSingle(); ok {
|
|
||||||
if key, err := e.KeyFunc(ctx, name); err == nil {
|
if key, err := e.KeyFunc(ctx, name); err == nil {
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
w, err := e.Storage.Watch(ctx, key, resourceVersion, filter)
|
w, err := e.Storage.Watch(ctx, key, resourceVersion, p)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
@ -892,7 +888,7 @@ func (e *Store) WatchPredicate(ctx api.Context, m *generic.SelectionPredicate, r
|
|||||||
// if we cannot extract a key based on the current context, the optimization is skipped
|
// if we cannot extract a key based on the current context, the optimization is skipped
|
||||||
}
|
}
|
||||||
|
|
||||||
w, err := e.Storage.WatchList(ctx, e.KeyRootFunc(ctx), resourceVersion, filter)
|
w, err := e.Storage.WatchList(ctx, e.KeyRootFunc(ctx), resourceVersion, p)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
@ -902,18 +898,6 @@ func (e *Store) WatchPredicate(ctx api.Context, m *generic.SelectionPredicate, r
|
|||||||
return w, nil
|
return w, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (e *Store) createFilter(m *generic.SelectionPredicate) storage.Filter {
|
|
||||||
filterFunc := func(obj runtime.Object) bool {
|
|
||||||
matches, err := m.Matches(obj)
|
|
||||||
if err != nil {
|
|
||||||
glog.Errorf("unable to match watch: %v", err)
|
|
||||||
return false
|
|
||||||
}
|
|
||||||
return matches
|
|
||||||
}
|
|
||||||
return storage.NewSimpleFilter(filterFunc, m.MatcherIndex)
|
|
||||||
}
|
|
||||||
|
|
||||||
// calculateTTL is a helper for retrieving the updated TTL for an object or returning an error
|
// calculateTTL is a helper for retrieving the updated TTL for an object or returning an error
|
||||||
// if the TTL cannot be calculated. The defaultTTL is changed to 1 if less than zero. Zero means
|
// if the TTL cannot be calculated. The defaultTTL is changed to 1 if less than zero. Zero means
|
||||||
// no TTL, not expire immediately.
|
// no TTL, not expire immediately.
|
||||||
|
@ -106,14 +106,14 @@ func NewTestGenericStoreRegistry(t *testing.T) (factory.DestroyFunc, *Store) {
|
|||||||
|
|
||||||
// matchPodName returns selection predicate that matches any pod with name in the set.
|
// matchPodName returns selection predicate that matches any pod with name in the set.
|
||||||
// Makes testing simpler.
|
// Makes testing simpler.
|
||||||
func matchPodName(names ...string) *generic.SelectionPredicate {
|
func matchPodName(names ...string) storage.SelectionPredicate {
|
||||||
// Note: even if pod name is a field, we have to use labels,
|
// Note: even if pod name is a field, we have to use labels,
|
||||||
// because field selector doesn't support "IN" operator.
|
// because field selector doesn't support "IN" operator.
|
||||||
l, err := labels.NewRequirement("name", selection.In, sets.NewString(names...))
|
l, err := labels.NewRequirement("name", selection.In, sets.NewString(names...))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
panic("Labels requirement must validate successfully")
|
panic("Labels requirement must validate successfully")
|
||||||
}
|
}
|
||||||
return &generic.SelectionPredicate{
|
return storage.SelectionPredicate{
|
||||||
Label: labels.Everything().Add(*l),
|
Label: labels.Everything().Add(*l),
|
||||||
Field: fields.Everything(),
|
Field: fields.Everything(),
|
||||||
GetAttrs: func(obj runtime.Object) (label labels.Set, field fields.Set, err error) {
|
GetAttrs: func(obj runtime.Object) (label labels.Set, field fields.Set, err error) {
|
||||||
@ -123,8 +123,8 @@ func matchPodName(names ...string) *generic.SelectionPredicate {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func matchEverything() *generic.SelectionPredicate {
|
func matchEverything() storage.SelectionPredicate {
|
||||||
return &generic.SelectionPredicate{
|
return storage.SelectionPredicate{
|
||||||
Label: labels.Everything(),
|
Label: labels.Everything(),
|
||||||
Field: fields.Everything(),
|
Field: fields.Everything(),
|
||||||
GetAttrs: func(obj runtime.Object) (label labels.Set, field fields.Set, err error) {
|
GetAttrs: func(obj runtime.Object) (label labels.Set, field fields.Set, err error) {
|
||||||
@ -148,7 +148,7 @@ func TestStoreList(t *testing.T) {
|
|||||||
|
|
||||||
table := map[string]struct {
|
table := map[string]struct {
|
||||||
in *api.PodList
|
in *api.PodList
|
||||||
m *generic.SelectionPredicate
|
m storage.SelectionPredicate
|
||||||
out runtime.Object
|
out runtime.Object
|
||||||
context api.Context
|
context api.Context
|
||||||
}{
|
}{
|
||||||
@ -1141,7 +1141,7 @@ func TestStoreWatch(t *testing.T) {
|
|||||||
noNamespaceContext := api.NewContext()
|
noNamespaceContext := api.NewContext()
|
||||||
|
|
||||||
table := map[string]struct {
|
table := map[string]struct {
|
||||||
selectPred *generic.SelectionPredicate
|
selectPred storage.SelectionPredicate
|
||||||
context api.Context
|
context api.Context
|
||||||
}{
|
}{
|
||||||
"single": {
|
"single": {
|
||||||
@ -1242,8 +1242,8 @@ func newTestGenericStoreRegistry(t *testing.T, hasCacheEnabled bool) (factory.De
|
|||||||
return path.Join(podPrefix, id), nil
|
return path.Join(podPrefix, id), nil
|
||||||
},
|
},
|
||||||
ObjectNameFunc: func(obj runtime.Object) (string, error) { return obj.(*api.Pod).Name, nil },
|
ObjectNameFunc: func(obj runtime.Object) (string, error) { return obj.(*api.Pod).Name, nil },
|
||||||
PredicateFunc: func(label labels.Selector, field fields.Selector) *generic.SelectionPredicate {
|
PredicateFunc: func(label labels.Selector, field fields.Selector) storage.SelectionPredicate {
|
||||||
return &generic.SelectionPredicate{
|
return storage.SelectionPredicate{
|
||||||
Label: label,
|
Label: label,
|
||||||
Field: field,
|
Field: field,
|
||||||
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
||||||
|
@ -27,6 +27,7 @@ import (
|
|||||||
"k8s.io/kubernetes/pkg/labels"
|
"k8s.io/kubernetes/pkg/labels"
|
||||||
"k8s.io/kubernetes/pkg/registry/generic"
|
"k8s.io/kubernetes/pkg/registry/generic"
|
||||||
"k8s.io/kubernetes/pkg/runtime"
|
"k8s.io/kubernetes/pkg/runtime"
|
||||||
|
"k8s.io/kubernetes/pkg/storage"
|
||||||
"k8s.io/kubernetes/pkg/util/validation/field"
|
"k8s.io/kubernetes/pkg/util/validation/field"
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -102,8 +103,8 @@ func PodDisruptionBudgetToSelectableFields(podDisruptionBudget *policy.PodDisrup
|
|||||||
|
|
||||||
// MatchPodDisruptionBudget is the filter used by the generic etcd backend to watch events
|
// MatchPodDisruptionBudget is the filter used by the generic etcd backend to watch events
|
||||||
// from etcd to clients of the apiserver only interested in specific labels/fields.
|
// from etcd to clients of the apiserver only interested in specific labels/fields.
|
||||||
func MatchPodDisruptionBudget(label labels.Selector, field fields.Selector) *generic.SelectionPredicate {
|
func MatchPodDisruptionBudget(label labels.Selector, field fields.Selector) storage.SelectionPredicate {
|
||||||
return &generic.SelectionPredicate{
|
return storage.SelectionPredicate{
|
||||||
Label: label,
|
Label: label,
|
||||||
Field: field,
|
Field: field,
|
||||||
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
||||||
|
@ -25,8 +25,8 @@ import (
|
|||||||
"k8s.io/kubernetes/pkg/apis/rbac/validation"
|
"k8s.io/kubernetes/pkg/apis/rbac/validation"
|
||||||
"k8s.io/kubernetes/pkg/fields"
|
"k8s.io/kubernetes/pkg/fields"
|
||||||
"k8s.io/kubernetes/pkg/labels"
|
"k8s.io/kubernetes/pkg/labels"
|
||||||
"k8s.io/kubernetes/pkg/registry/generic"
|
|
||||||
"k8s.io/kubernetes/pkg/runtime"
|
"k8s.io/kubernetes/pkg/runtime"
|
||||||
|
apistorage "k8s.io/kubernetes/pkg/storage"
|
||||||
"k8s.io/kubernetes/pkg/util/validation/field"
|
"k8s.io/kubernetes/pkg/util/validation/field"
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -102,8 +102,8 @@ func (s strategy) Export(ctx api.Context, obj runtime.Object, exact bool) error
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Matcher returns a generic matcher for a given label and field selector.
|
// Matcher returns a generic matcher for a given label and field selector.
|
||||||
func Matcher(label labels.Selector, field fields.Selector) *generic.SelectionPredicate {
|
func Matcher(label labels.Selector, field fields.Selector) apistorage.SelectionPredicate {
|
||||||
return &generic.SelectionPredicate{
|
return apistorage.SelectionPredicate{
|
||||||
Label: label,
|
Label: label,
|
||||||
Field: field,
|
Field: field,
|
||||||
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
||||||
|
@ -25,8 +25,8 @@ import (
|
|||||||
"k8s.io/kubernetes/pkg/apis/rbac/validation"
|
"k8s.io/kubernetes/pkg/apis/rbac/validation"
|
||||||
"k8s.io/kubernetes/pkg/fields"
|
"k8s.io/kubernetes/pkg/fields"
|
||||||
"k8s.io/kubernetes/pkg/labels"
|
"k8s.io/kubernetes/pkg/labels"
|
||||||
"k8s.io/kubernetes/pkg/registry/generic"
|
|
||||||
"k8s.io/kubernetes/pkg/runtime"
|
"k8s.io/kubernetes/pkg/runtime"
|
||||||
|
apistorage "k8s.io/kubernetes/pkg/storage"
|
||||||
"k8s.io/kubernetes/pkg/util/validation/field"
|
"k8s.io/kubernetes/pkg/util/validation/field"
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -102,8 +102,8 @@ func (s strategy) Export(ctx api.Context, obj runtime.Object, exact bool) error
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Matcher returns a generic matcher for a given label and field selector.
|
// Matcher returns a generic matcher for a given label and field selector.
|
||||||
func Matcher(label labels.Selector, field fields.Selector) *generic.SelectionPredicate {
|
func Matcher(label labels.Selector, field fields.Selector) apistorage.SelectionPredicate {
|
||||||
return &generic.SelectionPredicate{
|
return apistorage.SelectionPredicate{
|
||||||
Label: label,
|
Label: label,
|
||||||
Field: field,
|
Field: field,
|
||||||
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
||||||
|
@ -25,8 +25,8 @@ import (
|
|||||||
"k8s.io/kubernetes/pkg/apis/rbac/validation"
|
"k8s.io/kubernetes/pkg/apis/rbac/validation"
|
||||||
"k8s.io/kubernetes/pkg/fields"
|
"k8s.io/kubernetes/pkg/fields"
|
||||||
"k8s.io/kubernetes/pkg/labels"
|
"k8s.io/kubernetes/pkg/labels"
|
||||||
"k8s.io/kubernetes/pkg/registry/generic"
|
|
||||||
"k8s.io/kubernetes/pkg/runtime"
|
"k8s.io/kubernetes/pkg/runtime"
|
||||||
|
apistorage "k8s.io/kubernetes/pkg/storage"
|
||||||
"k8s.io/kubernetes/pkg/util/validation/field"
|
"k8s.io/kubernetes/pkg/util/validation/field"
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -102,8 +102,8 @@ func (s strategy) Export(ctx api.Context, obj runtime.Object, exact bool) error
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Matcher returns a generic matcher for a given label and field selector.
|
// Matcher returns a generic matcher for a given label and field selector.
|
||||||
func Matcher(label labels.Selector, field fields.Selector) *generic.SelectionPredicate {
|
func Matcher(label labels.Selector, field fields.Selector) apistorage.SelectionPredicate {
|
||||||
return &generic.SelectionPredicate{
|
return apistorage.SelectionPredicate{
|
||||||
Label: label,
|
Label: label,
|
||||||
Field: field,
|
Field: field,
|
||||||
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
||||||
|
@ -25,8 +25,8 @@ import (
|
|||||||
"k8s.io/kubernetes/pkg/apis/rbac/validation"
|
"k8s.io/kubernetes/pkg/apis/rbac/validation"
|
||||||
"k8s.io/kubernetes/pkg/fields"
|
"k8s.io/kubernetes/pkg/fields"
|
||||||
"k8s.io/kubernetes/pkg/labels"
|
"k8s.io/kubernetes/pkg/labels"
|
||||||
"k8s.io/kubernetes/pkg/registry/generic"
|
|
||||||
"k8s.io/kubernetes/pkg/runtime"
|
"k8s.io/kubernetes/pkg/runtime"
|
||||||
|
apistorage "k8s.io/kubernetes/pkg/storage"
|
||||||
"k8s.io/kubernetes/pkg/util/validation/field"
|
"k8s.io/kubernetes/pkg/util/validation/field"
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -102,8 +102,8 @@ func (s strategy) Export(ctx api.Context, obj runtime.Object, exact bool) error
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Matcher returns a generic matcher for a given label and field selector.
|
// Matcher returns a generic matcher for a given label and field selector.
|
||||||
func Matcher(label labels.Selector, field fields.Selector) *generic.SelectionPredicate {
|
func Matcher(label labels.Selector, field fields.Selector) apistorage.SelectionPredicate {
|
||||||
return &generic.SelectionPredicate{
|
return apistorage.SelectionPredicate{
|
||||||
Label: label,
|
Label: label,
|
||||||
Field: field,
|
Field: field,
|
||||||
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
||||||
|
@ -26,6 +26,7 @@ import (
|
|||||||
"k8s.io/kubernetes/pkg/labels"
|
"k8s.io/kubernetes/pkg/labels"
|
||||||
"k8s.io/kubernetes/pkg/registry/generic"
|
"k8s.io/kubernetes/pkg/registry/generic"
|
||||||
"k8s.io/kubernetes/pkg/runtime"
|
"k8s.io/kubernetes/pkg/runtime"
|
||||||
|
apistorage "k8s.io/kubernetes/pkg/storage"
|
||||||
"k8s.io/kubernetes/pkg/util/validation/field"
|
"k8s.io/kubernetes/pkg/util/validation/field"
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -77,8 +78,8 @@ func (storageClassStrategy) AllowUnconditionalUpdate() bool {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// MatchStorageClass returns a generic matcher for a given label and field selector.
|
// MatchStorageClass returns a generic matcher for a given label and field selector.
|
||||||
func MatchStorageClasses(label labels.Selector, field fields.Selector) *generic.SelectionPredicate {
|
func MatchStorageClasses(label labels.Selector, field fields.Selector) apistorage.SelectionPredicate {
|
||||||
return &generic.SelectionPredicate{
|
return apistorage.SelectionPredicate{
|
||||||
Label: label,
|
Label: label,
|
||||||
Field: field,
|
Field: field,
|
||||||
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
||||||
|
@ -278,7 +278,7 @@ func (c *Cacher) Delete(ctx context.Context, key string, out runtime.Object, pre
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Implements storage.Interface.
|
// Implements storage.Interface.
|
||||||
func (c *Cacher) Watch(ctx context.Context, key string, resourceVersion string, filter Filter) (watch.Interface, error) {
|
func (c *Cacher) Watch(ctx context.Context, key string, resourceVersion string, pred SelectionPredicate) (watch.Interface, error) {
|
||||||
watchRV, err := ParseWatchResourceVersion(resourceVersion)
|
watchRV, err := ParseWatchResourceVersion(resourceVersion)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
@ -302,17 +302,17 @@ func (c *Cacher) Watch(ctx context.Context, key string, resourceVersion string,
|
|||||||
}
|
}
|
||||||
|
|
||||||
triggerValue, triggerSupported := "", false
|
triggerValue, triggerSupported := "", false
|
||||||
// TODO: Currently we assume that in a given Cacher object, any <filter> that is
|
// TODO: Currently we assume that in a given Cacher object, any <predicate> that is
|
||||||
// passed here is aware of exactly the same trigger (at most one).
|
// passed here is aware of exactly the same trigger (at most one).
|
||||||
// Thus, either 0 or 1 values will be returned.
|
// Thus, either 0 or 1 values will be returned.
|
||||||
if matchValues := filter.Trigger(); len(matchValues) > 0 {
|
if matchValues := pred.MatcherIndex(); len(matchValues) > 0 {
|
||||||
triggerValue, triggerSupported = matchValues[0].Value, true
|
triggerValue, triggerSupported = matchValues[0].Value, true
|
||||||
}
|
}
|
||||||
|
|
||||||
c.Lock()
|
c.Lock()
|
||||||
defer c.Unlock()
|
defer c.Unlock()
|
||||||
forget := forgetWatcher(c, c.watcherIdx, triggerValue, triggerSupported)
|
forget := forgetWatcher(c, c.watcherIdx, triggerValue, triggerSupported)
|
||||||
watcher := newCacheWatcher(watchRV, initEvents, filterFunction(key, c.keyFunc, filter), forget)
|
watcher := newCacheWatcher(watchRV, initEvents, filterFunction(key, c.keyFunc, pred), forget)
|
||||||
|
|
||||||
c.watchers.addWatcher(watcher, c.watcherIdx, triggerValue, triggerSupported)
|
c.watchers.addWatcher(watcher, c.watcherIdx, triggerValue, triggerSupported)
|
||||||
c.watcherIdx++
|
c.watcherIdx++
|
||||||
@ -320,8 +320,8 @@ func (c *Cacher) Watch(ctx context.Context, key string, resourceVersion string,
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Implements storage.Interface.
|
// Implements storage.Interface.
|
||||||
func (c *Cacher) WatchList(ctx context.Context, key string, resourceVersion string, filter Filter) (watch.Interface, error) {
|
func (c *Cacher) WatchList(ctx context.Context, key string, resourceVersion string, pred SelectionPredicate) (watch.Interface, error) {
|
||||||
return c.Watch(ctx, key, resourceVersion, filter)
|
return c.Watch(ctx, key, resourceVersion, pred)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Implements storage.Interface.
|
// Implements storage.Interface.
|
||||||
@ -330,16 +330,16 @@ func (c *Cacher) Get(ctx context.Context, key string, objPtr runtime.Object, ign
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Implements storage.Interface.
|
// Implements storage.Interface.
|
||||||
func (c *Cacher) GetToList(ctx context.Context, key string, filter Filter, listObj runtime.Object) error {
|
func (c *Cacher) GetToList(ctx context.Context, key string, pred SelectionPredicate, listObj runtime.Object) error {
|
||||||
return c.storage.GetToList(ctx, key, filter, listObj)
|
return c.storage.GetToList(ctx, key, pred, listObj)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Implements storage.Interface.
|
// Implements storage.Interface.
|
||||||
func (c *Cacher) List(ctx context.Context, key string, resourceVersion string, filter Filter, listObj runtime.Object) error {
|
func (c *Cacher) List(ctx context.Context, key string, resourceVersion string, pred SelectionPredicate, listObj runtime.Object) error {
|
||||||
if resourceVersion == "" {
|
if resourceVersion == "" {
|
||||||
// If resourceVersion is not specified, serve it from underlying
|
// If resourceVersion is not specified, serve it from underlying
|
||||||
// storage (for backward compatibility).
|
// storage (for backward compatibility).
|
||||||
return c.storage.List(ctx, key, resourceVersion, filter, listObj)
|
return c.storage.List(ctx, key, resourceVersion, pred, listObj)
|
||||||
}
|
}
|
||||||
|
|
||||||
// If resourceVersion is specified, serve it from cache.
|
// If resourceVersion is specified, serve it from cache.
|
||||||
@ -362,7 +362,7 @@ func (c *Cacher) List(ctx context.Context, key string, resourceVersion string, f
|
|||||||
if err != nil || listVal.Kind() != reflect.Slice {
|
if err != nil || listVal.Kind() != reflect.Slice {
|
||||||
return fmt.Errorf("need a pointer to slice, got %v", listVal.Kind())
|
return fmt.Errorf("need a pointer to slice, got %v", listVal.Kind())
|
||||||
}
|
}
|
||||||
filterFunc := filterFunction(key, c.keyFunc, filter)
|
filter := filterFunction(key, c.keyFunc, pred)
|
||||||
|
|
||||||
objs, readResourceVersion, err := c.watchCache.WaitUntilFreshAndList(listRV)
|
objs, readResourceVersion, err := c.watchCache.WaitUntilFreshAndList(listRV)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@ -373,7 +373,7 @@ func (c *Cacher) List(ctx context.Context, key string, resourceVersion string, f
|
|||||||
if !ok {
|
if !ok {
|
||||||
return fmt.Errorf("non runtime.Object returned from storage: %v", obj)
|
return fmt.Errorf("non runtime.Object returned from storage: %v", obj)
|
||||||
}
|
}
|
||||||
if filterFunc.Filter(object) {
|
if filter(object) {
|
||||||
listVal.Set(reflect.Append(listVal, reflect.ValueOf(object).Elem()))
|
listVal.Set(reflect.Append(listVal, reflect.ValueOf(object).Elem()))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@ -502,19 +502,20 @@ func forgetWatcher(c *Cacher, index int, triggerValue string, triggerSupported b
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func filterFunction(key string, keyFunc func(runtime.Object) (string, error), filter Filter) Filter {
|
func filterFunction(key string, keyFunc func(runtime.Object) (string, error), p SelectionPredicate) FilterFunc {
|
||||||
|
f := SimpleFilter(p)
|
||||||
filterFunc := func(obj runtime.Object) bool {
|
filterFunc := func(obj runtime.Object) bool {
|
||||||
objKey, err := keyFunc(obj)
|
objKey, err := keyFunc(obj)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
glog.Errorf("invalid object for filter: %v", obj)
|
glog.Errorf("invalid object for filter. Obj: %v. Err: %v", obj, err)
|
||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
if !hasPathPrefix(objKey, key) {
|
if !hasPathPrefix(objKey, key) {
|
||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
return filter.Filter(obj)
|
return f(obj)
|
||||||
}
|
}
|
||||||
return NewSimpleFilter(filterFunc, filter.Trigger)
|
return filterFunc
|
||||||
}
|
}
|
||||||
|
|
||||||
// Returns resource version to which the underlying cache is synced.
|
// Returns resource version to which the underlying cache is synced.
|
||||||
@ -603,12 +604,12 @@ type cacheWatcher struct {
|
|||||||
sync.Mutex
|
sync.Mutex
|
||||||
input chan watchCacheEvent
|
input chan watchCacheEvent
|
||||||
result chan watch.Event
|
result chan watch.Event
|
||||||
filter Filter
|
filter FilterFunc
|
||||||
stopped bool
|
stopped bool
|
||||||
forget func(bool)
|
forget func(bool)
|
||||||
}
|
}
|
||||||
|
|
||||||
func newCacheWatcher(resourceVersion uint64, initEvents []watchCacheEvent, filter Filter, forget func(bool)) *cacheWatcher {
|
func newCacheWatcher(resourceVersion uint64, initEvents []watchCacheEvent, filter FilterFunc, forget func(bool)) *cacheWatcher {
|
||||||
watcher := &cacheWatcher{
|
watcher := &cacheWatcher{
|
||||||
input: make(chan watchCacheEvent, 10),
|
input: make(chan watchCacheEvent, 10),
|
||||||
result: make(chan watch.Event, 10),
|
result: make(chan watch.Event, 10),
|
||||||
@ -684,10 +685,10 @@ func (c *cacheWatcher) add(event *watchCacheEvent) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (c *cacheWatcher) sendWatchCacheEvent(event watchCacheEvent) {
|
func (c *cacheWatcher) sendWatchCacheEvent(event watchCacheEvent) {
|
||||||
curObjPasses := event.Type != watch.Deleted && c.filter.Filter(event.Object)
|
curObjPasses := event.Type != watch.Deleted && c.filter(event.Object)
|
||||||
oldObjPasses := false
|
oldObjPasses := false
|
||||||
if event.PrevObject != nil {
|
if event.PrevObject != nil {
|
||||||
oldObjPasses = c.filter.Filter(event.PrevObject)
|
oldObjPasses = c.filter(event.PrevObject)
|
||||||
}
|
}
|
||||||
if !curObjPasses && !oldObjPasses {
|
if !curObjPasses && !oldObjPasses {
|
||||||
// Watcher is not interested in that object.
|
// Watcher is not interested in that object.
|
||||||
|
@ -30,6 +30,7 @@ import (
|
|||||||
"k8s.io/kubernetes/pkg/api/testapi"
|
"k8s.io/kubernetes/pkg/api/testapi"
|
||||||
apitesting "k8s.io/kubernetes/pkg/api/testing"
|
apitesting "k8s.io/kubernetes/pkg/api/testing"
|
||||||
"k8s.io/kubernetes/pkg/api/unversioned"
|
"k8s.io/kubernetes/pkg/api/unversioned"
|
||||||
|
"k8s.io/kubernetes/pkg/fields"
|
||||||
"k8s.io/kubernetes/pkg/labels"
|
"k8s.io/kubernetes/pkg/labels"
|
||||||
"k8s.io/kubernetes/pkg/runtime"
|
"k8s.io/kubernetes/pkg/runtime"
|
||||||
"k8s.io/kubernetes/pkg/storage"
|
"k8s.io/kubernetes/pkg/storage"
|
||||||
@ -196,12 +197,12 @@ type injectListError struct {
|
|||||||
storage.Interface
|
storage.Interface
|
||||||
}
|
}
|
||||||
|
|
||||||
func (self *injectListError) List(ctx context.Context, key string, resourceVersion string, filter storage.Filter, listObj runtime.Object) error {
|
func (self *injectListError) List(ctx context.Context, key string, resourceVersion string, p storage.SelectionPredicate, listObj runtime.Object) error {
|
||||||
if self.errors > 0 {
|
if self.errors > 0 {
|
||||||
self.errors--
|
self.errors--
|
||||||
return fmt.Errorf("injected error")
|
return fmt.Errorf("injected error")
|
||||||
}
|
}
|
||||||
return self.Interface.List(ctx, key, resourceVersion, filter, listObj)
|
return self.Interface.List(ctx, key, resourceVersion, p, listObj)
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestWatch(t *testing.T) {
|
func TestWatch(t *testing.T) {
|
||||||
@ -355,17 +356,18 @@ func TestFiltering(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Set up Watch for object "podFoo" with label filter set.
|
// Set up Watch for object "podFoo" with label filter set.
|
||||||
selector := labels.SelectorFromSet(labels.Set{"filter": "foo"})
|
pred := storage.SelectionPredicate{
|
||||||
filterFunc := func(obj runtime.Object) bool {
|
Label: labels.SelectorFromSet(labels.Set{"filter": "foo"}),
|
||||||
|
Field: fields.Everything(),
|
||||||
|
GetAttrs: func(obj runtime.Object) (label labels.Set, field fields.Set, err error) {
|
||||||
metadata, err := meta.Accessor(obj)
|
metadata, err := meta.Accessor(obj)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Errorf("Unexpected error: %v", err)
|
t.Fatalf("Unexpected error: %v", err)
|
||||||
return false
|
|
||||||
}
|
}
|
||||||
return selector.Matches(labels.Set(metadata.GetLabels()))
|
return labels.Set(metadata.GetLabels()), nil, nil
|
||||||
|
},
|
||||||
}
|
}
|
||||||
filter := storage.NewSimpleFilter(filterFunc, storage.NoTriggerFunc)
|
watcher, err := cacher.Watch(context.TODO(), "pods/ns/foo", fooCreated.ResourceVersion, pred)
|
||||||
watcher, err := cacher.Watch(context.TODO(), "pods/ns/foo", fooCreated.ResourceVersion, filter)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("Unexpected error: %v", err)
|
t.Fatalf("Unexpected error: %v", err)
|
||||||
}
|
}
|
||||||
|
@ -202,7 +202,7 @@ func (h *etcdHelper) Delete(ctx context.Context, key string, out runtime.Object,
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Implements storage.Interface.
|
// Implements storage.Interface.
|
||||||
func (h *etcdHelper) Watch(ctx context.Context, key string, resourceVersion string, filter storage.Filter) (watch.Interface, error) {
|
func (h *etcdHelper) Watch(ctx context.Context, key string, resourceVersion string, pred storage.SelectionPredicate) (watch.Interface, error) {
|
||||||
if ctx == nil {
|
if ctx == nil {
|
||||||
glog.Errorf("Context is nil")
|
glog.Errorf("Context is nil")
|
||||||
}
|
}
|
||||||
@ -211,13 +211,13 @@ func (h *etcdHelper) Watch(ctx context.Context, key string, resourceVersion stri
|
|||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
key = h.prefixEtcdKey(key)
|
key = h.prefixEtcdKey(key)
|
||||||
w := newEtcdWatcher(false, h.quorum, nil, filter, h.codec, h.versioner, nil, h)
|
w := newEtcdWatcher(false, h.quorum, nil, storage.SimpleFilter(pred), h.codec, h.versioner, nil, h)
|
||||||
go w.etcdWatch(ctx, h.etcdKeysAPI, key, watchRV)
|
go w.etcdWatch(ctx, h.etcdKeysAPI, key, watchRV)
|
||||||
return w, nil
|
return w, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// Implements storage.Interface.
|
// Implements storage.Interface.
|
||||||
func (h *etcdHelper) WatchList(ctx context.Context, key string, resourceVersion string, filter storage.Filter) (watch.Interface, error) {
|
func (h *etcdHelper) WatchList(ctx context.Context, key string, resourceVersion string, pred storage.SelectionPredicate) (watch.Interface, error) {
|
||||||
if ctx == nil {
|
if ctx == nil {
|
||||||
glog.Errorf("Context is nil")
|
glog.Errorf("Context is nil")
|
||||||
}
|
}
|
||||||
@ -226,7 +226,7 @@ func (h *etcdHelper) WatchList(ctx context.Context, key string, resourceVersion
|
|||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
key = h.prefixEtcdKey(key)
|
key = h.prefixEtcdKey(key)
|
||||||
w := newEtcdWatcher(true, h.quorum, exceptKey(key), filter, h.codec, h.versioner, nil, h)
|
w := newEtcdWatcher(true, h.quorum, exceptKey(key), storage.SimpleFilter(pred), h.codec, h.versioner, nil, h)
|
||||||
go w.etcdWatch(ctx, h.etcdKeysAPI, key, watchRV)
|
go w.etcdWatch(ctx, h.etcdKeysAPI, key, watchRV)
|
||||||
return w, nil
|
return w, nil
|
||||||
}
|
}
|
||||||
@ -297,7 +297,7 @@ func (h *etcdHelper) extractObj(response *etcd.Response, inErr error, objPtr run
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Implements storage.Interface.
|
// Implements storage.Interface.
|
||||||
func (h *etcdHelper) GetToList(ctx context.Context, key string, filter storage.Filter, listObj runtime.Object) error {
|
func (h *etcdHelper) GetToList(ctx context.Context, key string, pred storage.SelectionPredicate, listObj runtime.Object) error {
|
||||||
if ctx == nil {
|
if ctx == nil {
|
||||||
glog.Errorf("Context is nil")
|
glog.Errorf("Context is nil")
|
||||||
}
|
}
|
||||||
@ -326,7 +326,7 @@ func (h *etcdHelper) GetToList(ctx context.Context, key string, filter storage.F
|
|||||||
nodes := make([]*etcd.Node, 0)
|
nodes := make([]*etcd.Node, 0)
|
||||||
nodes = append(nodes, response.Node)
|
nodes = append(nodes, response.Node)
|
||||||
|
|
||||||
if err := h.decodeNodeList(nodes, filter, listPtr); err != nil {
|
if err := h.decodeNodeList(nodes, storage.SimpleFilter(pred), listPtr); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
trace.Step("Object decoded")
|
trace.Step("Object decoded")
|
||||||
@ -337,7 +337,7 @@ func (h *etcdHelper) GetToList(ctx context.Context, key string, filter storage.F
|
|||||||
}
|
}
|
||||||
|
|
||||||
// decodeNodeList walks the tree of each node in the list and decodes into the specified object
|
// decodeNodeList walks the tree of each node in the list and decodes into the specified object
|
||||||
func (h *etcdHelper) decodeNodeList(nodes []*etcd.Node, filter storage.Filter, slicePtr interface{}) error {
|
func (h *etcdHelper) decodeNodeList(nodes []*etcd.Node, filter storage.FilterFunc, slicePtr interface{}) error {
|
||||||
trace := util.NewTrace("decodeNodeList " + getTypeName(slicePtr))
|
trace := util.NewTrace("decodeNodeList " + getTypeName(slicePtr))
|
||||||
defer trace.LogIfLong(400 * time.Millisecond)
|
defer trace.LogIfLong(400 * time.Millisecond)
|
||||||
v, err := conversion.EnforcePtr(slicePtr)
|
v, err := conversion.EnforcePtr(slicePtr)
|
||||||
@ -366,7 +366,7 @@ func (h *etcdHelper) decodeNodeList(nodes []*etcd.Node, filter storage.Filter, s
|
|||||||
}
|
}
|
||||||
// being unable to set the version does not prevent the object from being extracted
|
// being unable to set the version does not prevent the object from being extracted
|
||||||
_ = h.versioner.UpdateObject(obj, node.ModifiedIndex)
|
_ = h.versioner.UpdateObject(obj, node.ModifiedIndex)
|
||||||
if filter.Filter(obj) {
|
if filter(obj) {
|
||||||
v.Set(reflect.Append(v, reflect.ValueOf(obj).Elem()))
|
v.Set(reflect.Append(v, reflect.ValueOf(obj).Elem()))
|
||||||
}
|
}
|
||||||
if node.ModifiedIndex != 0 {
|
if node.ModifiedIndex != 0 {
|
||||||
@ -379,7 +379,7 @@ func (h *etcdHelper) decodeNodeList(nodes []*etcd.Node, filter storage.Filter, s
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Implements storage.Interface.
|
// Implements storage.Interface.
|
||||||
func (h *etcdHelper) List(ctx context.Context, key string, resourceVersion string, filter storage.Filter, listObj runtime.Object) error {
|
func (h *etcdHelper) List(ctx context.Context, key string, resourceVersion string, pred storage.SelectionPredicate, listObj runtime.Object) error {
|
||||||
if ctx == nil {
|
if ctx == nil {
|
||||||
glog.Errorf("Context is nil")
|
glog.Errorf("Context is nil")
|
||||||
}
|
}
|
||||||
@ -398,7 +398,7 @@ func (h *etcdHelper) List(ctx context.Context, key string, resourceVersion strin
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
if err := h.decodeNodeList(nodes, filter, listPtr); err != nil {
|
if err := h.decodeNodeList(nodes, storage.SimpleFilter(pred), listPtr); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
trace.Step("Node list decoded")
|
trace.Step("Node list decoded")
|
||||||
@ -548,7 +548,7 @@ func (h *etcdHelper) prefixEtcdKey(key string) string {
|
|||||||
// their Node.ModifiedIndex, which is unique across all types.
|
// their Node.ModifiedIndex, which is unique across all types.
|
||||||
// All implementations must be thread-safe.
|
// All implementations must be thread-safe.
|
||||||
type etcdCache interface {
|
type etcdCache interface {
|
||||||
getFromCache(index uint64, filter storage.Filter) (runtime.Object, bool)
|
getFromCache(index uint64, filter storage.FilterFunc) (runtime.Object, bool)
|
||||||
addToCache(index uint64, obj runtime.Object)
|
addToCache(index uint64, obj runtime.Object)
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -556,14 +556,14 @@ func getTypeName(obj interface{}) string {
|
|||||||
return reflect.TypeOf(obj).String()
|
return reflect.TypeOf(obj).String()
|
||||||
}
|
}
|
||||||
|
|
||||||
func (h *etcdHelper) getFromCache(index uint64, filter storage.Filter) (runtime.Object, bool) {
|
func (h *etcdHelper) getFromCache(index uint64, filter storage.FilterFunc) (runtime.Object, bool) {
|
||||||
startTime := time.Now()
|
startTime := time.Now()
|
||||||
defer func() {
|
defer func() {
|
||||||
metrics.ObserveGetCache(startTime)
|
metrics.ObserveGetCache(startTime)
|
||||||
}()
|
}()
|
||||||
obj, found := h.cache.Get(index)
|
obj, found := h.cache.Get(index)
|
||||||
if found {
|
if found {
|
||||||
if !filter.Filter(obj.(runtime.Object)) {
|
if !filter(obj.(runtime.Object)) {
|
||||||
return nil, true
|
return nil, true
|
||||||
}
|
}
|
||||||
// We should not return the object itself to avoid polluting the cache if someone
|
// We should not return the object itself to avoid polluting the cache if someone
|
||||||
|
@ -30,6 +30,8 @@ import (
|
|||||||
"k8s.io/kubernetes/pkg/api/testapi"
|
"k8s.io/kubernetes/pkg/api/testapi"
|
||||||
apitesting "k8s.io/kubernetes/pkg/api/testing"
|
apitesting "k8s.io/kubernetes/pkg/api/testing"
|
||||||
"k8s.io/kubernetes/pkg/conversion"
|
"k8s.io/kubernetes/pkg/conversion"
|
||||||
|
"k8s.io/kubernetes/pkg/fields"
|
||||||
|
"k8s.io/kubernetes/pkg/labels"
|
||||||
"k8s.io/kubernetes/pkg/runtime"
|
"k8s.io/kubernetes/pkg/runtime"
|
||||||
"k8s.io/kubernetes/pkg/runtime/serializer"
|
"k8s.io/kubernetes/pkg/runtime/serializer"
|
||||||
"k8s.io/kubernetes/pkg/storage"
|
"k8s.io/kubernetes/pkg/storage"
|
||||||
@ -154,14 +156,17 @@ func TestListFiltered(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
createPodList(t, helper, &list)
|
createPodList(t, helper, &list)
|
||||||
filterFunc := func(obj runtime.Object) bool {
|
// List only "bar" pod
|
||||||
|
p := storage.SelectionPredicate{
|
||||||
|
Label: labels.Everything(),
|
||||||
|
Field: fields.SelectorFromSet(fields.Set{"metadata.name": "bar"}),
|
||||||
|
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
||||||
pod := obj.(*api.Pod)
|
pod := obj.(*api.Pod)
|
||||||
return pod.Name == "bar"
|
return labels.Set(pod.Labels), fields.Set{"metadata.name": pod.Name}, nil
|
||||||
|
},
|
||||||
}
|
}
|
||||||
filter := storage.NewSimpleFilter(filterFunc, storage.NoTriggerFunc)
|
|
||||||
|
|
||||||
var got api.PodList
|
var got api.PodList
|
||||||
err := helper.List(context.TODO(), key, "", filter, &got)
|
err := helper.List(context.TODO(), key, "", p, &got)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Errorf("Unexpected error %v", err)
|
t.Errorf("Unexpected error %v", err)
|
||||||
}
|
}
|
||||||
|
@ -77,7 +77,7 @@ type etcdWatcher struct {
|
|||||||
list bool // If we're doing a recursive watch, should be true.
|
list bool // If we're doing a recursive watch, should be true.
|
||||||
quorum bool // If we enable quorum, shoule be true
|
quorum bool // If we enable quorum, shoule be true
|
||||||
include includeFunc
|
include includeFunc
|
||||||
filter storage.Filter
|
filter storage.FilterFunc
|
||||||
|
|
||||||
etcdIncoming chan *etcd.Response
|
etcdIncoming chan *etcd.Response
|
||||||
etcdError chan error
|
etcdError chan error
|
||||||
@ -105,7 +105,7 @@ const watchWaitDuration = 100 * time.Millisecond
|
|||||||
// newEtcdWatcher returns a new etcdWatcher; if list is true, watch sub-nodes.
|
// newEtcdWatcher returns a new etcdWatcher; if list is true, watch sub-nodes.
|
||||||
// The versioner must be able to handle the objects that transform creates.
|
// The versioner must be able to handle the objects that transform creates.
|
||||||
func newEtcdWatcher(
|
func newEtcdWatcher(
|
||||||
list bool, quorum bool, include includeFunc, filter storage.Filter,
|
list bool, quorum bool, include includeFunc, filter storage.FilterFunc,
|
||||||
encoding runtime.Codec, versioner storage.Versioner, transform TransformFunc,
|
encoding runtime.Codec, versioner storage.Versioner, transform TransformFunc,
|
||||||
cache etcdCache) *etcdWatcher {
|
cache etcdCache) *etcdWatcher {
|
||||||
w := &etcdWatcher{
|
w := &etcdWatcher{
|
||||||
@ -310,7 +310,7 @@ func (w *etcdWatcher) translate() {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (w *etcdWatcher) decodeObject(node *etcd.Node) (runtime.Object, error) {
|
func (w *etcdWatcher) decodeObject(node *etcd.Node) (runtime.Object, error) {
|
||||||
if obj, found := w.cache.getFromCache(node.ModifiedIndex, storage.Everything); found {
|
if obj, found := w.cache.getFromCache(node.ModifiedIndex, storage.SimpleFilter(storage.Everything)); found {
|
||||||
return obj, nil
|
return obj, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -355,7 +355,7 @@ func (w *etcdWatcher) sendAdd(res *etcd.Response) {
|
|||||||
// the resourceVersion to resume will never be able to get past a bad value.
|
// the resourceVersion to resume will never be able to get past a bad value.
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
if !w.filter.Filter(obj) {
|
if !w.filter(obj) {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
action := watch.Added
|
action := watch.Added
|
||||||
@ -384,7 +384,7 @@ func (w *etcdWatcher) sendModify(res *etcd.Response) {
|
|||||||
// the resourceVersion to resume will never be able to get past a bad value.
|
// the resourceVersion to resume will never be able to get past a bad value.
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
curObjPasses := w.filter.Filter(curObj)
|
curObjPasses := w.filter(curObj)
|
||||||
oldObjPasses := false
|
oldObjPasses := false
|
||||||
var oldObj runtime.Object
|
var oldObj runtime.Object
|
||||||
if res.PrevNode != nil && res.PrevNode.Value != "" {
|
if res.PrevNode != nil && res.PrevNode.Value != "" {
|
||||||
@ -393,7 +393,7 @@ func (w *etcdWatcher) sendModify(res *etcd.Response) {
|
|||||||
if err := w.versioner.UpdateObject(oldObj, res.Node.ModifiedIndex); err != nil {
|
if err := w.versioner.UpdateObject(oldObj, res.Node.ModifiedIndex); err != nil {
|
||||||
utilruntime.HandleError(fmt.Errorf("failure to version api object (%d) %#v: %v", res.Node.ModifiedIndex, oldObj, err))
|
utilruntime.HandleError(fmt.Errorf("failure to version api object (%d) %#v: %v", res.Node.ModifiedIndex, oldObj, err))
|
||||||
}
|
}
|
||||||
oldObjPasses = w.filter.Filter(oldObj)
|
oldObjPasses = w.filter(oldObj)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
// Some changes to an object may cause it to start or stop matching a filter.
|
// Some changes to an object may cause it to start or stop matching a filter.
|
||||||
@ -442,7 +442,7 @@ func (w *etcdWatcher) sendDelete(res *etcd.Response) {
|
|||||||
// the resourceVersion to resume will never be able to get past a bad value.
|
// the resourceVersion to resume will never be able to get past a bad value.
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
if !w.filter.Filter(obj) {
|
if !w.filter(obj) {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
w.emit(watch.Event{
|
w.emit(watch.Event{
|
||||||
|
@ -37,7 +37,7 @@ var versioner = APIObjectVersioner{}
|
|||||||
// Implements etcdCache interface as empty methods (i.e. does not cache any objects)
|
// Implements etcdCache interface as empty methods (i.e. does not cache any objects)
|
||||||
type fakeEtcdCache struct{}
|
type fakeEtcdCache struct{}
|
||||||
|
|
||||||
func (f *fakeEtcdCache) getFromCache(index uint64, filter storage.Filter) (runtime.Object, bool) {
|
func (f *fakeEtcdCache) getFromCache(index uint64, filter storage.FilterFunc) (runtime.Object, bool) {
|
||||||
return nil, false
|
return nil, false
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -46,18 +46,6 @@ func (f *fakeEtcdCache) addToCache(index uint64, obj runtime.Object) {
|
|||||||
|
|
||||||
var _ etcdCache = &fakeEtcdCache{}
|
var _ etcdCache = &fakeEtcdCache{}
|
||||||
|
|
||||||
// firstLetterIsB implements storage.Filter interface.
|
|
||||||
type firstLetterIsB struct {
|
|
||||||
}
|
|
||||||
|
|
||||||
func (f *firstLetterIsB) Filter(obj runtime.Object) bool {
|
|
||||||
return obj.(*api.Pod).Name[0] == 'b'
|
|
||||||
}
|
|
||||||
|
|
||||||
func (f *firstLetterIsB) Trigger() []storage.MatchValue {
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestWatchInterpretations(t *testing.T) {
|
func TestWatchInterpretations(t *testing.T) {
|
||||||
codec := testapi.Default.Codec()
|
codec := testapi.Default.Codec()
|
||||||
// Declare some pods to make the test cases compact.
|
// Declare some pods to make the test cases compact.
|
||||||
@ -135,10 +123,12 @@ func TestWatchInterpretations(t *testing.T) {
|
|||||||
expectEmit: false,
|
expectEmit: false,
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
|
firstLetterIsB := func(obj runtime.Object) bool {
|
||||||
|
return obj.(*api.Pod).Name[0] == 'b'
|
||||||
|
}
|
||||||
for name, item := range table {
|
for name, item := range table {
|
||||||
for _, action := range item.actions {
|
for _, action := range item.actions {
|
||||||
w := newEtcdWatcher(true, false, nil, &firstLetterIsB{}, codec, versioner, nil, &fakeEtcdCache{})
|
w := newEtcdWatcher(true, false, nil, firstLetterIsB, codec, versioner, nil, &fakeEtcdCache{})
|
||||||
emitCalled := false
|
emitCalled := false
|
||||||
w.emit = func(event watch.Event) {
|
w.emit = func(event watch.Event) {
|
||||||
emitCalled = true
|
emitCalled = true
|
||||||
@ -177,7 +167,7 @@ func TestWatchInterpretations(t *testing.T) {
|
|||||||
|
|
||||||
func TestWatchInterpretation_ResponseNotSet(t *testing.T) {
|
func TestWatchInterpretation_ResponseNotSet(t *testing.T) {
|
||||||
_, codec := testScheme(t)
|
_, codec := testScheme(t)
|
||||||
w := newEtcdWatcher(false, false, nil, storage.Everything, codec, versioner, nil, &fakeEtcdCache{})
|
w := newEtcdWatcher(false, false, nil, storage.SimpleFilter(storage.Everything), codec, versioner, nil, &fakeEtcdCache{})
|
||||||
w.emit = func(e watch.Event) {
|
w.emit = func(e watch.Event) {
|
||||||
t.Errorf("Unexpected emit: %v", e)
|
t.Errorf("Unexpected emit: %v", e)
|
||||||
}
|
}
|
||||||
@ -192,7 +182,7 @@ func TestWatchInterpretation_ResponseNoNode(t *testing.T) {
|
|||||||
_, codec := testScheme(t)
|
_, codec := testScheme(t)
|
||||||
actions := []string{"create", "set", "compareAndSwap", "delete"}
|
actions := []string{"create", "set", "compareAndSwap", "delete"}
|
||||||
for _, action := range actions {
|
for _, action := range actions {
|
||||||
w := newEtcdWatcher(false, false, nil, storage.Everything, codec, versioner, nil, &fakeEtcdCache{})
|
w := newEtcdWatcher(false, false, nil, storage.SimpleFilter(storage.Everything), codec, versioner, nil, &fakeEtcdCache{})
|
||||||
w.emit = func(e watch.Event) {
|
w.emit = func(e watch.Event) {
|
||||||
t.Errorf("Unexpected emit: %v", e)
|
t.Errorf("Unexpected emit: %v", e)
|
||||||
}
|
}
|
||||||
@ -207,7 +197,7 @@ func TestWatchInterpretation_ResponseBadData(t *testing.T) {
|
|||||||
_, codec := testScheme(t)
|
_, codec := testScheme(t)
|
||||||
actions := []string{"create", "set", "compareAndSwap", "delete"}
|
actions := []string{"create", "set", "compareAndSwap", "delete"}
|
||||||
for _, action := range actions {
|
for _, action := range actions {
|
||||||
w := newEtcdWatcher(false, false, nil, storage.Everything, codec, versioner, nil, &fakeEtcdCache{})
|
w := newEtcdWatcher(false, false, nil, storage.SimpleFilter(storage.Everything), codec, versioner, nil, &fakeEtcdCache{})
|
||||||
w.emit = func(e watch.Event) {
|
w.emit = func(e watch.Event) {
|
||||||
t.Errorf("Unexpected emit: %v", e)
|
t.Errorf("Unexpected emit: %v", e)
|
||||||
}
|
}
|
||||||
@ -229,10 +219,9 @@ func TestWatchInterpretation_ResponseBadData(t *testing.T) {
|
|||||||
|
|
||||||
func TestSendResultDeleteEventHaveLatestIndex(t *testing.T) {
|
func TestSendResultDeleteEventHaveLatestIndex(t *testing.T) {
|
||||||
codec := testapi.Default.Codec()
|
codec := testapi.Default.Codec()
|
||||||
filterFunc := func(obj runtime.Object) bool {
|
filter := func(obj runtime.Object) bool {
|
||||||
return obj.(*api.Pod).Name != "bar"
|
return obj.(*api.Pod).Name != "bar"
|
||||||
}
|
}
|
||||||
filter := storage.NewSimpleFilter(filterFunc, storage.NoTriggerFunc)
|
|
||||||
w := newEtcdWatcher(false, false, nil, filter, codec, versioner, nil, &fakeEtcdCache{})
|
w := newEtcdWatcher(false, false, nil, filter, codec, versioner, nil, &fakeEtcdCache{})
|
||||||
|
|
||||||
eventChan := make(chan watch.Event, 1)
|
eventChan := make(chan watch.Event, 1)
|
||||||
|
@ -255,7 +255,7 @@ func (s *store) GuaranteedUpdate(ctx context.Context, key string, out runtime.Ob
|
|||||||
}
|
}
|
||||||
|
|
||||||
// GetToList implements storage.Interface.GetToList.
|
// GetToList implements storage.Interface.GetToList.
|
||||||
func (s *store) GetToList(ctx context.Context, key string, filter storage.Filter, listObj runtime.Object) error {
|
func (s *store) GetToList(ctx context.Context, key string, pred storage.SelectionPredicate, listObj runtime.Object) error {
|
||||||
listPtr, err := meta.GetItemsPtr(listObj)
|
listPtr, err := meta.GetItemsPtr(listObj)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
@ -273,7 +273,7 @@ func (s *store) GetToList(ctx context.Context, key string, filter storage.Filter
|
|||||||
data: getResp.Kvs[0].Value,
|
data: getResp.Kvs[0].Value,
|
||||||
rev: uint64(getResp.Kvs[0].ModRevision),
|
rev: uint64(getResp.Kvs[0].ModRevision),
|
||||||
}}
|
}}
|
||||||
if err := decodeList(elems, filter, listPtr, s.codec, s.versioner); err != nil {
|
if err := decodeList(elems, storage.SimpleFilter(pred), listPtr, s.codec, s.versioner); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
// update version with cluster level revision
|
// update version with cluster level revision
|
||||||
@ -281,7 +281,7 @@ func (s *store) GetToList(ctx context.Context, key string, filter storage.Filter
|
|||||||
}
|
}
|
||||||
|
|
||||||
// List implements storage.Interface.List.
|
// List implements storage.Interface.List.
|
||||||
func (s *store) List(ctx context.Context, key, resourceVersion string, filter storage.Filter, listObj runtime.Object) error {
|
func (s *store) List(ctx context.Context, key, resourceVersion string, pred storage.SelectionPredicate, listObj runtime.Object) error {
|
||||||
listPtr, err := meta.GetItemsPtr(listObj)
|
listPtr, err := meta.GetItemsPtr(listObj)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
@ -305,7 +305,7 @@ func (s *store) List(ctx context.Context, key, resourceVersion string, filter st
|
|||||||
rev: uint64(kv.ModRevision),
|
rev: uint64(kv.ModRevision),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if err := decodeList(elems, filter, listPtr, s.codec, s.versioner); err != nil {
|
if err := decodeList(elems, storage.SimpleFilter(pred), listPtr, s.codec, s.versioner); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
// update version with cluster level revision
|
// update version with cluster level revision
|
||||||
@ -313,16 +313,16 @@ func (s *store) List(ctx context.Context, key, resourceVersion string, filter st
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Watch implements storage.Interface.Watch.
|
// Watch implements storage.Interface.Watch.
|
||||||
func (s *store) Watch(ctx context.Context, key string, resourceVersion string, filter storage.Filter) (watch.Interface, error) {
|
func (s *store) Watch(ctx context.Context, key string, resourceVersion string, pred storage.SelectionPredicate) (watch.Interface, error) {
|
||||||
return s.watch(ctx, key, resourceVersion, filter, false)
|
return s.watch(ctx, key, resourceVersion, storage.SimpleFilter(pred), false)
|
||||||
}
|
}
|
||||||
|
|
||||||
// WatchList implements storage.Interface.WatchList.
|
// WatchList implements storage.Interface.WatchList.
|
||||||
func (s *store) WatchList(ctx context.Context, key string, resourceVersion string, filter storage.Filter) (watch.Interface, error) {
|
func (s *store) WatchList(ctx context.Context, key string, resourceVersion string, pred storage.SelectionPredicate) (watch.Interface, error) {
|
||||||
return s.watch(ctx, key, resourceVersion, filter, true)
|
return s.watch(ctx, key, resourceVersion, storage.SimpleFilter(pred), true)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *store) watch(ctx context.Context, key string, rv string, filter storage.Filter, recursive bool) (watch.Interface, error) {
|
func (s *store) watch(ctx context.Context, key string, rv string, filter storage.FilterFunc, recursive bool) (watch.Interface, error) {
|
||||||
rev, err := storage.ParseWatchResourceVersion(rv)
|
rev, err := storage.ParseWatchResourceVersion(rv)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
@ -416,7 +416,7 @@ func decode(codec runtime.Codec, versioner storage.Versioner, value []byte, objP
|
|||||||
|
|
||||||
// decodeList decodes a list of values into a list of objects, with resource version set to corresponding rev.
|
// decodeList decodes a list of values into a list of objects, with resource version set to corresponding rev.
|
||||||
// On success, ListPtr would be set to the list of objects.
|
// On success, ListPtr would be set to the list of objects.
|
||||||
func decodeList(elems []*elemForDecode, filter storage.Filter, ListPtr interface{}, codec runtime.Codec, versioner storage.Versioner) error {
|
func decodeList(elems []*elemForDecode, filter storage.FilterFunc, ListPtr interface{}, codec runtime.Codec, versioner storage.Versioner) error {
|
||||||
v, err := conversion.EnforcePtr(ListPtr)
|
v, err := conversion.EnforcePtr(ListPtr)
|
||||||
if err != nil || v.Kind() != reflect.Slice {
|
if err != nil || v.Kind() != reflect.Slice {
|
||||||
panic("need ptr to slice")
|
panic("need ptr to slice")
|
||||||
@ -428,7 +428,7 @@ func decodeList(elems []*elemForDecode, filter storage.Filter, ListPtr interface
|
|||||||
}
|
}
|
||||||
// being unable to set the version does not prevent the object from being extracted
|
// being unable to set the version does not prevent the object from being extracted
|
||||||
versioner.UpdateObject(obj, elem.rev)
|
versioner.UpdateObject(obj, elem.rev)
|
||||||
if filter.Filter(obj) {
|
if filter(obj) {
|
||||||
v.Set(reflect.Append(v, reflect.ValueOf(obj).Elem()))
|
v.Set(reflect.Append(v, reflect.ValueOf(obj).Elem()))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@ -24,6 +24,8 @@ import (
|
|||||||
|
|
||||||
"k8s.io/kubernetes/pkg/api"
|
"k8s.io/kubernetes/pkg/api"
|
||||||
"k8s.io/kubernetes/pkg/api/testapi"
|
"k8s.io/kubernetes/pkg/api/testapi"
|
||||||
|
"k8s.io/kubernetes/pkg/fields"
|
||||||
|
"k8s.io/kubernetes/pkg/labels"
|
||||||
"k8s.io/kubernetes/pkg/runtime"
|
"k8s.io/kubernetes/pkg/runtime"
|
||||||
"k8s.io/kubernetes/pkg/storage"
|
"k8s.io/kubernetes/pkg/storage"
|
||||||
|
|
||||||
@ -226,36 +228,32 @@ func TestGetToList(t *testing.T) {
|
|||||||
|
|
||||||
tests := []struct {
|
tests := []struct {
|
||||||
key string
|
key string
|
||||||
filter func(runtime.Object) bool
|
pred storage.SelectionPredicate
|
||||||
trigger func() []storage.MatchValue
|
|
||||||
expectedOut []*api.Pod
|
expectedOut []*api.Pod
|
||||||
}{{ // test GetToList on existing key
|
}{{ // test GetToList on existing key
|
||||||
key: key,
|
key: key,
|
||||||
filter: storage.EverythingFunc,
|
pred: storage.Everything,
|
||||||
trigger: storage.NoTriggerFunc,
|
|
||||||
expectedOut: []*api.Pod{storedObj},
|
expectedOut: []*api.Pod{storedObj},
|
||||||
}, { // test GetToList on non-existing key
|
}, { // test GetToList on non-existing key
|
||||||
key: "/non-existing",
|
key: "/non-existing",
|
||||||
filter: storage.EverythingFunc,
|
pred: storage.Everything,
|
||||||
trigger: storage.NoTriggerFunc,
|
|
||||||
expectedOut: nil,
|
expectedOut: nil,
|
||||||
}, { // test GetToList with filter to reject the pod
|
}, { // test GetToList with matching pod name
|
||||||
key: "/non-existing",
|
key: "/non-existing",
|
||||||
filter: func(obj runtime.Object) bool {
|
pred: storage.SelectionPredicate{
|
||||||
pod, ok := obj.(*api.Pod)
|
Label: labels.Everything(),
|
||||||
if !ok {
|
Field: fields.ParseSelectorOrDie("metadata.name!=" + storedObj.Name),
|
||||||
t.Fatal("It should be able to convert obj to *api.Pod")
|
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
||||||
}
|
pod := obj.(*api.Pod)
|
||||||
return pod.Name != storedObj.Name
|
return nil, fields.Set{"metadata.name": pod.Name}, nil
|
||||||
|
},
|
||||||
},
|
},
|
||||||
trigger: storage.NoTriggerFunc,
|
|
||||||
expectedOut: nil,
|
expectedOut: nil,
|
||||||
}}
|
}}
|
||||||
|
|
||||||
for i, tt := range tests {
|
for i, tt := range tests {
|
||||||
out := &api.PodList{}
|
out := &api.PodList{}
|
||||||
filter := storage.NewSimpleFilter(tt.filter, tt.trigger)
|
err := store.GetToList(ctx, tt.key, tt.pred, out)
|
||||||
err := store.GetToList(ctx, tt.key, filter, out)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("GetToList failed: %v", err)
|
t.Fatalf("GetToList failed: %v", err)
|
||||||
}
|
}
|
||||||
@ -492,41 +490,36 @@ func TestList(t *testing.T) {
|
|||||||
|
|
||||||
tests := []struct {
|
tests := []struct {
|
||||||
prefix string
|
prefix string
|
||||||
filter func(runtime.Object) bool
|
pred storage.SelectionPredicate
|
||||||
trigger func() []storage.MatchValue
|
|
||||||
expectedOut []*api.Pod
|
expectedOut []*api.Pod
|
||||||
}{{ // test List on existing key
|
}{{ // test List on existing key
|
||||||
prefix: "/one-level/",
|
prefix: "/one-level/",
|
||||||
filter: storage.EverythingFunc,
|
pred: storage.Everything,
|
||||||
trigger: storage.NoTriggerFunc,
|
|
||||||
expectedOut: []*api.Pod{preset[0].storedObj},
|
expectedOut: []*api.Pod{preset[0].storedObj},
|
||||||
}, { // test List on non-existing key
|
}, { // test List on non-existing key
|
||||||
prefix: "/non-existing/",
|
prefix: "/non-existing/",
|
||||||
filter: storage.EverythingFunc,
|
pred: storage.Everything,
|
||||||
trigger: storage.NoTriggerFunc,
|
|
||||||
expectedOut: nil,
|
expectedOut: nil,
|
||||||
}, { // test List with filter
|
}, { // test List with pod name matching
|
||||||
prefix: "/one-level/",
|
prefix: "/one-level/",
|
||||||
filter: func(obj runtime.Object) bool {
|
pred: storage.SelectionPredicate{
|
||||||
pod, ok := obj.(*api.Pod)
|
Label: labels.Everything(),
|
||||||
if !ok {
|
Field: fields.ParseSelectorOrDie("metadata.name!=" + preset[0].storedObj.Name),
|
||||||
t.Fatal("It should be able to convert obj to *api.Pod")
|
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
||||||
}
|
pod := obj.(*api.Pod)
|
||||||
return pod.Name != preset[0].storedObj.Name
|
return nil, fields.Set{"metadata.name": pod.Name}, nil
|
||||||
|
},
|
||||||
},
|
},
|
||||||
trigger: storage.NoTriggerFunc,
|
|
||||||
expectedOut: nil,
|
expectedOut: nil,
|
||||||
}, { // test List with multiple levels of directories and expect flattened result
|
}, { // test List with multiple levels of directories and expect flattened result
|
||||||
prefix: "/two-level/",
|
prefix: "/two-level/",
|
||||||
filter: storage.EverythingFunc,
|
pred: storage.Everything,
|
||||||
trigger: storage.NoTriggerFunc,
|
|
||||||
expectedOut: []*api.Pod{preset[1].storedObj, preset[2].storedObj},
|
expectedOut: []*api.Pod{preset[1].storedObj, preset[2].storedObj},
|
||||||
}}
|
}}
|
||||||
|
|
||||||
for i, tt := range tests {
|
for i, tt := range tests {
|
||||||
out := &api.PodList{}
|
out := &api.PodList{}
|
||||||
filter := storage.NewSimpleFilter(tt.filter, tt.trigger)
|
err := store.List(ctx, tt.prefix, "0", tt.pred, out)
|
||||||
err := store.List(ctx, tt.prefix, "0", filter, out)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("List failed: %v", err)
|
t.Fatalf("List failed: %v", err)
|
||||||
}
|
}
|
||||||
|
@ -51,7 +51,7 @@ type watchChan struct {
|
|||||||
key string
|
key string
|
||||||
initialRev int64
|
initialRev int64
|
||||||
recursive bool
|
recursive bool
|
||||||
filter storage.Filter
|
filter storage.FilterFunc
|
||||||
ctx context.Context
|
ctx context.Context
|
||||||
cancel context.CancelFunc
|
cancel context.CancelFunc
|
||||||
incomingEventChan chan *event
|
incomingEventChan chan *event
|
||||||
@ -74,7 +74,7 @@ func newWatcher(client *clientv3.Client, codec runtime.Codec, versioner storage.
|
|||||||
// If recursive is false, it watches on given key.
|
// If recursive is false, it watches on given key.
|
||||||
// If recursive is true, it watches any children and directories under the key, excluding the root key itself.
|
// If recursive is true, it watches any children and directories under the key, excluding the root key itself.
|
||||||
// filter must be non-nil. Only if filter returns true will the changes be returned.
|
// filter must be non-nil. Only if filter returns true will the changes be returned.
|
||||||
func (w *watcher) Watch(ctx context.Context, key string, rev int64, recursive bool, filter storage.Filter) (watch.Interface, error) {
|
func (w *watcher) Watch(ctx context.Context, key string, rev int64, recursive bool, filter storage.FilterFunc) (watch.Interface, error) {
|
||||||
if recursive && !strings.HasSuffix(key, "/") {
|
if recursive && !strings.HasSuffix(key, "/") {
|
||||||
key += "/"
|
key += "/"
|
||||||
}
|
}
|
||||||
@ -83,7 +83,7 @@ func (w *watcher) Watch(ctx context.Context, key string, rev int64, recursive bo
|
|||||||
return wc, nil
|
return wc, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (w *watcher) createWatchChan(ctx context.Context, key string, rev int64, recursive bool, filter storage.Filter) *watchChan {
|
func (w *watcher) createWatchChan(ctx context.Context, key string, rev int64, recursive bool, filter storage.FilterFunc) *watchChan {
|
||||||
wc := &watchChan{
|
wc := &watchChan{
|
||||||
watcher: w,
|
watcher: w,
|
||||||
key: key,
|
key: key,
|
||||||
@ -241,7 +241,7 @@ func (wc *watchChan) transform(e *event) (res *watch.Event) {
|
|||||||
|
|
||||||
switch {
|
switch {
|
||||||
case e.isDeleted:
|
case e.isDeleted:
|
||||||
if !wc.filter.Filter(oldObj) {
|
if !wc.filter(oldObj) {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
res = &watch.Event{
|
res = &watch.Event{
|
||||||
@ -249,7 +249,7 @@ func (wc *watchChan) transform(e *event) (res *watch.Event) {
|
|||||||
Object: oldObj,
|
Object: oldObj,
|
||||||
}
|
}
|
||||||
case e.isCreated:
|
case e.isCreated:
|
||||||
if !wc.filter.Filter(curObj) {
|
if !wc.filter(curObj) {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
res = &watch.Event{
|
res = &watch.Event{
|
||||||
@ -257,8 +257,8 @@ func (wc *watchChan) transform(e *event) (res *watch.Event) {
|
|||||||
Object: curObj,
|
Object: curObj,
|
||||||
}
|
}
|
||||||
default:
|
default:
|
||||||
curObjPasses := wc.filter.Filter(curObj)
|
curObjPasses := wc.filter(curObj)
|
||||||
oldObjPasses := wc.filter.Filter(oldObj)
|
oldObjPasses := wc.filter(oldObj)
|
||||||
switch {
|
switch {
|
||||||
case curObjPasses && oldObjPasses:
|
case curObjPasses && oldObjPasses:
|
||||||
res = &watch.Event{
|
res = &watch.Event{
|
||||||
|
@ -30,6 +30,8 @@ import (
|
|||||||
"k8s.io/kubernetes/pkg/api"
|
"k8s.io/kubernetes/pkg/api"
|
||||||
"k8s.io/kubernetes/pkg/api/testapi"
|
"k8s.io/kubernetes/pkg/api/testapi"
|
||||||
"k8s.io/kubernetes/pkg/api/unversioned"
|
"k8s.io/kubernetes/pkg/api/unversioned"
|
||||||
|
"k8s.io/kubernetes/pkg/fields"
|
||||||
|
"k8s.io/kubernetes/pkg/labels"
|
||||||
"k8s.io/kubernetes/pkg/runtime"
|
"k8s.io/kubernetes/pkg/runtime"
|
||||||
"k8s.io/kubernetes/pkg/storage"
|
"k8s.io/kubernetes/pkg/storage"
|
||||||
"k8s.io/kubernetes/pkg/util/wait"
|
"k8s.io/kubernetes/pkg/util/wait"
|
||||||
@ -54,45 +56,42 @@ func testWatch(t *testing.T, recursive bool) {
|
|||||||
|
|
||||||
tests := []struct {
|
tests := []struct {
|
||||||
key string
|
key string
|
||||||
filter func(runtime.Object) bool
|
pred storage.SelectionPredicate
|
||||||
trigger func() []storage.MatchValue
|
|
||||||
watchTests []*testWatchStruct
|
watchTests []*testWatchStruct
|
||||||
}{{ // create a key
|
}{{ // create a key
|
||||||
key: "/somekey-1",
|
key: "/somekey-1",
|
||||||
watchTests: []*testWatchStruct{{podFoo, true, watch.Added}},
|
watchTests: []*testWatchStruct{{podFoo, true, watch.Added}},
|
||||||
filter: storage.EverythingFunc,
|
pred: storage.Everything,
|
||||||
trigger: storage.NoTriggerFunc,
|
|
||||||
}, { // create a key but obj gets filtered
|
|
||||||
key: "/somekey-2",
|
|
||||||
watchTests: []*testWatchStruct{{podFoo, false, ""}},
|
|
||||||
filter: func(runtime.Object) bool { return false },
|
|
||||||
trigger: storage.NoTriggerFunc,
|
|
||||||
}, { // create a key but obj gets filtered. Then update it with unfiltered obj
|
}, { // create a key but obj gets filtered. Then update it with unfiltered obj
|
||||||
key: "/somekey-3",
|
key: "/somekey-3",
|
||||||
watchTests: []*testWatchStruct{{podFoo, false, ""}, {podBar, true, watch.Added}},
|
watchTests: []*testWatchStruct{{podFoo, false, ""}, {podBar, true, watch.Added}},
|
||||||
filter: func(obj runtime.Object) bool {
|
pred: storage.SelectionPredicate{
|
||||||
|
Label: labels.Everything(),
|
||||||
|
Field: fields.ParseSelectorOrDie("metadata.name=bar"),
|
||||||
|
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
||||||
pod := obj.(*api.Pod)
|
pod := obj.(*api.Pod)
|
||||||
return pod.Name == "bar"
|
return nil, fields.Set{"metadata.name": pod.Name}, nil
|
||||||
|
},
|
||||||
},
|
},
|
||||||
trigger: storage.NoTriggerFunc,
|
|
||||||
}, { // update
|
}, { // update
|
||||||
key: "/somekey-4",
|
key: "/somekey-4",
|
||||||
watchTests: []*testWatchStruct{{podFoo, true, watch.Added}, {podBar, true, watch.Modified}},
|
watchTests: []*testWatchStruct{{podFoo, true, watch.Added}, {podBar, true, watch.Modified}},
|
||||||
filter: storage.EverythingFunc,
|
pred: storage.Everything,
|
||||||
trigger: storage.NoTriggerFunc,
|
|
||||||
}, { // delete because of being filtered
|
}, { // delete because of being filtered
|
||||||
key: "/somekey-5",
|
key: "/somekey-5",
|
||||||
watchTests: []*testWatchStruct{{podFoo, true, watch.Added}, {podBar, true, watch.Deleted}},
|
watchTests: []*testWatchStruct{{podFoo, true, watch.Added}, {podBar, true, watch.Deleted}},
|
||||||
filter: func(obj runtime.Object) bool {
|
pred: storage.SelectionPredicate{
|
||||||
|
Label: labels.Everything(),
|
||||||
|
Field: fields.ParseSelectorOrDie("metadata.name!=bar"),
|
||||||
|
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
||||||
pod := obj.(*api.Pod)
|
pod := obj.(*api.Pod)
|
||||||
return pod.Name != "bar"
|
return nil, fields.Set{"metadata.name": pod.Name}, nil
|
||||||
|
},
|
||||||
},
|
},
|
||||||
trigger: storage.NoTriggerFunc,
|
|
||||||
}}
|
}}
|
||||||
for i, tt := range tests {
|
for i, tt := range tests {
|
||||||
ctx, store, cluster := testSetup(t)
|
ctx, store, cluster := testSetup(t)
|
||||||
filter := storage.NewSimpleFilter(tt.filter, tt.trigger)
|
w, err := store.watch(ctx, tt.key, "0", storage.SimpleFilter(tt.pred), recursive)
|
||||||
w, err := store.watch(ctx, tt.key, "0", filter, recursive)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("Watch failed: %v", err)
|
t.Fatalf("Watch failed: %v", err)
|
||||||
}
|
}
|
||||||
@ -198,7 +197,7 @@ func TestWatchContextCancel(t *testing.T) {
|
|||||||
cancel()
|
cancel()
|
||||||
// When we watch with a canceled context, we should detect that it's context canceled.
|
// When we watch with a canceled context, we should detect that it's context canceled.
|
||||||
// We won't take it as error and also close the watcher.
|
// We won't take it as error and also close the watcher.
|
||||||
w, err := store.watcher.Watch(canceledCtx, "/abc", 0, false, storage.Everything)
|
w, err := store.watcher.Watch(canceledCtx, "/abc", 0, false, storage.SimpleFilter(storage.Everything))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
@ -217,7 +216,7 @@ func TestWatchErrResultNotBlockAfterCancel(t *testing.T) {
|
|||||||
origCtx, store, cluster := testSetup(t)
|
origCtx, store, cluster := testSetup(t)
|
||||||
defer cluster.Terminate(t)
|
defer cluster.Terminate(t)
|
||||||
ctx, cancel := context.WithCancel(origCtx)
|
ctx, cancel := context.WithCancel(origCtx)
|
||||||
w := store.watcher.createWatchChan(ctx, "/abc", 0, false, storage.Everything)
|
w := store.watcher.createWatchChan(ctx, "/abc", 0, false, storage.SimpleFilter(storage.Everything))
|
||||||
// make resutlChan and errChan blocking to ensure ordering.
|
// make resutlChan and errChan blocking to ensure ordering.
|
||||||
w.resultChan = make(chan watch.Event)
|
w.resultChan = make(chan watch.Event)
|
||||||
w.errChan = make(chan error)
|
w.errChan = make(chan error)
|
||||||
|
@ -18,6 +18,8 @@ package storage
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"golang.org/x/net/context"
|
"golang.org/x/net/context"
|
||||||
|
"k8s.io/kubernetes/pkg/fields"
|
||||||
|
"k8s.io/kubernetes/pkg/labels"
|
||||||
"k8s.io/kubernetes/pkg/runtime"
|
"k8s.io/kubernetes/pkg/runtime"
|
||||||
"k8s.io/kubernetes/pkg/types"
|
"k8s.io/kubernetes/pkg/types"
|
||||||
"k8s.io/kubernetes/pkg/watch"
|
"k8s.io/kubernetes/pkg/watch"
|
||||||
@ -62,34 +64,14 @@ type MatchValue struct {
|
|||||||
// to that function.
|
// to that function.
|
||||||
type TriggerPublisherFunc func(obj runtime.Object) []MatchValue
|
type TriggerPublisherFunc func(obj runtime.Object) []MatchValue
|
||||||
|
|
||||||
// Filter is interface that is used to pass filtering mechanism.
|
// FilterFunc takes an API object and returns true if the object satisfies some requirements.
|
||||||
type Filter interface {
|
// TODO: We will remove this type and use SelectionPredicate everywhere.
|
||||||
// Filter is a predicate which takes an API object and returns true
|
type FilterFunc func(obj runtime.Object) bool
|
||||||
// if and only if the object should remain in the set.
|
|
||||||
Filter(obj runtime.Object) bool
|
|
||||||
// For any triggers known to the Filter, if Filter() can return only
|
|
||||||
// (a subset of) objects for which indexing function returns <value>,
|
|
||||||
// (<index name>, <value> pair would be returned.
|
|
||||||
//
|
|
||||||
// This is optimization to avoid computing Filter() function (which are
|
|
||||||
// usually relatively expensive) in case we are sure they will return
|
|
||||||
// false anyway.
|
|
||||||
Trigger() []MatchValue
|
|
||||||
}
|
|
||||||
|
|
||||||
// Everything is a Filter which accepts all objects.
|
// Everything accepts all objects.
|
||||||
var Everything Filter = everything{}
|
var Everything = SelectionPredicate{
|
||||||
|
Label: labels.Everything(),
|
||||||
// everything is implementation of Everything.
|
Field: fields.Everything(),
|
||||||
type everything struct {
|
|
||||||
}
|
|
||||||
|
|
||||||
func (e everything) Filter(runtime.Object) bool {
|
|
||||||
return true
|
|
||||||
}
|
|
||||||
|
|
||||||
func (e everything) Trigger() []MatchValue {
|
|
||||||
return nil
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// Pass an UpdateFunc to Interface.GuaranteedUpdate to make an update
|
// Pass an UpdateFunc to Interface.GuaranteedUpdate to make an update
|
||||||
@ -125,18 +107,18 @@ type Interface interface {
|
|||||||
Delete(ctx context.Context, key string, out runtime.Object, preconditions *Preconditions) error
|
Delete(ctx context.Context, key string, out runtime.Object, preconditions *Preconditions) error
|
||||||
|
|
||||||
// Watch begins watching the specified key. Events are decoded into API objects,
|
// Watch begins watching the specified key. Events are decoded into API objects,
|
||||||
// and any items passing 'filter' are sent down to returned watch.Interface.
|
// and any items selected by 'p' are sent down to returned watch.Interface.
|
||||||
// resourceVersion may be used to specify what version to begin watching,
|
// resourceVersion may be used to specify what version to begin watching,
|
||||||
// which should be the current resourceVersion, and no longer rv+1
|
// which should be the current resourceVersion, and no longer rv+1
|
||||||
// (e.g. reconnecting without missing any updates).
|
// (e.g. reconnecting without missing any updates).
|
||||||
Watch(ctx context.Context, key string, resourceVersion string, filter Filter) (watch.Interface, error)
|
Watch(ctx context.Context, key string, resourceVersion string, p SelectionPredicate) (watch.Interface, error)
|
||||||
|
|
||||||
// WatchList begins watching the specified key's items. Items are decoded into API
|
// WatchList begins watching the specified key's items. Items are decoded into API
|
||||||
// objects and any item passing 'filter' are sent down to returned watch.Interface.
|
// objects and any item selected by 'p' are sent down to returned watch.Interface.
|
||||||
// resourceVersion may be used to specify what version to begin watching,
|
// resourceVersion may be used to specify what version to begin watching,
|
||||||
// which should be the current resourceVersion, and no longer rv+1
|
// which should be the current resourceVersion, and no longer rv+1
|
||||||
// (e.g. reconnecting without missing any updates).
|
// (e.g. reconnecting without missing any updates).
|
||||||
WatchList(ctx context.Context, key string, resourceVersion string, filter Filter) (watch.Interface, error)
|
WatchList(ctx context.Context, key string, resourceVersion string, p SelectionPredicate) (watch.Interface, error)
|
||||||
|
|
||||||
// Get unmarshals json found at key into objPtr. On a not found error, will either
|
// Get unmarshals json found at key into objPtr. On a not found error, will either
|
||||||
// return a zero object of the requested type, or an error, depending on ignoreNotFound.
|
// return a zero object of the requested type, or an error, depending on ignoreNotFound.
|
||||||
@ -145,13 +127,13 @@ type Interface interface {
|
|||||||
|
|
||||||
// GetToList unmarshals json found at key and opaque it into *List api object
|
// GetToList unmarshals json found at key and opaque it into *List api object
|
||||||
// (an object that satisfies the runtime.IsList definition).
|
// (an object that satisfies the runtime.IsList definition).
|
||||||
GetToList(ctx context.Context, key string, filter Filter, listObj runtime.Object) error
|
GetToList(ctx context.Context, key string, p SelectionPredicate, listObj runtime.Object) error
|
||||||
|
|
||||||
// List unmarshalls jsons found at directory defined by key and opaque them
|
// List unmarshalls jsons found at directory defined by key and opaque them
|
||||||
// into *List api object (an object that satisfies runtime.IsList definition).
|
// into *List api object (an object that satisfies runtime.IsList definition).
|
||||||
// The returned contents may be delayed, but it is guaranteed that they will
|
// The returned contents may be delayed, but it is guaranteed that they will
|
||||||
// be have at least 'resourceVersion'.
|
// be have at least 'resourceVersion'.
|
||||||
List(ctx context.Context, key string, resourceVersion string, filter Filter, listObj runtime.Object) error
|
List(ctx context.Context, key string, resourceVersion string, p SelectionPredicate, listObj runtime.Object) error
|
||||||
|
|
||||||
// GuaranteedUpdate keeps calling 'tryUpdate()' to update key 'key' (of type 'ptrToType')
|
// GuaranteedUpdate keeps calling 'tryUpdate()' to update key 'key' (of type 'ptrToType')
|
||||||
// retrying the update until success if there is index conflict.
|
// retrying the update until success if there is index conflict.
|
||||||
|
77
pkg/storage/selection_predicate.go
Normal file
77
pkg/storage/selection_predicate.go
Normal file
@ -0,0 +1,77 @@
|
|||||||
|
/*
|
||||||
|
Copyright 2016 The Kubernetes Authors.
|
||||||
|
|
||||||
|
Licensed under the Apache License, Version 2.0 (the "License");
|
||||||
|
you may not use this file except in compliance with the License.
|
||||||
|
You may obtain a copy of the License at
|
||||||
|
|
||||||
|
http://www.apache.org/licenses/LICENSE-2.0
|
||||||
|
|
||||||
|
Unless required by applicable law or agreed to in writing, software
|
||||||
|
distributed under the License is distributed on an "AS IS" BASIS,
|
||||||
|
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||||
|
See the License for the specific language governing permissions and
|
||||||
|
limitations under the License.
|
||||||
|
*/
|
||||||
|
|
||||||
|
package storage
|
||||||
|
|
||||||
|
import (
|
||||||
|
"k8s.io/kubernetes/pkg/fields"
|
||||||
|
"k8s.io/kubernetes/pkg/labels"
|
||||||
|
"k8s.io/kubernetes/pkg/runtime"
|
||||||
|
)
|
||||||
|
|
||||||
|
// AttrFunc returns label and field sets for List or Watch to match.
|
||||||
|
// In any failure to parse given object, it returns error.
|
||||||
|
type AttrFunc func(obj runtime.Object) (labels.Set, fields.Set, error)
|
||||||
|
|
||||||
|
// SelectionPredicate is used to represent the way to select objects from api storage.
|
||||||
|
type SelectionPredicate struct {
|
||||||
|
Label labels.Selector
|
||||||
|
Field fields.Selector
|
||||||
|
GetAttrs AttrFunc
|
||||||
|
IndexFields []string
|
||||||
|
}
|
||||||
|
|
||||||
|
// Matches returns true if the given object's labels and fields (as
|
||||||
|
// returned by s.GetAttrs) match s.Label and s.Field. An error is
|
||||||
|
// returned if s.GetAttrs fails.
|
||||||
|
func (s *SelectionPredicate) Matches(obj runtime.Object) (bool, error) {
|
||||||
|
if s.Label.Empty() && s.Field.Empty() {
|
||||||
|
return true, nil
|
||||||
|
}
|
||||||
|
labels, fields, err := s.GetAttrs(obj)
|
||||||
|
if err != nil {
|
||||||
|
return false, err
|
||||||
|
}
|
||||||
|
matched := s.Label.Matches(labels)
|
||||||
|
if s.Field != nil {
|
||||||
|
matched = (matched && s.Field.Matches(fields))
|
||||||
|
}
|
||||||
|
return matched, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// MatchesSingle will return (name, true) if and only if s.Field matches on the object's
|
||||||
|
// name.
|
||||||
|
func (s *SelectionPredicate) MatchesSingle() (string, bool) {
|
||||||
|
// TODO: should be namespace.name
|
||||||
|
if name, ok := s.Field.RequiresExactMatch("metadata.name"); ok {
|
||||||
|
return name, true
|
||||||
|
}
|
||||||
|
return "", false
|
||||||
|
}
|
||||||
|
|
||||||
|
// For any index defined by IndexFields, if a matcher can match only (a subset)
|
||||||
|
// of objects that return <value> for a given index, a pair (<index name>, <value>)
|
||||||
|
// wil be returned.
|
||||||
|
// TODO: Consider supporting also labels.
|
||||||
|
func (s *SelectionPredicate) MatcherIndex() []MatchValue {
|
||||||
|
var result []MatchValue
|
||||||
|
for _, field := range s.IndexFields {
|
||||||
|
if value, ok := s.Field.RequiresExactMatch(field); ok {
|
||||||
|
result = append(result, MatchValue{IndexName: field, Value: value})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return result
|
||||||
|
}
|
@ -1,5 +1,5 @@
|
|||||||
/*
|
/*
|
||||||
Copyright 2014 The Kubernetes Authors.
|
Copyright 2016 The Kubernetes Authors.
|
||||||
|
|
||||||
Licensed under the Apache License, Version 2.0 (the "License");
|
Licensed under the Apache License, Version 2.0 (the "License");
|
||||||
you may not use this file except in compliance with the License.
|
you may not use this file except in compliance with the License.
|
||||||
@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
|
|||||||
limitations under the License.
|
limitations under the License.
|
||||||
*/
|
*/
|
||||||
|
|
||||||
package generic
|
package storage
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"errors"
|
"errors"
|
@ -22,6 +22,8 @@ import (
|
|||||||
"strings"
|
"strings"
|
||||||
"sync/atomic"
|
"sync/atomic"
|
||||||
|
|
||||||
|
"github.com/golang/glog"
|
||||||
|
|
||||||
"k8s.io/kubernetes/pkg/api/meta"
|
"k8s.io/kubernetes/pkg/api/meta"
|
||||||
"k8s.io/kubernetes/pkg/api/validation/path"
|
"k8s.io/kubernetes/pkg/api/validation/path"
|
||||||
"k8s.io/kubernetes/pkg/runtime"
|
"k8s.io/kubernetes/pkg/runtime"
|
||||||
@ -38,26 +40,16 @@ func SimpleUpdate(fn SimpleUpdateFunc) UpdateFunc {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// SimpleFilter implements Filter interface.
|
// SimpleFilter converts a selection predicate into a FilterFunc.
|
||||||
type SimpleFilter struct {
|
// It ignores any error from Matches().
|
||||||
filterFunc func(runtime.Object) bool
|
func SimpleFilter(p SelectionPredicate) FilterFunc {
|
||||||
triggerFunc func() []MatchValue
|
return func(obj runtime.Object) bool {
|
||||||
|
matches, err := p.Matches(obj)
|
||||||
|
if err != nil {
|
||||||
|
glog.Errorf("invalid object for matching. Obj: %v. Err: %v", obj, err)
|
||||||
|
return false
|
||||||
}
|
}
|
||||||
|
return matches
|
||||||
func (s *SimpleFilter) Filter(obj runtime.Object) bool {
|
|
||||||
return s.filterFunc(obj)
|
|
||||||
}
|
|
||||||
|
|
||||||
func (s *SimpleFilter) Trigger() []MatchValue {
|
|
||||||
return s.triggerFunc()
|
|
||||||
}
|
|
||||||
|
|
||||||
func NewSimpleFilter(
|
|
||||||
filterFunc func(runtime.Object) bool,
|
|
||||||
triggerFunc func() []MatchValue) Filter {
|
|
||||||
return &SimpleFilter{
|
|
||||||
filterFunc: filterFunc,
|
|
||||||
triggerFunc: triggerFunc,
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
Loading…
Reference in New Issue
Block a user