k3s/pkg/dqlite/server.go

221 lines
5.5 KiB
Go
Raw Normal View History

2019-10-31 02:05:40 +00:00
package dqlite
import (
"context"
"crypto/tls"
"crypto/x509"
"fmt"
"io/ioutil"
"math/rand"
"net/http"
"os"
"path/filepath"
"strconv"
"strings"
"time"
"github.com/canonical/go-dqlite"
"github.com/canonical/go-dqlite/client"
"github.com/pkg/errors"
controllerclient "github.com/rancher/k3s/pkg/dqlite/controller/client"
"github.com/rancher/k3s/pkg/dqlite/dialer"
dqlitedriver "github.com/rancher/kine/pkg/drivers/dqlite"
v1 "github.com/rancher/wrangler-api/pkg/generated/controllers/core/v1"
"github.com/sirupsen/logrus"
"k8s.io/apimachinery/pkg/util/net"
)
type Certs struct {
ServerTrust *x509.Certificate
ClientTrust *x509.Certificate
ClientCert tls.Certificate
}
type DQLite struct {
ClientCA string
ClientCAKey string
ClientCert string
ClientCertKey string
ServerCA string
ServerCAKey string
AdvertiseIP string
AdvertisePort int
DataDir string
NodeStore client.NodeStore
NodeInfo client.NodeInfo
node *dqlite.Node
StorageEndpoint string
NodeControllerGetter NodeControllerGetter
clientOpts []client.Option
}
type NodeControllerGetter func() v1.NodeController
func New(dataDir, advertiseIP string, advertisePort int, getter NodeControllerGetter) *DQLite {
return &DQLite{
AdvertiseIP: advertiseIP,
AdvertisePort: advertisePort,
DataDir: dataDir,
NodeControllerGetter: getter,
}
}
2019-11-11 22:19:00 +00:00
func (d *DQLite) Start(ctx context.Context, initCluster, resetCluster bool, certs *Certs, next http.Handler) (http.Handler, error) {
2019-10-31 02:05:40 +00:00
bindAddress := d.getBindAddress()
clientTLSConfig, err := getClientTLSConfig(certs.ClientCert, certs.ServerTrust)
if err != nil {
return nil, err
}
advertise, err := getAdvertiseAddress(d.AdvertiseIP, d.AdvertisePort)
if err != nil {
return nil, errors.Wrap(err, "get advertise address")
}
dial, err := getDialer(advertise, bindAddress, clientTLSConfig)
if err != nil {
return nil, err
}
dqlitedriver.Dialer = dial
dqlitedriver.Logger = log()
d.clientOpts = append(d.clientOpts, client.WithDialFunc(dial), client.WithLogFunc(log()))
nodeInfo, node, err := getNode(d.DataDir, advertise, bindAddress, initCluster, dial)
if err != nil {
return nil, err
}
d.NodeInfo = nodeInfo
2019-11-11 22:19:00 +00:00
d.node = node
2019-10-31 02:05:40 +00:00
go func() {
<-ctx.Done()
node.Close()
}()
if err := d.nodeStore(ctx, initCluster); err != nil {
return nil, err
}
go d.startController(ctx)
2019-11-11 22:19:00 +00:00
if !resetCluster {
if err := node.Start(); err != nil {
return nil, err
}
}
return router(ctx, next, nodeInfo, certs.ClientTrust, "kube-apiserver", bindAddress), nil
2019-10-31 02:05:40 +00:00
}
func (d *DQLite) startController(ctx context.Context) {
for {
if nc := d.NodeControllerGetter(); nc != nil {
if os.Getenv("NODE_NAME") == "" {
logrus.Errorf("--disable-agent is not compatible with dqlite")
} else {
break
}
}
time.Sleep(time.Second)
}
controllerclient.Register(ctx, os.Getenv("NODE_NAME"), d.NodeInfo, d.NodeStore, d.NodeControllerGetter(), d.clientOpts)
}
func (d *DQLite) nodeStore(ctx context.Context, initCluster bool) error {
peerDB := filepath.Join(d.DataDir, "db", "state.dqlite", "peers.db")
ns, err := client.DefaultNodeStore(peerDB)
if err != nil {
return err
}
d.NodeStore = ns
d.StorageEndpoint = fmt.Sprintf("dqlite://?peer-file=%s", peerDB)
if initCluster {
if err := dqlitedriver.AddPeers(ctx, d.NodeStore, d.NodeInfo); err != nil {
return err
}
}
return nil
}
func getAdvertiseAddress(advertiseIP string, advertisePort int) (string, error) {
ip := advertiseIP
if ip == "" {
ipAddr, err := net.ChooseHostInterface()
if err != nil {
return "", err
}
ip = ipAddr.String()
}
return fmt.Sprintf("%s:%d", ip, advertisePort), nil
}
func getClientTLSConfig(cert tls.Certificate, ca *x509.Certificate) (*tls.Config, error) {
tlsConfig := &tls.Config{
RootCAs: x509.NewCertPool(),
Certificates: []tls.Certificate{
cert,
},
ServerName: "kubernetes",
}
tlsConfig.RootCAs.AddCert(ca)
return tlsConfig, nil
}
func getDialer(advertiseAddress, bindAddress string, tlsConfig *tls.Config) (client.DialFunc, error) {
return dialer.NewHTTPDialer(advertiseAddress, bindAddress, tlsConfig)
}
func getNode(dataDir string, advertiseAddress, bindAddress string, initCluster bool, dial client.DialFunc) (dqlite.NodeInfo, *dqlite.Node, error) {
id, err := getClusterID(initCluster, dataDir)
if err != nil {
return dqlite.NodeInfo{}, nil, errors.Wrap(err, "reading cluster id")
}
dbDir := filepath.Join(dataDir, "db", "state.dqlite")
node, err := dqlite.New(id, advertiseAddress, dbDir,
dqlite.WithBindAddress(bindAddress),
dqlite.WithDialFunc(dial),
dqlite.WithNetworkLatency(20*time.Millisecond))
return dqlite.NodeInfo{
ID: id,
Address: advertiseAddress,
}, node, err
}
func getClusterID(initCluster bool, dataDir string) (uint64, error) {
idFile := filepath.Join(dataDir, "db/state.dqlite/node-id")
content, err := ioutil.ReadFile(idFile)
if os.IsNotExist(err) {
content = nil
} else if err != nil {
return 0, err
}
idStr := strings.TrimSpace(string(content))
if idStr == "" {
if err := os.MkdirAll(filepath.Dir(idFile), 0700); err != nil {
return 0, err
}
id := rand.Uint64()
if initCluster {
id = 1
}
return id, ioutil.WriteFile(idFile, []byte(strconv.FormatUint(id, 10)), 0644)
}
return strconv.ParseUint(idStr, 10, 64)
}
func (d *DQLite) getBindAddress() string {
// only anonymous works???
return "@" + filepath.Join(d.DataDir, "db", "state.dqlite", "dqlite.sock")
}