AUTH-2587 add config watcher and reload logic for access client forwarder

cmd/cloudflared/access/carrier.go

@@ -12,6 +12,32 @@ import (
 	cli "gopkg.in/urfave/cli.v2"
 )
 
+// StartForwarder starts a client side websocket forward
+func StartForwarder(forwarder config.Forwarder, shutdown <-chan struct{}) error {
+	validURLString, err := validation.ValidateUrl(forwarder.Listener)
+	if err != nil {
+		logger.WithError(err).Error("Error validating origin URL")
+		return errors.Wrap(err, "error validating origin URL")
+	}
+
+	validURL, err := url.Parse(validURLString)
+	if err != nil {
+		logger.WithError(err).Error("Error parsing origin URL")
+		return errors.Wrap(err, "error parsing origin URL")
+	}
+
+	options := &carrier.StartOptions{
+		OriginURL: forwarder.URL,
+		Headers:   make(http.Header), //TODO: TUN-2688 support custom headers from config file
+	}
+
+	// we could add a cmd line variable for this bool if we want the SOCK5 server to be on the client side
+	wsConn := carrier.NewWSConnection(logger, false)
+
+	logger.Infof("Start Websocket listener on: %s", validURL.Host)
+	return carrier.StartForwarder(wsConn, validURL.Host, shutdown, options)
+}
+
 // ssh will start a WS proxy server for server mode
 // or copy from stdin/stdout for client mode
 // useful for proxying other protocols (like ssh) over websockets

cmd/cloudflared/app_service.go

@@ -0,0 +1,112 @@
+package main
+
+import (
+	"github.com/cloudflare/cloudflared/cmd/cloudflared/access"
+	"github.com/cloudflare/cloudflared/cmd/cloudflared/config"
+	"github.com/sirupsen/logrus"
+)
+
+type forwarderState struct {
+	forwarder config.Forwarder
+	shutdown  chan struct{}
+}
+
+func (s *forwarderState) Shutdown() {
+	s.shutdown <- struct{}{}
+}
+
+// AppService is the main service that runs when no command lines flags are passed to cloudflared
+// it manages all the running services such as tunnels, forwarders, DNS resolver, etc
+type AppService struct {
+	configManager    config.Manager
+	shutdownC        chan struct{}
+	forwarders       map[string]forwarderState
+	configUpdateChan chan config.Root
+	logger           *logrus.Logger
+}
+
+// NewAppService creates a new AppService with needed supporting services
+func NewAppService(configManager config.Manager, shutdownC chan struct{}, logger *logrus.Logger) *AppService {
+	return &AppService{
+		configManager:    configManager,
+		shutdownC:        shutdownC,
+		forwarders:       make(map[string]forwarderState),
+		configUpdateChan: make(chan config.Root),
+		logger:           logger,
+	}
+}
+
+// Run starts the run loop to handle config updates and run forwarders, tunnels, etc
+func (s *AppService) Run() error {
+	go s.actionLoop()
+	return s.configManager.Start(s)
+}
+
+// Shutdown kills all the running services
+func (s *AppService) Shutdown() error {
+	s.configManager.Shutdown()
+	return nil
+}
+
+// ConfigDidUpdate is a delegate notification from the config manager
+// it is trigger when the config file has been updated and now the service needs
+// to update its services accordingly
+func (s *AppService) ConfigDidUpdate(c config.Root) {
+	s.configUpdateChan <- c
+}
+
+// actionLoop handles the actions from running processes
+func (s *AppService) actionLoop() {
+	for {
+		select {
+		case c := <-s.configUpdateChan:
+			s.handleConfigUpdate(c)
+		case <-s.shutdownC:
+			for _, state := range s.forwarders {
+				state.Shutdown()
+			}
+			return
+		}
+	}
+}
+
+func (s *AppService) handleConfigUpdate(c config.Root) {
+	// handle the client forward listeners
+	activeListeners := map[string]struct{}{}
+	for _, f := range c.Forwarders {
+		s.handleForwarderUpdate(f)
+		activeListeners[f.Listener] = struct{}{}
+	}
+
+	// remove any listeners that are no longer active
+	for key, state := range s.forwarders {
+		if _, ok := activeListeners[key]; !ok {
+			state.Shutdown()
+			delete(s.forwarders, key)
+		}
+	}
+
+	// TODO: AUTH-2588, TUN-1451 - tunnels and dns proxy
+}
+
+// handle managing a forwarder service
+func (s *AppService) handleForwarderUpdate(f config.Forwarder) {
+	// check if we need to start a new listener or stop an old one
+	if state, ok := s.forwarders[f.Listener]; ok {
+		if state.forwarder.Hash() == f.Hash() {
+			return // the exact same listener, no changes, so move along
+		}
+		state.Shutdown() //shutdown the listener since a new one is starting
+	}
+	// add a new forwarder to the list
+	state := forwarderState{forwarder: f, shutdown: make(chan struct{}, 1)}
+	s.forwarders[f.Listener] = state
+
+	// start the forwarder
+	go func(f forwarderState) {
+		err := access.StartForwarder(f.forwarder, f.shutdown)
+		if err != nil {
+			s.logger.WithError(err).Errorf("Forwarder at address: %s", f.forwarder)
+		}
+	}(state)
+}

