open-consul/command/agent/command.go

303 lines
8.2 KiB
Go
Raw Normal View History

2013-12-19 20:18:06 +00:00
package agent
import (
2013-12-21 00:39:32 +00:00
"flag"
2013-12-20 01:14:46 +00:00
"fmt"
2013-12-21 00:39:32 +00:00
"github.com/hashicorp/logutils"
2013-12-19 20:18:06 +00:00
"github.com/mitchellh/cli"
2013-12-21 00:39:32 +00:00
"io"
2013-12-30 23:27:41 +00:00
"net"
2013-12-21 00:39:32 +00:00
"os"
"os/signal"
2013-12-19 20:18:06 +00:00
"strings"
2013-12-21 00:39:32 +00:00
"syscall"
"time"
2013-12-19 20:18:06 +00:00
)
2013-12-21 00:39:32 +00:00
// gracefulTimeout controls how long we wait before forcefully terminating
var gracefulTimeout = 5 * time.Second
// Command is a Command implementation that runs a Consul agent.
2013-12-19 20:18:06 +00:00
// The command will not end unless a shutdown message is sent on the
// ShutdownCh. If two messages are sent on the ShutdownCh it will forcibly
// exit.
type Command struct {
Ui cli.Ui
ShutdownCh <-chan struct{}
2013-12-21 00:39:32 +00:00
args []string
logFilter *logutils.LevelFilter
2013-12-23 19:38:51 +00:00
agent *Agent
2013-12-30 23:27:41 +00:00
rpcServer *AgentRPC
2013-12-23 19:38:51 +00:00
httpServer *HTTPServer
dnsServer *DNSServer
2013-12-21 00:39:32 +00:00
}
// readConfig is responsible for setup of our configuration using
// the command line and any file configs
func (c *Command) readConfig() *Config {
var cmdConfig Config
var configFiles []string
cmdFlags := flag.NewFlagSet("agent", flag.ContinueOnError)
cmdFlags.Usage = func() { c.Ui.Output(c.Help()) }
cmdFlags.StringVar(&cmdConfig.SerfBindAddr, "serf-bind", "", "address to bind serf listeners to")
cmdFlags.StringVar(&cmdConfig.ServerAddr, "server-addr", "", "address to bind server listeners to")
cmdFlags.Var((*AppendSliceValue)(&configFiles), "config-file",
"json file to read config from")
cmdFlags.Var((*AppendSliceValue)(&configFiles), "config-dir",
"directory of json files to read")
cmdFlags.StringVar(&cmdConfig.EncryptKey, "encrypt", "", "encryption key")
cmdFlags.StringVar(&cmdConfig.LogLevel, "log-level", "", "log level")
cmdFlags.StringVar(&cmdConfig.NodeName, "node", "", "node name")
cmdFlags.StringVar(&cmdConfig.RPCAddr, "rpc-addr", "",
"address to bind RPC listener to")
cmdFlags.StringVar(&cmdConfig.DataDir, "data", "", "path to the data directory")
cmdFlags.StringVar(&cmdConfig.Datacenter, "dc", "", "node datacenter")
2014-01-03 23:43:35 +00:00
cmdFlags.StringVar(&cmdConfig.DNSRecursor, "recursor", "", "address of dns recursor")
cmdFlags.BoolVar(&cmdConfig.Server, "server", false, "run agent as server")
cmdFlags.BoolVar(&cmdConfig.Bootstrap, "bootstrap", false, "enable server bootstrap mode")
2013-12-21 00:39:32 +00:00
if err := cmdFlags.Parse(c.args); err != nil {
return nil
}
config := DefaultConfig()
if len(configFiles) > 0 {
fileConfig, err := ReadConfigPaths(configFiles)
if err != nil {
c.Ui.Error(err.Error())
return nil
}
config = MergeConfig(config, fileConfig)
}
config = MergeConfig(config, &cmdConfig)
if config.NodeName == "" {
hostname, err := os.Hostname()
if err != nil {
c.Ui.Error(fmt.Sprintf("Error determining hostname: %s", err))
return nil
}
config.NodeName = hostname
}
if config.EncryptKey != "" {
if _, err := config.EncryptBytes(); err != nil {
c.Ui.Error(fmt.Sprintf("Invalid encryption key: %s", err))
return nil
}
}
return config
}
// setupLoggers is used to setup the logGate, logWriter, and our logOutput
func (c *Command) setupLoggers(config *Config) (*GatedWriter, *logWriter, io.Writer) {
// Setup logging. First create the gated log writer, which will
// store logs until we're ready to show them. Then create the level
// filter, filtering logs of the specified level.
logGate := &GatedWriter{
Writer: &cli.UiWriter{Ui: c.Ui},
}
c.logFilter = LevelFilter()
c.logFilter.MinLevel = logutils.LogLevel(strings.ToUpper(config.LogLevel))
c.logFilter.Writer = logGate
if !ValidateLevelFilter(c.logFilter.MinLevel, c.logFilter) {
c.Ui.Error(fmt.Sprintf(
"Invalid log level: %s. Valid log levels are: %v",
c.logFilter.MinLevel, c.logFilter.Levels))
return nil, nil, nil
}
// Create a log writer, and wrap a logOutput around it
logWriter := NewLogWriter(512)
logOutput := io.MultiWriter(c.logFilter, logWriter)
return logGate, logWriter, logOutput
2013-12-19 20:18:06 +00:00
}
2013-12-23 19:38:51 +00:00
// setupAgent is used to start the agent and various interfaces
2013-12-30 23:27:41 +00:00
func (c *Command) setupAgent(config *Config, logOutput io.Writer, logWriter *logWriter) error {
2013-12-23 19:38:51 +00:00
c.Ui.Output("Starting Consul agent...")
agent, err := Create(config, logOutput)
if err != nil {
c.Ui.Error(fmt.Sprintf("Error starting agent: %s", err))
return err
}
c.agent = agent
2013-12-30 23:27:41 +00:00
// Setup the RPC listener
rpcListener, err := net.Listen("tcp", config.RPCAddr)
if err != nil {
agent.Shutdown()
c.Ui.Error(fmt.Sprintf("Error starting RPC listener: %s", err))
return err
}
// Start the IPC layer
c.Ui.Output("Starting Consul agent RPC...")
2013-12-30 23:27:41 +00:00
c.rpcServer = NewAgentRPC(agent, rpcListener, logOutput, logWriter)
2013-12-23 19:38:51 +00:00
if config.HTTPAddr != "" {
2014-01-02 19:45:58 +00:00
server, err := NewHTTPServer(agent, logOutput, config.HTTPAddr)
2013-12-23 19:38:51 +00:00
if err != nil {
agent.Shutdown()
c.Ui.Error(fmt.Sprintf("Error starting http server: %s", err))
return err
}
c.httpServer = server
}
if config.DNSAddr != "" {
2014-01-03 23:43:35 +00:00
server, err := NewDNSServer(agent, logOutput, config.Domain,
config.DNSAddr, config.DNSRecursor)
if err != nil {
agent.Shutdown()
c.Ui.Error(fmt.Sprintf("Error starting dns server: %s", err))
return err
}
c.dnsServer = server
}
2013-12-23 19:38:51 +00:00
return nil
}
2013-12-19 20:18:06 +00:00
func (c *Command) Run(args []string) int {
2013-12-20 01:14:46 +00:00
c.Ui = &cli.PrefixedUi{
OutputPrefix: "==> ",
InfoPrefix: " ",
ErrorPrefix: "==> ",
Ui: c.Ui,
}
2013-12-21 00:39:32 +00:00
// Parse our configs
c.args = args
config := c.readConfig()
if config == nil {
return 1
}
c.args = args
// Setup the log outputs
logGate, logWriter, logOutput := c.setupLoggers(config)
if logWriter == nil {
return 1
}
// Create the agent
2013-12-30 23:27:41 +00:00
if err := c.setupAgent(config, logOutput, logWriter); err != nil {
2013-12-20 01:14:46 +00:00
return 1
}
2013-12-23 19:38:51 +00:00
defer c.agent.Shutdown()
2013-12-30 23:27:41 +00:00
if c.rpcServer != nil {
defer c.rpcServer.Shutdown()
}
2013-12-23 19:38:51 +00:00
if c.httpServer != nil {
defer c.httpServer.Shutdown()
}
2013-12-20 01:14:46 +00:00
2013-12-21 00:39:32 +00:00
c.Ui.Output("Consul agent running!")
2014-01-01 00:45:13 +00:00
c.Ui.Info(fmt.Sprintf(" Node name: '%s'", config.NodeName))
c.Ui.Info(fmt.Sprintf(" Datacenter: '%s'", config.Datacenter))
c.Ui.Info(fmt.Sprintf("Advertise addr: '%s'", config.AdvertiseAddr))
c.Ui.Info(fmt.Sprintf(" RPC addr: '%s'", config.RPCAddr))
c.Ui.Info(fmt.Sprintf(" HTTP addr: '%s'", config.HTTPAddr))
c.Ui.Info(fmt.Sprintf(" DNS addr: '%s'", config.DNSAddr))
2014-01-01 00:45:13 +00:00
c.Ui.Info(fmt.Sprintf(" Encrypted: %#v", config.EncryptKey != ""))
c.Ui.Info(fmt.Sprintf(" Server: %v (bootstrap: %v)", config.Server, config.Bootstrap))
2013-12-21 00:39:32 +00:00
// Enable log streaming
c.Ui.Info("")
c.Ui.Output("Log data will now stream in as it occurs:\n")
logGate.Flush()
// Wait for exit
2013-12-23 19:38:51 +00:00
return c.handleSignals(config)
2013-12-21 00:39:32 +00:00
}
// handleSignals blocks until we get an exit-causing signal
2013-12-23 19:38:51 +00:00
func (c *Command) handleSignals(config *Config) int {
2013-12-21 00:39:32 +00:00
signalCh := make(chan os.Signal, 4)
signal.Notify(signalCh, os.Interrupt, syscall.SIGTERM, syscall.SIGHUP)
// Wait for a signal
WAIT:
var sig os.Signal
2013-12-20 01:14:46 +00:00
select {
2013-12-21 00:39:32 +00:00
case s := <-signalCh:
sig = s
2013-12-20 01:14:46 +00:00
case <-c.ShutdownCh:
2013-12-21 00:39:32 +00:00
sig = os.Interrupt
2013-12-23 19:38:51 +00:00
case <-c.agent.ShutdownCh():
2013-12-21 00:39:32 +00:00
// Agent is already shutdown!
2013-12-20 01:14:46 +00:00
return 0
}
2013-12-21 00:39:32 +00:00
c.Ui.Output(fmt.Sprintf("Caught signal: %v", sig))
// Check if this is a SIGHUP
if sig == syscall.SIGHUP {
2013-12-23 19:38:51 +00:00
config = c.handleReload(config)
2013-12-21 00:39:32 +00:00
goto WAIT
}
// Check if we should do a graceful leave
graceful := false
if sig == os.Interrupt && !config.SkipLeaveOnInt {
graceful = true
} else if sig == syscall.SIGTERM && config.LeaveOnTerm {
graceful = true
}
// Bail fast if not doing a graceful leave
if !graceful {
return 1
}
// Attempt a graceful leave
gracefulCh := make(chan struct{})
c.Ui.Output("Gracefully shutting down agent...")
go func() {
2013-12-23 19:38:51 +00:00
if err := c.agent.Leave(); err != nil {
2013-12-21 00:39:32 +00:00
c.Ui.Error(fmt.Sprintf("Error: %s", err))
return
}
close(gracefulCh)
}()
// Wait for leave or another signal
select {
case <-signalCh:
return 1
case <-time.After(gracefulTimeout):
return 1
case <-gracefulCh:
return 0
}
}
// handleReload is invoked when we should reload our configs, e.g. SIGHUP
2013-12-23 19:38:51 +00:00
func (c *Command) handleReload(config *Config) *Config {
2013-12-21 00:39:32 +00:00
c.Ui.Output("Reloading configuration...")
// TODO
return config
2013-12-19 20:18:06 +00:00
}
func (c *Command) Synopsis() string {
return "Runs a Consul agent"
}
func (c *Command) Help() string {
helpText := `
Usage: consul agent [options]
Starts the Consul agent and runs until an interrupt is received. The
2013-12-20 01:14:46 +00:00
agent represents a single node in a cluster. An agent can also serve
as a server by configuraiton.
2013-12-19 20:18:06 +00:00
Options:
-rpc-addr=127.0.0.1:7373 Address to bind the RPC listener.
`
return strings.TrimSpace(helpText)
}