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
19package grpc
20
21import (
22	"fmt"
23	"strings"
24	"sync"
25
26	"google.golang.org/grpc/balancer"
27	"google.golang.org/grpc/credentials"
28	"google.golang.org/grpc/internal/channelz"
29	"google.golang.org/grpc/internal/grpcsync"
30	"google.golang.org/grpc/resolver"
31	"google.golang.org/grpc/serviceconfig"
32)
33
34// ccResolverWrapper is a wrapper on top of cc for resolvers.
35// It implements resolver.ClientConn interface.
36type ccResolverWrapper struct {
37	cc         *ClientConn
38	resolverMu sync.Mutex
39	resolver   resolver.Resolver
40	done       *grpcsync.Event
41	curState   resolver.State
42
43	incomingMu sync.Mutex // Synchronizes all the incoming calls.
44}
45
46// newCCResolverWrapper uses the resolver.Builder to build a Resolver and
47// returns a ccResolverWrapper object which wraps the newly built resolver.
48func newCCResolverWrapper(cc *ClientConn, rb resolver.Builder) (*ccResolverWrapper, error) {
49	ccr := &ccResolverWrapper{
50		cc:   cc,
51		done: grpcsync.NewEvent(),
52	}
53
54	var credsClone credentials.TransportCredentials
55	if creds := cc.dopts.copts.TransportCredentials; creds != nil {
56		credsClone = creds.Clone()
57	}
58	rbo := resolver.BuildOptions{
59		DisableServiceConfig: cc.dopts.disableServiceConfig,
60		DialCreds:            credsClone,
61		CredsBundle:          cc.dopts.copts.CredsBundle,
62		Dialer:               cc.dopts.copts.Dialer,
63	}
64
65	var err error
66	// We need to hold the lock here while we assign to the ccr.resolver field
67	// to guard against a data race caused by the following code path,
68	// rb.Build-->ccr.ReportError-->ccr.poll-->ccr.resolveNow, would end up
69	// accessing ccr.resolver which is being assigned here.
70	ccr.resolverMu.Lock()
71	defer ccr.resolverMu.Unlock()
72	ccr.resolver, err = rb.Build(cc.parsedTarget, ccr, rbo)
73	if err != nil {
74		return nil, err
75	}
76	return ccr, nil
77}
78
79func (ccr *ccResolverWrapper) resolveNow(o resolver.ResolveNowOptions) {
80	ccr.resolverMu.Lock()
81	if !ccr.done.HasFired() {
82		ccr.resolver.ResolveNow(o)
83	}
84	ccr.resolverMu.Unlock()
85}
86
87func (ccr *ccResolverWrapper) close() {
88	ccr.resolverMu.Lock()
89	ccr.resolver.Close()
90	ccr.done.Fire()
91	ccr.resolverMu.Unlock()
92}
93
94func (ccr *ccResolverWrapper) UpdateState(s resolver.State) error {
95	ccr.incomingMu.Lock()
96	defer ccr.incomingMu.Unlock()
97	if ccr.done.HasFired() {
98		return nil
99	}
100	channelz.Infof(logger, ccr.cc.channelzID, "ccResolverWrapper: sending update to cc: %v", s)
101	if channelz.IsOn() {
102		ccr.addChannelzTraceEvent(s)
103	}
104	ccr.curState = s
105	if err := ccr.cc.updateResolverState(ccr.curState, nil); err == balancer.ErrBadResolverState {
106		return balancer.ErrBadResolverState
107	}
108	return nil
109}
110
111func (ccr *ccResolverWrapper) ReportError(err error) {
112	ccr.incomingMu.Lock()
113	defer ccr.incomingMu.Unlock()
114	if ccr.done.HasFired() {
115		return
116	}
117	channelz.Warningf(logger, ccr.cc.channelzID, "ccResolverWrapper: reporting error to cc: %v", err)
118	ccr.cc.updateResolverState(resolver.State{}, err)
119}
120
121// NewAddress is called by the resolver implementation to send addresses to gRPC.
122func (ccr *ccResolverWrapper) NewAddress(addrs []resolver.Address) {
123	ccr.incomingMu.Lock()
124	defer ccr.incomingMu.Unlock()
125	if ccr.done.HasFired() {
126		return
127	}
128	channelz.Infof(logger, ccr.cc.channelzID, "ccResolverWrapper: sending new addresses to cc: %v", addrs)
129	if channelz.IsOn() {
130		ccr.addChannelzTraceEvent(resolver.State{Addresses: addrs, ServiceConfig: ccr.curState.ServiceConfig})
131	}
132	ccr.curState.Addresses = addrs
133	ccr.cc.updateResolverState(ccr.curState, nil)
134}
135
136// NewServiceConfig is called by the resolver implementation to send service
137// configs to gRPC.
138func (ccr *ccResolverWrapper) NewServiceConfig(sc string) {
139	ccr.incomingMu.Lock()
140	defer ccr.incomingMu.Unlock()
141	if ccr.done.HasFired() {
142		return
143	}
144	channelz.Infof(logger, ccr.cc.channelzID, "ccResolverWrapper: got new service config: %v", sc)
145	if ccr.cc.dopts.disableServiceConfig {
146		channelz.Info(logger, ccr.cc.channelzID, "Service config lookups disabled; ignoring config")
147		return
148	}
149	scpr := parseServiceConfig(sc)
150	if scpr.Err != nil {
151		channelz.Warningf(logger, ccr.cc.channelzID, "ccResolverWrapper: error parsing service config: %v", scpr.Err)
152		return
153	}
154	if channelz.IsOn() {
155		ccr.addChannelzTraceEvent(resolver.State{Addresses: ccr.curState.Addresses, ServiceConfig: scpr})
156	}
157	ccr.curState.ServiceConfig = scpr
158	ccr.cc.updateResolverState(ccr.curState, nil)
159}
160
161func (ccr *ccResolverWrapper) ParseServiceConfig(scJSON string) *serviceconfig.ParseResult {
162	return parseServiceConfig(scJSON)
163}
164
165func (ccr *ccResolverWrapper) addChannelzTraceEvent(s resolver.State) {
166	var updates []string
167	var oldSC, newSC *ServiceConfig
168	var oldOK, newOK bool
169	if ccr.curState.ServiceConfig != nil {
170		oldSC, oldOK = ccr.curState.ServiceConfig.Config.(*ServiceConfig)
171	}
172	if s.ServiceConfig != nil {
173		newSC, newOK = s.ServiceConfig.Config.(*ServiceConfig)
174	}
175	if oldOK != newOK || (oldOK && newOK && oldSC.rawJSONString != newSC.rawJSONString) {
176		updates = append(updates, "service config updated")
177	}
178	if len(ccr.curState.Addresses) > 0 && len(s.Addresses) == 0 {
179		updates = append(updates, "resolver returned an empty address list")
180	} else if len(ccr.curState.Addresses) == 0 && len(s.Addresses) > 0 {
181		updates = append(updates, "resolver returned new addresses")
182	}
183	channelz.AddTraceEvent(logger, ccr.cc.channelzID, 0, &channelz.TraceEventDesc{
184		Desc:     fmt.Sprintf("Resolver state updated: %+v (%v)", s, strings.Join(updates, "; ")),
185		Severity: channelz.CtInfo,
186	})
187}
188