Put etcd and Patroni's API on TLS, and move a running cluster there in maintenance mode
Members' etcd speaks TLS to peers and clients with a certificate from the cluster's authority and asks clients for one; Patroni reaches it with the member's certificate and serves its API over TLS. A running plain-HTTP cluster is moved with Patroni paused, so PostgreSQL serves throughout: Patroni's lease-free keys are carried over to the new etcd. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
1 parent
fed8c75c6c
commit
191c65e3db
12 files changed
+730
-105
No files matched your search
@@ -196,6 +196,12 @@ func newServerCmd() *cobra.Command {
|
||||
// Patroni. One host is a single PostgreSQL, which has none.
|
||||
if addr, err := storepkg.ResolveAddress(cfg.Auth.Store); err == nil && storepkg.IsPostgres(addr) {
|
||||
if hosts, _, _, err := storepkg.SplitPostgres(strings.TrimSpace(addr)); err == nil && len(hosts) > 1 {
|
||||
if cfg.AgentTLS.CA != "" {
|
||||
// Members' APIs, once on TLS, verified against it.
|
||||
if err := pgha.SetRESTCA(cfg.AgentTLS.CA); err != nil {
|
||||
log.Warn("the store's API cannot be verified", "err", err)
|
||||
}
|
||||
}
|
||||
storeHA = pgha.NewCollector(hosts)
|
||||
if err := rt.RegisterMetrics(storeHA); err != nil {
|
||||
return err
|
||||
|
||||
@@ -352,6 +352,7 @@ func (sa *serverAdd) storeAccess(ctx context.Context) error {
|
||||
if b, err := os.ReadFile(patroniRestFile(sa.etcDir)); err == nil { //nolint:gosec // our own secret
|
||||
rest = strings.TrimSpace(string(b))
|
||||
}
|
||||
_ = pgha.SetRESTCA(filepath.Join(sa.etcDir, "ca.pem"))
|
||||
changed, err := pgha.AllowClient(ctx, hosts, sa.ip, rest)
|
||||
if err != nil {
|
||||
return err
|
||||
|
||||
+76
-11
@@ -44,22 +44,68 @@ type Secrets struct {
|
||||
}
|
||||
|
||||
// Etcd is a member's /etc/default/etcd, for Ubuntu's etcd-server.
|
||||
func Etcd(m Member, all []Member) string {
|
||||
//
|
||||
// With tls, members speak TLS to each other and to clients, with a
|
||||
// certificate from the cluster's authority (EtcdTLSDir), and a client must
|
||||
// present one too: without, anyone reaching :2379 could take the leader lock
|
||||
// or rewrite Patroni's configuration.
|
||||
func Etcd(m Member, all []Member, tls bool) string {
|
||||
scheme := "http"
|
||||
if tls {
|
||||
scheme = "https"
|
||||
}
|
||||
var peers []string
|
||||
for _, p := range all {
|
||||
peers = append(peers, fmt.Sprintf("%s=http://%s:2380", p.Name, p.IP))
|
||||
peers = append(peers, fmt.Sprintf("%s=%s://%s:2380", p.Name, scheme, p.IP))
|
||||
}
|
||||
return fmt.Sprintf(`# llmgw: one member of the etcd that holds Patroni's leader lock.
|
||||
client := fmt.Sprintf("http://%s:2379,http://127.0.0.1:2379", m.IP)
|
||||
if tls {
|
||||
client = fmt.Sprintf("https://%s:2379,https://127.0.0.1:2379", m.IP)
|
||||
}
|
||||
conf := fmt.Sprintf(`# llmgw: one member of the etcd that holds Patroni's leader lock.
|
||||
ETCD_NAME=%[1]s
|
||||
ETCD_DATA_DIR=/var/lib/etcd/default
|
||||
ETCD_LISTEN_PEER_URLS=http://%[2]s:2380
|
||||
ETCD_LISTEN_CLIENT_URLS=http://%[2]s:2379,http://127.0.0.1:2379
|
||||
ETCD_INITIAL_ADVERTISE_PEER_URLS=http://%[2]s:2380
|
||||
ETCD_ADVERTISE_CLIENT_URLS=http://%[2]s:2379
|
||||
ETCD_LISTEN_PEER_URLS=%[5]s://%[2]s:2380
|
||||
ETCD_LISTEN_CLIENT_URLS=%[6]s
|
||||
ETCD_INITIAL_ADVERTISE_PEER_URLS=%[5]s://%[2]s:2380
|
||||
ETCD_ADVERTISE_CLIENT_URLS=%[5]s://%[2]s:2379
|
||||
ETCD_INITIAL_CLUSTER=%[3]s
|
||||
ETCD_INITIAL_CLUSTER_STATE=new
|
||||
ETCD_INITIAL_CLUSTER_TOKEN=%[4]s
|
||||
`, m.Name, m.IP, strings.Join(peers, ","), Scope+"-patroni")
|
||||
`, m.Name, m.IP, strings.Join(peers, ","), Scope+"-patroni", scheme, client)
|
||||
if tls {
|
||||
conf += `ETCD_CERT_FILE=` + EtcdCert + `
|
||||
ETCD_KEY_FILE=` + EtcdKey + `
|
||||
ETCD_TRUSTED_CA_FILE=` + EtcdCA + `
|
||||
ETCD_CLIENT_CERT_AUTH=true
|
||||
ETCD_PEER_CERT_FILE=` + EtcdCert + `
|
||||
ETCD_PEER_KEY_FILE=` + EtcdKey + `
|
||||
ETCD_PEER_TRUSTED_CA_FILE=` + EtcdCA + `
|
||||
ETCD_PEER_CLIENT_CERT_AUTH=true
|
||||
`
|
||||
}
|
||||
return conf
|
||||
}
|
||||
|
||||
// EtcdTLSDir holds an etcd member's certificate, which it presents both as
|
||||
// a server and as a client of the other members, and the authority's.
|
||||
const EtcdTLSDir = "/etc/etcd/tls"
|
||||
|
||||
// The files in EtcdTLSDir.
|
||||
const (
|
||||
EtcdCA = EtcdTLSDir + "/ca.crt"
|
||||
EtcdCert = EtcdTLSDir + "/peer.crt"
|
||||
EtcdKey = EtcdTLSDir + "/peer.key"
|
||||
)
|
||||
|
||||
// etcdctl is the command that asks the cluster, over TLS with a member's
|
||||
// certificate when tls; run on a member, as root, which reads it.
|
||||
func etcdctl(tls bool, endpoints []string) string {
|
||||
c := "sudo ETCDCTL_API=3 etcdctl --dial-timeout=3s --endpoints=" + strings.Join(endpoints, ",")
|
||||
if tls {
|
||||
c += " --cacert=" + EtcdCA + " --cert=" + EtcdCert + " --key=" + EtcdKey
|
||||
}
|
||||
return c
|
||||
}
|
||||
|
||||
// Patroni is a database member's /etc/patroni/config.yml.
|
||||
@@ -117,10 +163,10 @@ restapi:
|
||||
authentication:
|
||||
username: %[11]s
|
||||
password: "%[10]s"
|
||||
|
||||
%[13]s
|
||||
etcd3:
|
||||
hosts: %[5]s
|
||||
|
||||
%[14]s
|
||||
bootstrap:
|
||||
dcs:
|
||||
ttl: 30
|
||||
@@ -165,7 +211,7 @@ postgresql:
|
||||
parameters:
|
||||
unix_socket_directories: /var/run/postgresql
|
||||
`, Scope, m.Name, m.IP, PatroniPort, strings.Join(hosts, ","), version, s.Superuser, s.Replicator, hbaLines,
|
||||
s.RestAPI, RestUser, sslLines)
|
||||
s.RestAPI, RestUser, sslLines, restTLSLines(tls), etcdTLSLines(tls))
|
||||
b.WriteString(`
|
||||
tags:
|
||||
nofailover: false
|
||||
@@ -176,6 +222,25 @@ tags:
|
||||
return b.String()
|
||||
}
|
||||
|
||||
// restTLSLines put Patroni's API on TLS, with the member's certificate, and
|
||||
// have patronictl verify it.
|
||||
func restTLSLines(tls bool) string {
|
||||
if !tls {
|
||||
return ""
|
||||
}
|
||||
return " certfile: " + ServerCert + "\n keyfile: " + ServerKey + "\n cafile: " + CACert + "\n" +
|
||||
"\nctl:\n cacert: " + CACert + "\n"
|
||||
}
|
||||
|
||||
// etcdTLSLines have Patroni reach etcd over TLS, presenting the member's
|
||||
// certificate.
|
||||
func etcdTLSLines(tls bool) string {
|
||||
if !tls {
|
||||
return ""
|
||||
}
|
||||
return " protocol: https\n cacert: " + CACert + "\n cert: " + ServerCert + "\n key: " + ServerKey + "\n"
|
||||
}
|
||||
|
||||
// DSN is the store address the control planes are given: every member, and
|
||||
// the driver told to use whichever accepts writes, so a failover is followed
|
||||
// without anything in front of the database. With caFile — the cluster's
|
||||
|
||||
+17
-43
@@ -1,14 +1,11 @@
|
||||
package pgha
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net"
|
||||
"net/http"
|
||||
"slices"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
@@ -59,41 +56,29 @@ func hostssl(line string) string {
|
||||
|
||||
// certs gives each database member a certificate from the cluster's
|
||||
// authority, for its address and name — unless it has one for them already.
|
||||
// It serves PostgreSQL and Patroni's API, and is what Patroni presents to
|
||||
// etcd, so it identifies a client too. A member without PostgreSQL's user
|
||||
// yet is skipped: the Patroni step calls this again once it is installed.
|
||||
func (s *Setup) certs(ctx context.Context) error {
|
||||
if s.CA == nil {
|
||||
s.emit("tls", false, "no certificate authority on this machine: PostgreSQL stays without TLS")
|
||||
return nil
|
||||
}
|
||||
for _, d := range s.DB {
|
||||
have, _ := d.Remote.Run(ctx, "sudo openssl x509 -in "+ServerCert+" -noout -ext subjectAltName 2>/dev/null", nil)
|
||||
if strings.Contains(string(have), "IP Address:"+d.IP) {
|
||||
have, _ := d.Remote.Run(ctx, "id postgres >/dev/null 2>&1 || echo no-postgres; "+
|
||||
"sudo openssl x509 -in "+ServerCert+" -noout -ext subjectAltName,extendedKeyUsage 2>/dev/null", nil)
|
||||
if strings.Contains(string(have), "no-postgres") {
|
||||
continue
|
||||
}
|
||||
leaf, err := s.CA.IssueServer(d.Name, []string{d.IP, d.Name, "127.0.0.1", "localhost"})
|
||||
if strings.Contains(string(have), "IP Address:"+d.IP) && strings.Contains(string(have), "Client Authentication") {
|
||||
continue
|
||||
}
|
||||
leaf, err := s.CA.IssuePeer(d.Name, []string{d.IP, d.Name, "127.0.0.1", "localhost"})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
var body bytes.Buffer
|
||||
body.Write(s.CA.PEM)
|
||||
body.WriteString("\x00")
|
||||
body.Write(leaf.CertPEM)
|
||||
body.WriteString("\x00")
|
||||
body.Write(leaf.KeyPEM)
|
||||
if out, err := d.Remote.Run(ctx, `set -e
|
||||
sudo install -d -o postgres -g postgres -m 750 `+TLSDir+`
|
||||
t=$(mktemp -d); trap 'rm -rf "$t"' EXIT
|
||||
cat > "$t/all"
|
||||
python3 - "$t" <<'PY'
|
||||
import sys
|
||||
d = sys.argv[1]
|
||||
ca, crt, key = open(d + "/all", "rb").read().split(b"\x00")
|
||||
for name, data in (("ca.crt", ca), ("server.crt", crt), ("server.key", key)):
|
||||
open(d + "/" + name, "wb").write(data)
|
||||
PY
|
||||
sudo install -o postgres -g postgres -m 644 "$t/ca.crt" `+CACert+`
|
||||
sudo install -o postgres -g postgres -m 644 "$t/server.crt" `+ServerCert+`
|
||||
sudo install -o postgres -g postgres -m 600 "$t/server.key" `+ServerKey, body.Bytes()); err != nil {
|
||||
return fmt.Errorf("on %s: %v: %s", d.Name, err, strings.TrimSpace(string(out)))
|
||||
if err := putTLS(ctx, d, "postgres", TLSDir, CACert, ServerCert, ServerKey, s.CA.PEM, leaf.CertPEM, leaf.KeyPEM); err != nil {
|
||||
return err
|
||||
}
|
||||
s.emit("tls", true, "certificate for %s (%s)", d.Name, d.IP)
|
||||
}
|
||||
@@ -157,11 +142,9 @@ func (s *Setup) tlsInShared(ctx context.Context, sec Secrets) error {
|
||||
for i, d := range s.DB {
|
||||
hosts[i] = d.IP
|
||||
}
|
||||
client := &http.Client{Timeout: 15 * time.Second}
|
||||
var lastErr error
|
||||
for _, h := range hosts {
|
||||
base := "http://" + net.JoinHostPort(h, strconv.Itoa(restPort))
|
||||
cfg, err := getConfig(ctx, client, base)
|
||||
cfg, err := getConfig(ctx, h)
|
||||
if err != nil {
|
||||
lastErr = err
|
||||
continue
|
||||
@@ -192,7 +175,7 @@ func (s *Setup) tlsInShared(ctx context.Context, sec Secrets) error {
|
||||
if len(hba) > 0 {
|
||||
patch["pg_hba"] = next
|
||||
}
|
||||
if err := patchConfig(ctx, client, base, sec.RestAPI, map[string]any{"postgresql": patch}); err != nil {
|
||||
if err := patchConfig(ctx, h, sec.RestAPI, map[string]any{"postgresql": patch}); err != nil {
|
||||
return fmt.Errorf("Patroni on %s: %w", h, err)
|
||||
}
|
||||
s.emit("tls", true, "PostgreSQL speaks TLS on every member; remote rules are hostssl")
|
||||
@@ -203,17 +186,9 @@ func (s *Setup) tlsInShared(ctx context.Context, sec Secrets) error {
|
||||
|
||||
// patchConfig changes the cluster's shared configuration, with the REST
|
||||
// password.
|
||||
func patchConfig(ctx context.Context, client *http.Client, base, password string, body map[string]any) error {
|
||||
func patchConfig(ctx context.Context, host, password string, body map[string]any) error {
|
||||
b, _ := json.Marshal(body)
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodPatch, base+"/config", bytes.NewReader(b))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
if password != "" {
|
||||
req.SetBasicAuth(RestUser, password)
|
||||
}
|
||||
resp, err := client.Do(req)
|
||||
resp, err := restRequest(ctx, host, http.MethodPatch, "/config", b, password, 15*time.Second)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -228,9 +203,8 @@ func patchConfig(ctx context.Context, client *http.Client, base, password string
|
||||
// for a control plane without the authority's key, which signs nothing but
|
||||
// must still connect by TLS once the cluster asks for it.
|
||||
func (s *Setup) sharedTLS(ctx context.Context) bool {
|
||||
client := &http.Client{Timeout: 10 * time.Second}
|
||||
for _, d := range s.DB {
|
||||
cfg, err := getConfig(ctx, client, "http://"+net.JoinHostPort(d.IP, strconv.Itoa(restPort)))
|
||||
cfg, err := getConfig(ctx, d.IP)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
|
||||
+6
-15
@@ -4,10 +4,8 @@ import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net"
|
||||
"net/http"
|
||||
"slices"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
@@ -28,11 +26,9 @@ func ClientLine(ip string) string {
|
||||
// password is what Patroni's API asks for before a change (RestUser); a
|
||||
// cluster whose rules are hostssl lets the new control plane in by TLS too.
|
||||
func AllowClient(ctx context.Context, hosts []string, ip, password string) (bool, error) {
|
||||
client := &http.Client{Timeout: 5 * time.Second}
|
||||
var lastErr error
|
||||
for _, h := range hosts {
|
||||
base := "http://" + net.JoinHostPort(h, strconv.Itoa(restPort))
|
||||
cfg, err := getConfig(ctx, client, base)
|
||||
cfg, err := getConfig(ctx, h)
|
||||
if err != nil {
|
||||
lastErr = err
|
||||
continue
|
||||
@@ -59,7 +55,7 @@ func AllowClient(ctx context.Context, hosts []string, ip, password string) (bool
|
||||
if slices.Contains(hba, line) {
|
||||
return false, nil
|
||||
}
|
||||
if err := patchConfig(ctx, client, base, password,
|
||||
if err := patchConfig(ctx, h, password,
|
||||
map[string]any{"postgresql": map[string]any{"pg_hba": append(hba, line)}}); err != nil {
|
||||
return false, fmt.Errorf("Patroni on %s: %w", h, err)
|
||||
}
|
||||
@@ -68,19 +64,14 @@ func AllowClient(ctx context.Context, hosts []string, ip, password string) (bool
|
||||
return false, fmt.Errorf("no member's Patroni answered: %w", lastErr)
|
||||
}
|
||||
|
||||
func getConfig(ctx context.Context, client *http.Client, base string) (map[string]any, error) {
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, base+"/config", nil)
|
||||
func getConfig(ctx context.Context, host string) (map[string]any, error) {
|
||||
b, err := restJSON(ctx, host, http.MethodGet, "/config", nil, "", 10*time.Second)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
var cfg map[string]any
|
||||
if err := json.NewDecoder(resp.Body).Decode(&cfg); err != nil {
|
||||
return nil, fmt.Errorf("%s/config: %w", base, err)
|
||||
if err := json.Unmarshal(b, &cfg); err != nil {
|
||||
return nil, fmt.Errorf("%s /config: %w", host, err)
|
||||
}
|
||||
return cfg, nil
|
||||
}
|
||||
@@ -43,11 +43,26 @@ func TestTheConfigurationsNameEveryMemberAndOnlyTheClientsGiven(t *testing.T) {
|
||||
if strings.Count(cfg, "host llmgw llmgw 10.0.0.149/32") != 1 {
|
||||
t.Error("a member's address is allowed twice")
|
||||
}
|
||||
e := Etcd(ctl, []Member{pg01, ctl02, ctl})
|
||||
e := Etcd(ctl, []Member{pg01, ctl02, ctl}, false)
|
||||
if !strings.Contains(e, "ETCD_NAME=llmgw-ctl\n") ||
|
||||
!strings.Contains(e, "llmgw-pg01=http://10.0.0.149:2380,llmgw-ctl02=http://10.0.0.150:2380,llmgw-ctl=http://10.0.0.140:2380") {
|
||||
t.Errorf("etcd defaults:\n%s", e)
|
||||
}
|
||||
e = Etcd(ctl, []Member{pg01, ctl02, ctl}, true)
|
||||
for _, want := range []string{
|
||||
"llmgw-pg01=https://10.0.0.149:2380",
|
||||
"ETCD_LISTEN_CLIENT_URLS=https://10.0.0.140:2379,https://127.0.0.1:2379\n",
|
||||
"ETCD_ADVERTISE_CLIENT_URLS=https://10.0.0.140:2379\n",
|
||||
"ETCD_CLIENT_CERT_AUTH=true\n", "ETCD_PEER_CLIENT_CERT_AUTH=true\n",
|
||||
"ETCD_TRUSTED_CA_FILE=/etc/etcd/tls/ca.crt\n",
|
||||
} {
|
||||
if !strings.Contains(e, want) {
|
||||
t.Errorf("etcd over TLS lacks %q:\n%s", want, e)
|
||||
}
|
||||
}
|
||||
if strings.Contains(e, "http://") {
|
||||
t.Errorf("etcd over TLS still listens in the clear:\n%s", e)
|
||||
}
|
||||
if got := DSN([]Member{pg01, ctl02}, "pw", ""); got !=
|
||||
"postgres://llmgw:pw@10.0.0.149:5432,10.0.0.150:5432/llmgw?sslmode=disable&target_session_attrs=read-write" {
|
||||
t.Errorf("dsn %s", got)
|
||||
@@ -293,7 +308,15 @@ func TestANewClusterAsksForAPasswordAndSpeaksTLS(t *testing.T) {
|
||||
var doc struct {
|
||||
RestAPI struct {
|
||||
Authentication struct{ Username, Password string } `yaml:"authentication"`
|
||||
CertFile string `yaml:"certfile"`
|
||||
CAFile string `yaml:"cafile"`
|
||||
} `yaml:"restapi"`
|
||||
Ctl struct {
|
||||
CACert string `yaml:"cacert"`
|
||||
} `yaml:"ctl"`
|
||||
Etcd3 struct {
|
||||
Protocol, CACert, Cert, Key string
|
||||
} `yaml:"etcd3"`
|
||||
Bootstrap struct {
|
||||
DCS struct {
|
||||
PostgreSQL struct {
|
||||
@@ -309,6 +332,12 @@ func TestANewClusterAsksForAPasswordAndSpeaksTLS(t *testing.T) {
|
||||
if a := doc.RestAPI.Authentication; a.Username != RestUser || a.Password != "the-api-pw" {
|
||||
t.Errorf("rest api authentication %+v", a)
|
||||
}
|
||||
if doc.RestAPI.CertFile != ServerCert || doc.RestAPI.CAFile != CACert || doc.Ctl.CACert != CACert {
|
||||
t.Errorf("the API is not served over TLS:\n%s", cfg)
|
||||
}
|
||||
if e := doc.Etcd3; e.Protocol != "https" || e.CACert != CACert || e.Cert != ServerCert || e.Key != ServerKey {
|
||||
t.Errorf("etcd is not reached over TLS: %+v", e)
|
||||
}
|
||||
p := doc.Bootstrap.DCS.PostgreSQL.Parameters
|
||||
if p["ssl"] != "on" || p["ssl_cert_file"] != ServerCert || p["ssl_key_file"] != ServerKey {
|
||||
t.Errorf("parameters %v", p)
|
||||
|
||||
@@ -0,0 +1,116 @@
|
||||
package pgha
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/tls"
|
||||
"crypto/x509"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"net/http"
|
||||
"os"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Patroni's API answers over TLS once a member has its certificate, and
|
||||
// over plain HTTP on a cluster set up before that, or while it is being
|
||||
// moved. So every request is made over HTTPS first, verified against the
|
||||
// cluster's authority, and over HTTP only when the member answers in plain
|
||||
// HTTP — never when the certificate is wrong.
|
||||
|
||||
var (
|
||||
restMu sync.RWMutex
|
||||
restCA *x509.CertPool
|
||||
)
|
||||
|
||||
// SetRESTCA is the cluster's authority, against which members' API
|
||||
// certificates are verified: the control plane's ca.pem. Without one, the
|
||||
// API is asked over plain HTTP.
|
||||
func SetRESTCA(pemFile string) error {
|
||||
b, err := os.ReadFile(pemFile) //nolint:gosec // the operator's own CA
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return setRESTCAPEM(b)
|
||||
}
|
||||
|
||||
func setRESTCAPEM(b []byte) error {
|
||||
pool := x509.NewCertPool()
|
||||
if !pool.AppendCertsFromPEM(b) {
|
||||
return errors.New("no certificate in the authority's file")
|
||||
}
|
||||
restMu.Lock()
|
||||
restCA = pool
|
||||
restMu.Unlock()
|
||||
return nil
|
||||
}
|
||||
|
||||
// restRequest asks one member's API.
|
||||
func restRequest(ctx context.Context, host, method, path string, body []byte, password string,
|
||||
timeout time.Duration,
|
||||
) (*http.Response, error) {
|
||||
restMu.RLock()
|
||||
pool := restCA
|
||||
restMu.RUnlock()
|
||||
addr := net.JoinHostPort(host, strconv.Itoa(restPort))
|
||||
try := func(scheme string, client *http.Client) (*http.Response, error) {
|
||||
var r io.Reader
|
||||
if body != nil {
|
||||
r = bytes.NewReader(body)
|
||||
}
|
||||
req, err := http.NewRequestWithContext(ctx, method, scheme+"://"+addr+path, r)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if body != nil {
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
}
|
||||
if password != "" {
|
||||
req.SetBasicAuth(RestUser, password)
|
||||
}
|
||||
return client.Do(req)
|
||||
}
|
||||
plain := &http.Client{Timeout: timeout}
|
||||
if pool == nil {
|
||||
return try("http", plain)
|
||||
}
|
||||
secure := &http.Client{Timeout: timeout, Transport: &http.Transport{
|
||||
TLSClientConfig: &tls.Config{RootCAs: pool, MinVersion: tls.VersionTLS12},
|
||||
}}
|
||||
resp, err := try("https", secure)
|
||||
if err != nil && speaksPlainHTTP(err) {
|
||||
return try("http", plain)
|
||||
}
|
||||
return resp, err
|
||||
}
|
||||
|
||||
// speaksPlainHTTP reports whether a TLS attempt failed because the other
|
||||
// end answered in plain HTTP: a member whose API has no certificate yet.
|
||||
func speaksPlainHTTP(err error) bool {
|
||||
var rh tls.RecordHeaderError
|
||||
return errors.As(err, &rh) || strings.Contains(err.Error(), "server gave HTTP response to HTTPS client")
|
||||
}
|
||||
|
||||
// restJSON is restRequest for an answer that must be 200, read whole.
|
||||
func restJSON(ctx context.Context, host, method, path string, body []byte, password string,
|
||||
timeout time.Duration,
|
||||
) ([]byte, error) {
|
||||
resp, err := restRequest(ctx, host, method, path, body, password, timeout)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
b, err := io.ReadAll(io.LimitReader(resp.Body, 4<<20))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return b, fmt.Errorf("%s %s on %s: %s", method, path, host, resp.Status)
|
||||
}
|
||||
return b, nil
|
||||
}
|
||||
@@ -0,0 +1,65 @@
|
||||
package pgha
|
||||
|
||||
import (
|
||||
"crypto/tls"
|
||||
"encoding/pem"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strconv"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/tlsconf"
|
||||
)
|
||||
|
||||
func TestTheAPIIsAskedOverTLSAndOverHTTPOnlyWhenItSpeaksHTTP(t *testing.T) {
|
||||
answer := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
_, _ = w.Write([]byte(`{"scheme":"` + map[bool]string{true: "https", false: "http"}[r.TLS != nil] + `"}`))
|
||||
})
|
||||
secure := httptest.NewTLSServer(answer)
|
||||
defer secure.Close()
|
||||
plain := httptest.NewServer(answer)
|
||||
defer plain.Close()
|
||||
t.Cleanup(func() { restMu.Lock(); restCA = nil; restMu.Unlock(); restPort = PatroniPort })
|
||||
|
||||
if err := setRESTCAPEM(pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: secure.Certificate().Raw})); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ask := func(url string) (string, error) {
|
||||
_, port, _ := net.SplitHostPort(url[len("https://"):])
|
||||
if url[:5] == "http:" {
|
||||
_, port, _ = net.SplitHostPort(url[len("http://"):])
|
||||
}
|
||||
restPort, _ = strconv.Atoi(port)
|
||||
b, err := restJSON(t.Context(), "127.0.0.1", http.MethodGet, "/x", nil, "", 3*time.Second)
|
||||
return string(b), err
|
||||
}
|
||||
if got, err := ask(secure.URL); err != nil || got != `{"scheme":"https"}` {
|
||||
t.Errorf("a member on TLS: %s, %v", got, err)
|
||||
}
|
||||
if got, err := ask(plain.URL); err != nil || got != `{"scheme":"http"}` {
|
||||
t.Errorf("a member not yet on TLS: %s, %v", got, err)
|
||||
}
|
||||
|
||||
// A certificate from elsewhere is refused, not fallen back from.
|
||||
ca, err := tlsconf.NewCA("elsewhere")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
leaf, err := ca.IssueServer("other", []string{"127.0.0.1"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
cert, err := tls.X509KeyPair(leaf.CertPEM, leaf.KeyPEM)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
other := httptest.NewUnstartedServer(answer)
|
||||
other.TLS = &tls.Config{Certificates: []tls.Certificate{cert}}
|
||||
other.StartTLS()
|
||||
defer other.Close()
|
||||
if _, err := ask(other.URL); err == nil {
|
||||
t.Error("a certificate the authority did not sign was accepted")
|
||||
}
|
||||
}
|
||||
+162
-20
@@ -93,12 +93,22 @@ func (s *Setup) Run(ctx context.Context) (string, error) {
|
||||
if s.Status == nil {
|
||||
s.Status = HTTPStatus
|
||||
}
|
||||
// Members' APIs are verified against the cluster's authority once they
|
||||
// speak TLS.
|
||||
switch {
|
||||
case s.CA != nil:
|
||||
if err := setRESTCAPEM(s.CA.PEM); err != nil {
|
||||
return "", err
|
||||
}
|
||||
case s.CAFile != "":
|
||||
_ = SetRESTCA(s.CAFile)
|
||||
}
|
||||
sec, err := s.secrets(ctx)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("passwords: %w", err)
|
||||
}
|
||||
s.RestPassword = sec.RestAPI
|
||||
if err := s.etcd(ctx); err != nil {
|
||||
if err := s.etcd(ctx, sec); err != nil {
|
||||
return "", fmt.Errorf("etcd: %w", err)
|
||||
}
|
||||
if err := s.certs(ctx); err != nil {
|
||||
@@ -115,6 +125,9 @@ func (s *Setup) Run(ctx context.Context) (string, error) {
|
||||
if err := s.tlsInShared(ctx, sec); err != nil {
|
||||
return "", fmt.Errorf("tls: %w", err)
|
||||
}
|
||||
if err := s.ensurePatroniTLS(ctx, sec); err != nil {
|
||||
return "", fmt.Errorf("patroni api: %w", err)
|
||||
}
|
||||
if err := s.database(ctx); err != nil {
|
||||
return "", fmt.Errorf("database: %w", err)
|
||||
}
|
||||
@@ -214,27 +227,117 @@ func parseSecrets(out []byte) Secrets {
|
||||
|
||||
// etcd brings up the voting members, unless every one of them is already
|
||||
// healthy in one cluster — in which case nothing is touched, since rewriting
|
||||
// a member's configuration means emptying its data.
|
||||
func (s *Setup) etcd(ctx context.Context) error {
|
||||
// a member's configuration means emptying its data. With the cluster's
|
||||
// authority here, members speak TLS; a running cluster that does not yet is
|
||||
// moved to it (etcdToTLS).
|
||||
func (s *Setup) etcd(ctx context.Context, sec Secrets) error {
|
||||
nodes := s.etcdMembers()
|
||||
var members []Member
|
||||
var endpoints []string
|
||||
for _, n := range nodes {
|
||||
members = append(members, n.Member)
|
||||
endpoints = append(endpoints, "http://"+n.IP+":2379")
|
||||
}
|
||||
if s.etcdHealthy(ctx, nodes[0], endpoints, len(nodes)) {
|
||||
s.emit("etcd", false, "%d members healthy; left as they are", len(nodes))
|
||||
tls := s.CA != nil
|
||||
if s.etcdHealthy(ctx, nodes[0], etcdEndpoints(nodes, tls), len(nodes), tls) {
|
||||
s.emit("etcd", false, "%d members healthy%s; left as they are", len(nodes), map[bool]string{true: ", over TLS"}[tls])
|
||||
return nil
|
||||
}
|
||||
if tls && s.etcdHealthy(ctx, nodes[0], etcdEndpoints(nodes, false), len(nodes), false) {
|
||||
return s.etcdToTLS(ctx, sec)
|
||||
}
|
||||
return s.etcdBootstrap(ctx, tls)
|
||||
}
|
||||
|
||||
changed := make([]bool, len(nodes))
|
||||
func etcdEndpoints(nodes []Node, tls bool) []string {
|
||||
scheme := "http"
|
||||
if tls {
|
||||
scheme = "https"
|
||||
}
|
||||
out := make([]string, len(nodes))
|
||||
for i, n := range nodes {
|
||||
out[i] = scheme + "://" + n.IP + ":2379"
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// etcdInstall puts etcd on every member, stopped if it was not running.
|
||||
func (s *Setup) etcdInstall(ctx context.Context, nodes []Node) error {
|
||||
for _, n := range nodes {
|
||||
out, err := n.Remote.Run(ctx, `set -e
|
||||
if ! command -v etcd >/dev/null 2>&1; then
|
||||
sudo DEBIAN_FRONTEND=noninteractive apt-get -qq update >/dev/null
|
||||
sudo DEBIAN_FRONTEND=noninteractive apt-get -qq install -y etcd-server etcd-client >/dev/null
|
||||
fi
|
||||
fi`, nil)
|
||||
if err != nil {
|
||||
return fmt.Errorf("on %s: %v: %s", n.Name, err, strings.TrimSpace(string(out)))
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// etcdCerts gives each etcd member a certificate from the cluster's
|
||||
// authority, for its address, its name and loopback, unless it has one.
|
||||
func (s *Setup) etcdCerts(ctx context.Context, nodes []Node) error {
|
||||
for _, n := range nodes {
|
||||
have, _ := n.Remote.Run(ctx, "sudo openssl x509 -in "+EtcdCert+" -noout -ext subjectAltName,extendedKeyUsage 2>/dev/null", nil)
|
||||
if strings.Contains(string(have), "IP Address:"+n.IP) && strings.Contains(string(have), "Client Authentication") {
|
||||
continue
|
||||
}
|
||||
leaf, err := s.CA.IssuePeer(n.Name, []string{n.IP, n.Name, "127.0.0.1", "localhost"})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := putTLS(ctx, n, "etcd", EtcdTLSDir, EtcdCA, EtcdCert, EtcdKey, s.CA.PEM, leaf.CertPEM, leaf.KeyPEM); err != nil {
|
||||
return err
|
||||
}
|
||||
s.emit("etcd", true, "certificate for %s (%s)", n.Name, n.IP)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// putTLS installs an authority's certificate, a leaf and its key on a
|
||||
// machine, owned by user, the key readable by it alone.
|
||||
func putTLS(ctx context.Context, n Node, user, dir, caPath, certPath, keyPath string, ca, cert, key []byte) error {
|
||||
var body bytes.Buffer
|
||||
body.Write(ca)
|
||||
body.WriteString("\x00")
|
||||
body.Write(cert)
|
||||
body.WriteString("\x00")
|
||||
body.Write(key)
|
||||
out, err := n.Remote.Run(ctx, `set -e
|
||||
sudo install -d -o `+user+` -g `+user+` -m 750 `+dir+`
|
||||
t=$(mktemp -d); trap 'rm -rf "$t"' EXIT
|
||||
cat > "$t/all"
|
||||
python3 - "$t" <<'PY'
|
||||
import sys
|
||||
d = sys.argv[1]
|
||||
ca, crt, key = open(d + "/all", "rb").read().split(b"\x00")
|
||||
for name, data in (("ca", ca), ("crt", crt), ("key", key)):
|
||||
open(d + "/" + name, "wb").write(data)
|
||||
PY
|
||||
sudo install -o `+user+` -g `+user+` -m 644 "$t/ca" `+caPath+`
|
||||
sudo install -o `+user+` -g `+user+` -m 644 "$t/crt" `+certPath+`
|
||||
sudo install -o `+user+` -g `+user+` -m 600 "$t/key" `+keyPath, body.Bytes())
|
||||
if err != nil {
|
||||
return fmt.Errorf("on %s: %v: %s", n.Name, err, strings.TrimSpace(string(out)))
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// etcdBootstrap writes every member's configuration and starts them
|
||||
// together, emptying a member whose configuration changed.
|
||||
func (s *Setup) etcdBootstrap(ctx context.Context, tls bool) error {
|
||||
nodes := s.etcdMembers()
|
||||
var members []Member
|
||||
for _, n := range nodes {
|
||||
members = append(members, n.Member)
|
||||
}
|
||||
if err := s.etcdInstall(ctx, nodes); err != nil {
|
||||
return err
|
||||
}
|
||||
if tls {
|
||||
if err := s.etcdCerts(ctx, nodes); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
changed := make([]bool, len(nodes))
|
||||
for i, n := range nodes {
|
||||
out, err := n.Remote.Run(ctx, `set -e
|
||||
new=$(cat)
|
||||
if [ "$(sudo cat /etc/default/etcd 2>/dev/null)" != "$new" ]; then
|
||||
sudo systemctl stop etcd >/dev/null 2>&1 || true
|
||||
@@ -242,7 +345,7 @@ if [ "$(sudo cat /etc/default/etcd 2>/dev/null)" != "$new" ]; then
|
||||
printf '%s\n' "$new" | sudo tee /etc/default/etcd >/dev/null
|
||||
echo changed
|
||||
fi
|
||||
sudo systemctl enable etcd >/dev/null 2>&1`, []byte(Etcd(n.Member, members)))
|
||||
sudo systemctl enable etcd >/dev/null 2>&1`, []byte(Etcd(n.Member, members, tls)))
|
||||
if err != nil {
|
||||
return fmt.Errorf("on %s: %v: %s", n.Name, err, strings.TrimSpace(string(out)))
|
||||
}
|
||||
@@ -273,7 +376,7 @@ sudo systemctl enable etcd >/dev/null 2>&1`, []byte(Etcd(n.Member, members)))
|
||||
}
|
||||
}
|
||||
deadline := time.Now().Add(s.Wait)
|
||||
for !s.etcdHealthy(ctx, nodes[0], endpoints, len(nodes)) {
|
||||
for !s.etcdHealthy(ctx, nodes[0], etcdEndpoints(nodes, tls), len(nodes), tls) {
|
||||
if time.Now().After(deadline) {
|
||||
return fmt.Errorf("the %d members did not become healthy within %s; `journalctl -u etcd` on each says why", len(nodes), s.Wait)
|
||||
}
|
||||
@@ -281,16 +384,48 @@ sudo systemctl enable etcd >/dev/null 2>&1`, []byte(Etcd(n.Member, members)))
|
||||
return err
|
||||
}
|
||||
}
|
||||
s.emit("etcd", true, "%d members up", len(nodes))
|
||||
s.emit("etcd", true, "%d members up%s", len(nodes), map[bool]string{true: ", over TLS"}[tls])
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Setup) etcdHealthy(ctx context.Context, n Node, endpoints []string, want int) bool {
|
||||
out, err := n.Remote.Run(ctx, "ETCDCTL_API=3 etcdctl --dial-timeout=3s --endpoints="+
|
||||
strings.Join(endpoints, ",")+" endpoint health 2>&1", nil)
|
||||
func (s *Setup) etcdHealthy(ctx context.Context, n Node, endpoints []string, want int, tls bool) bool {
|
||||
out, err := n.Remote.Run(ctx, etcdctl(tls, endpoints)+" endpoint health 2>&1", nil)
|
||||
return err == nil && strings.Count(string(out), "is healthy") == want
|
||||
}
|
||||
|
||||
// ensurePatroniTLS puts a member whose Patroni does not yet reach etcd or
|
||||
// serve its API over TLS on it, in maintenance mode: an etcd moved to TLS by
|
||||
// a run that stopped before it got this far.
|
||||
func (s *Setup) ensurePatroniTLS(ctx context.Context, sec Secrets) error {
|
||||
if s.CA == nil {
|
||||
return nil
|
||||
}
|
||||
need := false
|
||||
for _, d := range s.DB {
|
||||
out, err := d.Remote.Run(ctx, patroniTLSEdit("check"), nil)
|
||||
if err != nil {
|
||||
return fmt.Errorf("on %s: %v: %s", d.Name, err, strings.TrimSpace(string(out)))
|
||||
}
|
||||
need = need || strings.TrimSpace(lastLine(out)) == "changed"
|
||||
}
|
||||
if !need {
|
||||
return nil
|
||||
}
|
||||
if err := s.pause(ctx, sec, true); err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err := s.patroniOnTLS(ctx, true); err != nil {
|
||||
return s.stuck(err)
|
||||
}
|
||||
if err := s.waitFor(ctx, s.dbHosts(), func(c Cluster) bool { return c.Leader() != "" }); err != nil {
|
||||
return s.stuck(err)
|
||||
}
|
||||
if err := s.pause(ctx, sec, false); err != nil {
|
||||
return s.stuck(err)
|
||||
}
|
||||
return s.waitFor(ctx, s.dbHosts(), func(c Cluster) bool { return c.Complete(len(s.DB)) })
|
||||
}
|
||||
|
||||
// patroni installs PostgreSQL and Patroni on every database machine, unless
|
||||
// the cluster already has a leader and every member streaming.
|
||||
func (s *Setup) patroni(ctx context.Context, sec Secrets) error {
|
||||
@@ -315,6 +450,10 @@ func (s *Setup) patroni(ctx context.Context, sec Secrets) error {
|
||||
}
|
||||
version = v
|
||||
}
|
||||
// Now that PostgreSQL's user exists, its certificates.
|
||||
if err := s.certs(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
var etcd, db []Member
|
||||
for _, n := range s.etcdMembers() {
|
||||
@@ -409,6 +548,7 @@ func (s *Setup) database(ctx context.Context) error {
|
||||
}
|
||||
out, err := leader.Remote.Run(ctx, `set -e
|
||||
pw=$(sudo cat `+secretDir+`/llmgw.pw)
|
||||
sudo -u postgres psql -Atc "select 'had=' || (select count(*) from pg_roles where rolname = 'llmgw') + (select count(*) from pg_database where datname = 'llmgw')"
|
||||
sudo -u postgres psql -q -v ON_ERROR_STOP=1 -v pw="$pw" <<'SQL'
|
||||
SELECT format('CREATE ROLE %I LOGIN', 'llmgw') WHERE NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'llmgw') \gexec
|
||||
ALTER ROLE llmgw PASSWORD :'pw';
|
||||
@@ -417,7 +557,9 @@ SQL`, nil)
|
||||
if err != nil {
|
||||
return fmt.Errorf("on %s: %v: %s", leader.Name, err, strings.TrimSpace(string(out)))
|
||||
}
|
||||
s.emit("database", true, "role and database %s on %s", Database, leader.Name)
|
||||
// Said as a change only when something was made: a rerun that finds
|
||||
// both is not a reason to restart the gateway.
|
||||
s.emit("database", !strings.Contains(string(out), "had=2"), "role and database %s on %s", Database, leader.Name)
|
||||
return nil
|
||||
}
|
||||
|
||||
|
||||
+3
-13
@@ -4,9 +4,7 @@ import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
@@ -70,24 +68,16 @@ func (c Cluster) String() string {
|
||||
// HTTPStatus asks each host's Patroni in turn and returns the first answer:
|
||||
// any member describes the whole cluster.
|
||||
func HTTPStatus(ctx context.Context, hosts []string) (Cluster, error) {
|
||||
client := &http.Client{Timeout: 3 * time.Second}
|
||||
var last error
|
||||
for _, h := range hosts {
|
||||
u := "http://" + net.JoinHostPort(h, strconv.Itoa(restPort)) + "/cluster"
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil)
|
||||
if err != nil {
|
||||
return Cluster{}, err
|
||||
}
|
||||
resp, err := client.Do(req)
|
||||
b, err := restJSON(ctx, h, http.MethodGet, "/cluster", nil, "", 3*time.Second)
|
||||
if err != nil {
|
||||
last = err
|
||||
continue
|
||||
}
|
||||
var c Cluster
|
||||
err = json.NewDecoder(resp.Body).Decode(&c)
|
||||
resp.Body.Close()
|
||||
if err != nil {
|
||||
last = fmt.Errorf("%s: %w", u, err)
|
||||
if err := json.Unmarshal(b, &c); err != nil {
|
||||
last = fmt.Errorf("%s /cluster: %w", h, err)
|
||||
continue
|
||||
}
|
||||
for i := range c.Members {
|
||||
|
||||
@@ -0,0 +1,236 @@
|
||||
package pgha
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/base64"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Moving a running cluster's etcd to TLS cannot be done member by member:
|
||||
// a member's peers reach it at the address the cluster recorded, http, and
|
||||
// changing that one member at a time loses the majority half way. So it is
|
||||
// done the way a first install is — every member's configuration rewritten
|
||||
// and started together, empty — with what Patroni keeps there carried over.
|
||||
//
|
||||
// While etcd is down Patroni would demote the primary. So the cluster is in
|
||||
// maintenance mode (pause) throughout: Patroni stops managing PostgreSQL,
|
||||
// which goes on serving, and a lost leader lock means nothing. Patroni is
|
||||
// then restarted against the new etcd, which in maintenance mode leaves
|
||||
// PostgreSQL running, and maintenance ends.
|
||||
|
||||
// etcdToTLS moves a healthy plain-HTTP etcd, and Patroni with it, to TLS.
|
||||
func (s *Setup) etcdToTLS(ctx context.Context, sec Secrets) error {
|
||||
nodes := s.etcdMembers()
|
||||
if err := s.etcdCerts(ctx, nodes); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := s.certs(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := s.pause(ctx, sec, true); err != nil {
|
||||
return fmt.Errorf("maintenance mode: %w", err)
|
||||
}
|
||||
s.emit("etcd", true, "Patroni in maintenance mode: PostgreSQL serves on while etcd moves to TLS")
|
||||
|
||||
kept, err := s.etcdDump(ctx, nodes[0])
|
||||
if err != nil {
|
||||
return s.stuck(fmt.Errorf("reading what Patroni keeps in etcd: %w", err))
|
||||
}
|
||||
if err := s.etcdBootstrap(ctx, true); err != nil {
|
||||
return s.stuck(err)
|
||||
}
|
||||
if err := s.etcdRestore(ctx, nodes[0], kept); err != nil {
|
||||
return s.stuck(fmt.Errorf("putting Patroni's keys back: %w", err))
|
||||
}
|
||||
s.emit("etcd", true, "%d of Patroni's keys carried over", len(kept))
|
||||
|
||||
if _, err := s.patroniOnTLS(ctx, true); err != nil {
|
||||
return s.stuck(err)
|
||||
}
|
||||
hosts := s.dbHosts()
|
||||
if err := s.waitFor(ctx, hosts, func(c Cluster) bool { return c.Leader() != "" }); err != nil {
|
||||
return s.stuck(fmt.Errorf("Patroni against the new etcd: %w", err))
|
||||
}
|
||||
if err := s.pause(ctx, sec, false); err != nil {
|
||||
return s.stuck(fmt.Errorf("ending maintenance mode: %w", err))
|
||||
}
|
||||
if err := s.waitFor(ctx, hosts, func(c Cluster) bool { return c.Complete(len(s.DB)) }); err != nil {
|
||||
return err
|
||||
}
|
||||
s.emit("etcd", true, "etcd and Patroni's API speak TLS; maintenance mode ended")
|
||||
return nil
|
||||
}
|
||||
|
||||
// stuck says where a move that failed half way leaves the cluster.
|
||||
func (s *Setup) stuck(err error) error {
|
||||
return fmt.Errorf("%w — the cluster is left in maintenance mode, PostgreSQL still serving; "+
|
||||
"run this again, or end it with `patronictl -c /etc/patroni/config.yml resume` on a member", err)
|
||||
}
|
||||
|
||||
func (s *Setup) dbHosts() []string {
|
||||
hosts := make([]string, len(s.DB))
|
||||
for i, d := range s.DB {
|
||||
hosts[i] = d.IP
|
||||
}
|
||||
return hosts
|
||||
}
|
||||
|
||||
// pause turns maintenance mode on or off, and waits until every member says
|
||||
// so.
|
||||
func (s *Setup) pause(ctx context.Context, sec Secrets, on bool) error {
|
||||
var err error
|
||||
for _, h := range s.dbHosts() {
|
||||
if err = patchConfig(ctx, h, sec.RestAPI, map[string]any{"pause": on}); err == nil {
|
||||
break
|
||||
}
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
deadline := time.Now().Add(s.Wait)
|
||||
for {
|
||||
all := true
|
||||
for _, h := range s.dbHosts() {
|
||||
b, err := restJSON(ctx, h, http.MethodGet, "/patroni", nil, "", 5*time.Second)
|
||||
var st struct {
|
||||
Pause bool `json:"pause"`
|
||||
}
|
||||
if err != nil || json.Unmarshal(b, &st) != nil || st.Pause != on {
|
||||
all = false
|
||||
}
|
||||
}
|
||||
if all {
|
||||
return nil
|
||||
}
|
||||
if time.Now().After(deadline) {
|
||||
return fmt.Errorf("not every member was in maintenance mode=%v within %s", on, s.Wait)
|
||||
}
|
||||
if err := sleep(ctx, 2*time.Second); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// etcdKV is one key Patroni keeps, and whether it is tied to a lease: a
|
||||
// leader lock or a member's entry, which Patroni writes again itself and
|
||||
// which, put back without its lease, would never expire.
|
||||
type etcdKV struct {
|
||||
Key, Value []byte
|
||||
Lease int64
|
||||
}
|
||||
|
||||
func (s *Setup) etcdDump(ctx context.Context, n Node) ([]etcdKV, error) {
|
||||
out, err := n.Remote.Run(ctx, etcdctl(false, []string{"http://" + n.IP + ":2379"})+
|
||||
" get --prefix /service/ -w json", nil)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("%v: %s", err, strings.TrimSpace(string(out)))
|
||||
}
|
||||
var got struct {
|
||||
KVs []struct {
|
||||
Key string `json:"key"`
|
||||
Value string `json:"value"`
|
||||
Lease int64 `json:"lease"`
|
||||
} `json:"kvs"`
|
||||
}
|
||||
if err := json.Unmarshal(out, &got); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var kvs []etcdKV
|
||||
for _, kv := range got.KVs {
|
||||
k, err1 := base64.StdEncoding.DecodeString(kv.Key)
|
||||
v, err2 := base64.StdEncoding.DecodeString(kv.Value)
|
||||
if err1 != nil || err2 != nil {
|
||||
return nil, fmt.Errorf("an unreadable key in etcd's answer")
|
||||
}
|
||||
if kv.Lease == 0 {
|
||||
kvs = append(kvs, etcdKV{Key: k, Value: v})
|
||||
}
|
||||
}
|
||||
if len(kvs) == 0 {
|
||||
return nil, fmt.Errorf("etcd holds nothing of Patroni's under /service/")
|
||||
}
|
||||
return kvs, nil
|
||||
}
|
||||
|
||||
func (s *Setup) etcdRestore(ctx context.Context, n Node, kvs []etcdKV) error {
|
||||
for _, kv := range kvs {
|
||||
if strings.ContainsAny(string(kv.Key), "'\n") {
|
||||
return fmt.Errorf("a key that cannot be written back: %q", kv.Key)
|
||||
}
|
||||
out, err := n.Remote.Run(ctx, etcdctl(true, []string{"https://" + n.IP + ":2379"})+
|
||||
" put '"+string(kv.Key)+"'", kv.Value)
|
||||
if err != nil {
|
||||
return fmt.Errorf("%s: %v: %s", kv.Key, err, strings.TrimSpace(string(out)))
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// patroniOnTLS has each member's Patroni reach etcd over TLS and serve its
|
||||
// API over TLS, restarting those whose configuration changed. With paused
|
||||
// false, maintenance mode is entered for the restart and left after.
|
||||
func (s *Setup) patroniOnTLS(ctx context.Context, paused bool) (bool, error) {
|
||||
var todo []Node
|
||||
for _, d := range s.DB {
|
||||
out, err := d.Remote.Run(ctx, patroniTLSEdit("check"), nil)
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("on %s: %v: %s", d.Name, err, strings.TrimSpace(string(out)))
|
||||
}
|
||||
switch strings.TrimSpace(lastLine(out)) {
|
||||
case "changed":
|
||||
todo = append(todo, d)
|
||||
case "missing":
|
||||
return false, fmt.Errorf("%s has no /etc/patroni/config.yml", d.Name)
|
||||
}
|
||||
}
|
||||
if len(todo) == 0 {
|
||||
return false, nil
|
||||
}
|
||||
if !paused {
|
||||
return false, fmt.Errorf("Patroni on %d member(s) is not yet on TLS, which needs maintenance mode", len(todo))
|
||||
}
|
||||
for _, d := range todo {
|
||||
if out, err := d.Remote.Run(ctx, patroniTLSEdit("apply")+`
|
||||
sudo systemctl restart patroni`, nil); err != nil {
|
||||
return false, fmt.Errorf("on %s: %v: %s", d.Name, err, strings.TrimSpace(string(out)))
|
||||
}
|
||||
s.emit("patroni api", true, "%s reaches etcd over TLS and serves its API over TLS", d.Name)
|
||||
}
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// patroniTLSEdit edits /etc/patroni/config.yml: "check" says whether it
|
||||
// would change, "apply" writes it.
|
||||
func patroniTLSEdit(mode string) string {
|
||||
return `set -e
|
||||
test -f /etc/patroni/config.yml || { echo missing; exit 0; }
|
||||
sudo python3 - ` + mode + ` <<'PY'
|
||||
import sys, yaml
|
||||
p = "/etc/patroni/config.yml"
|
||||
c = yaml.safe_load(open(p))
|
||||
want = {
|
||||
"restapi": {"certfile": "` + ServerCert + `", "keyfile": "` + ServerKey + `", "cafile": "` + CACert + `"},
|
||||
"ctl": {"cacert": "` + CACert + `"},
|
||||
"etcd3": {"protocol": "https", "cacert": "` + CACert + `", "cert": "` + ServerCert + `", "key": "` + ServerKey + `"},
|
||||
}
|
||||
changed = False
|
||||
for section, kv in want.items():
|
||||
sec = c.setdefault(section, {})
|
||||
for k, v in kv.items():
|
||||
if sec.get(k) != v:
|
||||
sec[k] = v
|
||||
changed = True
|
||||
if sys.argv[1] == "apply" and changed:
|
||||
open(p + ".new", "w").write("# llmgw: the control planes' store, kept by Patroni, its leader lock in etcd.\n" +
|
||||
yaml.safe_dump(c, default_flow_style=False, sort_keys=False))
|
||||
import os, shutil
|
||||
shutil.chown(p + ".new", "root", "postgres")
|
||||
os.chmod(p + ".new", 0o640)
|
||||
os.replace(p + ".new", p)
|
||||
print("changed" if changed else "same")
|
||||
PY`
|
||||
}
|
||||
+12
-2
@@ -151,7 +151,17 @@ func (ca *CA) IssueClient(nodeID string) (Leaf, error) {
|
||||
return ca.issue(nodeID, nil, x509.ExtKeyUsageClientAuth)
|
||||
}
|
||||
|
||||
func (ca *CA) issue(commonName string, hosts []string, usage x509.ExtKeyUsage) (Leaf, error) {
|
||||
// IssuePeer issues a certificate that serves and also identifies its holder
|
||||
// as a client: what an etcd member presents to the others, both ways, and
|
||||
// to the clients that must show one.
|
||||
func (ca *CA) IssuePeer(commonName string, hosts []string) (Leaf, error) {
|
||||
if len(hosts) == 0 {
|
||||
return Leaf{}, fmt.Errorf("a peer certificate needs at least one host")
|
||||
}
|
||||
return ca.issue(commonName, hosts, x509.ExtKeyUsageServerAuth, x509.ExtKeyUsageClientAuth)
|
||||
}
|
||||
|
||||
func (ca *CA) issue(commonName string, hosts []string, usage ...x509.ExtKeyUsage) (Leaf, error) {
|
||||
key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
|
||||
if err != nil {
|
||||
return Leaf{}, err
|
||||
@@ -167,7 +177,7 @@ func (ca *CA) issue(commonName string, hosts []string, usage x509.ExtKeyUsage) (
|
||||
NotBefore: time.Now().Add(-Backdate),
|
||||
NotAfter: time.Now().Add(LeafValidity),
|
||||
KeyUsage: x509.KeyUsageDigitalSignature | x509.KeyUsageKeyEncipherment,
|
||||
ExtKeyUsage: []x509.ExtKeyUsage{usage},
|
||||
ExtKeyUsage: usage,
|
||||
}
|
||||
for _, h := range hosts {
|
||||
if ip := net.ParseIP(h); ip != nil {
|
||||
|
||||
Reference in new issue
Block a user