-
Notifications
You must be signed in to change notification settings - Fork 479
/
Copy pathmain.go
242 lines (197 loc) · 8.44 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
234
235
236
237
238
239
240
241
242
// License: OpenFaaS Community Edition (CE) EULA
// Copyright (c) 2017,2019-2024 OpenFaaS Author(s)
// Copyright (c) Alex Ellis 2017. All rights reserved.
// Copyright (c) OpenFaaS Author(s) 2020. All rights reserved.
package main
import (
"context"
"flag"
"fmt"
"log"
"net/http"
"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/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"
// 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"
)
const defaultResync = time.Hour * 10
func main() {
var kubeconfig string
var masterURL string
var (
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.Bool("operator", false, "Run as an operator (not available in CE)")
flag.Parse()
sha, release := version.GetReleaseInfo()
fmt.Printf(`faas-netes - Community Edition (CE)
Warning: Commercial use limited to 60 days.
Learn more: https://github.com/openfaas/faas/blob/master/EULA.md
Version: %s Commit: %s
`, release, sha)
if err := config.ConnectivityCheck(); err != nil {
log.Fatalf("Error checking connectivity, OpenFaaS CE cannot be run in an offline environment: %s", err.Error())
}
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(2),
TimeoutSeconds: int32(1),
PeriodSeconds: int32(2),
},
LivenessProbe: &k8s.ProbeConfig{
InitialDelaySeconds: int32(2),
TimeoutSeconds: int32(1),
PeriodSeconds: int32(2),
},
}
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)
handlers.RegisterEventHandlers(listers.DeploymentInformer, kubeClient, config.DefaultFunctionNamespace)
deployLister := listers.DeploymentInformer.Lister()
functionLookup := k8s.NewFunctionLookup(config.DefaultFunctionNamespace, listers.EndpointsInformer.Lister())
functionList := k8s.NewFunctionList(config.DefaultFunctionNamespace, deployLister)
printFunctionExecutionTime := true
proxyHandler := proxy.NewHandlerFunc(config.FaaSConfig, functionLookup, printFunctionExecutionTime)
if err := handlers.Check(functionList); err != nil {
msg := fmt.Sprintf("Function invocations disabled due to error: %s.", err.Error())
log.Print(msg)
proxyHandler = func(w http.ResponseWriter, r *http.Request) {
http.Error(w, msg, http.StatusMethodNotAllowed)
}
}
bootstrapHandlers := providertypes.FaaSHandlers{
FunctionProxy: proxyHandler,
DeleteFunction: handlers.MakeDeleteHandler(config.DefaultFunctionNamespace, kubeClient),
DeployFunction: handlers.MakeDeployHandler(config.DefaultFunctionNamespace, factory, functionList),
FunctionLister: handlers.MakeFunctionReader(config.DefaultFunctionNamespace, deployLister),
FunctionStatus: handlers.MakeReplicaReader(config.DefaultFunctionNamespace, deployLister),
ScaleFunction: handlers.MakeReplicaUpdater(config.DefaultFunctionNamespace, kubeClient),
UpdateFunction: handlers.MakeUpdateHandler(config.DefaultFunctionNamespace, factory),
Health: handlers.MakeHealthHandler(),
Info: handlers.MakeInfoHandler(version.BuildVersion(), version.GitCommit),
Secrets: handlers.MakeSecretHandler(config.DefaultFunctionNamespace, kubeClient),
Logs: logs.NewLogHandlerFunc(k8s.NewLogRequestor(kubeClient, config.DefaultFunctionNamespace), config.FaaSConfig.WriteTimeout),
ListNamespaces: handlers.MakeNamespacesLister(config.DefaultFunctionNamespace, kubeClient),
}
ctx := context.Background()
faasProvider.Serve(ctx, &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
}