cmd/stdiscosrv: Add AMQP replication

cmd/stdiscosrv/amqp.go

@@ -0,0 +1,246 @@
+// Copyright (C) 2024 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"
+	"fmt"
+	"io"
+
+	amqp "github.com/rabbitmq/amqp091-go"
+	"github.com/thejerf/suture/v4"
+)
+
+type amqpReplicator struct {
+	suture.Service
+	broker   string
+	sender   *amqpSender
+	receiver *amqpReceiver
+	outbox   chan ReplicationRecord
+}
+
+func newAMQPReplicator(broker, clientID string, db database) *amqpReplicator {
+	svc := suture.New("amqpReplicator", suture.Spec{PassThroughPanics: true})
+
+	sender := &amqpSender{
+		broker:   broker,
+		clientID: clientID,
+		outbox:   make(chan ReplicationRecord, replicationOutboxSize),
+	}
+	svc.Add(sender)
+
+	receiver := &amqpReceiver{
+		broker:   broker,
+		clientID: clientID,
+		db:       db,
+	}
+	svc.Add(receiver)
+
+	return &amqpReplicator{
+		Service:  svc,
+		broker:   broker,
+		sender:   sender,
+		receiver: receiver,
+		outbox:   make(chan ReplicationRecord, replicationOutboxSize),
+	}
+}
+
+func (s *amqpReplicator) send(key string, ps []DatabaseAddress, seen int64) {
+	s.sender.send(key, ps, seen)
+}
+
+type amqpSender struct {
+	broker   string
+	clientID string
+	outbox   chan ReplicationRecord
+}
+
+func (s *amqpSender) Serve(ctx context.Context) error {
+	conn, ch, err := amqpChannel(s.broker)
+	if err != nil {
+		return err
+	}
+	defer ch.Close()
+	defer conn.Close()
+
+	buf := make([]byte, 1024)
+	for {
+		select {
+		case rec := <-s.outbox:
+			size := rec.Size()
+			if len(buf) < size {
+				buf = make([]byte, size)
+			}
+
+			n, err := rec.MarshalTo(buf)
+			if err != nil {
+				replicationSendsTotal.WithLabelValues("error").Inc()
+				return fmt.Errorf("replication marshal: %w", err)
+			}
+
+			err = ch.PublishWithContext(ctx,
+				"discovery", // exchange
+				"",          // routing key
+				false,       // mandatory
+				false,       // immediate
+				amqp.Publishing{
+					ContentType: "application/protobuf",
+					Body:        buf[:n],
+					AppId:       s.clientID,
+				})
+			if err != nil {
+				replicationSendsTotal.WithLabelValues("error").Inc()
+				return fmt.Errorf("replication publish: %w", err)
+			}
+
+			replicationSendsTotal.WithLabelValues("success").Inc()
+
+		case <-ctx.Done():
+			return nil
+		}
+	}
+}
+
+func (s *amqpSender) String() string {
+	return fmt.Sprintf("amqpSender(%q)", s.broker)
+}
+
+func (s *amqpSender) send(key string, 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()
+	}
+}
+
+type amqpReceiver struct {
+	broker   string
+	clientID string
+	db       database
+}
+
+func (s *amqpReceiver) Serve(ctx context.Context) error {
+	conn, ch, err := amqpChannel(s.broker)
+	if err != nil {
+		return err
+	}
+	defer ch.Close()
+	defer conn.Close()
+
+	msgs, err := amqpConsume(ch)
+	if err != nil {
+		return err
+	}
+
+	for {
+		select {
+		case msg, ok := <-msgs:
+			if !ok {
+				return fmt.Errorf("subscription closed: %w", io.EOF)
+			}
+
+			// ignore messages from ourself
+			if msg.AppId == s.clientID {
+				continue
+			}
+
+			var rec ReplicationRecord
+			if err := rec.Unmarshal(msg.Body); err != nil {
+				replicationRecvsTotal.WithLabelValues("error").Inc()
+				return fmt.Errorf("replication unmarshal: %w", err)
+			}
+
+			if err := s.db.merge(rec.Key, rec.Addresses, rec.Seen); err != nil {
+				return fmt.Errorf("replication database merge: %w", err)
+			}
+
+			replicationRecvsTotal.WithLabelValues("success").Inc()
+
+		case <-ctx.Done():
+			return nil
+		}
+	}
+}
+
+func (s *amqpReceiver) String() string {
+	return fmt.Sprintf("amqpReceiver(%q)", s.broker)
+}
+
+func amqpChannel(dst string) (*amqp.Connection, *amqp.Channel, error) {
+	conn, err := amqp.Dial(dst)
+	if err != nil {
+		return nil, nil, fmt.Errorf("AMQP dial: %w", err)
+	}
+
+	ch, err := conn.Channel()
+	if err != nil {
+		return nil, nil, fmt.Errorf("AMQP channel: %w", err)
+	}
+
+	err = ch.ExchangeDeclare(
+		"discovery", // name
+		"fanout",    // type
+		false,       // durable
+		false,       // auto-deleted
+		false,       // internal
+		false,       // no-wait
+		nil,         // arguments
+	)
+	if err != nil {
+		return nil, nil, fmt.Errorf("AMQP declare exchange: %w", err)
+	}
+
+	return conn, ch, nil
+}
+
+func amqpConsume(ch *amqp.Channel) (<-chan amqp.Delivery, error) {
+	q, err := ch.QueueDeclare(
+		"",    // name
+		false, // durable
+		false, // delete when unused
+		true,  // exclusive
+		false, // no-wait
+		nil,   // arguments
+	)
+	if err != nil {
+		return nil, fmt.Errorf("AMQP declare queue: %w", err)
+	}
+
+	err = ch.QueueBind(
+		q.Name,      // queue name
+		"",          // routing key
+		"discovery", // exchange
+		false,
+		nil,
+	)
+	if err != nil {
+		return nil, fmt.Errorf("AMQP bind queue: %w", err)
+	}
+
+	msgs, err := ch.Consume(
+		q.Name, // queue
+		"",     // consumer
+		true,   // auto-ack
+		false,  // exclusive
+		false,  // no-local
+		false,  // no-wait
+		nil,    // args
+	)
+	if err != nil {
+		return nil, fmt.Errorf("AMQP consume: %w", err)
+	}
+
+	return msgs, nil
+}

