mosn

1
//go:build MOSNTest
2
// +build MOSNTest
3

4
package simple
5

6
import (
7
	"bytes"
8
	"context"
9
	"crypto/tls"
10
	"io"
11
	"io/ioutil"
12
	"net"
13
	goHttp "net/http"
14
	"strings"
15
	"testing"
16
	"time"
17

18
	"golang.org/x/net/http2"
19
	"google.golang.org/grpc"
20
	"google.golang.org/grpc/codes"
21
	pb "google.golang.org/grpc/examples/features/proto/echo"
22
	"google.golang.org/grpc/status"
23

24
	"mosn.io/mosn/pkg/module/http2/h2c"
25
	. "mosn.io/mosn/test/framework"
26
	"mosn.io/mosn/test/lib/mosn"
27
)
28

29
var largeBody = make([]byte, 1<<18)
30

31
func TestHttp2NotUseStream(t *testing.T) {
32
	Scenario(t, "http2 not use stream", func() {
33
		var m *mosn.MosnOperator
34
		var server *goHttp.Server
35
		Setup(func() {
36
			m = mosn.StartMosn(ConfigSimpleHTTP2)
37
			Verify(m, NotNil)
38
			time.Sleep(2 * time.Second) // wait mosn start
39
		})
40
		Case("client-mosn-server", func() {
41
			go func() {
42
				h2s := &http2.Server{}
43
				handler := goHttp.HandlerFunc(func(w goHttp.ResponseWriter, r *goHttp.Request) {
44
					var body []byte
45
					var err error
46
					if body, err = ioutil.ReadAll(r.Body); err != nil {
47
						t.Fatalf("error request body %v", err)
48
					}
49
					w.Header().Set("Content-Type", "text/plain")
50
					w.Write(body)
51
				})
52
				server = &goHttp.Server{
53
					Addr:    "0.0.0.0:8080",
54
					Handler: h2c.NewHandler(handler, h2s),
55
				}
56
				_ = server.ListenAndServe()
57
			}()
58

59
			time.Sleep(time.Second)
60

61
			testcases := []struct {
62
				reqBody []byte
63
			}{
64
				{
65
					reqBody: []byte("xxxxx"),
66
				},
67
				{
68
					reqBody: largeBody,
69
				},
70
			}
71

72
			for _, tc := range testcases {
73
				client := goHttp.Client{
74
					Transport: &http2.Transport{
75
						AllowHTTP: true,
76
						DialTLS: func(network, addr string, cfg *tls.Config) (net.Conn, error) {
77
							return net.Dial(network, addr)
78
						},
79
					},
80
				}
81
				resp, err := client.Post("http://localhost:2046", "text/plain", bytes.NewReader(tc.reqBody))
82
				Verify(err, Equal, nil)
83
				respBody, err := ioutil.ReadAll(resp.Body)
84
				Verify(err, Equal, nil)
85
				Verify(len(respBody), Equal, len(tc.reqBody))
86
			}
87
		})
88
		TearDown(func() {
89
			m.Stop()
90
			_ = server.Shutdown(context.TODO())
91
		})
92
	})
93
}
94