cmd/cloudflared/config/manager.go

@@ -0,0 +1,101 @@
+package config
+
+import (
+	"errors"
+	"os"
+
+	"github.com/cloudflare/cloudflared/watcher"
+
+	"github.com/sirupsen/logrus"
+	"gopkg.in/yaml.v2"
+)
+
+// Notifier sends out config updates
+type Notifier interface {
+	ConfigDidUpdate(Root)
+}
+
+// Manager is the base functions of the config manager
+type Manager interface {
+	Start(Notifier) error
+	Shutdown()
+}
+
+// FileManager watches the yaml config for changes
+// sends updates to the service to reconfigure to match the updated config
+type FileManager struct {
+	watcher    watcher.Notifier
+	notifier   Notifier
+	configPath string
+	logger     *logrus.Logger
+}
+
+// NewFileManager creates a config manager
+func NewFileManager(watcher watcher.Notifier, configPath string, logger *logrus.Logger) (Manager, error) {
+	m := &FileManager{
+		watcher:    watcher,
+		configPath: configPath,
+		logger:     logger,
+	}
+	err := watcher.Add(configPath)
+	return m, err
+}
+
+// Start starts the runloop to watch for config changes
+func (m *FileManager) Start(notifier Notifier) error {
+	m.notifier = notifier
+
+	// update the notifier with a fresh config on start
+	config, err := m.GetConfig()
+	if err != nil {
+		return err
+	}
+	notifier.ConfigDidUpdate(config)
+
+	m.watcher.Start(m)
+	return nil
+}
+
+// GetConfig reads the yaml file from the disk
+func (m *FileManager) GetConfig() (Root, error) {
+	if m.configPath == "" {
+		return Root{}, errors.New("unable to find config file")
+	}
+
+	file, err := os.Open(m.configPath)
+	if err != nil {
+		return Root{}, err
+	}
+	defer file.Close()
+
+	var config Root
+	if err := yaml.NewDecoder(file).Decode(&config); err != nil {
+		return Root{}, err
+	}
+
+	return config, nil
+}
+
+// Shutdown stops the watcher
+func (m *FileManager) Shutdown() {
+	m.watcher.Shutdown()
+}
+
+// File change notifications from the watcher
+
+// WatcherItemDidChange triggers when the yaml config is updated
+// sends the updated config to the service to reload its state
+func (m *FileManager) WatcherItemDidChange(filepath string) {
+	config, err := m.GetConfig()
+	if err != nil {
+		m.logger.WithError(err).Error("Failed to read new config")
+		return
+	}
+	m.logger.Info("Config file has been updated")
+	m.notifier.ConfigDidUpdate(config)
+}
+
+// WatcherDidError notifies of errors with the file watcher
+func (m *FileManager) WatcherDidError(err error) {
+	m.logger.WithError(err).Error("Config watcher encountered an error")
+}

cmd/cloudflared/config/manager_test.go

