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
25fe1e0d
Commit
25fe1e0d
authored
Jan 12, 2017
by
Klaus Ma
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
Made cache.Controller to be interface.
parent
31483bf5
Hide whitespace changes
Inline
Side-by-side
Showing
36 changed files
with
115 additions
and
110 deletions
+115
-110
clustercontroller.go
...on/pkg/federation-controller/cluster/clustercontroller.go
+1
-1
configmap_controller.go
...g/federation-controller/configmap/configmap_controller.go
+2
-2
daemonset_controller.go
...g/federation-controller/daemonset/daemonset_controller.go
+2
-2
deploymentcontroller.go
.../federation-controller/deployment/deploymentcontroller.go
+3
-3
ingress_controller.go
...n/pkg/federation-controller/ingress/ingress_controller.go
+3
-3
namespace_controller.go
...g/federation-controller/namespace/namespace_controller.go
+2
-2
replicasetcontroller.go
.../federation-controller/replicaset/replicasetcontroller.go
+3
-3
secret_controller.go
...ion/pkg/federation-controller/secret/secret_controller.go
+2
-2
cluster_helper.go
...ation/pkg/federation-controller/service/cluster_helper.go
+2
-2
servicecontroller.go
...on/pkg/federation-controller/service/servicecontroller.go
+3
-3
federated_informer.go
...tion/pkg/federation-controller/util/federated_informer.go
+2
-2
federated_informer_test.go
...pkg/federation-controller/util/federated_informer_test.go
+1
-1
controller.go
pkg/client/cache/controller.go
+15
-18
shared_informer.go
pkg/client/cache/shared_informer.go
+9
-5
certificate_controller.go
pkg/controller/certificates/certificate_controller.go
+1
-1
disruption.go
pkg/controller/disruption/disruption.go
+6
-6
endpoints_controller.go
pkg/controller/endpoint/endpoints_controller.go
+2
-2
garbagecollector.go
pkg/controller/garbagecollector/garbagecollector.go
+1
-1
namespace_controller.go
pkg/controller/namespace/namespace_controller.go
+1
-1
pet_set.go
pkg/controller/petset/pet_set.go
+2
-2
horizontal.go
pkg/controller/podautoscaler/horizontal.go
+2
-2
gc_controller.go
pkg/controller/podgc/gc_controller.go
+1
-1
gc_controller_test.go
pkg/controller/podgc/gc_controller_test.go
+4
-0
replication_controller.go
pkg/controller/replication/replication_controller.go
+1
-1
replenishment_controller.go
pkg/controller/resourcequota/replenishment_controller.go
+5
-4
resource_quota_controller.go
pkg/controller/resourcequota/resource_quota_controller.go
+3
-3
routecontroller.go
pkg/controller/route/routecontroller.go
+1
-1
servicecontroller.go
pkg/controller/service/servicecontroller.go
+1
-1
tokens_controller.go
pkg/controller/serviceaccount/tokens_controller.go
+2
-2
pv_controller.go
pkg/controller/volume/persistentvolume/pv_controller.go
+2
-2
factory.go
plugin/pkg/scheduler/factory/factory.go
+6
-6
controller.go
staging/src/k8s.io/client-go/tools/cache/controller.go
+14
-18
shared_informer.go
staging/src/k8s.io/client-go/tools/cache/shared_informer.go
+7
-3
daemon_restart.go
test/e2e/daemon_restart.go
+1
-1
network_partition.go
test/e2e/network_partition.go
+1
-1
density_test.go
test/e2e_node/density_test.go
+1
-2
No files found.
federation/pkg/federation-controller/cluster/clustercontroller.go
View file @
25fe1e0d
...
...
@@ -49,7 +49,7 @@ type ClusterController struct {
clusterKubeClientMap
map
[
string
]
ClusterClient
// cluster framework and store
clusterController
*
cache
.
Controller
clusterController
cache
.
Controller
clusterStore
clustercache
.
StoreToClusterLister
}
...
...
federation/pkg/federation-controller/configmap/configmap_controller.go
View file @
25fe1e0d
...
...
@@ -58,7 +58,7 @@ type ConfigMapController struct {
// Definitions of configmaps that should be federated.
configmapInformerStore
cache
.
Store
// Informer controller for configmaps that should be federated.
configmapInformerController
cache
.
Controller
Interface
configmapInformerController
cache
.
Controller
// Client to federated api server.
federatedApiClient
federationclientset
.
Interface
...
...
@@ -112,7 +112,7 @@ func NewConfigMapController(client federationclientset.Interface) *ConfigMapCont
// Federated informer on configmaps in members of federation.
configmapcontroller
.
configmapFederatedInformer
=
util
.
NewFederatedInformer
(
client
,
func
(
cluster
*
federationapi
.
Cluster
,
targetClient
kubeclientset
.
Interface
)
(
cache
.
Store
,
cache
.
Controller
Interface
)
{
func
(
cluster
*
federationapi
.
Cluster
,
targetClient
kubeclientset
.
Interface
)
(
cache
.
Store
,
cache
.
Controller
)
{
return
cache
.
NewInformer
(
&
cache
.
ListWatch
{
ListFunc
:
func
(
options
apiv1
.
ListOptions
)
(
pkgruntime
.
Object
,
error
)
{
...
...
federation/pkg/federation-controller/daemonset/daemonset_controller.go
View file @
25fe1e0d
...
...
@@ -63,7 +63,7 @@ type DaemonSetController struct {
// Definitions of daemonsets that should be federated.
daemonsetInformerStore
cache
.
Store
// Informer controller for daemonsets that should be federated.
daemonsetInformerController
cache
.
Controller
Interface
daemonsetInformerController
cache
.
Controller
// Client to federated api server.
federatedApiClient
federationclientset
.
Interface
...
...
@@ -119,7 +119,7 @@ func NewDaemonSetController(client federationclientset.Interface) *DaemonSetCont
// Federated informer on daemonsets in members of federation.
daemonsetcontroller
.
daemonsetFederatedInformer
=
util
.
NewFederatedInformer
(
client
,
func
(
cluster
*
federationapi
.
Cluster
,
targetClient
kubeclientset
.
Interface
)
(
cache
.
Store
,
cache
.
Controller
Interface
)
{
func
(
cluster
*
federationapi
.
Cluster
,
targetClient
kubeclientset
.
Interface
)
(
cache
.
Store
,
cache
.
Controller
)
{
return
cache
.
NewInformer
(
&
cache
.
ListWatch
{
ListFunc
:
func
(
options
apiv1
.
ListOptions
)
(
pkgruntime
.
Object
,
error
)
{
...
...
federation/pkg/federation-controller/deployment/deploymentcontroller.go
View file @
25fe1e0d
...
...
@@ -80,7 +80,7 @@ func parseFederationDeploymentPreference(fd *extensionsv1.Deployment) (*fed.Fede
type
DeploymentController
struct
{
fedClient
fedclientset
.
Interface
deploymentController
*
cache
.
Controller
deploymentController
cache
.
Controller
deploymentStore
cache
.
Store
fedDeploymentInformer
fedutil
.
FederatedInformer
...
...
@@ -119,7 +119,7 @@ func NewDeploymentController(federationClient fedclientset.Interface) *Deploymen
eventRecorder
:
recorder
,
}
deploymentFedInformerFactory
:=
func
(
cluster
*
fedv1
.
Cluster
,
clientset
kubeclientset
.
Interface
)
(
cache
.
Store
,
cache
.
Controller
Interface
)
{
deploymentFedInformerFactory
:=
func
(
cluster
*
fedv1
.
Cluster
,
clientset
kubeclientset
.
Interface
)
(
cache
.
Store
,
cache
.
Controller
)
{
return
cache
.
NewInformer
(
&
cache
.
ListWatch
{
ListFunc
:
func
(
options
apiv1
.
ListOptions
)
(
runtime
.
Object
,
error
)
{
...
...
@@ -146,7 +146,7 @@ func NewDeploymentController(federationClient fedclientset.Interface) *Deploymen
}
fdc
.
fedDeploymentInformer
=
fedutil
.
NewFederatedInformer
(
federationClient
,
deploymentFedInformerFactory
,
&
clusterLifecycle
)
podFedInformerFactory
:=
func
(
cluster
*
fedv1
.
Cluster
,
clientset
kubeclientset
.
Interface
)
(
cache
.
Store
,
cache
.
Controller
Interface
)
{
podFedInformerFactory
:=
func
(
cluster
*
fedv1
.
Cluster
,
clientset
kubeclientset
.
Interface
)
(
cache
.
Store
,
cache
.
Controller
)
{
return
cache
.
NewInformer
(
&
cache
.
ListWatch
{
ListFunc
:
func
(
options
apiv1
.
ListOptions
)
(
runtime
.
Object
,
error
)
{
...
...
federation/pkg/federation-controller/ingress/ingress_controller.go
View file @
25fe1e0d
...
...
@@ -91,7 +91,7 @@ type IngressController struct {
// Definitions of ingresses that should be federated.
ingressInformerStore
cache
.
Store
// Informer controller for ingresses that should be federated.
ingressInformerController
cache
.
Controller
Interface
ingressInformerController
cache
.
Controller
// Client to federated api server.
federatedApiClient
federationclientset
.
Interface
...
...
@@ -157,7 +157,7 @@ func NewIngressController(client federationclientset.Interface) *IngressControll
// Federated informer on ingresses in members of federation.
ic
.
ingressFederatedInformer
=
util
.
NewFederatedInformer
(
client
,
func
(
cluster
*
federationapi
.
Cluster
,
targetClient
kubeclientset
.
Interface
)
(
cache
.
Store
,
cache
.
Controller
Interface
)
{
func
(
cluster
*
federationapi
.
Cluster
,
targetClient
kubeclientset
.
Interface
)
(
cache
.
Store
,
cache
.
Controller
)
{
return
cache
.
NewInformer
(
&
cache
.
ListWatch
{
ListFunc
:
func
(
options
v1
.
ListOptions
)
(
pkgruntime
.
Object
,
error
)
{
...
...
@@ -189,7 +189,7 @@ func NewIngressController(client federationclientset.Interface) *IngressControll
// Federated informer on configmaps for ingress controllers in members of the federation.
ic
.
configMapFederatedInformer
=
util
.
NewFederatedInformer
(
client
,
func
(
cluster
*
federationapi
.
Cluster
,
targetClient
kubeclientset
.
Interface
)
(
cache
.
Store
,
cache
.
Controller
Interface
)
{
func
(
cluster
*
federationapi
.
Cluster
,
targetClient
kubeclientset
.
Interface
)
(
cache
.
Store
,
cache
.
Controller
)
{
glog
.
V
(
4
)
.
Infof
(
"Returning new informer for cluster %q"
,
cluster
.
Name
)
return
cache
.
NewInformer
(
&
cache
.
ListWatch
{
...
...
federation/pkg/federation-controller/namespace/namespace_controller.go
View file @
25fe1e0d
...
...
@@ -60,7 +60,7 @@ type NamespaceController struct {
// Definitions of namespaces that should be federated.
namespaceInformerStore
cache
.
Store
// Informer controller for namespaces that should be federated.
namespaceInformerController
cache
.
Controller
Interface
namespaceInformerController
cache
.
Controller
// Client to federated api server.
federatedApiClient
federationclientset
.
Interface
...
...
@@ -116,7 +116,7 @@ func NewNamespaceController(client federationclientset.Interface) *NamespaceCont
// Federated informer on namespaces in members of federation.
nc
.
namespaceFederatedInformer
=
util
.
NewFederatedInformer
(
client
,
func
(
cluster
*
federationapi
.
Cluster
,
targetClient
kubeclientset
.
Interface
)
(
cache
.
Store
,
cache
.
Controller
Interface
)
{
func
(
cluster
*
federationapi
.
Cluster
,
targetClient
kubeclientset
.
Interface
)
(
cache
.
Store
,
cache
.
Controller
)
{
return
cache
.
NewInformer
(
&
cache
.
ListWatch
{
ListFunc
:
func
(
options
apiv1
.
ListOptions
)
(
runtime
.
Object
,
error
)
{
...
...
federation/pkg/federation-controller/replicaset/replicasetcontroller.go
View file @
25fe1e0d
...
...
@@ -80,7 +80,7 @@ func parseFederationReplicaSetReference(frs *extensionsv1.ReplicaSet) (*fed.Fede
type
ReplicaSetController
struct
{
fedClient
fedclientset
.
Interface
replicaSetController
*
cache
.
Controller
replicaSetController
cache
.
Controller
replicaSetStore
cache
.
StoreToReplicaSetLister
fedReplicaSetInformer
fedutil
.
FederatedInformer
...
...
@@ -121,7 +121,7 @@ func NewReplicaSetController(federationClient fedclientset.Interface) *ReplicaSe
eventRecorder
:
recorder
,
}
replicaSetFedInformerFactory
:=
func
(
cluster
*
fedv1
.
Cluster
,
clientset
kubeclientset
.
Interface
)
(
cache
.
Store
,
cache
.
Controller
Interface
)
{
replicaSetFedInformerFactory
:=
func
(
cluster
*
fedv1
.
Cluster
,
clientset
kubeclientset
.
Interface
)
(
cache
.
Store
,
cache
.
Controller
)
{
return
cache
.
NewInformer
(
&
cache
.
ListWatch
{
ListFunc
:
func
(
options
apiv1
.
ListOptions
)
(
runtime
.
Object
,
error
)
{
...
...
@@ -148,7 +148,7 @@ func NewReplicaSetController(federationClient fedclientset.Interface) *ReplicaSe
}
frsc
.
fedReplicaSetInformer
=
fedutil
.
NewFederatedInformer
(
federationClient
,
replicaSetFedInformerFactory
,
&
clusterLifecycle
)
podFedInformerFactory
:=
func
(
cluster
*
fedv1
.
Cluster
,
clientset
kubeclientset
.
Interface
)
(
cache
.
Store
,
cache
.
Controller
Interface
)
{
podFedInformerFactory
:=
func
(
cluster
*
fedv1
.
Cluster
,
clientset
kubeclientset
.
Interface
)
(
cache
.
Store
,
cache
.
Controller
)
{
return
cache
.
NewInformer
(
&
cache
.
ListWatch
{
ListFunc
:
func
(
options
apiv1
.
ListOptions
)
(
runtime
.
Object
,
error
)
{
...
...
federation/pkg/federation-controller/secret/secret_controller.go
View file @
25fe1e0d
...
...
@@ -61,7 +61,7 @@ type SecretController struct {
// Definitions of secrets that should be federated.
secretInformerStore
cache
.
Store
// Informer controller for secrets that should be federated.
secretInformerController
cache
.
Controller
Interface
secretInformerController
cache
.
Controller
// Client to federated api server.
federatedApiClient
federationclientset
.
Interface
...
...
@@ -117,7 +117,7 @@ func NewSecretController(client federationclientset.Interface) *SecretController
// Federated informer on secrets in members of federation.
secretcontroller
.
secretFederatedInformer
=
util
.
NewFederatedInformer
(
client
,
func
(
cluster
*
federationapi
.
Cluster
,
targetClient
kubeclientset
.
Interface
)
(
cache
.
Store
,
cache
.
Controller
Interface
)
{
func
(
cluster
*
federationapi
.
Cluster
,
targetClient
kubeclientset
.
Interface
)
(
cache
.
Store
,
cache
.
Controller
)
{
return
cache
.
NewInformer
(
&
cache
.
ListWatch
{
ListFunc
:
func
(
options
apiv1
.
ListOptions
)
(
pkgruntime
.
Object
,
error
)
{
...
...
federation/pkg/federation-controller/service/cluster_helper.go
View file @
25fe1e0d
...
...
@@ -41,11 +41,11 @@ type clusterCache struct {
// A store of services, populated by the serviceController
serviceStore
cache
.
StoreToServiceLister
// Watches changes to all services
serviceController
*
cache
.
Controller
serviceController
cache
.
Controller
// A store of endpoint, populated by the serviceController
endpointStore
cache
.
StoreToEndpointsLister
// Watches changes to all endpoints
endpointController
*
cache
.
Controller
endpointController
cache
.
Controller
// services that need to be synced
serviceQueue
*
workqueue
.
Type
// endpoints that need to be synced
...
...
federation/pkg/federation-controller/service/servicecontroller.go
View file @
25fe1e0d
...
...
@@ -121,12 +121,12 @@ type ServiceController struct {
// A store of services, populated by the serviceController
serviceStore
cache
.
StoreToServiceLister
// Watches changes to all services
serviceController
*
cache
.
Controller
serviceController
cache
.
Controller
federatedInformer
fedutil
.
FederatedInformer
// A store of services, populated by the serviceController
clusterStore
federationcache
.
StoreToClusterLister
// Watches changes to all services
clusterController
*
cache
.
Controller
clusterController
cache
.
Controller
eventBroadcaster
record
.
EventBroadcaster
eventRecorder
record
.
EventRecorder
// services that need to be synced
...
...
@@ -245,7 +245,7 @@ func New(federationClient fedclientset.Interface, dns dnsprovider.Interface,
s
.
clusterDeliverer
.
DeliverAfter
(
allClustersKey
,
nil
,
clusterAvailableDelay
)
},
}
fedInformerFactory
:=
func
(
cluster
*
v1beta1
.
Cluster
,
targetClient
kubeclientset
.
Interface
)
(
cache
.
Store
,
cache
.
Controller
Interface
)
{
fedInformerFactory
:=
func
(
cluster
*
v1beta1
.
Cluster
,
targetClient
kubeclientset
.
Interface
)
(
cache
.
Store
,
cache
.
Controller
)
{
return
cache
.
NewInformer
(
&
cache
.
ListWatch
{
ListFunc
:
func
(
options
v1
.
ListOptions
)
(
pkgruntime
.
Object
,
error
)
{
...
...
federation/pkg/federation-controller/util/federated_informer.go
View file @
25fe1e0d
...
...
@@ -116,7 +116,7 @@ type FederatedInformerForTestOnly interface {
// A function that should be used to create an informer on the target object. Store should use
// cache.DeletionHandlingMetaNamespaceKeyFunc as a keying function.
type
TargetInformerFactory
func
(
*
federationapi
.
Cluster
,
kubeclientset
.
Interface
)
(
cache
.
Store
,
cache
.
Controller
Interface
)
type
TargetInformerFactory
func
(
*
federationapi
.
Cluster
,
kubeclientset
.
Interface
)
(
cache
.
Store
,
cache
.
Controller
)
// A structure with cluster lifecycle handler functions. Cluster is available (and ClusterAvailable is fired)
// when it is created in federated etcd and ready. Cluster becomes unavailable (and ClusterUnavailable is fired)
...
...
@@ -242,7 +242,7 @@ func isClusterReady(cluster *federationapi.Cluster) bool {
}
type
informer
struct
{
controller
cache
.
Controller
Interface
controller
cache
.
Controller
store
cache
.
Store
stopChan
chan
struct
{}
}
...
...
federation/pkg/federation-controller/util/federated_informer_test.go
View file @
25fe1e0d
...
...
@@ -77,7 +77,7 @@ func TestFederatedInformer(t *testing.T) {
return
true
,
watch
.
NewFake
(),
nil
})
targetInformerFactory
:=
func
(
cluster
*
federationapi
.
Cluster
,
clientset
kubeclientset
.
Interface
)
(
cache
.
Store
,
cache
.
Controller
Interface
)
{
targetInformerFactory
:=
func
(
cluster
*
federationapi
.
Cluster
,
clientset
kubeclientset
.
Interface
)
(
cache
.
Store
,
cache
.
Controller
)
{
return
cache
.
NewInformer
(
&
cache
.
ListWatch
{
ListFunc
:
func
(
options
apiv1
.
ListOptions
)
(
runtime
.
Object
,
error
)
{
...
...
pkg/client/cache/controller.go
View file @
25fe1e0d
...
...
@@ -61,21 +61,21 @@ type Config struct {
type
ProcessFunc
func
(
obj
interface
{})
error
// Controller is a generic controller framework.
type
C
ontroller
struct
{
type
c
ontroller
struct
{
config
Config
reflector
*
Reflector
reflectorMutex
sync
.
RWMutex
}
// TODO make the "Controller" private, and convert all references to use ControllerInterface instead
type
ControllerInterface
interface
{
type
Controller
interface
{
Run
(
stopCh
<-
chan
struct
{})
HasSynced
()
bool
LastSyncResourceVersion
()
string
}
// New makes a new Controller from the given Config.
func
New
(
c
*
Config
)
*
Controller
{
ctlr
:=
&
C
ontroller
{
func
New
(
c
*
Config
)
Controller
{
ctlr
:=
&
c
ontroller
{
config
:
*
c
,
}
return
ctlr
...
...
@@ -84,7 +84,7 @@ func New(c *Config) *Controller {
// Run begins processing items, and will continue until a value is sent down stopCh.
// It's an error to call Run more than once.
// Run blocks; call via go.
func
(
c
*
C
ontroller
)
Run
(
stopCh
<-
chan
struct
{})
{
func
(
c
*
c
ontroller
)
Run
(
stopCh
<-
chan
struct
{})
{
defer
utilruntime
.
HandleCrash
()
r
:=
NewReflector
(
c
.
config
.
ListerWatcher
,
...
...
@@ -103,18 +103,15 @@ func (c *Controller) Run(stopCh <-chan struct{}) {
}
// Returns true once this controller has completed an initial resource listing
func
(
c
*
C
ontroller
)
HasSynced
()
bool
{
func
(
c
*
c
ontroller
)
HasSynced
()
bool
{
return
c
.
config
.
Queue
.
HasSynced
()
}
// Requeue adds the provided object back into the queue if it does not already exist.
func
(
c
*
Controller
)
Requeue
(
obj
interface
{})
error
{
return
c
.
config
.
Queue
.
AddIfNotPresent
(
Deltas
{
Delta
{
Type
:
Sync
,
Object
:
obj
,
},
})
func
(
c
*
controller
)
LastSyncResourceVersion
()
string
{
if
c
.
reflector
==
nil
{
return
""
}
return
c
.
reflector
.
LastSyncResourceVersion
()
}
// processLoop drains the work queue.
...
...
@@ -126,7 +123,7 @@ func (c *Controller) Requeue(obj interface{}) error {
// actually exit when the controller is stopped. Or just give up on this stuff
// ever being stoppable. Converting this whole package to use Context would
// also be helpful.
func
(
c
*
C
ontroller
)
processLoop
()
{
func
(
c
*
c
ontroller
)
processLoop
()
{
for
{
obj
,
err
:=
c
.
config
.
Queue
.
Pop
(
PopProcessFunc
(
c
.
config
.
Process
))
if
err
!=
nil
{
...
...
@@ -218,7 +215,7 @@ func NewInformer(
objType
runtime
.
Object
,
resyncPeriod
time
.
Duration
,
h
ResourceEventHandler
,
)
(
Store
,
*
Controller
)
{
)
(
Store
,
Controller
)
{
// This will hold the client state, as we know it.
clientState
:=
NewStore
(
DeletionHandlingMetaNamespaceKeyFunc
)
...
...
@@ -284,7 +281,7 @@ func NewIndexerInformer(
resyncPeriod
time
.
Duration
,
h
ResourceEventHandler
,
indexers
Indexers
,
)
(
Indexer
,
*
Controller
)
{
)
(
Indexer
,
Controller
)
{
// This will hold the client state, as we know it.
clientState
:=
NewIndexer
(
DeletionHandlingMetaNamespaceKeyFunc
,
indexers
)
...
...
pkg/client/cache/shared_informer.go
View file @
25fe1e0d
...
...
@@ -43,7 +43,7 @@ type SharedInformer interface {
AddEventHandler
(
handler
ResourceEventHandler
)
error
GetStore
()
Store
// GetController gives back a synthetic interface that "votes" to start the informer
GetController
()
Controller
Interface
GetController
()
Controller
Run
(
stopCh
<-
chan
struct
{})
HasSynced
()
bool
LastSyncResourceVersion
()
string
...
...
@@ -108,7 +108,7 @@ func WaitForCacheSync(stopCh <-chan struct{}, cacheSyncs ...InformerSynced) bool
type
sharedIndexInformer
struct
{
indexer
Indexer
controller
*
Controller
controller
Controller
processor
*
sharedProcessor
cacheMutationDetector
CacheMutationDetector
...
...
@@ -145,6 +145,10 @@ func (v *dummyController) HasSynced() bool {
return
v
.
informer
.
HasSynced
()
}
func
(
c
*
dummyController
)
LastSyncResourceVersion
()
string
{
return
""
}
type
updateNotification
struct
{
oldObj
interface
{}
newObj
interface
{}
...
...
@@ -207,10 +211,10 @@ func (s *sharedIndexInformer) LastSyncResourceVersion() string {
s
.
startedLock
.
Lock
()
defer
s
.
startedLock
.
Unlock
()
if
s
.
controller
==
nil
||
s
.
controller
.
reflector
==
nil
{
if
s
.
controller
==
nil
{
return
""
}
return
s
.
controller
.
reflector
.
LastSyncResourceVersion
()
return
s
.
controller
.
LastSyncResourceVersion
()
}
func
(
s
*
sharedIndexInformer
)
GetStore
()
Store
{
...
...
@@ -232,7 +236,7 @@ func (s *sharedIndexInformer) AddIndexers(indexers Indexers) error {
return
s
.
indexer
.
AddIndexers
(
indexers
)
}
func
(
s
*
sharedIndexInformer
)
GetController
()
Controller
Interface
{
func
(
s
*
sharedIndexInformer
)
GetController
()
Controller
{
return
&
dummyController
{
informer
:
s
}
}
...
...
pkg/controller/certificates/certificate_controller.go
View file @
25fe1e0d
...
...
@@ -48,7 +48,7 @@ type CertificateController struct {
kubeClient
clientset
.
Interface
// CSR framework and store
csrController
*
cache
.
Controller
csrController
cache
.
Controller
csrStore
cache
.
StoreToCertificateRequestLister
syncHandler
func
(
csrKey
string
)
error
...
...
pkg/controller/disruption/disruption.go
View file @
25fe1e0d
...
...
@@ -63,26 +63,26 @@ type DisruptionController struct {
kubeClient
clientset
.
Interface
pdbStore
cache
.
Store
pdbController
*
cache
.
Controller
pdbController
cache
.
Controller
pdbLister
cache
.
StoreToPodDisruptionBudgetLister
podController
cache
.
Controller
Interface
podController
cache
.
Controller
podLister
cache
.
StoreToPodLister
rcIndexer
cache
.
Indexer
rcController
*
cache
.
Controller
rcController
cache
.
Controller
rcLister
cache
.
StoreToReplicationControllerLister
rsStore
cache
.
Store
rsController
*
cache
.
Controller
rsController
cache
.
Controller
rsLister
cache
.
StoreToReplicaSetLister
dIndexer
cache
.
Indexer
dController
*
cache
.
Controller
dController
cache
.
Controller
dLister
cache
.
StoreToDeploymentLister
ssStore
cache
.
Store
ssController
*
cache
.
Controller
ssController
cache
.
Controller
ssLister
cache
.
StoreToStatefulSetLister
// PodDisruptionBudget keys that need to be synced.
...
...
pkg/controller/endpoint/endpoints_controller.go
View file @
25fe1e0d
...
...
@@ -147,8 +147,8 @@ type EndpointController struct {
// Since we join two objects, we'll watch both of them with
// controllers.
serviceController
*
cache
.
Controller
podController
cache
.
Controller
Interface
serviceController
cache
.
Controller
podController
cache
.
Controller
// podStoreSynced returns true if the pod store has been synced at least once.
// Added as a member to the struct to allow injection for testing.
podStoreSynced
func
()
bool
...
...
pkg/controller/garbagecollector/garbagecollector.go
View file @
25fe1e0d
...
...
@@ -48,7 +48,7 @@ const ResourceResyncTime time.Duration = 0
type
monitor
struct
{
store
cache
.
Store
controller
*
cache
.
Controller
controller
cache
.
Controller
}
type
objectReference
struct
{
...
...
pkg/controller/namespace/namespace_controller.go
View file @
25fe1e0d
...
...
@@ -56,7 +56,7 @@ type NamespaceController struct {
// store that holds the namespaces
store
cache
.
Store
// controller that observes the namespaces
controller
*
cache
.
Controller
controller
cache
.
Controller
// namespaces that have been queued up for processing by workers
queue
workqueue
.
RateLimitingInterface
// function to list of preferred resources for namespace deletion
...
...
pkg/controller/petset/pet_set.go
View file @
25fe1e0d
...
...
@@ -64,12 +64,12 @@ type StatefulSetController struct {
// podStoreSynced returns true if the pod store has synced at least once.
podStoreSynced
func
()
bool
// Watches changes to all pods.
podController
cache
.
Controller
Interface
podController
cache
.
Controller
// A store of StatefulSets, populated by the psController.
psStore
cache
.
StoreToStatefulSetLister
// Watches changes to all StatefulSets.
psController
*
cache
.
Controller
psController
cache
.
Controller
// A store of the 1 unhealthy pet blocking progress for a given ps
blockingPetStore
*
unhealthyPetTracker
...
...
pkg/controller/podautoscaler/horizontal.go
View file @
25fe1e0d
...
...
@@ -66,13 +66,13 @@ type HorizontalController struct {
// A store of HPA objects, populated by the controller.
store
cache
.
Store
// Watches changes to all HPA objects.
controller
*
cache
.
Controller
controller
cache
.
Controller
}
var
downscaleForbiddenWindow
=
5
*
time
.
Minute
var
upscaleForbiddenWindow
=
3
*
time
.
Minute
func
newInformer
(
controller
*
HorizontalController
,
resyncPeriod
time
.
Duration
)
(
cache
.
Store
,
*
cache
.
Controller
)
{
func
newInformer
(
controller
*
HorizontalController
,
resyncPeriod
time
.
Duration
)
(
cache
.
Store
,
cache
.
Controller
)
{
return
cache
.
NewInformer
(
&
cache
.
ListWatch
{
ListFunc
:
func
(
options
v1
.
ListOptions
)
(
runtime
.
Object
,
error
)
{
...
...
pkg/controller/podgc/gc_controller.go
View file @
25fe1e0d
...
...
@@ -50,7 +50,7 @@ type PodGCController struct {
internalPodInformer
cache
.
SharedIndexInformer
podStore
cache
.
StoreToPodLister
podController
cache
.
Controller
Interface
podController
cache
.
Controller
deletePod
func
(
namespace
,
name
string
)
error
terminatedPodThreshold
int
...
...
pkg/controller/podgc/gc_controller_test.go
View file @
25fe1e0d
...
...
@@ -37,6 +37,10 @@ func (*FakeController) HasSynced() bool {
return
true
}
func
(
*
FakeController
)
LastSyncResourceVersion
()
string
{
return
""
}
func
TestGCTerminated
(
t
*
testing
.
T
)
{
type
nameToPhase
struct
{
name
string
...
...
pkg/controller/replication/replication_controller.go
View file @
25fe1e0d
...
...
@@ -86,7 +86,7 @@ type ReplicationManager struct {
// A store of pods, populated by the podController
podLister
cache
.
StoreToPodLister
// Watches changes to all pods
podController
cache
.
Controller
Interface
podController
cache
.
Controller
// podListerSynced returns true if the pod store has been synced at least once.
// Added as a member to the struct to allow injection for testing.
podListerSynced
func
()
bool
...
...
pkg/controller/resourcequota/replenishment_controller.go
View file @
25fe1e0d
...
...
@@ -90,7 +90,7 @@ func ObjectReplenishmentDeleteFunc(options *ReplenishmentControllerOptions) func
type
ReplenishmentControllerFactory
interface
{
// NewController returns a controller configured with the specified options.
// This method is NOT thread-safe.
NewController
(
options
*
ReplenishmentControllerOptions
)
(
cache
.
Controller
Interface
,
error
)
NewController
(
options
*
ReplenishmentControllerOptions
)
(
cache
.
Controller
,
error
)
}
// replenishmentControllerFactory implements ReplenishmentControllerFactory
...
...
@@ -118,7 +118,8 @@ func NewReplenishmentControllerFactoryFromClient(kubeClient clientset.Interface)
func
controllerFor
(
groupResource
schema
.
GroupResource
,
f
informers
.
SharedInformerFactory
,
handlerFuncs
cache
.
ResourceEventHandlerFuncs
)
(
cache
.
ControllerInterface
,
error
)
{
handlerFuncs
cache
.
ResourceEventHandlerFuncs
,
)
(
cache
.
Controller
,
error
)
{
genericInformer
,
err
:=
f
.
ForResource
(
groupResource
)
if
err
!=
nil
{
return
nil
,
err
...
...
@@ -128,7 +129,7 @@ func controllerFor(
return
informer
.
GetController
(),
nil
}
func
(
r
*
replenishmentControllerFactory
)
NewController
(
options
*
ReplenishmentControllerOptions
)
(
result
cache
.
Controller
Interface
,
err
error
)
{
func
(
r
*
replenishmentControllerFactory
)
NewController
(
options
*
ReplenishmentControllerOptions
)
(
result
cache
.
Controller
,
err
error
)
{
if
r
.
kubeClient
!=
nil
&&
r
.
kubeClient
.
Core
()
.
RESTClient
()
.
GetRateLimiter
()
!=
nil
{
metrics
.
RegisterMetricAndTrackRateLimiterUsage
(
"replenishment_controller"
,
r
.
kubeClient
.
Core
()
.
RESTClient
()
.
GetRateLimiter
())
}
...
...
@@ -276,7 +277,7 @@ func IsUnhandledGroupKindError(err error) bool {
// returning the first success or failure it hits. If there are no hits either way, it return an UnhandledGroupKind error
type
UnionReplenishmentControllerFactory
[]
ReplenishmentControllerFactory
func
(
f
UnionReplenishmentControllerFactory
)
NewController
(
options
*
ReplenishmentControllerOptions
)
(
cache
.
Controller
Interface
,
error
)
{
func
(
f
UnionReplenishmentControllerFactory
)
NewController
(
options
*
ReplenishmentControllerOptions
)
(
cache
.
Controller
,
error
)
{
for
_
,
factory
:=
range
f
{
controller
,
err
:=
factory
.
NewController
(
options
)
if
!
IsUnhandledGroupKindError
(
err
)
{
...
...
pkg/controller/resourcequota/resource_quota_controller.go
View file @
25fe1e0d
...
...
@@ -60,7 +60,7 @@ type ResourceQuotaController struct {
// An index of resource quota objects by namespace
rqIndexer
cache
.
Indexer
// Watches changes to all resource quota
rqController
*
cache
.
Controller
rqController
cache
.
Controller
// ResourceQuota objects that need to be synchronized
queue
workqueue
.
RateLimitingInterface
// missingUsageQueue holds objects that are missing the initial usage informatino
...
...
@@ -72,7 +72,7 @@ type ResourceQuotaController struct {
// knows how to calculate usage
registry
quota
.
Registry
// controllers monitoring to notify for replenishment
replenishmentControllers
[]
cache
.
Controller
Interface
replenishmentControllers
[]
cache
.
Controller
}
func
NewResourceQuotaController
(
options
*
ResourceQuotaControllerOptions
)
*
ResourceQuotaController
{
...
...
@@ -83,7 +83,7 @@ func NewResourceQuotaController(options *ResourceQuotaControllerOptions) *Resour
missingUsageQueue
:
workqueue
.
NewNamedRateLimitingQueue
(
workqueue
.
DefaultControllerRateLimiter
(),
"resourcequota_priority"
),
resyncPeriod
:
options
.
ResyncPeriod
,
registry
:
options
.
Registry
,
replenishmentControllers
:
[]
cache
.
Controller
Interface
{},
replenishmentControllers
:
[]
cache
.
Controller
{},
}
if
options
.
KubeClient
!=
nil
&&
options
.
KubeClient
.
Core
()
.
RESTClient
()
.
GetRateLimiter
()
!=
nil
{
metrics
.
RegisterMetricAndTrackRateLimiterUsage
(
"resource_quota_controller"
,
options
.
KubeClient
.
Core
()
.
RESTClient
()
.
GetRateLimiter
())
...
...
pkg/controller/route/routecontroller.go
View file @
25fe1e0d
...
...
@@ -54,7 +54,7 @@ type RouteController struct {
clusterName
string
clusterCIDR
*
net
.
IPNet
// Node framework and store
nodeController
*
cache
.
Controller
nodeController
cache
.
Controller
nodeStore
cache
.
StoreToNodeLister
}
...
...
pkg/controller/service/servicecontroller.go
View file @
25fe1e0d
...
...
@@ -87,7 +87,7 @@ type ServiceController struct {
// A store of services, populated by the serviceController
serviceStore
cache
.
StoreToServiceLister
// Watches changes to all services
serviceController
*
cache
.
Controller
serviceController
cache
.
Controller
eventBroadcaster
record
.
EventBroadcaster
eventRecorder
record
.
EventRecorder
nodeLister
cache
.
StoreToNodeLister
...
...
pkg/controller/serviceaccount/tokens_controller.go
View file @
25fe1e0d
...
...
@@ -145,8 +145,8 @@ type TokensController struct {
secrets
cache
.
Indexer
// Since we join two objects, we'll watch both of them with controllers.
serviceAccountController
*
cache
.
Controller
secretController
*
cache
.
Controller
serviceAccountController
cache
.
Controller
secretController
cache
.
Controller
// syncServiceAccountQueue handles service account events:
// * ensures a referenced token exists for service accounts which still exist
...
...
pkg/controller/volume/persistentvolume/pv_controller.go
View file @
25fe1e0d
...
...
@@ -146,10 +146,10 @@ const createProvisionedPVInterval = 10 * time.Second
// cache.Controllers that watch PersistentVolume and PersistentVolumeClaim
// changes.
type
PersistentVolumeController
struct
{
volumeController
*
cache
.
Controller
volumeController
cache
.
Controller
volumeInformer
cache
.
Indexer
volumeSource
cache
.
ListerWatcher
claimController
*
cache
.
Controller
claimController
cache
.
Controller
claimInformer
cache
.
Store
claimSource
cache
.
ListerWatcher
classReflector
*
cache
.
Reflector
...
...
plugin/pkg/scheduler/factory/factory.go
View file @
25fe1e0d
...
...
@@ -80,12 +80,12 @@ type ConfigFactory struct {
StopEverything
chan
struct
{}
informerFactory
informers
.
SharedInformerFactory
scheduledPodPopulator
cache
.
Controller
Interface
nodePopulator
cache
.
Controller
Interface
pvPopulator
cache
.
Controller
Interface
pvcPopulator
cache
.
Controller
Interface
servicePopulator
cache
.
Controller
Interface
controllerPopulator
cache
.
Controller
Interface
scheduledPodPopulator
cache
.
Controller
nodePopulator
cache
.
Controller
pvPopulator
cache
.
Controller
pvcPopulator
cache
.
Controller
servicePopulator
cache
.
Controller
controllerPopulator
cache
.
Controller
schedulerCache
schedulercache
.
Cache
...
...
staging/src/k8s.io/client-go/tools/cache/controller.go
View file @
25fe1e0d
...
...
@@ -61,21 +61,20 @@ type Config struct {
type
ProcessFunc
func
(
obj
interface
{})
error
// Controller is a generic controller framework.
type
C
ontroller
struct
{
type
c
ontroller
struct
{
config
Config
reflector
*
Reflector
reflectorMutex
sync
.
RWMutex
}
// TODO make the "Controller" private, and convert all references to use ControllerInterface instead
type
ControllerInterface
interface
{
type
Controller
interface
{
Run
(
stopCh
<-
chan
struct
{})
HasSynced
()
bool
}
// New makes a new Controller from the given Config.
func
New
(
c
*
Config
)
*
Controller
{
ctlr
:=
&
C
ontroller
{
func
New
(
c
*
Config
)
Controller
{
ctlr
:=
&
c
ontroller
{
config
:
*
c
,
}
return
ctlr
...
...
@@ -84,7 +83,7 @@ func New(c *Config) *Controller {
// Run begins processing items, and will continue until a value is sent down stopCh.
// It's an error to call Run more than once.
// Run blocks; call via go.
func
(
c
*
C
ontroller
)
Run
(
stopCh
<-
chan
struct
{})
{
func
(
c
*
c
ontroller
)
Run
(
stopCh
<-
chan
struct
{})
{
defer
utilruntime
.
HandleCrash
()
r
:=
NewReflector
(
c
.
config
.
ListerWatcher
,
...
...
@@ -103,18 +102,15 @@ func (c *Controller) Run(stopCh <-chan struct{}) {
}
// Returns true once this controller has completed an initial resource listing
func
(
c
*
C
ontroller
)
HasSynced
()
bool
{
func
(
c
*
c
ontroller
)
HasSynced
()
bool
{
return
c
.
config
.
Queue
.
HasSynced
()
}
// Requeue adds the provided object back into the queue if it does not already exist.
func
(
c
*
Controller
)
Requeue
(
obj
interface
{})
error
{
return
c
.
config
.
Queue
.
AddIfNotPresent
(
Deltas
{
Delta
{
Type
:
Sync
,
Object
:
obj
,
},
})
func
(
c
*
controller
)
LastSyncResourceVersion
()
string
{
if
c
.
reflector
==
nil
{
return
""
}
return
c
.
reflector
.
LastSyncResourceVersion
()
}
// processLoop drains the work queue.
...
...
@@ -126,7 +122,7 @@ func (c *Controller) Requeue(obj interface{}) error {
// actually exit when the controller is stopped. Or just give up on this stuff
// ever being stoppable. Converting this whole package to use Context would
// also be helpful.
func
(
c
*
C
ontroller
)
processLoop
()
{
func
(
c
*
c
ontroller
)
processLoop
()
{
for
{
obj
,
err
:=
c
.
config
.
Queue
.
Pop
(
PopProcessFunc
(
c
.
config
.
Process
))
if
err
!=
nil
{
...
...
@@ -218,7 +214,7 @@ func NewInformer(
objType
runtime
.
Object
,
resyncPeriod
time
.
Duration
,
h
ResourceEventHandler
,
)
(
Store
,
*
Controller
)
{
)
(
Store
,
Controller
)
{
// This will hold the client state, as we know it.
clientState
:=
NewStore
(
DeletionHandlingMetaNamespaceKeyFunc
)
...
...
@@ -284,7 +280,7 @@ func NewIndexerInformer(
resyncPeriod
time
.
Duration
,
h
ResourceEventHandler
,
indexers
Indexers
,
)
(
Indexer
,
*
Controller
)
{
)
(
Indexer
,
Controller
)
{
// This will hold the client state, as we know it.
clientState
:=
NewIndexer
(
DeletionHandlingMetaNamespaceKeyFunc
,
indexers
)
...
...
staging/src/k8s.io/client-go/tools/cache/shared_informer.go
View file @
25fe1e0d
...
...
@@ -43,7 +43,7 @@ type SharedInformer interface {
AddEventHandler
(
handler
ResourceEventHandler
)
error
GetStore
()
Store
// GetController gives back a synthetic interface that "votes" to start the informer
GetController
()
Controller
Interface
GetController
()
Controller
Run
(
stopCh
<-
chan
struct
{})
HasSynced
()
bool
LastSyncResourceVersion
()
string
...
...
@@ -108,7 +108,7 @@ func WaitForCacheSync(stopCh <-chan struct{}, cacheSyncs ...InformerSynced) bool
type
sharedIndexInformer
struct
{
indexer
Indexer
controller
*
Controller
controller
Controller
processor
*
sharedProcessor
cacheMutationDetector
CacheMutationDetector
...
...
@@ -145,6 +145,10 @@ func (v *dummyController) HasSynced() bool {
return
v
.
informer
.
HasSynced
()
}
func
(
c
*
dummyController
)
LastSyncResourceVersion
()
string
{
return
""
}
type
updateNotification
struct
{
oldObj
interface
{}
newObj
interface
{}
...
...
@@ -232,7 +236,7 @@ func (s *sharedIndexInformer) AddIndexers(indexers Indexers) error {
return
s
.
indexer
.
AddIndexers
(
indexers
)
}
func
(
s
*
sharedIndexInformer
)
GetController
()
Controller
Interface
{
func
(
s
*
sharedIndexInformer
)
GetController
()
Controller
{
return
&
dummyController
{
informer
:
s
}
}
...
...
test/e2e/daemon_restart.go
View file @
25fe1e0d
...
...
@@ -192,7 +192,7 @@ var _ = framework.KubeDescribe("DaemonRestart [Disruptive]", func() {
existingPods
:=
cache
.
NewStore
(
cache
.
MetaNamespaceKeyFunc
)
var
ns
string
var
config
testutils
.
RCConfig
var
controller
*
cache
.
Controller
var
controller
cache
.
Controller
var
newPods
cache
.
Store
var
stopCh
chan
struct
{}
var
tracker
*
podTracker
...
...
test/e2e/network_partition.go
View file @
25fe1e0d
...
...
@@ -188,7 +188,7 @@ var _ = framework.KubeDescribe("Network Partition [Disruptive] [Slow]", func() {
nodeSelector
:=
fields
.
OneTermEqualSelector
(
"metadata.name"
,
node
.
Name
)
stopCh
:=
make
(
chan
struct
{})
newNode
:=
make
(
chan
*
v1
.
Node
)
var
controller
*
cache
.
Controller
var
controller
cache
.
Controller
_
,
controller
=
cache
.
NewInformer
(
&
cache
.
ListWatch
{
ListFunc
:
func
(
options
v1
.
ListOptions
)
(
runtime
.
Object
,
error
)
{
...
...
test/e2e_node/density_test.go
View file @
25fe1e0d
...
...
@@ -476,8 +476,7 @@ func verifyPodStartupLatency(expect, actual framework.LatencyMetric) error {
}
// newInformerWatchPod creates an informer to check whether all pods are running.
func
newInformerWatchPod
(
f
*
framework
.
Framework
,
mutex
*
sync
.
Mutex
,
watchTimes
map
[
string
]
metav1
.
Time
,
podType
string
)
*
cache
.
Controller
{
func
newInformerWatchPod
(
f
*
framework
.
Framework
,
mutex
*
sync
.
Mutex
,
watchTimes
map
[
string
]
metav1
.
Time
,
podType
string
)
cache
.
Controller
{
ns
:=
f
.
Namespace
.
Name
checkPodRunning
:=
func
(
p
*
v1
.
Pod
)
{
mutex
.
Lock
()
...
...
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