forked from rancher/rke
/
etcd.go
175 lines (164 loc) · 5.89 KB
/
etcd.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
package services
import (
"fmt"
"time"
"context"
etcdclient "github.com/coreos/etcd/client"
"github.com/docker/docker/api/types/container"
"github.com/rancher/rke/docker"
"github.com/rancher/rke/hosts"
"github.com/rancher/rke/log"
"github.com/rancher/types/apis/management.cattle.io/v3"
"github.com/sirupsen/logrus"
)
func RunEtcdPlane(ctx context.Context, etcdHosts []*hosts.Host, etcdService v3.ETCDService, localConnDialerFactory hosts.DialerFactory) error {
log.Infof(ctx, "[%s] Building up Etcd Plane..", ETCDRole)
initCluster := getEtcdInitialCluster(etcdHosts)
for _, host := range etcdHosts {
imageCfg, hostCfg := buildEtcdConfig(host, etcdService, initCluster)
err := docker.DoRunContainer(ctx, host.DClient, imageCfg, hostCfg, EtcdContainerName, host.Address, ETCDRole)
if err != nil {
return err
}
}
log.Infof(ctx, "[%s] Successfully started Etcd Plane..", ETCDRole)
return nil
}
func RemoveEtcdPlane(ctx context.Context, etcdHosts []*hosts.Host, force bool) error {
log.Infof(ctx, "[%s] Tearing down Etcd Plane..", ETCDRole)
for _, host := range etcdHosts {
err := docker.DoRemoveContainer(ctx, host.DClient, EtcdContainerName, host.Address)
if err != nil {
return err
}
if !host.IsWorker || !host.IsControl || force {
// remove unschedulable kubelet on etcd host
if err := removeKubelet(ctx, host); err != nil {
return err
}
if err := removeKubeproxy(ctx, host); err != nil {
return err
}
if err := removeNginxProxy(ctx, host); err != nil {
return err
}
if err := removeSidekick(ctx, host); err != nil {
return err
}
}
}
log.Infof(ctx, "[%s] Successfully tore down Etcd Plane..", ETCDRole)
return nil
}
func buildEtcdConfig(host *hosts.Host, etcdService v3.ETCDService, initCluster string) (*container.Config, *container.HostConfig) {
clusterState := "new"
if host.ExistingEtcdCluster {
clusterState = "existing"
}
imageCfg := &container.Config{
Image: etcdService.Image,
Cmd: []string{"/usr/local/bin/etcd",
"--name=etcd-" + host.HostnameOverride,
"--data-dir=/etcd-data",
"--advertise-client-urls=http://" + host.InternalAddress + ":2379,http://" + host.InternalAddress + ":4001",
"--listen-client-urls=http://0.0.0.0:2379",
"--initial-advertise-peer-urls=http://" + host.InternalAddress + ":2380",
"--listen-peer-urls=http://0.0.0.0:2380",
"--initial-cluster-token=etcd-cluster-1",
"--initial-cluster=" + initCluster,
"--initial-cluster-state=" + clusterState},
}
hostCfg := &container.HostConfig{
RestartPolicy: container.RestartPolicy{Name: "always"},
Binds: []string{
"/var/lib/etcd:/etcd-data"},
NetworkMode: "host",
}
for arg, value := range etcdService.ExtraArgs {
cmd := fmt.Sprintf("--%s=%s", arg, value)
imageCfg.Entrypoint = append(imageCfg.Entrypoint, cmd)
}
return imageCfg, hostCfg
}
func AddEtcdMember(ctx context.Context, etcdHost *hosts.Host, etcdHosts []*hosts.Host, localConnDialerFactory hosts.DialerFactory) error {
log.Infof(ctx, "[add/%s] Adding member [etcd-%s] to etcd cluster", ETCDRole, etcdHost.HostnameOverride)
peerURL := fmt.Sprintf("http://%s:2380", etcdHost.InternalAddress)
added := false
for _, host := range etcdHosts {
etcdClient, err := getEtcdClient(ctx, host, localConnDialerFactory)
if err != nil {
logrus.Debugf("Failed to create etcd client for host [%s]: %v", host.Address, err)
continue
}
memAPI := etcdclient.NewMembersAPI(etcdClient)
if _, err := memAPI.Add(ctx, peerURL); err != nil {
logrus.Debugf("Failed to list etcd members from host [%s]: %v", host.Address, err)
continue
}
added = true
break
}
if !added {
return fmt.Errorf("Failed to add etcd member [etcd-%s] from etcd cluster", etcdHost.HostnameOverride)
}
log.Infof(ctx, "[add/%s] Successfully Added member [etcd-%s] to etcd cluster", ETCDRole, etcdHost.HostnameOverride)
return nil
}
func RemoveEtcdMember(ctx context.Context, etcdHost *hosts.Host, etcdHosts []*hosts.Host, localConnDialerFactory hosts.DialerFactory) error {
log.Infof(ctx, "[remove/%s] Removing member [etcd-%s] from etcd cluster", ETCDRole, etcdHost.HostnameOverride)
var mID string
removed := false
for _, host := range etcdHosts {
etcdClient, err := getEtcdClient(ctx, host, localConnDialerFactory)
if err != nil {
logrus.Debugf("Failed to create etcd client for host [%s]: %v", host.Address, err)
continue
}
memAPI := etcdclient.NewMembersAPI(etcdClient)
members, err := memAPI.List(ctx)
if err != nil {
logrus.Debugf("Failed to list etcd members from host [%s]: %v", host.Address, err)
continue
}
for _, member := range members {
if member.Name == fmt.Sprintf("etcd-%s", etcdHost.HostnameOverride) {
mID = member.ID
break
}
}
if err := memAPI.Remove(ctx, mID); err != nil {
logrus.Debugf("Failed to list etcd members from host [%s]: %v", host.Address, err)
continue
}
removed = true
break
}
if !removed {
return fmt.Errorf("Failed to delete etcd member [etcd-%s] from etcd cluster", etcdHost.HostnameOverride)
}
log.Infof(ctx, "[remove/%s] Successfully removed member [etcd-%s] from etcd cluster", ETCDRole, etcdHost.HostnameOverride)
return nil
}
func ReloadEtcdCluster(ctx context.Context, etcdHosts []*hosts.Host, etcdService v3.ETCDService, localConnDialerFactory hosts.DialerFactory) error {
readyEtcdHosts := []*hosts.Host{}
for _, host := range etcdHosts {
if !host.ToAddEtcdMember {
readyEtcdHosts = append(readyEtcdHosts, host)
host.ExistingEtcdCluster = true
}
}
initCluster := getEtcdInitialCluster(readyEtcdHosts)
for _, host := range readyEtcdHosts {
imageCfg, hostCfg := buildEtcdConfig(host, etcdService, initCluster)
if err := docker.DoRunContainer(ctx, host.DClient, imageCfg, hostCfg, EtcdContainerName, host.Address, ETCDRole); err != nil {
return err
}
}
time.Sleep(10 * time.Second)
for _, host := range readyEtcdHosts {
if healthy := isEtcdHealthy(ctx, localConnDialerFactory, host); healthy {
break
}
}
return nil
}