mirror of https://github.com/status-im/consul.git
466 lines
12 KiB
Go
466 lines
12 KiB
Go
package agent
|
|
|
|
import (
|
|
"encoding/base64"
|
|
"encoding/json"
|
|
"fmt"
|
|
"github.com/hashicorp/consul/consul"
|
|
"github.com/mitchellh/mapstructure"
|
|
"io"
|
|
"os"
|
|
"path/filepath"
|
|
"sort"
|
|
"strings"
|
|
"time"
|
|
)
|
|
|
|
// Config is the configuration that can be set for an Agent.
|
|
// Some of this is configurable as CLI flags, but most must
|
|
// be set using a configuration file.
|
|
type Config struct {
|
|
// AEInterval controls the anti-entropy interval. This is how often
|
|
// the agent attempts to reconcile it's local state with the server'
|
|
// representation of our state. Defaults to every 60s.
|
|
AEInterval time.Duration `mapstructure:"-"`
|
|
|
|
// Bootstrap is used to bring up the first Consul server, and
|
|
// permits that node to elect itself leader
|
|
Bootstrap bool `mapstructure:"bootstrap"`
|
|
|
|
// Datacenter is the datacenter this node is in. Defaults to dc1
|
|
Datacenter string `mapstructure:"datacenter"`
|
|
|
|
// DataDir is the directory to store our state in
|
|
DataDir string `mapstructure:"data_dir"`
|
|
|
|
// DNSAddr is the address of the DNS server for the agent
|
|
DNSAddr string `mapstructure:"dns_addr"`
|
|
|
|
// DNSRecursor can be set to allow the DNS server to recursively
|
|
// resolve non-consul domains
|
|
DNSRecursor string `mapstructure:"recursor"`
|
|
|
|
// Domain is the DNS domain for the records. Defaults to "consul."
|
|
Domain string `mapstructure:"domain"`
|
|
|
|
// Encryption key to use for the Serf communication
|
|
EncryptKey string `mapstructure:"encrypt"`
|
|
|
|
// HTTP interface address
|
|
HTTPAddr string `mapstructure:"http_addr"`
|
|
|
|
// LogLevel is the level of the logs to putout
|
|
LogLevel string `mapstructure:"log_level"`
|
|
|
|
// Node name is the name we use to advertise. Defaults to hostname.
|
|
NodeName string `mapstructure:"node_name"`
|
|
|
|
// RPCAddr is the address and port to listen on for the
|
|
// agent's RPC interface.
|
|
RPCAddr string `mapstructure:"rpc_addr"`
|
|
|
|
// BindAddr is the address that Consul's RPC and Serf's will
|
|
// bind to. This address should be routable by all other hosts.
|
|
SerfBindAddr string `mapstructure:"serf_bind_addr"`
|
|
|
|
// SerfLanPort is the port we use for the lan-local serf cluster
|
|
// This is used for all nodes.
|
|
SerfLanPort int `mapstructure:"serf_lan_port"`
|
|
|
|
// SerfWanPort is the port we use for the wan serf cluster.
|
|
// This is only for the Consul servers
|
|
SerfWanPort int `mapstructure:"serf_wan_port"`
|
|
|
|
// ServerAddr is the address we use for Consul server communication.
|
|
// Defaults to 0.0.0.0:8300
|
|
ServerAddr string `mapstructure:"server_addr"`
|
|
|
|
// AdvertiseAddr is the address we use for advertising our Serf,
|
|
// and Consul RPC IP. If not specified, the first private IP we
|
|
// find is used.
|
|
AdvertiseAddr string `mapstructure:"advertise_addr"`
|
|
|
|
// Server controls if this agent acts like a Consul server,
|
|
// or merely as a client. Servers have more state, take part
|
|
// in leader election, etc.
|
|
Server bool `mapstructure:"server"`
|
|
|
|
// LeaveOnTerm controls if Serf does a graceful leave when receiving
|
|
// the TERM signal. Defaults false. This can be changed on reload.
|
|
LeaveOnTerm bool `mapstructure:"leave_on_terminate"`
|
|
|
|
// SkipLeaveOnInt controls if Serf skips a graceful leave when receiving
|
|
// the INT signal. Defaults false. This can be changed on reload.
|
|
SkipLeaveOnInt bool `mapstructure:"skip_leave_on_interrupt"`
|
|
|
|
// StatsiteAddr is the address of a statsite instance. If provided,
|
|
// metrics will be streamed to that instance.
|
|
StatsiteAddr string `mapstructure:"statsite_addr"`
|
|
|
|
// Protocol is the Consul protocol version to use.
|
|
Protocol int `mapstructure:"protocol"`
|
|
|
|
// EnableDebug is used to enable various debugging features
|
|
EnableDebug bool `mapstructure:"enable_debug"`
|
|
|
|
// VerifyIncoming is used to verify the authenticity of incoming connections.
|
|
// This means that TCP requests are forbidden, only allowing for TLS. TLS connections
|
|
// must match a provided certificate authority. This can be used to force client auth.
|
|
VerifyIncoming bool `mapstructure:"verify_incoming"`
|
|
|
|
// VerifyOutgoing is used to verify the authenticity of outgoing connections.
|
|
// This means that TLS requests are used. TLS connections must match a provided
|
|
// certificate authority. This is used to verify authenticity of server nodes.
|
|
VerifyOutgoing bool `mapstructure:"verify_outgoing"`
|
|
|
|
// CAFile is a path to a certificate authority file. This is used with VerifyIncoming
|
|
// or VerifyOutgoing to verify the TLS connection.
|
|
CAFile string `mapstructure:"ca_file"`
|
|
|
|
// CertFile is used to provide a TLS certificate that is used for serving TLS connections.
|
|
// Must be provided to serve TLS connections.
|
|
CertFile string `mapstructure:"cert_file"`
|
|
|
|
// KeyFile is used to provide a TLS key that is used for serving TLS connections.
|
|
// Must be provided to serve TLS connections.
|
|
KeyFile string `mapstructure:"key_file"`
|
|
|
|
// Checks holds the provided check definitions
|
|
Checks []*CheckDefinition `mapstructure:"-"`
|
|
|
|
// Services holds the provided service definitions
|
|
Services []*ServiceDefinition `mapstructure:"-"`
|
|
|
|
// ConsulConfig can either be provided or a default one created
|
|
ConsulConfig *consul.Config `mapstructure:"-"`
|
|
}
|
|
|
|
type dirEnts []os.FileInfo
|
|
|
|
// DefaultConfig is used to return a sane default configuration
|
|
func DefaultConfig() *Config {
|
|
return &Config{
|
|
AEInterval: time.Minute,
|
|
Datacenter: consul.DefaultDC,
|
|
DNSAddr: "127.0.0.1:8600",
|
|
Domain: "consul.",
|
|
HTTPAddr: "127.0.0.1:8500",
|
|
LogLevel: "INFO",
|
|
RPCAddr: "127.0.0.1:8400",
|
|
SerfLanPort: consul.DefaultLANSerfPort,
|
|
SerfWanPort: consul.DefaultWANSerfPort,
|
|
Server: false,
|
|
Protocol: consul.ProtocolVersionMax,
|
|
}
|
|
}
|
|
|
|
// EncryptBytes returns the encryption key configured.
|
|
func (c *Config) EncryptBytes() ([]byte, error) {
|
|
return base64.StdEncoding.DecodeString(c.EncryptKey)
|
|
}
|
|
|
|
// DecodeConfig reads the configuration from the given reader in JSON
|
|
// format and decodes it into a proper Config structure.
|
|
func DecodeConfig(r io.Reader) (*Config, error) {
|
|
var raw interface{}
|
|
var result Config
|
|
dec := json.NewDecoder(r)
|
|
if err := dec.Decode(&raw); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
// Check the result type
|
|
if obj, ok := raw.(map[string]interface{}); ok {
|
|
// Check for a "service" or "check" key, meaning
|
|
// this is actually a definition entry
|
|
if sub, ok := obj["service"]; ok {
|
|
service, err := DecodeServiceDefinition(sub)
|
|
result.Services = append(result.Services, service)
|
|
return &result, err
|
|
} else if sub, ok := obj["check"]; ok {
|
|
check, err := DecodeCheckDefinition(sub)
|
|
result.Checks = append(result.Checks, check)
|
|
return &result, err
|
|
}
|
|
}
|
|
|
|
// Decode
|
|
var md mapstructure.Metadata
|
|
msdec, err := mapstructure.NewDecoder(&mapstructure.DecoderConfig{
|
|
Metadata: &md,
|
|
Result: &result,
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if err := msdec.Decode(raw); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return &result, nil
|
|
}
|
|
|
|
// DecodeServiceDefinition is used to decode a service definition
|
|
func DecodeServiceDefinition(raw interface{}) (*ServiceDefinition, error) {
|
|
var sub interface{}
|
|
rawMap, ok := raw.(map[string]interface{})
|
|
if !ok {
|
|
goto AFTER_FIX
|
|
}
|
|
|
|
// If no 'tags', handle the deprecated 'tag' value.
|
|
if _, ok := rawMap["tags"]; !ok {
|
|
if tag, ok := rawMap["tag"]; ok {
|
|
rawMap["tags"] = []interface{}{tag}
|
|
}
|
|
}
|
|
|
|
sub, ok = rawMap["check"]
|
|
if !ok {
|
|
goto AFTER_FIX
|
|
}
|
|
if err := FixupCheckType(sub); err != nil {
|
|
return nil, err
|
|
}
|
|
AFTER_FIX:
|
|
var md mapstructure.Metadata
|
|
var result ServiceDefinition
|
|
msdec, err := mapstructure.NewDecoder(&mapstructure.DecoderConfig{
|
|
Metadata: &md,
|
|
Result: &result,
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if err := msdec.Decode(raw); err != nil {
|
|
return nil, err
|
|
}
|
|
return &result, nil
|
|
}
|
|
|
|
func FixupCheckType(raw interface{}) error {
|
|
// Handle decoding of time durations
|
|
rawMap, ok := raw.(map[string]interface{})
|
|
if !ok {
|
|
return nil
|
|
}
|
|
if ttl, ok := rawMap["ttl"]; ok {
|
|
ttlS, ok := ttl.(string)
|
|
if ok {
|
|
if dur, err := time.ParseDuration(ttlS); err != nil {
|
|
return err
|
|
} else {
|
|
rawMap["ttl"] = dur
|
|
}
|
|
}
|
|
}
|
|
if interval, ok := rawMap["interval"]; ok {
|
|
intervalS, ok := interval.(string)
|
|
if ok {
|
|
if dur, err := time.ParseDuration(intervalS); err != nil {
|
|
return err
|
|
} else {
|
|
rawMap["interval"] = dur
|
|
}
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// DecodeCheckDefinition is used to decode a check definition
|
|
func DecodeCheckDefinition(raw interface{}) (*CheckDefinition, error) {
|
|
if err := FixupCheckType(raw); err != nil {
|
|
return nil, err
|
|
}
|
|
var md mapstructure.Metadata
|
|
var result CheckDefinition
|
|
msdec, err := mapstructure.NewDecoder(&mapstructure.DecoderConfig{
|
|
Metadata: &md,
|
|
Result: &result,
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if err := msdec.Decode(raw); err != nil {
|
|
return nil, err
|
|
}
|
|
return &result, nil
|
|
}
|
|
|
|
// MergeConfig merges two configurations together to make a single new
|
|
// configuration.
|
|
func MergeConfig(a, b *Config) *Config {
|
|
var result Config = *a
|
|
|
|
// Copy the strings if they're set
|
|
if b.Bootstrap {
|
|
result.Bootstrap = true
|
|
}
|
|
if b.Datacenter != "" {
|
|
result.Datacenter = b.Datacenter
|
|
}
|
|
if b.DataDir != "" {
|
|
result.DataDir = b.DataDir
|
|
}
|
|
if b.DNSAddr != "" {
|
|
result.DNSAddr = b.DNSAddr
|
|
}
|
|
if b.DNSRecursor != "" {
|
|
result.DNSRecursor = b.DNSRecursor
|
|
}
|
|
if b.Domain != "" {
|
|
result.Domain = b.Domain
|
|
}
|
|
if b.EncryptKey != "" {
|
|
result.EncryptKey = b.EncryptKey
|
|
}
|
|
if b.HTTPAddr != "" {
|
|
result.HTTPAddr = b.HTTPAddr
|
|
}
|
|
if b.LogLevel != "" {
|
|
result.LogLevel = b.LogLevel
|
|
}
|
|
if b.Protocol > 0 {
|
|
result.Protocol = b.Protocol
|
|
}
|
|
if b.NodeName != "" {
|
|
result.NodeName = b.NodeName
|
|
}
|
|
if b.RPCAddr != "" {
|
|
result.RPCAddr = b.RPCAddr
|
|
}
|
|
if b.SerfBindAddr != "" {
|
|
result.SerfBindAddr = b.SerfBindAddr
|
|
}
|
|
if b.SerfLanPort > 0 {
|
|
result.SerfLanPort = b.SerfLanPort
|
|
}
|
|
if b.SerfWanPort > 0 {
|
|
result.SerfWanPort = b.SerfWanPort
|
|
}
|
|
if b.ServerAddr != "" {
|
|
result.ServerAddr = b.ServerAddr
|
|
}
|
|
if b.AdvertiseAddr != "" {
|
|
result.AdvertiseAddr = b.AdvertiseAddr
|
|
}
|
|
if b.Server == true {
|
|
result.Server = b.Server
|
|
}
|
|
if b.LeaveOnTerm == true {
|
|
result.LeaveOnTerm = true
|
|
}
|
|
if b.SkipLeaveOnInt == true {
|
|
result.SkipLeaveOnInt = true
|
|
}
|
|
if b.EnableDebug {
|
|
result.EnableDebug = true
|
|
}
|
|
if b.VerifyIncoming {
|
|
result.VerifyIncoming = true
|
|
}
|
|
if b.VerifyOutgoing {
|
|
result.VerifyOutgoing = true
|
|
}
|
|
if b.CAFile != "" {
|
|
result.CAFile = b.CAFile
|
|
}
|
|
if b.CertFile != "" {
|
|
result.CertFile = b.CertFile
|
|
}
|
|
if b.KeyFile != "" {
|
|
result.KeyFile = b.KeyFile
|
|
}
|
|
if b.Checks != nil {
|
|
result.Checks = append(result.Checks, b.Checks...)
|
|
}
|
|
if b.Services != nil {
|
|
result.Services = append(result.Services, b.Services...)
|
|
}
|
|
return &result
|
|
}
|
|
|
|
// ReadConfigPaths reads the paths in the given order to load configurations.
|
|
// The paths can be to files or directories. If the path is a directory,
|
|
// we read one directory deep and read any files ending in ".json" as
|
|
// configuration files.
|
|
func ReadConfigPaths(paths []string) (*Config, error) {
|
|
result := new(Config)
|
|
for _, path := range paths {
|
|
f, err := os.Open(path)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Error reading '%s': %s", path, err)
|
|
}
|
|
|
|
fi, err := f.Stat()
|
|
if err != nil {
|
|
f.Close()
|
|
return nil, fmt.Errorf("Error reading '%s': %s", path, err)
|
|
}
|
|
|
|
if !fi.IsDir() {
|
|
config, err := DecodeConfig(f)
|
|
f.Close()
|
|
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Error decoding '%s': %s", path, err)
|
|
}
|
|
|
|
result = MergeConfig(result, config)
|
|
continue
|
|
}
|
|
|
|
contents, err := f.Readdir(-1)
|
|
f.Close()
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Error reading '%s': %s", path, err)
|
|
}
|
|
|
|
// Sort the contents, ensures lexical order
|
|
sort.Sort(dirEnts(contents))
|
|
|
|
for _, fi := range contents {
|
|
// Don't recursively read contents
|
|
if fi.IsDir() {
|
|
continue
|
|
}
|
|
|
|
// If it isn't a JSON file, ignore it
|
|
if !strings.HasSuffix(fi.Name(), ".json") {
|
|
continue
|
|
}
|
|
|
|
subpath := filepath.Join(path, fi.Name())
|
|
f, err := os.Open(subpath)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Error reading '%s': %s", subpath, err)
|
|
}
|
|
|
|
config, err := DecodeConfig(f)
|
|
f.Close()
|
|
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Error decoding '%s': %s", subpath, err)
|
|
}
|
|
|
|
result = MergeConfig(result, config)
|
|
}
|
|
}
|
|
|
|
return result, nil
|
|
}
|
|
|
|
// Implement the sort interface for dirEnts
|
|
func (d dirEnts) Len() int {
|
|
return len(d)
|
|
}
|
|
|
|
func (d dirEnts) Less(i, j int) bool {
|
|
return d[i].Name() < d[j].Name()
|
|
}
|
|
|
|
func (d dirEnts) Swap(i, j int) {
|
|
d[i], d[j] = d[j], d[i]
|
|
}
|