cubefs

1
/*
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

19
package grpc
20

21
import (
22
	"fmt"
23
	"strings"
24
	"sync"
25
	"time"
26

27
	"google.golang.org/grpc/balancer"
28
	"google.golang.org/grpc/credentials"
29
	"google.golang.org/grpc/internal/channelz"
30
	"google.golang.org/grpc/internal/grpcsync"
31
	"google.golang.org/grpc/resolver"
32
	"google.golang.org/grpc/serviceconfig"
33
)
34

35
// ccResolverWrapper is a wrapper on top of cc for resolvers.
36
// It implements resolver.ClientConn interface.
37
type ccResolverWrapper struct {
38
	cc         *ClientConn
39
	resolverMu sync.Mutex
40
	resolver   resolver.Resolver
41
	done       *grpcsync.Event
42
	curState   resolver.State
43

44
	pollingMu sync.Mutex
45
	polling   chan struct{}
46
}
47

48
// newCCResolverWrapper uses the resolver.Builder to build a Resolver and
49
// returns a ccResolverWrapper object which wraps the newly built resolver.
50
func newCCResolverWrapper(cc *ClientConn, rb resolver.Builder) (*ccResolverWrapper, error) {
51
	ccr := &ccResolverWrapper{
52
		cc:   cc,
53
		done: grpcsync.NewEvent(),
54
	}
55

56
	var credsClone credentials.TransportCredentials
57
	if creds := cc.dopts.copts.TransportCredentials; creds != nil {
58
		credsClone = creds.Clone()
59
	}
60
	rbo := resolver.BuildOptions{
61
		DisableServiceConfig: cc.dopts.disableServiceConfig,
62
		DialCreds:            credsClone,
63
		CredsBundle:          cc.dopts.copts.CredsBundle,
64
		Dialer:               cc.dopts.copts.Dialer,
65
	}
66

67
	var err error
68
	// We need to hold the lock here while we assign to the ccr.resolver field
69
	// to guard against a data race caused by the following code path,
70
	// rb.Build-->ccr.ReportError-->ccr.poll-->ccr.resolveNow, would end up
71
	// accessing ccr.resolver which is being assigned here.
72
	ccr.resolverMu.Lock()
73
	defer ccr.resolverMu.Unlock()
74
	ccr.resolver, err = rb.Build(cc.parsedTarget, ccr, rbo)
75
	if err != nil {
76
		return nil, err
77
	}
78
	return ccr, nil
79
}
80

81
func (ccr *ccResolverWrapper) resolveNow(o resolver.ResolveNowOptions) {
82
	ccr.resolverMu.Lock()
83
	if !ccr.done.HasFired() {
84
		ccr.resolver.ResolveNow(o)
85
	}
86
	ccr.resolverMu.Unlock()
87
}
88

89
func (ccr *ccResolverWrapper) close() {
90
	ccr.resolverMu.Lock()
91
	ccr.resolver.Close()
92
	ccr.done.Fire()
93
	ccr.resolverMu.Unlock()
94
}
95

96
// poll begins or ends asynchronous polling of the resolver based on whether
97
// err is ErrBadResolverState.
98
func (ccr *ccResolverWrapper) poll(err error) {
99
	ccr.pollingMu.Lock()
100
	defer ccr.pollingMu.Unlock()
101
	if err != balancer.ErrBadResolverState {
102
		// stop polling
103
		if ccr.polling != nil {
104
			close(ccr.polling)
105
			ccr.polling = nil
106
		}
107
		return
108
	}
109
	if ccr.polling != nil {
110
		// already polling
111
		return
112
	}
113
	p := make(chan struct{})
114
	ccr.polling = p
115
	go func() {
116
		for i := 0; ; i++ {
117
			ccr.resolveNow(resolver.ResolveNowOptions{})
118
			t := time.NewTimer(ccr.cc.dopts.resolveNowBackoff(i))
119
			select {
120
			case <-p:
121
				t.Stop()
122
				return
123
			case <-ccr.done.Done():
124
				// Resolver has been closed.
125
				t.Stop()
126
				return
127
			case <-t.C:
128
				select {
129
				case <-p:
130
					return
131
				default:
132
				}
133
				// Timer expired; re-resolve.
134
			}
135
		}
136
	}()
137
}
138

139
func (ccr *ccResolverWrapper) UpdateState(s resolver.State) {
140
	if ccr.done.HasFired() {
141
		return
142
	}
143
	channelz.Infof(logger, ccr.cc.channelzID, "ccResolverWrapper: sending update to cc: %v", s)
144
	if channelz.IsOn() {
145
		ccr.addChannelzTraceEvent(s)
146
	}
147
	ccr.curState = s
148
	ccr.poll(ccr.cc.updateResolverState(ccr.curState, nil))
149
}
150

151
func (ccr *ccResolverWrapper) ReportError(err error) {
152
	if ccr.done.HasFired() {
153
		return
154
	}
155
	channelz.Warningf(logger, ccr.cc.channelzID, "ccResolverWrapper: reporting error to cc: %v", err)
156
	ccr.poll(ccr.cc.updateResolverState(resolver.State{}, err))
157
}
158

159
// NewAddress is called by the resolver implementation to send addresses to gRPC.
160
func (ccr *ccResolverWrapper) NewAddress(addrs []resolver.Address) {
161
	if ccr.done.HasFired() {
162
		return
163
	}
164
	channelz.Infof(logger, ccr.cc.channelzID, "ccResolverWrapper: sending new addresses to cc: %v", addrs)
165
	if channelz.IsOn() {
166
		ccr.addChannelzTraceEvent(resolver.State{Addresses: addrs, ServiceConfig: ccr.curState.ServiceConfig})
167
	}
168
	ccr.curState.Addresses = addrs
169
	ccr.poll(ccr.cc.updateResolverState(ccr.curState, nil))
170
}
171

172
// NewServiceConfig is called by the resolver implementation to send service
173
// configs to gRPC.
174
func (ccr *ccResolverWrapper) NewServiceConfig(sc string) {
175
	if ccr.done.HasFired() {
176
		return
177
	}
178
	channelz.Infof(logger, ccr.cc.channelzID, "ccResolverWrapper: got new service config: %v", sc)
179
	if ccr.cc.dopts.disableServiceConfig {
180
		channelz.Info(logger, ccr.cc.channelzID, "Service config lookups disabled; ignoring config")
181
		return
182
	}
183
	scpr := parseServiceConfig(sc)
184
	if scpr.Err != nil {
185
		channelz.Warningf(logger, ccr.cc.channelzID, "ccResolverWrapper: error parsing service config: %v", scpr.Err)
186
		ccr.poll(balancer.ErrBadResolverState)
187
		return
188
	}
189
	if channelz.IsOn() {
190
		ccr.addChannelzTraceEvent(resolver.State{Addresses: ccr.curState.Addresses, ServiceConfig: scpr})
191
	}
192
	ccr.curState.ServiceConfig = scpr
193
	ccr.poll(ccr.cc.updateResolverState(ccr.curState, nil))
194
}
195

196
func (ccr *ccResolverWrapper) ParseServiceConfig(scJSON string) *serviceconfig.ParseResult {
197
	return parseServiceConfig(scJSON)
198
}
199

200
func (ccr *ccResolverWrapper) addChannelzTraceEvent(s resolver.State) {
201
	var updates []string
202
	var oldSC, newSC *ServiceConfig
203
	var oldOK, newOK bool
204
	if ccr.curState.ServiceConfig != nil {
205
		oldSC, oldOK = ccr.curState.ServiceConfig.Config.(*ServiceConfig)
206
	}
207
	if s.ServiceConfig != nil {
208
		newSC, newOK = s.ServiceConfig.Config.(*ServiceConfig)
209
	}
210
	if oldOK != newOK || (oldOK && newOK && oldSC.rawJSONString != newSC.rawJSONString) {
211
		updates = append(updates, "service config updated")
212
	}
213
	if len(ccr.curState.Addresses) > 0 && len(s.Addresses) == 0 {
214
		updates = append(updates, "resolver returned an empty address list")
215
	} else if len(ccr.curState.Addresses) == 0 && len(s.Addresses) > 0 {
216
		updates = append(updates, "resolver returned new addresses")
217
	}
218
	channelz.AddTraceEvent(logger, ccr.cc.channelzID, 0, &channelz.TraceEventDesc{
219
		Desc:     fmt.Sprintf("Resolver state updated: %+v (%v)", s, strings.Join(updates, "; ")),
220
		Severity: channelz.CtInfo,
221
	})
222
}
223