Reload Handling for combined Reg Server (#148)

* clientconf reload for bidirectional API registrar and phantom subnet reload for processor

* Fix config decoding

The toml library complains about nil pointer when decoding.

* Add synchronization

Co-authored-by: Mingye Chen <[email protected]>

Author: jmwample

cmd/registration-server/main.go

@@ -4,8 +4,10 @@ import (
 	"flag"
 	"fmt"
 	"os"
+	"os/signal"
 	"strconv"
 	"sync"
+	"syscall"
 	"time"
 
 	"github.com/BurntSushi/toml"
@@ -36,8 +38,9 @@ type config struct {
 	ZMQPrivateKeyPath  string   `toml:"zmq_privkey_path"`
 	StationPublicKeys  []string `toml:"station_pubkeys"`
 	ClientConfPath     string   `toml:"clientconf_path"`
-	LogLevel           string   `toml:"log_level"`
-	LogMetricsInterval uint16   `toml:"log_metrics_interval"`
+	latestClientConf   *pb.ClientConf
+	LogLevel           string `toml:"log_level"`
+	LogMetricsInterval uint16 `toml:"log_metrics_interval"`
 }
 
 // parseClientConf parse the latest ClientConf based on path file
@@ -108,6 +111,27 @@ func readKeyAndEncode(path string) (string, error) {
 	return privkey, nil
 }
 
+// loadConfig is intended to re-parse portions of the config in conjunction with
+// setupReloadHandler. This is specifically for settings where we do not want to
+// restart the station. This is not intended to be a full re-build of the
+// station (i.e. auth, workers, and loglevel are not changed), Mostly this
+// should allow us to dynamically reload when there is an update to the latest
+// client configuration or the phantom subnets that we select from.
+func loadConfig(configPath string) (*config, error) {
+	conf := &config{}
+	_, err := toml.DecodeFile(configPath, conf)
+	if err != nil {
+		return nil, err
+	}
+
+	conf.latestClientConf, err = parseClientConf(conf.ClientConfPath)
+	if err != nil {
+		return nil, err
+	}
+
+	return conf, nil
+}
+
 func main() {
 	var configPath string
 
@@ -126,10 +150,9 @@ func main() {
 
 	log.SetFormatter(logFormatter)
 
-	var conf config
-	_, err := toml.DecodeFile(configPath, &conf)
+	conf, err := loadConfig(configPath)
 	if err != nil {
-		log.Fatalf("Error in reading config file: %v", err)
+		log.Fatalf("error occurred while parsing config: %v", err)
 	}
 
 	logClientIP, err := strconv.ParseBool(os.Getenv("LOG_CLIENT_IP"))
@@ -166,27 +189,50 @@ func main() {
 		log.Fatal(err)
 	}
 
-	latestClientConf, err := parseClientConf(conf.ClientConfPath)
-	if err != nil {
-		log.Fatal(err)
-	}
-
 	dnsPrivKey, err := readKey(conf.DNSPrivkeyPath)
 	if err != nil {
 		log.Fatal(err)
 	}
 
-	dnsRegServer, err := dnsregserver.NewDNSRegServer(conf.Domain, conf.DNSListenAddr, dnsPrivKey, processor, latestClientConf.GetGeneration(), log.WithField("registrar", "DNS"), metrics)
+	dnsRegServer, err := dnsregserver.NewDNSRegServer(conf.Domain, conf.DNSListenAddr, dnsPrivKey, processor, conf.latestClientConf.GetGeneration(), log.WithField("registrar", "DNS"), metrics)
 	if err != nil {
 		log.Fatal(err)
 	}
 
-	apiRegServer, err := apiregserver.NewAPIRegServer(conf.APIPort, processor, latestClientConf, log.WithField("registrar", "API"), logClientIP, metrics)
+	apiRegServer, err := apiregserver.NewAPIRegServer(conf.APIPort, processor, conf.latestClientConf, log.WithField("registrar", "API"), logClientIP, metrics)
 	if err != nil {
 		log.Fatal(err)
 	}
 
 	regServers := []regServer{dnsRegServer, apiRegServer}
 
+	signalChan := make(chan os.Signal, 1)
+
+	signal.Notify(
+		signalChan,
+		syscall.SIGHUP, // listen for SIGHUP as reload signal
+	)
+
+	// spawn a goroutine to handle os signals continuously
+	go func() {
+		for {
+			sig := <-signalChan
+
+			if sig == syscall.SIGHUP {
+				conf, err = loadConfig(configPath)
+				if err != nil {
+					log.Errorf("error occurred while reloading config -- aborting reload: %v", err)
+				} else {
+					err := processor.ReloadSubnets()
+					if err != nil {
+						log.Errorf("failed to reload phantom subnets - aborting reload: %v", err)
+					}
+					apiRegServer.NewClientConf(conf.latestClientConf)
+					dnsRegServer.UpdateLatestCCGen(conf.latestClientConf.GetGeneration())
+				}
+			}
+		}
+	}()
+
 	run(regServers)
 }

pkg/apiregserver/apiregserver.go

@@ -8,6 +8,7 @@ import (
 	"net"
 	"net/http"
 	"strings"
+	"sync"
 
 	"github.com/gorilla/mux"
 	"github.com/refraction-networking/conjure/pkg/metrics"
@@ -25,6 +26,7 @@ type registrar interface {
 type APIRegServer struct {
 	apiPort          uint16
 	latestClientConf *pb.ClientConf // Latest clientConf for sharing over RegistrationResponse channel.
+	ccMutex          sync.RWMutex
 	processor        registrar
 	logger           log.FieldLogger
 	logClientIP      bool
@@ -190,6 +192,9 @@ func (s *APIRegServer) registerBidirectional(w http.ResponseWriter, r *http.Requ
 // Use this function in registerBidirectional, if the returned ClientConfig is
 // not nil add it to the RegistrationResponse.
 func (s *APIRegServer) compareClientConfGen(genNum uint32) *pb.ClientConf {
+	s.ccMutex.RLock()
+	defer s.ccMutex.RUnlock()
+
 	// Check that server has a currnet (latest) client config
 	if s.latestClientConf == nil {
 		s.logger.Debugf("Server latest ClientConf is nil")
@@ -228,6 +233,14 @@ func parseIP(addrPort string) *net.IP {
 
 }
 
+func (s *APIRegServer) NewClientConf(c *pb.ClientConf) {
+	if c != nil {
+		s.ccMutex.Lock()
+		defer s.ccMutex.Unlock()
+		s.latestClientConf = c
+	}
+}
+
 func (s *APIRegServer) ListenAndServe() error {
 	r := mux.NewRouter()
 	r.HandleFunc("/register", s.register)
@@ -247,6 +260,7 @@ func NewAPIRegServer(apiPort uint16, regprocessor *regprocessor.RegProcessor, la
 		apiPort:          apiPort,
 		processor:        regprocessor,
 		latestClientConf: latestCC,
+		ccMutex:          sync.RWMutex{},
 		logger:           logger,
 		logClientIP:      logClientIP,
 		metrics:          metrics,

pkg/dnsregserver/dnsregserver.go

@@ -3,6 +3,7 @@ package dnsregserver
 import (
 	"encoding/hex"
 	"errors"
+	"sync/atomic"
 
 	"github.com/refraction-networking/conjure/pkg/metrics"
 	"github.com/refraction-networking/conjure/pkg/regprocessor"
@@ -70,7 +71,7 @@ func (s *DNSRegServer) processRequest(reqIn []byte) ([]byte, error) {
 	reqLogger.Tracef("Request received: [%+v]", c2sPayload)
 
 	clientconfOutdated := false
-	if c2sPayload.RegistrationPayload.GetDecoyListGeneration() < s.latestCCGen {
+	if c2sPayload.RegistrationPayload.GetDecoyListGeneration() < atomic.LoadUint32(&s.latestCCGen) {
 		clientconfOutdated = true
 	}
 
@@ -117,3 +118,8 @@ func (s *DNSRegServer) processRequest(reqIn []byte) ([]byte, error) {
 func (f *DNSRegServer) Close() error {
 	return f.dnsResponder.Close()
 }
+
+// Close closes the underlying dns responder.
+func (f *DNSRegServer) UpdateLatestCCGen(gen uint32) {
+	atomic.StoreUint32(&f.latestCCGen, gen)
+}

pkg/regprocessor/regprocessor.go

@@ -44,10 +44,11 @@ type ipSelector interface {
 
 // RegProcessor provides an interface to publish registrations and helper functions to process registration requests
 type RegProcessor struct {
-	zmqMutex   sync.Mutex
-	ipSelector ipSelector
-	sock       zmqSender
-	metrics    *metrics.Metrics
+	zmqMutex      sync.Mutex
+	selectorMutex sync.RWMutex
+	ipSelector    ipSelector
+	sock          zmqSender
+	metrics       *metrics.Metrics
 }
 
 // NewRegProcessor initialize a new RegProcessor
@@ -76,10 +77,11 @@ func NewRegProcessor(zmqBindAddr string, zmqPort uint16, privkey string, authVer
 	}
 
 	return &RegProcessor{
-		zmqMutex:   sync.Mutex{},
-		ipSelector: phantomSelector,
-		sock:       sock,
-		metrics:    metrics,
+		zmqMutex:      sync.Mutex{},
+		selectorMutex: sync.RWMutex{},
+		ipSelector:    phantomSelector,
+		sock:          sock,
+		metrics:       metrics,
 	}, nil
 }
 
@@ -101,10 +103,11 @@ func NewRegProcessorNoAuth(zmqBindAddr string, zmqPort uint16, metrics *metrics.
 	}
 
 	return &RegProcessor{
-		zmqMutex:   sync.Mutex{},
-		ipSelector: phantomSelector,
-		sock:       sock,
-		metrics:    metrics,
+		zmqMutex:      sync.Mutex{},
+		selectorMutex: sync.RWMutex{},
+		ipSelector:    phantomSelector,
+		sock:          sock,
+		metrics:       metrics,
 	}, nil
 }
 
@@ -173,6 +176,8 @@ func (p *RegProcessor) processBdReq(c2sPayload *pb.C2SWrapper) (*pb.Registration
 	}
 
 	if *c2sPayload.RegistrationPayload.V4Support {
+		p.selectorMutex.RLock()
+		defer p.selectorMutex.RUnlock()
 		phantom4, err := p.ipSelector.Select(
 			cjkeys.DarkDecoySeed,
 			uint(c2sPayload.GetRegistrationPayload().GetDecoyListGeneration()), //generation type uint
@@ -190,6 +195,8 @@ func (p *RegProcessor) processBdReq(c2sPayload *pb.C2SWrapper) (*pb.Registration
 	}
 
 	if *c2sPayload.RegistrationPayload.V6Support {
+		p.selectorMutex.RLock()
+		defer p.selectorMutex.RUnlock()
 		phantom6, err := p.ipSelector.Select(
 			cjkeys.DarkDecoySeed,
 			uint(c2sPayload.GetRegistrationPayload().GetDecoyListGeneration()),
@@ -254,3 +261,16 @@ func (p *RegProcessor) processC2SWrapper(c2sPayload *pb.C2SWrapper, clientAddr [
 
 	return proto.Marshal(payload)
 }
+
+func (p *RegProcessor) ReloadSubnets() error {
+	phantomSelector, err := lib.GetPhantomSubnetSelector()
+	if err != nil {
+		return err
+	}
+
+	p.selectorMutex.Lock()
+	defer p.selectorMutex.Unlock()
+	p.ipSelector = phantomSelector
+
+	return nil
+}

sysconfig/conjure-app.service

@@ -18,6 +18,9 @@ EnvironmentFile=/opt/conjure/sysconfig/conjure.conf
 ExecStartPre=/bin/sleep 10
 ExecStart=/opt/conjure/application/application
 
+# send SIGHUP to the station process
+ExecReload=/bin/kill -HUP $MAINPID
+
 # on stop processes will get SIGTERM, and after 10 secs - SIGKILL (default 90)
 TimeoutStopSec=10
 

sysconfig/conjure-registration-server.service

@@ -11,6 +11,9 @@ EnvironmentFile=/opt/conjure/sysconfig/conjure.conf
 
 ExecStart=/opt/conjure/cmd/registration-server/registration-server --config /opt/conjure/cmd/registration-server/config.toml
 
+# send SIGHUP to the registration server process
+ExecReload=/bin/kill -HUP $MAINPID
+
 # on stop processes will get SIGTERM, and after 10 secs - SIGKILL (default 90)
 TimeoutStopSec=10