Skip to content
Snippets Groups Projects
Commit bb4c0d15 authored by scott.wang's avatar scott.wang
Browse files

Finish etcd registry

parent 0e485401
No related branches found
No related tags found
No related merge requests found
Hide whitespace changes
Inline Side-by-side
Showing
with 64 additions and 335 deletions
registry/etcdv3/listener_test.go
+ 34
− 30
View file @ bb4c0d15
......@@ -2,51 +2,51 @@ package etcdv3
import (
"context"
"os"
"os/exec"
"testing"
)
import (
"time"
"github.com/apache/dubbo-go/common"
"github.com/apache/dubbo-go/remoting"
"github.com/stretchr/testify/suite"
"go.etcd.io/etcd/embed"
)
func startETCDServer() {
cmd := exec.Command("./load.sh", "start")
//cmd := exec.Command("pwd")
cmd.Stdout = os.Stdout
cmd.Stderr = os.Stdout
cmd.Dir = "../../remoting/etcdv3/single"
if err := cmd.Run(); err != nil {
panic(err)
}
type RegistryTestSuite struct {
suite.Suite
etcd *embed.Etcd
}
func stopETCDServer() {
// start etcd server
func (suite *RegistryTestSuite) SetupSuite() {
cmd := exec.Command("./load.sh", "stop")
//cmd := exec.Command("pwd")
cmd.Stdout = os.Stdout
cmd.Stderr = os.Stdout
cmd.Dir = "../../remoting/etcdv3/single"
t := suite.T()
if err := cmd.Run(); err != nil {
panic(err)
cfg := embed.NewConfig()
cfg.Dir = "/tmp/default.etcd"
e, err := embed.StartEtcd(cfg)
if err != nil {
t.Fatal(err)
}
select {
case <-e.Server.ReadyNotify():
t.Log("Server is ready!")
case <-time.After(60 * time.Second):
e.Server.Stop() // trigger a shutdown
t.Logf("Server took too long to start!")
}
suite.etcd = e
return
}
func TestMain(m *testing.M) {
startETCDServer()
m.Run()
stopETCDServer()
// stop etcd server
func (suite *RegistryTestSuite) TearDownSuite() {
suite.etcd.Close()
}
func Test_DataChange(t *testing.T) {
func (suite *RegistryTestSuite) TestDataChange() {
t := suite.T()
listener := NewRegistryDataListener(&MockDataListener{})
url, _ := common.NewURL(context.TODO(), "jsonrpc%3A%2F%2F127.0.0.1%3A20001%2Fcom.ikurento.user.UserProvider%3Fanyhost%3Dtrue%26app.version%3D0.0.1%26application%3DBDTService%26category%3Dproviders%26cluster%3Dfailover%26dubbo%3Ddubbo-provider-golang-2.6.0%26environment%3Ddev%26group%3D%26interface%3Dcom.ikurento.user.UserProvider%26ip%3D10.32.20.124%26loadbalance%3Drandom%26methods.GetUser.loadbalance%3Drandom%26methods.GetUser.retries%3D1%26methods.GetUser.weight%3D0%26module%3Ddubbogo%2Buser-info%2Bserver%26name%3DBDTService%26organization%3Dikurento.com%26owner%3DZX%26pid%3D74500%26retries%3D0%26service.filter%3Decho%26side%3Dprovider%26timestamp%3D1560155407%26version%3D%26warmup%3D100")
......@@ -56,6 +56,10 @@ func Test_DataChange(t *testing.T) {
}
}
func TestRegistrySuite(t *testing.T) {
suite.Run(t, &RegistryTestSuite{})
}
type MockDataListener struct {
}
......
registry/etcdv3/registry_test.go
+ 9
− 4
View file @ bb4c0d15
......@@ -33,7 +33,9 @@ func initRegistry(t *testing.T) *etcdV3Registry {
return out
}
func Test_Register(t *testing.T) {
func (suite *RegistryTestSuite) TestRegister() {
t := suite.T()
url, _ := common.NewURL(context.TODO(), "dubbo://127.0.0.1:20000/com.ikurento.user.UserProvider", common.WithParamsValue(constant.CLUSTER_KEY, "mock"), common.WithMethods([]string{"GetUser", "AddUser"}))
......@@ -47,8 +49,9 @@ func Test_Register(t *testing.T) {
assert.NoError(t, err)
}
func Test_Subscribe(t *testing.T) {
func (suite *RegistryTestSuite) TestSubscribe() {
t := suite.T()
regurl, _ := common.NewURL(context.TODO(), "registry://127.0.0.1:1111", common.WithParamsValue(constant.ROLE_KEY, strconv.Itoa(common.PROVIDER)))
url, _ := common.NewURL(context.TODO(), "dubbo://127.0.0.1:20000/com.ikurento.user.UserProvider", common.WithParamsValue(constant.CLUSTER_KEY, "mock"), common.WithMethods([]string{"GetUser", "AddUser"}))
......@@ -76,8 +79,9 @@ func Test_Subscribe(t *testing.T) {
assert.Regexp(t, ".*ServiceEvent{Action{add}.*", serviceEvent.String())
}
func Test_ConsumerDestory(t *testing.T) {
func (suite *RegistryTestSuite) TestConsumerDestory() {
t := suite.T()
url, _ := common.NewURL(context.TODO(), "dubbo://127.0.0.1:20000/com.ikurento.user.UserProvider", common.WithParamsValue(constant.CLUSTER_KEY, "mock"), common.WithMethods([]string{"GetUser", "AddUser"}))
reg := initRegistry(t)
......@@ -94,8 +98,9 @@ func Test_ConsumerDestory(t *testing.T) {
}
func Test_ProviderDestory(t *testing.T) {
func (suite *RegistryTestSuite) TestProviderDestory() {
t := suite.T()
reg := initRegistry(t)
url, _ := common.NewURL(context.TODO(), "dubbo://127.0.0.1:20000/com.ikurento.user.UserProvider", common.WithParamsValue(constant.CLUSTER_KEY, "mock"), common.WithMethods([]string{"GetUser", "AddUser"}))
reg.Register(url)
......
remoting/etcdv3/client_test.go
+ 21
− 16
View file @ bb4c0d15
......@@ -2,8 +2,7 @@ package etcdv3
import (
"fmt"
"os"
"os/exec"
"go.etcd.io/etcd/embed"
"path"
"reflect"
"strings"
......@@ -56,31 +55,37 @@ type ClientTestSuite struct {
heartbeat int
}
etcd *embed.Etcd
client *Client
}
// start etcd server
func (suite *ClientTestSuite) SetupSuite() {
cmd := exec.Command("./load.sh", "start")
cmd.Stdout = os.Stdout
cmd.Stderr = os.Stdout
cmd.Dir = "./single"
if err := cmd.Run(); err != nil {
suite.T().Fatal(err)
t := suite.T()
cfg := embed.NewConfig()
cfg.Dir = "/tmp/default.etcd"
e, err := embed.StartEtcd(cfg)
if err != nil {
t.Fatal(err)
}
select {
case <-e.Server.ReadyNotify():
t.Log("Server is ready!")
case <-time.After(60 * time.Second):
e.Server.Stop() // trigger a shutdown
t.Logf("Server took too long to start!")
}
suite.etcd = e
return
}
// stop etcd server
func (suite *ClientTestSuite) TearDownSuite() {
cmd := exec.Command("./load.sh", "stop")
cmd.Stdout = os.Stdout
cmd.Stderr = os.Stdout
cmd.Dir = "./single"
if err := cmd.Run(); err != nil {
suite.T().Fatal(err)
}
suite.etcd.Close()
}
func (suite *ClientTestSuite) setUpClient() *Client {
......
remoting/etcdv3/single/config.sh deleted 100755 → 0
+ 0
− 140
View file @ 0e485401
#!/usr/bin/env bash
# ******************************************************
# DESC : etcd devops script
# AUTHOR : Alex Stocks
# VERSION : 1.0
# LICENCE : LGPL V3
# EMAIL : alexstocks@foxmail.com
# MOD : 2018-01-05 11:37
# FILE : load.sh
# ******************************************************
cluster_name="etcd-test-cluster"
name_prefix="etcd_node"
data_dir=.
wal_dir=.
log_dir=.
peer0_ip="127.0.0.1"
peer0_client_port=2379
peer0_peer_port=2380
export ETCDCTL_API=3
etcd_endpoints="http://${peer0_ip}:${peer0_client_port}"
ctl="etcdctl --endpoints=$etcd_endpoints"
usage() {
echo "Usage: $0 start"
echo " $0 stop"
echo " $0 restart"
echo " $0 list"
echo " $0 clear"
exit
}
start() {
etcd --name=${name} \
--initial-advertise-peer-urls http://${!ip}:${!peer_port} \
--listen-peer-urls http://${!ip}:${!peer_port} \
--listen-client-urls http://${!ip}:${!client_port},http://127.0.0.1:${!client_port} \
--advertise-client-urls http://${!ip}:${!client_port} \
--initial-cluster-token ${cluster_name} \
--initial-cluster etcd_node0=http://${peer0_ip}:${peer0_peer_port} \
--initial-cluster-state new >> ${log_dir}/${name}.log 2>&1 &
sleep 5
PID=`ps aux | grep -w "name=${name}" | grep ${!client_port} | grep -v grep | awk '{print $2}'`
if [ "$PID" != "" ];
then
for p in $PID
do
echo "start $name ( pid =" $p ")"
done
fi
}
term() {
PID=`ps aux | grep -w "name=${name}" | grep ${!client_port} | grep -v grep | awk '{print $2}'`
if [ "$PID" != "" ];
then
for ps in $PID
do
echo "kill -9 $name ( pid =" $ps ")"
done
fi
kill -9 $ps
}
stop() {
id=`${ctl} member list | grep ${name} | grep -v grep | awk -F ',' '{print $1;}'`
echo "stop member name:${name} id:${id}"
# echo "etcdctl member remove ${id}"
# $ctl member remove ${id}
term
}
list() {
PID=`ps aux | grep -w "name=${name}" | grep ${!client_port} | grep -v grep | awk '{printf("%s,%s,%s,%s\n", $1, $2, $9, $10)}'`
if [ "$PID" != "" ];
then
echo "list ${name} $role"
echo "index: user, pid, start, duration"
idx=0
for ps in $PID
do
echo "$idx: $ps"
((idx ++))
done
fi
}
status() {
echo "-----------------member list----------------"
$ctl member list
export ETCDCTL_API=2
echo "-----------------cluster health-------------"
$ctl cluster-health
export ETCDCTL_API=3
}
client() {
local host=$1
local port=$2
sh bin/zkCli.sh -server $host:$port
}
clear() {
cd ${data_dir} && rm -rf ./* && cd ..
cd ${wal_dir} && rm -rf ./* && cd ..
cd ${log_dir} && rm -rf ./* && cd ..
}
switch() {
opt=$1
case C"$opt" in
Cstart)
start
;;
Cstop)
stop
;;
Crestart)
stop
start
;;
Clist)
list
;;
Cstatus)
status
;;
Cclear)
clear
;;
C*)
usage
;;
esac
}
remoting/etcdv3/single/ctl.sh deleted 100644 → 0
+ 0
− 124
View file @ 0e485401
#!/usr/bin/env bash
# ******************************************************
# DESC : etcd ctrl kv devops script
# AUTHOR : Alex Stocks
# VERSION : 1.0
# LICENCE : LGPL V3
# EMAIL : alexstocks@foxmail.com
# MOD : 2018-01-08 23:32
# FILE : ctl.sh
# ******************************************************
export ETCDCTL_API=3
peer0_ip="127.0.0.1"
peer0_client_port=2379
etcd_endpoints="http://${peer0_ip}:${peer0_client_port}"
ctl="etcdctl --endpoints=$etcd_endpoints"
usage() {
echo "Usage: $0 set key value"
echo " $0 get key"
echo " $0 get2 key # output value with properties"
echo " $0 getj key # output value in json"
echo " $0 getr [from to)"
echo " $0 ls dir" # get all info about key"
echo " $0 lr dir"
echo " $0 rm key"
echo " $0 rmr dir"
echo " $0 watch key"
echo " $0 watcha dir"
echo " $0 lease ID" # ID should be hex
echo " $0 status # cluster status"
exit
}
opt=$1
case C"$opt" in
Cset)
if [ $# != 3 ]; then
usage
fi
$ctl put $2 $3 --prev-kv=true
;;
Cget)
if [ $# != 2 ]; then
usage
fi
$ctl --write-out="json" get $2 --order=ASCEND
;;
Cget2)
if [ $# != 2 ]; then
usage
fi
$ctl --write-out="fields" get $2 --order=ASCEND
;;
Cgetj)
if [ $# != 2 ]; then
usage
fi
$ctl --write-out="json" get $2 --order=ASCEND
;;
Cgetr)
if [ $# != 3 ]; then
usage
fi
$ctl get "$2" "$3" --order=ASCEND
;;
Clr)
if [ $# != 2 ]; then
usage
fi
$ctl get $2 --order=DESCEND --prefix=true
;;
Cls)
dir=""
if [ $# == 2 ]; then
dir=$2
fi
$ctl get "$dir" --prefix=true
;;
Crm)
if [ $# != 2 ]; then
usage
fi
$ctl del $2 --prev-kv=true
;;
Crmr)
if [ $# != 2 ]; then
usage
fi
$ctl del $2 --prefix=true --prev-kv=true
;;
Cwatch)
if [ $# != 2 ]; then
usage
fi
$ctl watch $2 --prev-kv=true
;;
Cwatcha)
if [ $# != 2 ]; then
usage
fi
$ctl watch $2 --prev-kv=true --prefix=true
;;
Clease)
if [ $# != 2 ]; then
usage
fi
$ctl lease timetolive $2
;;
Cstatus)
export ETCDCTL_API=2
echo "-----------------member list----------------"
$ctl member list
echo "-----------------cluster health-------------"
$ctl cluster-health
export ETCDCTL_API=3
;;
C*)
usage
;;
esac
remoting/etcdv3/single/etcd deleted 100755 → 0
+ 0
− 0
View file @ 0e485401
File deleted
remoting/etcdv3/single/etcdctl deleted 100755 → 0
+ 0
− 0
View file @ 0e485401
File deleted
remoting/etcdv3/single/load.sh deleted 100755 → 0
+ 0
− 21
View file @ 0e485401
#!/usr/bin/env bash
# ******************************************************
# DESC : etcd devops script
# AUTHOR : Alex Stocks
# VERSION : 1.0
# LICENCE : LGPL V3
# EMAIL : alexstocks@foxmail.com
# MOD : 2018-01-05 11:37
# FILE : load.sh
# ******************************************************
source ./config.sh
idx=0
name="${name_prefix}${idx}"
ip="peer${idx}_ip"
client_port="peer${idx}_client_port"
peer_port="peer${idx}_peer_port"
opt=$1
switch $opt
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment