/
start.go
110 lines (91 loc) · 3.47 KB
/
start.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
/*
Copyright 2019 The OpenEBS 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 volume
import (
"sync"
"time"
"github.com/pkg/errors"
kubeinformers "k8s.io/client-go/informers"
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/rest"
"k8s.io/client-go/tools/clientcmd"
"k8s.io/klog/v2"
clientset "github.com/openebs/device-localpv/pkg/generated/clientset/internalclientset"
informers "github.com/openebs/device-localpv/pkg/generated/informer/externalversions"
)
var (
masterURL string
kubeconfig string
)
// Start starts the devicevolume controller.
func Start(controllerMtx *sync.RWMutex, stopCh <-chan struct{}) error {
// Get in cluster config
cfg, err := getClusterConfig(kubeconfig)
if err != nil {
return errors.Wrap(err, "error building kubeconfig")
}
// Building Kubernetes Clientset
kubeClient, err := kubernetes.NewForConfig(cfg)
if err != nil {
return errors.Wrap(err, "error building kubernetes clientset")
}
// Building OpenEBS Clientset
openebsClient, err := clientset.NewForConfig(cfg)
if err != nil {
return errors.Wrap(err, "error building openebs clientset")
}
kubeInformerFactory := kubeinformers.NewSharedInformerFactory(kubeClient, time.Second*30)
VolInformerFactory := informers.NewSharedInformerFactory(openebsClient, time.Second*30)
// Build() fn of all controllers calls AddToScheme to adds all types of this
// clientset into the given scheme.
// If multiple controllers happen to call this AddToScheme same time,
// it causes panic with error saying concurrent map access.
// This lock is used to serialize the AddToScheme call of all controllers.
controllerMtx.Lock()
controller, err := NewVolControllerBuilder().
withKubeClient(kubeClient).
withOpenEBSClient(openebsClient).
withVolSynced(VolInformerFactory).
withVolLister(VolInformerFactory).
withRecorder(kubeClient).
withEventHandler(VolInformerFactory).
withWorkqueueRateLimiting().Build()
// blocking call, can't use defer to release the lock
controllerMtx.Unlock()
if err != nil {
return errors.Wrapf(err, "error building controller instance")
}
go kubeInformerFactory.Start(stopCh)
go VolInformerFactory.Start(stopCh)
// Threadiness defines the number of workers to be launched in Run function
// The no.of threads is set to 1 here as using `parted` command for creation/deletion
// of partitions from multiple threads can lead to race condition and eventually some
// partitions not getting cleaned up from the disk.
// Ref: https://github.com/openebs/device-localpv/issues/21
return controller.Run(1, stopCh)
}
// GetClusterConfig return the config for k8s.
func getClusterConfig(kubeconfig string) (*rest.Config, error) {
cfg, err := rest.InClusterConfig()
if err != nil {
klog.Errorf("Failed to get k8s Incluster config. %+v", err)
if kubeconfig == "" {
return nil, errors.Wrap(err, "kubeconfig is empty")
}
cfg, err = clientcmd.BuildConfigFromFlags(masterURL, kubeconfig)
if err != nil {
return nil, errors.Wrap(err, "error building kubeconfig")
}
}
return cfg, err
}