agent: start proxy manager

agent/agent.go

@@ -27,6 +27,7 @@ import (
 	"github.com/hashicorp/consul/agent/config"
 	"github.com/hashicorp/consul/agent/consul"
 	"github.com/hashicorp/consul/agent/local"
+	"github.com/hashicorp/consul/agent/proxy"
 	"github.com/hashicorp/consul/agent/structs"
 	"github.com/hashicorp/consul/agent/systemd"
 	"github.com/hashicorp/consul/agent/token"
@@ -200,6 +201,9 @@ type Agent struct {
 	// be updated at runtime, so should always be used instead of going to
 	// the configuration directly.
 	tokens *token.Store
+
+	// proxyManager is the proxy process manager for managed Connect proxies.
+	proxyManager *proxy.Manager
 }
 
 func New(c *config.RuntimeConfig) (*Agent, error) {
@@ -353,6 +357,14 @@ func (a *Agent) Start() error {
 		return err
 	}
 
+	// create the proxy process manager and start it. This is purposely
+	// done here after the local state above is loaded in so we can have
+	// a more accurate initial state view.
+	a.proxyManager = proxy.NewManager()
+	a.proxyManager.State = a.State
+	a.proxyManager.Logger = a.logger
+	go a.proxyManager.Run()
+
 	// Start watching for critical services to deregister, based on their
 	// checks.
 	go a.reapServices()
@@ -1269,9 +1281,11 @@ func (a *Agent) ShutdownAgent() error {
 		chk.Stop()
 	}
 
-	// Unload all our proxies so that we stop the running processes.
+	// Stop the proxy manager
-	if err := a.unloadProxies(); err != nil {
+	// NOTE(mitchellh): we use Kill for now to kill the processes since
-		a.logger.Printf("[WARN] agent: error stopping managed proxies: %s", err)
+	// snapshotting isn't implemented. This should change to Close later.
+	if err := a.proxyManager.Kill(); err != nil {
+		a.logger.Printf("[WARN] agent: error shutting down proxy manager: %s", err)
 	}
 
 	var err error
@@ -2038,23 +2052,21 @@ func (a *Agent) AddProxy(proxy *structs.ConnectManagedProxy, persist bool) error
 	// Lookup the target service token in state if there is one.
 	token := a.State.ServiceToken(proxy.TargetServiceID)
 
-	/*
+	// Determine if we need to default the command
-		// Determine if we need to default the command
+	if proxy.ExecMode == structs.ProxyExecModeDaemon && len(proxy.Command) == 0 {
-		if proxy.ExecMode == structs.ProxyExecModeDaemon && len(proxy.Command) == 0 {
+		// We use the globally configured default command. If it is empty
-			// We use the globally configured default command. If it is empty
+		// then we need to determine the subcommand for this agent.
-			// then we need to determine the subcommand for this agent.
+		cmd := a.config.ConnectProxyDefaultDaemonCommand
-			cmd := a.config.ConnectProxyDefaultDaemonCommand
+		if len(cmd) == 0 {
-			if len(cmd) == 0 {
+			var err error
-				var err error
+			cmd, err = a.defaultProxyCommand()
-				cmd, err = a.defaultProxyCommand()
+			if err != nil {
-				if err != nil {
+				return err
-					return err
-				}
 			}
-
-			proxy.CommandDefault = cmd
 		}
-	*/
+
+		proxy.CommandDefault = cmd
+	}
 
 	// Add the proxy to local state first since we may need to assign a port which
 	// needs to be coordinate under state lock. AddProxy will generate the

agent/agent_endpoint_test.go

@@ -70,7 +70,7 @@ func TestAgent_Services(t *testing.T) {
 		},
 		TargetServiceID: "mysql",
 	}
-	_, _, err := a.State.AddProxy(prxy1, "")
+	_, err := a.State.AddProxy(prxy1, "")
 	require.NoError(t, err)
 
 	req, _ := http.NewRequest("GET", "/v1/agent/services", nil)

agent/proxy/daemon.go

@@ -7,6 +7,7 @@ import (
 	"os/exec"
 	"reflect"
 	"sync"
+	"syscall"
 	"time"
 )
 
@@ -146,9 +147,15 @@ func (p *Daemon) keepAlive(stopCh <-chan struct{}) {
 
 		}
 
-		_, err := process.Wait()
+		ps, err := process.Wait()
 		process = nil
-		p.Logger.Printf("[INFO] agent/proxy: daemon exited: %s", err)
+		if err != nil {
+			p.Logger.Printf("[INFO] agent/proxy: daemon exited with error: %s", err)
+		} else if status, ok := ps.Sys().(syscall.WaitStatus); ok {
+			p.Logger.Printf(
+				"[INFO] agent/proxy: daemon exited with exit code: %d",
+				status.ExitStatus())
+		}
 	}
 }
 
@@ -165,7 +172,12 @@ func (p *Daemon) start() (*os.Process, error) {
 	copy(cmd.Env, p.Command.Env)
 	cmd.Env = append(cmd.Env, fmt.Sprintf("%s=%s", EnvProxyToken, p.ProxyToken))
 
+	// TODO(mitchellh): temporary until we introduce the file based logging
+	cmd.Stdout = os.Stdout
+	cmd.Stderr = os.Stderr
+
 	// Start it
+	p.Logger.Printf("[DEBUG] agent/proxy: starting proxy: %q %#v", cmd.Path, cmd.Args)
 	err := cmd.Start()
 	return cmd.Process, err
 }

agent/proxy/manager.go

@@ -192,6 +192,7 @@ func (m *Manager) Run() {
 	m.State.NotifyProxy(notifyCh)
 	defer m.State.StopNotifyProxy(notifyCh)
 
+	m.Logger.Println("[DEBUG] agent/proxy: managed Connect proxy manager started")
 	for {
 		// Sync first, before waiting on further notifications so that
 		// we can start with a known-current state.
@@ -203,6 +204,7 @@ func (m *Manager) Run() {
 
 		case <-stopCh:
 			// Stop immediately, no cleanup
+			m.Logger.Println("[DEBUG] agent/proxy: Stopping managed Connect proxy manager")
 			return
 		}
 	}
@@ -298,7 +300,7 @@ func (m *Manager) newProxy(mp *local.ManagedProxy) (Proxy, error) {
 		// Build the command to execute.
 		var cmd exec.Cmd
 		cmd.Path = command[0]
-		cmd.Args = command[1:]
+		cmd.Args = command // idx 0 is path but preserved since it should be
 
 		// Build the daemon structure
 		return &Daemon{

agent/structs/connect.go

@@ -49,6 +49,8 @@ func (m ProxyExecMode) String() string {
 		return "daemon"
 	case ProxyExecModeScript:
 		return "script"
+	case ProxyExecModeTest:
+		return "test"
 	default:
 		return "unknown"
 	}