embed

Documentation

Overview

Package embed provides bindings for embedding an etcd server in a program.

Launch an embedded etcd server using the configuration defaults:

import (
	"log"
	"time"

	"go.etcd.io/etcd/embed"
)

func main() {
	cfg := embed.NewConfig()
	cfg.Dir = "default.etcd"
	e, err := embed.StartEtcd(cfg)
	if err != nil {
		log.Fatal(err)
	}
	defer e.Close()
	select {
	case <-e.Server.ReadyNotify():
		log.Printf("Server is ready!")
	case <-time.After(60 * time.Second):
		e.Server.Stop() // trigger a shutdown
		log.Printf("Server took too long to start!")
	}
	log.Fatal(<-e.Err())
}

Index

• Constants
• Variables
• func NewZapCoreLoggerBuilder(lg *zap.Logger, cr zapcore.Core, syncer zapcore.WriteSyncer) func(*Config) error
• func WrapCORS(cors map[string]struct{}, h http.Handler) http.Handler
• type Config
  • func ConfigFromFile(path string) (*Config, error)
  • func NewConfig() *Config
  • func (cfg *Config) ClientSelfCert() (err error)
  • func (cfg Config) ElectionTicks() int
  • func (cfg *Config) GetDNSClusterNames() ([]string, error)
  • func (cfg Config) GetLogger() *zap.Logger
  • func (cfg Config) InitialClusterFromName(name string) (ret string)
  • func (cfg Config) IsNewCluster() bool
  • func (cfg *Config) PeerSelfCert() (err error)
  • func (cfg *Config) PeerURLsMapAndToken(which string) (urlsmap types.URLsMap, token string, err error)
  • func (cfg *Config) UpdateDefaultClusterFromName(defaultInitialCluster string) (string, error)
  • func (cfg *Config) Validate() error
• type Etcd
  • func StartEtcd(inCfg *Config) (e *Etcd, err error)
  • func (e *Etcd) Close()
  • func (e *Etcd) Config() Config
  • func (e *Etcd) Err() <-chan error
  • func (e *Etcd) GetLogger() *zap.Logger

Constants

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"
	JournalLogOutput = "systemd/journal"
	StdErrLogOutput  = "stderr"
	StdOutLogOutput  = "stdout"

	// 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 API is disabled by default.
	DefaultEnableV2 = false
)

Variables

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

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 = v3compactor.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 = v3compactor.ModeRevision
)

Functions

func NewZapCoreLoggerBuilder

func NewZapCoreLoggerBuilder(lg *zap.Logger, cr zapcore.Core, syncer zapcore.WriteSyncer) func(*Config) error

NewZapCoreLoggerBuilder generates a zap core logger builder.

func WrapCORS

func WrapCORS(cors map[string]struct{}, h http.Handler) http.Handler

WrapCORS wraps existing handler with CORS. TODO: deprecate this after v2 proxy deprecate

Types

type Config

type Config struct {
	Name   string `json:"name"`
	Dir    string `json:"data-dir"`
	WalDir string `json:"wal-dir"`

	SnapshotCount uint64 `json:"snapshot-count"`

	// SnapshotCatchUpEntries is the number of entries for a slow follower
	// to catch-up after compacting the raft storage entries.
	// We expect the follower has a millisecond level latency with the leader.
	// The max throughput is around 10K. Keep a 5K entries is enough for helping
	// follower to catch up.
	// WARNING: only change this for tests.
	// Always use "DefaultSnapshotCatchUpEntries"
	SnapshotCatchUpEntries uint64

	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"`

	// InitialElectionTickAdvance is true, then local member fast-forwards
	// election ticks to speed up "initial" leader election trigger. This
	// benefits the case of larger election ticks. For instance, cross
	// datacenter deployment may require longer election timeout of 10-second.
	// If true, local node does not need wait up to 10-second. Instead,
	// forwards its election ticks to 8-second, and have only 2-second left
	// before leader election.
	//
	// Major assumptions are that:
	//  - cluster has no active leader thus advancing ticks enables faster
	//    leader election, or
	//  - cluster already has an established leader, and rejoining follower
	//    is likely to receive heartbeats from the leader after tick advance
	//    and before election timeout.
	//
	// However, when network from leader to rejoining follower is congested,
	// and the follower does not receive leader heartbeat within left election
	// ticks, disruptive election has to happen thus affecting cluster
	// availabilities.
	//
	// Disabling this would slow down initial bootstrap process for cross
	// datacenter deployments. Make your own tradeoffs by configuring
	// --initial-election-tick-advance at the cost of slow initial bootstrap.
	//
	// If single-node, it advances ticks regardless.
	//
	// See https://github.com/etcd-io/etcd/issues/9333 for more detail.
	InitialElectionTickAdvance bool `json:"initial-election-tick-advance"`

	// BackendBatchInterval is the maximum time before commit the backend transaction.
	BackendBatchInterval time.Duration `json:"backend-batch-interval"`
	// BackendBatchLimit is the maximum operations before commit the backend transaction.
	BackendBatchLimit int `json:"backend-batch-limit"`
	// BackendFreelistType specifies the type of freelist that boltdb backend uses (array and map are supported types).
	BackendFreelistType string `json:"backend-bbolt-freelist-type"`
	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

	// CipherSuites is a list of supported TLS cipher suites between
	// client/server and peers. If empty, Go auto-populates the list.
	// Note that cipher suites are prioritized in the given order.
	CipherSuites []string `json:"cipher-suites"`

	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/etcd-io/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:"-"`

	AuthToken  string `json:"auth-token"`
	BcryptCost uint   `json:"bcrypt-cost"`

	ExperimentalInitialCorruptCheck bool          `json:"experimental-initial-corrupt-check"`
	ExperimentalCorruptCheckTime    time.Duration `json:"experimental-corrupt-check-time"`
	ExperimentalEnableV2V3          string        `json:"experimental-enable-v2v3"`
	// ExperimentalEnableLeaseCheckpoint enables primary lessor to persist lease remainingTTL to prevent indefinite auto-renewal of long lived leases.
	ExperimentalEnableLeaseCheckpoint bool `json:"experimental-enable-lease-checkpoint"`
	ExperimentalCompactionBatchLimit  int  `json:"experimental-compaction-batch-limit"`

	// 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 is logger options: "zap", "capnslog".
	// WARN: "capnslog" is being deprecated in v3.5.
	Logger string `json:"logger"`
	// LogLevel configures log level. Only supports debug, info, warn, error, panic, or fatal. Default 'info'.
	LogLevel string `json:"log-level"`
	// LogOutputs is either:
	//  - "default" as os.Stderr,
	//  - "stderr" as os.Stderr,
	//  - "stdout" as os.Stdout,
	//  - file path to append server logs to.
	// It can be multiple when "Logger" is zap.
	LogOutputs []string `json:"log-outputs"`

	// ZapLoggerBuilder is used to build the zap logger.
	ZapLoggerBuilder func(*Config) error

	// EnableGRPCGateway is false to disable grpc gateway.
	EnableGRPCGateway bool `json:"enable-grpc-gateway"`

	// DeprecatedLogOutput is to be deprecated in v3.5.
	// Just here for safe migration in v3.4.
	DeprecatedLogOutput []string `json:"log-output"`
	// Debug is true, to enable debug level logging.
	// WARNING: to be deprecated in 3.5. Use "--log-level=debug" instead.
	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"`
	// contains filtered or unexported fields
}