95
func TestHttp2UseStream(t *testing.T) {
96
	Scenario(t, "http2 use stream", func() {
97
		var m *mosn.MosnOperator
98
		var server *goHttp.Server
99
		Setup(func() {
100
			m = mosn.StartMosn(ConfigSimpleHTTP2UseStream)
101
			Verify(m, NotNil)
102
			time.Sleep(2 * time.Second) // wait mosn start
103
		})
104
		Case("client-mosn-server", func() {
105
			go func() {
106
				h2s := &http2.Server{}
107
				handler := goHttp.HandlerFunc(func(w goHttp.ResponseWriter, r *goHttp.Request) {
108
					var body []byte
109
					var err error
110
					if body, err = ioutil.ReadAll(r.Body); err != nil {
111
						t.Fatalf("error request body %v", err)
112
					}
113
					w.Header().Set("Content-Type", "text/plain")
114
					w.Write(body)
115
				})
116
				server = &goHttp.Server{
117
					Addr:    "0.0.0.0:8080",
118
					Handler: h2c.NewHandler(handler, h2s),
119
				}
120
				_ = server.ListenAndServe()
121
			}()
122

123
			time.Sleep(time.Second)
124

125
			testcases := []struct {
126
				reqBody []byte
127
			}{
128
				{
129
					reqBody: []byte("xxxxx"),
130
				},
131
				{
132
					reqBody: largeBody,
133
				},
134
			}
135

136
			for _, tc := range testcases {
137
				client := goHttp.Client{
138
					Transport: &http2.Transport{
139
						AllowHTTP: true,
140
						DialTLS: func(network, addr string, cfg *tls.Config) (net.Conn, error) {
141
							return net.Dial(network, addr)
142
						},
143
					},
144
				}
145
				resp, err := client.Post("http://localhost:2046", "text/plain", bytes.NewReader(tc.reqBody))
146
				Verify(err, Equal, nil)
147
				respBody, err := ioutil.ReadAll(resp.Body)
148
				Verify(err, Equal, nil)
149
				Verify(len(respBody), Equal, len(tc.reqBody))
150
			}
151
		})
152
		TearDown(func() {
153
			m.Stop()
154
			_ = server.Shutdown(context.TODO())
155
		})
156
	})
157
}
158

159
type bidirectionalStreamingGrpcServer struct {
160
	pb.UnimplementedEchoServer
161
}
162

163
func (s *bidirectionalStreamingGrpcServer) BidirectionalStreamingEcho(stream pb.Echo_BidirectionalStreamingEchoServer) error {
164
	for {
165
		in, err := stream.Recv()
166
		if err != nil {
167
			if err == io.EOF {
168
				return nil
169
			}
170
			return err
171
		}
172
		stream.Send(&pb.EchoResponse{Message: in.Message})
173
	}
174
}
175

176
func TestGRPCBidirectionalStreaming(t *testing.T) {
177
	Scenario(t, "http2 use stream", func() {
178
		var m *mosn.MosnOperator
179
		var server *grpc.Server
180
		Setup(func() {
181
			m = mosn.StartMosn(ConfigSimpleHTTP2UseStream)
182
			Verify(m, NotNil)
183
			time.Sleep(2 * time.Second) // wait mosn start
184
		})
185
		Case("client-mosn-server", func() {
186
			go func() {
187
				lis, err := net.Listen("tcp", "0.0.0.0:8080")
188
				Verify(err, Equal, nil)
189
				server = grpc.NewServer()
190
				pb.RegisterEchoServer(server, &bidirectionalStreamingGrpcServer{})
191
				server.Serve(lis)
192
			}()
193

194
			time.Sleep(time.Second)
195

196
			// Set up a connection to the server.
197
			conn, err := grpc.Dial("127.0.0.1:2046", grpc.WithInsecure())
198
			Verify(err, Equal, nil)
199
			defer conn.Close()
200

201
			c := pb.NewEchoClient(conn)
202

203
			testcases := []struct {
204
				reqMessage string
205
			}{
206
				{
207
					reqMessage: "hello world",
208
				},
209
			}
210

211
			for _, testcase := range testcases {
212
				// Initiate the stream with a context that supports cancellation.
213
				ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
214
				stream, err := c.BidirectionalStreamingEcho(ctx)
215
				Verify(err, Equal, nil)
216
				// Send some test messages.
217
				for i := 0; i < 5; i++ {
218
					err = stream.Send(&pb.EchoRequest{Message: testcase.reqMessage})
219
					Verify(err, Equal, nil)
220

221
					res, err := stream.Recv()
222
					Verify(status.Code(err), Equal, codes.OK)
223
					Verify(res.Message, Equal, testcase.reqMessage)
224
				}
225
				cancel()
226
			}
227
		})
228
		TearDown(func() {
229
			m.Stop()
230
			server.GracefulStop()
231
		})
232
	})
233
}
234

