blob: 6934905b0f6cb7e6fcbd4cd6e9e3f1469b7b93f5 [file] [log] [blame]
Matteo Scandoloa6a3aee2019-11-26 13:30:14 -07001/*
2 *
3 * Copyright 2017 gRPC authors.
4 *
5 * Licensed under the Apache License, Version 2.0 (the "License");
6 * you may not use this file except in compliance with the License.
7 * You may obtain a copy of the License at
8 *
9 * http://www.apache.org/licenses/LICENSE-2.0
10 *
11 * Unless required by applicable law or agreed to in writing, software
12 * distributed under the License is distributed on an "AS IS" BASIS,
13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 * See the License for the specific language governing permissions and
15 * limitations under the License.
16 *
17 */
18
19package grpc
20
21import (
22 "fmt"
23 "strings"
24 "sync/atomic"
25
26 "google.golang.org/grpc/grpclog"
27 "google.golang.org/grpc/internal/channelz"
28 "google.golang.org/grpc/resolver"
29)
30
31// ccResolverWrapper is a wrapper on top of cc for resolvers.
32// It implements resolver.ClientConnection interface.
33type ccResolverWrapper struct {
34 cc *ClientConn
35 resolver resolver.Resolver
36 addrCh chan []resolver.Address
37 scCh chan string
38 done uint32 // accessed atomically; set to 1 when closed.
39 curState resolver.State
40}
41
42// split2 returns the values from strings.SplitN(s, sep, 2).
43// If sep is not found, it returns ("", "", false) instead.
44func split2(s, sep string) (string, string, bool) {
45 spl := strings.SplitN(s, sep, 2)
46 if len(spl) < 2 {
47 return "", "", false
48 }
49 return spl[0], spl[1], true
50}
51
52// parseTarget splits target into a struct containing scheme, authority and
53// endpoint.
54//
55// If target is not a valid scheme://authority/endpoint, it returns {Endpoint:
56// target}.
57func parseTarget(target string) (ret resolver.Target) {
58 var ok bool
59 ret.Scheme, ret.Endpoint, ok = split2(target, "://")
60 if !ok {
61 return resolver.Target{Endpoint: target}
62 }
63 ret.Authority, ret.Endpoint, ok = split2(ret.Endpoint, "/")
64 if !ok {
65 return resolver.Target{Endpoint: target}
66 }
67 return ret
68}
69
70// newCCResolverWrapper parses cc.target for scheme and gets the resolver
71// builder for this scheme and builds the resolver. The monitoring goroutine
72// for it is not started yet and can be created by calling start().
73//
74// If withResolverBuilder dial option is set, the specified resolver will be
75// used instead.
76func newCCResolverWrapper(cc *ClientConn) (*ccResolverWrapper, error) {
77 rb := cc.dopts.resolverBuilder
78 if rb == nil {
79 return nil, fmt.Errorf("could not get resolver for scheme: %q", cc.parsedTarget.Scheme)
80 }
81
82 ccr := &ccResolverWrapper{
83 cc: cc,
84 addrCh: make(chan []resolver.Address, 1),
85 scCh: make(chan string, 1),
86 }
87
88 var err error
89 ccr.resolver, err = rb.Build(cc.parsedTarget, ccr, resolver.BuildOption{DisableServiceConfig: cc.dopts.disableServiceConfig})
90 if err != nil {
91 return nil, err
92 }
93 return ccr, nil
94}
95
96func (ccr *ccResolverWrapper) resolveNow(o resolver.ResolveNowOption) {
97 ccr.resolver.ResolveNow(o)
98}
99
100func (ccr *ccResolverWrapper) close() {
101 ccr.resolver.Close()
102 atomic.StoreUint32(&ccr.done, 1)
103}
104
105func (ccr *ccResolverWrapper) isDone() bool {
106 return atomic.LoadUint32(&ccr.done) == 1
107}
108
109func (ccr *ccResolverWrapper) UpdateState(s resolver.State) {
110 if ccr.isDone() {
111 return
112 }
113 grpclog.Infof("ccResolverWrapper: sending update to cc: %v", s)
114 if channelz.IsOn() {
115 ccr.addChannelzTraceEvent(s)
116 }
117 ccr.cc.updateResolverState(s)
118 ccr.curState = s
119}
120
121// NewAddress is called by the resolver implementation to send addresses to gRPC.
122func (ccr *ccResolverWrapper) NewAddress(addrs []resolver.Address) {
123 if ccr.isDone() {
124 return
125 }
126 grpclog.Infof("ccResolverWrapper: sending new addresses to cc: %v", addrs)
127 if channelz.IsOn() {
128 ccr.addChannelzTraceEvent(resolver.State{Addresses: addrs, ServiceConfig: ccr.curState.ServiceConfig})
129 }
130 ccr.curState.Addresses = addrs
131 ccr.cc.updateResolverState(ccr.curState)
132}
133
134// NewServiceConfig is called by the resolver implementation to send service
135// configs to gRPC.
136func (ccr *ccResolverWrapper) NewServiceConfig(sc string) {
137 if ccr.isDone() {
138 return
139 }
140 grpclog.Infof("ccResolverWrapper: got new service config: %v", sc)
141 c, err := parseServiceConfig(sc)
142 if err != nil {
143 return
144 }
145 if channelz.IsOn() {
146 ccr.addChannelzTraceEvent(resolver.State{Addresses: ccr.curState.Addresses, ServiceConfig: c})
147 }
148 ccr.curState.ServiceConfig = c
149 ccr.cc.updateResolverState(ccr.curState)
150}
151
152func (ccr *ccResolverWrapper) addChannelzTraceEvent(s resolver.State) {
153 var updates []string
154 oldSC, oldOK := ccr.curState.ServiceConfig.(*ServiceConfig)
155 newSC, newOK := s.ServiceConfig.(*ServiceConfig)
156 if oldOK != newOK || (oldOK && newOK && oldSC.rawJSONString != newSC.rawJSONString) {
157 updates = append(updates, "service config updated")
158 }
159 if len(ccr.curState.Addresses) > 0 && len(s.Addresses) == 0 {
160 updates = append(updates, "resolver returned an empty address list")
161 } else if len(ccr.curState.Addresses) == 0 && len(s.Addresses) > 0 {
162 updates = append(updates, "resolver returned new addresses")
163 }
164 channelz.AddTraceEvent(ccr.cc.channelzID, &channelz.TraceEventDesc{
165 Desc: fmt.Sprintf("Resolver state updated: %+v (%v)", s, strings.Join(updates, "; ")),
166 Severity: channelz.CtINFO,
167 })
168}