podman

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
	"encoding/json"
23
	"errors"
24
	"fmt"
25

26
	"google.golang.org/grpc/balancer"
27
	"google.golang.org/grpc/connectivity"
28
	internalgrpclog "google.golang.org/grpc/internal/grpclog"
29
	"google.golang.org/grpc/internal/grpcrand"
30
	"google.golang.org/grpc/internal/pretty"
31
	"google.golang.org/grpc/resolver"
32
	"google.golang.org/grpc/serviceconfig"
33
)
34

35
const (
36
	// PickFirstBalancerName is the name of the pick_first balancer.
37
	PickFirstBalancerName = "pick_first"
38
	logPrefix             = "[pick-first-lb %p] "
39
)
40

41
func newPickfirstBuilder() balancer.Builder {
42
	return &pickfirstBuilder{}
43
}
44

45
type pickfirstBuilder struct{}
46

47
func (*pickfirstBuilder) Build(cc balancer.ClientConn, opt balancer.BuildOptions) balancer.Balancer {
48
	b := &pickfirstBalancer{cc: cc}
49
	b.logger = internalgrpclog.NewPrefixLogger(logger, fmt.Sprintf(logPrefix, b))
50
	return b
51
}
52

53
func (*pickfirstBuilder) Name() string {
54
	return PickFirstBalancerName
55
}
56

57
type pfConfig struct {
58
	serviceconfig.LoadBalancingConfig `json:"-"`
59

60
	// If set to true, instructs the LB policy to shuffle the order of the list
61
	// of addresses received from the name resolver before attempting to
62
	// connect to them.
63
	ShuffleAddressList bool `json:"shuffleAddressList"`
64
}
65

66
func (*pickfirstBuilder) ParseConfig(js json.RawMessage) (serviceconfig.LoadBalancingConfig, error) {
67
	var cfg pfConfig
68
	if err := json.Unmarshal(js, &cfg); err != nil {
69
		return nil, fmt.Errorf("pickfirst: unable to unmarshal LB policy config: %s, error: %v", string(js), err)
70
	}
71
	return cfg, nil
72
}
73

74
type pickfirstBalancer struct {
75
	logger  *internalgrpclog.PrefixLogger
76
	state   connectivity.State
77
	cc      balancer.ClientConn
78
	subConn balancer.SubConn
79
}
80

81
func (b *pickfirstBalancer) ResolverError(err error) {
82
	if b.logger.V(2) {
83
		b.logger.Infof("Received error from the name resolver: %v", err)
84
	}
85
	if b.subConn == nil {
86
		b.state = connectivity.TransientFailure
87
	}
88

89
	if b.state != connectivity.TransientFailure {
90
		// The picker will not change since the balancer does not currently
91
		// report an error.
92
		return
93
	}
94
	b.cc.UpdateState(balancer.State{
95
		ConnectivityState: connectivity.TransientFailure,
96
		Picker:            &picker{err: fmt.Errorf("name resolver error: %v", err)},
97
	})
98
}
99

100
func (b *pickfirstBalancer) UpdateClientConnState(state balancer.ClientConnState) error {
101
	addrs := state.ResolverState.Addresses
102
	if len(addrs) == 0 {
103
		// The resolver reported an empty address list. Treat it like an error by
104
		// calling b.ResolverError.
105
		if b.subConn != nil {
106
			// Shut down the old subConn. All addresses were removed, so it is
107
			// no longer valid.
108
			b.subConn.Shutdown()
109
			b.subConn = nil
110
		}
111
		b.ResolverError(errors.New("produced zero addresses"))
112
		return balancer.ErrBadResolverState
113
	}
114

115
	// We don't have to guard this block with the env var because ParseConfig
116
	// already does so.
117
	cfg, ok := state.BalancerConfig.(pfConfig)
118
	if state.BalancerConfig != nil && !ok {
119
		return fmt.Errorf("pickfirst: received illegal BalancerConfig (type %T): %v", state.BalancerConfig, state.BalancerConfig)
120
	}
121
	if cfg.ShuffleAddressList {
122
		addrs = append([]resolver.Address{}, addrs...)
123
		grpcrand.Shuffle(len(addrs), func(i, j int) { addrs[i], addrs[j] = addrs[j], addrs[i] })
124
	}
125

126
	if b.logger.V(2) {
127
		b.logger.Infof("Received new config %s, resolver state %s", pretty.ToJSON(cfg), pretty.ToJSON(state.ResolverState))
128
	}
129

130
	if b.subConn != nil {
131
		b.cc.UpdateAddresses(b.subConn, addrs)
132
		return nil
133
	}
134

135
	var subConn balancer.SubConn
136
	subConn, err := b.cc.NewSubConn(addrs, balancer.NewSubConnOptions{
137
		StateListener: func(state balancer.SubConnState) {
138
			b.updateSubConnState(subConn, state)
139
		},
140
	})
141
	if err != nil {
142
		if b.logger.V(2) {
143
			b.logger.Infof("Failed to create new SubConn: %v", err)
144
		}
145
		b.state = connectivity.TransientFailure
146
		b.cc.UpdateState(balancer.State{
147
			ConnectivityState: connectivity.TransientFailure,
148
			Picker:            &picker{err: fmt.Errorf("error creating connection: %v", err)},
149
		})
150
		return balancer.ErrBadResolverState
151
	}
152
	b.subConn = subConn
153
	b.state = connectivity.Idle
154
	b.cc.UpdateState(balancer.State{
155
		ConnectivityState: connectivity.Connecting,
156
		Picker:            &picker{err: balancer.ErrNoSubConnAvailable},
157
	})
158
	b.subConn.Connect()
159
	return nil
160
}
161

