-
Notifications
You must be signed in to change notification settings - Fork 463
/
kube-router.go
233 lines (193 loc) · 6.88 KB
/
kube-router.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
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
package cmd
import (
"errors"
"fmt"
"os"
"os/signal"
"runtime"
"sync"
"syscall"
"github.com/cloudnativelabs/kube-router/pkg/controllers/netpol"
"github.com/cloudnativelabs/kube-router/pkg/controllers/proxy"
"github.com/cloudnativelabs/kube-router/pkg/controllers/routing"
"github.com/cloudnativelabs/kube-router/pkg/healthcheck"
"github.com/cloudnativelabs/kube-router/pkg/metrics"
"github.com/cloudnativelabs/kube-router/pkg/options"
"github.com/golang/glog"
"time"
"k8s.io/client-go/informers"
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/rest"
"k8s.io/client-go/tools/clientcmd"
)
// These get set at build time via -ldflags magic
var version string
var buildDate string
// KubeRouter holds the information needed to run server
type KubeRouter struct {
Client kubernetes.Interface
Config *options.KubeRouterConfig
}
// NewKubeRouterDefault returns a KubeRouter object
func NewKubeRouterDefault(config *options.KubeRouterConfig) (*KubeRouter, error) {
var clientconfig *rest.Config
var err error
PrintVersion(true)
// Use out of cluster config if the URL or kubeconfig have been specified. Otherwise use incluster config.
if len(config.Master) != 0 || len(config.Kubeconfig) != 0 {
clientconfig, err = clientcmd.BuildConfigFromFlags(config.Master, config.Kubeconfig)
if err != nil {
return nil, errors.New("Failed to build configuration from CLI: " + err.Error())
}
} else {
clientconfig, err = rest.InClusterConfig()
if err != nil {
return nil, errors.New("unable to initialize inclusterconfig: " + err.Error())
}
}
clientset, err := kubernetes.NewForConfig(clientconfig)
if err != nil {
return nil, errors.New("Failed to create Kubernetes client: " + err.Error())
}
return &KubeRouter{Client: clientset, Config: config}, nil
}
// CleanupConfigAndExit performs Cleanup on all three controllers
func CleanupConfigAndExit() {
npc := netpol.NetworkPolicyController{}
npc.Cleanup()
nsc := proxy.NetworkServicesController{}
nsc.Cleanup()
nrc := routing.NetworkRoutingController{}
nrc.Cleanup()
}
// Run starts the controllers and waits forever till we get SIGINT or SIGTERM
func (kr *KubeRouter) Run() error {
var err error
var wg sync.WaitGroup
healthChan := make(chan *healthcheck.ControllerHeartbeat, 10)
defer close(healthChan)
stopCh := make(chan struct{})
if !(kr.Config.RunFirewall || kr.Config.RunServiceProxy || kr.Config.RunRouter) {
glog.Info("Router, Firewall or Service proxy functionality must be specified. Exiting!")
os.Exit(0)
}
hc, err := healthcheck.NewHealthController(kr.Config)
if err != nil {
return errors.New("Failed to create health controller: " + err.Error())
}
wg.Add(1)
go hc.RunServer(stopCh, &wg)
informerFactory := informers.NewSharedInformerFactory(kr.Client, 0)
svcInformer := informerFactory.Core().V1().Services().Informer()
epInformer := informerFactory.Core().V1().Endpoints().Informer()
podInformer := informerFactory.Core().V1().Pods().Informer()
nodeInformer := informerFactory.Core().V1().Nodes().Informer()
nsInformer := informerFactory.Core().V1().Namespaces().Informer()
npInformer := informerFactory.Networking().V1().NetworkPolicies().Informer()
informerFactory.Start(stopCh)
err = kr.CacheSyncOrTimeout(informerFactory, stopCh)
if err != nil {
return errors.New("Failed to synchronize cache: " + err.Error())
}
hc.SetAlive()
wg.Add(1)
go hc.RunCheck(healthChan, stopCh, &wg)
if (kr.Config.MetricsPort > 0) && (kr.Config.MetricsPort <= 65535) {
kr.Config.MetricsEnabled = true
mc, err := metrics.NewMetricsController(kr.Client, kr.Config)
if err != nil {
return errors.New("Failed to create metrics controller: " + err.Error())
}
wg.Add(1)
go mc.Run(healthChan, stopCh, &wg)
} else if kr.Config.MetricsPort > 65535 {
glog.Errorf("Metrics port must be over 0 and under 65535, given port: %d", kr.Config.MetricsPort)
kr.Config.MetricsEnabled = false
} else {
kr.Config.MetricsEnabled = false
}
if kr.Config.BGPGracefulRestart {
if kr.Config.BGPGracefulRestartTime > time.Second*4095 {
return errors.New("BGPGracefuleRestartTime should be less than 4095 seconds")
}
if kr.Config.BGPGracefulRestartTime <= 0 {
return errors.New("BGPGracefuleRestartTime must be positive")
}
if kr.Config.BGPGracefulRestartDeferralTime > time.Hour*18 {
return errors.New("BGPGracefuleRestartDeferralTime should be less than 18 hours")
}
if kr.Config.BGPGracefulRestartDeferralTime <= 0 {
return errors.New("BGPGracefuleRestartDeferralTime must be positive")
}
}
if kr.Config.RunRouter {
nrc, err := routing.NewNetworkRoutingController(kr.Client, kr.Config, nodeInformer, svcInformer, epInformer)
if err != nil {
return errors.New("Failed to create network routing controller: " + err.Error())
}
nodeInformer.AddEventHandler(nrc.NodeEventHandler)
svcInformer.AddEventHandler(nrc.ServiceEventHandler)
epInformer.AddEventHandler(nrc.EndpointsEventHandler)
wg.Add(1)
go nrc.Run(healthChan, stopCh, &wg)
}
if kr.Config.RunServiceProxy {
nsc, err := proxy.NewNetworkServicesController(kr.Client, kr.Config,
svcInformer, epInformer, podInformer)
if err != nil {
return errors.New("Failed to create network services controller: " + err.Error())
}
svcInformer.AddEventHandler(nsc.ServiceEventHandler)
epInformer.AddEventHandler(nsc.EndpointsEventHandler)
wg.Add(1)
go nsc.Run(healthChan, stopCh, &wg)
// wait for the proxy firewall rules to be setup before network policies
if kr.Config.RunFirewall {
nsc.ProxyFirewallSetup.L.Lock()
nsc.ProxyFirewallSetup.Wait()
nsc.ProxyFirewallSetup.L.Unlock()
}
}
if kr.Config.RunFirewall {
npc, err := netpol.NewNetworkPolicyController(kr.Client,
kr.Config, podInformer, npInformer, nsInformer)
if err != nil {
return errors.New("Failed to create network policy controller: " + err.Error())
}
podInformer.AddEventHandler(npc.PodEventHandler)
nsInformer.AddEventHandler(npc.NamespaceEventHandler)
npInformer.AddEventHandler(npc.NetworkPolicyEventHandler)
wg.Add(1)
go npc.Run(healthChan, stopCh, &wg)
}
// Handle SIGINT and SIGTERM
ch := make(chan os.Signal, 1)
signal.Notify(ch, syscall.SIGINT, syscall.SIGTERM)
<-ch
glog.Infof("Shutting down the controllers")
close(stopCh)
wg.Wait()
return nil
}
// CacheSync performs cache synchronization under timeout limit
func (kr *KubeRouter) CacheSyncOrTimeout(informerFactory informers.SharedInformerFactory, stopCh <-chan struct{}) error {
syncOverCh := make(chan struct{})
go func() {
informerFactory.WaitForCacheSync(stopCh)
close(syncOverCh)
}()
select {
case <-time.After(kr.Config.CacheSyncTimeout):
return errors.New(kr.Config.CacheSyncTimeout.String() + " timeout")
case <-syncOverCh:
return nil
}
}
func PrintVersion(logOutput bool) {
output := fmt.Sprintf("Running %v version %s, built on %s, %s\n", os.Args[0], version, buildDate, runtime.Version())
if !logOutput {
fmt.Fprintf(os.Stderr, "%s", output)
} else {
glog.Info(output)
}
}