bump(github.com/coreos/go-etcd): 526d936ffe75284ca80290ea6386f883f573c232
parent
72514f8ab2
commit
40021ab72e
2
etcd.go
2
etcd.go
|
@ -26,8 +26,8 @@ import (
|
||||||
|
|
||||||
"github.com/coreos/etcd/third_party/github.com/coreos/raft"
|
"github.com/coreos/etcd/third_party/github.com/coreos/raft"
|
||||||
|
|
||||||
ehttp "github.com/coreos/etcd/http"
|
|
||||||
"github.com/coreos/etcd/config"
|
"github.com/coreos/etcd/config"
|
||||||
|
ehttp "github.com/coreos/etcd/http"
|
||||||
"github.com/coreos/etcd/log"
|
"github.com/coreos/etcd/log"
|
||||||
"github.com/coreos/etcd/metrics"
|
"github.com/coreos/etcd/metrics"
|
||||||
"github.com/coreos/etcd/server"
|
"github.com/coreos/etcd/server"
|
||||||
|
|
|
@ -12,15 +12,9 @@ import (
|
||||||
"net/url"
|
"net/url"
|
||||||
"os"
|
"os"
|
||||||
"path"
|
"path"
|
||||||
"strings"
|
|
||||||
"time"
|
"time"
|
||||||
)
|
)
|
||||||
|
|
||||||
const (
|
|
||||||
HTTP = iota
|
|
||||||
HTTPS
|
|
||||||
)
|
|
||||||
|
|
||||||
// See SetConsistency for how to use these constants.
|
// See SetConsistency for how to use these constants.
|
||||||
const (
|
const (
|
||||||
// Using strings rather than iota because the consistency level
|
// Using strings rather than iota because the consistency level
|
||||||
|
@ -34,45 +28,27 @@ const (
|
||||||
defaultBufferSize = 10
|
defaultBufferSize = 10
|
||||||
)
|
)
|
||||||
|
|
||||||
type Cluster struct {
|
|
||||||
Leader string `json:"leader"`
|
|
||||||
Machines []string `json:"machines"`
|
|
||||||
}
|
|
||||||
|
|
||||||
type Config struct {
|
type Config struct {
|
||||||
CertFile string `json:"certFile"`
|
CertFile string `json:"certFile"`
|
||||||
KeyFile string `json:"keyFile"`
|
KeyFile string `json:"keyFile"`
|
||||||
CaCertFile string `json:"caCertFile"`
|
CaCertFile []string `json:"caCertFiles"`
|
||||||
Scheme string `json:"scheme"`
|
|
||||||
Timeout time.Duration `json:"timeout"`
|
Timeout time.Duration `json:"timeout"`
|
||||||
Consistency string `json: "consistency"`
|
Consistency string `json: "consistency"`
|
||||||
}
|
}
|
||||||
|
|
||||||
type Client struct {
|
type Client struct {
|
||||||
cluster Cluster `json:"cluster"`
|
|
||||||
config Config `json:"config"`
|
config Config `json:"config"`
|
||||||
|
cluster *Cluster `json:"cluster"`
|
||||||
httpClient *http.Client
|
httpClient *http.Client
|
||||||
persistence io.Writer
|
persistence io.Writer
|
||||||
cURLch chan string
|
cURLch chan string
|
||||||
|
keyPrefix string
|
||||||
}
|
}
|
||||||
|
|
||||||
// NewClient create a basic client that is configured to be used
|
// NewClient create a basic client that is configured to be used
|
||||||
// with the given machine list.
|
// with the given machine list.
|
||||||
func NewClient(machines []string) *Client {
|
func NewClient(machines []string) *Client {
|
||||||
// if an empty slice was sent in then just assume localhost
|
|
||||||
if len(machines) == 0 {
|
|
||||||
machines = []string{"http://127.0.0.1:4001"}
|
|
||||||
}
|
|
||||||
|
|
||||||
// default leader and machines
|
|
||||||
cluster := Cluster{
|
|
||||||
Leader: machines[0],
|
|
||||||
Machines: machines,
|
|
||||||
}
|
|
||||||
|
|
||||||
config := Config{
|
config := Config{
|
||||||
// default use http
|
|
||||||
Scheme: "http",
|
|
||||||
// default timeout is one second
|
// default timeout is one second
|
||||||
Timeout: time.Second,
|
Timeout: time.Second,
|
||||||
// default consistency level is STRONG
|
// default consistency level is STRONG
|
||||||
|
@ -80,75 +56,146 @@ func NewClient(machines []string) *Client {
|
||||||
}
|
}
|
||||||
|
|
||||||
client := &Client{
|
client := &Client{
|
||||||
cluster: cluster,
|
cluster: NewCluster(machines),
|
||||||
config: config,
|
config: config,
|
||||||
|
keyPrefix: path.Join(version, "keys"),
|
||||||
}
|
}
|
||||||
|
|
||||||
err := setupHttpClient(client)
|
client.initHTTPClient()
|
||||||
if err != nil {
|
client.saveConfig()
|
||||||
panic(err)
|
|
||||||
}
|
|
||||||
|
|
||||||
return client
|
return client
|
||||||
}
|
}
|
||||||
|
|
||||||
// NewClientFile creates a client from a given file path.
|
// NewTLSClient create a basic client with TLS configuration
|
||||||
|
func NewTLSClient(machines []string, cert, key, caCert string) (*Client, error) {
|
||||||
|
// overwrite the default machine to use https
|
||||||
|
if len(machines) == 0 {
|
||||||
|
machines = []string{"https://127.0.0.1:4001"}
|
||||||
|
}
|
||||||
|
|
||||||
|
config := Config{
|
||||||
|
// default timeout is one second
|
||||||
|
Timeout: time.Second,
|
||||||
|
// default consistency level is STRONG
|
||||||
|
Consistency: STRONG_CONSISTENCY,
|
||||||
|
CertFile: cert,
|
||||||
|
KeyFile: key,
|
||||||
|
CaCertFile: make([]string, 0),
|
||||||
|
}
|
||||||
|
|
||||||
|
client := &Client{
|
||||||
|
cluster: NewCluster(machines),
|
||||||
|
config: config,
|
||||||
|
keyPrefix: path.Join(version, "keys"),
|
||||||
|
}
|
||||||
|
|
||||||
|
err := client.initHTTPSClient(cert, key)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
|
||||||
|
err = client.AddRootCA(caCert)
|
||||||
|
|
||||||
|
client.saveConfig()
|
||||||
|
|
||||||
|
return client, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// NewClientFromFile creates a client from a given file path.
|
||||||
// The given file is expected to use the JSON format.
|
// The given file is expected to use the JSON format.
|
||||||
func NewClientFile(fpath string) (*Client, error) {
|
func NewClientFromFile(fpath string) (*Client, error) {
|
||||||
fi, err := os.Open(fpath)
|
fi, err := os.Open(fpath)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
defer func() {
|
defer func() {
|
||||||
if err := fi.Close(); err != nil {
|
if err := fi.Close(); err != nil {
|
||||||
panic(err)
|
panic(err)
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
|
|
||||||
return NewClientReader(fi)
|
return NewClientFromReader(fi)
|
||||||
}
|
}
|
||||||
|
|
||||||
// NewClientReader creates a Client configured from a given reader.
|
// NewClientFromReader creates a Client configured from a given reader.
|
||||||
// The config is expected to use the JSON format.
|
// The configuration is expected to use the JSON format.
|
||||||
func NewClientReader(reader io.Reader) (*Client, error) {
|
func NewClientFromReader(reader io.Reader) (*Client, error) {
|
||||||
var client Client
|
c := new(Client)
|
||||||
|
|
||||||
b, err := ioutil.ReadAll(reader)
|
b, err := ioutil.ReadAll(reader)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
err = json.Unmarshal(b, &client)
|
err = json.Unmarshal(b, c)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
if c.config.CertFile == "" {
|
||||||
|
c.initHTTPClient()
|
||||||
|
} else {
|
||||||
|
err = c.initHTTPSClient(c.config.CertFile, c.config.KeyFile)
|
||||||
|
}
|
||||||
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
err = setupHttpClient(&client)
|
for _, caCert := range c.config.CaCertFile {
|
||||||
if err != nil {
|
if err := c.AddRootCA(caCert); err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
|
||||||
return &client, nil
|
return c, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func setupHttpClient(client *Client) error {
|
// Override the Client's HTTP Transport object
|
||||||
if client.config.CertFile != "" && client.config.KeyFile != "" {
|
func (c *Client) SetTransport(tr *http.Transport) {
|
||||||
err := client.SetCertAndKey(client.config.CertFile, client.config.KeyFile, client.config.CaCertFile)
|
c.httpClient.Transport = tr
|
||||||
if err != nil {
|
}
|
||||||
return err
|
|
||||||
}
|
// SetKeyPrefix changes the key prefix from the default `/v2/keys` to whatever
|
||||||
} else {
|
// is set.
|
||||||
client.config.CertFile = ""
|
func (c *Client) SetKeyPrefix(prefix string) {
|
||||||
client.config.KeyFile = ""
|
c.keyPrefix = prefix
|
||||||
|
}
|
||||||
|
|
||||||
|
// initHTTPClient initializes a HTTP client for etcd client
|
||||||
|
func (c *Client) initHTTPClient() {
|
||||||
tr := &http.Transport{
|
tr := &http.Transport{
|
||||||
Dial: dialTimeout,
|
Dial: dialTimeout,
|
||||||
TLSClientConfig: &tls.Config{
|
TLSClientConfig: &tls.Config{
|
||||||
InsecureSkipVerify: true,
|
InsecureSkipVerify: true,
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
client.httpClient = &http.Client{Transport: tr}
|
c.httpClient = &http.Client{Transport: tr}
|
||||||
|
}
|
||||||
|
|
||||||
|
// initHTTPClient initializes a HTTPS client for etcd client
|
||||||
|
func (c *Client) initHTTPSClient(cert, key string) error {
|
||||||
|
if cert == "" || key == "" {
|
||||||
|
return errors.New("Require both cert and key path")
|
||||||
}
|
}
|
||||||
|
|
||||||
|
tlsCert, err := tls.LoadX509KeyPair(cert, key)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
tlsConfig := &tls.Config{
|
||||||
|
Certificates: []tls.Certificate{tlsCert},
|
||||||
|
InsecureSkipVerify: true,
|
||||||
|
}
|
||||||
|
|
||||||
|
tr := &http.Transport{
|
||||||
|
TLSClientConfig: tlsConfig,
|
||||||
|
Dial: dialTimeout,
|
||||||
|
}
|
||||||
|
|
||||||
|
c.httpClient = &http.Client{Transport: tr}
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -179,114 +226,45 @@ func (c *Client) SetConsistency(consistency string) error {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// MarshalJSON implements the Marshaller interface
|
// AddRootCA adds a root CA cert for the etcd client
|
||||||
// as defined by the standard JSON package.
|
func (c *Client) AddRootCA(caCert string) error {
|
||||||
func (c *Client) MarshalJSON() ([]byte, error) {
|
if c.httpClient == nil {
|
||||||
b, err := json.Marshal(struct {
|
return errors.New("Client has not been initialized yet!")
|
||||||
Config Config `json:"config"`
|
|
||||||
Cluster Cluster `json:"cluster"`
|
|
||||||
}{
|
|
||||||
Config: c.config,
|
|
||||||
Cluster: c.cluster,
|
|
||||||
})
|
|
||||||
|
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
}
|
||||||
|
|
||||||
return b, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// UnmarshalJSON implements the Unmarshaller interface
|
|
||||||
// as defined by the standard JSON package.
|
|
||||||
func (c *Client) UnmarshalJSON(b []byte) error {
|
|
||||||
temp := struct {
|
|
||||||
Config Config `json: "config"`
|
|
||||||
Cluster Cluster `json: "cluster"`
|
|
||||||
}{}
|
|
||||||
err := json.Unmarshal(b, &temp)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
c.cluster = temp.Cluster
|
|
||||||
c.config = temp.Config
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// saveConfig saves the current config using c.persistence.
|
|
||||||
func (c *Client) saveConfig() error {
|
|
||||||
if c.persistence != nil {
|
|
||||||
b, err := json.Marshal(c)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
_, err = c.persistence.Write(b)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func (c *Client) SetCertAndKey(cert string, key string, caCert string) error {
|
|
||||||
if cert != "" && key != "" {
|
|
||||||
tlsCert, err := tls.LoadX509KeyPair(cert, key)
|
|
||||||
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
tlsConfig := &tls.Config{
|
|
||||||
Certificates: []tls.Certificate{tlsCert},
|
|
||||||
}
|
|
||||||
|
|
||||||
if caCert != "" {
|
|
||||||
caCertPool := x509.NewCertPool()
|
|
||||||
|
|
||||||
certBytes, err := ioutil.ReadFile(caCert)
|
certBytes, err := ioutil.ReadFile(caCert)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
if !caCertPool.AppendCertsFromPEM(certBytes) {
|
tr, ok := c.httpClient.Transport.(*http.Transport)
|
||||||
return errors.New("Unable to load caCert")
|
|
||||||
|
if !ok {
|
||||||
|
panic("AddRootCA(): Transport type assert should not fail")
|
||||||
}
|
}
|
||||||
|
|
||||||
tlsConfig.RootCAs = caCertPool
|
if tr.TLSClientConfig.RootCAs == nil {
|
||||||
|
caCertPool := x509.NewCertPool()
|
||||||
|
ok = caCertPool.AppendCertsFromPEM(certBytes)
|
||||||
|
if ok {
|
||||||
|
tr.TLSClientConfig.RootCAs = caCertPool
|
||||||
|
}
|
||||||
|
tr.TLSClientConfig.InsecureSkipVerify = false
|
||||||
} else {
|
} else {
|
||||||
tlsConfig.InsecureSkipVerify = true
|
ok = tr.TLSClientConfig.RootCAs.AppendCertsFromPEM(certBytes)
|
||||||
}
|
}
|
||||||
|
|
||||||
tr := &http.Transport{
|
if !ok {
|
||||||
TLSClientConfig: tlsConfig,
|
err = errors.New("Unable to load caCert")
|
||||||
Dial: dialTimeout,
|
|
||||||
}
|
}
|
||||||
|
|
||||||
c.httpClient = &http.Client{Transport: tr}
|
c.config.CaCertFile = append(c.config.CaCertFile, caCert)
|
||||||
c.saveConfig()
|
c.saveConfig()
|
||||||
return nil
|
|
||||||
}
|
return err
|
||||||
return errors.New("Require both cert and key path")
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func (c *Client) SetScheme(scheme int) error {
|
// SetCluster updates cluster information using the given machine list.
|
||||||
if scheme == HTTP {
|
|
||||||
c.config.Scheme = "http"
|
|
||||||
c.saveConfig()
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
if scheme == HTTPS {
|
|
||||||
c.config.Scheme = "https"
|
|
||||||
c.saveConfig()
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
return errors.New("Unknown Scheme")
|
|
||||||
}
|
|
||||||
|
|
||||||
// SetCluster updates config using the given machine list.
|
|
||||||
func (c *Client) SetCluster(machines []string) bool {
|
func (c *Client) SetCluster(machines []string) bool {
|
||||||
success := c.internalSyncCluster(machines)
|
success := c.internalSyncCluster(machines)
|
||||||
return success
|
return success
|
||||||
|
@ -296,16 +274,15 @@ func (c *Client) GetCluster() []string {
|
||||||
return c.cluster.Machines
|
return c.cluster.Machines
|
||||||
}
|
}
|
||||||
|
|
||||||
// SyncCluster updates config using the internal machine list.
|
// SyncCluster updates the cluster information using the internal machine list.
|
||||||
func (c *Client) SyncCluster() bool {
|
func (c *Client) SyncCluster() bool {
|
||||||
success := c.internalSyncCluster(c.cluster.Machines)
|
return c.internalSyncCluster(c.cluster.Machines)
|
||||||
return success
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// internalSyncCluster syncs cluster information using the given machine list.
|
// internalSyncCluster syncs cluster information using the given machine list.
|
||||||
func (c *Client) internalSyncCluster(machines []string) bool {
|
func (c *Client) internalSyncCluster(machines []string) bool {
|
||||||
for _, machine := range machines {
|
for _, machine := range machines {
|
||||||
httpPath := c.createHttpPath(machine, version+"/machines")
|
httpPath := c.createHttpPath(machine, path.Join(version, "machines"))
|
||||||
resp, err := c.httpClient.Get(httpPath)
|
resp, err := c.httpClient.Get(httpPath)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
// try another machine in the cluster
|
// try another machine in the cluster
|
||||||
|
@ -319,12 +296,11 @@ func (c *Client) internalSyncCluster(machines []string) bool {
|
||||||
}
|
}
|
||||||
|
|
||||||
// update Machines List
|
// update Machines List
|
||||||
c.cluster.Machines = strings.Split(string(b), ", ")
|
c.cluster.updateFromStr(string(b))
|
||||||
|
|
||||||
// update leader
|
// update leader
|
||||||
// the first one in the machine list is the leader
|
// the first one in the machine list is the leader
|
||||||
logger.Debugf("update.leader[%s,%s]", c.cluster.Leader, c.cluster.Machines[0])
|
c.cluster.switchLeader(0)
|
||||||
c.cluster.Leader = c.cluster.Machines[0]
|
|
||||||
|
|
||||||
logger.Debug("sync.machines ", c.cluster.Machines)
|
logger.Debug("sync.machines ", c.cluster.Machines)
|
||||||
c.saveConfig()
|
c.saveConfig()
|
||||||
|
@ -337,8 +313,12 @@ func (c *Client) internalSyncCluster(machines []string) bool {
|
||||||
// createHttpPath creates a complete HTTP URL.
|
// createHttpPath creates a complete HTTP URL.
|
||||||
// serverName should contain both the host name and a port number, if any.
|
// serverName should contain both the host name and a port number, if any.
|
||||||
func (c *Client) createHttpPath(serverName string, _path string) string {
|
func (c *Client) createHttpPath(serverName string, _path string) string {
|
||||||
u, _ := url.Parse(serverName)
|
u, err := url.Parse(serverName)
|
||||||
u.Path = path.Join(u.Path, "/", _path)
|
if err != nil {
|
||||||
|
panic(err)
|
||||||
|
}
|
||||||
|
|
||||||
|
u.Path = path.Join(u.Path, _path)
|
||||||
|
|
||||||
if u.Scheme == "" {
|
if u.Scheme == "" {
|
||||||
u.Scheme = "http"
|
u.Scheme = "http"
|
||||||
|
@ -351,27 +331,6 @@ func dialTimeout(network, addr string) (net.Conn, error) {
|
||||||
return net.DialTimeout(network, addr, time.Second)
|
return net.DialTimeout(network, addr, time.Second)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (c *Client) updateLeader(u *url.URL) {
|
|
||||||
var leader string
|
|
||||||
if u.Scheme == "" {
|
|
||||||
leader = "http://" + u.Host
|
|
||||||
} else {
|
|
||||||
leader = u.Scheme + "://" + u.Host
|
|
||||||
}
|
|
||||||
|
|
||||||
logger.Debugf("update.leader[%s,%s]", c.cluster.Leader, leader)
|
|
||||||
c.cluster.Leader = leader
|
|
||||||
c.saveConfig()
|
|
||||||
}
|
|
||||||
|
|
||||||
// switchLeader switch the current leader to machines[num]
|
|
||||||
func (c *Client) switchLeader(num int) {
|
|
||||||
logger.Debugf("switch.leader[from %v to %v]",
|
|
||||||
c.cluster.Leader, c.cluster.Machines[num])
|
|
||||||
|
|
||||||
c.cluster.Leader = c.cluster.Machines[num]
|
|
||||||
}
|
|
||||||
|
|
||||||
func (c *Client) OpenCURL() {
|
func (c *Client) OpenCURL() {
|
||||||
c.cURLch = make(chan string, defaultBufferSize)
|
c.cURLch = make(chan string, defaultBufferSize)
|
||||||
}
|
}
|
||||||
|
@ -392,3 +351,55 @@ func (c *Client) sendCURL(command string) {
|
||||||
func (c *Client) RecvCURL() string {
|
func (c *Client) RecvCURL() string {
|
||||||
return <-c.cURLch
|
return <-c.cURLch
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// saveConfig saves the current config using c.persistence.
|
||||||
|
func (c *Client) saveConfig() error {
|
||||||
|
if c.persistence != nil {
|
||||||
|
b, err := json.Marshal(c)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
_, err = c.persistence.Write(b)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// MarshalJSON implements the Marshaller interface
|
||||||
|
// as defined by the standard JSON package.
|
||||||
|
func (c *Client) MarshalJSON() ([]byte, error) {
|
||||||
|
b, err := json.Marshal(struct {
|
||||||
|
Config Config `json:"config"`
|
||||||
|
Cluster *Cluster `json:"cluster"`
|
||||||
|
}{
|
||||||
|
Config: c.config,
|
||||||
|
Cluster: c.cluster,
|
||||||
|
})
|
||||||
|
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
|
||||||
|
return b, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// UnmarshalJSON implements the Unmarshaller interface
|
||||||
|
// as defined by the standard JSON package.
|
||||||
|
func (c *Client) UnmarshalJSON(b []byte) error {
|
||||||
|
temp := struct {
|
||||||
|
Config Config `json: "config"`
|
||||||
|
Cluster *Cluster `json: "cluster"`
|
||||||
|
}{}
|
||||||
|
err := json.Unmarshal(b, &temp)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
c.cluster = temp.Cluster
|
||||||
|
c.config = temp.Config
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
|
@ -14,7 +14,9 @@ import (
|
||||||
func TestSync(t *testing.T) {
|
func TestSync(t *testing.T) {
|
||||||
fmt.Println("Make sure there are three nodes at 0.0.0.0:4001-4003")
|
fmt.Println("Make sure there are three nodes at 0.0.0.0:4001-4003")
|
||||||
|
|
||||||
c := NewClient(nil)
|
// Explicit trailing slash to ensure this doesn't reproduce:
|
||||||
|
// https://github.com/coreos/go-etcd/issues/82
|
||||||
|
c := NewClient([]string{"http://127.0.0.1:4001/"})
|
||||||
|
|
||||||
success := c.SyncCluster()
|
success := c.SyncCluster()
|
||||||
if !success {
|
if !success {
|
||||||
|
@ -79,7 +81,7 @@ func TestPersistence(t *testing.T) {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
|
|
||||||
c2, err := NewClientFile("config.json")
|
c2, err := NewClientFromFile("config.json")
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
|
|
|
@ -0,0 +1,51 @@
|
||||||
|
package etcd
|
||||||
|
|
||||||
|
import (
|
||||||
|
"net/url"
|
||||||
|
"strings"
|
||||||
|
)
|
||||||
|
|
||||||
|
type Cluster struct {
|
||||||
|
Leader string `json:"leader"`
|
||||||
|
Machines []string `json:"machines"`
|
||||||
|
}
|
||||||
|
|
||||||
|
func NewCluster(machines []string) *Cluster {
|
||||||
|
// if an empty slice was sent in then just assume HTTP 4001 on localhost
|
||||||
|
if len(machines) == 0 {
|
||||||
|
machines = []string{"http://127.0.0.1:4001"}
|
||||||
|
}
|
||||||
|
|
||||||
|
// default leader and machines
|
||||||
|
return &Cluster{
|
||||||
|
Leader: machines[0],
|
||||||
|
Machines: machines,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// switchLeader switch the current leader to machines[num]
|
||||||
|
func (cl *Cluster) switchLeader(num int) {
|
||||||
|
logger.Debugf("switch.leader[from %v to %v]",
|
||||||
|
cl.Leader, cl.Machines[num])
|
||||||
|
|
||||||
|
cl.Leader = cl.Machines[num]
|
||||||
|
}
|
||||||
|
|
||||||
|
func (cl *Cluster) updateFromStr(machines string) {
|
||||||
|
cl.Machines = strings.Split(machines, ", ")
|
||||||
|
}
|
||||||
|
|
||||||
|
func (cl *Cluster) updateLeader(leader string) {
|
||||||
|
logger.Debugf("update.leader[%s,%s]", cl.Leader, leader)
|
||||||
|
cl.Leader = leader
|
||||||
|
}
|
||||||
|
|
||||||
|
func (cl *Cluster) updateLeaderFromURL(u *url.URL) {
|
||||||
|
var leader string
|
||||||
|
if u.Scheme == "" {
|
||||||
|
leader = "http://" + u.Host
|
||||||
|
} else {
|
||||||
|
leader = u.Scheme + "://" + u.Host
|
||||||
|
}
|
||||||
|
cl.updateLeader(leader)
|
||||||
|
}
|
|
@ -0,0 +1,34 @@
|
||||||
|
package etcd
|
||||||
|
|
||||||
|
import "fmt"
|
||||||
|
|
||||||
|
func (c *Client) CompareAndDelete(key string, prevValue string, prevIndex uint64) (*Response, error) {
|
||||||
|
raw, err := c.RawCompareAndDelete(key, prevValue, prevIndex)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
|
||||||
|
return raw.toResponse()
|
||||||
|
}
|
||||||
|
|
||||||
|
func (c *Client) RawCompareAndDelete(key string, prevValue string, prevIndex uint64) (*RawResponse, error) {
|
||||||
|
if prevValue == "" && prevIndex == 0 {
|
||||||
|
return nil, fmt.Errorf("You must give either prevValue or prevIndex.")
|
||||||
|
}
|
||||||
|
|
||||||
|
options := options{}
|
||||||
|
if prevValue != "" {
|
||||||
|
options["prevValue"] = prevValue
|
||||||
|
}
|
||||||
|
if prevIndex != 0 {
|
||||||
|
options["prevIndex"] = prevIndex
|
||||||
|
}
|
||||||
|
|
||||||
|
raw, err := c.delete(key, options)
|
||||||
|
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
|
||||||
|
return raw, err
|
||||||
|
}
|
|
@ -0,0 +1,46 @@
|
||||||
|
package etcd
|
||||||
|
|
||||||
|
import (
|
||||||
|
"testing"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestCompareAndDelete(t *testing.T) {
|
||||||
|
c := NewClient(nil)
|
||||||
|
defer func() {
|
||||||
|
c.Delete("foo", true)
|
||||||
|
}()
|
||||||
|
|
||||||
|
c.Set("foo", "bar", 5)
|
||||||
|
|
||||||
|
// This should succeed an correct prevValue
|
||||||
|
resp, err := c.CompareAndDelete("foo", "bar", 0)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if !(resp.PrevNode.Value == "bar" && resp.PrevNode.Key == "/foo" && resp.PrevNode.TTL == 5) {
|
||||||
|
t.Fatalf("CompareAndDelete 1 prevNode failed: %#v", resp)
|
||||||
|
}
|
||||||
|
|
||||||
|
resp, _ = c.Set("foo", "bar", 5)
|
||||||
|
// This should fail because it gives an incorrect prevValue
|
||||||
|
_, err = c.CompareAndDelete("foo", "xxx", 0)
|
||||||
|
if err == nil {
|
||||||
|
t.Fatalf("CompareAndDelete 2 should have failed. The response is: %#v", resp)
|
||||||
|
}
|
||||||
|
|
||||||
|
// This should succeed because it gives an correct prevIndex
|
||||||
|
resp, err = c.CompareAndDelete("foo", "", resp.Node.ModifiedIndex)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if !(resp.PrevNode.Value == "bar" && resp.PrevNode.Key == "/foo" && resp.PrevNode.TTL == 5) {
|
||||||
|
t.Fatalf("CompareAndSwap 3 prevNode failed: %#v", resp)
|
||||||
|
}
|
||||||
|
|
||||||
|
c.Set("foo", "bar", 5)
|
||||||
|
// This should fail because it gives an incorrect prevIndex
|
||||||
|
resp, err = c.CompareAndDelete("foo", "", 29817514)
|
||||||
|
if err == nil {
|
||||||
|
t.Fatalf("CompareAndDelete 4 should have failed. The response is: %#v", resp)
|
||||||
|
}
|
||||||
|
}
|
|
@ -2,7 +2,18 @@ package etcd
|
||||||
|
|
||||||
import "fmt"
|
import "fmt"
|
||||||
|
|
||||||
func (c *Client) CompareAndSwap(key string, value string, ttl uint64, prevValue string, prevIndex uint64) (*Response, error) {
|
func (c *Client) CompareAndSwap(key string, value string, ttl uint64,
|
||||||
|
prevValue string, prevIndex uint64) (*Response, error) {
|
||||||
|
raw, err := c.RawCompareAndSwap(key, value, ttl, prevValue, prevIndex)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
|
||||||
|
return raw.toResponse()
|
||||||
|
}
|
||||||
|
|
||||||
|
func (c *Client) RawCompareAndSwap(key string, value string, ttl uint64,
|
||||||
|
prevValue string, prevIndex uint64) (*RawResponse, error) {
|
||||||
if prevValue == "" && prevIndex == 0 {
|
if prevValue == "" && prevIndex == 0 {
|
||||||
return nil, fmt.Errorf("You must give either prevValue or prevIndex.")
|
return nil, fmt.Errorf("You must give either prevValue or prevIndex.")
|
||||||
}
|
}
|
||||||
|
@ -21,5 +32,5 @@ func (c *Client) CompareAndSwap(key string, value string, ttl uint64, prevValue
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
return raw.toResponse()
|
return raw, err
|
||||||
}
|
}
|
||||||
|
|
|
@ -0,0 +1 @@
|
||||||
|
{"config":{"certFile":"","keyFile":"","caCertFiles":null,"timeout":1000000000,"Consistency":"STRONG"},"cluster":{"leader":"http://127.0.0.1:4001","machines":["http://127.0.0.1:4001","http://127.0.0.1:4002"]}}
|
|
@ -1,28 +1,53 @@
|
||||||
package etcd
|
package etcd
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"os"
|
"io/ioutil"
|
||||||
|
"log"
|
||||||
"github.com/coreos/etcd/third_party/github.com/coreos/go-log/log"
|
"strings"
|
||||||
)
|
)
|
||||||
|
|
||||||
var logger *log.Logger
|
type Logger interface {
|
||||||
|
Debug(args ...interface{})
|
||||||
|
Debugf(fmt string, args ...interface{})
|
||||||
|
Warning(args ...interface{})
|
||||||
|
Warningf(fmt string, args ...interface{})
|
||||||
|
}
|
||||||
|
|
||||||
|
var logger Logger
|
||||||
|
|
||||||
|
func SetLogger(log Logger) {
|
||||||
|
logger = log
|
||||||
|
}
|
||||||
|
|
||||||
|
func GetLogger() Logger {
|
||||||
|
return logger
|
||||||
|
}
|
||||||
|
|
||||||
|
type defaultLogger struct {
|
||||||
|
log *log.Logger
|
||||||
|
}
|
||||||
|
|
||||||
|
func (p *defaultLogger) Debug(args ...interface{}) {
|
||||||
|
p.log.Println(args)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (p *defaultLogger) Debugf(fmt string, args ...interface{}) {
|
||||||
|
// Append newline if necessary
|
||||||
|
if !strings.HasSuffix(fmt, "\n") {
|
||||||
|
fmt = fmt + "\n"
|
||||||
|
}
|
||||||
|
p.log.Printf(fmt, args)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (p *defaultLogger) Warning(args ...interface{}) {
|
||||||
|
p.Debug(args)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (p *defaultLogger) Warningf(fmt string, args ...interface{}) {
|
||||||
|
p.Debugf(fmt, args)
|
||||||
|
}
|
||||||
|
|
||||||
func init() {
|
func init() {
|
||||||
setLogger(log.PriErr)
|
// Default logger uses the go default log.
|
||||||
}
|
SetLogger(&defaultLogger{log.New(ioutil.Discard, "go-etcd", log.LstdFlags)})
|
||||||
|
|
||||||
func OpenDebug() {
|
|
||||||
setLogger(log.PriDebug)
|
|
||||||
}
|
|
||||||
|
|
||||||
func CloseDebug() {
|
|
||||||
setLogger(log.PriErr)
|
|
||||||
}
|
|
||||||
|
|
||||||
func setLogger(priority log.Priority) {
|
|
||||||
logger = log.NewSimple(
|
|
||||||
log.PriorityFilter(
|
|
||||||
priority,
|
|
||||||
log.WriterSink(os.Stdout, log.BasicFormat, log.BasicFields)))
|
|
||||||
}
|
}
|
||||||
|
|
|
@ -10,7 +10,7 @@ package etcd
|
||||||
// then everything under the directory (including all child directories)
|
// then everything under the directory (including all child directories)
|
||||||
// will be deleted.
|
// will be deleted.
|
||||||
func (c *Client) Delete(key string, recursive bool) (*Response, error) {
|
func (c *Client) Delete(key string, recursive bool) (*Response, error) {
|
||||||
raw, err := c.DeleteRaw(key, recursive, false)
|
raw, err := c.RawDelete(key, recursive, false)
|
||||||
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
|
@ -21,7 +21,7 @@ func (c *Client) Delete(key string, recursive bool) (*Response, error) {
|
||||||
|
|
||||||
// DeleteDir deletes an empty directory or a key value pair
|
// DeleteDir deletes an empty directory or a key value pair
|
||||||
func (c *Client) DeleteDir(key string) (*Response, error) {
|
func (c *Client) DeleteDir(key string) (*Response, error) {
|
||||||
raw, err := c.DeleteRaw(key, false, true)
|
raw, err := c.RawDelete(key, false, true)
|
||||||
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
|
@ -30,7 +30,7 @@ func (c *Client) DeleteDir(key string) (*Response, error) {
|
||||||
return raw.toResponse()
|
return raw.toResponse()
|
||||||
}
|
}
|
||||||
|
|
||||||
func (c *Client) DeleteRaw(key string, recursive bool, dir bool) (*RawResponse, error) {
|
func (c *Client) RawDelete(key string, recursive bool, dir bool) (*RawResponse, error) {
|
||||||
ops := options{
|
ops := options{
|
||||||
"recursive": recursive,
|
"recursive": recursive,
|
||||||
"dir": dir,
|
"dir": dir,
|
||||||
|
|
|
@ -36,9 +36,13 @@ func newError(errorCode int, cause string, index uint64) *EtcdError {
|
||||||
}
|
}
|
||||||
|
|
||||||
func handleError(b []byte) error {
|
func handleError(b []byte) error {
|
||||||
var err EtcdError
|
etcdErr := new(EtcdError)
|
||||||
|
|
||||||
json.Unmarshal(b, &err)
|
|
||||||
|
|
||||||
|
err := json.Unmarshal(b, etcdErr)
|
||||||
|
if err != nil {
|
||||||
|
logger.Warningf("cannot unmarshal etcd error: %v", err)
|
||||||
return err
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
return etcdErr
|
||||||
}
|
}
|
||||||
|
|
|
@ -5,6 +5,26 @@ import (
|
||||||
"testing"
|
"testing"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
// cleanNode scrubs Expiration, ModifiedIndex and CreatedIndex of a node.
|
||||||
|
func cleanNode(n *Node) {
|
||||||
|
n.Expiration = nil
|
||||||
|
n.ModifiedIndex = 0
|
||||||
|
n.CreatedIndex = 0
|
||||||
|
}
|
||||||
|
|
||||||
|
// cleanResult scrubs a result object two levels deep of Expiration,
|
||||||
|
// ModifiedIndex and CreatedIndex.
|
||||||
|
func cleanResult(result *Response) {
|
||||||
|
// TODO(philips): make this recursive.
|
||||||
|
cleanNode(result.Node)
|
||||||
|
for i, _ := range result.Node.Nodes {
|
||||||
|
cleanNode(&result.Node.Nodes[i])
|
||||||
|
for j, _ := range result.Node.Nodes[i].Nodes {
|
||||||
|
cleanNode(&result.Node.Nodes[i].Nodes[j])
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestGet(t *testing.T) {
|
func TestGet(t *testing.T) {
|
||||||
c := NewClient(nil)
|
c := NewClient(nil)
|
||||||
defer func() {
|
defer func() {
|
||||||
|
@ -51,22 +71,15 @@ func TestGetAll(t *testing.T) {
|
||||||
Key: "/fooDir/k0",
|
Key: "/fooDir/k0",
|
||||||
Value: "v0",
|
Value: "v0",
|
||||||
TTL: 5,
|
TTL: 5,
|
||||||
ModifiedIndex: 31,
|
|
||||||
CreatedIndex: 31,
|
|
||||||
},
|
},
|
||||||
Node{
|
Node{
|
||||||
Key: "/fooDir/k1",
|
Key: "/fooDir/k1",
|
||||||
Value: "v1",
|
Value: "v1",
|
||||||
TTL: 5,
|
TTL: 5,
|
||||||
ModifiedIndex: 32,
|
|
||||||
CreatedIndex: 32,
|
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
|
|
||||||
// do not check expiration time, too hard to fake
|
cleanResult(result)
|
||||||
for i, _ := range result.Node.Nodes {
|
|
||||||
result.Node.Nodes[i].Expiration = nil
|
|
||||||
}
|
|
||||||
|
|
||||||
if !reflect.DeepEqual(result.Node.Nodes, expected) {
|
if !reflect.DeepEqual(result.Node.Nodes, expected) {
|
||||||
t.Fatalf("(actual) %v != (expected) %v", result.Node.Nodes, expected)
|
t.Fatalf("(actual) %v != (expected) %v", result.Node.Nodes, expected)
|
||||||
|
@ -79,16 +92,7 @@ func TestGetAll(t *testing.T) {
|
||||||
// Return kv-pairs in sorted order
|
// Return kv-pairs in sorted order
|
||||||
result, err = c.Get("fooDir", true, true)
|
result, err = c.Get("fooDir", true, true)
|
||||||
|
|
||||||
// do not check expiration time, too hard to fake
|
cleanResult(result)
|
||||||
result.Node.Expiration = nil
|
|
||||||
for i, _ := range result.Node.Nodes {
|
|
||||||
result.Node.Nodes[i].Expiration = nil
|
|
||||||
if result.Node.Nodes[i].Nodes != nil {
|
|
||||||
for j, _ := range result.Node.Nodes[i].Nodes {
|
|
||||||
result.Node.Nodes[i].Nodes[j].Expiration = nil
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
|
@ -103,30 +107,24 @@ func TestGetAll(t *testing.T) {
|
||||||
Key: "/fooDir/childDir/k2",
|
Key: "/fooDir/childDir/k2",
|
||||||
Value: "v2",
|
Value: "v2",
|
||||||
TTL: 5,
|
TTL: 5,
|
||||||
ModifiedIndex: 34,
|
|
||||||
CreatedIndex: 34,
|
|
||||||
},
|
},
|
||||||
},
|
},
|
||||||
TTL: 5,
|
TTL: 5,
|
||||||
ModifiedIndex: 33,
|
|
||||||
CreatedIndex: 33,
|
|
||||||
},
|
},
|
||||||
Node{
|
Node{
|
||||||
Key: "/fooDir/k0",
|
Key: "/fooDir/k0",
|
||||||
Value: "v0",
|
Value: "v0",
|
||||||
TTL: 5,
|
TTL: 5,
|
||||||
ModifiedIndex: 31,
|
|
||||||
CreatedIndex: 31,
|
|
||||||
},
|
},
|
||||||
Node{
|
Node{
|
||||||
Key: "/fooDir/k1",
|
Key: "/fooDir/k1",
|
||||||
Value: "v1",
|
Value: "v1",
|
||||||
TTL: 5,
|
TTL: 5,
|
||||||
ModifiedIndex: 32,
|
|
||||||
CreatedIndex: 32,
|
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
|
|
||||||
|
cleanResult(result)
|
||||||
|
|
||||||
if !reflect.DeepEqual(result.Node.Nodes, expected) {
|
if !reflect.DeepEqual(result.Node.Nodes, expected) {
|
||||||
t.Fatalf("(actual) %v != (expected) %v", result.Node.Nodes, expected)
|
t.Fatalf("(actual) %v != (expected) %v", result.Node.Nodes, expected)
|
||||||
}
|
}
|
||||||
|
|
|
@ -36,6 +36,8 @@ var (
|
||||||
VALID_DELETE_OPTIONS = validOptions{
|
VALID_DELETE_OPTIONS = validOptions{
|
||||||
"recursive": reflect.Bool,
|
"recursive": reflect.Bool,
|
||||||
"dir": reflect.Bool,
|
"dir": reflect.Bool,
|
||||||
|
"prevValue": reflect.String,
|
||||||
|
"prevIndex": reflect.Uint64,
|
||||||
}
|
}
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
|
@ -13,9 +13,6 @@ import (
|
||||||
|
|
||||||
// get issues a GET request
|
// get issues a GET request
|
||||||
func (c *Client) get(key string, options options) (*RawResponse, error) {
|
func (c *Client) get(key string, options options) (*RawResponse, error) {
|
||||||
logger.Debugf("get %s [%s]", key, c.cluster.Leader)
|
|
||||||
p := keyToPath(key)
|
|
||||||
|
|
||||||
// If consistency level is set to STRONG, append
|
// If consistency level is set to STRONG, append
|
||||||
// the `consistent` query string.
|
// the `consistent` query string.
|
||||||
if c.config.Consistency == STRONG_CONSISTENCY {
|
if c.config.Consistency == STRONG_CONSISTENCY {
|
||||||
|
@ -26,9 +23,8 @@ func (c *Client) get(key string, options options) (*RawResponse, error) {
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
p += str
|
|
||||||
|
|
||||||
resp, err := c.sendRequest("GET", p, nil)
|
resp, err := c.sendKeyRequest("GET", key, str, nil)
|
||||||
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
|
@ -41,16 +37,12 @@ func (c *Client) get(key string, options options) (*RawResponse, error) {
|
||||||
func (c *Client) put(key string, value string, ttl uint64,
|
func (c *Client) put(key string, value string, ttl uint64,
|
||||||
options options) (*RawResponse, error) {
|
options options) (*RawResponse, error) {
|
||||||
|
|
||||||
logger.Debugf("put %s, %s, ttl: %d, [%s]", key, value, ttl, c.cluster.Leader)
|
|
||||||
p := keyToPath(key)
|
|
||||||
|
|
||||||
str, err := options.toParameters(VALID_PUT_OPTIONS)
|
str, err := options.toParameters(VALID_PUT_OPTIONS)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
p += str
|
|
||||||
|
|
||||||
resp, err := c.sendRequest("PUT", p, buildValues(value, ttl))
|
resp, err := c.sendKeyRequest("PUT", key, str, buildValues(value, ttl))
|
||||||
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
|
@ -61,10 +53,7 @@ func (c *Client) put(key string, value string, ttl uint64,
|
||||||
|
|
||||||
// post issues a POST request
|
// post issues a POST request
|
||||||
func (c *Client) post(key string, value string, ttl uint64) (*RawResponse, error) {
|
func (c *Client) post(key string, value string, ttl uint64) (*RawResponse, error) {
|
||||||
logger.Debugf("post %s, %s, ttl: %d, [%s]", key, value, ttl, c.cluster.Leader)
|
resp, err := c.sendKeyRequest("POST", key, "", buildValues(value, ttl))
|
||||||
p := keyToPath(key)
|
|
||||||
|
|
||||||
resp, err := c.sendRequest("POST", p, buildValues(value, ttl))
|
|
||||||
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
|
@ -75,16 +64,12 @@ func (c *Client) post(key string, value string, ttl uint64) (*RawResponse, error
|
||||||
|
|
||||||
// delete issues a DELETE request
|
// delete issues a DELETE request
|
||||||
func (c *Client) delete(key string, options options) (*RawResponse, error) {
|
func (c *Client) delete(key string, options options) (*RawResponse, error) {
|
||||||
logger.Debugf("delete %s [%s]", key, c.cluster.Leader)
|
|
||||||
p := keyToPath(key)
|
|
||||||
|
|
||||||
str, err := options.toParameters(VALID_DELETE_OPTIONS)
|
str, err := options.toParameters(VALID_DELETE_OPTIONS)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
p += str
|
|
||||||
|
|
||||||
resp, err := c.sendRequest("DELETE", p, nil)
|
resp, err := c.sendKeyRequest("DELETE", key, str, nil)
|
||||||
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
|
@ -93,8 +78,8 @@ func (c *Client) delete(key string, options options) (*RawResponse, error) {
|
||||||
return resp, nil
|
return resp, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// sendRequest sends a HTTP request and returns a Response as defined by etcd
|
// sendKeyRequest sends a HTTP request and returns a Response as defined by etcd
|
||||||
func (c *Client) sendRequest(method string, relativePath string,
|
func (c *Client) sendKeyRequest(method string, key string, params string,
|
||||||
values url.Values) (*RawResponse, error) {
|
values url.Values) (*RawResponse, error) {
|
||||||
|
|
||||||
var req *http.Request
|
var req *http.Request
|
||||||
|
@ -105,6 +90,11 @@ func (c *Client) sendRequest(method string, relativePath string,
|
||||||
|
|
||||||
trial := 0
|
trial := 0
|
||||||
|
|
||||||
|
logger.Debugf("%s %s %s [%s]", method, key, params, c.cluster.Leader)
|
||||||
|
|
||||||
|
// Build the request path if no prefix exists
|
||||||
|
relativePath := path.Join(c.keyPrefix, key) + params
|
||||||
|
|
||||||
// if we connect to a follower, we will retry until we found a leader
|
// if we connect to a follower, we will retry until we found a leader
|
||||||
for {
|
for {
|
||||||
trial++
|
trial++
|
||||||
|
@ -146,7 +136,8 @@ func (c *Client) sendRequest(method string, relativePath string,
|
||||||
|
|
||||||
// network error, change a machine!
|
// network error, change a machine!
|
||||||
if resp, err = c.httpClient.Do(req); err != nil {
|
if resp, err = c.httpClient.Do(req); err != nil {
|
||||||
c.switchLeader(trial % len(c.cluster.Machines))
|
logger.Debug("network error: ", err.Error())
|
||||||
|
c.cluster.switchLeader(trial % len(c.cluster.Machines))
|
||||||
time.Sleep(time.Millisecond * 200)
|
time.Sleep(time.Millisecond * 200)
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
@ -195,7 +186,7 @@ func (c *Client) handleResp(resp *http.Response) (bool, []byte) {
|
||||||
if err != nil {
|
if err != nil {
|
||||||
logger.Warning(err)
|
logger.Warning(err)
|
||||||
} else {
|
} else {
|
||||||
c.updateLeader(u)
|
c.cluster.updateLeaderFromURL(u)
|
||||||
}
|
}
|
||||||
|
|
||||||
return false, nil
|
return false, nil
|
||||||
|
@ -219,18 +210,14 @@ func (c *Client) handleResp(resp *http.Response) (bool, []byte) {
|
||||||
|
|
||||||
func (c *Client) getHttpPath(random bool, s ...string) string {
|
func (c *Client) getHttpPath(random bool, s ...string) string {
|
||||||
var machine string
|
var machine string
|
||||||
|
|
||||||
if random {
|
if random {
|
||||||
machine = c.cluster.Machines[rand.Intn(len(c.cluster.Machines))]
|
machine = c.cluster.Machines[rand.Intn(len(c.cluster.Machines))]
|
||||||
} else {
|
} else {
|
||||||
machine = c.cluster.Leader
|
machine = c.cluster.Leader
|
||||||
}
|
}
|
||||||
|
|
||||||
fullPath := machine + "/" + version
|
return machine + "/" + strings.Join(s, "/")
|
||||||
for _, seg := range s {
|
|
||||||
fullPath = fullPath + "/" + seg
|
|
||||||
}
|
|
||||||
|
|
||||||
return fullPath
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// buildValues builds a url.Values map according to the given value and ttl
|
// buildValues builds a url.Values map according to the given value and ttl
|
||||||
|
@ -249,17 +236,14 @@ func buildValues(value string, ttl uint64) url.Values {
|
||||||
}
|
}
|
||||||
|
|
||||||
// convert key string to http path exclude version
|
// convert key string to http path exclude version
|
||||||
// for example: key[foo] -> path[keys/foo]
|
// for example: key[foo] -> path[foo]
|
||||||
// key[/] -> path[keys/]
|
// key[] -> path[/]
|
||||||
func keyToPath(key string) string {
|
func keyToPath(key string) string {
|
||||||
p := path.Join("keys", key)
|
clean := path.Clean(key)
|
||||||
|
|
||||||
// corner case: if key is "/" or "//" ect
|
if clean == "" || clean == "." {
|
||||||
// path join will clear the tailing "/"
|
return "/"
|
||||||
// we need to add it back
|
|
||||||
if p == "keys" {
|
|
||||||
p = "keys/"
|
|
||||||
}
|
}
|
||||||
|
|
||||||
return p
|
return clean
|
||||||
}
|
}
|
||||||
|
|
|
@ -0,0 +1,50 @@
|
||||||
|
package etcd
|
||||||
|
|
||||||
|
import (
|
||||||
|
"path"
|
||||||
|
"testing"
|
||||||
|
)
|
||||||
|
|
||||||
|
func testKey(t *testing.T, in, exp string) {
|
||||||
|
if keyToPath(in) != exp {
|
||||||
|
t.Errorf("Expected %s got %s", exp, keyToPath(in))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestKeyToPath ensures the key cleaning funciton keyToPath works in a number
|
||||||
|
// of cases.
|
||||||
|
func TestKeyToPath(t *testing.T) {
|
||||||
|
testKey(t, "", "/")
|
||||||
|
testKey(t, "/", "/")
|
||||||
|
testKey(t, "///", "/")
|
||||||
|
testKey(t, "hello/world/", "hello/world")
|
||||||
|
testKey(t, "///hello////world/../", "/hello")
|
||||||
|
}
|
||||||
|
|
||||||
|
func testPath(t *testing.T, c *Client, in, exp string) {
|
||||||
|
out := c.getHttpPath(false, in)
|
||||||
|
|
||||||
|
if out != exp {
|
||||||
|
t.Errorf("Expected %s got %s", exp, out)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestHttpPath ensures that the URLs generated make sense for the given keys
|
||||||
|
func TestHttpPath(t *testing.T) {
|
||||||
|
c := NewClient(nil)
|
||||||
|
|
||||||
|
testPath(t, c,
|
||||||
|
path.Join(c.keyPrefix, "hello") + "?prevInit=true",
|
||||||
|
"http://127.0.0.1:4001/v2/keys/hello?prevInit=true")
|
||||||
|
|
||||||
|
testPath(t, c,
|
||||||
|
path.Join(c.keyPrefix, "///hello///world") + "?prevInit=true",
|
||||||
|
"http://127.0.0.1:4001/v2/keys/hello/world?prevInit=true")
|
||||||
|
|
||||||
|
c = NewClient([]string{"https://discovery.etcd.io"})
|
||||||
|
c.SetKeyPrefix("")
|
||||||
|
|
||||||
|
testPath(t, c,
|
||||||
|
path.Join(c.keyPrefix, "hello") + "?prevInit=true",
|
||||||
|
"https://discovery.etcd.io/hello?prevInit=true")
|
||||||
|
}
|
|
@ -1,33 +0,0 @@
|
||||||
// Utility functions
|
|
||||||
|
|
||||||
package etcd
|
|
||||||
|
|
||||||
import (
|
|
||||||
"fmt"
|
|
||||||
"net/url"
|
|
||||||
"reflect"
|
|
||||||
)
|
|
||||||
|
|
||||||
// Convert options to a string of HTML parameters
|
|
||||||
func optionsToString(options options, vops validOptions) (string, error) {
|
|
||||||
p := "?"
|
|
||||||
v := url.Values{}
|
|
||||||
for opKey, opVal := range options {
|
|
||||||
// Check if the given option is valid (that it exists)
|
|
||||||
kind := vops[opKey]
|
|
||||||
if kind == reflect.Invalid {
|
|
||||||
return "", fmt.Errorf("Invalid option: %v", opKey)
|
|
||||||
}
|
|
||||||
|
|
||||||
// Check if the given option is of the valid type
|
|
||||||
t := reflect.TypeOf(opVal)
|
|
||||||
if kind != t.Kind() {
|
|
||||||
return "", fmt.Errorf("Option %s should be of %v kind, not of %v kind.",
|
|
||||||
opKey, kind, t.Kind())
|
|
||||||
}
|
|
||||||
|
|
||||||
v.Set(opKey, fmt.Sprintf("%v", opVal))
|
|
||||||
}
|
|
||||||
p += v.Encode()
|
|
||||||
return p, nil
|
|
||||||
}
|
|
Loading…
Reference in New Issue