forked from rancher/rke
/
state.go
111 lines (105 loc) · 3.16 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
package cluster
import (
"fmt"
"time"
"github.com/Sirupsen/logrus"
"github.com/rancher/rke/hosts"
"github.com/rancher/rke/k8s"
"github.com/rancher/rke/pki"
yaml "gopkg.in/yaml.v2"
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/pkg/api/v1"
)
func (c *Cluster) SaveClusterState(clusterFile string) error {
// Reinitialize kubernetes Client
var err error
c.KClient, err = k8s.NewClient(pki.KubeAdminConfigPath)
if err != nil {
return fmt.Errorf("Failed to re-initialize Kubernetes Client: %v", err)
}
err = saveClusterCerts(c.KClient, c.Certificates)
if err != nil {
return fmt.Errorf("[certificates] Failed to Save Kubernetes certificates: %v", err)
}
err = saveStateToKubernetes(c.KClient, pki.KubeAdminConfigPath, []byte(clusterFile))
if err != nil {
return fmt.Errorf("[state] Failed to save configuration state: %v", err)
}
return nil
}
func (c *Cluster) GetClusterState() (*Cluster, error) {
var err error
var currentCluster *Cluster
c.KClient, err = k8s.NewClient(pki.KubeAdminConfigPath)
if err != nil {
logrus.Warnf("Failed to initiate new Kubernetes Client: %v", err)
} else {
// Handle pervious kubernetes state and certificate generation
currentCluster = getStateFromKubernetes(c.KClient, pki.KubeAdminConfigPath)
if currentCluster != nil {
err = currentCluster.InvertIndexHosts()
if err != nil {
return nil, fmt.Errorf("Failed to classify hosts from fetched cluster: %v", err)
}
err = hosts.ReconcileWorkers(currentCluster.WorkerHosts, c.WorkerHosts, c.KClient)
if err != nil {
return nil, fmt.Errorf("Failed to reconcile hosts: %v", err)
}
}
}
return currentCluster, nil
}
func saveStateToKubernetes(kClient *kubernetes.Clientset, kubeConfigPath string, clusterFile []byte) error {
logrus.Infof("[state] Saving cluster state to Kubernetes")
timeout := make(chan bool, 1)
go func() {
for {
err := k8s.UpdateConfigMap(kClient, clusterFile, StateConfigMapName)
if err != nil {
time.Sleep(time.Second * 5)
continue
}
logrus.Infof("[state] Successfully Saved cluster state to Kubernetes ConfigMap: %s", StateConfigMapName)
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(kClient *kubernetes.Clientset, kubeConfigPath string) *Cluster {
logrus.Infof("[state] Fetching cluster state from Kubernetes")
var cfgMap *v1.ConfigMap
var currentCluster Cluster
var err error
timeout := make(chan bool, 1)
go func() {
for {
cfgMap, err = k8s.GetConfigMap(kClient, StateConfigMapName)
if err != nil {
time.Sleep(time.Second * 5)
continue
}
logrus.Infof("[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
}
return ¤tCluster
case <-time.After(time.Second * GetStateTimeout):
logrus.Warnf("Timed out waiting for kubernetes cluster")
return nil
}
return nil
}