This repository has been archived by the owner on May 12, 2021. It is now read-only.
/
client.go
423 lines (361 loc) · 11.1 KB
/
client.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
// Copyright 2017 HyperHQ Inc.
//
// SPDX-License-Identifier: Apache-2.0
//
// gRPC client wrapper
package client
import (
"context"
"fmt"
"net"
"net/url"
"strconv"
"strings"
"time"
"github.com/grpc-ecosystem/grpc-opentracing/go/otgrpc"
"github.com/hashicorp/yamux"
"github.com/mdlayher/vsock"
opentracing "github.com/opentracing/opentracing-go"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
grpcStatus "google.golang.org/grpc/status"
agentgrpc "github.com/kata-containers/agent/protocols/grpc"
)
const (
UnixSocketScheme = "unix"
VSockSocketScheme = "vsock"
HybridVSockScheme = "hvsock"
)
var defaultDialTimeout = 15 * time.Second
var defaultCloseTimeout = 5 * time.Second
var hybridVSockPort uint32
// AgentClient is an agent gRPC client connection wrapper for agentgrpc.AgentServiceClient
type AgentClient struct {
agentgrpc.AgentServiceClient
agentgrpc.HealthClient
conn *grpc.ClientConn
}
type yamuxSessionStream struct {
net.Conn
session *yamux.Session
}
func (y *yamuxSessionStream) Close() error {
waitCh := y.session.CloseChan()
timeout := time.NewTimer(defaultCloseTimeout)
if err := y.Conn.Close(); err != nil {
return err
}
if err := y.session.Close(); err != nil {
return err
}
// block until session is really closed
select {
case <-waitCh:
timeout.Stop()
case <-timeout.C:
return fmt.Errorf("timeout waiting for session close")
}
return nil
}
type dialer func(string, time.Duration) (net.Conn, error)
// NewAgentClient creates a new agent gRPC client and handles both unix and vsock addresses.
//
// Supported sock address formats are:
// - unix://<unix socket path>
// - vsock://<cid>:<port>
// - <unix socket path>
// - hvsock://<path>:<port>. Firecracker implements the virtio-vsock device
// model, and mediates communication between AF_UNIX sockets (on the host end)
// and AF_VSOCK sockets (on the guest end).
func NewAgentClient(ctx context.Context, sock string, enableYamux bool) (*AgentClient, error) {
grpcAddr, parsedAddr, err := parse(sock)
if err != nil {
return nil, err
}
dialOpts := []grpc.DialOption{grpc.WithInsecure(), grpc.WithBlock()}
dialOpts = append(dialOpts, grpc.WithDialer(agentDialer(parsedAddr, enableYamux)))
var tracer opentracing.Tracer
span := opentracing.SpanFromContext(ctx)
// If the context contains a trace span, trace all client comms
if span != nil {
tracer = span.Tracer()
dialOpts = append(dialOpts,
grpc.WithUnaryInterceptor(otgrpc.OpenTracingClientInterceptor(tracer)))
dialOpts = append(dialOpts,
grpc.WithStreamInterceptor(otgrpc.OpenTracingStreamClientInterceptor(tracer)))
}
ctx, cancel := context.WithTimeout(ctx, defaultDialTimeout)
defer cancel()
conn, err := grpc.DialContext(ctx, grpcAddr, dialOpts...)
if err != nil {
return nil, err
}
return &AgentClient{
AgentServiceClient: agentgrpc.NewAgentServiceClient(conn),
HealthClient: agentgrpc.NewHealthClient(conn),
conn: conn,
}, nil
}
// Close an existing connection to the agent gRPC server.
func (c *AgentClient) Close() error {
return c.conn.Close()
}
// vsock scheme is self-defined to be kept from being parsed by grpc.
// Any format starting with "scheme://" will be parsed by grpc and we lose
// all address information because vsock scheme is not supported by grpc.
// Therefore we use the format vsock:<cid>:<port> for vsock address.
//
// See https://github.com/grpc/grpc/blob/master/doc/naming.md
//
// In the long term, we should patch grpc to support vsock scheme and also
// upstream the timed vsock dialer.
func parse(sock string) (string, *url.URL, error) {
addr, err := url.Parse(sock)
if err != nil {
return "", nil, err
}
var grpcAddr string
// validate more
switch addr.Scheme {
case VSockSocketScheme:
if addr.Hostname() == "" || addr.Port() == "" || addr.Path != "" {
return "", nil, grpcStatus.Errorf(codes.InvalidArgument, "Invalid vsock scheme: %s", sock)
}
if _, err := strconv.ParseUint(addr.Hostname(), 10, 32); err != nil {
return "", nil, grpcStatus.Errorf(codes.InvalidArgument, "Invalid vsock cid: %s", sock)
}
if _, err := strconv.ParseUint(addr.Port(), 10, 32); err != nil {
return "", nil, grpcStatus.Errorf(codes.InvalidArgument, "Invalid vsock port: %s", sock)
}
grpcAddr = VSockSocketScheme + ":" + addr.Host
case UnixSocketScheme:
fallthrough
case "":
if (addr.Host == "" && addr.Path == "") || addr.Port() != "" {
return "", nil, grpcStatus.Errorf(codes.InvalidArgument, "Invalid unix scheme: %s", sock)
}
if addr.Host == "" {
grpcAddr = UnixSocketScheme + ":///" + addr.Path
} else {
grpcAddr = UnixSocketScheme + ":///" + addr.Host + "/" + addr.Path
}
case HybridVSockScheme:
if addr.Path == "" {
return "", nil, grpcStatus.Errorf(codes.InvalidArgument, "Invalid hybrid vsock scheme: %s", sock)
}
hvsocket := strings.Split(addr.Path, ":")
if len(hvsocket) != 2 {
return "", nil, grpcStatus.Errorf(codes.InvalidArgument, "Invalid hybrid vsock scheme: %s", sock)
}
// Save port since agent dialer not sent the port to the hybridVSock dialer
var port uint64
if port, err = strconv.ParseUint(hvsocket[1], 10, 32); err != nil {
return "", nil, grpcStatus.Errorf(codes.InvalidArgument, "Invalid hybrid vsock port %s: %v", sock, err)
}
hybridVSockPort = uint32(port)
grpcAddr = HybridVSockScheme + ":" + hvsocket[0]
default:
return "", nil, grpcStatus.Errorf(codes.InvalidArgument, "Invalid scheme: %s", sock)
}
return grpcAddr, addr, nil
}
// This function is meant to run in a go routine since it will send ping
// commands every second. It behaves as a heartbeat to maintain a proper
// communication state with the Yamux server in the agent.
func heartBeat(session *yamux.Session) {
if session == nil {
return
}
for {
if session.IsClosed() {
break
}
session.Ping()
// 1 Hz heartbeat
time.Sleep(time.Second)
}
}
func agentDialer(addr *url.URL, enableYamux bool) dialer {
var d dialer
switch addr.Scheme {
case VSockSocketScheme:
d = vsockDialer
case HybridVSockScheme:
d = HybridVSockDialer
case UnixSocketScheme:
fallthrough
default:
d = unixDialer
}
if !enableYamux {
return d
}
// yamux dialer
return func(sock string, timeout time.Duration) (net.Conn, error) {
conn, err := d(sock, timeout)
if err != nil {
return nil, err
}
defer func() {
if err != nil {
conn.Close()
}
}()
var session *yamux.Session
sessionConfig := yamux.DefaultConfig()
// Disable keepAlive since we don't know how much time a container can be paused
sessionConfig.EnableKeepAlive = false
sessionConfig.ConnectionWriteTimeout = time.Second
session, err = yamux.Client(conn, sessionConfig)
if err != nil {
return nil, err
}
// Start the heartbeat in a separate go routine
go heartBeat(session)
var stream net.Conn
stream, err = session.Open()
if err != nil {
return nil, err
}
y := &yamuxSessionStream{
Conn: stream.(net.Conn),
session: session,
}
return y, nil
}
}
func unixDialer(sock string, timeout time.Duration) (net.Conn, error) {
if strings.HasPrefix(sock, "unix:") {
sock = strings.Trim(sock, "unix:")
}
dialFunc := func() (net.Conn, error) {
return net.DialTimeout("unix", sock, timeout)
}
timeoutErr := grpcStatus.Errorf(codes.DeadlineExceeded, "timed out connecting to unix socket %s", sock)
return commonDialer(timeout, dialFunc, timeoutErr)
}
func parseGrpcVsockAddr(sock string) (uint32, uint32, error) {
sp := strings.Split(sock, ":")
if len(sp) != 3 {
return 0, 0, grpcStatus.Errorf(codes.InvalidArgument, "Invalid vsock address: %s", sock)
}
if sp[0] != VSockSocketScheme {
return 0, 0, grpcStatus.Errorf(codes.InvalidArgument, "Invalid vsock URL scheme: %s", sp[0])
}
cid, err := strconv.ParseUint(sp[1], 10, 32)
if err != nil {
return 0, 0, grpcStatus.Errorf(codes.InvalidArgument, "Invalid vsock cid: %s", sp[1])
}
port, err := strconv.ParseUint(sp[2], 10, 32)
if err != nil {
return 0, 0, grpcStatus.Errorf(codes.InvalidArgument, "Invalid vsock port: %s", sp[2])
}
return uint32(cid), uint32(port), nil
}
func parseGrpcHybridVSockAddr(sock string) (string, uint32, error) {
sp := strings.Split(sock, ":")
// scheme and host are required
if len(sp) < 2 {
return "", 0, grpcStatus.Errorf(codes.InvalidArgument, "Invalid hybrid vsock address: %s", sock)
}
if sp[0] != HybridVSockScheme {
return "", 0, grpcStatus.Errorf(codes.InvalidArgument, "Invalid hybrid vsock URL scheme: %s", sock)
}
port := uint32(0)
// the third is the port
if len(sp) == 3 {
p, err := strconv.ParseUint(sp[2], 10, 32)
if err == nil {
port = uint32(p)
}
}
return sp[1], port, nil
}
// This would bypass the grpc dialer backoff strategy and handle dial timeout
// internally. Because we do not have a large number of concurrent dialers,
// it is not reasonable to have such aggressive backoffs which would kill kata
// containers boot up speed. For more information, see
// https://github.com/grpc/grpc/blob/master/doc/connection-backoff.md
func commonDialer(timeout time.Duration, dialFunc func() (net.Conn, error), timeoutErrMsg error) (net.Conn, error) {
t := time.NewTimer(timeout)
cancel := make(chan bool)
ch := make(chan net.Conn)
go func() {
for {
select {
case <-cancel:
// canceled or channel closed
return
default:
}
conn, err := dialFunc()
if err == nil {
// Send conn back iff timer is not fired
// Otherwise there might be no one left reading it
if t.Stop() {
ch <- conn
} else {
conn.Close()
}
return
}
}
}()
var conn net.Conn
var ok bool
select {
case conn, ok = <-ch:
if !ok {
return nil, timeoutErrMsg
}
case <-t.C:
cancel <- true
return nil, timeoutErrMsg
}
return conn, nil
}
func vsockDialer(sock string, timeout time.Duration) (net.Conn, error) {
cid, port, err := parseGrpcVsockAddr(sock)
if err != nil {
return nil, err
}
dialFunc := func() (net.Conn, error) {
return vsock.Dial(cid, port)
}
timeoutErr := grpcStatus.Errorf(codes.DeadlineExceeded, "timed out connecting to vsock %d:%d", cid, port)
return commonDialer(timeout, dialFunc, timeoutErr)
}
// HybridVSockDialer dials to a hybrid virtio socket
func HybridVSockDialer(sock string, timeout time.Duration) (net.Conn, error) {
udsPath, port, err := parseGrpcHybridVSockAddr(sock)
if err != nil {
return nil, err
}
dialFunc := func() (net.Conn, error) {
conn, err := net.DialTimeout("unix", udsPath, timeout)
if err != nil {
return nil, err
}
if port == 0 {
// use the port read at parse()
port = hybridVSockPort
}
// Once the connection is opened, the following command MUST BE sent,
// the hypervisor needs to know the port number where the agent is listening in order to
// create the connection
if _, err = conn.Write([]byte(fmt.Sprintf("CONNECT %d\n", port))); err != nil {
conn.Close()
return nil, err
}
// Read EOT (End of transmission) byte
eot := make([]byte, 32)
if _, err = conn.Read(eot); err != nil {
// Just close the connection, gRPC will dial again
// without errors
conn.Close()
}
return conn, nil
}
timeoutErr := grpcStatus.Errorf(codes.DeadlineExceeded, "timed out connecting to hybrid vsocket %s", sock)
return commonDialer(timeout, dialFunc, timeoutErr)
}