agent: start proxy manager

This commit is contained in:
Mitchell Hashimoto 2018-05-02 11:38:18 -07:00
parent 6884654c9d
commit 669268f85c
No known key found for this signature in database
GPG Key ID: 744E147AA52F5B0A
5 changed files with 50 additions and 22 deletions

View File

@ -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.
if err := a.unloadProxies(); err != nil {
a.logger.Printf("[WARN] agent: error stopping managed proxies: %s", err)
// Stop the proxy manager
// NOTE(mitchellh): we use Kill for now to kill the processes since
// 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
if proxy.ExecMode == structs.ProxyExecModeDaemon && len(proxy.Command) == 0 {
// We use the globally configured default command. If it is empty
// then we need to determine the subcommand for this agent.
cmd := a.config.ConnectProxyDefaultDaemonCommand
if len(cmd) == 0 {
var err error
cmd, err = a.defaultProxyCommand()
if err != nil {
return err
}
// Determine if we need to default the command
if proxy.ExecMode == structs.ProxyExecModeDaemon && len(proxy.Command) == 0 {
// We use the globally configured default command. If it is empty
// then we need to determine the subcommand for this agent.
cmd := a.config.ConnectProxyDefaultDaemonCommand
if len(cmd) == 0 {
var err error
cmd, err = a.defaultProxyCommand()
if err != nil {
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

View File

@ -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)

View File

@ -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
}

View File

@ -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{

View File

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