inotnako/grpc_lb_exp.go

## grpc_lb_exp.go
package grpc_lb_exp

import (
	"context"
	"net"
	"testing"

	"github.com/lygo/runner"

	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/grpc/peer"
	"google.golang.org/grpc/balancer/roundrobin"
	"google.golang.org/grpc/resolver"
	"google.golang.org/grpc/resolver/manual"

	pb "google.golang.org/grpc/examples/helloworld/helloworld"
)

// server is used to implement helloworld.GreeterServer.
type server struct{
	name string
	err bool
}

// SayHello implements helloworld.GreeterServer
func (s *server) SayHello(ctx context.Context, in *pb.HelloRequest) (*pb.HelloReply, error) {
	if s.err {
		return nil, status.Error(codes.Internal, `BOOM`)
	}

	return &pb.HelloReply{Message: "Hello " + in.Name +" from "+s.name}, nil
}

func startTestServers(withErr bool,addrs ...resolver.Address ) (stop func () , err error) {
	app := runner.New()

	defer func  () {
		if err != nil && app != nil {
			app.Shutdown()
		}
	}()

	for _, addr:= range addrs {
		var lis net.Listener

		lis , err = net.Listen(`tcp`,addr.Addr)
		if err != nil {
			return
		}

		srv := grpc.NewServer()
		pb.RegisterGreeterServer(srv,&server{
			name:addr.Addr,
			err:withErr,
		})
		app.Runners = append(app.Runners, func () error {
			return srv.Serve(lis)
		})
		app.Slams = append(app.Slams,func () error {
			srv.Stop()
			return nil
		})
	}

	app.Run()
	<- app.Started
	stop = app.Shutdown
	return
}

