mosn

1
/*
2
 * Licensed to the Apache Software Foundation (ASF) under one or more
3
 * contributor license agreements.  See the NOTICE file distributed with
4
 * this work for additional information regarding copyright ownership.
5
 * The ASF licenses this file to You under the Apache License, Version 2.0
6
 * (the "License"); you may not use this file except in compliance with
7
 * the License.  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
package integrate
19

20
import (
21
	"math"
22
	"testing"
23
	"time"
24

25
	"mosn.io/mosn/pkg/protocol/xprotocol/bolt"
26

27
	testutil "mosn.io/mosn/test/util"
28
	"mosn.io/mosn/test/util/mosn"
29
)
30

31
type weightCase struct {
32
	*XTestCase
33
	appServers map[string]testutil.UpstreamServer
34
	clusters   []*testutil.WeightCluster
35
}
36

37
// support SofaRPC with xprotocol only
38
func newWeightCase(t *testing.T, clusters []*testutil.WeightCluster) *weightCase {
39
	tc := NewXTestCase(t, bolt.ProtocolName, testutil.NewRPCServer(t, "", bolt.ProtocolName))
40
	appServers := make(map[string]testutil.UpstreamServer)
41
	for _, cluster := range clusters {
42
		for _, host := range cluster.Hosts {
43
			appServers[host.Addr] = testutil.NewRPCServer(t, host.Addr, bolt.ProtocolName)
44
		}
45
	}
46
	return &weightCase{
47
		XTestCase:  tc,
48
		appServers: appServers,
49
		clusters:   clusters,
50
	}
51
}
52

53
func (c *weightCase) Start() {
54
	for _, appserver := range c.appServers {
55
		appserver.GoServe()
56
	}
57

58
	meshAddr := testutil.CurrentMeshAddr()
59
	c.ClientMeshAddr = meshAddr
60
	// use bolt as example
61
	cfg := testutil.CreateXWeightProxyMesh(meshAddr, bolt.ProtocolName, c.clusters)
62
	mesh := mosn.NewMosn(cfg)
63
	go mesh.Start()
64
	go func() {
65
		<-c.Finish
66
		for _, appserver := range c.appServers {
67
			appserver.Close()
68
		}
69
		mesh.Close()
70
		c.Finish <- true
71
	}()
72
	time.Sleep(5 * time.Second) //wait server and mesh start
73
}
74

75
const thres float64 = 0.3
76

77
func (c *weightCase) Verify(t *testing.T) {
78
	clusterCounts := []uint32{}
79
	clusterTotalCount := uint32(0)
80
	for index, cluster := range c.clusters {
81
		hostWeights := []uint32{}
82
		hostCounts := []uint32{}
83
		totalWeight := uint32(0)
84
		totalCount := uint32(0)
85
		for _, host := range cluster.Hosts {
86
			hostWeights = append(hostWeights, host.Weight)
87
			count := c.appServers[host.Addr].(*testutil.RPCServer).Count
88
			hostCounts = append(hostCounts, count)
89
			totalWeight += host.Weight
90
			totalCount += count
91
		}
92
		// compare count percent and weight percent, in range 'thres'
93
		for i := range cluster.Hosts {
94
			ratio := float64(hostCounts[i]) / float64(totalCount)
95
			expected := float64(hostWeights[i]) / float64(totalWeight)
96
			if math.Abs(ratio-expected) > thres {
97
				t.Errorf("cluster #%d weighted host selected error, host#%d expected: %f, got: %f", index, i, expected, ratio)
98
			}
99
			t.Logf("cluster #%d weighted host #%d expected: %f, got: %f", index, i, expected, ratio)
100
		}
101
		// cluster info
102
		clusterCounts = append(clusterCounts, totalCount)
103
		clusterTotalCount += totalCount
104
	}
105
	for i := range c.clusters {
106
		// cluster total weight should be 100
107
		expected := float64(c.clusters[i].Weight) / 100.0
108
		ratio := float64(clusterCounts[i]) / float64(clusterTotalCount)
109
		if math.Abs(ratio-expected) > thres {
110
			t.Errorf("cluster #%d weighted cluster selected error,  expected: %f, got: %f", i, expected, ratio)
111
		}
112
		t.Logf("cluster #%d weighted expected: %f, got: %f", i, expected, ratio)
113
	}
114
}
115

116
func TestWeightProxy(t *testing.T) {
117
	hostsAddress := []string{"127.0.0.1:8081", "127.0.0.1:8082", "127.0.0.1:8083", "127.0.0.1:8084"}
118
	cluster1 := &testutil.WeightCluster{
119
		Name:   "cluster1",
120
		Weight: 40,
121
		Hosts: []*testutil.WeightHost{
122
			{Addr: hostsAddress[0], Weight: 1},
123
			{Addr: hostsAddress[1], Weight: 2},
124
		},
125
	}
126
	cluster2 := &testutil.WeightCluster{
127
		Name:   "cluster2",
128
		Weight: 60,
129
		Hosts: []*testutil.WeightHost{
130
			{Addr: hostsAddress[2], Weight: 1},
131
			{Addr: hostsAddress[3], Weight: 2},
132
		},
133
	}
134
	tc := newWeightCase(t, []*testutil.WeightCluster{cluster1, cluster2})
135
	pass := false
136
	tc.Start()
137
	go tc.RunCase(100, 0)
138
	select {
139
	case err := <-tc.C:
140
		if err != nil {
141
			t.Errorf("[ERROR MESSAGE] weighted proxy test failed, error: %v\n", err)
142
		}
143
		pass = true
144
	case <-time.After(100 * time.Second):
145
		t.Error("[ERROR MESSAGE] weighted proxy hangn")
146
	}
147
	tc.FinishCase()
148
	// Verify Weight
149
	if pass {
150
		tc.Verify(t)
151
	}
152
}
153