cubefs

1
/*
2
 *
3
 * Copyright 2018 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 dns implements a dns resolver to be installed as the default resolver
20
// in grpc.
21
package dns
22

23
import (
24
	"context"
25
	"encoding/json"
26
	"errors"
27
	"fmt"
28
	"net"
29
	"os"
30
	"strconv"
31
	"strings"
32
	"sync"
33
	"time"
34

35
	grpclbstate "google.golang.org/grpc/balancer/grpclb/state"
36
	"google.golang.org/grpc/grpclog"
37
	"google.golang.org/grpc/internal/envconfig"
38
	"google.golang.org/grpc/internal/grpcrand"
39
	"google.golang.org/grpc/resolver"
40
	"google.golang.org/grpc/serviceconfig"
41
)
42

43
// EnableSRVLookups controls whether the DNS resolver attempts to fetch gRPCLB
44
// addresses from SRV records.  Must not be changed after init time.
45
var EnableSRVLookups = false
46

47
var logger = grpclog.Component("dns")
48

49
func init() {
50
	resolver.Register(NewBuilder())
51
}
52

53
const (
54
	defaultPort       = "443"
55
	defaultDNSSvrPort = "53"
56
	golang            = "GO"
57
	// txtPrefix is the prefix string to be prepended to the host name for txt record lookup.
58
	txtPrefix = "_grpc_config."
59
	// In DNS, service config is encoded in a TXT record via the mechanism
60
	// described in RFC-1464 using the attribute name grpc_config.
61
	txtAttribute = "grpc_config="
62
)
63

64
var (
65
	errMissingAddr = errors.New("dns resolver: missing address")
66

67
	// Addresses ending with a colon that is supposed to be the separator
68
	// between host and port is not allowed.  E.g. "::" is a valid address as
69
	// it is an IPv6 address (host only) and "[::]:" is invalid as it ends with
70
	// a colon as the host and port separator
71
	errEndsWithColon = errors.New("dns resolver: missing port after port-separator colon")
72
)
73

74
var (
75
	defaultResolver netResolver = net.DefaultResolver
76
	// To prevent excessive re-resolution, we enforce a rate limit on DNS
77
	// resolution requests.
78
	minDNSResRate = 30 * time.Second
79
)
80

81
var customAuthorityDialler = func(authority string) func(ctx context.Context, network, address string) (net.Conn, error) {
82
	return func(ctx context.Context, network, address string) (net.Conn, error) {
83
		var dialer net.Dialer
84
		return dialer.DialContext(ctx, network, authority)
85
	}
86
}
87

88
var customAuthorityResolver = func(authority string) (netResolver, error) {
89
	host, port, err := parseTarget(authority, defaultDNSSvrPort)
90
	if err != nil {
91
		return nil, err
92
	}
93

94
	authorityWithPort := net.JoinHostPort(host, port)
95

96
	return &net.Resolver{
97
		PreferGo: true,
98
		Dial:     customAuthorityDialler(authorityWithPort),
99
	}, nil
100
}
101

102
// NewBuilder creates a dnsBuilder which is used to factory DNS resolvers.
103
func NewBuilder() resolver.Builder {
104
	return &dnsBuilder{}
105
}
106

107
type dnsBuilder struct{}
108

109
// Build creates and starts a DNS resolver that watches the name resolution of the target.
110
func (b *dnsBuilder) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOptions) (resolver.Resolver, error) {
111
	host, port, err := parseTarget(target.Endpoint, defaultPort)
112
	if err != nil {
113
		return nil, err
114
	}
115

116
	// IP address.
117
	if ipAddr, ok := formatIP(host); ok {
118
		addr := []resolver.Address{{Addr: ipAddr + ":" + port}}
119
		cc.UpdateState(resolver.State{Addresses: addr})
120
		return deadResolver{}, nil
121
	}
122

123
	// DNS address (non-IP).
124
	ctx, cancel := context.WithCancel(context.Background())
125
	d := &dnsResolver{
126
		host:                 host,
127
		port:                 port,
128
		ctx:                  ctx,
129
		cancel:               cancel,
130
		cc:                   cc,
131
		rn:                   make(chan struct{}, 1),
132
		disableServiceConfig: opts.DisableServiceConfig,
133
	}
134

135
	if target.Authority == "" {
136
		d.resolver = defaultResolver
137
	} else {
138
		d.resolver, err = customAuthorityResolver(target.Authority)
139
		if err != nil {
140
			return nil, err
141
		}
142
	}
143

144
	d.wg.Add(1)
145
	go d.watcher()
146
	d.ResolveNow(resolver.ResolveNowOptions{})
147
	return d, nil
148
}
149

150
// Scheme returns the naming scheme of this resolver builder, which is "dns".
151
func (b *dnsBuilder) Scheme() string {
152
	return "dns"
153
}
154

155
type netResolver interface {
156
	LookupHost(ctx context.Context, host string) (addrs []string, err error)
157
	LookupSRV(ctx context.Context, service, proto, name string) (cname string, addrs []*net.SRV, err error)
158
	LookupTXT(ctx context.Context, name string) (txts []string, err error)
159
}
160

161
// deadResolver is a resolver that does nothing.
162
type deadResolver struct{}
163

164
func (deadResolver) ResolveNow(resolver.ResolveNowOptions) {}
165

166
func (deadResolver) Close() {}
167

168
// dnsResolver watches for the name resolution update for a non-IP target.
169
type dnsResolver struct {
170
	host     string
171
	port     string
172
	resolver netResolver
173
	ctx      context.Context
174
	cancel   context.CancelFunc
175
	cc       resolver.ClientConn
176
	// rn channel is used by ResolveNow() to force an immediate resolution of the target.
177
	rn chan struct{}
178
	// wg is used to enforce Close() to return after the watcher() goroutine has finished.
179
	// Otherwise, data race will be possible. [Race Example] in dns_resolver_test we
180
	// replace the real lookup functions with mocked ones to facilitate testing.
181
	// If Close() doesn't wait for watcher() goroutine finishes, race detector sometimes
182
	// will warns lookup (READ the lookup function pointers) inside watcher() goroutine
183
	// has data race with replaceNetFunc (WRITE the lookup function pointers).
184
	wg                   sync.WaitGroup
185
	disableServiceConfig bool
186
}
187

188
// ResolveNow invoke an immediate resolution of the target that this dnsResolver watches.
189
func (d *dnsResolver) ResolveNow(resolver.ResolveNowOptions) {
190
	select {
191
	case d.rn <- struct{}{}:
192
	default:
193
	}
194
}
195

196
// Close closes the dnsResolver.
197
func (d *dnsResolver) Close() {
198
	d.cancel()
199
	d.wg.Wait()
200
}
201

202
func (d *dnsResolver) watcher() {
203
	defer d.wg.Done()
204
	for {
205
		select {
206
		case <-d.ctx.Done():
207
			return
208
		case <-d.rn:
209
		}
210

211
		state, err := d.lookup()
212
		if err != nil {
213
			d.cc.ReportError(err)
214
		} else {
215
			d.cc.UpdateState(*state)
216
		}
217

218
		// Sleep to prevent excessive re-resolutions. Incoming resolution requests
219
		// will be queued in d.rn.
220
		t := time.NewTimer(minDNSResRate)
221
		select {
222
		case <-t.C:
223
		case <-d.ctx.Done():
224
			t.Stop()
225
			return
226
		}
227
	}
228
}
229

230
func (d *dnsResolver) lookupSRV() ([]resolver.Address, error) {
231
	if !EnableSRVLookups {
232
		return nil, nil
233
	}
234
	var newAddrs []resolver.Address
235
	_, srvs, err := d.resolver.LookupSRV(d.ctx, "grpclb", "tcp", d.host)
236
	if err != nil {
237
		err = handleDNSError(err, "SRV") // may become nil
238
		return nil, err
239
	}
240
	for _, s := range srvs {
241
		lbAddrs, err := d.resolver.LookupHost(d.ctx, s.Target)
242
		if err != nil {
243
			err = handleDNSError(err, "A") // may become nil
244
			if err == nil {
245
				// If there are other SRV records, look them up and ignore this
246
				// one that does not exist.
247
				continue
248
			}
249
			return nil, err
250
		}
251
		for _, a := range lbAddrs {
252
			ip, ok := formatIP(a)
253
			if !ok {
254
				return nil, fmt.Errorf("dns: error parsing A record IP address %v", a)
255
			}
256
			addr := ip + ":" + strconv.Itoa(int(s.Port))
257
			newAddrs = append(newAddrs, resolver.Address{Addr: addr, ServerName: s.Target})
258
		}
259
	}
260
	return newAddrs, nil
261
}
262

263
var filterError = func(err error) error {
264
	if dnsErr, ok := err.(*net.DNSError); ok && !dnsErr.IsTimeout && !dnsErr.IsTemporary {
265
		// Timeouts and temporary errors should be communicated to gRPC to
266
		// attempt another DNS query (with backoff).  Other errors should be
267
		// suppressed (they may represent the absence of a TXT record).
268
		return nil
269
	}
270
	return err
271
}
272

273
func handleDNSError(err error, lookupType string) error {
274
	err = filterError(err)
275
	if err != nil {
276
		err = fmt.Errorf("dns: %v record lookup error: %v", lookupType, err)
277
		logger.Info(err)
278
	}
279
	return err
280
}
281

282
func (d *dnsResolver) lookupTXT() *serviceconfig.ParseResult {
283
	ss, err := d.resolver.LookupTXT(d.ctx, txtPrefix+d.host)
284
	if err != nil {
285
		if envconfig.TXTErrIgnore {
286
			return nil
287
		}
288
		if err = handleDNSError(err, "TXT"); err != nil {
289
			return &serviceconfig.ParseResult{Err: err}
290
		}
291
		return nil
292
	}
293
	var res string
294
	for _, s := range ss {
295
		res += s
296
	}
297

298
	// TXT record must have "grpc_config=" attribute in order to be used as service config.
299
	if !strings.HasPrefix(res, txtAttribute) {
300
		logger.Warningf("dns: TXT record %v missing %v attribute", res, txtAttribute)
301
		// This is not an error; it is the equivalent of not having a service config.
302
		return nil
303
	}
304
	sc := canaryingSC(strings.TrimPrefix(res, txtAttribute))
305
	return d.cc.ParseServiceConfig(sc)
306
}
307

308
func (d *dnsResolver) lookupHost() ([]resolver.Address, error) {
309
	var newAddrs []resolver.Address
310
	addrs, err := d.resolver.LookupHost(d.ctx, d.host)
311
	if err != nil {
312
		err = handleDNSError(err, "A")
313
		return nil, err
314
	}
315
	for _, a := range addrs {
316
		ip, ok := formatIP(a)
317
		if !ok {
318
			return nil, fmt.Errorf("dns: error parsing A record IP address %v", a)
319
		}
320
		addr := ip + ":" + d.port
321
		newAddrs = append(newAddrs, resolver.Address{Addr: addr})
322
	}
323
	return newAddrs, nil
324
}
325

326
func (d *dnsResolver) lookup() (*resolver.State, error) {
327
	srv, srvErr := d.lookupSRV()
328
	addrs, hostErr := d.lookupHost()
329
	if hostErr != nil && (srvErr != nil || len(srv) == 0) {
330
		return nil, hostErr
331
	}
332

333
	state := resolver.State{Addresses: addrs}
334
	if len(srv) > 0 {
335
		state = grpclbstate.Set(state, &grpclbstate.State{BalancerAddresses: srv})
336
	}
337
	if !d.disableServiceConfig {
338
		state.ServiceConfig = d.lookupTXT()
339
	}
340
	return &state, nil
341
}
342

343
// formatIP returns ok = false if addr is not a valid textual representation of an IP address.
344
// If addr is an IPv4 address, return the addr and ok = true.
345
// If addr is an IPv6 address, return the addr enclosed in square brackets and ok = true.
346
func formatIP(addr string) (addrIP string, ok bool) {
347
	ip := net.ParseIP(addr)
348
	if ip == nil {
349
		return "", false
350
	}
351
	if ip.To4() != nil {
352
		return addr, true
353
	}
354
	return "[" + addr + "]", true
355
}
356

357
// parseTarget takes the user input target string and default port, returns formatted host and port info.
358
// If target doesn't specify a port, set the port to be the defaultPort.
359
// If target is in IPv6 format and host-name is enclosed in square brackets, brackets
360
// are stripped when setting the host.
361
// examples:
362
// target: "www.google.com" defaultPort: "443" returns host: "www.google.com", port: "443"
363
// target: "ipv4-host:80" defaultPort: "443" returns host: "ipv4-host", port: "80"
364
// target: "[ipv6-host]" defaultPort: "443" returns host: "ipv6-host", port: "443"
365
// target: ":80" defaultPort: "443" returns host: "localhost", port: "80"
366
func parseTarget(target, defaultPort string) (host, port string, err error) {
367
	if target == "" {
368
		return "", "", errMissingAddr
369
	}
370
	if ip := net.ParseIP(target); ip != nil {
371
		// target is an IPv4 or IPv6(without brackets) address
372
		return target, defaultPort, nil
373
	}
374
	if host, port, err = net.SplitHostPort(target); err == nil {
375
		if port == "" {
376
			// If the port field is empty (target ends with colon), e.g. "[::1]:", this is an error.
377
			return "", "", errEndsWithColon
378
		}
379
		// target has port, i.e ipv4-host:port, [ipv6-host]:port, host-name:port
380
		if host == "" {
381
			// Keep consistent with net.Dial(): If the host is empty, as in ":80", the local system is assumed.
382
			host = "localhost"
383
		}
384
		return host, port, nil
385
	}
386
	if host, port, err = net.SplitHostPort(target + ":" + defaultPort); err == nil {
387
		// target doesn't have port
388
		return host, port, nil
389
	}
390
	return "", "", fmt.Errorf("invalid target address %v, error info: %v", target, err)
391
}
392

393
type rawChoice struct {
394
	ClientLanguage *[]string        `json:"clientLanguage,omitempty"`
395
	Percentage     *int             `json:"percentage,omitempty"`
396
	ClientHostName *[]string        `json:"clientHostName,omitempty"`
397
	ServiceConfig  *json.RawMessage `json:"serviceConfig,omitempty"`
398
}
399

400
func containsString(a *[]string, b string) bool {
401
	if a == nil {
402
		return true
403
	}
404
	for _, c := range *a {
405
		if c == b {
406
			return true
407
		}
408
	}
409
	return false
410
}
411

412
func chosenByPercentage(a *int) bool {
413
	if a == nil {
414
		return true
415
	}
416
	return grpcrand.Intn(100)+1 <= *a
417
}
418

419
func canaryingSC(js string) string {
420
	if js == "" {
421
		return ""
422
	}
423
	var rcs []rawChoice
424
	err := json.Unmarshal([]byte(js), &rcs)
425
	if err != nil {
426
		logger.Warningf("dns: error parsing service config json: %v", err)
427
		return ""
428
	}
429
	cliHostname, err := os.Hostname()
430
	if err != nil {
431
		logger.Warningf("dns: error getting client hostname: %v", err)
432
		return ""
433
	}
434
	var sc string
435
	for _, c := range rcs {
436
		if !containsString(c.ClientLanguage, golang) ||
437
			!chosenByPercentage(c.Percentage) ||
438
			!containsString(c.ClientHostName, cliHostname) ||
439
			c.ServiceConfig == nil {
440
			continue
441
		}
442
		sc = string(*c.ServiceConfig)
443
		break
444
	}
445
	return sc
446
}
447