/
state.go
365 lines (336 loc) · 12.8 KB
/
state.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
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
package cluster
import (
"context"
"encoding/json"
"fmt"
"io"
"os"
"path"
"path/filepath"
"strings"
"time"
"k8s.io/client-go/transport"
"github.com/rancher/rke/hosts"
"github.com/rancher/rke/k8s"
"github.com/rancher/rke/log"
"github.com/rancher/rke/pki"
"github.com/rancher/rke/services"
v3 "github.com/rancher/rke/types"
"github.com/sirupsen/logrus"
"gopkg.in/yaml.v2"
v1 "k8s.io/api/core/v1"
)
const (
stateFileExt = ".rkestate"
certDirExt = "_certs"
)
type FullState struct {
DesiredState State `json:"desiredState,omitempty"`
CurrentState State `json:"currentState,omitempty"`
}
type State struct {
RancherKubernetesEngineConfig *v3.RancherKubernetesEngineConfig `json:"rkeConfig,omitempty"`
CertificatesBundle map[string]pki.CertificatePKI `json:"certificatesBundle,omitempty"`
EncryptionConfig string `json:"encryptionConfig,omitempty"`
}
func (c *Cluster) UpdateClusterCurrentState(ctx context.Context, fullState *FullState) error {
fullState.CurrentState.RancherKubernetesEngineConfig = c.RancherKubernetesEngineConfig.DeepCopy()
fullState.CurrentState.CertificatesBundle = c.Certificates
fullState.CurrentState.EncryptionConfig = c.EncryptionConfig.EncryptionProviderFile
return fullState.WriteStateFile(ctx, c.StateFilePath)
}
func (c *Cluster) GetClusterState(ctx context.Context, fullState *FullState) (*Cluster, error) {
var err error
if fullState.CurrentState.RancherKubernetesEngineConfig == nil {
return nil, nil
}
// resetup external flags
flags := GetExternalFlags(false, false, false, false, c.ConfigDir, c.ConfigPath)
currentCluster, err := InitClusterObject(ctx, fullState.CurrentState.RancherKubernetesEngineConfig, flags, fullState.CurrentState.EncryptionConfig)
if err != nil {
return nil, err
}
currentCluster.Certificates = fullState.CurrentState.CertificatesBundle
currentCluster.EncryptionConfig.EncryptionProviderFile = fullState.CurrentState.EncryptionConfig
// resetup dialers
dialerOptions := hosts.GetDialerOptions(c.DockerDialerFactory, c.LocalConnDialerFactory, c.K8sWrapTransport)
if err := currentCluster.SetupDialers(ctx, dialerOptions); err != nil {
return nil, err
}
return currentCluster, nil
}
func (c *Cluster) GetStateFileFromConfigMap(ctx context.Context) (string, error) {
kubeletImage := c.Services.Kubelet.Image
for _, host := range c.ControlPlaneHosts {
stateFile, err := services.RunGetStateFileFromConfigMap(ctx, host, c.PrivateRegistriesMap, kubeletImage, c.Version)
if err != nil || stateFile == "" {
logrus.Infof("Could not get ConfigMap with cluster state from host [%s]", host.Address)
continue
}
return stateFile, nil
}
return "", fmt.Errorf("Unable to get ConfigMap with cluster state from any Control Plane host")
}
func SaveFullStateToKubernetes(ctx context.Context, kubeCluster *Cluster, fullState *FullState) error {
k8sClient, err := k8s.NewClient(kubeCluster.LocalKubeConfigPath, kubeCluster.K8sWrapTransport)
if err != nil {
return fmt.Errorf("Failed to create Kubernetes Client: %v", err)
}
log.Infof(ctx, "[state] Saving full cluster state to Kubernetes")
stateFile, err := json.Marshal(*fullState)
if err != nil {
return err
}
timeout := make(chan bool, 1)
go func() {
for {
_, err := k8s.UpdateConfigMap(k8sClient, stateFile, FullStateConfigMapName)
if err != nil {
time.Sleep(time.Second * 5)
continue
}
log.Infof(ctx, "[state] Successfully Saved full cluster state to Kubernetes ConfigMap: %s", FullStateConfigMapName)
timeout <- true
break
}
}()
select {
case <-timeout:
return nil
case <-time.After(time.Second * UpdateStateTimeout):
return fmt.Errorf("[state] Timeout waiting for kubernetes to be ready")
}
}
func GetStateFromKubernetes(ctx context.Context, kubeCluster *Cluster) (*Cluster, error) {
log.Infof(ctx, "[state] Fetching cluster state from Kubernetes")
k8sClient, err := k8s.NewClient(kubeCluster.LocalKubeConfigPath, kubeCluster.K8sWrapTransport)
if err != nil {
return nil, fmt.Errorf("Failed to create Kubernetes Client: %v", err)
}
var cfgMap *v1.ConfigMap
var currentCluster Cluster
timeout := make(chan bool, 1)
go func() {
for {
cfgMap, err = k8s.GetConfigMap(k8sClient, StateConfigMapName)
if err != nil {
time.Sleep(time.Second * 5)
continue
}
log.Infof(ctx, "[state] Successfully Fetched cluster state to Kubernetes ConfigMap: %s", StateConfigMapName)
timeout <- true
break
}
}()
select {
case <-timeout:
clusterData := cfgMap.Data[StateConfigMapName]
err := yaml.Unmarshal([]byte(clusterData), ¤tCluster)
if err != nil {
return nil, fmt.Errorf("Failed to unmarshal cluster data")
}
return ¤tCluster, nil
case <-time.After(time.Second * GetStateTimeout):
log.Infof(ctx, "Timed out waiting for kubernetes cluster to get state")
return nil, fmt.Errorf("Timeout waiting for kubernetes cluster to get state")
}
}
func GetK8sVersion(localConfigPath string, k8sWrapTransport transport.WrapperFunc) (string, error) {
logrus.Debugf("[version] Using %s to connect to Kubernetes cluster..", localConfigPath)
k8sClient, err := k8s.NewClient(localConfigPath, k8sWrapTransport)
if err != nil {
return "", fmt.Errorf("Failed to create Kubernetes Client: %v", err)
}
discoveryClient := k8sClient.DiscoveryClient
logrus.Debugf("[version] Getting Kubernetes server version..")
serverVersion, err := discoveryClient.ServerVersion()
if err != nil {
return "", fmt.Errorf("Failed to get Kubernetes server version: %v", err)
}
return fmt.Sprintf("%#v", *serverVersion), nil
}
func RebuildState(ctx context.Context, kubeCluster *Cluster, oldState *FullState, flags ExternalFlags) (*FullState, error) {
rkeConfig := &kubeCluster.RancherKubernetesEngineConfig
newState := &FullState{
DesiredState: State{
RancherKubernetesEngineConfig: rkeConfig.DeepCopy(),
},
}
if flags.CustomCerts {
certBundle, err := pki.ReadCertsAndKeysFromDir(flags.CertificateDir)
if err != nil {
return nil, fmt.Errorf("Failed to read certificates from dir [%s]: %v", flags.CertificateDir, err)
}
// make sure all custom certs are included
if err := pki.ValidateBundleContent(rkeConfig, certBundle, flags.ClusterFilePath, flags.ConfigDir); err != nil {
return nil, fmt.Errorf("Failed to validates certificates from dir [%s]: %v", flags.CertificateDir, err)
}
newState.DesiredState.CertificatesBundle = certBundle
newState.CurrentState = oldState.CurrentState
err = updateEncryptionConfig(kubeCluster, oldState, newState)
if err != nil {
return nil, err
}
return newState, nil
}
// Rebuilding the certificates of the desired state
if oldState.DesiredState.CertificatesBundle == nil { // this is a fresh cluster
if err := buildFreshState(ctx, kubeCluster, newState); err != nil {
return nil, err
}
} else { // This is an existing cluster with an old DesiredState
if err := rebuildExistingState(ctx, kubeCluster, oldState, newState, flags); err != nil {
return nil, err
}
}
newState.CurrentState = oldState.CurrentState
return newState, nil
}
func (s *FullState) WriteStateFile(ctx context.Context, statePath string) error {
stateFile, err := json.MarshalIndent(s, "", " ")
if err != nil {
return fmt.Errorf("Failed to Marshal state object: %v", err)
}
logrus.Tracef("Writing state file: %s", stateFile)
if err := os.WriteFile(statePath, stateFile, 0600); err != nil {
return fmt.Errorf("Failed to write state file: %v", err)
}
log.Infof(ctx, "Successfully Deployed state file at [%s]", statePath)
return nil
}
func GetStateFilePath(configPath, configDir string) string {
if configPath == "" {
configPath = pki.ClusterConfig
}
baseDir := filepath.Dir(configPath)
if len(configDir) > 0 {
baseDir = filepath.Dir(configDir)
}
fileName := filepath.Base(configPath)
baseDir += "/"
fullPath := fmt.Sprintf("%s%s", baseDir, fileName)
trimmedName := strings.TrimSuffix(fullPath, filepath.Ext(fullPath))
return trimmedName + stateFileExt
}
func GetCertificateDirPath(configPath, configDir string) string {
if configPath == "" {
configPath = pki.ClusterConfig
}
baseDir := filepath.Dir(configPath)
if len(configDir) > 0 {
baseDir = filepath.Dir(configDir)
}
fileName := filepath.Base(configPath)
baseDir += "/"
fullPath := fmt.Sprintf("%s%s", baseDir, fileName)
trimmedName := strings.TrimSuffix(fullPath, filepath.Ext(fullPath))
return trimmedName + certDirExt
}
func StringToFullState(ctx context.Context, stateFileContent string) (*FullState, error) {
rkeFullState := &FullState{}
logrus.Tracef("stateFileContent: %s", stateFileContent)
if err := json.Unmarshal([]byte(stateFileContent), rkeFullState); err != nil {
return rkeFullState, err
}
rkeFullState.DesiredState.CertificatesBundle = pki.TransformPEMToObject(rkeFullState.DesiredState.CertificatesBundle)
rkeFullState.CurrentState.CertificatesBundle = pki.TransformPEMToObject(rkeFullState.CurrentState.CertificatesBundle)
logrus.Tracef("rkeFullState: %+v", rkeFullState)
return rkeFullState, nil
}
func ReadStateFile(ctx context.Context, statePath string) (*FullState, error) {
rkeFullState := &FullState{}
fp, err := filepath.Abs(statePath)
if err != nil {
return rkeFullState, fmt.Errorf("failed to lookup current directory name: %v", err)
}
file, err := os.Open(fp)
if err != nil {
return rkeFullState, fmt.Errorf("Can not find RKE state file: %v", err)
}
defer file.Close()
buf, err := io.ReadAll(file)
if err != nil {
return rkeFullState, fmt.Errorf("failed to read state file: %v", err)
}
if err := json.Unmarshal(buf, rkeFullState); err != nil {
return rkeFullState, fmt.Errorf("failed to unmarshal the state file: %v", err)
}
rkeFullState.DesiredState.CertificatesBundle = pki.TransformPEMToObject(rkeFullState.DesiredState.CertificatesBundle)
rkeFullState.CurrentState.CertificatesBundle = pki.TransformPEMToObject(rkeFullState.CurrentState.CertificatesBundle)
return rkeFullState, nil
}
func RemoveStateFile(ctx context.Context, statePath string) {
log.Infof(ctx, "Removing state file: %s", statePath)
if err := os.Remove(statePath); err != nil {
logrus.Warningf("Failed to remove state file: %v", err)
return
}
log.Infof(ctx, "State file removed successfully")
}
func GetStateFromNodes(ctx context.Context, kubeCluster *Cluster) *Cluster {
var currentCluster Cluster
var clusterFile string
var err error
uniqueHosts := hosts.GetUniqueHostList(kubeCluster.EtcdHosts, kubeCluster.ControlPlaneHosts, kubeCluster.WorkerHosts)
for _, host := range uniqueHosts {
filePath := path.Join(pki.TempCertPath, pki.ClusterStateFile)
clusterFile, err = pki.FetchFileFromHost(ctx, filePath, kubeCluster.SystemImages.Alpine, host, kubeCluster.PrivateRegistriesMap, pki.StateDeployerContainerName, "state", kubeCluster.Version)
if err == nil {
break
}
}
if len(clusterFile) == 0 {
return nil
}
err = yaml.Unmarshal([]byte(clusterFile), ¤tCluster)
if err != nil {
logrus.Debugf("[state] Failed to unmarshal the cluster file fetched from nodes: %v", err)
return nil
}
log.Infof(ctx, "[state] Successfully fetched cluster state from Nodes")
return ¤tCluster
}
func buildFreshState(ctx context.Context, kubeCluster *Cluster, newState *FullState) error {
rkeConfig := &kubeCluster.RancherKubernetesEngineConfig
// Get the certificate Bundle
certBundle, err := pki.GenerateRKECerts(ctx, *rkeConfig, "", "")
if err != nil {
return fmt.Errorf("Failed to generate certificate bundle: %v", err)
}
newState.DesiredState.CertificatesBundle = certBundle
if isEncryptionEnabled(rkeConfig) {
if newState.DesiredState.EncryptionConfig, err = kubeCluster.getEncryptionProviderFile(); err != nil {
return err
}
}
return nil
}
func rebuildExistingState(ctx context.Context, kubeCluster *Cluster, oldState, newState *FullState, flags ExternalFlags) error {
rkeConfig := &kubeCluster.RancherKubernetesEngineConfig
pkiCertBundle := oldState.DesiredState.CertificatesBundle
// check for legacy clusters prior to requestheaderca
if pkiCertBundle[pki.RequestHeaderCACertName].Certificate == nil {
if err := pki.GenerateRKERequestHeaderCACert(ctx, pkiCertBundle, flags.ClusterFilePath, flags.ConfigDir); err != nil {
return err
}
}
if err := pki.GenerateRKEServicesCerts(ctx, pkiCertBundle, *rkeConfig, flags.ClusterFilePath, flags.ConfigDir, false); err != nil {
return err
}
newState.DesiredState.CertificatesBundle = pkiCertBundle
err := updateEncryptionConfig(kubeCluster, oldState, newState)
return err
}
func updateEncryptionConfig(kubeCluster *Cluster, oldState *FullState, newState *FullState) error {
if isEncryptionEnabled(&kubeCluster.RancherKubernetesEngineConfig) {
if oldState.DesiredState.EncryptionConfig != "" {
newState.DesiredState.EncryptionConfig = oldState.DesiredState.EncryptionConfig
} else {
var err error
if newState.DesiredState.EncryptionConfig, err = kubeCluster.getEncryptionProviderFile(); err != nil {
return err
}
}
}
return nil
}