@@ -0,0 +1,80 @@
+package config
+
+import (
+	"bufio"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/cloudflare/cloudflared/log"
+	"github.com/cloudflare/cloudflared/watcher"
+	"github.com/stretchr/testify/assert"
+	"gopkg.in/yaml.v2"
+)
+
+type mockNotifier struct {
+	configs []Root
+}
+
+func (n *mockNotifier) ConfigDidUpdate(c Root) {
+	n.configs = append(n.configs, c)
+}
+
+func writeConfig(t *testing.T, f *os.File, c *Root) {
+	f.Sync()
+	b, err := yaml.Marshal(c)
+	assert.NoError(t, err)
+
+	w := bufio.NewWriter(f)
+	_, err = w.Write(b)
+	assert.NoError(t, err)
+
+	err = w.Flush()
+	assert.NoError(t, err)
+}
+
+func TestConfigChanged(t *testing.T) {
+	filePath := "config.yaml"
+	f, err := os.Create(filePath)
+	assert.NoError(t, err)
+	defer func() {
+		f.Close()
+		os.Remove(filePath)
+	}()
+	c := &Root{
+		OrgKey:          "abcd",
+		ConfigType:      "mytype",
+		CheckinInterval: 1,
+		Forwarders: []Forwarder{
+			{
+				URL:      "test.daltoniam.com",
+				Listener: "127.0.0.1:8080",
+			},
+		},
+	}
+	writeConfig(t, f, c)
+
+	w, err := watcher.NewFile()
+	assert.NoError(t, err)
+	logger := log.CreateLogger()
+	service, err := NewFileManager(w, filePath, logger)
+	assert.NoError(t, err)
+
+	n := &mockNotifier{}
+	go service.Start(n)
+
+	c.Forwarders = append(c.Forwarders, Forwarder{URL: "add.daltoniam.com", Listener: "127.0.0.1:8081"})
+	writeConfig(t, f, c)
+
+	// give it time to trigger
+	time.Sleep(10 * time.Millisecond)
+	service.Shutdown()
+
+	assert.Len(t, n.configs, 2, "did not get 2 config updates as expected")
+	assert.Len(t, n.configs[0].Forwarders, 1, "not the amount of forwarders expected")
+	assert.Len(t, n.configs[1].Forwarders, 2, "not the amount of forwarders expected")
+
+	assert.Equal(t, n.configs[0].Forwarders[0].Hash(), c.Forwarders[0].Hash(), "forwarder hashes don't match")
+	assert.Equal(t, n.configs[1].Forwarders[0].Hash(), c.Forwarders[0].Hash(), "forwarder hashes don't match")
+	assert.Equal(t, n.configs[1].Forwarders[1].Hash(), c.Forwarders[1].Hash(), "forwarder hashes don't match")
+}

cmd/cloudflared/config/model.go

@@ -0,0 +1,37 @@
+package config
+
+import (
+	"crypto/md5"
+	"fmt"
+	"io"
+)
+
+// Forwarder represents a client side listener to forward traffic to the edge
+type Forwarder struct {
+	URL      string `json:"url"`
+	Listener string `json:"listener"`
+}
+
+// Tunnel represents a tunnel that should be started
+type Tunnel struct {
+	URL          string `json:"url"`
+	Origin       string `json:"origin"`
+	ProtocolType string `json:"type"`
+}
+
+// Root is the base options to configure the service
+type Root struct {
+	OrgKey          string      `json:"org_key"`
+	ConfigType      string      `json:"type"`
+	CheckinInterval int         `json:"checkin_interval"`
+	Forwarders      []Forwarder `json:"forwarders,omitempty"`
+	Tunnels         []Tunnel    `json:"tunnels,omitempty"`
+}
+
+// Hash returns the computed values to see if the forwarder values change
+func (f *Forwarder) Hash() string {
+	h := md5.New()
+	io.WriteString(h, f.URL)
+	io.WriteString(h, f.Listener)
+	return fmt.Sprintf("%x", h.Sum(nil))
+}

cmd/cloudflared/main.go

@@ -6,10 +6,12 @@ import (
 	"time"
 
 	"github.com/cloudflare/cloudflared/cmd/cloudflared/access"
+	"github.com/cloudflare/cloudflared/cmd/cloudflared/config"
 	"github.com/cloudflare/cloudflared/cmd/cloudflared/tunnel"
 	"github.com/cloudflare/cloudflared/cmd/cloudflared/updater"
 	"github.com/cloudflare/cloudflared/log"
 	"github.com/cloudflare/cloudflared/metrics"
+	"github.com/cloudflare/cloudflared/watcher"
 
 	raven "github.com/getsentry/raven-go"
 	homedir "github.com/mitchellh/go-homedir"
@@ -121,7 +123,7 @@ func isEmptyInvocation(c *cli.Context) bool {
 func action(version string, shutdownC, graceShutdownC chan struct{}) cli.ActionFunc {
 	return func(c *cli.Context) (err error) {
 		if isEmptyInvocation(c) {
-			cli.ShowAppHelpAndExit(c, 1)
+			return handleServiceMode(shutdownC)
 		}
 		tags := make(map[string]string)
 		tags["hostname"] = c.String("hostname")
@@ -161,3 +163,27 @@ func handleError(err error) {
 	}
 	raven.CaptureError(err, nil)
 }
+
+// cloudflared was started without any flags
+func handleServiceMode(shutdownC chan struct{}) error {
+	// start the main run loop that reads from the config file
+	f, err := watcher.NewFile()
+	if err != nil {
+		logger.WithError(err).Error("Cannot load config file")
+		return err
+	}
+
+	configPath := config.FindDefaultConfigPath()
+	configManager, err := config.NewFileManager(f, configPath, logger)
+	if err != nil {
+		logger.WithError(err).Error("Cannot setup config file for monitoring")
+		return err
+	}
+
+	appService := NewAppService(configManager, shutdownC, logger)
+	if err := appService.Run(); err != nil {
+		logger.WithError(err).Error("Failed to start app service")
+		return err
+	}
+	return nil
+}