cmd/stdiscosrv/apisrv.go

@@ -39,12 +39,13 @@ type announcement struct {
 }
 
 type apiSrv struct {
-	addr     string
-	cert     tls.Certificate
-	db       database
-	listener net.Listener
-	repl     replicator // optional
-	useHTTP  bool
+	addr           string
+	cert           tls.Certificate
+	db             database
+	listener       net.Listener
+	repl           replicator // optional
+	useHTTP        bool
+	missesIncrease int
 
 	mapsMut sync.Mutex
 	misses  map[string]int32
@@ -60,14 +61,15 @@ type contextKey int
 
 const idKey contextKey = iota
 
-func newAPISrv(addr string, cert tls.Certificate, db database, repl replicator, useHTTP bool) *apiSrv {
+func newAPISrv(addr string, cert tls.Certificate, db database, repl replicator, useHTTP bool, missesIncrease int) *apiSrv {
 	return &apiSrv{
-		addr:    addr,
-		cert:    cert,
-		db:      db,
-		repl:    repl,
-		useHTTP: useHTTP,
-		misses:  make(map[string]int32),
+		addr:           addr,
+		cert:           cert,
+		db:             db,
+		repl:           repl,
+		useHTTP:        useHTTP,
+		misses:         make(map[string]int32),
+		missesIncrease: missesIncrease,
 	}
 }
 
@@ -197,14 +199,13 @@ func (s *apiSrv) handleGET(w http.ResponseWriter, req *http.Request) {
 		s.mapsMut.Lock()
 		misses := s.misses[key]
 		if misses < rec.Misses {
-			misses = rec.Misses + 1
-		} else {
-			misses++
+			misses = rec.Misses
 		}
+		misses += int32(s.missesIncrease)
 		s.misses[key] = misses
 		s.mapsMut.Unlock()
 
-		if misses%notFoundMissesWriteInterval == 0 {
+		if misses >= notFoundMissesWriteInterval {
 			rec.Misses = misses
 			rec.Missed = time.Now().UnixNano()
 			rec.Addresses = nil
@@ -444,7 +445,6 @@ func fixupAddresses(remote *net.TCPAddr, addresses []string) []string {
 				// remote is nil, unable to determine host IP
 				continue
 			}
-
 		}
 
 		// If zero port was specified, use remote port.

cmd/stdiscosrv/main.go

@@ -22,6 +22,7 @@ import (
 	_ "github.com/syncthing/syncthing/lib/automaxprocs"
 	"github.com/syncthing/syncthing/lib/build"
 	"github.com/syncthing/syncthing/lib/protocol"
+	"github.com/syncthing/syncthing/lib/rand"
 	"github.com/syncthing/syncthing/lib/tlsutil"
 	"github.com/syndtr/goleveldb/leveldb/opt"
 	"github.com/thejerf/suture/v4"
@@ -80,6 +81,8 @@ func main() {
 	var replKeyFile string
 	var useHTTP bool
 	var largeDB bool
+	var amqpAddress string
+	missesIncrease := 1
 
 	log.SetOutput(os.Stdout)
 	log.SetFlags(0)
@@ -96,6 +99,8 @@ func main() {
 	flag.StringVar(&replCertFile, "replication-cert", "", "Certificate file for replication")
 	flag.StringVar(&replKeyFile, "replication-key", "", "Key file for replication")
 	flag.BoolVar(&largeDB, "large-db", false, "Use larger database settings")
+	flag.StringVar(&amqpAddress, "amqp-address", "", "Address to AMQP broker")
+	flag.IntVar(&missesIncrease, "misses-increase", 1, "How many times to increase the misses counter on each miss")
 	showVersion := flag.Bool("version", false, "Show version")
 	flag.Parse()
 
@@ -203,8 +208,24 @@ func main() {
 		main.Add(rl)
 	}
 
+	// If we have an AMQP broker, start that
+	if amqpAddress != "" {
+		clientID := rand.String(10)
+		kr := newAMQPReplicator(amqpAddress, clientID, db)
+		repl = append(repl, kr)
+		main.Add(kr)
+	}
+
+	go func() {
+		for range time.NewTicker(time.Second).C {
+			for _, r := range repl {
+				r.send("<heartbeat>", nil, time.Now().UnixNano())
+			}
+		}
+	}()
+
 	// Start the main API server.
-	qs := newAPISrv(listen, cert, db, repl, useHTTP)
+	qs := newAPISrv(listen, cert, db, repl, useHTTP, missesIncrease)
 	main.Add(qs)
 
 	// If we have a metrics port configured, start a metrics handler.

cmd/stdiscosrv/replication.go

@@ -144,10 +144,11 @@ func (s *replicationSender) String() string {
 	return fmt.Sprintf("replicationSender(%q)", s.dst)
 }
 
-func (s *replicationSender) send(key string, ps []DatabaseAddress, _ int64) {
+func (s *replicationSender) send(key string, ps []DatabaseAddress, seen int64) {
 	item := ReplicationRecord{
 		Key:       key,
 		Addresses: ps,
+		Seen:      seen,
 	}
 
 	// The send should never block. The inbox is suitably buffered for at

cmd/stdiscosrv/stats.go

@@ -7,10 +7,7 @@
 package main
 
 import (
-	"os"
-
 	"github.com/prometheus/client_golang/prometheus"
-	"github.com/prometheus/client_golang/prometheus/collectors"
 )
 
 var (
@@ -127,15 +124,4 @@ func init() {
 		databaseKeys, databaseStatisticsSeconds,
 		databaseOperations, databaseOperationSeconds,
 		retryAfterHistogram)
-
-	processCollectorOpts := collectors.ProcessCollectorOpts{
-		Namespace: "syncthing_discovery",
-		PidFn: func() (int, error) {
-			return os.Getpid(), nil
-		},
-	}
-
-	prometheus.MustRegister(
-		collectors.NewProcessCollector(processCollectorOpts),
-	)
 }