235
func TestHttp2UseStreamNoRetry(t *testing.T) {
236
	Scenario(t, "http2 use stream no retry", func() {
237
		var m *mosn.MosnOperator
238
		var serverHTTP2, serverHTTP1 *goHttp.Server
239
		Setup(func() {
240
			m = mosn.StartMosn(ConfigSimpleHTTP2UseStreamNoRetry)
241
			Verify(m, NotNil)
242
			time.Sleep(2 * time.Second) // wait mosn start
243
		})
244
		Case("client-mosn-server", func() {
245
			go func() {
246
				h2s := &http2.Server{}
247
				handler := goHttp.HandlerFunc(func(w goHttp.ResponseWriter, r *goHttp.Request) {
248
					_, err := ioutil.ReadAll(r.Body)
249
					if err != nil {
250
						t.Fatalf(err.Error())
251
					}
252
					w.WriteHeader(200)
253
				})
254
				serverHTTP2 = &goHttp.Server{
255
					Addr:    "0.0.0.0:8080",
256
					Handler: h2c.NewHandler(handler, h2s),
257
				}
258
				_ = serverHTTP2.ListenAndServe()
259
			}()
260
			// http1 protocol error
261
			// https://github.com/mosn/mosn/issues/1751
262
			go func() {
263
				serverHTTP1 = &goHttp.Server{Addr: "0.0.0.0:8081", Handler: nil}
264
				serverHTTP1.ListenAndServe()
265
			}()
266

267
			time.Sleep(time.Second)
268

269
			client := goHttp.Client{
270
				Timeout: time.Second * 2,
271
				Transport: &http2.Transport{
272
					AllowHTTP: true,
273
					DialTLS: func(network, addr string, cfg *tls.Config) (net.Conn, error) {
274
						return net.Dial(network, addr)
275
					},
276
				},
277
			}
278
			for i := 0; i < 10; i++ {
279
				_, err := client.Post("http://localhost:2046", "text/plain", bytes.NewReader(largeBody))
280
				if err != nil && strings.Contains(err.Error(), "Client.Timeout exceeded while awaiting headers") {
281
					t.Fatalf("error request hangs")
282
				}
283
			}
284
		})
285
		TearDown(func() {
286
			m.Stop()
287
			_ = serverHTTP1.Shutdown(context.TODO())
288
			_ = serverHTTP2.Shutdown(context.TODO())
289
		})
290
	})
291
}
292

293
type unaryGrpcServer struct {
294
	pb.UnimplementedEchoServer
295
	event chan struct{}
296
}
297

298
func (s *unaryGrpcServer) UnaryEcho(ctx context.Context, req *pb.EchoRequest) (*pb.EchoResponse, error) {
299
	select {
300
	case <-ctx.Done():
301
		close(s.event)
302
	case <-time.After(time.Second):
303
	}
304
	return &pb.EchoResponse{}, nil
305
}
306

307
func TestHttp2TransferResetStream(t *testing.T) {
308
	Scenario(t, "http2 transfer reset stream", func() {
309
		var m *mosn.MosnOperator
310
		var server *grpc.Server
311
		Setup(func() {
312
			m = mosn.StartMosn(ConfigSimpleHTTP2UseStream)
313
			Verify(m, NotNil)
314
			time.Sleep(2 * time.Second) // wait mosn start
315
		})
316
		Case("client-mosn-server", func() {
317
			serverEvent := make(chan struct{})
318
			go func() {
319
				lis, err := net.Listen("tcp", "0.0.0.0:8080")
320
				Verify(err, Equal, nil)
321
				server = grpc.NewServer()
322
				pb.RegisterEchoServer(server, &unaryGrpcServer{
323
					event: serverEvent,
324
				})
325
				server.Serve(lis)
326
			}()
327

328
			time.Sleep(time.Second)
329

330
			// Set up a connection to the server.
331
			conn, err := grpc.Dial("127.0.0.1:2046", grpc.WithInsecure())
332
			Verify(err, Equal, nil)
333
			defer conn.Close()
334

335
			c := pb.NewEchoClient(conn)
336

337
			// Initiate the stream with a context that supports cancellation.
338
			ctx, cancel := context.WithCancel(context.Background())
339
			go func() {
340
				time.Sleep(500 * time.Millisecond)
341
				cancel()
342
			}()
343
			_, err = c.UnaryEcho(ctx, &pb.EchoRequest{
344
				Message: "hello jack",
345
			})
346
			Verify(err, NotEqual, nil)
347
			select {
348
			case <-serverEvent:
349
				t.Log("context transferred successfully")
350
			case <-time.After(500 * time.Millisecond):
351
				t.Fatalf("error context not transferred")
352
			}
353
		})
354
		TearDown(func() {
355
			m.Stop()
356
			server.GracefulStop()
357
		})
358
	})
359
}
360

