From 1616edcee3dcf43f916b95aa6fda2c9fca27df8f Mon Sep 17 00:00:00 2001 From: Jakob Borg Date: Wed, 11 Sep 2024 11:43:58 +0200 Subject: [PATCH] chore(stdiscosrv): remove legacy replication --- cmd/stdiscosrv/amqp.go | 4 - cmd/stdiscosrv/apisrv.go | 4 + cmd/stdiscosrv/main.go | 94 ++-------- cmd/stdiscosrv/replication.go | 334 ---------------------------------- 4 files changed, 19 insertions(+), 417 deletions(-) delete mode 100644 cmd/stdiscosrv/replication.go diff --git a/cmd/stdiscosrv/amqp.go b/cmd/stdiscosrv/amqp.go index e32eea49e..bb99c1e6d 100644 --- a/cmd/stdiscosrv/amqp.go +++ b/cmd/stdiscosrv/amqp.go @@ -7,7 +7,6 @@ package main import ( - "bytes" "context" "fmt" "io" @@ -164,9 +163,6 @@ func (s *amqpReceiver) Serve(ctx context.Context) error { replicationRecvsTotal.WithLabelValues("error").Inc() return fmt.Errorf("replication unmarshal: %w", err) } - if bytes.Equal(rec.Key, []byte("")) { - continue - } id, err := protocol.DeviceIDFromBytes(rec.Key) if err != nil { id, err = protocol.DeviceIDFromString(string(rec.Key)) diff --git a/cmd/stdiscosrv/apisrv.go b/cmd/stdiscosrv/apisrv.go index 2c7c874c9..fced38e09 100644 --- a/cmd/stdiscosrv/apisrv.go +++ b/cmd/stdiscosrv/apisrv.go @@ -51,6 +51,10 @@ type apiSrv struct { notSeenTracker *retryAfterTracker } +type replicator interface { + send(key *protocol.DeviceID, addrs []DatabaseAddress, seen int64) +} + type requestID int64 func (i requestID) String() string { diff --git a/cmd/stdiscosrv/main.go b/cmd/stdiscosrv/main.go index daeb4b3c7..069dd02a5 100644 --- a/cmd/stdiscosrv/main.go +++ b/cmd/stdiscosrv/main.go @@ -10,12 +10,10 @@ import ( "context" "crypto/tls" "log" - "net" "net/http" "os" "os/signal" "runtime" - "strings" "time" _ "net/http/pprof" @@ -66,11 +64,6 @@ type CLI struct { Listen string `group:"Listen" help:"Listen address" default:":8443" env:"DISCOVERY_LISTEN"` MetricsListen string `group:"Listen" help:"Metrics listen address" env:"DISCOVERY_METRICS_LISTEN"` - Replicate []string `group:"Legacy replication" help:"Replication peers, id@address, comma separated" env:"DISCOVERY_REPLICATE"` - ReplicationListen string `group:"Legacy replication" help:"Replication listen address" default:":19200" env:"DISCOVERY_REPLICATION_LISTEN"` - ReplicationCert string `group:"Legacy replication" help:"Certificate file for replication" env:"DISCOVERY_REPLICATION_CERT_FILE"` - ReplicationKey string `group:"Legacy replication" help:"Key file for replication" env:"DISCOVERY_REPLICATION_KEY_FILE"` - AMQPAddress string `group:"AMQP replication" help:"Address to AMQP broker" env:"DISCOVERY_AMQP_ADDRESS"` DBDir string `group:"Database" help:"Database directory" default:"." env:"DISCOVERY_DB_DIR"` @@ -100,65 +93,21 @@ func main() { buildInfo.WithLabelValues(build.Version, runtime.Version(), build.User, build.Date.UTC().Format("2006-01-02T15:04:05Z")).Set(1) - cert, err := tls.LoadX509KeyPair(cli.Cert, cli.Key) - if os.IsNotExist(err) { - log.Println("Failed to load keypair. Generating one, this might take a while...") - cert, err = tlsutil.NewCertificate(cli.Cert, cli.Key, "stdiscosrv", 20*365) - if err != nil { - log.Fatalln("Failed to generate X509 key pair:", err) - } - } else if err != nil { - log.Fatalln("Failed to load keypair:", err) - } - devID := protocol.NewDeviceID(cert.Certificate[0]) - log.Println("Server device ID is", devID) - - replCert := cert - if cli.ReplicationCert != "" && cli.ReplicationKey != "" { - replCert, err = tls.LoadX509KeyPair(cli.ReplicationCert, cli.ReplicationKey) - if err != nil { - log.Fatalln("Failed to load replication keypair:", err) - } - } - replDevID := protocol.NewDeviceID(replCert.Certificate[0]) - log.Println("Replication device ID is", replDevID) - - // Parse the replication specs, if any. - var allowedReplicationPeers []protocol.DeviceID - var replicationDestinations []string - for _, part := range cli.Replicate { - if part == "" { - continue - } - - fields := strings.Split(part, "@") - switch len(fields) { - case 2: - // This is an id@address specification. Grab the address for the - // destination list. Try to resolve it once to catch obvious - // syntax errors here rather than having the sender service fail - // repeatedly later. - _, err := net.ResolveTCPAddr("tcp", fields[1]) + var cert tls.Certificate + if !cli.HTTP { + var err error + cert, err = tls.LoadX509KeyPair(cli.Cert, cli.Key) + if os.IsNotExist(err) { + log.Println("Failed to load keypair. Generating one, this might take a while...") + cert, err = tlsutil.NewCertificate(cli.Cert, cli.Key, "stdiscosrv", 20*365) if err != nil { - log.Fatalln("Resolving address:", err) + log.Fatalln("Failed to generate X509 key pair:", err) } - replicationDestinations = append(replicationDestinations, fields[1]) - fallthrough // N.B. - - case 1: - // The first part is always a device ID. - id, err := protocol.DeviceIDFromString(fields[0]) - if err != nil { - log.Fatalln("Parsing device ID:", err) - } - if id == protocol.EmptyDeviceID { - log.Fatalf("Missing device ID for peer in %q", part) - } - allowedReplicationPeers = append(allowedReplicationPeers, id) - - default: - log.Fatalln("Unrecognized replication spec:", part) + } else if err != nil { + log.Fatalln("Failed to load keypair:", err) } + devID := protocol.NewDeviceID(cert.Certificate[0]) + log.Println("Server device ID is", devID) } // Root of the service tree. @@ -182,26 +131,13 @@ func main() { db := newInMemoryStore(cli.DBDir, cli.DBFlushInterval, s3c) main.Add(db) - // Start any replication senders. - var repl replicationMultiplexer - for _, dst := range replicationDestinations { - rs := newReplicationSender(dst, replCert, allowedReplicationPeers) - main.Add(rs) - repl = append(repl, rs) - } - - // If we have replication configured, start the replication listener. - if len(allowedReplicationPeers) > 0 { - rl := newReplicationListener(cli.ReplicationListen, replCert, allowedReplicationPeers, db) - main.Add(rl) - } - - // If we have an AMQP broker, start that + // If we have an AMQP broker for replication, start that + var repl replicator if cli.AMQPAddress != "" { clientID := rand.String(10) kr := newAMQPReplicator(cli.AMQPAddress, clientID, db) - repl = append(repl, kr) main.Add(kr) + repl = kr } // Start the main API server. diff --git a/cmd/stdiscosrv/replication.go b/cmd/stdiscosrv/replication.go deleted file mode 100644 index a39304e1a..000000000 --- a/cmd/stdiscosrv/replication.go +++ /dev/null @@ -1,334 +0,0 @@ -// Copyright (C) 2018 The Syncthing Authors. -// -// This Source Code Form is subject to the terms of the Mozilla Public -// License, v. 2.0. If a copy of the MPL was not distributed with this file, -// You can obtain one at https://mozilla.org/MPL/2.0/. - -package main - -import ( - "context" - "crypto/tls" - "encoding/binary" - "fmt" - io "io" - "log" - "net" - "time" - - "github.com/syncthing/syncthing/lib/protocol" -) - -const ( - replicationReadTimeout = time.Minute - replicationWriteTimeout = 30 * time.Second - replicationHeartbeatInterval = time.Second * 30 -) - -type replicator interface { - send(key *protocol.DeviceID, addrs []DatabaseAddress, seen int64) -} - -// a replicationSender tries to connect to the remote address and provide -// them with a feed of replication updates. -type replicationSender struct { - dst string - cert tls.Certificate // our certificate - allowedIDs []protocol.DeviceID - outbox chan ReplicationRecord -} - -func newReplicationSender(dst string, cert tls.Certificate, allowedIDs []protocol.DeviceID) *replicationSender { - return &replicationSender{ - dst: dst, - cert: cert, - allowedIDs: allowedIDs, - outbox: make(chan ReplicationRecord, replicationOutboxSize), - } -} - -func (s *replicationSender) Serve(ctx context.Context) error { - // Sleep a little at startup. Peers often restart at the same time, and - // this avoid the service failing and entering backoff state - // unnecessarily, while also reducing the reconnect rate to something - // reasonable by default. - time.Sleep(2 * time.Second) - - tlsCfg := &tls.Config{ - Certificates: []tls.Certificate{s.cert}, - MinVersion: tls.VersionTLS12, - InsecureSkipVerify: true, - } - - // Dial the TLS connection. - conn, err := tls.Dial("tcp", s.dst, tlsCfg) - if err != nil { - log.Println("Replication connect:", err) - return err - } - defer func() { - conn.SetWriteDeadline(time.Now().Add(time.Second)) - conn.Close() - }() - - // The replication stream is not especially latency sensitive, but it is - // quite a lot of data in small writes. Make it more efficient. - if tcpc, ok := conn.NetConn().(*net.TCPConn); ok { - _ = tcpc.SetNoDelay(false) - } - - // Get the other side device ID. - remoteID, err := deviceID(conn) - if err != nil { - log.Println("Replication connect:", err) - return err - } - - // Verify it's in the set of allowed device IDs. - if !deviceIDIn(remoteID, s.allowedIDs) { - log.Println("Replication connect: unexpected device ID:", remoteID) - return err - } - - heartBeatTicker := time.NewTicker(replicationHeartbeatInterval) - defer heartBeatTicker.Stop() - - // Send records. - buf := make([]byte, 1024) - for { - select { - case <-heartBeatTicker.C: - if len(s.outbox) > 0 { - // No need to send heartbeats if there are events/prevrious - // heartbeats to send, they will keep the connection alive. - continue - } - // Empty replication message is the heartbeat: - s.outbox <- ReplicationRecord{} - - case rec := <-s.outbox: - // Buffer must hold record plus four bytes for size - size := rec.Size() - if len(buf) < size+4 { - buf = make([]byte, size+4) - } - - // Record comes after the four bytes size - n, err := rec.MarshalTo(buf[4:]) - if err != nil { - // odd to get an error here, but we haven't sent anything - // yet so it's not fatal - replicationSendsTotal.WithLabelValues("error").Inc() - log.Println("Replication marshal:", err) - continue - } - binary.BigEndian.PutUint32(buf, uint32(n)) - - // Send - conn.SetWriteDeadline(time.Now().Add(replicationWriteTimeout)) - if _, err := conn.Write(buf[:4+n]); err != nil { - replicationSendsTotal.WithLabelValues("error").Inc() - log.Println("Replication write:", err) - // Yes, we are losing the replication event here. - return err - } - replicationSendsTotal.WithLabelValues("success").Inc() - - case <-ctx.Done(): - return nil - } - } -} - -func (s *replicationSender) String() string { - return fmt.Sprintf("replicationSender(%q)", s.dst) -} - -func (s *replicationSender) send(key *protocol.DeviceID, ps []DatabaseAddress, seen int64) { - item := ReplicationRecord{ - Key: key[:], - Addresses: ps, - Seen: seen, - } - - // The send should never block. The inbox is suitably buffered for at - // least a few seconds of stalls, which shouldn't happen in practice. - select { - case s.outbox <- item: - default: - replicationSendsTotal.WithLabelValues("drop").Inc() - } -} - -// a replicationMultiplexer sends to multiple replicators -type replicationMultiplexer []replicator - -func (m replicationMultiplexer) send(key *protocol.DeviceID, ps []DatabaseAddress, seen int64) { - for _, s := range m { - // each send is nonblocking - s.send(key, ps, seen) - } -} - -// replicationListener accepts incoming connections and reads replication -// items from them. Incoming items are applied to the KV store. -type replicationListener struct { - addr string - cert tls.Certificate - allowedIDs []protocol.DeviceID - db database -} - -func newReplicationListener(addr string, cert tls.Certificate, allowedIDs []protocol.DeviceID, db database) *replicationListener { - return &replicationListener{ - addr: addr, - cert: cert, - allowedIDs: allowedIDs, - db: db, - } -} - -func (l *replicationListener) Serve(ctx context.Context) error { - tlsCfg := &tls.Config{ - Certificates: []tls.Certificate{l.cert}, - ClientAuth: tls.RequestClientCert, - MinVersion: tls.VersionTLS12, - InsecureSkipVerify: true, - } - - lst, err := tls.Listen("tcp", l.addr, tlsCfg) - if err != nil { - log.Println("Replication listen:", err) - return err - } - defer lst.Close() - - for { - select { - case <-ctx.Done(): - return nil - default: - } - - // Accept a connection - conn, err := lst.Accept() - if err != nil { - log.Println("Replication accept:", err) - return err - } - - // Figure out the other side device ID - remoteID, err := deviceID(conn.(*tls.Conn)) - if err != nil { - log.Println("Replication accept:", err) - conn.SetWriteDeadline(time.Now().Add(time.Second)) - conn.Close() - continue - } - - // Verify it is in the set of allowed device IDs - if !deviceIDIn(remoteID, l.allowedIDs) { - log.Println("Replication accept: unexpected device ID:", remoteID) - conn.SetWriteDeadline(time.Now().Add(time.Second)) - conn.Close() - continue - } - - go l.handle(ctx, conn) - } -} - -func (l *replicationListener) String() string { - return fmt.Sprintf("replicationListener(%q)", l.addr) -} - -func (l *replicationListener) handle(ctx context.Context, conn net.Conn) { - defer func() { - conn.SetWriteDeadline(time.Now().Add(time.Second)) - conn.Close() - }() - - buf := make([]byte, 1024) - - for { - select { - case <-ctx.Done(): - return - default: - } - - conn.SetReadDeadline(time.Now().Add(replicationReadTimeout)) - - // First four bytes are the size - if _, err := io.ReadFull(conn, buf[:4]); err != nil { - log.Println("Replication read size:", err) - replicationRecvsTotal.WithLabelValues("error").Inc() - return - } - - // Read the rest of the record - size := int(binary.BigEndian.Uint32(buf[:4])) - if len(buf) < size { - buf = make([]byte, size) - } - - if size == 0 { - // Heartbeat, ignore - continue - } - - if _, err := io.ReadFull(conn, buf[:size]); err != nil { - log.Println("Replication read record:", err) - replicationRecvsTotal.WithLabelValues("error").Inc() - return - } - - // Unmarshal - var rec ReplicationRecord - if err := rec.Unmarshal(buf[:size]); err != nil { - log.Println("Replication unmarshal:", err) - replicationRecvsTotal.WithLabelValues("error").Inc() - continue - } - id, err := protocol.DeviceIDFromBytes(rec.Key) - if err != nil { - id, err = protocol.DeviceIDFromString(string(rec.Key)) - } - if err != nil { - log.Println("Replication device ID:", err) - replicationRecvsTotal.WithLabelValues("error").Inc() - continue - } - - // Store - l.db.merge(&id, rec.Addresses, rec.Seen) - replicationRecvsTotal.WithLabelValues("success").Inc() - } -} - -func deviceID(conn *tls.Conn) (protocol.DeviceID, error) { - // Handshake may not be complete on the server side yet, which we need - // to get the client certificate. - if !conn.ConnectionState().HandshakeComplete { - if err := conn.Handshake(); err != nil { - return protocol.DeviceID{}, err - } - } - - // We expect exactly one certificate. - certs := conn.ConnectionState().PeerCertificates - if len(certs) != 1 { - return protocol.DeviceID{}, fmt.Errorf("unexpected number of certificates (%d != 1)", len(certs)) - } - - return protocol.NewDeviceID(certs[0].Raw), nil -} - -func deviceIDIn(id protocol.DeviceID, ids []protocol.DeviceID) bool { - for _, candidate := range ids { - if id == candidate { - return true - } - } - return false -}