-
Notifications
You must be signed in to change notification settings - Fork 300
/
controller.go
137 lines (118 loc) · 3.95 KB
/
controller.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
/*
Copyright 2017 The Kubernetes Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package instancegroups
import (
"time"
apiv1 "k8s.io/api/core/v1"
meta_v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/client-go/tools/cache"
"k8s.io/ingress-gce/pkg/utils"
"k8s.io/ingress-gce/pkg/utils/zonegetter"
"k8s.io/klog/v2"
)
// Controller synchronizes the state of the nodes to the unmanaged instance
// groups.
type Controller struct {
// lister is a cache of the k8s Node resources.
lister cache.Indexer
// queue is the TaskQueue used to manage the node worker updates.
queue utils.TaskQueue
// igManager is an interface to manage instance groups.
igManager Manager
// hasSynced returns true if relevant caches have done their initial
// synchronization.
hasSynced func() bool
enableMultiSubnetCluster bool
stopCh <-chan struct{}
zoneGetter *zonegetter.ZoneGetter
logger klog.Logger
}
type ControllerConfig struct {
NodeInformer cache.SharedIndexInformer
ZoneGetter *zonegetter.ZoneGetter
IGManager Manager
HasSynced func() bool
EnableMultiSubnetCluster bool
StopCh <-chan struct{}
}
var defaultNodeObj = &apiv1.Node{
ObjectMeta: meta_v1.ObjectMeta{
Name: "default",
},
}
// NewController returns a new node update controller.
func NewController(config *ControllerConfig, logger klog.Logger) *Controller {
logger = logger.WithName("InstanceGroupsController")
c := &Controller{
lister: config.NodeInformer.GetIndexer(),
zoneGetter: config.ZoneGetter,
igManager: config.IGManager,
hasSynced: config.HasSynced,
enableMultiSubnetCluster: config.EnableMultiSubnetCluster,
stopCh: config.StopCh,
logger: logger,
}
c.queue = utils.NewPeriodicTaskQueue("", "nodes", c.sync, logger)
config.NodeInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
AddFunc: func(obj interface{}) {
c.queue.Enqueue(defaultNodeObj)
},
DeleteFunc: func(obj interface{}) {
c.queue.Enqueue(defaultNodeObj)
},
UpdateFunc: func(oldObj, newObj interface{}) {
if nodeStatusChanged(oldObj.(*apiv1.Node), newObj.(*apiv1.Node)) {
c.queue.Enqueue(defaultNodeObj)
}
},
})
return c
}
// Run the queue to process updates for the controller. This must be run in a
// separate goroutine (method will block until queue shutdown).
func (c *Controller) Run() {
start := time.Now()
for !c.hasSynced() {
c.logger.V(2).Info("Waiting for hasSynced", "elapsedTime", time.Now().Sub(start))
time.Sleep(1 * time.Second)
}
c.logger.V(2).Info("Caches synced", "timeTaken", time.Now().Sub(start))
go c.queue.Run()
<-c.stopCh
c.Shutdown()
}
func nodeStatusChanged(old, cur *apiv1.Node) bool {
if old.Spec.Unschedulable != cur.Spec.Unschedulable {
return true
}
if utils.NodeIsReady(old) != utils.NodeIsReady(cur) {
return true
}
return false
}
// Shutdown shuts down the goroutine that processes node updates.
func (c *Controller) Shutdown() {
c.queue.Shutdown()
}
func (c *Controller) sync(key string) error {
start := time.Now()
c.logger.V(4).Info("Instance groups controller: Start processing", "key", key)
defer func() {
c.logger.V(4).Info("Instance groups controller: Processing key finished", "key", key, "timeTaken", time.Since(start))
}()
nodes, err := c.zoneGetter.ListNodes(zonegetter.CandidateNodesFilter, c.logger)
if err != nil {
return err
}
return c.igManager.Sync(utils.GetNodeNames(nodes))
}