etcd/embed/config.go

873 lines
29 KiB
Go
Raw Normal View History

// Copyright 2016 The etcd Authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package embed
import (
"crypto/tls"
"fmt"
"io/ioutil"
2016-09-13 05:49:22 +03:00
"net"
"net/http"
"net/url"
"os"
"path/filepath"
"strings"
"sync"
"syscall"
"time"
"github.com/coreos/etcd/compactor"
"github.com/coreos/etcd/etcdserver"
"github.com/coreos/etcd/pkg/flags"
"github.com/coreos/etcd/pkg/logutil"
"github.com/coreos/etcd/pkg/netutil"
"github.com/coreos/etcd/pkg/srv"
"github.com/coreos/etcd/pkg/transport"
"github.com/coreos/etcd/pkg/types"
"github.com/coreos/pkg/capnslog"
"github.com/ghodss/yaml"
"go.uber.org/zap"
"google.golang.org/grpc"
"google.golang.org/grpc/grpclog"
)
const (
ClusterStateFlagNew = "new"
ClusterStateFlagExisting = "existing"
DefaultName = "default"
DefaultMaxSnapshots = 5
DefaultMaxWALs = 5
DefaultMaxTxnOps = uint(128)
DefaultMaxRequestBytes = 1.5 * 1024 * 1024
DefaultGRPCKeepAliveMinTime = 5 * time.Second
DefaultGRPCKeepAliveInterval = 2 * time.Hour
DefaultGRPCKeepAliveTimeout = 20 * time.Second
DefaultListenPeerURLs = "http://localhost:2380"
DefaultListenClientURLs = "http://localhost:2379"
DefaultLogOutput = "default"
// DefaultStrictReconfigCheck is the default value for "--strict-reconfig-check" flag.
// It's enabled by default.
DefaultStrictReconfigCheck = true
// DefaultEnableV2 is the default value for "--enable-v2" flag.
// v2 is enabled by default.
// TODO: disable v2 when deprecated.
DefaultEnableV2 = true
// maxElectionMs specifies the maximum value of election timeout.
// More details are listed in ../Documentation/tuning.md#time-parameters.
maxElectionMs = 50000
)
var (
ErrConflictBootstrapFlags = fmt.Errorf("multiple discovery or bootstrap flags are set. " +
"Choose one of \"initial-cluster\", \"discovery\" or \"discovery-srv\"")
ErrUnsetAdvertiseClientURLsFlag = fmt.Errorf("--advertise-client-urls is required when --listen-client-urls is set explicitly")
DefaultInitialAdvertisePeerURLs = "http://localhost:2380"
DefaultAdvertiseClientURLs = "http://localhost:2379"
defaultHostname string
defaultHostStatus error
)
var (
// CompactorModePeriodic is periodic compaction mode
// for "Config.AutoCompactionMode" field.
// If "AutoCompactionMode" is CompactorModePeriodic and
// "AutoCompactionRetention" is "1h", it automatically compacts
// compacts storage every hour.
CompactorModePeriodic = compactor.ModePeriodic
// CompactorModeRevision is revision-based compaction mode
// for "Config.AutoCompactionMode" field.
// If "AutoCompactionMode" is CompactorModeRevision and
// "AutoCompactionRetention" is "1000", it compacts log on
// revision 5000 when the current revision is 6000.
// This runs every 5-minute if enough of logs have proceeded.
CompactorModeRevision = compactor.ModeRevision
)
func init() {
defaultHostname, defaultHostStatus = netutil.GetDefaultHost()
}
// Config holds the arguments for configuring an etcd server.
type Config struct {
Name string `json:"name"`
Dir string `json:"data-dir"`
WalDir string `json:"wal-dir"`
SnapCount uint64 `json:"snapshot-count"`
MaxSnapFiles uint `json:"max-snapshots"`
MaxWalFiles uint `json:"max-wals"`
// TickMs is the number of milliseconds between heartbeat ticks.
// TODO: decouple tickMs and heartbeat tick (current heartbeat tick = 1).
// make ticks a cluster wide configuration.
TickMs uint `json:"heartbeat-interval"`
ElectionMs uint `json:"election-timeout"`
QuotaBackendBytes int64 `json:"quota-backend-bytes"`
MaxTxnOps uint `json:"max-txn-ops"`
MaxRequestBytes uint `json:"max-request-bytes"`
LPUrls, LCUrls []url.URL
APUrls, ACUrls []url.URL
ClientTLSInfo transport.TLSInfo
ClientAutoTLS bool
PeerTLSInfo transport.TLSInfo
PeerAutoTLS bool
ClusterState string `json:"initial-cluster-state"`
DNSCluster string `json:"discovery-srv"`
DNSClusterServiceName string `json:"discovery-srv-name"`
Dproxy string `json:"discovery-proxy"`
Durl string `json:"discovery"`
InitialCluster string `json:"initial-cluster"`
InitialClusterToken string `json:"initial-cluster-token"`
StrictReconfigCheck bool `json:"strict-reconfig-check"`
EnableV2 bool `json:"enable-v2"`
// AutoCompactionMode is either 'periodic' or 'revision'.
AutoCompactionMode string `json:"auto-compaction-mode"`
// AutoCompactionRetention is either duration string with time unit
// (e.g. '5m' for 5-minute), or revision unit (e.g. '5000').
// If no time unit is provided and compaction mode is 'periodic',
// the unit defaults to hour. For example, '5' translates into 5-hour.
AutoCompactionRetention string `json:"auto-compaction-retention"`
// GRPCKeepAliveMinTime is the minimum interval that a client should
// wait before pinging server. When client pings "too fast", server
// sends goaway and closes the connection (errors: too_many_pings,
// http2.ErrCodeEnhanceYourCalm). When too slow, nothing happens.
// Server expects client pings only when there is any active streams
// (PermitWithoutStream is set false).
GRPCKeepAliveMinTime time.Duration `json:"grpc-keepalive-min-time"`
// GRPCKeepAliveInterval is the frequency of server-to-client ping
// to check if a connection is alive. Close a non-responsive connection
// after an additional duration of Timeout. 0 to disable.
GRPCKeepAliveInterval time.Duration `json:"grpc-keepalive-interval"`
// GRPCKeepAliveTimeout is the additional duration of wait
// before closing a non-responsive connection. 0 to disable.
GRPCKeepAliveTimeout time.Duration `json:"grpc-keepalive-timeout"`
// PreVote is true to enable Raft Pre-Vote.
// If enabled, Raft runs an additional election phase
// to check whether it would get enough votes to win
// an election, thus minimizing disruptions.
// TODO: enable by default in 3.5.
PreVote bool `json:"pre-vote"`
CORS map[string]struct{}
// HostWhitelist lists acceptable hostnames from HTTP client requests.
// Client origin policy protects against "DNS Rebinding" attacks
// to insecure etcd servers. That is, any website can simply create
// an authorized DNS name, and direct DNS to "localhost" (or any
// other address). Then, all HTTP endpoints of etcd server listening
// on "localhost" becomes accessible, thus vulnerable to DNS rebinding
// attacks. See "CVE-2018-5702" for more detail.
//
// 1. If client connection is secure via HTTPS, allow any hostnames.
// 2. If client connection is not secure and "HostWhitelist" is not empty,
// only allow HTTP requests whose Host field is listed in whitelist.
//
// Note that the client origin policy is enforced whether authentication
// is enabled or not, for tighter controls.
//
// By default, "HostWhitelist" is "*", which allows any hostnames.
// Note that when specifying hostnames, loopback addresses are not added
// automatically. To allow loopback interfaces, leave it empty or set it "*",
// or add them to whitelist manually (e.g. "localhost", "127.0.0.1", etc.).
//
// CVE-2018-5702 reference:
// - https://bugs.chromium.org/p/project-zero/issues/detail?id=1447#c2
// - https://github.com/transmission/transmission/pull/468
// - https://github.com/coreos/etcd/issues/9353
HostWhitelist map[string]struct{}
// UserHandlers is for registering users handlers and only used for
// embedding etcd into other applications.
// The map key is the route path for the handler, and
// you must ensure it can't be conflicted with etcd's.
UserHandlers map[string]http.Handler `json:"-"`
// ServiceRegister is for registering users' gRPC services. A simple usage example:
// cfg := embed.NewConfig()
// cfg.ServerRegister = func(s *grpc.Server) {
// pb.RegisterFooServer(s, &fooServer{})
// pb.RegisterBarServer(s, &barServer{})
// }
// embed.StartEtcd(cfg)
ServiceRegister func(*grpc.Server) `json:"-"`
*: support jwt token in v3 auth API This commit adds jwt token support in v3 auth API. Remaining major ToDos: - Currently token type isn't hidden from etcdserver. In the near future the information should be completely invisible from etcdserver package. - Configurable expiration of token. Currently tokens can be valid until keys are changed. How to use: 1. generate keys for signing and verfying jwt tokens: $ openssl genrsa -out app.rsa 1024 $ openssl rsa -in app.rsa -pubout > app.rsa.pub 2. add command line options to etcd like below: --auth-token-type jwt \ --auth-jwt-pub-key app.rsa.pub --auth-jwt-priv-key app.rsa \ --auth-jwt-sign-method RS512 3. launch etcd cluster Below is a performance comparison of serializable read w/ and w/o jwt token. Every (3) etcd node is executed on a single machine. Signing method is RS512 and key length is 1024 bit. As the results show, jwt based token introduces a performance overhead but it would be acceptable for a case that requires authentication. w/o jwt token auth (no auth): Summary: Total: 1.6172 secs. Slowest: 0.0125 secs. Fastest: 0.0001 secs. Average: 0.0002 secs. Stddev: 0.0004 secs. Requests/sec: 6183.5877 Response time histogram: 0.000 [1] | 0.001 [9982] |∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎ 0.003 [1] | 0.004 [1] | 0.005 [0] | 0.006 [0] | 0.008 [6] | 0.009 [0] | 0.010 [1] | 0.011 [5] | 0.013 [3] | Latency distribution: 10% in 0.0001 secs. 25% in 0.0001 secs. 50% in 0.0001 secs. 75% in 0.0001 secs. 90% in 0.0002 secs. 95% in 0.0002 secs. 99% in 0.0003 secs. w/ jwt token auth: Summary: Total: 2.5364 secs. Slowest: 0.0182 secs. Fastest: 0.0002 secs. Average: 0.0003 secs. Stddev: 0.0005 secs. Requests/sec: 3942.5185 Response time histogram: 0.000 [1] | 0.002 [9975] |∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎ 0.004 [0] | 0.006 [1] | 0.007 [11] | 0.009 [2] | 0.011 [4] | 0.013 [5] | 0.015 [0] | 0.016 [0] | 0.018 [1] | Latency distribution: 10% in 0.0002 secs. 25% in 0.0002 secs. 50% in 0.0002 secs. 75% in 0.0002 secs. 90% in 0.0003 secs. 95% in 0.0003 secs. 99% in 0.0004 secs.
2016-07-21 08:13:57 +03:00
AuthToken string `json:"auth-token"`
ExperimentalInitialCorruptCheck bool `json:"experimental-initial-corrupt-check"`
ExperimentalCorruptCheckTime time.Duration `json:"experimental-corrupt-check-time"`
ExperimentalEnableV2V3 string `json:"experimental-enable-v2v3"`
// ForceNewCluster starts a new cluster even if previously started; unsafe.
ForceNewCluster bool `json:"force-new-cluster"`
EnablePprof bool `json:"enable-pprof"`
Metrics string `json:"metrics"`
ListenMetricsUrls []url.URL
ListenMetricsUrlsJSON string `json:"listen-metrics-urls"`
// logger logs server-side operations. The default is nil,
// and "setupLogging" must be called before starting server.
// Do not set logger directly.
loggerMu *sync.RWMutex
logger *zap.Logger
loggerConfig zap.Config
// Logger is logger options: "zap", "capnslog".
// WARN: "capnslog" is being deprecated in v3.5.
Logger string `json:"logger"`
// LogOutput is either:
// - "default" as os.Stderr
// - "stderr" as os.Stderr
// - "stdout" as os.Stdout
// - file path to append server logs to
LogOutput string `json:"log-output"`
// Debug is true, to enable debug level logging.
Debug bool `json:"debug"`
// LogPkgLevels is being deprecated in v3.5.
// Only valid if "logger" option is "capnslog".
// WARN: DO NOT USE THIS!
LogPkgLevels string `json:"log-package-levels"`
}
// configYAML holds the config suitable for yaml parsing
type configYAML struct {
Config
configJSON
}
// configJSON has file options that are translated into Config options
type configJSON struct {
LPUrlsJSON string `json:"listen-peer-urls"`
LCUrlsJSON string `json:"listen-client-urls"`
APUrlsJSON string `json:"initial-advertise-peer-urls"`
ACUrlsJSON string `json:"advertise-client-urls"`
CORSJSON string `json:"cors"`
HostWhitelistJSON string `json:"host-whitelist"`
ClientSecurityJSON securityConfig `json:"client-transport-security"`
PeerSecurityJSON securityConfig `json:"peer-transport-security"`
}
type securityConfig struct {
CertFile string `json:"cert-file"`
KeyFile string `json:"key-file"`
CertAuth bool `json:"client-cert-auth"`
TrustedCAFile string `json:"trusted-ca-file"`
AutoTLS bool `json:"auto-tls"`
}
// NewConfig creates a new Config populated with default values.
func NewConfig() *Config {
lpurl, _ := url.Parse(DefaultListenPeerURLs)
apurl, _ := url.Parse(DefaultInitialAdvertisePeerURLs)
lcurl, _ := url.Parse(DefaultListenClientURLs)
acurl, _ := url.Parse(DefaultAdvertiseClientURLs)
cfg := &Config{
MaxSnapFiles: DefaultMaxSnapshots,
MaxWalFiles: DefaultMaxWALs,
Name: DefaultName,
SnapCount: etcdserver.DefaultSnapCount,
MaxTxnOps: DefaultMaxTxnOps,
MaxRequestBytes: DefaultMaxRequestBytes,
GRPCKeepAliveMinTime: DefaultGRPCKeepAliveMinTime,
GRPCKeepAliveInterval: DefaultGRPCKeepAliveInterval,
GRPCKeepAliveTimeout: DefaultGRPCKeepAliveTimeout,
TickMs: 100,
ElectionMs: 1000,
LPUrls: []url.URL{*lpurl},
LCUrls: []url.URL{*lcurl},
APUrls: []url.URL{*apurl},
ACUrls: []url.URL{*acurl},
ClusterState: ClusterStateFlagNew,
InitialClusterToken: "etcd-cluster",
StrictReconfigCheck: DefaultStrictReconfigCheck,
Metrics: "basic",
EnableV2: DefaultEnableV2,
CORS: map[string]struct{}{"*": {}},
HostWhitelist: map[string]struct{}{"*": {}},
AuthToken: "simple",
PreVote: false, // TODO: enable by default in v3.5
loggerMu: new(sync.RWMutex),
logger: nil,
Logger: "capnslog",
LogOutput: DefaultLogOutput,
Debug: false,
LogPkgLevels: "",
}
cfg.InitialCluster = cfg.InitialClusterFromName(cfg.Name)
return cfg
}
func logTLSHandshakeFailure(conn *tls.Conn, err error) {
state := conn.ConnectionState()
remoteAddr := conn.RemoteAddr().String()
serverName := state.ServerName
if len(state.PeerCertificates) > 0 {
cert := state.PeerCertificates[0]
ips, dns := cert.IPAddresses, cert.DNSNames
plog.Infof("rejected connection from %q (error %q, ServerName %q, IPAddresses %q, DNSNames %q)", remoteAddr, err.Error(), serverName, ips, dns)
} else {
plog.Infof("rejected connection from %q (error %q, ServerName %q)", remoteAddr, err.Error(), serverName)
}
}
// GetLogger returns the logger.
func (cfg Config) GetLogger() *zap.Logger {
cfg.loggerMu.RLock()
l := cfg.logger
cfg.loggerMu.RUnlock()
return l
}
// for testing
var grpcLogOnce = new(sync.Once)
// setupLogging initializes etcd logging.
// Must be called after flag parsing or finishing configuring embed.Config.
func (cfg *Config) setupLogging() error {
switch cfg.Logger {
case "capnslog": // TODO: deprecate this in v3.5
cfg.ClientTLSInfo.HandshakeFailure = logTLSHandshakeFailure
cfg.PeerTLSInfo.HandshakeFailure = logTLSHandshakeFailure
if cfg.Debug {
capnslog.SetGlobalLogLevel(capnslog.DEBUG)
grpc.EnableTracing = true
// enable info, warning, error
grpclog.SetLoggerV2(grpclog.NewLoggerV2(os.Stderr, os.Stderr, os.Stderr))
} else {
capnslog.SetGlobalLogLevel(capnslog.INFO)
// only discard info
grpclog.SetLoggerV2(grpclog.NewLoggerV2(ioutil.Discard, os.Stderr, os.Stderr))
}
// TODO: deprecate with "capnslog"
if cfg.LogPkgLevels != "" {
repoLog := capnslog.MustRepoLogger("github.com/coreos/etcd")
settings, err := repoLog.ParseLogLevelConfig(cfg.LogPkgLevels)
if err != nil {
plog.Warningf("couldn't parse log level string: %s, continuing with default levels", err.Error())
return nil
}
repoLog.SetLogLevel(settings)
}
// capnslog initially SetFormatter(NewDefaultFormatter(os.Stderr))
// where NewDefaultFormatter returns NewJournaldFormatter when syscall.Getppid() == 1
// specify 'stdout' or 'stderr' to skip journald logging even when running under systemd
switch cfg.LogOutput {
case "stdout":
capnslog.SetFormatter(capnslog.NewPrettyFormatter(os.Stdout, cfg.Debug))
case "stderr":
capnslog.SetFormatter(capnslog.NewPrettyFormatter(os.Stderr, cfg.Debug))
case DefaultLogOutput:
default:
plog.Panicf(`unknown log-output %q (only supports %q, "stdout", "stderr")`, cfg.LogOutput, DefaultLogOutput)
}
case "zap":
// TODO: make this more configurable
lcfg := zap.Config{
Level: zap.NewAtomicLevelAt(zap.InfoLevel),
Development: false,
Sampling: &zap.SamplingConfig{
Initial: 100,
Thereafter: 100,
},
Encoding: "json",
EncoderConfig: zap.NewProductionEncoderConfig(),
}
switch cfg.LogOutput {
case DefaultLogOutput:
if syscall.Getppid() == 1 {
// capnslog initially SetFormatter(NewDefaultFormatter(os.Stderr))
// where "NewDefaultFormatter" returns "NewJournaldFormatter"
// when syscall.Getppid() == 1, specify 'stdout' or 'stderr' to
// skip journald logging even when running under systemd
fmt.Println("running under init, which may be systemd!")
// TODO: capnlog.NewJournaldFormatter()
lcfg.OutputPaths = []string{"stderr"}
lcfg.ErrorOutputPaths = []string{"stderr"}
} else {
lcfg.OutputPaths = []string{"stderr"}
lcfg.ErrorOutputPaths = []string{"stderr"}
}
case "stderr":
lcfg.OutputPaths = []string{"stderr"}
lcfg.ErrorOutputPaths = []string{"stderr"}
case "stdout":
lcfg.OutputPaths = []string{"stdout"}
lcfg.ErrorOutputPaths = []string{"stdout"}
default:
lcfg.OutputPaths = []string{cfg.LogOutput}
lcfg.ErrorOutputPaths = []string{cfg.LogOutput}
}
if cfg.Debug {
lcfg.Level = zap.NewAtomicLevelAt(zap.DebugLevel)
grpc.EnableTracing = true
}
var err error
cfg.logger, err = lcfg.Build()
if err != nil {
return err
}
cfg.loggerConfig = lcfg
grpcLogOnce.Do(func() {
// debug true, enable info, warning, error
// debug false, only discard info
var gl grpclog.LoggerV2
gl, err = logutil.NewGRPCLoggerV2(lcfg)
if err == nil {
grpclog.SetLoggerV2(gl)
}
})
if err != nil {
return err
}
logTLSHandshakeFailure := func(conn *tls.Conn, err error) {
state := conn.ConnectionState()
remoteAddr := conn.RemoteAddr().String()
serverName := state.ServerName
if len(state.PeerCertificates) > 0 {
cert := state.PeerCertificates[0]
ips := make([]string, 0, len(cert.IPAddresses))
for i := range cert.IPAddresses {
ips[i] = cert.IPAddresses[i].String()
}
cfg.logger.Warn(
"rejected connection",
zap.String("remote-addr", remoteAddr),
zap.String("server-name", serverName),
zap.Strings("ip-addresses", ips),
zap.Strings("dns-names", cert.DNSNames),
zap.Error(err),
)
} else {
cfg.logger.Warn(
"rejected connection",
zap.String("remote-addr", remoteAddr),
zap.String("server-name", serverName),
zap.Error(err),
)
}
}
cfg.ClientTLSInfo.HandshakeFailure = logTLSHandshakeFailure
cfg.PeerTLSInfo.HandshakeFailure = logTLSHandshakeFailure
default:
return fmt.Errorf("unknown logger option %q", cfg.Logger)
}
return nil
}
func ConfigFromFile(path string) (*Config, error) {
cfg := &configYAML{Config: *NewConfig()}
if err := cfg.configFromFile(path); err != nil {
return nil, err
}
return &cfg.Config, nil
}
func (cfg *configYAML) configFromFile(path string) error {
b, err := ioutil.ReadFile(path)
if err != nil {
return err
}
defaultInitialCluster := cfg.InitialCluster
err = yaml.Unmarshal(b, cfg)
if err != nil {
return err
}
if cfg.LPUrlsJSON != "" {
u, err := types.NewURLs(strings.Split(cfg.LPUrlsJSON, ","))
if err != nil {
fmt.Fprintf(os.Stderr, "unexpected error setting up listen-peer-urls: %v\n", err)
os.Exit(1)
}
cfg.LPUrls = []url.URL(u)
}
if cfg.LCUrlsJSON != "" {
u, err := types.NewURLs(strings.Split(cfg.LCUrlsJSON, ","))
if err != nil {
fmt.Fprintf(os.Stderr, "unexpected error setting up listen-client-urls: %v\n", err)
os.Exit(1)
}
cfg.LCUrls = []url.URL(u)
}
if cfg.APUrlsJSON != "" {
u, err := types.NewURLs(strings.Split(cfg.APUrlsJSON, ","))
if err != nil {
fmt.Fprintf(os.Stderr, "unexpected error setting up initial-advertise-peer-urls: %v\n", err)
os.Exit(1)
}
cfg.APUrls = []url.URL(u)
}
if cfg.ACUrlsJSON != "" {
u, err := types.NewURLs(strings.Split(cfg.ACUrlsJSON, ","))
if err != nil {
fmt.Fprintf(os.Stderr, "unexpected error setting up advertise-peer-urls: %v\n", err)
os.Exit(1)
}
cfg.ACUrls = []url.URL(u)
}
if cfg.ListenMetricsUrlsJSON != "" {
u, err := types.NewURLs(strings.Split(cfg.ListenMetricsUrlsJSON, ","))
if err != nil {
fmt.Fprintf(os.Stderr, "unexpected error setting up listen-metrics-urls: %v\n", err)
os.Exit(1)
}
cfg.ListenMetricsUrls = []url.URL(u)
}
if cfg.CORSJSON != "" {
uv := flags.NewUniqueURLsWithExceptions(cfg.CORSJSON, "*")
cfg.CORS = uv.Values
}
if cfg.HostWhitelistJSON != "" {
uv := flags.NewUniqueStringsValue(cfg.HostWhitelistJSON)
cfg.HostWhitelist = uv.Values
}
// If a discovery flag is set, clear default initial cluster set by InitialClusterFromName
if (cfg.Durl != "" || cfg.DNSCluster != "") && cfg.InitialCluster == defaultInitialCluster {
cfg.InitialCluster = ""
}
if cfg.ClusterState == "" {
cfg.ClusterState = ClusterStateFlagNew
}
copySecurityDetails := func(tls *transport.TLSInfo, ysc *securityConfig) {
tls.CertFile = ysc.CertFile
tls.KeyFile = ysc.KeyFile
tls.ClientCertAuth = ysc.CertAuth
tls.TrustedCAFile = ysc.TrustedCAFile
}
copySecurityDetails(&cfg.ClientTLSInfo, &cfg.ClientSecurityJSON)
copySecurityDetails(&cfg.PeerTLSInfo, &cfg.PeerSecurityJSON)
cfg.ClientAutoTLS = cfg.ClientSecurityJSON.AutoTLS
cfg.PeerAutoTLS = cfg.PeerSecurityJSON.AutoTLS
return cfg.Validate()
}
// Validate ensures that '*embed.Config' fields are properly configured.
func (cfg *Config) Validate() error {
if err := cfg.setupLogging(); err != nil {
return err
}
2016-09-13 05:49:22 +03:00
if err := checkBindURLs(cfg.LPUrls); err != nil {
return err
}
if err := checkBindURLs(cfg.LCUrls); err != nil {
return err
}
if err := checkBindURLs(cfg.ListenMetricsUrls); err != nil {
return err
}
if err := checkHostURLs(cfg.APUrls); err != nil {
addrs := make([]string, len(cfg.APUrls))
for i := range cfg.APUrls {
addrs[i] = cfg.APUrls[i].String()
}
return fmt.Errorf(`--initial-advertise-peer-urls %q must be "host:port" (%v)`, strings.Join(addrs, ","), err)
}
if err := checkHostURLs(cfg.ACUrls); err != nil {
addrs := make([]string, len(cfg.ACUrls))
for i := range cfg.ACUrls {
addrs[i] = cfg.ACUrls[i].String()
}
return fmt.Errorf(`--advertise-client-urls %q must be "host:port" (%v)`, strings.Join(addrs, ","), err)
}
// Check if conflicting flags are passed.
nSet := 0
for _, v := range []bool{cfg.Durl != "", cfg.InitialCluster != "", cfg.DNSCluster != ""} {
if v {
nSet++
}
}
if cfg.ClusterState != ClusterStateFlagNew && cfg.ClusterState != ClusterStateFlagExisting {
return fmt.Errorf("unexpected clusterState %q", cfg.ClusterState)
}
if nSet > 1 {
return ErrConflictBootstrapFlags
}
if cfg.TickMs <= 0 {
return fmt.Errorf("--heartbeat-interval must be >0 (set to %dms)", cfg.TickMs)
}
if cfg.ElectionMs <= 0 {
return fmt.Errorf("--election-timeout must be >0 (set to %dms)", cfg.ElectionMs)
}
if 5*cfg.TickMs > cfg.ElectionMs {
return fmt.Errorf("--election-timeout[%vms] should be at least as 5 times as --heartbeat-interval[%vms]", cfg.ElectionMs, cfg.TickMs)
}
if cfg.ElectionMs > maxElectionMs {
return fmt.Errorf("--election-timeout[%vms] is too long, and should be set less than %vms", cfg.ElectionMs, maxElectionMs)
}
// check this last since proxying in etcdmain may make this OK
if cfg.LCUrls != nil && cfg.ACUrls == nil {
return ErrUnsetAdvertiseClientURLsFlag
}
switch cfg.AutoCompactionMode {
case "":
case CompactorModeRevision, CompactorModePeriodic:
default:
return fmt.Errorf("unknown auto-compaction-mode %q", cfg.AutoCompactionMode)
}
return nil
}
// PeerURLsMapAndToken sets up an initial peer URLsMap and cluster token for bootstrap or discovery.
func (cfg *Config) PeerURLsMapAndToken(which string) (urlsmap types.URLsMap, token string, err error) {
token = cfg.InitialClusterToken
switch {
case cfg.Durl != "":
urlsmap = types.URLsMap{}
// If using discovery, generate a temporary cluster based on
// self's advertised peer URLs
urlsmap[cfg.Name] = cfg.APUrls
token = cfg.Durl
case cfg.DNSCluster != "":
clusterStrs, cerr := cfg.GetDNSClusterNames()
lg := cfg.logger
if cerr != nil {
if lg != nil {
lg.Error("failed to resolve during SRV discovery", zap.Error(cerr))
} else {
plog.Errorf("couldn't resolve during SRV discovery (%v)", cerr)
}
return nil, "", cerr
}
for _, s := range clusterStrs {
if lg != nil {
lg.Info("got bootstrap from DNS for etcd-server", zap.String("node", s))
} else {
plog.Noticef("got bootstrap from DNS for etcd-server at %s", s)
}
}
clusterStr := strings.Join(clusterStrs, ",")
if strings.Contains(clusterStr, "https://") && cfg.PeerTLSInfo.TrustedCAFile == "" {
cfg.PeerTLSInfo.ServerName = cfg.DNSCluster
}
urlsmap, err = types.NewURLsMap(clusterStr)
// only etcd member must belong to the discovered cluster.
// proxy does not need to belong to the discovered cluster.
if which == "etcd" {
if _, ok := urlsmap[cfg.Name]; !ok {
return nil, "", fmt.Errorf("cannot find local etcd member %q in SRV records", cfg.Name)
}
}
default:
// We're statically configured, and cluster has appropriately been set.
urlsmap, err = types.NewURLsMap(cfg.InitialCluster)
}
return urlsmap, token, err
}
// GetDNSClusterNames uses DNS SRV records to get a list of initial nodes for cluster bootstrapping.
func (cfg *Config) GetDNSClusterNames() ([]string, error) {
var (
clusterStrs []string
cerr error
serviceNameSuffix string
)
if cfg.DNSClusterServiceName != "" {
serviceNameSuffix = "-" + cfg.DNSClusterServiceName
}
// Use both etcd-server-ssl and etcd-server for discovery. Combine the results if both are available.
clusterStrs, cerr = srv.GetCluster("https", "etcd-server-ssl"+serviceNameSuffix, cfg.Name, cfg.DNSCluster, cfg.APUrls)
defaultHTTPClusterStrs, httpCerr := srv.GetCluster("http", "etcd-server"+serviceNameSuffix, cfg.Name, cfg.DNSCluster, cfg.APUrls)
if cerr != nil {
clusterStrs = make([]string, 0)
}
if httpCerr != nil {
clusterStrs = append(clusterStrs, defaultHTTPClusterStrs...)
}
return clusterStrs, cerr
}
func (cfg Config) InitialClusterFromName(name string) (ret string) {
if len(cfg.APUrls) == 0 {
return ""
}
n := name
if name == "" {
n = DefaultName
}
for i := range cfg.APUrls {
ret = ret + "," + n + "=" + cfg.APUrls[i].String()
}
return ret[1:]
}
func (cfg Config) IsNewCluster() bool { return cfg.ClusterState == ClusterStateFlagNew }
func (cfg Config) ElectionTicks() int { return int(cfg.ElectionMs / cfg.TickMs) }
func (cfg Config) defaultPeerHost() bool {
return len(cfg.APUrls) == 1 && cfg.APUrls[0].String() == DefaultInitialAdvertisePeerURLs
}
2016-09-13 05:49:22 +03:00
func (cfg Config) defaultClientHost() bool {
return len(cfg.ACUrls) == 1 && cfg.ACUrls[0].String() == DefaultAdvertiseClientURLs
}
func (cfg *Config) ClientSelfCert() (err error) {
if cfg.ClientAutoTLS && cfg.ClientTLSInfo.Empty() {
chosts := make([]string, len(cfg.LCUrls))
for i, u := range cfg.LCUrls {
chosts[i] = u.Host
}
cfg.ClientTLSInfo, err = transport.SelfCert(cfg.logger, filepath.Join(cfg.Dir, "fixtures", "client"), chosts)
return err
} else if cfg.ClientAutoTLS {
if cfg.logger != nil {
cfg.logger.Warn("ignoring client auto TLS since certs given")
} else {
plog.Warningf("ignoring client auto TLS since certs given")
}
}
return nil
}
func (cfg *Config) PeerSelfCert() (err error) {
if cfg.PeerAutoTLS && cfg.PeerTLSInfo.Empty() {
phosts := make([]string, len(cfg.LPUrls))
for i, u := range cfg.LPUrls {
phosts[i] = u.Host
}
cfg.PeerTLSInfo, err = transport.SelfCert(cfg.logger, filepath.Join(cfg.Dir, "fixtures", "peer"), phosts)
return err
} else if cfg.PeerAutoTLS {
if cfg.logger != nil {
cfg.logger.Warn("ignoring peer auto TLS since certs given")
} else {
plog.Warningf("ignoring peer auto TLS since certs given")
}
}
return nil
}
// UpdateDefaultClusterFromName updates cluster advertise URLs with, if available, default host,
// if advertise URLs are default values(localhost:2379,2380) AND if listen URL is 0.0.0.0.
// e.g. advertise peer URL localhost:2380 or listen peer URL 0.0.0.0:2380
// then the advertise peer host would be updated with machine's default host,
// while keeping the listen URL's port.
// User can work around this by explicitly setting URL with 127.0.0.1.
// It returns the default hostname, if used, and the error, if any, from getting the machine's default host.
// TODO: check whether fields are set instead of whether fields have default value
func (cfg *Config) UpdateDefaultClusterFromName(defaultInitialCluster string) (string, error) {
if defaultHostname == "" || defaultHostStatus != nil {
// update 'initial-cluster' when only the name is specified (e.g. 'etcd --name=abc')
if cfg.Name != DefaultName && cfg.InitialCluster == defaultInitialCluster {
cfg.InitialCluster = cfg.InitialClusterFromName(cfg.Name)
}
return "", defaultHostStatus
}
used := false
pip, pport := cfg.LPUrls[0].Hostname(), cfg.LPUrls[0].Port()
if cfg.defaultPeerHost() && pip == "0.0.0.0" {
cfg.APUrls[0] = url.URL{Scheme: cfg.APUrls[0].Scheme, Host: fmt.Sprintf("%s:%s", defaultHostname, pport)}
used = true
}
// update 'initial-cluster' when only the name is specified (e.g. 'etcd --name=abc')
if cfg.Name != DefaultName && cfg.InitialCluster == defaultInitialCluster {
cfg.InitialCluster = cfg.InitialClusterFromName(cfg.Name)
}
cip, cport := cfg.LCUrls[0].Hostname(), cfg.LCUrls[0].Port()
if cfg.defaultClientHost() && cip == "0.0.0.0" {
cfg.ACUrls[0] = url.URL{Scheme: cfg.ACUrls[0].Scheme, Host: fmt.Sprintf("%s:%s", defaultHostname, cport)}
used = true
}
dhost := defaultHostname
if !used {
dhost = ""
}
return dhost, defaultHostStatus
}
2016-09-13 05:49:22 +03:00
// checkBindURLs returns an error if any URL uses a domain name.
func checkBindURLs(urls []url.URL) error {
for _, url := range urls {
if url.Scheme == "unix" || url.Scheme == "unixs" {
continue
}
host, _, err := net.SplitHostPort(url.Host)
if err != nil {
return err
}
if host == "localhost" {
// special case for local address
// TODO: support /etc/hosts ?
continue
}
if net.ParseIP(host) == nil {
return fmt.Errorf("expected IP in URL for binding (%s)", url.String())
2016-09-13 05:49:22 +03:00
}
}
return nil
}
func checkHostURLs(urls []url.URL) error {
for _, url := range urls {
host, _, err := net.SplitHostPort(url.Host)
if err != nil {
return err
}
if host == "" {
return fmt.Errorf("unexpected empty host (%s)", url.String())
}
}
return nil
}