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
|
package agent
import (
"context"
"fmt"
"net/http"
"time"
notificationv1 "github.com/fluxcd/notification-controller/api/v1"
sourcev1 "github.com/fluxcd/source-controller/api/v1"
"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v16/internal/module/flux"
"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v16/internal/module/flux/rpc"
"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v16/internal/module/modagent"
"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v16/internal/module/modshared"
"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v16/internal/tool/logz"
"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v16/internal/tool/retry"
"go.uber.org/zap"
apiextensionsv1api "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
apiextensionsv1client "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset/typed/apiextensions/v1"
kubeerrors "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime/schema"
"k8s.io/client-go/dynamic"
"k8s.io/client-go/dynamic/dynamicinformer"
"k8s.io/client-go/informers"
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/tools/cache"
"k8s.io/client-go/transport"
)
const (
// resyncDuration defines the duration for the shared informer cache resync interval.
resyncDuration = 10 * time.Minute
reconcileProjectsInitBackoff = 10 * time.Second
reconcileProjectsMaxBackoff = 5 * time.Minute
reconcileProjectsResetDuration = 10 * time.Minute
reconcileProjectsBackoffFactor = 2.0
reconcileProjectsJitter = 1.0
)
var (
requiredFluxCRDs = [...]schema.GroupVersionResource{
sourcev1.GroupVersion.WithResource("gitrepositories"),
notificationv1.GroupVersion.WithResource("receivers"),
}
)
type Factory struct{}
func (f *Factory) IsProducingLeaderModules() bool {
return true
}
func (f *Factory) New(config *modagent.Config) (modagent.Module, error) {
restConfig, err := config.K8sUtilFactory.ToRESTConfig()
if err != nil {
return nil, err
}
extAPIClient, err := apiextensionsv1client.NewForConfig(restConfig)
if err != nil {
return nil, err
}
if !isFluxInstalled(context.Background(), config.Log, extAPIClient) {
config.Log.Info("Flux is not installed, skipping module. A restart is required for this to be checked again")
return nil, nil
}
dynamicClient, err := dynamic.NewForConfig(restConfig)
if err != nil {
return nil, err
}
clientset, err := kubernetes.NewForConfig(restConfig)
if err != nil {
return nil, err
}
receiverClient := dynamicClient.Resource(notificationv1.GroupVersion.WithResource("receivers"))
kubeAPIURL, _, err := defaultServerURLFor(restConfig)
if err != nil {
return nil, err
}
transportCfg, err := restConfig.TransportConfig()
if err != nil {
return nil, err
}
kubeAPIRoundTripper, err := transport.New(transportCfg)
if err != nil {
return nil, err
}
return &module{
log: config.Log,
k8sExtAPIClient: extAPIClient,
runner: &fluxReconciliationRunner{
informersFactory: func() (informers.GenericInformer, informers.GenericInformer, cache.Indexer) {
informerFactory := dynamicinformer.NewDynamicSharedInformerFactory(dynamicClient, resyncDuration)
gitRepositoryInformer := informerFactory.ForResource(sourcev1.GroupVersion.WithResource("gitrepositories"))
receiverInformer := informerFactory.ForResource(notificationv1.GroupVersion.WithResource("receivers"))
receiverIndexer := receiverInformer.Informer().GetIndexer()
return gitRepositoryInformer, receiverInformer, receiverIndexer
},
clientFactory: func(ctx context.Context, cfgURL string, receiverIndexer cache.Indexer) (*client, error) {
agentID, err := config.API.GetAgentID(ctx)
if err != nil {
return nil, err
}
rt, err := newGitRepositoryReconcileTrigger(cfgURL, kubeAPIURL, kubeAPIRoundTripper, http.DefaultTransport)
if err != nil {
return nil, err
}
return newClient(
config.Log,
config.API,
agentID,
rpc.NewGitLabFluxClient(config.KASConn),
retry.NewPollConfigFactory(0, retry.NewExponentialBackoffFactory(
reconcileProjectsInitBackoff, reconcileProjectsMaxBackoff, reconcileProjectsResetDuration, reconcileProjectsBackoffFactor, reconcileProjectsJitter),
),
receiverIndexer,
rt,
)
},
controllerFactory: func(ctx context.Context, gitRepositoryInformer informers.GenericInformer, receiverInformer informers.GenericInformer, projectReconciler projectReconciler) (controller, error) {
agentID, err := config.API.GetAgentID(ctx)
if err != nil {
return nil, err
}
gitLabExternalURL, err := config.API.GetGitLabExternalURL(ctx)
if err != nil {
return nil, err
}
return newGitRepositoryController(ctx, config.Log, config.API, agentID, gitLabExternalURL, gitRepositoryInformer, receiverInformer, projectReconciler, receiverClient, clientset.CoreV1())
},
},
}, nil
}
func (f *Factory) Name() string {
return flux.ModuleName
}
func (f *Factory) StartStopPhase() modshared.ModuleStartStopPhase {
return modshared.ModuleStartBeforeServers
}
func isFluxInstalled(ctx context.Context, log *zap.Logger, client apiextensionsv1client.ApiextensionsV1Interface) bool {
for _, crd := range requiredFluxCRDs {
ok, err := checkCRDExistsAndEstablished(ctx, client, crd)
if err != nil {
log.Error("Unable to check if CRD is installed", logz.K8sGroup(crd.Group), logz.Error(err))
return false
}
if !ok {
log.Debug("Required Flux CRD is not established", logz.K8sResource(crd.String()))
return false
}
}
return true
}
func checkCRDExistsAndEstablished(ctx context.Context, client apiextensionsv1client.ApiextensionsV1Interface, crd schema.GroupVersionResource) (bool, error) {
obj, err := client.CustomResourceDefinitions().Get(ctx, crd.GroupResource().String(), metav1.GetOptions{})
if err != nil {
if kubeerrors.IsNotFound(err) {
return false, nil
}
return false, fmt.Errorf("unable to get CRD %s: %w", crd.String(), err)
}
if !isCRDVersionServed(obj, crd.Version) {
return false, nil
}
return isCRDEstablished(obj), nil
}
func isCRDVersionServed(obj *apiextensionsv1api.CustomResourceDefinition, version string) bool {
// Check if the appropriate API version of the CRD is served.
for _, v := range obj.Spec.Versions {
if v.Served && v.Name == version {
return true
}
}
return false
}
func isCRDEstablished(obj *apiextensionsv1api.CustomResourceDefinition) bool {
for _, cond := range obj.Status.Conditions {
switch cond.Type { //nolint:exhaustive
case apiextensionsv1api.Established:
if cond.Status == apiextensionsv1api.ConditionTrue {
return true
}
// we don't really care about any other conditions for now, because we don't own this CRD
// and expect the owner to make sure it becomes established.
}
}
return false
}
|