func TestGrpcLbManual(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	var (
		org1MspID  = `org1msp`
		org2MspID  = `org2msp`
	)

	resolverForOrg1 := manual.NewBuilderWithScheme(org1MspID)
	resolver.Register(resolverForOrg1)
	defer resolverForOrg1.Close()
	org1Addrs := []resolver.Address{
		{
			Addr:`localhost:3456`,
			Type:resolver.Backend,
			Metadata:&map[string]string{
				"mspid":org1MspID,
			},
		},
		{
			Addr:`localhost:3457`,
			Type:resolver.Backend,
			Metadata:&map[string]string{
				"mspid":org1MspID,
			},
		},
	}
	org1SrvsStop, err := startTestServers(false,org1Addrs...)
	if err != nil {
		t.Fatal(err)
	}
	defer org1SrvsStop()
	resolverForOrg1.InitialAddrs(org1Addrs)


	resolverForOrg2 := manual.NewBuilderWithScheme(org2MspID)
	resolver.Register(resolverForOrg2)

	defer resolverForOrg2.Close()

	org2Addrs := []resolver.Address{
		{
			Addr:`localhost:3458`,
			Type:resolver.Backend,
			Metadata:&map[string]string{
				"mspid":org2MspID,
			},
		},
		{
			Addr:`localhost:3459`,
			Type:resolver.Backend,
			Metadata:&map[string]string{
				"mspid":org2MspID,
			},
		},
	}

	org2SrvsStop, err := startTestServers(false,org2Addrs...)
	if err != nil {
		t.Fatal(err)
	}
	defer org2SrvsStop()

	resolverForOrg2.InitialAddrs(org2Addrs)

	defer func () {
		resolver.UnregisterForTesting(org1MspID)
		resolver.UnregisterForTesting(org2MspID)
	}()

	org1Conn,err := grpc.DialContext(
		ctx,
		resolverForOrg1.Scheme()+":///peer.service",
		grpc.WithInsecure(),
		grpc.WithBalancerName(roundrobin.Name),
	)
	if err != nil {
		t.Fatal(err)
	}
	defer org1Conn.Close()

	org2Conn,err := grpc.DialContext(
		ctx,
		resolverForOrg2.Scheme()+":///peer.service",
		grpc.WithInsecure(),
		grpc.WithBalancerName(roundrobin.Name),
	)
	if err != nil {
		t.Fatal(err)
	}
	defer org2Conn.Close()

	org1Cli := pb.NewGreeterClient(org1Conn)
	org2Cli := pb.NewGreeterClient(org2Conn)

	var p peer.Peer

	for i := 0; i < 10; i++ {
		reply, err := org1Cli.SayHello(ctx,&pb.HelloRequest{Name:`Aloxa`},grpc.Peer(&p))
		if err != nil {
			t.Error(err)
		} else {
			t.Log(p.Addr.Network(),p.Addr.String(),reply.Message)
		}
	}

	for i := 0; i < 10; i++ {
		reply, err := org2Cli.SayHello(ctx,&pb.HelloRequest{Name:`Bloxa`},grpc.Peer(&p))
		if err != nil {
			t.Error(err)
		} else {
			t.Log(p.Addr.Network(),p.Addr.String(),reply.Message)
		}
	}

	resolverForOrg2.NewAddress(append(org2Addrs,resolver.Address{
		Addr:`localhost:3457`,
		Type:resolver.Backend,
		Metadata:&map[string]string{
			"mspid":org2MspID,
		},
	}))

	t.Log("expected only for localhost:3457,3458,3459")

	for i := 0; i < 10; i++ {
		reply, err := org2Cli.SayHello(ctx,&pb.HelloRequest{Name:`Cloxa`},grpc.Peer(&p))
		if err != nil {
			t.Error(err, p.Addr.Network(),p.Addr.String())
		} else {
			t.Log(p.Addr.Network(),p.Addr.String(),reply.Message)
		}
	}

	resolverForOrg2.NewAddress([]resolver.Address{
		{
			Addr:`localhost:3458`,
			Type:resolver.Backend,
			Metadata:&map[string]string{
				"mspid":org2MspID,
			},
		},
	})


	t.Log("expected only for localhost:3458")
	for i := 0; i < 10; i++ {
		reply, err := org2Cli.SayHello(ctx,&pb.HelloRequest{Name:`Dloxa`},grpc.Peer(&p))
		if err != nil {
			t.Error(err, p.Addr.Network(),p.Addr.String())
		} else {
			t.Log(p.Addr.Network(),p.Addr.String(),reply.Message)
		}
	}

	addrWithError := resolver.Address{
		Addr:`localhost:3460`,
		Type:resolver.Backend,
		Metadata:&map[string]string{
			"mspid":org2MspID,
		},
	}

	resolverForOrg2.NewAddress([]resolver.Address{
		{
			Addr:`localhost:3458`,
			Type:resolver.Backend,
			Metadata:&map[string]string{
				"mspid":org2MspID,
			},
		},
		addrWithError,
	})

	org2SrvsErrStop, err := startTestServers(true,addrWithError)
	if err != nil {
		t.Fatal(err)
	}
	defer org2SrvsErrStop()


	for i := 0; i < 100; i++ {
		reply, err := org2Cli.SayHello(ctx,&pb.HelloRequest{Name:`Eloxa`},grpc.Peer(&p), grpc.FailFast(true))
		if err != nil {
			t.Error(err, p.Addr.Network(),p.Addr.String())
		} else {
			t.Log(p.Addr.Network(),p.Addr.String(),reply.Message)
		}
	}

}
	package grpc_lb_exp

	import (
	"context"
	"net"
	"testing"

	"github.com/lygo/runner"

	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/grpc/peer"
	"google.golang.org/grpc/balancer/roundrobin"
	"google.golang.org/grpc/resolver"
	"google.golang.org/grpc/resolver/manual"

	pb "google.golang.org/grpc/examples/helloworld/helloworld"
	)

	// server is used to implement helloworld.GreeterServer.
	type server struct{
	name string
	err bool
	}

	// SayHello implements helloworld.GreeterServer
	func (s server) SayHello(ctx context.Context, in pb.HelloRequest) (*pb.HelloReply, error) {
	if s.err {
	return nil, status.Error(codes.Internal, `BOOM`)
	}

	return &pb.HelloReply{Message: "Hello " + in.Name +" from "+s.name}, nil
	}

	func startTestServers(withErr bool,addrs ...resolver.Address ) (stop func () , err error) {
	app := runner.New()

	defer func () {
	if err != nil && app != nil {
	app.Shutdown()
	}
	}()

	for _, addr:= range addrs {
	var lis net.Listener

	lis , err = net.Listen(`tcp`,addr.Addr)
	if err != nil {
	return
	}

	srv := grpc.NewServer()
	pb.RegisterGreeterServer(srv,&server{
	name:addr.Addr,
	err:withErr,
	})
	app.Runners = append(app.Runners, func () error {
	return srv.Serve(lis)
	})
	app.Slams = append(app.Slams,func () error {
	srv.Stop()
	return nil
	})
	}

	app.Run()
	<- app.Started
	stop = app.Shutdown
	return
	}

	func TestGrpcLbManual(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	var (
	org1MspID = `org1msp`
	org2MspID = `org2msp`
	)

	resolverForOrg1 := manual.NewBuilderWithScheme(org1MspID)
	resolver.Register(resolverForOrg1)
	defer resolverForOrg1.Close()
	org1Addrs := []resolver.Address{
	{
	Addr:`localhost:3456`,
	Type:resolver.Backend,
	Metadata:&map[string]string{
	"mspid":org1MspID,
	},
	},
	{
	Addr:`localhost:3457`,
	Type:resolver.Backend,
	Metadata:&map[string]string{
	"mspid":org1MspID,
	},
	},
	}
	org1SrvsStop, err := startTestServers(false,org1Addrs...)
	if err != nil {
	t.Fatal(err)
	}
	defer org1SrvsStop()
	resolverForOrg1.InitialAddrs(org1Addrs)


	resolverForOrg2 := manual.NewBuilderWithScheme(org2MspID)
	resolver.Register(resolverForOrg2)

	defer resolverForOrg2.Close()

	org2Addrs := []resolver.Address{
	{
	Addr:`localhost:3458`,
	Type:resolver.Backend,
	Metadata:&map[string]string{
	"mspid":org2MspID,
	},
	},
	{
	Addr:`localhost:3459`,
	Type:resolver.Backend,
	Metadata:&map[string]string{
	"mspid":org2MspID,
	},
	},
	}

	org2SrvsStop, err := startTestServers(false,org2Addrs...)
	if err != nil {
	t.Fatal(err)
	}
	defer org2SrvsStop()

	resolverForOrg2.InitialAddrs(org2Addrs)

	defer func () {
	resolver.UnregisterForTesting(org1MspID)
	resolver.UnregisterForTesting(org2MspID)
	}()

	org1Conn,err := grpc.DialContext(
	ctx,
	resolverForOrg1.Scheme()+":///peer.service",
	grpc.WithInsecure(),
	grpc.WithBalancerName(roundrobin.Name),
	)
	if err != nil {
	t.Fatal(err)
	}
	defer org1Conn.Close()

	org2Conn,err := grpc.DialContext(
	ctx,
	resolverForOrg2.Scheme()+":///peer.service",
	grpc.WithInsecure(),
	grpc.WithBalancerName(roundrobin.Name),
	)
	if err != nil {
	t.Fatal(err)
	}
	defer org2Conn.Close()

	org1Cli := pb.NewGreeterClient(org1Conn)
	org2Cli := pb.NewGreeterClient(org2Conn)

	var p peer.Peer

	for i := 0; i < 10; i++ {
	reply, err := org1Cli.SayHello(ctx,&pb.HelloRequest{Name:`Aloxa`},grpc.Peer(&p))
	if err != nil {
	t.Error(err)
	} else {
	t.Log(p.Addr.Network(),p.Addr.String(),reply.Message)
	}
	}

	for i := 0; i < 10; i++ {
	reply, err := org2Cli.SayHello(ctx,&pb.HelloRequest{Name:`Bloxa`},grpc.Peer(&p))
	if err != nil {
	t.Error(err)
	} else {
	t.Log(p.Addr.Network(),p.Addr.String(),reply.Message)
	}
	}

	resolverForOrg2.NewAddress(append(org2Addrs,resolver.Address{
	Addr:`localhost:3457`,
	Type:resolver.Backend,
	Metadata:&map[string]string{
	"mspid":org2MspID,
	},
	}))

	t.Log("expected only for localhost:3457,3458,3459")

	for i := 0; i < 10; i++ {
	reply, err := org2Cli.SayHello(ctx,&pb.HelloRequest{Name:`Cloxa`},grpc.Peer(&p))
	if err != nil {
	t.Error(err, p.Addr.Network(),p.Addr.String())
	} else {
	t.Log(p.Addr.Network(),p.Addr.String(),reply.Message)
	}
	}

	resolverForOrg2.NewAddress([]resolver.Address{
	{
	Addr:`localhost:3458`,
	Type:resolver.Backend,
	Metadata:&map[string]string{
	"mspid":org2MspID,
	},
	},
	})


	t.Log("expected only for localhost:3458")
	for i := 0; i < 10; i++ {
	reply, err := org2Cli.SayHello(ctx,&pb.HelloRequest{Name:`Dloxa`},grpc.Peer(&p))
	if err != nil {
	t.Error(err, p.Addr.Network(),p.Addr.String())
	} else {
	t.Log(p.Addr.Network(),p.Addr.String(),reply.Message)
	}
	}

	addrWithError := resolver.Address{
	Addr:`localhost:3460`,
	Type:resolver.Backend,
	Metadata:&map[string]string{
	"mspid":org2MspID,
	},
	}

	resolverForOrg2.NewAddress([]resolver.Address{
	{
	Addr:`localhost:3458`,
	Type:resolver.Backend,
	Metadata:&map[string]string{
	"mspid":org2MspID,
	},
	},
	addrWithError,
	})

	org2SrvsErrStop, err := startTestServers(true,addrWithError)
	if err != nil {
	t.Fatal(err)
	}
	defer org2SrvsErrStop()



	for i := 0; i < 100; i++ {
	reply, err := org2Cli.SayHello(ctx,&pb.HelloRequest{Name:`Eloxa`},grpc.Peer(&p), grpc.FailFast(true))
	if err != nil {
	t.Error(err, p.Addr.Network(),p.Addr.String())
	} else {
	t.Log(p.Addr.Network(),p.Addr.String(),reply.Message)
	}
	}

	}