Skip to content
Projects
Groups
Snippets
Help
This project
Loading...
Sign in / Register
Toggle navigation
K
k3s
Project
Project
Details
Activity
Cycle Analytics
Repository
Repository
Files
Commits
Branches
Tags
Contributors
Graph
Compare
Charts
Issues
0
Issues
0
List
Board
Labels
Milestones
Merge Requests
0
Merge Requests
0
CI / CD
CI / CD
Pipelines
Jobs
Schedules
Charts
Registry
Registry
Wiki
Wiki
Snippets
Snippets
Members
Members
Collapse sidebar
Close sidebar
Activity
Graph
Charts
Create a new issue
Jobs
Commits
Issue Boards
Open sidebar
Jacklull
k3s
Commits
9a6ed3d0
Commit
9a6ed3d0
authored
Sep 21, 2016
by
Antoine Pelisse
Committed by
GitHub
Sep 21, 2016
Browse files
Options
Browse Files
Download
Plain Diff
Merge pull request #33230 from kubernetes/revert-32888-client-10-fixup-remaining-listers
Revert "simplify RC and SVC listers"
parents
ca00e596
93887258
Expand all
Show whitespace changes
Inline
Side-by-side
Showing
23 changed files
with
251 additions
and
240 deletions
+251
-240
endpoints_controller.go
contrib/mesos/pkg/service/endpoints_controller.go
+2
-3
cluster_helper.go
...ation/pkg/federation-controller/service/cluster_helper.go
+1
-2
service_helper.go
...ation/pkg/federation-controller/service/service_helper.go
+1
-1
servicecontroller.go
...on/pkg/federation-controller/service/servicecontroller.go
+2
-3
listers.go
pkg/client/cache/listers.go
+151
-1
listers_core.go
pkg/client/cache/listers_core.go
+9
-141
listers_test.go
pkg/client/cache/listers_test.go
+9
-9
endpoints_controller.go
pkg/controller/endpoint/endpoints_controller.go
+2
-3
endpoints_controller_test.go
pkg/controller/endpoint/endpoints_controller_test.go
+11
-11
replication_controller.go
pkg/controller/replication/replication_controller.go
+4
-4
replication_controller_utils.go
pkg/controller/replication/replication_controller_utils.go
+1
-1
servicecontroller.go
pkg/controller/service/servicecontroller.go
+2
-3
envvars.go
pkg/kubelet/envvars/envvars.go
+3
-3
envvars_test.go
pkg/kubelet/envvars/envvars_test.go
+4
-2
kubelet.go
pkg/kubelet/kubelet.go
+11
-14
kubelet_test.go
pkg/kubelet/kubelet_test.go
+8
-7
listers.go
plugin/pkg/scheduler/algorithm/listers.go
+15
-17
predicates.go
plugin/pkg/scheduler/algorithm/predicates/predicates.go
+1
-1
predicates_test.go
plugin/pkg/scheduler/algorithm/predicates/predicates_test.go
+9
-9
selector_spreading.go
.../pkg/scheduler/algorithm/priorities/selector_spreading.go
+1
-1
selector_spreading_test.go
...scheduler/algorithm/priorities/selector_spreading_test.go
+0
-0
factory.go
plugin/pkg/scheduler/factory/factory.go
+2
-2
generic_scheduler_test.go
plugin/pkg/scheduler/generic_scheduler_test.go
+2
-2
No files found.
contrib/mesos/pkg/service/endpoints_controller.go
View file @
9a6ed3d0
...
@@ -55,7 +55,7 @@ func NewEndpointController(client *clientset.Clientset) *endpointController {
...
@@ -55,7 +55,7 @@ func NewEndpointController(client *clientset.Clientset) *endpointController {
client
:
client
,
client
:
client
,
queue
:
workqueue
.
NewNamedRateLimitingQueue
(
workqueue
.
DefaultControllerRateLimiter
(),
"endpoint"
),
queue
:
workqueue
.
NewNamedRateLimitingQueue
(
workqueue
.
DefaultControllerRateLimiter
(),
"endpoint"
),
}
}
e
.
serviceStore
.
Indexer
,
e
.
serviceController
=
cache
.
NewIndexer
Informer
(
e
.
serviceStore
.
Store
,
e
.
serviceController
=
cache
.
New
Informer
(
&
cache
.
ListWatch
{
&
cache
.
ListWatch
{
ListFunc
:
func
(
options
api
.
ListOptions
)
(
runtime
.
Object
,
error
)
{
ListFunc
:
func
(
options
api
.
ListOptions
)
(
runtime
.
Object
,
error
)
{
return
e
.
client
.
Core
()
.
Services
(
api
.
NamespaceAll
)
.
List
(
options
)
return
e
.
client
.
Core
()
.
Services
(
api
.
NamespaceAll
)
.
List
(
options
)
...
@@ -73,7 +73,6 @@ func NewEndpointController(client *clientset.Clientset) *endpointController {
...
@@ -73,7 +73,6 @@ func NewEndpointController(client *clientset.Clientset) *endpointController {
},
},
DeleteFunc
:
e
.
enqueueService
,
DeleteFunc
:
e
.
enqueueService
,
},
},
cache
.
Indexers
{
cache
.
NamespaceIndex
:
cache
.
MetaNamespaceIndexFunc
},
)
)
e
.
podStore
.
Indexer
,
e
.
podController
=
cache
.
NewIndexerInformer
(
e
.
podStore
.
Indexer
,
e
.
podController
=
cache
.
NewIndexerInformer
(
...
@@ -263,7 +262,7 @@ func (e *endpointController) syncService(key string) error {
...
@@ -263,7 +262,7 @@ func (e *endpointController) syncService(key string) error {
defer
func
()
{
defer
func
()
{
glog
.
V
(
4
)
.
Infof
(
"Finished syncing service %q endpoints. (%v)"
,
key
,
time
.
Now
()
.
Sub
(
startTime
))
glog
.
V
(
4
)
.
Infof
(
"Finished syncing service %q endpoints. (%v)"
,
key
,
time
.
Now
()
.
Sub
(
startTime
))
}()
}()
obj
,
exists
,
err
:=
e
.
serviceStore
.
Indexer
.
GetByKey
(
key
)
obj
,
exists
,
err
:=
e
.
serviceStore
.
Store
.
GetByKey
(
key
)
if
err
!=
nil
||
!
exists
{
if
err
!=
nil
||
!
exists
{
// Delete the corresponding endpoint, as the service has been deleted.
// Delete the corresponding endpoint, as the service has been deleted.
// TODO: Please note that this will delete an endpoint when a
// TODO: Please note that this will delete an endpoint when a
...
...
federation/pkg/federation-controller/service/cluster_helper.go
View file @
9a6ed3d0
...
@@ -114,7 +114,7 @@ func (cc *clusterClientCache) startClusterLW(cluster *v1beta1.Cluster, clusterNa
...
@@ -114,7 +114,7 @@ func (cc *clusterClientCache) startClusterLW(cluster *v1beta1.Cluster, clusterNa
},
},
)
)
cachedClusterClient
.
serviceStore
.
Indexer
,
cachedClusterClient
.
serviceController
=
cache
.
NewIndexer
Informer
(
cachedClusterClient
.
serviceStore
.
Store
,
cachedClusterClient
.
serviceController
=
cache
.
New
Informer
(
&
cache
.
ListWatch
{
&
cache
.
ListWatch
{
ListFunc
:
func
(
options
api
.
ListOptions
)
(
pkg_runtime
.
Object
,
error
)
{
ListFunc
:
func
(
options
api
.
ListOptions
)
(
pkg_runtime
.
Object
,
error
)
{
return
clientset
.
Core
()
.
Services
(
v1
.
NamespaceAll
)
.
List
(
options
)
return
clientset
.
Core
()
.
Services
(
v1
.
NamespaceAll
)
.
List
(
options
)
...
@@ -149,7 +149,6 @@ func (cc *clusterClientCache) startClusterLW(cluster *v1beta1.Cluster, clusterNa
...
@@ -149,7 +149,6 @@ func (cc *clusterClientCache) startClusterLW(cluster *v1beta1.Cluster, clusterNa
glog
.
V
(
2
)
.
Infof
(
"Service %s/%s deletion found and enque to service store %s"
,
service
.
Namespace
,
service
.
Name
,
clusterName
)
glog
.
V
(
2
)
.
Infof
(
"Service %s/%s deletion found and enque to service store %s"
,
service
.
Namespace
,
service
.
Name
,
clusterName
)
},
},
},
},
cache
.
Indexers
{
cache
.
NamespaceIndex
:
cache
.
MetaNamespaceIndexFunc
},
)
)
cc
.
clientMap
[
clusterName
]
=
cachedClusterClient
cc
.
clientMap
[
clusterName
]
=
cachedClusterClient
go
cachedClusterClient
.
serviceController
.
Run
(
wait
.
NeverStop
)
go
cachedClusterClient
.
serviceController
.
Run
(
wait
.
NeverStop
)
...
...
federation/pkg/federation-controller/service/service_helper.go
View file @
9a6ed3d0
...
@@ -63,7 +63,7 @@ func (cc *clusterClientCache) syncService(key, clusterName string, clusterCache
...
@@ -63,7 +63,7 @@ func (cc *clusterClientCache) syncService(key, clusterName string, clusterCache
// if serviceCache does not exists, that means the service is not created by federation, we should skip it
// if serviceCache does not exists, that means the service is not created by federation, we should skip it
return
nil
return
nil
}
}
serviceInterface
,
exists
,
err
:=
clusterCache
.
serviceStore
.
Indexer
.
GetByKey
(
key
)
serviceInterface
,
exists
,
err
:=
clusterCache
.
serviceStore
.
GetByKey
(
key
)
if
err
!=
nil
{
if
err
!=
nil
{
glog
.
Errorf
(
"Did not successfully get %v from store: %v, will retry later"
,
key
,
err
)
glog
.
Errorf
(
"Did not successfully get %v from store: %v, will retry later"
,
key
,
err
)
clusterCache
.
serviceQueue
.
Add
(
key
)
clusterCache
.
serviceQueue
.
Add
(
key
)
...
...
federation/pkg/federation-controller/service/servicecontroller.go
View file @
9a6ed3d0
...
@@ -144,7 +144,7 @@ func New(federationClient federation_release_1_4.Interface, dns dnsprovider.Inte
...
@@ -144,7 +144,7 @@ func New(federationClient federation_release_1_4.Interface, dns dnsprovider.Inte
queue
:
workqueue
.
New
(),
queue
:
workqueue
.
New
(),
knownClusterSet
:
make
(
sets
.
String
),
knownClusterSet
:
make
(
sets
.
String
),
}
}
s
.
serviceStore
.
Indexer
,
s
.
serviceController
=
cache
.
NewIndexer
Informer
(
s
.
serviceStore
.
Store
,
s
.
serviceController
=
cache
.
New
Informer
(
&
cache
.
ListWatch
{
&
cache
.
ListWatch
{
ListFunc
:
func
(
options
api
.
ListOptions
)
(
pkg_runtime
.
Object
,
error
)
{
ListFunc
:
func
(
options
api
.
ListOptions
)
(
pkg_runtime
.
Object
,
error
)
{
return
s
.
federationClient
.
Core
()
.
Services
(
v1
.
NamespaceAll
)
.
List
(
options
)
return
s
.
federationClient
.
Core
()
.
Services
(
v1
.
NamespaceAll
)
.
List
(
options
)
...
@@ -165,7 +165,6 @@ func New(federationClient federation_release_1_4.Interface, dns dnsprovider.Inte
...
@@ -165,7 +165,6 @@ func New(federationClient federation_release_1_4.Interface, dns dnsprovider.Inte
},
},
DeleteFunc
:
s
.
enqueueService
,
DeleteFunc
:
s
.
enqueueService
,
},
},
cache
.
Indexers
{
cache
.
NamespaceIndex
:
cache
.
MetaNamespaceIndexFunc
},
)
)
s
.
clusterStore
.
Store
,
s
.
clusterController
=
cache
.
NewInformer
(
s
.
clusterStore
.
Store
,
s
.
clusterController
=
cache
.
NewInformer
(
&
cache
.
ListWatch
{
&
cache
.
ListWatch
{
...
@@ -817,7 +816,7 @@ func (s *ServiceController) syncService(key string) error {
...
@@ -817,7 +816,7 @@ func (s *ServiceController) syncService(key string) error {
glog
.
V
(
4
)
.
Infof
(
"Finished syncing service %q (%v)"
,
key
,
time
.
Now
()
.
Sub
(
startTime
))
glog
.
V
(
4
)
.
Infof
(
"Finished syncing service %q (%v)"
,
key
,
time
.
Now
()
.
Sub
(
startTime
))
}()
}()
// obj holds the latest service info from apiserver
// obj holds the latest service info from apiserver
obj
,
exists
,
err
:=
s
.
serviceStore
.
Indexer
.
GetByKey
(
key
)
obj
,
exists
,
err
:=
s
.
serviceStore
.
Store
.
GetByKey
(
key
)
if
err
!=
nil
{
if
err
!=
nil
{
glog
.
Errorf
(
"Unable to retrieve service %v from store: %v"
,
key
,
err
)
glog
.
Errorf
(
"Unable to retrieve service %v from store: %v"
,
key
,
err
)
s
.
queue
.
Add
(
key
)
s
.
queue
.
Add
(
key
)
...
...
pkg/client/cache/listers.go
View file @
9a6ed3d0
...
@@ -32,7 +32,6 @@ import (
...
@@ -32,7 +32,6 @@ import (
"k8s.io/kubernetes/pkg/labels"
"k8s.io/kubernetes/pkg/labels"
)
)
// AppendFunc is used to add a matching item to whatever list the caller is using
type
AppendFunc
func
(
interface
{})
type
AppendFunc
func
(
interface
{})
func
ListAll
(
store
Store
,
selector
labels
.
Selector
,
appendFn
AppendFunc
)
error
{
func
ListAll
(
store
Store
,
selector
labels
.
Selector
,
appendFn
AppendFunc
)
error
{
...
@@ -137,6 +136,116 @@ func (s storeToNodeConditionLister) List() (nodes []*api.Node, err error) {
...
@@ -137,6 +136,116 @@ func (s storeToNodeConditionLister) List() (nodes []*api.Node, err error) {
return
return
}
}
// StoreToReplicationControllerLister gives a store List and Exists methods. The store must contain only ReplicationControllers.
type
StoreToReplicationControllerLister
struct
{
Indexer
}
// Exists checks if the given rc exists in the store.
func
(
s
*
StoreToReplicationControllerLister
)
Exists
(
controller
*
api
.
ReplicationController
)
(
bool
,
error
)
{
_
,
exists
,
err
:=
s
.
Indexer
.
Get
(
controller
)
if
err
!=
nil
{
return
false
,
err
}
return
exists
,
nil
}
// StoreToReplicationControllerLister lists all controllers in the store.
// TODO: converge on the interface in pkg/client
func
(
s
*
StoreToReplicationControllerLister
)
List
()
(
controllers
[]
api
.
ReplicationController
,
err
error
)
{
for
_
,
c
:=
range
s
.
Indexer
.
List
()
{
controllers
=
append
(
controllers
,
*
(
c
.
(
*
api
.
ReplicationController
)))
}
return
controllers
,
nil
}
func
(
s
*
StoreToReplicationControllerLister
)
ReplicationControllers
(
namespace
string
)
storeReplicationControllersNamespacer
{
return
storeReplicationControllersNamespacer
{
s
.
Indexer
,
namespace
}
}
type
storeReplicationControllersNamespacer
struct
{
indexer
Indexer
namespace
string
}
func
(
s
storeReplicationControllersNamespacer
)
List
(
selector
labels
.
Selector
)
([]
api
.
ReplicationController
,
error
)
{
controllers
:=
[]
api
.
ReplicationController
{}
if
s
.
namespace
==
api
.
NamespaceAll
{
for
_
,
m
:=
range
s
.
indexer
.
List
()
{
rc
:=
*
(
m
.
(
*
api
.
ReplicationController
))
if
selector
.
Matches
(
labels
.
Set
(
rc
.
Labels
))
{
controllers
=
append
(
controllers
,
rc
)
}
}
return
controllers
,
nil
}
key
:=
&
api
.
ReplicationController
{
ObjectMeta
:
api
.
ObjectMeta
{
Namespace
:
s
.
namespace
}}
items
,
err
:=
s
.
indexer
.
Index
(
NamespaceIndex
,
key
)
if
err
!=
nil
{
// Ignore error; do slow search without index.
glog
.
Warningf
(
"can not retrieve list of objects using index : %v"
,
err
)
for
_
,
m
:=
range
s
.
indexer
.
List
()
{
rc
:=
*
(
m
.
(
*
api
.
ReplicationController
))
if
s
.
namespace
==
rc
.
Namespace
&&
selector
.
Matches
(
labels
.
Set
(
rc
.
Labels
))
{
controllers
=
append
(
controllers
,
rc
)
}
}
return
controllers
,
nil
}
for
_
,
m
:=
range
items
{
rc
:=
*
(
m
.
(
*
api
.
ReplicationController
))
if
selector
.
Matches
(
labels
.
Set
(
rc
.
Labels
))
{
controllers
=
append
(
controllers
,
rc
)
}
}
return
controllers
,
nil
}
func
(
s
storeReplicationControllersNamespacer
)
Get
(
name
string
)
(
*
api
.
ReplicationController
,
error
)
{
obj
,
exists
,
err
:=
s
.
indexer
.
GetByKey
(
s
.
namespace
+
"/"
+
name
)
if
err
!=
nil
{
return
nil
,
err
}
if
!
exists
{
return
nil
,
errors
.
NewNotFound
(
api
.
Resource
(
"replicationcontroller"
),
name
)
}
return
obj
.
(
*
api
.
ReplicationController
),
nil
}
// GetPodControllers returns a list of replication controllers managing a pod. Returns an error only if no matching controllers are found.
func
(
s
*
StoreToReplicationControllerLister
)
GetPodControllers
(
pod
*
api
.
Pod
)
(
controllers
[]
api
.
ReplicationController
,
err
error
)
{
var
selector
labels
.
Selector
var
rc
api
.
ReplicationController
if
len
(
pod
.
Labels
)
==
0
{
err
=
fmt
.
Errorf
(
"no controllers found for pod %v because it has no labels"
,
pod
.
Name
)
return
}
key
:=
&
api
.
ReplicationController
{
ObjectMeta
:
api
.
ObjectMeta
{
Namespace
:
pod
.
Namespace
}}
items
,
err
:=
s
.
Indexer
.
Index
(
NamespaceIndex
,
key
)
if
err
!=
nil
{
return
}
for
_
,
m
:=
range
items
{
rc
=
*
m
.
(
*
api
.
ReplicationController
)
selector
=
labels
.
Set
(
rc
.
Spec
.
Selector
)
.
AsSelectorPreValidated
()
// If an rc with a nil or empty selector creeps in, it should match nothing, not everything.
if
selector
.
Empty
()
||
!
selector
.
Matches
(
labels
.
Set
(
pod
.
Labels
))
{
continue
}
controllers
=
append
(
controllers
,
rc
)
}
if
len
(
controllers
)
==
0
{
err
=
fmt
.
Errorf
(
"could not find controller for pod %s in namespace %s with labels: %v"
,
pod
.
Name
,
pod
.
Namespace
,
pod
.
Labels
)
}
return
}
// StoreToDeploymentLister gives a store List and Exists methods. The store must contain only Deployments.
// StoreToDeploymentLister gives a store List and Exists methods. The store must contain only Deployments.
type
StoreToDeploymentLister
struct
{
type
StoreToDeploymentLister
struct
{
Indexer
Indexer
...
@@ -409,6 +518,47 @@ func (s *StoreToDaemonSetLister) GetPodDaemonSets(pod *api.Pod) (daemonSets []ex
...
@@ -409,6 +518,47 @@ func (s *StoreToDaemonSetLister) GetPodDaemonSets(pod *api.Pod) (daemonSets []ex
return
return
}
}
// StoreToServiceLister makes a Store that has the List method of the client.ServiceInterface
// The Store must contain (only) Services.
type
StoreToServiceLister
struct
{
Store
}
func
(
s
*
StoreToServiceLister
)
List
()
(
services
api
.
ServiceList
,
err
error
)
{
for
_
,
m
:=
range
s
.
Store
.
List
()
{
services
.
Items
=
append
(
services
.
Items
,
*
(
m
.
(
*
api
.
Service
)))
}
return
services
,
nil
}
// TODO: Move this back to scheduler as a helper function that takes a Store,
// rather than a method of StoreToServiceLister.
func
(
s
*
StoreToServiceLister
)
GetPodServices
(
pod
*
api
.
Pod
)
(
services
[]
api
.
Service
,
err
error
)
{
var
selector
labels
.
Selector
var
service
api
.
Service
for
_
,
m
:=
range
s
.
Store
.
List
()
{
service
=
*
m
.
(
*
api
.
Service
)
// consider only services that are in the same namespace as the pod
if
service
.
Namespace
!=
pod
.
Namespace
{
continue
}
if
service
.
Spec
.
Selector
==
nil
{
// services with nil selectors match nothing, not everything.
continue
}
selector
=
labels
.
Set
(
service
.
Spec
.
Selector
)
.
AsSelectorPreValidated
()
if
selector
.
Matches
(
labels
.
Set
(
pod
.
Labels
))
{
services
=
append
(
services
,
service
)
}
}
if
len
(
services
)
==
0
{
err
=
fmt
.
Errorf
(
"could not find service for pod %s in namespace %s with labels: %v"
,
pod
.
Name
,
pod
.
Namespace
,
pod
.
Labels
)
}
return
}
// StoreToEndpointsLister makes a Store that lists endpoints.
// StoreToEndpointsLister makes a Store that lists endpoints.
type
StoreToEndpointsLister
struct
{
type
StoreToEndpointsLister
struct
{
Store
Store
...
...
pkg/client/cache/listers_core.go
View file @
9a6ed3d0
...
@@ -17,8 +17,6 @@ limitations under the License.
...
@@ -17,8 +17,6 @@ limitations under the License.
package
cache
package
cache
import
(
import
(
"fmt"
"k8s.io/kubernetes/pkg/api"
"k8s.io/kubernetes/pkg/api"
"k8s.io/kubernetes/pkg/api/errors"
"k8s.io/kubernetes/pkg/api/errors"
"k8s.io/kubernetes/pkg/labels"
"k8s.io/kubernetes/pkg/labels"
...
@@ -27,24 +25,24 @@ import (
...
@@ -27,24 +25,24 @@ import (
// TODO: generate these classes and methods for all resources of interest using
// TODO: generate these classes and methods for all resources of interest using
// a script. Can use "go generate" once 1.4 is supported by all users.
// a script. Can use "go generate" once 1.4 is supported by all users.
// Lister makes an Index have the List method. The Stores must contain only the expected type
// StoreToPodLister makes a Store have the List method of the client.PodInterface
// The Store must contain (only) Pods.
//
// Example:
// Example:
// s := cache.NewStore()
// s := cache.NewStore()
// lw := cache.ListWatch{Client: c, FieldSelector: sel, Resource: "pods"}
// lw := cache.ListWatch{Client: c, FieldSelector: sel, Resource: "pods"}
// r := cache.NewReflector(lw, &api.Pod{}, s).Run()
// r := cache.NewReflector(lw, &api.Pod{}, s).Run()
// l := StoreToPodLister{s}
// l := StoreToPodLister{s}
// l.List()
// l.List()
// StoreToPodLister helps list pods
type
StoreToPodLister
struct
{
type
StoreToPodLister
struct
{
Indexer
Indexer
Indexer
Indexer
}
}
func
(
s
*
StoreToPodLister
)
List
(
selector
labels
.
Selector
)
(
ret
[]
*
api
.
Pod
,
err
error
)
{
func
(
s
*
StoreToPodLister
)
List
(
selector
labels
.
Selector
)
(
pods
[]
*
api
.
Pod
,
err
error
)
{
err
=
ListAll
(
s
.
Indexer
,
selector
,
func
(
m
interface
{})
{
err
=
ListAll
(
s
.
Indexer
,
selector
,
func
(
m
interface
{})
{
ret
=
append
(
ret
,
m
.
(
*
api
.
Pod
))
pods
=
append
(
pods
,
m
.
(
*
api
.
Pod
))
})
})
return
ret
,
err
return
pods
,
err
}
}
func
(
s
*
StoreToPodLister
)
Pods
(
namespace
string
)
storePodsNamespacer
{
func
(
s
*
StoreToPodLister
)
Pods
(
namespace
string
)
storePodsNamespacer
{
...
@@ -56,11 +54,11 @@ type storePodsNamespacer struct {
...
@@ -56,11 +54,11 @@ type storePodsNamespacer struct {
namespace
string
namespace
string
}
}
func
(
s
storePodsNamespacer
)
List
(
selector
labels
.
Selector
)
(
ret
[]
*
api
.
Pod
,
err
error
)
{
func
(
s
storePodsNamespacer
)
List
(
selector
labels
.
Selector
)
(
pods
[]
*
api
.
Pod
,
err
error
)
{
err
=
ListAllByNamespace
(
s
.
Indexer
,
s
.
namespace
,
selector
,
func
(
m
interface
{})
{
err
=
ListAllByNamespace
(
s
.
Indexer
,
s
.
namespace
,
selector
,
func
(
m
interface
{})
{
ret
=
append
(
ret
,
m
.
(
*
api
.
Pod
))
pods
=
append
(
pods
,
m
.
(
*
api
.
Pod
))
})
})
return
ret
,
err
return
pods
,
err
}
}
func
(
s
storePodsNamespacer
)
Get
(
name
string
)
(
*
api
.
Pod
,
error
)
{
func
(
s
storePodsNamespacer
)
Get
(
name
string
)
(
*
api
.
Pod
,
error
)
{
...
@@ -73,133 +71,3 @@ func (s storePodsNamespacer) Get(name string) (*api.Pod, error) {
...
@@ -73,133 +71,3 @@ func (s storePodsNamespacer) Get(name string) (*api.Pod, error) {
}
}
return
obj
.
(
*
api
.
Pod
),
nil
return
obj
.
(
*
api
.
Pod
),
nil
}
}
// StoreToReplicationControllerLister helps list rcs
type
StoreToReplicationControllerLister
struct
{
Indexer
Indexer
}
func
(
s
*
StoreToReplicationControllerLister
)
List
(
selector
labels
.
Selector
)
(
ret
[]
*
api
.
ReplicationController
,
err
error
)
{
err
=
ListAll
(
s
.
Indexer
,
selector
,
func
(
m
interface
{})
{
ret
=
append
(
ret
,
m
.
(
*
api
.
ReplicationController
))
})
return
ret
,
err
}
func
(
s
*
StoreToReplicationControllerLister
)
ReplicationControllers
(
namespace
string
)
storeReplicationControllersNamespacer
{
return
storeReplicationControllersNamespacer
{
s
.
Indexer
,
namespace
}
}
type
storeReplicationControllersNamespacer
struct
{
indexer
Indexer
namespace
string
}
func
(
s
storeReplicationControllersNamespacer
)
List
(
selector
labels
.
Selector
)
(
ret
[]
*
api
.
ReplicationController
,
err
error
)
{
err
=
ListAllByNamespace
(
s
.
indexer
,
s
.
namespace
,
selector
,
func
(
m
interface
{})
{
ret
=
append
(
ret
,
m
.
(
*
api
.
ReplicationController
))
})
return
ret
,
err
}
func
(
s
storeReplicationControllersNamespacer
)
Get
(
name
string
)
(
*
api
.
ReplicationController
,
error
)
{
obj
,
exists
,
err
:=
s
.
indexer
.
GetByKey
(
s
.
namespace
+
"/"
+
name
)
if
err
!=
nil
{
return
nil
,
err
}
if
!
exists
{
return
nil
,
errors
.
NewNotFound
(
api
.
Resource
(
"replicationcontroller"
),
name
)
}
return
obj
.
(
*
api
.
ReplicationController
),
nil
}
// GetPodControllers returns a list of replication controllers managing a pod. Returns an error only if no matching controllers are found.
func
(
s
*
StoreToReplicationControllerLister
)
GetPodControllers
(
pod
*
api
.
Pod
)
(
controllers
[]
*
api
.
ReplicationController
,
err
error
)
{
if
len
(
pod
.
Labels
)
==
0
{
err
=
fmt
.
Errorf
(
"no controllers found for pod %v because it has no labels"
,
pod
.
Name
)
return
}
key
:=
&
api
.
ReplicationController
{
ObjectMeta
:
api
.
ObjectMeta
{
Namespace
:
pod
.
Namespace
}}
items
,
err
:=
s
.
Indexer
.
Index
(
NamespaceIndex
,
key
)
if
err
!=
nil
{
return
}
for
_
,
m
:=
range
items
{
rc
:=
m
.
(
*
api
.
ReplicationController
)
selector
:=
labels
.
Set
(
rc
.
Spec
.
Selector
)
.
AsSelectorPreValidated
()
// If an rc with a nil or empty selector creeps in, it should match nothing, not everything.
if
selector
.
Empty
()
||
!
selector
.
Matches
(
labels
.
Set
(
pod
.
Labels
))
{
continue
}
controllers
=
append
(
controllers
,
rc
)
}
if
len
(
controllers
)
==
0
{
err
=
fmt
.
Errorf
(
"could not find controller for pod %s in namespace %s with labels: %v"
,
pod
.
Name
,
pod
.
Namespace
,
pod
.
Labels
)
}
return
}
// StoreToServiceLister helps list services
type
StoreToServiceLister
struct
{
Indexer
Indexer
}
func
(
s
*
StoreToServiceLister
)
List
(
selector
labels
.
Selector
)
(
ret
[]
*
api
.
Service
,
err
error
)
{
err
=
ListAll
(
s
.
Indexer
,
selector
,
func
(
m
interface
{})
{
ret
=
append
(
ret
,
m
.
(
*
api
.
Service
))
})
return
ret
,
err
}
func
(
s
*
StoreToServiceLister
)
Services
(
namespace
string
)
storeServicesNamespacer
{
return
storeServicesNamespacer
{
s
.
Indexer
,
namespace
}
}
type
storeServicesNamespacer
struct
{
indexer
Indexer
namespace
string
}
func
(
s
storeServicesNamespacer
)
List
(
selector
labels
.
Selector
)
(
ret
[]
*
api
.
Service
,
err
error
)
{
err
=
ListAllByNamespace
(
s
.
indexer
,
s
.
namespace
,
selector
,
func
(
m
interface
{})
{
ret
=
append
(
ret
,
m
.
(
*
api
.
Service
))
})
return
ret
,
err
}
func
(
s
storeServicesNamespacer
)
Get
(
name
string
)
(
*
api
.
Service
,
error
)
{
obj
,
exists
,
err
:=
s
.
indexer
.
GetByKey
(
s
.
namespace
+
"/"
+
name
)
if
err
!=
nil
{
return
nil
,
err
}
if
!
exists
{
return
nil
,
errors
.
NewNotFound
(
api
.
Resource
(
"service"
),
name
)
}
return
obj
.
(
*
api
.
Service
),
nil
}
// TODO: Move this back to scheduler as a helper function that takes a Store,
// rather than a method of StoreToServiceLister.
func
(
s
*
StoreToServiceLister
)
GetPodServices
(
pod
*
api
.
Pod
)
(
services
[]
*
api
.
Service
,
err
error
)
{
allServices
,
err
:=
s
.
Services
(
pod
.
Namespace
)
.
List
(
labels
.
Everything
())
if
err
!=
nil
{
return
nil
,
err
}
for
i
:=
range
allServices
{
service
:=
allServices
[
i
]
if
service
.
Spec
.
Selector
==
nil
{
// services with nil selectors match nothing, not everything.
continue
}
selector
:=
labels
.
Set
(
service
.
Spec
.
Selector
)
.
AsSelectorPreValidated
()
if
selector
.
Matches
(
labels
.
Set
(
pod
.
Labels
))
{
services
=
append
(
services
,
service
)
}
}
return
services
,
nil
}
pkg/client/cache/listers_test.go
View file @
9a6ed3d0
...
@@ -128,7 +128,7 @@ func TestStoreToReplicationControllerLister(t *testing.T) {
...
@@ -128,7 +128,7 @@ func TestStoreToReplicationControllerLister(t *testing.T) {
testCases
:=
[]
struct
{
testCases
:=
[]
struct
{
description
string
description
string
inRCs
[]
*
api
.
ReplicationController
inRCs
[]
*
api
.
ReplicationController
list
func
(
StoreToReplicationControllerLister
)
([]
*
api
.
ReplicationController
,
error
)
list
func
(
StoreToReplicationControllerLister
)
([]
api
.
ReplicationController
,
error
)
outRCNames
sets
.
String
outRCNames
sets
.
String
expectErr
bool
expectErr
bool
onlyIfIndexedByNamespace
bool
onlyIfIndexedByNamespace
bool
...
@@ -143,7 +143,7 @@ func TestStoreToReplicationControllerLister(t *testing.T) {
...
@@ -143,7 +143,7 @@ func TestStoreToReplicationControllerLister(t *testing.T) {
ObjectMeta
:
api
.
ObjectMeta
{
Name
:
"hmm"
,
Namespace
:
"hmm"
},
ObjectMeta
:
api
.
ObjectMeta
{
Name
:
"hmm"
,
Namespace
:
"hmm"
},
},
},
},
},
list
:
func
(
lister
StoreToReplicationControllerLister
)
([]
*
api
.
ReplicationController
,
error
)
{
list
:
func
(
lister
StoreToReplicationControllerLister
)
([]
api
.
ReplicationController
,
error
)
{
return
lister
.
ReplicationControllers
(
api
.
NamespaceAll
)
.
List
(
labels
.
Set
{}
.
AsSelectorPreValidated
())
return
lister
.
ReplicationControllers
(
api
.
NamespaceAll
)
.
List
(
labels
.
Set
{}
.
AsSelectorPreValidated
())
},
},
outRCNames
:
sets
.
NewString
(
"hmm"
,
"foo"
),
outRCNames
:
sets
.
NewString
(
"hmm"
,
"foo"
),
...
@@ -158,7 +158,7 @@ func TestStoreToReplicationControllerLister(t *testing.T) {
...
@@ -158,7 +158,7 @@ func TestStoreToReplicationControllerLister(t *testing.T) {
ObjectMeta
:
api
.
ObjectMeta
{
Name
:
"hmm"
,
Namespace
:
"hmm"
},
ObjectMeta
:
api
.
ObjectMeta
{
Name
:
"hmm"
,
Namespace
:
"hmm"
},
},
},
},
},
list
:
func
(
lister
StoreToReplicationControllerLister
)
([]
*
api
.
ReplicationController
,
error
)
{
list
:
func
(
lister
StoreToReplicationControllerLister
)
([]
api
.
ReplicationController
,
error
)
{
return
lister
.
ReplicationControllers
(
"hmm"
)
.
List
(
labels
.
Set
{}
.
AsSelectorPreValidated
())
return
lister
.
ReplicationControllers
(
"hmm"
)
.
List
(
labels
.
Set
{}
.
AsSelectorPreValidated
())
},
},
outRCNames
:
sets
.
NewString
(
"hmm"
),
outRCNames
:
sets
.
NewString
(
"hmm"
),
...
@@ -168,8 +168,8 @@ func TestStoreToReplicationControllerLister(t *testing.T) {
...
@@ -168,8 +168,8 @@ func TestStoreToReplicationControllerLister(t *testing.T) {
inRCs
:
[]
*
api
.
ReplicationController
{
inRCs
:
[]
*
api
.
ReplicationController
{
{
ObjectMeta
:
api
.
ObjectMeta
{
Name
:
"basic"
}},
{
ObjectMeta
:
api
.
ObjectMeta
{
Name
:
"basic"
}},
},
},
list
:
func
(
lister
StoreToReplicationControllerLister
)
([]
*
api
.
ReplicationController
,
error
)
{
list
:
func
(
lister
StoreToReplicationControllerLister
)
([]
api
.
ReplicationController
,
error
)
{
return
lister
.
List
(
labels
.
Everything
()
)
return
lister
.
List
()
},
},
outRCNames
:
sets
.
NewString
(
"basic"
),
outRCNames
:
sets
.
NewString
(
"basic"
),
},
},
...
@@ -183,7 +183,7 @@ func TestStoreToReplicationControllerLister(t *testing.T) {
...
@@ -183,7 +183,7 @@ func TestStoreToReplicationControllerLister(t *testing.T) {
},
},
},
},
},
},
list
:
func
(
lister
StoreToReplicationControllerLister
)
([]
*
api
.
ReplicationController
,
error
)
{
list
:
func
(
lister
StoreToReplicationControllerLister
)
([]
api
.
ReplicationController
,
error
)
{
pod
:=
&
api
.
Pod
{
pod
:=
&
api
.
Pod
{
ObjectMeta
:
api
.
ObjectMeta
{
Name
:
"pod1"
,
Namespace
:
"ns"
},
ObjectMeta
:
api
.
ObjectMeta
{
Name
:
"pod1"
,
Namespace
:
"ns"
},
}
}
...
@@ -199,7 +199,7 @@ func TestStoreToReplicationControllerLister(t *testing.T) {
...
@@ -199,7 +199,7 @@ func TestStoreToReplicationControllerLister(t *testing.T) {
ObjectMeta
:
api
.
ObjectMeta
{
Name
:
"basic"
,
Namespace
:
"ns"
},
ObjectMeta
:
api
.
ObjectMeta
{
Name
:
"basic"
,
Namespace
:
"ns"
},
},
},
},
},
list
:
func
(
lister
StoreToReplicationControllerLister
)
([]
*
api
.
ReplicationController
,
error
)
{
list
:
func
(
lister
StoreToReplicationControllerLister
)
([]
api
.
ReplicationController
,
error
)
{
pod
:=
&
api
.
Pod
{
pod
:=
&
api
.
Pod
{
ObjectMeta
:
api
.
ObjectMeta
{
ObjectMeta
:
api
.
ObjectMeta
{
Name
:
"pod1"
,
Name
:
"pod1"
,
...
@@ -228,7 +228,7 @@ func TestStoreToReplicationControllerLister(t *testing.T) {
...
@@ -228,7 +228,7 @@ func TestStoreToReplicationControllerLister(t *testing.T) {
},
},
},
},
},
},
list
:
func
(
lister
StoreToReplicationControllerLister
)
([]
*
api
.
ReplicationController
,
error
)
{
list
:
func
(
lister
StoreToReplicationControllerLister
)
([]
api
.
ReplicationController
,
error
)
{
pod
:=
&
api
.
Pod
{
pod
:=
&
api
.
Pod
{
ObjectMeta
:
api
.
ObjectMeta
{
ObjectMeta
:
api
.
ObjectMeta
{
Name
:
"pod1"
,
Name
:
"pod1"
,
...
@@ -754,7 +754,7 @@ func TestStoreToPodLister(t *testing.T) {
...
@@ -754,7 +754,7 @@ func TestStoreToPodLister(t *testing.T) {
}
}
func
TestStoreToServiceLister
(
t
*
testing
.
T
)
{
func
TestStoreToServiceLister
(
t
*
testing
.
T
)
{
store
:=
New
Indexer
(
MetaNamespaceKeyFunc
,
Indexers
{
NamespaceIndex
:
MetaNamespaceIndexFunc
}
)
store
:=
New
Store
(
MetaNamespaceKeyFunc
)
store
.
Add
(
&
api
.
Service
{
store
.
Add
(
&
api
.
Service
{
ObjectMeta
:
api
.
ObjectMeta
{
Name
:
"foo"
},
ObjectMeta
:
api
.
ObjectMeta
{
Name
:
"foo"
},
Spec
:
api
.
ServiceSpec
{
Spec
:
api
.
ServiceSpec
{
...
...
pkg/controller/endpoint/endpoints_controller.go
View file @
9a6ed3d0
...
@@ -80,7 +80,7 @@ func NewEndpointController(podInformer cache.SharedIndexInformer, client *client
...
@@ -80,7 +80,7 @@ func NewEndpointController(podInformer cache.SharedIndexInformer, client *client
queue
:
workqueue
.
NewNamedRateLimitingQueue
(
workqueue
.
DefaultControllerRateLimiter
(),
"endpoint"
),
queue
:
workqueue
.
NewNamedRateLimitingQueue
(
workqueue
.
DefaultControllerRateLimiter
(),
"endpoint"
),
}
}
e
.
serviceStore
.
Indexer
,
e
.
serviceController
=
cache
.
NewIndexer
Informer
(
e
.
serviceStore
.
Store
,
e
.
serviceController
=
cache
.
New
Informer
(
&
cache
.
ListWatch
{
&
cache
.
ListWatch
{
ListFunc
:
func
(
options
api
.
ListOptions
)
(
runtime
.
Object
,
error
)
{
ListFunc
:
func
(
options
api
.
ListOptions
)
(
runtime
.
Object
,
error
)
{
return
e
.
client
.
Core
()
.
Services
(
api
.
NamespaceAll
)
.
List
(
options
)
return
e
.
client
.
Core
()
.
Services
(
api
.
NamespaceAll
)
.
List
(
options
)
...
@@ -99,7 +99,6 @@ func NewEndpointController(podInformer cache.SharedIndexInformer, client *client
...
@@ -99,7 +99,6 @@ func NewEndpointController(podInformer cache.SharedIndexInformer, client *client
},
},
DeleteFunc
:
e
.
enqueueService
,
DeleteFunc
:
e
.
enqueueService
,
},
},
cache
.
Indexers
{
cache
.
NamespaceIndex
:
cache
.
MetaNamespaceIndexFunc
},
)
)
podInformer
.
AddEventHandler
(
cache
.
ResourceEventHandlerFuncs
{
podInformer
.
AddEventHandler
(
cache
.
ResourceEventHandlerFuncs
{
...
@@ -336,7 +335,7 @@ func (e *EndpointController) syncService(key string) error {
...
@@ -336,7 +335,7 @@ func (e *EndpointController) syncService(key string) error {
glog
.
V
(
4
)
.
Infof
(
"Finished syncing service %q endpoints. (%v)"
,
key
,
time
.
Now
()
.
Sub
(
startTime
))
glog
.
V
(
4
)
.
Infof
(
"Finished syncing service %q endpoints. (%v)"
,
key
,
time
.
Now
()
.
Sub
(
startTime
))
}()
}()
obj
,
exists
,
err
:=
e
.
serviceStore
.
Indexer
.
GetByKey
(
key
)
obj
,
exists
,
err
:=
e
.
serviceStore
.
Store
.
GetByKey
(
key
)
if
err
!=
nil
||
!
exists
{
if
err
!=
nil
||
!
exists
{
// Delete the corresponding endpoint, as the service has been deleted.
// Delete the corresponding endpoint, as the service has been deleted.
// TODO: Please note that this will delete an endpoint when a
// TODO: Please note that this will delete an endpoint when a
...
...
pkg/controller/endpoint/endpoints_controller_test.go
View file @
9a6ed3d0
...
@@ -110,7 +110,7 @@ func TestSyncEndpointsItemsPreserveNoSelector(t *testing.T) {
...
@@ -110,7 +110,7 @@ func TestSyncEndpointsItemsPreserveNoSelector(t *testing.T) {
client
:=
clientset
.
NewForConfigOrDie
(
&
restclient
.
Config
{
Host
:
testServer
.
URL
,
ContentConfig
:
restclient
.
ContentConfig
{
GroupVersion
:
testapi
.
Default
.
GroupVersion
()}})
client
:=
clientset
.
NewForConfigOrDie
(
&
restclient
.
Config
{
Host
:
testServer
.
URL
,
ContentConfig
:
restclient
.
ContentConfig
{
GroupVersion
:
testapi
.
Default
.
GroupVersion
()}})
endpoints
:=
NewEndpointControllerFromClient
(
client
,
controller
.
NoResyncPeriodFunc
)
endpoints
:=
NewEndpointControllerFromClient
(
client
,
controller
.
NoResyncPeriodFunc
)
endpoints
.
podStoreSynced
=
alwaysReady
endpoints
.
podStoreSynced
=
alwaysReady
endpoints
.
serviceStore
.
Indexer
.
Add
(
&
api
.
Service
{
endpoints
.
serviceStore
.
Store
.
Add
(
&
api
.
Service
{
ObjectMeta
:
api
.
ObjectMeta
{
Name
:
"foo"
,
Namespace
:
ns
},
ObjectMeta
:
api
.
ObjectMeta
{
Name
:
"foo"
,
Namespace
:
ns
},
Spec
:
api
.
ServiceSpec
{
Ports
:
[]
api
.
ServicePort
{{
Port
:
80
}}},
Spec
:
api
.
ServiceSpec
{
Ports
:
[]
api
.
ServicePort
{{
Port
:
80
}}},
})
})
...
@@ -174,7 +174,7 @@ func TestSyncEndpointsProtocolTCP(t *testing.T) {
...
@@ -174,7 +174,7 @@ func TestSyncEndpointsProtocolTCP(t *testing.T) {
endpoints
.
podStoreSynced
=
alwaysReady
endpoints
.
podStoreSynced
=
alwaysReady
addPods
(
endpoints
.
podStore
.
Indexer
,
ns
,
1
,
1
,
0
)
addPods
(
endpoints
.
podStore
.
Indexer
,
ns
,
1
,
1
,
0
)
endpoints
.
serviceStore
.
Indexer
.
Add
(
&
api
.
Service
{
endpoints
.
serviceStore
.
Store
.
Add
(
&
api
.
Service
{
ObjectMeta
:
api
.
ObjectMeta
{
Name
:
"foo"
,
Namespace
:
ns
},
ObjectMeta
:
api
.
ObjectMeta
{
Name
:
"foo"
,
Namespace
:
ns
},
Spec
:
api
.
ServiceSpec
{
Spec
:
api
.
ServiceSpec
{
Selector
:
map
[
string
]
string
{},
Selector
:
map
[
string
]
string
{},
...
@@ -216,7 +216,7 @@ func TestSyncEndpointsProtocolUDP(t *testing.T) {
...
@@ -216,7 +216,7 @@ func TestSyncEndpointsProtocolUDP(t *testing.T) {
endpoints
:=
NewEndpointControllerFromClient
(
client
,
controller
.
NoResyncPeriodFunc
)
endpoints
:=
NewEndpointControllerFromClient
(
client
,
controller
.
NoResyncPeriodFunc
)
endpoints
.
podStoreSynced
=
alwaysReady
endpoints
.
podStoreSynced
=
alwaysReady
addPods
(
endpoints
.
podStore
.
Indexer
,
ns
,
1
,
1
,
0
)
addPods
(
endpoints
.
podStore
.
Indexer
,
ns
,
1
,
1
,
0
)
endpoints
.
serviceStore
.
Indexer
.
Add
(
&
api
.
Service
{
endpoints
.
serviceStore
.
Store
.
Add
(
&
api
.
Service
{
ObjectMeta
:
api
.
ObjectMeta
{
Name
:
"foo"
,
Namespace
:
ns
},
ObjectMeta
:
api
.
ObjectMeta
{
Name
:
"foo"
,
Namespace
:
ns
},
Spec
:
api
.
ServiceSpec
{
Spec
:
api
.
ServiceSpec
{
Selector
:
map
[
string
]
string
{},
Selector
:
map
[
string
]
string
{},
...
@@ -255,7 +255,7 @@ func TestSyncEndpointsItemsEmptySelectorSelectsAll(t *testing.T) {
...
@@ -255,7 +255,7 @@ func TestSyncEndpointsItemsEmptySelectorSelectsAll(t *testing.T) {
endpoints
:=
NewEndpointControllerFromClient
(
client
,
controller
.
NoResyncPeriodFunc
)
endpoints
:=
NewEndpointControllerFromClient
(
client
,
controller
.
NoResyncPeriodFunc
)
endpoints
.
podStoreSynced
=
alwaysReady
endpoints
.
podStoreSynced
=
alwaysReady
addPods
(
endpoints
.
podStore
.
Indexer
,
ns
,
1
,
1
,
0
)
addPods
(
endpoints
.
podStore
.
Indexer
,
ns
,
1
,
1
,
0
)
endpoints
.
serviceStore
.
Indexer
.
Add
(
&
api
.
Service
{
endpoints
.
serviceStore
.
Store
.
Add
(
&
api
.
Service
{
ObjectMeta
:
api
.
ObjectMeta
{
Name
:
"foo"
,
Namespace
:
ns
},
ObjectMeta
:
api
.
ObjectMeta
{
Name
:
"foo"
,
Namespace
:
ns
},
Spec
:
api
.
ServiceSpec
{
Spec
:
api
.
ServiceSpec
{
Selector
:
map
[
string
]
string
{},
Selector
:
map
[
string
]
string
{},
...
@@ -293,7 +293,7 @@ func TestSyncEndpointsItemsEmptySelectorSelectsAllNotReady(t *testing.T) {
...
@@ -293,7 +293,7 @@ func TestSyncEndpointsItemsEmptySelectorSelectsAllNotReady(t *testing.T) {
endpoints
:=
NewEndpointControllerFromClient
(
client
,
controller
.
NoResyncPeriodFunc
)
endpoints
:=
NewEndpointControllerFromClient
(
client
,
controller
.
NoResyncPeriodFunc
)
endpoints
.
podStoreSynced
=
alwaysReady
endpoints
.
podStoreSynced
=
alwaysReady
addPods
(
endpoints
.
podStore
.
Indexer
,
ns
,
0
,
1
,
1
)
addPods
(
endpoints
.
podStore
.
Indexer
,
ns
,
0
,
1
,
1
)
endpoints
.
serviceStore
.
Indexer
.
Add
(
&
api
.
Service
{
endpoints
.
serviceStore
.
Store
.
Add
(
&
api
.
Service
{
ObjectMeta
:
api
.
ObjectMeta
{
Name
:
"foo"
,
Namespace
:
ns
},
ObjectMeta
:
api
.
ObjectMeta
{
Name
:
"foo"
,
Namespace
:
ns
},
Spec
:
api
.
ServiceSpec
{
Spec
:
api
.
ServiceSpec
{
Selector
:
map
[
string
]
string
{},
Selector
:
map
[
string
]
string
{},
...
@@ -331,7 +331,7 @@ func TestSyncEndpointsItemsEmptySelectorSelectsAllMixed(t *testing.T) {
...
@@ -331,7 +331,7 @@ func TestSyncEndpointsItemsEmptySelectorSelectsAllMixed(t *testing.T) {
endpoints
:=
NewEndpointControllerFromClient
(
client
,
controller
.
NoResyncPeriodFunc
)
endpoints
:=
NewEndpointControllerFromClient
(
client
,
controller
.
NoResyncPeriodFunc
)
endpoints
.
podStoreSynced
=
alwaysReady
endpoints
.
podStoreSynced
=
alwaysReady
addPods
(
endpoints
.
podStore
.
Indexer
,
ns
,
1
,
1
,
1
)
addPods
(
endpoints
.
podStore
.
Indexer
,
ns
,
1
,
1
,
1
)
endpoints
.
serviceStore
.
Indexer
.
Add
(
&
api
.
Service
{
endpoints
.
serviceStore
.
Store
.
Add
(
&
api
.
Service
{
ObjectMeta
:
api
.
ObjectMeta
{
Name
:
"foo"
,
Namespace
:
ns
},
ObjectMeta
:
api
.
ObjectMeta
{
Name
:
"foo"
,
Namespace
:
ns
},
Spec
:
api
.
ServiceSpec
{
Spec
:
api
.
ServiceSpec
{
Selector
:
map
[
string
]
string
{},
Selector
:
map
[
string
]
string
{},
...
@@ -373,7 +373,7 @@ func TestSyncEndpointsItemsPreexisting(t *testing.T) {
...
@@ -373,7 +373,7 @@ func TestSyncEndpointsItemsPreexisting(t *testing.T) {
endpoints
:=
NewEndpointControllerFromClient
(
client
,
controller
.
NoResyncPeriodFunc
)
endpoints
:=
NewEndpointControllerFromClient
(
client
,
controller
.
NoResyncPeriodFunc
)
endpoints
.
podStoreSynced
=
alwaysReady
endpoints
.
podStoreSynced
=
alwaysReady
addPods
(
endpoints
.
podStore
.
Indexer
,
ns
,
1
,
1
,
0
)
addPods
(
endpoints
.
podStore
.
Indexer
,
ns
,
1
,
1
,
0
)
endpoints
.
serviceStore
.
Indexer
.
Add
(
&
api
.
Service
{
endpoints
.
serviceStore
.
Store
.
Add
(
&
api
.
Service
{
ObjectMeta
:
api
.
ObjectMeta
{
Name
:
"foo"
,
Namespace
:
ns
},
ObjectMeta
:
api
.
ObjectMeta
{
Name
:
"foo"
,
Namespace
:
ns
},
Spec
:
api
.
ServiceSpec
{
Spec
:
api
.
ServiceSpec
{
Selector
:
map
[
string
]
string
{
"foo"
:
"bar"
},
Selector
:
map
[
string
]
string
{
"foo"
:
"bar"
},
...
@@ -414,7 +414,7 @@ func TestSyncEndpointsItemsPreexistingIdentical(t *testing.T) {
...
@@ -414,7 +414,7 @@ func TestSyncEndpointsItemsPreexistingIdentical(t *testing.T) {
endpoints
:=
NewEndpointControllerFromClient
(
client
,
controller
.
NoResyncPeriodFunc
)
endpoints
:=
NewEndpointControllerFromClient
(
client
,
controller
.
NoResyncPeriodFunc
)
endpoints
.
podStoreSynced
=
alwaysReady
endpoints
.
podStoreSynced
=
alwaysReady
addPods
(
endpoints
.
podStore
.
Indexer
,
api
.
NamespaceDefault
,
1
,
1
,
0
)
addPods
(
endpoints
.
podStore
.
Indexer
,
api
.
NamespaceDefault
,
1
,
1
,
0
)
endpoints
.
serviceStore
.
Indexer
.
Add
(
&
api
.
Service
{
endpoints
.
serviceStore
.
Store
.
Add
(
&
api
.
Service
{
ObjectMeta
:
api
.
ObjectMeta
{
Name
:
"foo"
,
Namespace
:
api
.
NamespaceDefault
},
ObjectMeta
:
api
.
ObjectMeta
{
Name
:
"foo"
,
Namespace
:
api
.
NamespaceDefault
},
Spec
:
api
.
ServiceSpec
{
Spec
:
api
.
ServiceSpec
{
Selector
:
map
[
string
]
string
{
"foo"
:
"bar"
},
Selector
:
map
[
string
]
string
{
"foo"
:
"bar"
},
...
@@ -435,7 +435,7 @@ func TestSyncEndpointsItems(t *testing.T) {
...
@@ -435,7 +435,7 @@ func TestSyncEndpointsItems(t *testing.T) {
endpoints
.
podStoreSynced
=
alwaysReady
endpoints
.
podStoreSynced
=
alwaysReady
addPods
(
endpoints
.
podStore
.
Indexer
,
ns
,
3
,
2
,
0
)
addPods
(
endpoints
.
podStore
.
Indexer
,
ns
,
3
,
2
,
0
)
addPods
(
endpoints
.
podStore
.
Indexer
,
"blah"
,
5
,
2
,
0
)
// make sure these aren't found!
addPods
(
endpoints
.
podStore
.
Indexer
,
"blah"
,
5
,
2
,
0
)
// make sure these aren't found!
endpoints
.
serviceStore
.
Indexer
.
Add
(
&
api
.
Service
{
endpoints
.
serviceStore
.
Store
.
Add
(
&
api
.
Service
{
ObjectMeta
:
api
.
ObjectMeta
{
Name
:
"foo"
,
Namespace
:
ns
},
ObjectMeta
:
api
.
ObjectMeta
{
Name
:
"foo"
,
Namespace
:
ns
},
Spec
:
api
.
ServiceSpec
{
Spec
:
api
.
ServiceSpec
{
Selector
:
map
[
string
]
string
{
"foo"
:
"bar"
},
Selector
:
map
[
string
]
string
{
"foo"
:
"bar"
},
...
@@ -478,7 +478,7 @@ func TestSyncEndpointsItemsWithLabels(t *testing.T) {
...
@@ -478,7 +478,7 @@ func TestSyncEndpointsItemsWithLabels(t *testing.T) {
endpoints
.
podStoreSynced
=
alwaysReady
endpoints
.
podStoreSynced
=
alwaysReady
addPods
(
endpoints
.
podStore
.
Indexer
,
ns
,
3
,
2
,
0
)
addPods
(
endpoints
.
podStore
.
Indexer
,
ns
,
3
,
2
,
0
)
serviceLabels
:=
map
[
string
]
string
{
"foo"
:
"bar"
}
serviceLabels
:=
map
[
string
]
string
{
"foo"
:
"bar"
}
endpoints
.
serviceStore
.
Indexer
.
Add
(
&
api
.
Service
{
endpoints
.
serviceStore
.
Store
.
Add
(
&
api
.
Service
{
ObjectMeta
:
api
.
ObjectMeta
{
ObjectMeta
:
api
.
ObjectMeta
{
Name
:
"foo"
,
Name
:
"foo"
,
Namespace
:
ns
,
Namespace
:
ns
,
...
@@ -539,7 +539,7 @@ func TestSyncEndpointsItemsPreexistingLabelsChange(t *testing.T) {
...
@@ -539,7 +539,7 @@ func TestSyncEndpointsItemsPreexistingLabelsChange(t *testing.T) {
endpoints
.
podStoreSynced
=
alwaysReady
endpoints
.
podStoreSynced
=
alwaysReady
addPods
(
endpoints
.
podStore
.
Indexer
,
ns
,
1
,
1
,
0
)
addPods
(
endpoints
.
podStore
.
Indexer
,
ns
,
1
,
1
,
0
)
serviceLabels
:=
map
[
string
]
string
{
"baz"
:
"blah"
}
serviceLabels
:=
map
[
string
]
string
{
"baz"
:
"blah"
}
endpoints
.
serviceStore
.
Indexer
.
Add
(
&
api
.
Service
{
endpoints
.
serviceStore
.
Store
.
Add
(
&
api
.
Service
{
ObjectMeta
:
api
.
ObjectMeta
{
ObjectMeta
:
api
.
ObjectMeta
{
Name
:
"foo"
,
Name
:
"foo"
,
Namespace
:
ns
,
Namespace
:
ns
,
...
...
pkg/controller/replication/replication_controller.go
View file @
9a6ed3d0
...
@@ -269,16 +269,16 @@ func (rm *ReplicationManager) getPodController(pod *api.Pod) *api.ReplicationCon
...
@@ -269,16 +269,16 @@ func (rm *ReplicationManager) getPodController(pod *api.Pod) *api.ReplicationCon
}
}
// update lookup cache
// update lookup cache
rm
.
lookupCache
.
Update
(
pod
,
controllers
[
0
])
rm
.
lookupCache
.
Update
(
pod
,
&
controllers
[
0
])
return
controllers
[
0
]
return
&
controllers
[
0
]
}
}
// isCacheValid check if the cache is valid
// isCacheValid check if the cache is valid
func
(
rm
*
ReplicationManager
)
isCacheValid
(
pod
*
api
.
Pod
,
cachedRC
*
api
.
ReplicationController
)
bool
{
func
(
rm
*
ReplicationManager
)
isCacheValid
(
pod
*
api
.
Pod
,
cachedRC
*
api
.
ReplicationController
)
bool
{
_
,
err
:=
rm
.
rcStore
.
ReplicationControllers
(
cachedRC
.
Namespace
)
.
Get
(
cachedRC
.
Name
)
exists
,
err
:=
rm
.
rcStore
.
Exists
(
cachedRC
)
// rc has been deleted or updated, cache is invalid
// rc has been deleted or updated, cache is invalid
if
err
!=
nil
||
!
isControllerMatch
(
pod
,
cachedRC
)
{
if
err
!=
nil
||
!
exists
||
!
isControllerMatch
(
pod
,
cachedRC
)
{
return
false
return
false
}
}
return
true
return
true
...
...
pkg/controller/replication/replication_controller_utils.go
View file @
9a6ed3d0
...
@@ -71,7 +71,7 @@ func updateReplicaCount(rcClient unversionedcore.ReplicationControllerInterface,
...
@@ -71,7 +71,7 @@ func updateReplicaCount(rcClient unversionedcore.ReplicationControllerInterface,
}
}
// OverlappingControllers sorts a list of controllers by creation timestamp, using their names as a tie breaker.
// OverlappingControllers sorts a list of controllers by creation timestamp, using their names as a tie breaker.
type
OverlappingControllers
[]
*
api
.
ReplicationController
type
OverlappingControllers
[]
api
.
ReplicationController
func
(
o
OverlappingControllers
)
Len
()
int
{
return
len
(
o
)
}
func
(
o
OverlappingControllers
)
Len
()
int
{
return
len
(
o
)
}
func
(
o
OverlappingControllers
)
Swap
(
i
,
j
int
)
{
o
[
i
],
o
[
j
]
=
o
[
j
],
o
[
i
]
}
func
(
o
OverlappingControllers
)
Swap
(
i
,
j
int
)
{
o
[
i
],
o
[
j
]
=
o
[
j
],
o
[
i
]
}
...
...
pkg/controller/service/servicecontroller.go
View file @
9a6ed3d0
...
@@ -119,7 +119,7 @@ func New(cloud cloudprovider.Interface, kubeClient clientset.Interface, clusterN
...
@@ -119,7 +119,7 @@ func New(cloud cloudprovider.Interface, kubeClient clientset.Interface, clusterN
},
},
workingQueue
:
workqueue
.
NewDelayingQueue
(),
workingQueue
:
workqueue
.
NewDelayingQueue
(),
}
}
s
.
serviceStore
.
Indexer
,
s
.
serviceController
=
cache
.
NewIndexer
Informer
(
s
.
serviceStore
.
Store
,
s
.
serviceController
=
cache
.
New
Informer
(
&
cache
.
ListWatch
{
&
cache
.
ListWatch
{
ListFunc
:
func
(
options
api
.
ListOptions
)
(
pkg_runtime
.
Object
,
error
)
{
ListFunc
:
func
(
options
api
.
ListOptions
)
(
pkg_runtime
.
Object
,
error
)
{
return
s
.
kubeClient
.
Core
()
.
Services
(
api
.
NamespaceAll
)
.
List
(
options
)
return
s
.
kubeClient
.
Core
()
.
Services
(
api
.
NamespaceAll
)
.
List
(
options
)
...
@@ -141,7 +141,6 @@ func New(cloud cloudprovider.Interface, kubeClient clientset.Interface, clusterN
...
@@ -141,7 +141,6 @@ func New(cloud cloudprovider.Interface, kubeClient clientset.Interface, clusterN
},
},
DeleteFunc
:
s
.
enqueueService
,
DeleteFunc
:
s
.
enqueueService
,
},
},
cache
.
Indexers
{
cache
.
NamespaceIndex
:
cache
.
MetaNamespaceIndexFunc
},
)
)
if
err
:=
s
.
init
();
err
!=
nil
{
if
err
:=
s
.
init
();
err
!=
nil
{
return
nil
,
err
return
nil
,
err
...
@@ -725,7 +724,7 @@ func (s *ServiceController) syncService(key string) error {
...
@@ -725,7 +724,7 @@ func (s *ServiceController) syncService(key string) error {
glog
.
V
(
4
)
.
Infof
(
"Finished syncing service %q (%v)"
,
key
,
time
.
Now
()
.
Sub
(
startTime
))
glog
.
V
(
4
)
.
Infof
(
"Finished syncing service %q (%v)"
,
key
,
time
.
Now
()
.
Sub
(
startTime
))
}()
}()
// obj holds the latest service info from apiserver
// obj holds the latest service info from apiserver
obj
,
exists
,
err
:=
s
.
serviceStore
.
Indexer
.
GetByKey
(
key
)
obj
,
exists
,
err
:=
s
.
serviceStore
.
Store
.
GetByKey
(
key
)
if
err
!=
nil
{
if
err
!=
nil
{
glog
.
Infof
(
"Unable to retrieve service %v from store: %v"
,
key
,
err
)
glog
.
Infof
(
"Unable to retrieve service %v from store: %v"
,
key
,
err
)
s
.
workingQueue
.
Add
(
key
)
s
.
workingQueue
.
Add
(
key
)
...
...
pkg/kubelet/envvars/envvars.go
View file @
9a6ed3d0
...
@@ -27,10 +27,10 @@ import (
...
@@ -27,10 +27,10 @@ import (
// FromServices builds environment variables that a container is started with,
// FromServices builds environment variables that a container is started with,
// which tell the container where to find the services it may need, which are
// which tell the container where to find the services it may need, which are
// provided as an argument.
// provided as an argument.
func
FromServices
(
services
[]
*
api
.
Service
)
[]
api
.
EnvVar
{
func
FromServices
(
services
*
api
.
ServiceList
)
[]
api
.
EnvVar
{
var
result
[]
api
.
EnvVar
var
result
[]
api
.
EnvVar
for
i
:=
range
services
{
for
i
:=
range
services
.
Items
{
service
:=
service
s
[
i
]
service
:=
&
services
.
Item
s
[
i
]
// ignore services where ClusterIP is "None" or empty
// ignore services where ClusterIP is "None" or empty
// the services passed to this method should be pre-filtered
// the services passed to this method should be pre-filtered
...
...
pkg/kubelet/envvars/envvars_test.go
View file @
9a6ed3d0
...
@@ -25,7 +25,8 @@ import (
...
@@ -25,7 +25,8 @@ import (
)
)
func
TestFromServices
(
t
*
testing
.
T
)
{
func
TestFromServices
(
t
*
testing
.
T
)
{
sl
:=
[]
*
api
.
Service
{
sl
:=
api
.
ServiceList
{
Items
:
[]
api
.
Service
{
{
{
ObjectMeta
:
api
.
ObjectMeta
{
Name
:
"foo-bar"
},
ObjectMeta
:
api
.
ObjectMeta
{
Name
:
"foo-bar"
},
Spec
:
api
.
ServiceSpec
{
Spec
:
api
.
ServiceSpec
{
...
@@ -78,8 +79,9 @@ func TestFromServices(t *testing.T) {
...
@@ -78,8 +79,9 @@ func TestFromServices(t *testing.T) {
},
},
},
},
},
},
},
}
}
vars
:=
envvars
.
FromServices
(
sl
)
vars
:=
envvars
.
FromServices
(
&
sl
)
expected
:=
[]
api
.
EnvVar
{
expected
:=
[]
api
.
EnvVar
{
{
Name
:
"FOO_BAR_SERVICE_HOST"
,
Value
:
"1.2.3.4"
},
{
Name
:
"FOO_BAR_SERVICE_HOST"
,
Value
:
"1.2.3.4"
},
{
Name
:
"FOO_BAR_SERVICE_PORT"
,
Value
:
"8080"
},
{
Name
:
"FOO_BAR_SERVICE_PORT"
,
Value
:
"8080"
},
...
...
pkg/kubelet/kubelet.go
View file @
9a6ed3d0
...
@@ -77,7 +77,6 @@ import (
...
@@ -77,7 +77,6 @@ import (
"k8s.io/kubernetes/pkg/kubelet/util/queue"
"k8s.io/kubernetes/pkg/kubelet/util/queue"
"k8s.io/kubernetes/pkg/kubelet/util/sliceutils"
"k8s.io/kubernetes/pkg/kubelet/util/sliceutils"
"k8s.io/kubernetes/pkg/kubelet/volumemanager"
"k8s.io/kubernetes/pkg/kubelet/volumemanager"
"k8s.io/kubernetes/pkg/labels"
"k8s.io/kubernetes/pkg/runtime"
"k8s.io/kubernetes/pkg/runtime"
"k8s.io/kubernetes/pkg/security/apparmor"
"k8s.io/kubernetes/pkg/security/apparmor"
"k8s.io/kubernetes/pkg/types"
"k8s.io/kubernetes/pkg/types"
...
@@ -372,7 +371,7 @@ func NewMainKubelet(kubeCfg *componentconfig.KubeletConfiguration, kubeDeps *Kub
...
@@ -372,7 +371,7 @@ func NewMainKubelet(kubeCfg *componentconfig.KubeletConfiguration, kubeDeps *Kub
dockerExecHandler
=
&
dockertools
.
NativeExecHandler
{}
dockerExecHandler
=
&
dockertools
.
NativeExecHandler
{}
}
}
serviceStore
:=
cache
.
New
Indexer
(
cache
.
MetaNamespaceKeyFunc
,
cache
.
Indexers
{
cache
.
NamespaceIndex
:
cache
.
MetaNamespaceIndexFunc
}
)
serviceStore
:=
cache
.
New
Store
(
cache
.
MetaNamespaceKeyFunc
)
if
kubeClient
!=
nil
{
if
kubeClient
!=
nil
{
// TODO: cache.NewListWatchFromClient is limited as it takes a client implementation rather
// TODO: cache.NewListWatchFromClient is limited as it takes a client implementation rather
// than an interface. There is no way to construct a list+watcher using resource name.
// than an interface. There is no way to construct a list+watcher using resource name.
...
@@ -386,7 +385,7 @@ func NewMainKubelet(kubeCfg *componentconfig.KubeletConfiguration, kubeDeps *Kub
...
@@ -386,7 +385,7 @@ func NewMainKubelet(kubeCfg *componentconfig.KubeletConfiguration, kubeDeps *Kub
}
}
cache
.
NewReflector
(
listWatch
,
&
api
.
Service
{},
serviceStore
,
0
)
.
Run
()
cache
.
NewReflector
(
listWatch
,
&
api
.
Service
{},
serviceStore
,
0
)
.
Run
()
}
}
serviceLister
:=
&
cache
.
StoreToServiceLister
{
Indexer
:
serviceStore
}
serviceLister
:=
&
cache
.
StoreToServiceLister
{
Store
:
serviceStore
}
nodeStore
:=
cache
.
NewStore
(
cache
.
MetaNamespaceKeyFunc
)
nodeStore
:=
cache
.
NewStore
(
cache
.
MetaNamespaceKeyFunc
)
if
kubeClient
!=
nil
{
if
kubeClient
!=
nil
{
...
@@ -778,7 +777,7 @@ func NewMainKubelet(kubeCfg *componentconfig.KubeletConfiguration, kubeDeps *Kub
...
@@ -778,7 +777,7 @@ func NewMainKubelet(kubeCfg *componentconfig.KubeletConfiguration, kubeDeps *Kub
}
}
type
serviceLister
interface
{
type
serviceLister
interface
{
List
(
labels
.
Selector
)
([]
*
api
.
Service
,
error
)
List
(
)
(
api
.
ServiceList
,
error
)
}
}
type
nodeLister
interface
{
type
nodeLister
interface
{
...
@@ -1448,7 +1447,7 @@ var masterServices = sets.NewString("kubernetes")
...
@@ -1448,7 +1447,7 @@ var masterServices = sets.NewString("kubernetes")
// pod in namespace ns should see.
// pod in namespace ns should see.
func
(
kl
*
Kubelet
)
getServiceEnvVarMap
(
ns
string
)
(
map
[
string
]
string
,
error
)
{
func
(
kl
*
Kubelet
)
getServiceEnvVarMap
(
ns
string
)
(
map
[
string
]
string
,
error
)
{
var
(
var
(
serviceMap
=
make
(
map
[
string
]
*
api
.
Service
)
serviceMap
=
make
(
map
[
string
]
api
.
Service
)
m
=
make
(
map
[
string
]
string
)
m
=
make
(
map
[
string
]
string
)
)
)
...
@@ -1458,16 +1457,15 @@ func (kl *Kubelet) getServiceEnvVarMap(ns string) (map[string]string, error) {
...
@@ -1458,16 +1457,15 @@ func (kl *Kubelet) getServiceEnvVarMap(ns string) (map[string]string, error) {
// Kubelets without masters (e.g. plain GCE ContainerVM) don't set env vars.
// Kubelets without masters (e.g. plain GCE ContainerVM) don't set env vars.
return
m
,
nil
return
m
,
nil
}
}
services
,
err
:=
kl
.
serviceLister
.
List
(
labels
.
Everything
()
)
services
,
err
:=
kl
.
serviceLister
.
List
()
if
err
!=
nil
{
if
err
!=
nil
{
return
m
,
fmt
.
Errorf
(
"failed to list services when setting up env vars."
)
return
m
,
fmt
.
Errorf
(
"failed to list services when setting up env vars."
)
}
}
// project the services in namespace ns onto the master services
// project the services in namespace ns onto the master services
for
i
:=
range
services
{
for
_
,
service
:=
range
services
.
Items
{
service
:=
services
[
i
]
// ignore services where ClusterIP is "None" or empty
// ignore services where ClusterIP is "None" or empty
if
!
api
.
IsServiceIPSet
(
service
)
{
if
!
api
.
IsServiceIPSet
(
&
service
)
{
continue
continue
}
}
serviceName
:=
service
.
Name
serviceName
:=
service
.
Name
...
@@ -1487,13 +1485,12 @@ func (kl *Kubelet) getServiceEnvVarMap(ns string) (map[string]string, error) {
...
@@ -1487,13 +1485,12 @@ func (kl *Kubelet) getServiceEnvVarMap(ns string) (map[string]string, error) {
}
}
}
}
}
}
services
.
Items
=
[]
api
.
Service
{}
mappedServices
:=
[]
*
api
.
Service
{}
for
_
,
service
:=
range
serviceMap
{
for
key
:=
range
serviceMap
{
services
.
Items
=
append
(
services
.
Items
,
service
)
mappedServices
=
append
(
mappedServices
,
serviceMap
[
key
])
}
}
for
_
,
e
:=
range
envvars
.
FromServices
(
mappedS
ervices
)
{
for
_
,
e
:=
range
envvars
.
FromServices
(
&
s
ervices
)
{
m
[
e
.
Name
]
=
e
.
Value
m
[
e
.
Name
]
=
e
.
Value
}
}
return
m
,
nil
return
m
,
nil
...
...
pkg/kubelet/kubelet_test.go
View file @
9a6ed3d0
...
@@ -61,7 +61,6 @@ import (
...
@@ -61,7 +61,6 @@ import (
kubetypes
"k8s.io/kubernetes/pkg/kubelet/types"
kubetypes
"k8s.io/kubernetes/pkg/kubelet/types"
"k8s.io/kubernetes/pkg/kubelet/util/queue"
"k8s.io/kubernetes/pkg/kubelet/util/queue"
kubeletvolume
"k8s.io/kubernetes/pkg/kubelet/volumemanager"
kubeletvolume
"k8s.io/kubernetes/pkg/kubelet/volumemanager"
"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/util/clock"
"k8s.io/kubernetes/pkg/util/clock"
...
@@ -897,11 +896,13 @@ func TestDNSConfigurationParams(t *testing.T) {
...
@@ -897,11 +896,13 @@ func TestDNSConfigurationParams(t *testing.T) {
}
}
type
testServiceLister
struct
{
type
testServiceLister
struct
{
services
[]
*
api
.
Service
services
[]
api
.
Service
}
}
func
(
ls
testServiceLister
)
List
(
labels
.
Selector
)
([]
*
api
.
Service
,
error
)
{
func
(
ls
testServiceLister
)
List
()
(
api
.
ServiceList
,
error
)
{
return
ls
.
services
,
nil
return
api
.
ServiceList
{
Items
:
ls
.
services
,
},
nil
}
}
type
testNodeLister
struct
{
type
testNodeLister
struct
{
...
@@ -937,8 +938,8 @@ func (e envs) Swap(i, j int) { e[i], e[j] = e[j], e[i] }
...
@@ -937,8 +938,8 @@ func (e envs) Swap(i, j int) { e[i], e[j] = e[j], e[i] }
func
(
e
envs
)
Less
(
i
,
j
int
)
bool
{
return
e
[
i
]
.
Name
<
e
[
j
]
.
Name
}
func
(
e
envs
)
Less
(
i
,
j
int
)
bool
{
return
e
[
i
]
.
Name
<
e
[
j
]
.
Name
}
func
buildService
(
name
,
namespace
,
clusterIP
,
protocol
string
,
port
int
)
*
api
.
Service
{
func
buildService
(
name
,
namespace
,
clusterIP
,
protocol
string
,
port
int
)
api
.
Service
{
return
&
api
.
Service
{
return
api
.
Service
{
ObjectMeta
:
api
.
ObjectMeta
{
Name
:
name
,
Namespace
:
namespace
},
ObjectMeta
:
api
.
ObjectMeta
{
Name
:
name
,
Namespace
:
namespace
},
Spec
:
api
.
ServiceSpec
{
Spec
:
api
.
ServiceSpec
{
Ports
:
[]
api
.
ServicePort
{{
Ports
:
[]
api
.
ServicePort
{{
...
@@ -951,7 +952,7 @@ func buildService(name, namespace, clusterIP, protocol string, port int) *api.Se
...
@@ -951,7 +952,7 @@ func buildService(name, namespace, clusterIP, protocol string, port int) *api.Se
}
}
func
TestMakeEnvironmentVariables
(
t
*
testing
.
T
)
{
func
TestMakeEnvironmentVariables
(
t
*
testing
.
T
)
{
services
:=
[]
*
api
.
Service
{
services
:=
[]
api
.
Service
{
buildService
(
"kubernetes"
,
api
.
NamespaceDefault
,
"1.2.3.1"
,
"TCP"
,
8081
),
buildService
(
"kubernetes"
,
api
.
NamespaceDefault
,
"1.2.3.1"
,
"TCP"
,
8081
),
buildService
(
"test"
,
"test1"
,
"1.2.3.3"
,
"TCP"
,
8083
),
buildService
(
"test"
,
"test1"
,
"1.2.3.3"
,
"TCP"
,
8083
),
buildService
(
"kubernetes"
,
"test2"
,
"1.2.3.4"
,
"TCP"
,
8084
),
buildService
(
"kubernetes"
,
"test2"
,
"1.2.3.4"
,
"TCP"
,
8084
),
...
...
plugin/pkg/scheduler/algorithm/listers.go
View file @
9a6ed3d0
...
@@ -63,25 +63,24 @@ func (f FakePodLister) List(s labels.Selector) (selected []*api.Pod, err error)
...
@@ -63,25 +63,24 @@ func (f FakePodLister) List(s labels.Selector) (selected []*api.Pod, err error)
// ServiceLister interface represents anything that can produce a list of services; the list is consumed by a scheduler.
// ServiceLister interface represents anything that can produce a list of services; the list is consumed by a scheduler.
type
ServiceLister
interface
{
type
ServiceLister
interface
{
// Lists all the services
// Lists all the services
List
(
labels
.
Selector
)
([]
*
api
.
Service
,
error
)
List
(
)
(
api
.
ServiceList
,
error
)
// Gets the services for the given pod
// Gets the services for the given pod
GetPodServices
(
*
api
.
Pod
)
([]
*
api
.
Service
,
error
)
GetPodServices
(
*
api
.
Pod
)
([]
api
.
Service
,
error
)
}
}
// FakeServiceLister implements ServiceLister on []api.Service for test purposes.
// FakeServiceLister implements ServiceLister on []api.Service for test purposes.
type
FakeServiceLister
[]
*
api
.
Service
type
FakeServiceLister
[]
api
.
Service
// List returns api.ServiceList, the list of all services.
// List returns api.ServiceList, the list of all services.
func
(
f
FakeServiceLister
)
List
(
labels
.
Selector
)
([]
*
api
.
Service
,
error
)
{
func
(
f
FakeServiceLister
)
List
(
)
(
api
.
ServiceList
,
error
)
{
return
f
,
nil
return
api
.
ServiceList
{
Items
:
f
}
,
nil
}
}
// GetPodServices gets the services that have the selector that match the labels on the given pod
// GetPodServices gets the services that have the selector that match the labels on the given pod
func
(
f
FakeServiceLister
)
GetPodServices
(
pod
*
api
.
Pod
)
(
services
[]
*
api
.
Service
,
err
error
)
{
func
(
f
FakeServiceLister
)
GetPodServices
(
pod
*
api
.
Pod
)
(
services
[]
api
.
Service
,
err
error
)
{
var
selector
labels
.
Selector
var
selector
labels
.
Selector
for
i
:=
range
f
{
for
_
,
service
:=
range
f
{
service
:=
f
[
i
]
// consider only services that are in the same namespace as the pod
// consider only services that are in the same namespace as the pod
if
service
.
Namespace
!=
pod
.
Namespace
{
if
service
.
Namespace
!=
pod
.
Namespace
{
continue
continue
...
@@ -101,38 +100,37 @@ func (f FakeServiceLister) GetPodServices(pod *api.Pod) (services []*api.Service
...
@@ -101,38 +100,37 @@ func (f FakeServiceLister) GetPodServices(pod *api.Pod) (services []*api.Service
// ControllerLister interface represents anything that can produce a list of ReplicationController; the list is consumed by a scheduler.
// ControllerLister interface represents anything that can produce a list of ReplicationController; the list is consumed by a scheduler.
type
ControllerLister
interface
{
type
ControllerLister
interface
{
// Lists all the replication controllers
// Lists all the replication controllers
List
(
labels
.
Selector
)
([]
*
api
.
ReplicationController
,
error
)
List
(
)
([]
api
.
ReplicationController
,
error
)
// Gets the services for the given pod
// Gets the services for the given pod
GetPodControllers
(
*
api
.
Pod
)
([]
*
api
.
ReplicationController
,
error
)
GetPodControllers
(
*
api
.
Pod
)
([]
api
.
ReplicationController
,
error
)
}
}
// EmptyControllerLister implements ControllerLister on []api.ReplicationController returning empty data
// EmptyControllerLister implements ControllerLister on []api.ReplicationController returning empty data
type
EmptyControllerLister
struct
{}
type
EmptyControllerLister
struct
{}
// List returns nil
// List returns nil
func
(
f
EmptyControllerLister
)
List
(
labels
.
Selector
)
([]
*
api
.
ReplicationController
,
error
)
{
func
(
f
EmptyControllerLister
)
List
(
)
([]
api
.
ReplicationController
,
error
)
{
return
nil
,
nil
return
nil
,
nil
}
}
// GetPodControllers returns nil
// GetPodControllers returns nil
func
(
f
EmptyControllerLister
)
GetPodControllers
(
pod
*
api
.
Pod
)
(
controllers
[]
*
api
.
ReplicationController
,
err
error
)
{
func
(
f
EmptyControllerLister
)
GetPodControllers
(
pod
*
api
.
Pod
)
(
controllers
[]
api
.
ReplicationController
,
err
error
)
{
return
nil
,
nil
return
nil
,
nil
}
}
// FakeControllerLister implements ControllerLister on []api.ReplicationController for test purposes.
// FakeControllerLister implements ControllerLister on []api.ReplicationController for test purposes.
type
FakeControllerLister
[]
*
api
.
ReplicationController
type
FakeControllerLister
[]
api
.
ReplicationController
// List returns []api.ReplicationController, the list of all ReplicationControllers.
// List returns []api.ReplicationController, the list of all ReplicationControllers.
func
(
f
FakeControllerLister
)
List
(
labels
.
Selector
)
([]
*
api
.
ReplicationController
,
error
)
{
func
(
f
FakeControllerLister
)
List
(
)
([]
api
.
ReplicationController
,
error
)
{
return
f
,
nil
return
f
,
nil
}
}
// GetPodControllers gets the ReplicationControllers that have the selector that match the labels on the given pod
// GetPodControllers gets the ReplicationControllers that have the selector that match the labels on the given pod
func
(
f
FakeControllerLister
)
GetPodControllers
(
pod
*
api
.
Pod
)
(
controllers
[]
*
api
.
ReplicationController
,
err
error
)
{
func
(
f
FakeControllerLister
)
GetPodControllers
(
pod
*
api
.
Pod
)
(
controllers
[]
api
.
ReplicationController
,
err
error
)
{
var
selector
labels
.
Selector
var
selector
labels
.
Selector
for
i
:=
range
f
{
for
_
,
controller
:=
range
f
{
controller
:=
f
[
i
]
if
controller
.
Namespace
!=
pod
.
Namespace
{
if
controller
.
Namespace
!=
pod
.
Namespace
{
continue
continue
}
}
...
...
plugin/pkg/scheduler/algorithm/predicates/predicates.go
View file @
9a6ed3d0
...
@@ -670,7 +670,7 @@ func (s *ServiceAffinity) CheckServiceAffinity(pod *api.Pod, meta interface{}, n
...
@@ -670,7 +670,7 @@ func (s *ServiceAffinity) CheckServiceAffinity(pod *api.Pod, meta interface{}, n
// skip looking at other pods in the service if the current pod defines all the required affinity labels
// skip looking at other pods in the service if the current pod defines all the required affinity labels
if
!
labelsExist
{
if
!
labelsExist
{
services
,
err
:=
s
.
serviceLister
.
GetPodServices
(
pod
)
services
,
err
:=
s
.
serviceLister
.
GetPodServices
(
pod
)
if
err
==
nil
&&
len
(
services
)
>
0
{
if
err
==
nil
{
// just use the first service and get the other pods within the service
// just use the first service and get the other pods within the service
// TODO: a separate predicate can be created that tries to handle all services for the pod
// TODO: a separate predicate can be created that tries to handle all services for the pod
selector
:=
labels
.
SelectorFromSet
(
services
[
0
]
.
Spec
.
Selector
)
selector
:=
labels
.
SelectorFromSet
(
services
[
0
]
.
Spec
.
Selector
)
...
...
plugin/pkg/scheduler/algorithm/predicates/predicates_test.go
View file @
9a6ed3d0
...
@@ -1209,7 +1209,7 @@ func TestServiceAffinity(t *testing.T) {
...
@@ -1209,7 +1209,7 @@ func TestServiceAffinity(t *testing.T) {
tests
:=
[]
struct
{
tests
:=
[]
struct
{
pod
*
api
.
Pod
pod
*
api
.
Pod
pods
[]
*
api
.
Pod
pods
[]
*
api
.
Pod
services
[]
*
api
.
Service
services
[]
api
.
Service
node
*
api
.
Node
node
*
api
.
Node
labels
[]
string
labels
[]
string
fits
bool
fits
bool
...
@@ -1240,7 +1240,7 @@ func TestServiceAffinity(t *testing.T) {
...
@@ -1240,7 +1240,7 @@ func TestServiceAffinity(t *testing.T) {
pod
:
&
api
.
Pod
{
ObjectMeta
:
api
.
ObjectMeta
{
Labels
:
selector
}},
pod
:
&
api
.
Pod
{
ObjectMeta
:
api
.
ObjectMeta
{
Labels
:
selector
}},
pods
:
[]
*
api
.
Pod
{{
Spec
:
api
.
PodSpec
{
NodeName
:
"machine1"
},
ObjectMeta
:
api
.
ObjectMeta
{
Labels
:
selector
}}},
pods
:
[]
*
api
.
Pod
{{
Spec
:
api
.
PodSpec
{
NodeName
:
"machine1"
},
ObjectMeta
:
api
.
ObjectMeta
{
Labels
:
selector
}}},
node
:
&
node1
,
node
:
&
node1
,
services
:
[]
*
api
.
Service
{{
Spec
:
api
.
ServiceSpec
{
Selector
:
selector
}}},
services
:
[]
api
.
Service
{{
Spec
:
api
.
ServiceSpec
{
Selector
:
selector
}}},
fits
:
true
,
fits
:
true
,
labels
:
[]
string
{
"region"
},
labels
:
[]
string
{
"region"
},
test
:
"service pod on same node"
,
test
:
"service pod on same node"
,
...
@@ -1249,7 +1249,7 @@ func TestServiceAffinity(t *testing.T) {
...
@@ -1249,7 +1249,7 @@ func TestServiceAffinity(t *testing.T) {
pod
:
&
api
.
Pod
{
ObjectMeta
:
api
.
ObjectMeta
{
Labels
:
selector
}},
pod
:
&
api
.
Pod
{
ObjectMeta
:
api
.
ObjectMeta
{
Labels
:
selector
}},
pods
:
[]
*
api
.
Pod
{{
Spec
:
api
.
PodSpec
{
NodeName
:
"machine2"
},
ObjectMeta
:
api
.
ObjectMeta
{
Labels
:
selector
}}},
pods
:
[]
*
api
.
Pod
{{
Spec
:
api
.
PodSpec
{
NodeName
:
"machine2"
},
ObjectMeta
:
api
.
ObjectMeta
{
Labels
:
selector
}}},
node
:
&
node1
,
node
:
&
node1
,
services
:
[]
*
api
.
Service
{{
Spec
:
api
.
ServiceSpec
{
Selector
:
selector
}}},
services
:
[]
api
.
Service
{{
Spec
:
api
.
ServiceSpec
{
Selector
:
selector
}}},
fits
:
true
,
fits
:
true
,
labels
:
[]
string
{
"region"
},
labels
:
[]
string
{
"region"
},
test
:
"service pod on different node, region match"
,
test
:
"service pod on different node, region match"
,
...
@@ -1258,7 +1258,7 @@ func TestServiceAffinity(t *testing.T) {
...
@@ -1258,7 +1258,7 @@ func TestServiceAffinity(t *testing.T) {
pod
:
&
api
.
Pod
{
ObjectMeta
:
api
.
ObjectMeta
{
Labels
:
selector
}},
pod
:
&
api
.
Pod
{
ObjectMeta
:
api
.
ObjectMeta
{
Labels
:
selector
}},
pods
:
[]
*
api
.
Pod
{{
Spec
:
api
.
PodSpec
{
NodeName
:
"machine3"
},
ObjectMeta
:
api
.
ObjectMeta
{
Labels
:
selector
}}},
pods
:
[]
*
api
.
Pod
{{
Spec
:
api
.
PodSpec
{
NodeName
:
"machine3"
},
ObjectMeta
:
api
.
ObjectMeta
{
Labels
:
selector
}}},
node
:
&
node1
,
node
:
&
node1
,
services
:
[]
*
api
.
Service
{{
Spec
:
api
.
ServiceSpec
{
Selector
:
selector
}}},
services
:
[]
api
.
Service
{{
Spec
:
api
.
ServiceSpec
{
Selector
:
selector
}}},
fits
:
false
,
fits
:
false
,
labels
:
[]
string
{
"region"
},
labels
:
[]
string
{
"region"
},
test
:
"service pod on different node, region mismatch"
,
test
:
"service pod on different node, region mismatch"
,
...
@@ -1267,7 +1267,7 @@ func TestServiceAffinity(t *testing.T) {
...
@@ -1267,7 +1267,7 @@ func TestServiceAffinity(t *testing.T) {
pod
:
&
api
.
Pod
{
ObjectMeta
:
api
.
ObjectMeta
{
Labels
:
selector
,
Namespace
:
"ns1"
}},
pod
:
&
api
.
Pod
{
ObjectMeta
:
api
.
ObjectMeta
{
Labels
:
selector
,
Namespace
:
"ns1"
}},
pods
:
[]
*
api
.
Pod
{{
Spec
:
api
.
PodSpec
{
NodeName
:
"machine3"
},
ObjectMeta
:
api
.
ObjectMeta
{
Labels
:
selector
,
Namespace
:
"ns1"
}}},
pods
:
[]
*
api
.
Pod
{{
Spec
:
api
.
PodSpec
{
NodeName
:
"machine3"
},
ObjectMeta
:
api
.
ObjectMeta
{
Labels
:
selector
,
Namespace
:
"ns1"
}}},
node
:
&
node1
,
node
:
&
node1
,
services
:
[]
*
api
.
Service
{{
Spec
:
api
.
ServiceSpec
{
Selector
:
selector
},
ObjectMeta
:
api
.
ObjectMeta
{
Namespace
:
"ns2"
}}},
services
:
[]
api
.
Service
{{
Spec
:
api
.
ServiceSpec
{
Selector
:
selector
},
ObjectMeta
:
api
.
ObjectMeta
{
Namespace
:
"ns2"
}}},
fits
:
true
,
fits
:
true
,
labels
:
[]
string
{
"region"
},
labels
:
[]
string
{
"region"
},
test
:
"service in different namespace, region mismatch"
,
test
:
"service in different namespace, region mismatch"
,
...
@@ -1276,7 +1276,7 @@ func TestServiceAffinity(t *testing.T) {
...
@@ -1276,7 +1276,7 @@ func TestServiceAffinity(t *testing.T) {
pod
:
&
api
.
Pod
{
ObjectMeta
:
api
.
ObjectMeta
{
Labels
:
selector
,
Namespace
:
"ns1"
}},
pod
:
&
api
.
Pod
{
ObjectMeta
:
api
.
ObjectMeta
{
Labels
:
selector
,
Namespace
:
"ns1"
}},
pods
:
[]
*
api
.
Pod
{{
Spec
:
api
.
PodSpec
{
NodeName
:
"machine3"
},
ObjectMeta
:
api
.
ObjectMeta
{
Labels
:
selector
,
Namespace
:
"ns2"
}}},
pods
:
[]
*
api
.
Pod
{{
Spec
:
api
.
PodSpec
{
NodeName
:
"machine3"
},
ObjectMeta
:
api
.
ObjectMeta
{
Labels
:
selector
,
Namespace
:
"ns2"
}}},
node
:
&
node1
,
node
:
&
node1
,
services
:
[]
*
api
.
Service
{{
Spec
:
api
.
ServiceSpec
{
Selector
:
selector
},
ObjectMeta
:
api
.
ObjectMeta
{
Namespace
:
"ns1"
}}},
services
:
[]
api
.
Service
{{
Spec
:
api
.
ServiceSpec
{
Selector
:
selector
},
ObjectMeta
:
api
.
ObjectMeta
{
Namespace
:
"ns1"
}}},
fits
:
true
,
fits
:
true
,
labels
:
[]
string
{
"region"
},
labels
:
[]
string
{
"region"
},
test
:
"pod in different namespace, region mismatch"
,
test
:
"pod in different namespace, region mismatch"
,
...
@@ -1285,7 +1285,7 @@ func TestServiceAffinity(t *testing.T) {
...
@@ -1285,7 +1285,7 @@ func TestServiceAffinity(t *testing.T) {
pod
:
&
api
.
Pod
{
ObjectMeta
:
api
.
ObjectMeta
{
Labels
:
selector
,
Namespace
:
"ns1"
}},
pod
:
&
api
.
Pod
{
ObjectMeta
:
api
.
ObjectMeta
{
Labels
:
selector
,
Namespace
:
"ns1"
}},
pods
:
[]
*
api
.
Pod
{{
Spec
:
api
.
PodSpec
{
NodeName
:
"machine3"
},
ObjectMeta
:
api
.
ObjectMeta
{
Labels
:
selector
,
Namespace
:
"ns1"
}}},
pods
:
[]
*
api
.
Pod
{{
Spec
:
api
.
PodSpec
{
NodeName
:
"machine3"
},
ObjectMeta
:
api
.
ObjectMeta
{
Labels
:
selector
,
Namespace
:
"ns1"
}}},
node
:
&
node1
,
node
:
&
node1
,
services
:
[]
*
api
.
Service
{{
Spec
:
api
.
ServiceSpec
{
Selector
:
selector
},
ObjectMeta
:
api
.
ObjectMeta
{
Namespace
:
"ns1"
}}},
services
:
[]
api
.
Service
{{
Spec
:
api
.
ServiceSpec
{
Selector
:
selector
},
ObjectMeta
:
api
.
ObjectMeta
{
Namespace
:
"ns1"
}}},
fits
:
false
,
fits
:
false
,
labels
:
[]
string
{
"region"
},
labels
:
[]
string
{
"region"
},
test
:
"service and pod in same namespace, region mismatch"
,
test
:
"service and pod in same namespace, region mismatch"
,
...
@@ -1294,7 +1294,7 @@ func TestServiceAffinity(t *testing.T) {
...
@@ -1294,7 +1294,7 @@ func TestServiceAffinity(t *testing.T) {
pod
:
&
api
.
Pod
{
ObjectMeta
:
api
.
ObjectMeta
{
Labels
:
selector
}},
pod
:
&
api
.
Pod
{
ObjectMeta
:
api
.
ObjectMeta
{
Labels
:
selector
}},
pods
:
[]
*
api
.
Pod
{{
Spec
:
api
.
PodSpec
{
NodeName
:
"machine2"
},
ObjectMeta
:
api
.
ObjectMeta
{
Labels
:
selector
}}},
pods
:
[]
*
api
.
Pod
{{
Spec
:
api
.
PodSpec
{
NodeName
:
"machine2"
},
ObjectMeta
:
api
.
ObjectMeta
{
Labels
:
selector
}}},
node
:
&
node1
,
node
:
&
node1
,
services
:
[]
*
api
.
Service
{{
Spec
:
api
.
ServiceSpec
{
Selector
:
selector
}}},
services
:
[]
api
.
Service
{{
Spec
:
api
.
ServiceSpec
{
Selector
:
selector
}}},
fits
:
false
,
fits
:
false
,
labels
:
[]
string
{
"region"
,
"zone"
},
labels
:
[]
string
{
"region"
,
"zone"
},
test
:
"service pod on different node, multiple labels, not all match"
,
test
:
"service pod on different node, multiple labels, not all match"
,
...
@@ -1303,7 +1303,7 @@ func TestServiceAffinity(t *testing.T) {
...
@@ -1303,7 +1303,7 @@ func TestServiceAffinity(t *testing.T) {
pod
:
&
api
.
Pod
{
ObjectMeta
:
api
.
ObjectMeta
{
Labels
:
selector
}},
pod
:
&
api
.
Pod
{
ObjectMeta
:
api
.
ObjectMeta
{
Labels
:
selector
}},
pods
:
[]
*
api
.
Pod
{{
Spec
:
api
.
PodSpec
{
NodeName
:
"machine5"
},
ObjectMeta
:
api
.
ObjectMeta
{
Labels
:
selector
}}},
pods
:
[]
*
api
.
Pod
{{
Spec
:
api
.
PodSpec
{
NodeName
:
"machine5"
},
ObjectMeta
:
api
.
ObjectMeta
{
Labels
:
selector
}}},
node
:
&
node4
,
node
:
&
node4
,
services
:
[]
*
api
.
Service
{{
Spec
:
api
.
ServiceSpec
{
Selector
:
selector
}}},
services
:
[]
api
.
Service
{{
Spec
:
api
.
ServiceSpec
{
Selector
:
selector
}}},
fits
:
true
,
fits
:
true
,
labels
:
[]
string
{
"region"
,
"zone"
},
labels
:
[]
string
{
"region"
,
"zone"
},
test
:
"service pod on different node, multiple labels, all match"
,
test
:
"service pod on different node, multiple labels, all match"
,
...
...
plugin/pkg/scheduler/algorithm/priorities/selector_spreading.go
View file @
9a6ed3d0
...
@@ -194,7 +194,7 @@ func NewServiceAntiAffinityPriority(podLister algorithm.PodLister, serviceLister
...
@@ -194,7 +194,7 @@ func NewServiceAntiAffinityPriority(podLister algorithm.PodLister, serviceLister
// The label to be considered is provided to the struct (ServiceAntiAffinity).
// The label to be considered is provided to the struct (ServiceAntiAffinity).
func
(
s
*
ServiceAntiAffinity
)
CalculateAntiAffinityPriority
(
pod
*
api
.
Pod
,
nodeNameToInfo
map
[
string
]
*
schedulercache
.
NodeInfo
,
nodes
[]
*
api
.
Node
)
(
schedulerapi
.
HostPriorityList
,
error
)
{
func
(
s
*
ServiceAntiAffinity
)
CalculateAntiAffinityPriority
(
pod
*
api
.
Pod
,
nodeNameToInfo
map
[
string
]
*
schedulercache
.
NodeInfo
,
nodes
[]
*
api
.
Node
)
(
schedulerapi
.
HostPriorityList
,
error
)
{
var
nsServicePods
[]
*
api
.
Pod
var
nsServicePods
[]
*
api
.
Pod
if
services
,
err
:=
s
.
serviceLister
.
GetPodServices
(
pod
);
err
==
nil
&&
len
(
services
)
>
0
{
if
services
,
err
:=
s
.
serviceLister
.
GetPodServices
(
pod
);
err
==
nil
{
// just use the first service and get the other pods within the service
// just use the first service and get the other pods within the service
// TODO: a separate predicate can be created that tries to handle all services for the pod
// TODO: a separate predicate can be created that tries to handle all services for the pod
selector
:=
labels
.
SelectorFromSet
(
services
[
0
]
.
Spec
.
Selector
)
selector
:=
labels
.
SelectorFromSet
(
services
[
0
]
.
Spec
.
Selector
)
...
...
plugin/pkg/scheduler/algorithm/priorities/selector_spreading_test.go
View file @
9a6ed3d0
This diff is collapsed.
Click to expand it.
plugin/pkg/scheduler/factory/factory.go
View file @
9a6ed3d0
...
@@ -108,7 +108,7 @@ func NewConfigFactory(client *client.Client, schedulerName string, hardPodAffini
...
@@ -108,7 +108,7 @@ func NewConfigFactory(client *client.Client, schedulerName string, hardPodAffini
NodeLister
:
&
cache
.
StoreToNodeLister
{},
NodeLister
:
&
cache
.
StoreToNodeLister
{},
PVLister
:
&
cache
.
StoreToPVFetcher
{
Store
:
cache
.
NewStore
(
cache
.
MetaNamespaceKeyFunc
)},
PVLister
:
&
cache
.
StoreToPVFetcher
{
Store
:
cache
.
NewStore
(
cache
.
MetaNamespaceKeyFunc
)},
PVCLister
:
&
cache
.
StoreToPVCFetcher
{
Store
:
cache
.
NewStore
(
cache
.
MetaNamespaceKeyFunc
)},
PVCLister
:
&
cache
.
StoreToPVCFetcher
{
Store
:
cache
.
NewStore
(
cache
.
MetaNamespaceKeyFunc
)},
ServiceLister
:
&
cache
.
StoreToServiceLister
{
Indexer
:
cache
.
NewIndexer
(
cache
.
MetaNamespaceKeyFunc
,
cache
.
Indexers
{
cache
.
NamespaceIndex
:
cache
.
MetaNamespaceIndexFunc
}
)},
ServiceLister
:
&
cache
.
StoreToServiceLister
{
Store
:
cache
.
NewStore
(
cache
.
MetaNamespaceKeyFunc
)},
ControllerLister
:
&
cache
.
StoreToReplicationControllerLister
{
Indexer
:
cache
.
NewIndexer
(
cache
.
MetaNamespaceKeyFunc
,
cache
.
Indexers
{
cache
.
NamespaceIndex
:
cache
.
MetaNamespaceIndexFunc
})},
ControllerLister
:
&
cache
.
StoreToReplicationControllerLister
{
Indexer
:
cache
.
NewIndexer
(
cache
.
MetaNamespaceKeyFunc
,
cache
.
Indexers
{
cache
.
NamespaceIndex
:
cache
.
MetaNamespaceIndexFunc
})},
ReplicaSetLister
:
&
cache
.
StoreToReplicaSetLister
{
Store
:
cache
.
NewStore
(
cache
.
MetaNamespaceKeyFunc
)},
ReplicaSetLister
:
&
cache
.
StoreToReplicaSetLister
{
Store
:
cache
.
NewStore
(
cache
.
MetaNamespaceKeyFunc
)},
schedulerCache
:
schedulerCache
,
schedulerCache
:
schedulerCache
,
...
@@ -401,7 +401,7 @@ func (f *ConfigFactory) Run() {
...
@@ -401,7 +401,7 @@ func (f *ConfigFactory) Run() {
// Watch and cache all service objects. Scheduler needs to find all pods
// Watch and cache all service objects. Scheduler needs to find all pods
// created by the same services or ReplicationControllers/ReplicaSets, so that it can spread them correctly.
// created by the same services or ReplicationControllers/ReplicaSets, so that it can spread them correctly.
// Cache this locally.
// Cache this locally.
cache
.
NewReflector
(
f
.
createServiceLW
(),
&
api
.
Service
{},
f
.
ServiceLister
.
Indexer
,
0
)
.
RunUntil
(
f
.
StopEverything
)
cache
.
NewReflector
(
f
.
createServiceLW
(),
&
api
.
Service
{},
f
.
ServiceLister
.
Store
,
0
)
.
RunUntil
(
f
.
StopEverything
)
// Watch and cache all ReplicationController objects. Scheduler needs to find all pods
// Watch and cache all ReplicationController objects. Scheduler needs to find all pods
// created by the same services or ReplicationControllers/ReplicaSets, so that it can spread them correctly.
// created by the same services or ReplicationControllers/ReplicaSets, so that it can spread them correctly.
...
...
plugin/pkg/scheduler/generic_scheduler_test.go
View file @
9a6ed3d0
...
@@ -493,8 +493,8 @@ func TestZeroRequest(t *testing.T) {
...
@@ -493,8 +493,8 @@ func TestZeroRequest(t *testing.T) {
{
{
Function
:
algorithmpriorities
.
NewSelectorSpreadPriority
(
Function
:
algorithmpriorities
.
NewSelectorSpreadPriority
(
algorithm
.
FakePodLister
(
test
.
pods
),
algorithm
.
FakePodLister
(
test
.
pods
),
algorithm
.
FakeServiceLister
([]
*
api
.
Service
{}),
algorithm
.
FakeServiceLister
([]
api
.
Service
{}),
algorithm
.
FakeControllerLister
([]
*
api
.
ReplicationController
{}),
algorithm
.
FakeControllerLister
([]
api
.
ReplicationController
{}),
algorithm
.
FakeReplicaSetLister
([]
extensions
.
ReplicaSet
{})),
algorithm
.
FakeReplicaSetLister
([]
extensions
.
ReplicaSet
{})),
Weight
:
1
,
Weight
:
1
,
},
},
...
...
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment