forked from k3s-io/k3s
/
cluster.go
157 lines (136 loc) · 4.7 KB
/
cluster.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
package cluster
import (
"context"
"net/url"
"runtime"
"strings"
"github.com/k3s-io/k3s/pkg/clientaccess"
"github.com/k3s-io/k3s/pkg/cluster/managed"
"github.com/k3s-io/k3s/pkg/daemons/config"
"github.com/k3s-io/k3s/pkg/etcd"
"github.com/k3s-io/kine/pkg/endpoint"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
utilsnet "k8s.io/utils/net"
)
type Cluster struct {
clientAccessInfo *clientaccess.Info
config *config.Control
managedDB managed.Driver
joining bool
storageStarted bool
saveBootstrap bool
shouldBootstrap bool
cnFilterFunc func(...string) []string
}
// Start creates the dynamic tls listener, http request handler,
// handles starting and writing/reading bootstrap data, and returns a channel
// that will be closed when datastore is ready. If embedded etcd is in use,
// a secondary call to Cluster.save is made.
func (c *Cluster) Start(ctx context.Context) (<-chan struct{}, error) {
// Set up the dynamiclistener and http request handlers
if err := c.initClusterAndHTTPS(ctx); err != nil {
return nil, errors.Wrap(err, "init cluster datastore and https")
}
if c.config.DisableETCD {
ready := make(chan struct{})
defer close(ready)
// try to get /db/info urls first before attempting to use join url
clientURLs, _, err := etcd.ClientURLs(ctx, c.clientAccessInfo, c.config.PrivateIP)
if err != nil {
return nil, err
}
if len(clientURLs) < 1 {
clientURL, err := url.Parse(c.config.JoinURL)
if err != nil {
return nil, err
}
clientURL.Host = clientURL.Hostname() + ":2379"
clientURLs = append(clientURLs, clientURL.String())
}
etcdProxy, err := etcd.NewETCDProxy(ctx, true, c.config.DataDir, clientURLs[0], utilsnet.IsIPv6CIDR(c.config.ServiceIPRanges[0]))
if err != nil {
return nil, err
}
c.setupEtcdProxy(ctx, etcdProxy)
// remove etcd member if it exists
if err := c.managedDB.RemoveSelf(ctx); err != nil {
logrus.Warnf("Failed to remove this node from etcd members")
}
c.config.Runtime.EtcdConfig.Endpoints = strings.Split(c.config.Datastore.Endpoint, ",")
c.config.Runtime.EtcdConfig.TLSConfig = c.config.Datastore.BackendTLSConfig
return ready, nil
}
// start managed database (if necessary)
if err := c.start(ctx); err != nil {
return nil, errors.Wrap(err, "start managed database")
}
// get the wait channel for testing managed database readiness
ready, err := c.testClusterDB(ctx)
if err != nil {
return nil, err
}
if err := c.startStorage(ctx); err != nil {
return nil, err
}
// if necessary, store bootstrap data to datastore
if c.saveBootstrap {
if err := Save(ctx, c.config, false); err != nil {
return nil, err
}
}
// at this point, if etcd is in use, it's bootstrapping is complete
// so save the bootstrap data. We will need for etcd to be up. If
// the save call returns an error, we panic since subsequent etcd
// snapshots will be empty.
if c.managedDB != nil {
go func() {
for {
select {
case <-ready:
if err := Save(ctx, c.config, false); err != nil {
panic(err)
}
if !c.config.EtcdDisableSnapshots {
if err := c.managedDB.ReconcileSnapshotData(ctx); err != nil {
logrus.Errorf("Failed to record snapshots for cluster: %v", err)
}
}
return
default:
runtime.Gosched()
}
}
}()
}
return ready, nil
}
// startStorage starts the kine listener and configures the endpoints, if necessary.
// This calls into the kine endpoint code, which sets up the database client
// and unix domain socket listener if using an external database. In the case of an etcd
// backend it just returns the user-provided etcd endpoints and tls config.
func (c *Cluster) startStorage(ctx context.Context) error {
if c.storageStarted {
return nil
}
c.storageStarted = true
// start listening on the kine socket as an etcd endpoint, or return the external etcd endpoints
etcdConfig, err := endpoint.Listen(ctx, c.config.Datastore)
if err != nil {
return errors.Wrap(err, "creating storage endpoint")
}
// Persist the returned etcd configuration. We decide if we're doing leader election for embedded controllers
// based on what the kine wrapper tells us about the datastore. Single-node datastores like sqlite don't require
// leader election, while basically all others (etcd, external database, etc) do since they allow multiple servers.
c.config.Runtime.EtcdConfig = etcdConfig
c.config.Datastore.BackendTLSConfig = etcdConfig.TLSConfig
c.config.Datastore.Endpoint = strings.Join(etcdConfig.Endpoints, ",")
c.config.NoLeaderElect = !etcdConfig.LeaderElect
return nil
}
// New creates an initial cluster using the provided configuration.
func New(config *config.Control) *Cluster {
return &Cluster{
config: config,
}
}