forked from openfaas/faas-netes
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathmain.go
233 lines (190 loc) · 8.55 KB
/
main.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
// Copyright (c) Alex Ellis 2017. All rights reserved.
// Copyright (c) OpenFaaS Author(s) 2020. All rights reserved.
// Licensed under the MIT license. See LICENSE file in the project root for full license information.
package main
import (
"flag"
"fmt"
"log"
"os"
"time"
clientset "github.com/openfaas/faas-netes/pkg/client/clientset/versioned"
informers "github.com/openfaas/faas-netes/pkg/client/informers/externalversions"
v1 "github.com/openfaas/faas-netes/pkg/client/informers/externalversions/openfaas/v1"
"github.com/openfaas/faas-netes/pkg/config"
"github.com/openfaas/faas-netes/pkg/controller"
"github.com/openfaas/faas-netes/pkg/handlers"
"github.com/openfaas/faas-netes/pkg/k8s"
"github.com/openfaas/faas-netes/pkg/signals"
version "github.com/openfaas/faas-netes/version"
faasProvider "github.com/openfaas/faas-provider"
"github.com/openfaas/faas-provider/logs"
"github.com/openfaas/faas-provider/proxy"
providertypes "github.com/openfaas/faas-provider/types"
kubeinformers "k8s.io/client-go/informers"
v1apps "k8s.io/client-go/informers/apps/v1"
v1core "k8s.io/client-go/informers/core/v1"
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/tools/cache"
"k8s.io/client-go/tools/clientcmd"
"k8s.io/klog"
// required to authenticate against GKE clusters
_ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
// required for updating and validating the CRD clientset
_ "k8s.io/code-generator/cmd/client-gen/generators"
// main.go:36:2: import "sigs.k8s.io/controller-tools/cmd/controller-gen" is a program, not an importable package
// _ "sigs.k8s.io/controller-tools/cmd/controller-gen"
)
func main() {
var kubeconfig string
var masterURL string
var (
operator,
verbose bool
)
flag.StringVar(&kubeconfig, "kubeconfig", "",
"Path to a kubeconfig. Only required if out-of-cluster.")
flag.BoolVar(&verbose, "verbose", false, "Print verbose config information")
flag.StringVar(&masterURL, "master", "",
"The address of the Kubernetes API server. Overrides any value in kubeconfig. Only required if out-of-cluster.")
flag.BoolVar(&operator, "operator", false, "Use the operator mode instead of faas-netes")
flag.Parse()
if operator {
klog.Errorf("The operator mode is deprecated in OpenFaaS Community Edition (CE), upgrade to OpenFaaS Pro to continue using it")
os.Exit(1)
}
mode := "controller"
sha, release := version.GetReleaseInfo()
fmt.Printf("faas-netes - Community Edition (CE)\n"+
"\nVersion: %s Commit: %s Mode: %s\n", release, sha, mode)
clientCmdConfig, err := clientcmd.BuildConfigFromFlags(masterURL, kubeconfig)
if err != nil {
log.Fatalf("Error building kubeconfig: %s", err.Error())
}
kubeconfigQPS := 100
kubeconfigBurst := 250
clientCmdConfig.QPS = float32(kubeconfigQPS)
clientCmdConfig.Burst = kubeconfigBurst
kubeClient, err := kubernetes.NewForConfig(clientCmdConfig)
if err != nil {
log.Fatalf("Error building Kubernetes clientset: %s", err.Error())
}
faasClient, err := clientset.NewForConfig(clientCmdConfig)
if err != nil {
log.Fatalf("Error building OpenFaaS clientset: %s", err.Error())
}
readConfig := config.ReadConfig{}
osEnv := providertypes.OsEnv{}
config, err := readConfig.Read(osEnv)
if err != nil {
log.Fatalf("Error reading config: %s", err.Error())
}
config.Fprint(verbose)
deployConfig := k8s.DeploymentConfig{
RuntimeHTTPPort: 8080,
HTTPProbe: config.HTTPProbe,
SetNonRootUser: config.SetNonRootUser,
ReadinessProbe: &k8s.ProbeConfig{
InitialDelaySeconds: int32(config.ReadinessProbeInitialDelaySeconds),
TimeoutSeconds: int32(config.ReadinessProbeTimeoutSeconds),
PeriodSeconds: int32(config.ReadinessProbePeriodSeconds),
},
LivenessProbe: &k8s.ProbeConfig{
InitialDelaySeconds: int32(config.LivenessProbeInitialDelaySeconds),
TimeoutSeconds: int32(config.LivenessProbeTimeoutSeconds),
PeriodSeconds: int32(config.LivenessProbePeriodSeconds),
},
ImagePullPolicy: config.ImagePullPolicy,
ProfilesNamespace: config.ProfilesNamespace,
}
// the sync interval does not affect the scale to/from zero feature
// auto-scaling is does via the HTTP API that acts on the deployment Spec.Replicas
defaultResync := time.Minute * 5
namespaceScope := config.DefaultFunctionNamespace
if namespaceScope == "" {
klog.Fatal("DefaultFunctionNamespace must be set")
}
kubeInformerOpt := kubeinformers.WithNamespace(namespaceScope)
kubeInformerFactory := kubeinformers.NewSharedInformerFactoryWithOptions(kubeClient, defaultResync, kubeInformerOpt)
faasInformerOpt := informers.WithNamespace(namespaceScope)
faasInformerFactory := informers.NewSharedInformerFactoryWithOptions(faasClient, defaultResync, faasInformerOpt)
factory := k8s.NewFunctionFactory(kubeClient, deployConfig, faasClient.OpenfaasV1())
setup := serverSetup{
config: config,
functionFactory: factory,
kubeInformerFactory: kubeInformerFactory,
faasInformerFactory: faasInformerFactory,
kubeClient: kubeClient,
faasClient: faasClient,
}
runController(setup)
}
type customInformers struct {
EndpointsInformer v1core.EndpointsInformer
DeploymentInformer v1apps.DeploymentInformer
FunctionsInformer v1.FunctionInformer
}
func startInformers(setup serverSetup, stopCh <-chan struct{}, operator bool) customInformers {
kubeInformerFactory := setup.kubeInformerFactory
faasInformerFactory := setup.faasInformerFactory
var functions v1.FunctionInformer
if operator {
functions = faasInformerFactory.Openfaas().V1().Functions()
go functions.Informer().Run(stopCh)
if ok := cache.WaitForNamedCacheSync("faas-netes:functions", stopCh, functions.Informer().HasSynced); !ok {
log.Fatalf("failed to wait for cache to sync")
}
}
deployments := kubeInformerFactory.Apps().V1().Deployments()
go deployments.Informer().Run(stopCh)
if ok := cache.WaitForNamedCacheSync("faas-netes:deployments", stopCh, deployments.Informer().HasSynced); !ok {
log.Fatalf("failed to wait for cache to sync")
}
endpoints := kubeInformerFactory.Core().V1().Endpoints()
go endpoints.Informer().Run(stopCh)
if ok := cache.WaitForNamedCacheSync("faas-netes:endpoints", stopCh, endpoints.Informer().HasSynced); !ok {
log.Fatalf("failed to wait for cache to sync")
}
return customInformers{
EndpointsInformer: endpoints,
DeploymentInformer: deployments,
FunctionsInformer: functions,
}
}
// runController runs the faas-netes imperative controller
func runController(setup serverSetup) {
config := setup.config
kubeClient := setup.kubeClient
factory := setup.functionFactory
// set up signals so we handle the first shutdown signal gracefully
stopCh := signals.SetupSignalHandler()
operator := false
listers := startInformers(setup, stopCh, operator)
controller.RegisterEventHandlers(listers.DeploymentInformer, kubeClient, config.DefaultFunctionNamespace)
functionLookup := k8s.NewFunctionLookup(config.DefaultFunctionNamespace, listers.EndpointsInformer.Lister())
bootstrapHandlers := providertypes.FaaSHandlers{
FunctionProxy: proxy.NewHandlerFunc(config.FaaSConfig, functionLookup),
DeleteHandler: handlers.MakeDeleteHandler(config.DefaultFunctionNamespace, kubeClient),
DeployHandler: handlers.MakeDeployHandler(config.DefaultFunctionNamespace, factory),
FunctionReader: handlers.MakeFunctionReader(config.DefaultFunctionNamespace, listers.DeploymentInformer.Lister()),
ReplicaReader: handlers.MakeReplicaReader(config.DefaultFunctionNamespace, listers.DeploymentInformer.Lister()),
ReplicaUpdater: handlers.MakeReplicaUpdater(config.DefaultFunctionNamespace, kubeClient),
UpdateHandler: handlers.MakeUpdateHandler(config.DefaultFunctionNamespace, factory),
HealthHandler: handlers.MakeHealthHandler(),
InfoHandler: handlers.MakeInfoHandler(version.BuildVersion(), version.GitCommit),
SecretHandler: handlers.MakeSecretHandler(config.DefaultFunctionNamespace, kubeClient),
LogHandler: logs.NewLogHandlerFunc(k8s.NewLogRequestor(kubeClient, config.DefaultFunctionNamespace), config.FaaSConfig.WriteTimeout),
ListNamespaceHandler: handlers.MakeNamespacesLister(config.DefaultFunctionNamespace, kubeClient),
}
faasProvider.Serve(&bootstrapHandlers, &config.FaaSConfig)
}
// serverSetup is a container for the config and clients needed to start the
// faas-netes controller or operator
type serverSetup struct {
config config.BootstrapConfig
kubeClient *kubernetes.Clientset
faasClient *clientset.Clientset
functionFactory k8s.FunctionFactory
kubeInformerFactory kubeinformers.SharedInformerFactory
faasInformerFactory informers.SharedInformerFactory
}