mosn

1
package integrate
2

3
import (
4
	"crypto/tls"
5
	"fmt"
6
	"io/ioutil"
7
	"net"
8
	"net/http"
9
	"testing"
10
	"time"
11

12
	"mosn.io/mosn/pkg/protocol/xprotocol/bolt"
13
	"mosn.io/mosn/pkg/protocol/xprotocol/dubbo"
14
	"mosn.io/mosn/pkg/protocol/xprotocol/dubbothrift"
15
	"mosn.io/mosn/pkg/protocol/xprotocol/tars"
16

17
	"golang.org/x/net/http2"
18
	_ "mosn.io/mosn/pkg/filter/network/proxy"
19
	_ "mosn.io/mosn/pkg/filter/network/streamproxy"
20
	"mosn.io/mosn/pkg/protocol"
21
	"mosn.io/mosn/pkg/types"
22
	"mosn.io/mosn/test/util"
23
	"mosn.io/mosn/test/util/mosn"
24
)
25

26
// TODO: base interface
27
//type Case interface {
28
//	StartProxy()
29
//	Start(tls bool)
30
//	RunCase(n int, interval int)
31
//}
32

33
type TestCase struct {
34
	AppProtocol    types.ProtocolName
35
	MeshProtocol   types.ProtocolName
36
	C              chan error
37
	T              *testing.T
38
	AppServer      util.UpstreamServer
39
	ClientMeshAddr string
40
	ServerMeshAddr string
41
	Finish         chan bool
42
}
43

44
func NewTestCase(t *testing.T, app, mesh types.ProtocolName, server util.UpstreamServer) *TestCase {
45
	return &TestCase{
46
		AppProtocol:  app,
47
		MeshProtocol: mesh,
48
		C:            make(chan error),
49
		T:            t,
50
		AppServer:    server,
51
		Finish:       make(chan bool),
52
	}
53
}
54

55
// client - mesh - server
56
// not support tls
57
// ignore parameter : mesh protocol
58
func (c *TestCase) StartProxy() {
59
	c.AppServer.GoServe()
60
	appAddr := c.AppServer.Addr()
61
	clientMeshAddr := util.CurrentMeshAddr()
62
	c.ClientMeshAddr = clientMeshAddr
63
	cfg := util.CreateProxyMesh(clientMeshAddr, []string{appAddr}, c.AppProtocol)
64
	mesh := mosn.NewMosn(cfg)
65
	go mesh.Start()
66
	go func() {
67
		<-c.Finish
68
		c.AppServer.Close()
69
		mesh.Close()
70
		c.Finish <- true
71
	}()
72
	time.Sleep(5 * time.Second) //wait server and mesh start
73
}
74

75
// client - mesh - mesh - server
76
func (c *TestCase) Start(tls bool) {
77
	c.AppServer.GoServe()
78
	appAddr := c.AppServer.Addr()
79
	clientMeshAddr := util.CurrentMeshAddr()
80
	c.ClientMeshAddr = clientMeshAddr
81
	serverMeshAddr := util.CurrentMeshAddr()
82
	cfg := util.CreateMeshToMeshConfig(clientMeshAddr, serverMeshAddr, c.AppProtocol, c.MeshProtocol, []string{appAddr}, tls)
83
	mesh := mosn.NewMosn(cfg)
84
	go mesh.Start()
85
	go func() {
86
		<-c.Finish
87
		c.AppServer.Close()
88
		mesh.Close()
89
		c.Finish <- true
90
	}()
91
	time.Sleep(5 * time.Second) //wait server and mesh start
92
}
93

94
// Finish case and wait close returns
95
func (c *TestCase) FinishCase() {
96
	c.Finish <- true
97
	<-c.Finish
98
}
99

100
const HTTPTestPath = "test/path"
101

