Skip to content

Commit bb29615

Browse files
jingyihjingyih
authored
proxy: remove capnslog (etcd-io#11614)
* proxy: remove capnslog * CHANGELOG: function signature change
1 parent 71e3220 commit bb29615

File tree

16 files changed

+121
-76
lines changed

16 files changed

+121
-76
lines changed

‎CHANGELOG-3.5.md

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -49,6 +49,10 @@ See [code changes](https://github.com/etcd-io/etcd/compare/v3.4.0...v3.5.0) and
4949
- Deprecated `etcd_debugging_mvcc_txn_total` Prometheus metric. Use `etcd_mvcc_txn_total` instead.
5050
- Deprecated `etcd_debugging_mvcc_range_total` Prometheus metric. Use `etcd_mvcc_range_total` instead.
5151
- Master branch `/version` outputs `3.5.0-pre`, instead of `3.4.0+git`.
52+
- Changed `proxy` package function signature to [support structured logger](https://github.com/etcd-io/etcd/pull/11614).
53+
- Previously, `NewClusterProxy(c *clientv3.Client, advaddr string, prefix string) (pb.ClusterServer, <-chan struct{})`, now `NewClusterProxy(lg *zap.Logger, c *clientv3.Client, advaddr string, prefix string) (pb.ClusterServer, <-chan struct{})`.
54+
- Previously, `Register(c *clientv3.Client, prefix string, addr string, ttl int)`, now `Register(lg *zap.Logger, c *clientv3.Client, prefix string, addr string, ttl int) <-chan struct{}`.
55+
- Previously, `NewHandler(t *http.Transport, urlsFunc GetProxyURLs, failureWait time.Duration, refreshInterval time.Duration) http.Handler`, now `NewHandler(lg *zap.Logger, t *http.Transport, urlsFunc GetProxyURLs, failureWait time.Duration, refreshInterval time.Duration) http.Handler`.
5256

5357
### Metrics, Monitoring
5458

‎etcdmain/etcd.go

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -364,7 +364,7 @@ func startProxy(cfg *config) error {
364364

365365
return clientURLs
366366
}
367-
ph := httpproxy.NewHandler(pt, uf, time.Duration(cfg.cp.ProxyFailureWaitMs)*time.Millisecond, time.Duration(cfg.cp.ProxyRefreshIntervalMs)*time.Millisecond)
367+
ph := httpproxy.NewHandler(lg, pt, uf, time.Duration(cfg.cp.ProxyFailureWaitMs)*time.Millisecond, time.Duration(cfg.cp.ProxyRefreshIntervalMs)*time.Millisecond)
368368
ph = embed.WrapCORS(cfg.ec.CORS, ph)
369369

370370
if cfg.isReadonlyProxy() {

‎etcdmain/grpc_proxy.go

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -349,9 +349,9 @@ func newGRPCProxyServer(lg *zap.Logger, client *clientv3.Client) *grpc.Server {
349349
kvp, _ := grpcproxy.NewKvProxy(client)
350350
watchp, _ := grpcproxy.NewWatchProxy(client)
351351
if grpcProxyResolverPrefix != "" {
352-
grpcproxy.Register(client, grpcProxyResolverPrefix, grpcProxyAdvertiseClientURL, grpcProxyResolverTTL)
352+
grpcproxy.Register(lg, client, grpcProxyResolverPrefix, grpcProxyAdvertiseClientURL, grpcProxyResolverTTL)
353353
}
354-
clusterp, _ := grpcproxy.NewClusterProxy(client, grpcProxyAdvertiseClientURL, grpcProxyResolverPrefix)
354+
clusterp, _ := grpcproxy.NewClusterProxy(lg, client, grpcProxyAdvertiseClientURL, grpcProxyResolverPrefix)
355355
leasep, _ := grpcproxy.NewLeaseProxy(client)
356356
mainp := grpcproxy.NewMaintenanceProxy(client)
357357
authp := grpcproxy.NewAuthProxy(client)

‎integration/cluster_proxy.go

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,8 @@ import (
2323
"go.etcd.io/etcd/clientv3/namespace"
2424
"go.etcd.io/etcd/proxy/grpcproxy"
2525
"go.etcd.io/etcd/proxy/grpcproxy/adapter"
26+
27+
"go.uber.org/zap"
2628
)
2729

2830
var (
@@ -56,7 +58,7 @@ func toGRPC(c *clientv3.Client) grpcAPI {
5658
wp, wpch := grpcproxy.NewWatchProxy(c)
5759
lp, lpch := grpcproxy.NewLeaseProxy(c)
5860
mp := grpcproxy.NewMaintenanceProxy(c)
59-
clp, _ := grpcproxy.NewClusterProxy(c, "", "") // without registering proxy URLs
61+
clp, _ := grpcproxy.NewClusterProxy(zap.NewExample(), c, "", "") // without registering proxy URLs
6062
authp := grpcproxy.NewAuthProxy(c)
6163
lockp := grpcproxy.NewLockProxy(c)
6264
electp := grpcproxy.NewElectionProxy(c)

‎proxy/grpcproxy/cluster.go

Lines changed: 9 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@ import (
2626
"go.etcd.io/etcd/etcdserver/api/v3rpc/rpctypes"
2727
pb "go.etcd.io/etcd/etcdserver/etcdserverpb"
2828

29+
"go.uber.org/zap"
2930
"golang.org/x/time/rate"
3031
gnaming "google.golang.org/grpc/naming"
3132
)
@@ -34,6 +35,7 @@ import (
3435
const resolveRetryRate = 1
3536

3637
type clusterProxy struct {
38+
lg *zap.Logger
3739
clus clientv3.Cluster
3840
ctx context.Context
3941
gr *naming.GRPCResolver
@@ -49,8 +51,12 @@ type clusterProxy struct {
4951
// NewClusterProxy takes optional prefix to fetch grpc-proxy member endpoints.
5052
// The returned channel is closed when there is grpc-proxy endpoint registered
5153
// and the client's context is canceled so the 'register' loop returns.
52-
func NewClusterProxy(c *clientv3.Client, advaddr string, prefix string) (pb.ClusterServer, <-chan struct{}) {
54+
func NewClusterProxy(lg *zap.Logger, c *clientv3.Client, advaddr string, prefix string) (pb.ClusterServer, <-chan struct{}) {
55+
if lg == nil {
56+
lg = zap.NewNop()
57+
}
5358
cp := &clusterProxy{
59+
lg: lg,
5460
clus: c.Cluster,
5561
ctx: c.Ctx(),
5662
gr: &naming.GRPCResolver{Client: c},
@@ -78,7 +84,7 @@ func (cp *clusterProxy) resolve(prefix string) {
7884
for rm.Wait(cp.ctx) == nil {
7985
wa, err := cp.gr.Resolve(prefix)
8086
if err != nil {
81-
plog.Warningf("failed to resolve %q (%v)", prefix, err)
87+
cp.lg.Warn("failed to resolve prefix", zap.String("prefix", prefix), zap.Error(err))
8288
continue
8389
}
8490
cp.monitor(wa)
@@ -89,7 +95,7 @@ func (cp *clusterProxy) monitor(wa gnaming.Watcher) {
8995
for cp.ctx.Err() == nil {
9096
ups, err := wa.Next()
9197
if err != nil {
92-
plog.Warningf("clusterProxy watcher error (%v)", err)
98+
cp.lg.Warn("clusterProxy watcher error", zap.Error(err))
9399
if rpctypes.ErrorDesc(err) == naming.ErrWatcherClosed.Error() {
94100
return
95101
}

‎proxy/grpcproxy/cluster_test.go

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@ import (
2525
"go.etcd.io/etcd/integration"
2626
"go.etcd.io/etcd/pkg/testutil"
2727

28+
"go.uber.org/zap"
2829
"google.golang.org/grpc"
2930
)
3031

@@ -34,7 +35,7 @@ func TestClusterProxyMemberList(t *testing.T) {
3435
clus := integration.NewClusterV3(t, &integration.ClusterConfig{Size: 1})
3536
defer clus.Terminate(t)
3637

37-
cts := newClusterProxyServer([]string{clus.Members[0].GRPCAddr()}, t)
38+
cts := newClusterProxyServer(zap.NewExample(), []string{clus.Members[0].GRPCAddr()}, t)
3839
defer cts.close(t)
3940

4041
cfg := clientv3.Config{
@@ -88,7 +89,7 @@ func (cts *clusterproxyTestServer) close(t *testing.T) {
8889
}
8990
}
9091

91-
func newClusterProxyServer(endpoints []string, t *testing.T) *clusterproxyTestServer {
92+
func newClusterProxyServer(lg *zap.Logger, endpoints []string, t *testing.T) *clusterproxyTestServer {
9293
cfg := clientv3.Config{
9394
Endpoints: endpoints,
9495
DialTimeout: 5 * time.Second,
@@ -113,8 +114,8 @@ func newClusterProxyServer(endpoints []string, t *testing.T) *clusterproxyTestSe
113114
cts.server.Serve(cts.l)
114115
}()
115116

116-
Register(client, "test-prefix", cts.l.Addr().String(), 7)
117-
cts.cp, cts.donec = NewClusterProxy(client, cts.l.Addr().String(), "test-prefix")
117+
Register(lg, client, "test-prefix", cts.l.Addr().String(), 7)
118+
cts.cp, cts.donec = NewClusterProxy(lg, client, cts.l.Addr().String(), "test-prefix")
118119
cts.caddr = cts.l.Addr().String()
119120
pb.RegisterClusterServer(cts.server, cts.cp)
120121
close(servec)

‎proxy/grpcproxy/logger.go

Lines changed: 0 additions & 19 deletions
This file was deleted.

‎proxy/grpcproxy/register.go

Lines changed: 12 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@ import (
2222
"go.etcd.io/etcd/clientv3/concurrency"
2323
"go.etcd.io/etcd/clientv3/naming"
2424

25+
"go.uber.org/zap"
2526
"golang.org/x/time/rate"
2627
gnaming "google.golang.org/grpc/naming"
2728
)
@@ -32,17 +33,17 @@ const registerRetryRate = 1
3233
// Register registers itself as a grpc-proxy server by writing prefixed-key
3334
// with session of specified TTL (in seconds). The returned channel is closed
3435
// when the client's context is canceled.
35-
func Register(c *clientv3.Client, prefix string, addr string, ttl int) <-chan struct{} {
36+
func Register(lg *zap.Logger, c *clientv3.Client, prefix string, addr string, ttl int) <-chan struct{} {
3637
rm := rate.NewLimiter(rate.Limit(registerRetryRate), registerRetryRate)
3738

3839
donec := make(chan struct{})
3940
go func() {
4041
defer close(donec)
4142

4243
for rm.Wait(c.Ctx()) == nil {
43-
ss, err := registerSession(c, prefix, addr, ttl)
44+
ss, err := registerSession(lg, c, prefix, addr, ttl)
4445
if err != nil {
45-
plog.Warningf("failed to create a session %v", err)
46+
lg.Warn("failed to create a session", zap.Error(err))
4647
continue
4748
}
4849
select {
@@ -51,8 +52,8 @@ func Register(c *clientv3.Client, prefix string, addr string, ttl int) <-chan st
5152
return
5253

5354
case <-ss.Done():
54-
plog.Warning("session expired; possible network partition or server restart")
55-
plog.Warning("creating a new session to rejoin")
55+
lg.Warn("session expired; possible network partition or server restart")
56+
lg.Warn("creating a new session to rejoin")
5657
continue
5758
}
5859
}
@@ -61,7 +62,7 @@ func Register(c *clientv3.Client, prefix string, addr string, ttl int) <-chan st
6162
return donec
6263
}
6364

64-
func registerSession(c *clientv3.Client, prefix string, addr string, ttl int) (*concurrency.Session, error) {
65+
func registerSession(lg *zap.Logger, c *clientv3.Client, prefix string, addr string, ttl int) (*concurrency.Session, error) {
6566
ss, err := concurrency.NewSession(c, concurrency.WithTTL(ttl))
6667
if err != nil {
6768
return nil, err
@@ -72,7 +73,11 @@ func registerSession(c *clientv3.Client, prefix string, addr string, ttl int) (*
7273
return nil, err
7374
}
7475

75-
plog.Infof("registered %q with %d-second lease", addr, ttl)
76+
lg.Info(
77+
"registered session with lease",
78+
zap.String("addr", addr),
79+
zap.Int("lease-ttl", ttl),
80+
)
7681
return ss, nil
7782
}
7883

‎proxy/grpcproxy/register_test.go

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@ import (
2323
"go.etcd.io/etcd/integration"
2424
"go.etcd.io/etcd/pkg/testutil"
2525

26+
"go.uber.org/zap"
2627
gnaming "google.golang.org/grpc/naming"
2728
)
2829

@@ -44,7 +45,7 @@ func TestRegister(t *testing.T) {
4445
t.Fatalf("len(ups) expected 0, got %d (%v)", len(ups), ups)
4546
}
4647

47-
donec := Register(cli, testPrefix, paddr, 5)
48+
donec := Register(zap.NewExample(), cli, testPrefix, paddr, 5)
4849

4950
ups, err = wa.Next()
5051
if err != nil {

‎proxy/httpproxy/director.go

Lines changed: 29 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,8 @@ import (
1919
"net/url"
2020
"sync"
2121
"time"
22+
23+
"go.uber.org/zap"
2224
)
2325

2426
// defaultRefreshInterval is the default proxyRefreshIntervalMs value
@@ -31,8 +33,12 @@ func init() {
3133
rand.Seed(time.Now().UnixNano())
3234
}
3335

34-
func newDirector(urlsFunc GetProxyURLs, failureWait time.Duration, refreshInterval time.Duration) *director {
36+
func newDirector(lg *zap.Logger, urlsFunc GetProxyURLs, failureWait time.Duration, refreshInterval time.Duration) *director {
37+
if lg == nil {
38+
lg = zap.NewNop()
39+
}
3540
d := &director{
41+
lg: lg,
3642
uf: urlsFunc,
3743
failureWait: failureWait,
3844
}
@@ -56,7 +62,7 @@ func newDirector(urlsFunc GetProxyURLs, failureWait time.Duration, refreshInterv
5662
for _, e := range es {
5763
sl = append(sl, e.URL.String())
5864
}
59-
plog.Infof("endpoints found %q", sl)
65+
lg.Info("endpoints found", zap.Strings("endpoints", sl))
6066
})
6167
}
6268
time.Sleep(ri)
@@ -68,6 +74,7 @@ func newDirector(urlsFunc GetProxyURLs, failureWait time.Duration, refreshInterv
6874

6975
type director struct {
7076
sync.Mutex
77+
lg *zap.Logger
7178
ep []*endpoint
7279
uf GetProxyURLs
7380
failureWait time.Duration
@@ -81,10 +88,10 @@ func (d *director) refresh() {
8188
for _, u := range urls {
8289
uu, err := url.Parse(u)
8390
if err != nil {
84-
plog.Printf("upstream URL invalid: %v", err)
91+
d.lg.Info("upstream URL invalid", zap.Error(err))
8592
continue
8693
}
87-
endpoints = append(endpoints, newEndpoint(*uu, d.failureWait))
94+
endpoints = append(endpoints, newEndpoint(d.lg, *uu, d.failureWait))
8895
}
8996

9097
// shuffle array to avoid connections being "stuck" to a single endpoint
@@ -109,8 +116,9 @@ func (d *director) endpoints() []*endpoint {
109116
return filtered
110117
}
111118

112-
func newEndpoint(u url.URL, failureWait time.Duration) *endpoint {
119+
func newEndpoint(lg *zap.Logger, u url.URL, failureWait time.Duration) *endpoint {
113120
ep := endpoint{
121+
lg: lg,
114122
URL: u,
115123
Available: true,
116124
failFunc: timedUnavailabilityFunc(failureWait),
@@ -122,6 +130,7 @@ func newEndpoint(u url.URL, failureWait time.Duration) *endpoint {
122130
type endpoint struct {
123131
sync.Mutex
124132

133+
lg *zap.Logger
125134
URL url.URL
126135
Available bool
127136

@@ -138,10 +147,17 @@ func (ep *endpoint) Failed() {
138147
ep.Available = false
139148
ep.Unlock()
140149

141-
plog.Printf("marked endpoint %s unavailable", ep.URL.String())
150+
if ep.lg != nil {
151+
ep.lg.Info("marked endpoint unavailable", zap.String("endpoint", ep.URL.String()))
152+
}
142153

143154
if ep.failFunc == nil {
144-
plog.Printf("no failFunc defined, endpoint %s will be unavailable forever.", ep.URL.String())
155+
if ep.lg != nil {
156+
ep.lg.Info(
157+
"no failFunc defined, endpoint will be unavailable forever",
158+
zap.String("endpoint", ep.URL.String()),
159+
)
160+
}
145161
return
146162
}
147163

@@ -152,7 +168,12 @@ func timedUnavailabilityFunc(wait time.Duration) func(*endpoint) {
152168
return func(ep *endpoint) {
153169
time.AfterFunc(wait, func() {
154170
ep.Available = true
155-
plog.Printf("marked endpoint %s available, to retest connectivity", ep.URL.String())
171+
if ep.lg != nil {
172+
ep.lg.Info(
173+
"marked endpoint available, to retest connectivity",
174+
zap.String("endpoint", ep.URL.String()),
175+
)
176+
}
156177
})
157178
}
158179
}

0 commit comments

Comments
 (0)