/
operator.go
318 lines (270 loc) · 11.6 KB
/
operator.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
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
/*
Licensed to the Apache Software Foundation (ASF) under one or more
contributor license agreements. See the NOTICE file distributed with
this work for additional information regarding copyright ownership.
The ASF licenses this file to You 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 operator
import (
"context"
"flag"
"fmt"
"os"
"reflect"
"runtime"
"strconv"
"strings"
"time"
"k8s.io/klog/v2"
"go.uber.org/automaxprocs/maxprocs"
"go.uber.org/zap"
"go.uber.org/zap/zapcore"
appsv1 "k8s.io/api/apps/v1"
batchv1 "k8s.io/api/batch/v1"
coordination "k8s.io/api/coordination/v1"
corev1 "k8s.io/api/core/v1"
k8serrors "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/labels"
"k8s.io/apimachinery/pkg/selection"
"k8s.io/client-go/tools/leaderelection/resourcelock"
"k8s.io/client-go/tools/record"
"sigs.k8s.io/controller-runtime/pkg/cache"
ctrl "sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/client/config"
"sigs.k8s.io/controller-runtime/pkg/healthz"
logf "sigs.k8s.io/controller-runtime/pkg/log"
zapctrl "sigs.k8s.io/controller-runtime/pkg/log/zap"
"sigs.k8s.io/controller-runtime/pkg/manager"
"sigs.k8s.io/controller-runtime/pkg/manager/signals"
servingv1 "knative.dev/serving/pkg/apis/serving/v1"
"github.com/apache/camel-k/v2/pkg/apis"
v1 "github.com/apache/camel-k/v2/pkg/apis/camel/v1"
"github.com/apache/camel-k/v2/pkg/client"
"github.com/apache/camel-k/v2/pkg/controller"
"github.com/apache/camel-k/v2/pkg/controller/synthetic"
"github.com/apache/camel-k/v2/pkg/event"
"github.com/apache/camel-k/v2/pkg/install"
"github.com/apache/camel-k/v2/pkg/platform"
"github.com/apache/camel-k/v2/pkg/util/defaults"
"github.com/apache/camel-k/v2/pkg/util/kubernetes"
logutil "github.com/apache/camel-k/v2/pkg/util/log"
)
var log = logutil.Log.WithName("cmd")
func printVersion() {
log.Info(fmt.Sprintf("Go Version: %s", runtime.Version()))
log.Info(fmt.Sprintf("Go OS/Arch: %s/%s", runtime.GOOS, runtime.GOARCH))
log.Info(fmt.Sprintf("Camel K Operator Version: %v", defaults.Version))
log.Info(fmt.Sprintf("Camel K Default Runtime Version: %v", defaults.DefaultRuntimeVersion))
log.Info(fmt.Sprintf("Camel K Git Commit: %v", defaults.GitCommit))
log.Info(fmt.Sprintf("Camel K Operator ID: %v", defaults.OperatorID()))
// Will only appear if DEBUG level has been enabled using the env var LOG_LEVEL
log.Debug("*** DEBUG level messages will be logged ***")
}
// Run starts the Camel K operator.
func Run(healthPort, monitoringPort int32, leaderElection bool, leaderElectionID string) {
flag.Parse()
// The logger instantiated here can be changed to any logger
// implementing the logr.Logger interface. This logger will
// be propagated through the whole operator, generating
// uniform and structured logs.
// The constants specified here are zap specific
var logLevel zapcore.Level
logLevelVal, ok := os.LookupEnv("LOG_LEVEL")
if ok {
switch strings.ToLower(logLevelVal) {
case "error":
logLevel = zapcore.ErrorLevel
case "info":
logLevel = zapcore.InfoLevel
case "debug":
logLevel = zapcore.DebugLevel
default:
customLevel, err := strconv.Atoi(strings.ToLower(logLevelVal))
exitOnError(err, "Invalid log-level")
// Need to multiply by -1 to turn logr expected level into zap level
logLevel = zapcore.Level(int8(customLevel) * -1)
}
} else {
logLevel = zapcore.InfoLevel
}
// Use and set atomic level that all following log events are compared with
// in order to evaluate if a given log level on the event is enabled.
logf.SetLogger(zapctrl.New(func(o *zapctrl.Options) {
o.Development = false
o.Level = zap.NewAtomicLevelAt(logLevel)
}))
klog.SetLogger(log.AsLogger())
_, err := maxprocs.Set(maxprocs.Logger(func(f string, a ...interface{}) { log.Info(fmt.Sprintf(f, a)) }))
if err != nil {
log.Error(err, "failed to set GOMAXPROCS from cgroups")
}
printVersion()
watchNamespace, err := getWatchNamespace()
exitOnError(err, "failed to get watch namespace")
ctx := signals.SetupSignalHandler()
cfg, err := config.GetConfig()
exitOnError(err, "cannot get client config")
// Increase maximum burst that is used by client-side throttling,
// to prevent the requests made to apply the bundled Kamelets
// from being throttled.
cfg.QPS = 20
cfg.Burst = 200
bootstrapClient, err := client.NewClientWithConfig(false, cfg)
exitOnError(err, "cannot initialize client")
// We do not rely on the event broadcaster managed by controller runtime,
// so that we can check the operator has been granted permission to create
// Events. This is required for the operator to be installable by standard
// admin users, that are not granted create permission on Events by default.
broadcaster := record.NewBroadcaster()
defer broadcaster.Shutdown()
if ok, err := kubernetes.CheckPermission(ctx, bootstrapClient, corev1.GroupName, "events", watchNamespace, "", "create"); err != nil || !ok {
// Do not sink Events to the server as they'll be rejected
broadcaster = event.NewSinkLessBroadcaster(broadcaster)
exitOnError(err, "cannot check permissions for creating Events")
log.Info("Event broadcasting is disabled because of missing permissions to create Events")
}
operatorNamespace := platform.GetOperatorNamespace()
if operatorNamespace == "" {
// Fallback to using the watch namespace when the operator is not in-cluster.
// It does not support local (off-cluster) operator watching resources globally,
// in which case it's not possible to determine a namespace.
operatorNamespace = watchNamespace
if operatorNamespace == "" {
leaderElection = false
log.Info("unable to determine namespace for leader election")
}
}
// Set the operator container image if it runs in-container
platform.OperatorImage, err = getOperatorImage(ctx, bootstrapClient)
exitOnError(err, "cannot get operator container image")
if ok, err := kubernetes.CheckPermission(ctx, bootstrapClient, coordination.GroupName, "leases", operatorNamespace, "", "create"); err != nil || !ok {
leaderElection = false
exitOnError(err, "cannot check permissions for creating Leases")
log.Info("The operator is not granted permissions to create Leases")
}
if !leaderElection {
log.Info("Leader election is disabled!")
}
hasIntegrationLabel, err := labels.NewRequirement(v1.IntegrationLabel, selection.Exists, []string{})
exitOnError(err, "cannot create Integration label selector")
selector := labels.NewSelector().Add(*hasIntegrationLabel)
selectors := map[ctrl.Object]cache.ByObject{
&corev1.Pod{}: {Label: selector},
&appsv1.Deployment{}: {Label: selector},
&batchv1.Job{}: {Label: selector},
&servingv1.Service{}: {Label: selector},
}
if ok, err := kubernetes.IsAPIResourceInstalled(bootstrapClient, batchv1.SchemeGroupVersion.String(), reflect.TypeOf(batchv1.CronJob{}).Name()); ok && err == nil {
selectors[&batchv1.CronJob{}] = cache.ByObject{
Label: selector,
}
}
options := cache.Options{
ByObject: selectors,
}
mgr, err := manager.New(cfg, manager.Options{
Namespace: watchNamespace,
EventBroadcaster: broadcaster,
LeaderElection: leaderElection,
LeaderElectionNamespace: operatorNamespace,
LeaderElectionID: leaderElectionID,
LeaderElectionResourceLock: resourcelock.LeasesResourceLock,
LeaderElectionReleaseOnCancel: true,
HealthProbeBindAddress: ":" + strconv.Itoa(int(healthPort)),
MetricsBindAddress: ":" + strconv.Itoa(int(monitoringPort)),
Cache: options,
})
exitOnError(err, "")
log.Info("Configuring manager")
exitOnError(mgr.AddHealthzCheck("health-probe", healthz.Ping), "Unable add liveness check")
exitOnError(apis.AddToScheme(mgr.GetScheme()), "")
ctrlClient, err := client.FromManager(mgr)
exitOnError(err, "")
exitOnError(controller.AddToManager(ctx, mgr, ctrlClient), "")
log.Info("Installing operator resources")
installCtx, installCancel := context.WithTimeout(ctx, 1*time.Minute)
defer installCancel()
install.OperatorStartupOptionalTools(installCtx, bootstrapClient, watchNamespace, operatorNamespace, log)
exitOnError(findOrCreateIntegrationPlatform(installCtx, bootstrapClient, operatorNamespace), "failed to create integration platform")
synthEnvVal, synth := os.LookupEnv("CAMEL_K_SYNTHETIC_INTEGRATIONS")
if synth && synthEnvVal == "true" {
log.Info("Starting the synthetic Integration manager")
exitOnError(synthetic.ManageSyntheticIntegrations(ctx, ctrlClient, mgr.GetCache()), "synthetic Integration manager error")
} else {
log.Info("Synthetic Integration manager not configured, skipping")
}
log.Info("Starting the manager")
exitOnError(mgr.Start(ctx), "manager exited non-zero")
}
// findOrCreateIntegrationPlatform create default integration platform in operator namespace if not already exists.
func findOrCreateIntegrationPlatform(ctx context.Context, c client.Client, operatorNamespace string) error {
operatorID := defaults.OperatorID()
var platformName string
if operatorID != "" {
platformName = operatorID
} else {
platformName = platform.DefaultPlatformName
}
if pl, err := kubernetes.GetIntegrationPlatform(ctx, c, platformName, operatorNamespace); pl == nil || k8serrors.IsNotFound(err) {
defaultPlatform := v1.NewIntegrationPlatform(operatorNamespace, platformName)
if defaultPlatform.Labels == nil {
defaultPlatform.Labels = make(map[string]string)
}
defaultPlatform.Labels["app"] = "camel-k"
defaultPlatform.Labels["camel.apache.org/platform.generated"] = "true"
if operatorID != "" {
defaultPlatform.SetOperatorID(operatorID)
}
if _, err := c.CamelV1().IntegrationPlatforms(operatorNamespace).Create(ctx, &defaultPlatform, metav1.CreateOptions{}); err != nil {
return err
}
// Make sure that IntegrationPlatform installed in operator namespace can be seen by others
if err := install.IntegrationPlatformViewerRole(ctx, c, operatorNamespace); err != nil && !k8serrors.IsAlreadyExists(err) {
return fmt.Errorf("error while installing global IntegrationPlatform viewer role: %w", err)
}
} else {
return err
}
return nil
}
// getWatchNamespace returns the Namespace the operator should be watching for changes.
func getWatchNamespace() (string, error) {
ns, found := os.LookupEnv(platform.OperatorWatchNamespaceEnvVariable)
if !found {
return "", fmt.Errorf("%s must be set", platform.OperatorWatchNamespaceEnvVariable)
}
return ns, nil
}
// getOperatorImage returns the image currently used by the running operator if present (when running out of cluster, it may be absent).
func getOperatorImage(ctx context.Context, c ctrl.Reader) (string, error) {
ns := platform.GetOperatorNamespace()
name := platform.GetOperatorPodName()
if ns == "" || name == "" {
return "", nil
}
pod := corev1.Pod{}
if err := c.Get(ctx, ctrl.ObjectKey{Namespace: ns, Name: name}, &pod); err != nil && k8serrors.IsNotFound(err) {
return "", nil
} else if err != nil {
return "", err
}
if len(pod.Spec.Containers) == 0 {
return "", fmt.Errorf("no containers found in operator pod")
}
return pod.Spec.Containers[0].Image, nil
}
func exitOnError(err error, msg string) {
if err != nil {
log.Error(err, msg)
os.Exit(1)
}
}