162
// UpdateSubConnState is unused as a StateListener is always registered when
163
// creating SubConns.
164
func (b *pickfirstBalancer) UpdateSubConnState(subConn balancer.SubConn, state balancer.SubConnState) {
165
	b.logger.Errorf("UpdateSubConnState(%v, %+v) called unexpectedly", subConn, state)
166
}
167

168
func (b *pickfirstBalancer) updateSubConnState(subConn balancer.SubConn, state balancer.SubConnState) {
169
	if b.logger.V(2) {
170
		b.logger.Infof("Received SubConn state update: %p, %+v", subConn, state)
171
	}
172
	if b.subConn != subConn {
173
		if b.logger.V(2) {
174
			b.logger.Infof("Ignored state change because subConn is not recognized")
175
		}
176
		return
177
	}
178
	if state.ConnectivityState == connectivity.Shutdown {
179
		b.subConn = nil
180
		return
181
	}
182

183
	switch state.ConnectivityState {
184
	case connectivity.Ready:
185
		b.cc.UpdateState(balancer.State{
186
			ConnectivityState: state.ConnectivityState,
187
			Picker:            &picker{result: balancer.PickResult{SubConn: subConn}},
188
		})
189
	case connectivity.Connecting:
190
		if b.state == connectivity.TransientFailure {
191
			// We stay in TransientFailure until we are Ready. See A62.
192
			return
193
		}
194
		b.cc.UpdateState(balancer.State{
195
			ConnectivityState: state.ConnectivityState,
196
			Picker:            &picker{err: balancer.ErrNoSubConnAvailable},
197
		})
198
	case connectivity.Idle:
199
		if b.state == connectivity.TransientFailure {
200
			// We stay in TransientFailure until we are Ready. Also kick the
201
			// subConn out of Idle into Connecting. See A62.
202
			b.subConn.Connect()
203
			return
204
		}
205
		b.cc.UpdateState(balancer.State{
206
			ConnectivityState: state.ConnectivityState,
207
			Picker:            &idlePicker{subConn: subConn},
208
		})
209
	case connectivity.TransientFailure:
210
		b.cc.UpdateState(balancer.State{
211
			ConnectivityState: state.ConnectivityState,
212
			Picker:            &picker{err: state.ConnectionError},
213
		})
214
	}
215
	b.state = state.ConnectivityState
216
}
217

218
func (b *pickfirstBalancer) Close() {
219
}
220

221
func (b *pickfirstBalancer) ExitIdle() {
222
	if b.subConn != nil && b.state == connectivity.Idle {
223
		b.subConn.Connect()
224
	}
225
}
226

227
type picker struct {
228
	result balancer.PickResult
229
	err    error
230
}
231

232
func (p *picker) Pick(balancer.PickInfo) (balancer.PickResult, error) {
233
	return p.result, p.err
234
}
235

236
// idlePicker is used when the SubConn is IDLE and kicks the SubConn into
237
// CONNECTING when Pick is called.
238
type idlePicker struct {
239
	subConn balancer.SubConn
240
}
241

242
func (i *idlePicker) Pick(balancer.PickInfo) (balancer.PickResult, error) {
243
	i.subConn.Connect()
244
	return balancer.PickResult{}, balancer.ErrNoSubConnAvailable
245
}
246

247
func init() {
248
	balancer.Register(newPickfirstBuilder())
249
}
250