func (g *Gateway) LeaderAddress() (string, error) {
	if g.memoryDial != nil {
		return "", errors.New("node is not clustered")
	}
	ctx, cancel := context.WithTimeout(g.ctx, 5*time.Second)
	defer cancel()
	if g.raft != nil {
		for ctx.Err() == nil {
			if address := string(g.raft.Raft().Leader()); address != "" {
				return address, nil
			}
			time.Sleep(time.Second)
		}
		return "", ctx.Err()
	}
	config, err := cert.TLSClientConfig(g.cert)
	if err != nil {
		return "", err
	}
	var addresses []string
	if err := g.database.Transaction(func(tx *db.NodeTx) error {
		nodes, err := tx.RaftNodes()
		if err != nil {
			return errors.WithStack(err)
		}
		for _, node := range nodes {
			addresses = append(addresses, node.Address)
		}
		return nil
	}); err != nil {
		return "", errors.Wrap(err, "failed to fetch raft nodes addresses")
	}
	if len(addresses) == 0 {
		return "", errors.New("no raft node known")
	}
	for _, address := range addresses {
		url := fmt.Sprintf("https://%s%s", address, heartbeat.DatabaseEndpoint)
		request, err := http.NewRequest("GET", url, nil)
		if err != nil {
			return "", errors.WithStack(err)
		}
		request = request.WithContext(ctx)
		client := &http.Client{
			Transport: &http.Transport{
				TLSClientConfig: config,
			},
		}
		response, err := client.Do(request)
		if err != nil {
			level.Debug(g.logger).Log("msg", "failed to fetch leader address", "address", address, "err", err)
			continue
		}
		defer response.Body.Close()
		if response.StatusCode != http.StatusOK {
			level.Debug(g.logger).Log("msg", "request for leader address failed", "address", address, "status", response.Status)
			continue
		}
		info := map[string]string{}
		if err := json.Read(response.Body, &info); err != nil {
			level.Debug(g.logger).Log("msg", "failed to parse leader address", "address", address, "err", err)
			continue
		}
		leader, ok := info["leader"]
		if !ok || leader == "" {
			level.Debug(g.logger).Log("msg", "raft node return no leader address", "address", address)
			continue
		}
		return leader, nil
	}
	return "", errors.New("raft cluster is unavailable")
}
LeaderAddress returns the address of the current raft leader.
package cluster
import (
	"context"
	"fmt"
	"net"
	"net/http"
	"net/url"
	"path/filepath"
	"strconv"
	"time"
	dqlite "github.com/CanonicalLtd/go-dqlite"
	raftmembership "github.com/CanonicalLtd/raft-membership"
	"github.com/go-kit/kit/log"
	"github.com/go-kit/kit/log/level"
	hashiraft "github.com/hashicorp/raft"
	"github.com/pkg/errors"
	"github.com/spoke-d/thermionic/internal/cert"
	"github.com/spoke-d/thermionic/internal/clock"
	"github.com/spoke-d/thermionic/internal/cluster/heartbeat"
	"github.com/spoke-d/thermionic/internal/cluster/raft"
	"github.com/spoke-d/thermionic/internal/config"
	"github.com/spoke-d/thermionic/internal/db"
	"github.com/spoke-d/thermionic/internal/db/cluster"
	"github.com/spoke-d/thermionic/internal/db/database"
	"github.com/spoke-d/thermionic/internal/fsys"
	"github.com/spoke-d/thermionic/internal/json"
	"github.com/spoke-d/thermionic/internal/retrier"
)
type Node interface {
	database.DBAccessor
	db.NodeOpener
	db.NodeTransactioner
	Dir() string
	Close() error
}
type RaftProvider interface {
	New(Node, *cert.Info, config.Schema, fsys.FileSystem, log.Logger, float64) RaftInstance
}
type RaftInstance interface {
	Init() error
	HandlerFunc() http.HandlerFunc
	Raft() *hashiraft.Raft
	Registry() *dqlite.Registry
	Servers() ([]hashiraft.Server, error)
	Shutdown() error
	MembershipChanger() raftmembership.Changer
}
type ServerProvider interface {
	New(RaftInstance, net.Listener, *raft.AddressProvider, log.Logger) (Server, error)
}
type Server interface {
	Dump(string, string) error
	Close() error
}
type Net interface {
	UnixListen(string) (net.Listener, error)
	UnixDial(string) (net.Conn, error)
}
type StoreProvider interface {
	Memory() cluster.ServerStore
	Disk(database.DB) (cluster.ServerStore, error)
}
type Gateway struct {
	database         Node
	fileSystem       fsys.FileSystem
	cert             *cert.Info
	nodeConfigSchema config.Schema
	raft             RaftInstance
	server   Server
	acceptCh chan net.Conn
	memoryDial dqlite.DialFunc
	ctx    context.Context
	cancel context.CancelFunc
	upgradeCh chan struct{}
	store *ServerStore
	raftProvider    RaftProvider
	netProvider     Net
	storeProvider   StoreProvider
	serverProvider  ServerProvider
	sleeper         clock.Sleeper
	addressProvider *raft.AddressProvider
	latency float64
	logger  log.Logger
}
func NewGateway(database Node, nodeConfigSchema config.Schema, fileSystem fsys.FileSystem, options ...Option) *Gateway {
	opts := newOptions()
	opts.database = database
	opts.nodeConfigSchema = nodeConfigSchema
	opts.fileSystem = fileSystem
	for _, option := range options {
		option(opts)
	}
	ctx, cancel := context.WithCancel(context.Background())
	gateway := &Gateway{
		database:         opts.database,
		cert:             opts.cert,
		nodeConfigSchema: opts.nodeConfigSchema,
		fileSystem:       opts.fileSystem,
		ctx:              ctx,
		cancel:           cancel,
		upgradeCh:        make(chan struct{}, 16),
		acceptCh:         make(chan net.Conn),
		store:            &ServerStore{},
		raftProvider:    opts.raftProvider,
		netProvider:     opts.netProvider,
		storeProvider:   opts.storeProvider,
		serverProvider:  opts.serverProvider,
		addressProvider: raft.NewAddressProvider(opts.database),
		sleeper:         opts.sleeper,
		latency:         opts.latency,
		logger:          opts.logger,
	}
	return gateway
}
func (g *Gateway) Init(cert *cert.Info) error {
	level.Info(g.logger).Log("msg", "initializing database gateway")
	g.cert = cert
	raftInst := g.raftProvider.New(
		g.database,
		g.cert,
		g.nodeConfigSchema,
		g.fileSystem,
		g.logger,
		g.latency,
	)
	if err := raftInst.Init(); err != nil {
		return errors.Wrap(err, "failed to create raft factory")
	}
	if raftInst != nil && raftInst.Raft() != nil {
		listener, err := g.netProvider.UnixListen("")
		if err != nil {
			return errors.Wrap(err, "failed to allocate loopback port")
		}
		if raftInst.HandlerFunc() == nil {
			g.memoryDial = dqliteMemoryDial(g.netProvider, listener)
			g.store.inMemory = g.storeProvider.Memory()
			if err := g.store.Set(context.Background(), []cluster.ServerInfo{
				{Address: "0"},
			}); err != nil {
				return errors.Wrap(err, "failed to set address on memory store")
			}
		} else {
			go func(proxy DqliteProxyFn) {
				proxy(listener, g.acceptCh)
			}(DqliteProxy(g.logger, g.netProvider))
			g.store.inMemory = nil
		}
		provider := raft.NewAddressProvider(g.database)
		server, err := g.serverProvider.New(raftInst, listener, provider, g.logger)
		if err != nil {
			return errors.Wrap(err, "failed to create dqlite server")
		}
		g.addressProvider = provider
		g.server = server
		g.raft = raftInst
	} else {
		g.server = nil
		g.raft = nil
		g.store.inMemory = nil
	}
	store, err := g.storeProvider.Disk(g.database.DB())
	if err != nil {
		return errors.WithStack(err)
	}
	g.store.onDisk = store
	return nil
}
func (g *Gateway) HandlerFuncs() map[string]http.HandlerFunc {
	databaseHandler := func(w http.ResponseWriter, r *http.Request) {
		ok, err := cert.TLSCheckCert(r, g.cert)
		if err != nil {
			http.Error(w, "500 server error", http.StatusInternalServerError)
			return
		}
		if !ok {
			http.Error(w, "403 invalid client certificate", http.StatusForbidden)
			return
		}
		if r.Method == "PUT" {
			g.handleHeartbeats(w, r)
			return
		}
		if !g.Clustered() {
			http.NotFound(w, r)
			return
		}
		if r.Method == "PATCH" {
			select {
			case g.upgradeCh <- struct{}{}:
			default:
			}
			return
		}
		if r.Method == "HEAD" {
			g.handleLeadershipState(w, r)
			return
		}
		if r.Method == "GET" {
			g.handleLeadership(w, r)
			return
		}
		if r.Header.Get("Upgrade") != "dqlite" {
			http.Error(w, "Missing or invalid upgrade header", http.StatusBadRequest)
			return
		}
		hijacker, ok := w.(http.Hijacker)
		if !ok {
			http.Error(w, "Webserver doesn't support hijacking", http.StatusInternalServerError)
			return
		}
		conn, _, err := hijacker.Hijack()
		if err != nil {
			message := errors.Wrap(err, "Failed to hijack connection").Error()
			http.Error(w, message, http.StatusInternalServerError)
			return
		}
		data := []byte("HTTP/1.1 101 Switching Protocols\r\nUpgrade: dqlite\r\n\r\n")
		if n, err := conn.Write(data); err != nil || n != len(data) {
			conn.Close()
			return
		}
		g.acceptCh <- conn
	}
	raftHandler := func(w http.ResponseWriter, r *http.Request) {
		if g.raft == nil {
			var address string
			if err := g.database.Transaction(func(tx *db.NodeTx) error {
				nodes, err := tx.RaftNodes()
				if err != nil {
					return errors.WithStack(err)
				}
				if len(nodes) == 0 {
					return errors.Errorf("no raft nodes found")
				}
				address = nodes[0].Address
				return nil
			}); err != nil {
				http.Error(w, "500 failed to fetch raft nodes", http.StatusInternalServerError)
				return
			}
			url := &url.URL{
				Scheme:   "http",
				Path:     r.URL.Path,
				RawQuery: r.URL.RawQuery,
				Host:     address,
			}
			http.Redirect(w, r, url.String(), http.StatusPermanentRedirect)
			return
		}
		handlerFunc := g.raft.HandlerFunc()
		if handlerFunc == nil {
			http.NotFound(w, r)
			return
		}
		handlerFunc(w, r)
	}
	return map[string]http.HandlerFunc{
		heartbeat.DatabaseEndpoint: databaseHandler,
		raft.Endpoint:              raftHandler,
	}
}
func (g *Gateway) WaitLeadership() error {
	retry := retrier.New(g.sleeper, 100, 250*time.Millisecond)
	err := retry.Run(func() error {
		if g.raft.Raft().State() == hashiraft.Leader {
			return nil
		}
		return errors.New("node is not leader")
	})
	return errors.WithMessage(err, "Node did not self-elect with timeframe")
}
func (g *Gateway) IsDatabaseNode() bool {
	return g.raft != nil
}
func (g *Gateway) WaitUpgradeNotification() {
	<-g.upgradeCh
}
func (g *Gateway) DB() Node {
	return g.database
}
func (g *Gateway) Cert() *cert.Info {
	return g.cert
}
func (g *Gateway) Clustered() bool {
	return g.server != nil && g.memoryDial == nil
}
func (g *Gateway) Raft() RaftInstance {
	return g.raft
}
func (g *Gateway) RaftNodes() ([]db.RaftNode, error) {
	if g.raft == nil {
		return nil, hashiraft.ErrNotLeader
	}
	servers, err := g.raft.Servers()
	if err != nil {
		return nil, errors.WithStack(err)
	}
	nodes := make([]db.RaftNode, len(servers))
	addressProvider := g.addressProvider
	for i, server := range servers {
		address, err := addressProvider.ServerAddr(server.ID)
		if err != nil {
			if err != db.ErrNoSuchObject {
				return nil, errors.Wrap(err, "failed to fetch raft server address")
			}
			address = server.Address
		}
		id, err := strconv.Atoi(string(server.ID))
		if err != nil {
			return nil, errors.Wrap(err, "non-numeric server id")
		}
		nodes[i].ID = int64(id)
		nodes[i].Address = string(address)
	}
	return nodes, nil
}