discovery: support structured logger
Signed-off-by: Gyuho Lee <gyuhox@gmail.com>release-3.4
parent
f269c42aad
commit
6a016cbd86
|
@ -35,6 +35,7 @@ import (
|
||||||
|
|
||||||
"github.com/coreos/pkg/capnslog"
|
"github.com/coreos/pkg/capnslog"
|
||||||
"github.com/jonboulle/clockwork"
|
"github.com/jonboulle/clockwork"
|
||||||
|
"go.uber.org/zap"
|
||||||
)
|
)
|
||||||
|
|
||||||
var (
|
var (
|
||||||
|
@ -59,8 +60,8 @@ var (
|
||||||
|
|
||||||
// JoinCluster will connect to the discovery service at the given url, and
|
// JoinCluster will connect to the discovery service at the given url, and
|
||||||
// register the server represented by the given id and config to the cluster
|
// register the server represented by the given id and config to the cluster
|
||||||
func JoinCluster(durl, dproxyurl string, id types.ID, config string) (string, error) {
|
func JoinCluster(lg *zap.Logger, durl, dproxyurl string, id types.ID, config string) (string, error) {
|
||||||
d, err := newDiscovery(durl, dproxyurl, id)
|
d, err := newDiscovery(lg, durl, dproxyurl, id)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return "", err
|
return "", err
|
||||||
}
|
}
|
||||||
|
@ -69,8 +70,8 @@ func JoinCluster(durl, dproxyurl string, id types.ID, config string) (string, er
|
||||||
|
|
||||||
// GetCluster will connect to the discovery service at the given url and
|
// GetCluster will connect to the discovery service at the given url and
|
||||||
// retrieve a string describing the cluster
|
// retrieve a string describing the cluster
|
||||||
func GetCluster(durl, dproxyurl string) (string, error) {
|
func GetCluster(lg *zap.Logger, durl, dproxyurl string) (string, error) {
|
||||||
d, err := newDiscovery(durl, dproxyurl, 0)
|
d, err := newDiscovery(lg, durl, dproxyurl, 0)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return "", err
|
return "", err
|
||||||
}
|
}
|
||||||
|
@ -78,6 +79,7 @@ func GetCluster(durl, dproxyurl string) (string, error) {
|
||||||
}
|
}
|
||||||
|
|
||||||
type discovery struct {
|
type discovery struct {
|
||||||
|
lg *zap.Logger
|
||||||
cluster string
|
cluster string
|
||||||
id types.ID
|
id types.ID
|
||||||
c client.KeysAPI
|
c client.KeysAPI
|
||||||
|
@ -90,7 +92,7 @@ type discovery struct {
|
||||||
// newProxyFunc builds a proxy function from the given string, which should
|
// newProxyFunc builds a proxy function from the given string, which should
|
||||||
// represent a URL that can be used as a proxy. It performs basic
|
// represent a URL that can be used as a proxy. It performs basic
|
||||||
// sanitization of the URL and returns any error encountered.
|
// sanitization of the URL and returns any error encountered.
|
||||||
func newProxyFunc(proxy string) (func(*http.Request) (*url.URL, error), error) {
|
func newProxyFunc(lg *zap.Logger, proxy string) (func(*http.Request) (*url.URL, error), error) {
|
||||||
if proxy == "" {
|
if proxy == "" {
|
||||||
return nil, nil
|
return nil, nil
|
||||||
}
|
}
|
||||||
|
@ -111,18 +113,22 @@ func newProxyFunc(proxy string) (func(*http.Request) (*url.URL, error), error) {
|
||||||
return nil, fmt.Errorf("invalid proxy address %q: %v", proxy, err)
|
return nil, fmt.Errorf("invalid proxy address %q: %v", proxy, err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if lg != nil {
|
||||||
|
lg.Info("running proxy with discovery", zap.String("proxy-url", proxyURL.String()))
|
||||||
|
} else {
|
||||||
plog.Infof("using proxy %q", proxyURL.String())
|
plog.Infof("using proxy %q", proxyURL.String())
|
||||||
|
}
|
||||||
return http.ProxyURL(proxyURL), nil
|
return http.ProxyURL(proxyURL), nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func newDiscovery(durl, dproxyurl string, id types.ID) (*discovery, error) {
|
func newDiscovery(lg *zap.Logger, durl, dproxyurl string, id types.ID) (*discovery, error) {
|
||||||
u, err := url.Parse(durl)
|
u, err := url.Parse(durl)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
token := u.Path
|
token := u.Path
|
||||||
u.Path = ""
|
u.Path = ""
|
||||||
pf, err := newProxyFunc(dproxyurl)
|
pf, err := newProxyFunc(lg, dproxyurl)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
@ -143,6 +149,7 @@ func newDiscovery(durl, dproxyurl string, id types.ID) (*discovery, error) {
|
||||||
}
|
}
|
||||||
dc := client.NewKeysAPIWithPrefix(c, "")
|
dc := client.NewKeysAPIWithPrefix(c, "")
|
||||||
return &discovery{
|
return &discovery{
|
||||||
|
lg: lg,
|
||||||
cluster: token,
|
cluster: token,
|
||||||
c: dc,
|
c: dc,
|
||||||
id: id,
|
id: id,
|
||||||
|
@ -225,7 +232,17 @@ func (d *discovery) checkCluster() ([]*client.Node, int, uint64, error) {
|
||||||
return nil, 0, 0, ErrBadDiscoveryEndpoint
|
return nil, 0, 0, ErrBadDiscoveryEndpoint
|
||||||
}
|
}
|
||||||
if ce, ok := err.(*client.ClusterError); ok {
|
if ce, ok := err.(*client.ClusterError); ok {
|
||||||
|
if d.lg != nil {
|
||||||
|
d.lg.Warn(
|
||||||
|
"failed to get from discovery server",
|
||||||
|
zap.String("discovery-url", d.url.String()),
|
||||||
|
zap.String("path", path.Join(configKey, "size")),
|
||||||
|
zap.Error(err),
|
||||||
|
zap.String("err-detail", ce.Detail()),
|
||||||
|
)
|
||||||
|
} else {
|
||||||
plog.Error(ce.Detail())
|
plog.Error(ce.Detail())
|
||||||
|
}
|
||||||
return d.checkClusterRetry()
|
return d.checkClusterRetry()
|
||||||
}
|
}
|
||||||
return nil, 0, 0, err
|
return nil, 0, 0, err
|
||||||
|
@ -240,7 +257,17 @@ func (d *discovery) checkCluster() ([]*client.Node, int, uint64, error) {
|
||||||
cancel()
|
cancel()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
if ce, ok := err.(*client.ClusterError); ok {
|
if ce, ok := err.(*client.ClusterError); ok {
|
||||||
|
if d.lg != nil {
|
||||||
|
d.lg.Warn(
|
||||||
|
"failed to get from discovery server",
|
||||||
|
zap.String("discovery-url", d.url.String()),
|
||||||
|
zap.String("path", d.cluster),
|
||||||
|
zap.Error(err),
|
||||||
|
zap.String("err-detail", ce.Detail()),
|
||||||
|
)
|
||||||
|
} else {
|
||||||
plog.Error(ce.Detail())
|
plog.Error(ce.Detail())
|
||||||
|
}
|
||||||
return d.checkClusterRetry()
|
return d.checkClusterRetry()
|
||||||
}
|
}
|
||||||
return nil, 0, 0, err
|
return nil, 0, 0, err
|
||||||
|
@ -276,7 +303,16 @@ func (d *discovery) logAndBackoffForRetry(step string) {
|
||||||
retries = maxExpoentialRetries
|
retries = maxExpoentialRetries
|
||||||
}
|
}
|
||||||
retryTimeInSecond := time.Duration(0x1<<retries) * time.Second
|
retryTimeInSecond := time.Duration(0x1<<retries) * time.Second
|
||||||
|
if d.lg != nil {
|
||||||
|
d.lg.Info(
|
||||||
|
"retry connecting to discovery service",
|
||||||
|
zap.String("url", d.url.String()),
|
||||||
|
zap.String("reason", step),
|
||||||
|
zap.Duration("backoff", retryTimeInSecond),
|
||||||
|
)
|
||||||
|
} else {
|
||||||
plog.Infof("%s: error connecting to %s, retrying in %s", step, d.url, retryTimeInSecond)
|
plog.Infof("%s: error connecting to %s, retrying in %s", step, d.url, retryTimeInSecond)
|
||||||
|
}
|
||||||
d.clock.Sleep(retryTimeInSecond)
|
d.clock.Sleep(retryTimeInSecond)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -310,15 +346,40 @@ func (d *discovery) waitNodes(nodes []*client.Node, size int, index uint64) ([]*
|
||||||
copy(all, nodes)
|
copy(all, nodes)
|
||||||
for _, n := range all {
|
for _, n := range all {
|
||||||
if path.Base(n.Key) == path.Base(d.selfKey()) {
|
if path.Base(n.Key) == path.Base(d.selfKey()) {
|
||||||
|
if d.lg != nil {
|
||||||
|
d.lg.Info(
|
||||||
|
"found self from discovery server",
|
||||||
|
zap.String("discovery-url", d.url.String()),
|
||||||
|
zap.String("self", path.Base(d.selfKey())),
|
||||||
|
)
|
||||||
|
} else {
|
||||||
plog.Noticef("found self %s in the cluster", path.Base(d.selfKey()))
|
plog.Noticef("found self %s in the cluster", path.Base(d.selfKey()))
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
if d.lg != nil {
|
||||||
|
d.lg.Info(
|
||||||
|
"found peer from discovery server",
|
||||||
|
zap.String("discovery-url", d.url.String()),
|
||||||
|
zap.String("peer", path.Base(n.Key)),
|
||||||
|
)
|
||||||
} else {
|
} else {
|
||||||
plog.Noticef("found peer %s in the cluster", path.Base(n.Key))
|
plog.Noticef("found peer %s in the cluster", path.Base(n.Key))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// wait for others
|
// wait for others
|
||||||
for len(all) < size {
|
for len(all) < size {
|
||||||
|
if d.lg != nil {
|
||||||
|
d.lg.Info(
|
||||||
|
"found peers from discovery server; waiting for more",
|
||||||
|
zap.String("discovery-url", d.url.String()),
|
||||||
|
zap.Int("found-peers", len(all)),
|
||||||
|
zap.Int("needed-peers", size-len(all)),
|
||||||
|
)
|
||||||
|
} else {
|
||||||
plog.Noticef("found %d peer(s), waiting for %d more", len(all), size-len(all))
|
plog.Noticef("found %d peer(s), waiting for %d more", len(all), size-len(all))
|
||||||
|
}
|
||||||
resp, err := w.Next(context.Background())
|
resp, err := w.Next(context.Background())
|
||||||
if err != nil {
|
if err != nil {
|
||||||
if ce, ok := err.(*client.ClusterError); ok {
|
if ce, ok := err.(*client.ClusterError); ok {
|
||||||
|
@ -327,10 +388,26 @@ func (d *discovery) waitNodes(nodes []*client.Node, size int, index uint64) ([]*
|
||||||
}
|
}
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
if d.lg != nil {
|
||||||
|
d.lg.Info(
|
||||||
|
"found peer from discovery server",
|
||||||
|
zap.String("discovery-url", d.url.String()),
|
||||||
|
zap.String("peer", path.Base(resp.Node.Key)),
|
||||||
|
)
|
||||||
|
} else {
|
||||||
plog.Noticef("found peer %s in the cluster", path.Base(resp.Node.Key))
|
plog.Noticef("found peer %s in the cluster", path.Base(resp.Node.Key))
|
||||||
|
}
|
||||||
all = append(all, resp.Node)
|
all = append(all, resp.Node)
|
||||||
}
|
}
|
||||||
|
if d.lg != nil {
|
||||||
|
d.lg.Info(
|
||||||
|
"found all needed peers from discovery server",
|
||||||
|
zap.String("discovery-url", d.url.String()),
|
||||||
|
zap.Int("found-peers", len(all)),
|
||||||
|
)
|
||||||
|
} else {
|
||||||
plog.Noticef("found %d needed peer(s)", len(all))
|
plog.Noticef("found %d needed peer(s)", len(all))
|
||||||
|
}
|
||||||
return all, nil
|
return all, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -26,6 +26,8 @@ import (
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
"go.uber.org/zap"
|
||||||
|
|
||||||
"github.com/coreos/etcd/client"
|
"github.com/coreos/etcd/client"
|
||||||
|
|
||||||
"github.com/jonboulle/clockwork"
|
"github.com/jonboulle/clockwork"
|
||||||
|
@ -36,7 +38,7 @@ const (
|
||||||
)
|
)
|
||||||
|
|
||||||
func TestNewProxyFuncUnset(t *testing.T) {
|
func TestNewProxyFuncUnset(t *testing.T) {
|
||||||
pf, err := newProxyFunc("")
|
pf, err := newProxyFunc(zap.NewExample(), "")
|
||||||
if pf != nil {
|
if pf != nil {
|
||||||
t.Fatal("unexpected non-nil proxyFunc")
|
t.Fatal("unexpected non-nil proxyFunc")
|
||||||
}
|
}
|
||||||
|
@ -51,7 +53,7 @@ func TestNewProxyFuncBad(t *testing.T) {
|
||||||
"http://foo.com/%1",
|
"http://foo.com/%1",
|
||||||
}
|
}
|
||||||
for i, in := range tests {
|
for i, in := range tests {
|
||||||
pf, err := newProxyFunc(in)
|
pf, err := newProxyFunc(zap.NewExample(), in)
|
||||||
if pf != nil {
|
if pf != nil {
|
||||||
t.Errorf("#%d: unexpected non-nil proxyFunc", i)
|
t.Errorf("#%d: unexpected non-nil proxyFunc", i)
|
||||||
}
|
}
|
||||||
|
@ -67,7 +69,7 @@ func TestNewProxyFunc(t *testing.T) {
|
||||||
"http://disco.foo.bar": "http://disco.foo.bar",
|
"http://disco.foo.bar": "http://disco.foo.bar",
|
||||||
}
|
}
|
||||||
for in, w := range tests {
|
for in, w := range tests {
|
||||||
pf, err := newProxyFunc(in)
|
pf, err := newProxyFunc(zap.NewExample(), in)
|
||||||
if pf == nil {
|
if pf == nil {
|
||||||
t.Errorf("%s: unexpected nil proxyFunc", in)
|
t.Errorf("%s: unexpected nil proxyFunc", in)
|
||||||
continue
|
continue
|
||||||
|
|
Loading…
Reference in New Issue