102
// mesh to mesh use tls if "istls" is true
103
// client do "n" times request, interval time (ms)
104
func (c *TestCase) RunCase(n int, interval int) {
105
	// Client Call
106
	var call func() error
107
	switch c.AppProtocol {
108
	case protocol.HTTP1:
109
		call = func() error {
110
			resp, err := http.Get(fmt.Sprintf("http://%s/%s", c.ClientMeshAddr, HTTPTestPath))
111
			if err != nil {
112
				return err
113
			}
114
			defer resp.Body.Close()
115
			if resp.StatusCode != http.StatusOK {
116
				return fmt.Errorf("response status: %d", resp.StatusCode)
117
			}
118
			b, err := ioutil.ReadAll(resp.Body)
119
			if err != nil {
120
				return err
121
			}
122
			c.T.Logf("HTTP client receive data: %s\n", string(b))
123
			return nil
124
		}
125
	case protocol.HTTP2:
126
		tr := &http2.Transport{
127
			AllowHTTP: true,
128
			DialTLS: func(netw, addr string, cfg *tls.Config) (net.Conn, error) {
129
				return net.Dial(netw, addr)
130
			},
131
		}
132
		httpClient := http.Client{Transport: tr}
133
		call = func() error {
134
			resp, err := httpClient.Get(fmt.Sprintf("http://%s/%s", c.ClientMeshAddr, HTTPTestPath))
135
			if err != nil {
136
				return err
137
			}
138
			defer resp.Body.Close()
139
			if resp.StatusCode != http.StatusOK {
140
				return fmt.Errorf("response status: %d", resp.StatusCode)
141

142
			}
143
			b, err := ioutil.ReadAll(resp.Body)
144
			if err != nil {
145
				return err
146
			}
147
			c.T.Logf("HTTP2 client receive data: %s\n", string(b))
148
			return nil
149
		}
150
	default:
151
		c.C <- fmt.Errorf("unsupported protocol: %v", c.AppProtocol)
152
		return
153
	}
154
	for i := 0; i < n; i++ {
155
		if err := call(); err != nil {
156
			c.C <- err
157
			return
158
		}
159
		time.Sleep(time.Duration(interval) * time.Millisecond)
160
	}
161
	c.C <- nil
162
}
163

164
// XProtocol case, should use sub-protocol
165
type XTestCase struct {
166
	TestCase
167
	SubProtocol types.ProtocolName
168
}
169

170
func NewXTestCase(t *testing.T, subProtocol types.ProtocolName, server util.UpstreamServer) *XTestCase {
171
	return &XTestCase{
172
		TestCase: TestCase{
173
			AppProtocol:  types.ProtocolName("X"),
174
			MeshProtocol: types.ProtocolName("X"),
175
			C:            make(chan error),
176
			T:            t,
177
			AppServer:    server,
178
			Finish:       make(chan bool),
179
		},
180
		SubProtocol: subProtocol,
181
	}
182
}
183

184
// client - mesh - server
185
// not support tls
186
// ignore parameter : mesh protocol
187
func (c *XTestCase) StartProxy() {
188
	c.AppServer.GoServe()
189
	appAddr := c.AppServer.Addr()
190
	clientMeshAddr := util.CurrentMeshAddr()
191
	c.ClientMeshAddr = clientMeshAddr
192
	cfg := util.CreateXProtocolProxyMesh(clientMeshAddr, []string{appAddr}, c.SubProtocol)
193
	mesh := mosn.NewMosn(cfg)
194
	go mesh.Start()
195
	go func() {
196
		<-c.Finish
197
		c.AppServer.Close()
198
		mesh.Close()
199
		c.Finish <- true
200
	}()
201
	time.Sleep(5 * time.Second) //wait server and mesh start
202
}
203

204
func (c *XTestCase) Start(tls bool) {
205
	c.AppServer.GoServe()
206
	appAddr := c.AppServer.Addr()
207
	clientMeshAddr := util.CurrentMeshAddr()
208
	c.ClientMeshAddr = clientMeshAddr
209
	serverMeshAddr := util.CurrentMeshAddr()
210
	cfg := util.CreateXProtocolMesh(clientMeshAddr, serverMeshAddr, c.SubProtocol, []string{appAddr}, tls)
211
	mesh := mosn.NewMosn(cfg)
212
	go mesh.Start()
213
	go func() {
214
		<-c.Finish
215
		c.AppServer.Close()
216
		mesh.Close()
217
		c.Finish <- true
218
	}()
219
	time.Sleep(5 * time.Second) //wait server and mesh start
220
}
221

222
// mesh to mesh use tls if "istls" is true
223
// client do "n" times request, interval time (ms)
224
func (c *XTestCase) RunCase(n int, interval int) {
225
	// Client Call
226
	var call func() error
227
	switch c.SubProtocol {
228
	case bolt.ProtocolName, dubbo.ProtocolName, tars.ProtocolName, dubbothrift.ProtocolName:
229
		server, ok := c.AppServer.(*util.RPCServer)
230
		if !ok {
231
			c.C <- fmt.Errorf("need a xprotocol rpc server")
232
			return
233
		}
234
		client := server.Client
235
		if err := client.Connect(c.ClientMeshAddr); err != nil {
236
			c.C <- err
237
			return
238
		}
239
		defer client.Close()
240
		call = func() error {
241
			client.SendRequest()
242
			if !util.WaitMapEmpty(&client.Waits, 2*time.Second) {
243
				return fmt.Errorf("request get no response")
244
			}
245
			return nil
246
		}
247
	default:
248
		c.C <- fmt.Errorf("unsupported protocol: %v", c.AppProtocol)
249
		return
250
	}
251
	for i := 0; i < n; i++ {
252
		if err := call(); err != nil {
253
			c.C <- err
254
			return
255
		}
256
		time.Sleep(time.Duration(interval) * time.Millisecond)
257
	}
258
	c.C <- nil
259
}
260