Config holds the arguments for configuring an etcd server.

func ConfigFromFile

func ConfigFromFile(path string) (*Config, error)

func NewConfig

func NewConfig() *Config

NewConfig creates a new Config populated with default values.

func (*Config) ClientSelfCert

func (cfg *Config) ClientSelfCert() (err error)

func (Config) ElectionTicks

func (cfg Config) ElectionTicks() int

func (*Config) GetDNSClusterNames

func (cfg *Config) GetDNSClusterNames() ([]string, error)

GetDNSClusterNames uses DNS SRV records to get a list of initial nodes for cluster bootstrapping.

func (Config) GetLogger

func (cfg Config) GetLogger() *zap.Logger

GetLogger returns the logger.

func (Config) InitialClusterFromName

func (cfg Config) InitialClusterFromName(name string) (ret string)

func (Config) IsNewCluster

func (cfg Config) IsNewCluster() bool

func (*Config) PeerSelfCert

func (cfg *Config) PeerSelfCert() (err error)

func (*Config) PeerURLsMapAndToken

func (cfg *Config) PeerURLsMapAndToken(which string) (urlsmap types.URLsMap, token string, err error)

PeerURLsMapAndToken sets up an initial peer URLsMap and cluster token for bootstrap or discovery.

func (*Config) UpdateDefaultClusterFromName

func (cfg *Config) UpdateDefaultClusterFromName(defaultInitialCluster string) (string, error)

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 (*Config) Validate

func (cfg *Config) Validate() error

Validate ensures that '*embed.Config' fields are properly configured.

type Etcd

type Etcd struct {
	// 此Peers和raftNode中的peers不是一码事，是ETCD启动时参数指定的。
	// Listener的数组，一般一个就够用了（我觉得）
	Peers []*peerListener

	// 向etcd集群发送HTTP请求的客户端
	// 类似于Peers，Clients是netListener，监听client端的网络请求
	Clients []net.Listener

	Server *etcdserver.EtcdServer // Etcd核心服务
	// contains filtered or unexported fields
}

Etcd contains a running etcd server and its listeners. Etcd结构体

func StartEtcd

func StartEtcd(inCfg *Config) (e *Etcd, err error)

StartEtcd . |-embed.StartEtcd() —> embed/etcd.go | |-configurePeerListeners() peer网络监听，服务于raft | |-configureClientListeners() client网络监听，服务于client | |-EtcdServer.ServerConfig() 生成新的配置 | | | |-EtcdServer.NewServer() etcdserver/server.go 正式启动RAFT服务<<<1>>> | | | |-EtcdServer.Start() -> etcdserver/server.go 开始启动服务 | | |-EtcdServer.start() -> | | |-wait.New() 新建WaitGroup组以及一些管道服务 | | | | |-EtcdServer.run() etcdserver/raft.go 启动应用层的处理协程<<<2>>> | | | |-Etcd.servePeers() 启动集群内部通讯 | | 好的 | | |-v3rpc.Server() 启动gRPC服务 api/v3rpc/grpc.go，这里真正监听了frpc请求 | | |-grpc.NewServer() 调用gRPC的接口创建 | | |-pb.RegisterKVServer() 注册各种的服务，这里包含了多个 | | |-pb.RegisterWatchServer() | | | |-Etcd.serveClients() 启动协程处理客户请求

func (*Etcd) Close

func (e *Etcd) Close()

Close gracefully shuts down all servers/listeners. Client requests will be terminated with request timeout. After timeout, enforce remaning requests be closed immediately.

func (*Etcd) Config

func (e *Etcd) Config() Config

Config returns the current configuration.

func (*Etcd) Err

func (e *Etcd) Err() <-chan error

func (*Etcd) GetLogger

func (e *Etcd) GetLogger() *zap.Logger

GetLogger returns the logger.