syncthing/cmd/stdiscosrv/main.go

// 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"
	"flag"
	"log"
	"net"
	"net/http"
	"os"
	"os/signal"
	"runtime"
	"strings"
	"time"

	_ "net/http/pprof"

	"github.com/prometheus/client_golang/prometheus/promhttp"
	_ "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/thejerf/suture/v4"
)

const (
	addressExpiryTime          = 2 * time.Hour
	databaseStatisticsInterval = 5 * time.Minute

	// Reannounce-After is set to reannounceAfterSeconds +
	// random(reannounzeFuzzSeconds), similar for Retry-After
	reannounceAfterSeconds = 3300
	reannounzeFuzzSeconds  = 300
	errorRetryAfterSeconds = 1500
	errorRetryFuzzSeconds  = 300

	// Retry for not found is notFoundRetrySeenSeconds for records we have
	// seen an announcement for (but it's not active right now) and
	// notFoundRetryUnknownSeconds for records we have never seen (or not
	// seen within the last week).
	notFoundRetryUnknownMinSeconds = 60
	notFoundRetryUnknownMaxSeconds = 3600

	httpReadTimeout    = 5 * time.Second
	httpWriteTimeout   = 5 * time.Second
	httpMaxHeaderBytes = 1 << 10

	// Size of the replication outbox channel
	replicationOutboxSize = 10000
)

var debug = false

func main() {
	var listen string
	var dir string
	var metricsListen string
	var replicationListen string
	var replicationPeers string
	var certFile string
	var keyFile string
	var replCertFile string
	var replKeyFile string
	var useHTTP bool
	var compression bool
	var amqpAddress string
	var flushInterval time.Duration

	log.SetOutput(os.Stdout)
	log.SetFlags(0)

	flag.StringVar(&certFile, "cert", "./cert.pem", "Certificate file")
	flag.StringVar(&keyFile, "key", "./key.pem", "Key file")
	flag.StringVar(&dir, "db-dir", ".", "Database directory")
	flag.BoolVar(&debug, "debug", false, "Print debug output")
	flag.BoolVar(&useHTTP, "http", false, "Listen on HTTP (behind an HTTPS proxy)")
	flag.BoolVar(&compression, "compression", true, "Enable GZIP compression of responses")
	flag.StringVar(&listen, "listen", ":8443", "Listen address")
	flag.StringVar(&metricsListen, "metrics-listen", "", "Metrics listen address")
	flag.StringVar(&replicationPeers, "replicate", "", "Replication peers, id@address, comma separated")
	flag.StringVar(&replicationListen, "replication-listen", ":19200", "Replication listen address")
	flag.StringVar(&replCertFile, "replication-cert", "", "Certificate file for replication")
	flag.StringVar(&replKeyFile, "replication-key", "", "Key file for replication")
	flag.StringVar(&amqpAddress, "amqp-address", "", "Address to AMQP broker")
	flag.DurationVar(&flushInterval, "flush-interval", 5*time.Minute, "Interval between database flushes")
	showVersion := flag.Bool("version", false, "Show version")
	flag.Parse()

	log.Println(build.LongVersionFor("stdiscosrv"))
	if *showVersion {
		return
	}

	buildInfo.WithLabelValues(build.Version, runtime.Version(), build.User, build.Date.UTC().Format("2006-01-02T15:04:05Z")).Set(1)

	cert, err := tls.LoadX509KeyPair(certFile, keyFile)
	if os.IsNotExist(err) {
		log.Println("Failed to load keypair. Generating one, this might take a while...")
		cert, err = tlsutil.NewCertificate(certFile, keyFile, "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 replCertFile != "" && replKeyFile != "" {
		replCert, err = tls.LoadX509KeyPair(replCertFile, replKeyFile)
		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
	parts := strings.Split(replicationPeers, ",")
	for _, part := range parts {
		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])
			if err != nil {
				log.Fatalln("Resolving address:", 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)
		}
	}

	// Root of the service tree.
	main := suture.New("main", suture.Spec{
		PassThroughPanics: true,
	})

	// Start the database.
	db := newInMemoryStore(dir, flushInterval)
	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(replicationListen, replCert, allowedReplicationPeers, db)
		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)
	}

	// Start the main API server.
	qs := newAPISrv(listen, cert, db, repl, useHTTP, compression)
	main.Add(qs)

	// If we have a metrics port configured, start a metrics handler.
	if metricsListen != "" {
		go func() {
			mux := http.NewServeMux()
			mux.Handle("/metrics", promhttp.Handler())
			log.Fatal(http.ListenAndServe(metricsListen, mux))
		}()
	}

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// Cancel on signal
	signalChan := make(chan os.Signal, 1)
	signal.Notify(signalChan, os.Interrupt)
	go func() {
		sig := <-signalChan
		log.Printf("Received signal %s; shutting down", sig)
		cancel()
	}()

	// Engage!
	main.Serve(ctx)
}