361
const ConfigSimpleHTTP2UseStream = `{
362
        "servers":[
363
                {
364
                        "default_log_path":"stdout",
365
                        "default_log_level": "ERROR",
366
                        "routers": [
367
                                {
368
                                        "router_config_name":"router_to_server",
369
                                        "virtual_hosts":[{
370
                                                "name":"server_hosts",
371
                                                "domains": ["*"],
372
                                                "routers": [
373
                                                        {
374
                                                                "match":{"prefix":"/"},
375
                                                                "route":{"cluster_name":"server_cluster"}
376
                                                        }
377
                                                ]
378
                                        }]
379
                                }
380
                        ],
381
                        "listeners":[
382
                                {
383
                                        "address":"127.0.0.1:2046",
384
                                        "bind_port": true,
385
                                        "filter_chains": [{
386
                                                "filters": [
387
                                                        {
388
                                                                "type": "proxy",
389
                                                                "config": {
390
                                                                        "downstream_protocol": "Http2",
391
                                                                        "upstream_protocol": "Http2",
392
                                                                        "router_config_name":"router_to_server",
393
                                                                        "extend_config": {
394
                                                                                "http2_use_stream": true
395
                                                                        }
396
                                                                }
397
                                                        }
398
                                                ]
399
                                        }]
400
                                }
401
                        ]
402
                }
403
        ],
404
        "cluster_manager":{
405
                "clusters":[
406
                        {
407
                                "name": "server_cluster",
408
                                "type": "SIMPLE",
409
                                "lb_type": "LB_RANDOM",
410
                                "hosts":[
411
                                        {"address":"127.0.0.1:8080"}
412
                                ]
413
                        }
414
                ]
415
        }
416
}`
417

418
const ConfigSimpleHTTP2UseStreamNoRetry = `{
419
        "servers":[
420
                {
421
                        "default_log_path":"stdout",
422
                        "default_log_level": "ERROR",
423
                        "routers": [
424
                                {
425
                                        "router_config_name":"router_to_server",
426
                                        "virtual_hosts":[{
427
                                                "name":"server_hosts",
428
                                                "domains": ["*"],
429
                                                "routers": [
430
                                                        {
431
                                                                "match":{"prefix":"/"},
432
                                                                "route":{"cluster_name":"server_cluster"}
433
                                                        }
434
                                                ]
435
                                        }]
436
                                }
437
                        ],
438
                        "listeners":[
439
                                {
440
                                        "address":"127.0.0.1:2046",
441
                                        "bind_port": true,
442
                                        "filter_chains": [{
443
                                                "filters": [
444
                                                        {
445
                                                                "type": "proxy",
446
                                                                "config": {
447
                                                                        "downstream_protocol": "Http2",
448
                                                                        "upstream_protocol": "Http2",
449
                                                                        "router_config_name":"router_to_server",
450
                                                                        "extend_config": {
451
                                                                                "http2_use_stream": true
452
                                                                        }
453
                                                                }
454
                                                        }
455
                                                ]
456
                                        }]
457
                                }
458
                        ]
459
                }
460
        ],
461
        "cluster_manager":{
462
                "clusters":[
463
                        {
464
                                "name": "server_cluster",
465
                                "type": "SIMPLE",
466
                                "lb_type": "LB_RANDOM",
467
                                "hosts":[
468
                                        {"address":"127.0.0.1:8080"},
469
                                        {"address":"127.0.0.1:8081"}
470
                                ]
471
                        }
472
                ]
473
        }
474
}`
475