vendor/github.com/coreos/etcd/clientv3/health_balancer_test.go - cloudstack-kubernetes-provider - Git at Google

 // Copyright 2017 The etcd Authors
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
 //     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 // See the License for the specific language governing permissions and
 // limitations under the License.

 package clientv3

 import (
 	"context"
 	"errors"
 	"net"
 	"sync"
 	"testing"
 	"time"

 	pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
 	"github.com/coreos/etcd/pkg/testutil"

 	"google.golang.org/grpc"
 )

 var endpoints = []string{"localhost:2379", "localhost:22379", "localhost:32379"}

 func TestBalancerGetUnblocking(t *testing.T) {
 	hb := newHealthBalancer(endpoints, minHealthRetryDuration, func(string) (bool, error) { return true, nil })
 	defer hb.Close()
 	if addrs := <-hb.Notify(); len(addrs) != len(endpoints) {
 		t.Errorf("Initialize newHealthBalancer should have triggered Notify() chan, but it didn't")
 	}
 	unblockingOpts := grpc.BalancerGetOptions{BlockingWait: false}

 	_, _, err := hb.Get(context.Background(), unblockingOpts)
 	if err != ErrNoAddrAvilable {
 		t.Errorf("Get() with no up endpoints should return ErrNoAddrAvailable, got: %v", err)
 	}

 	down1 := hb.Up(grpc.Address{Addr: endpoints[1]})
 	if addrs := <-hb.Notify(); len(addrs) != 1 {
 		t.Errorf("first Up() should have triggered balancer to send the first connected address via Notify chan so that other connections can be closed")
 	}
 	down2 := hb.Up(grpc.Address{Addr: endpoints[2]})
 	addrFirst, putFun, err := hb.Get(context.Background(), unblockingOpts)
 	if err != nil {
 		t.Errorf("Get() with up endpoints should success, got %v", err)
 	}
 	if addrFirst.Addr != endpoints[1] {
 		t.Errorf("Get() didn't return expected address, got %v", addrFirst)
 	}
 	if putFun == nil {
 		t.Errorf("Get() returned unexpected nil put function")
 	}
 	addrSecond, _, _ := hb.Get(context.Background(), unblockingOpts)
 	if addrFirst.Addr != addrSecond.Addr {
 		t.Errorf("Get() didn't return the same address as previous call, got %v and %v", addrFirst, addrSecond)
 	}

 	down1(errors.New("error"))
 	if addrs := <-hb.Notify(); len(addrs) != len(endpoints)-1 { // we call down on one endpoint
 		t.Errorf("closing the only connection should triggered balancer to send the %d endpoints via Notify chan so that we can establish a connection", len(endpoints)-1)
 	}
 	down2(errors.New("error"))
 	_, _, err = hb.Get(context.Background(), unblockingOpts)
 	if err != ErrNoAddrAvilable {
 		t.Errorf("Get() with no up endpoints should return ErrNoAddrAvailable, got: %v", err)
 	}
 }

 func TestBalancerGetBlocking(t *testing.T) {
 	hb := newHealthBalancer(endpoints, minHealthRetryDuration, func(string) (bool, error) { return true, nil })
 	defer hb.Close()
 	if addrs := <-hb.Notify(); len(addrs) != len(endpoints) {
 		t.Errorf("Initialize newHealthBalancer should have triggered Notify() chan, but it didn't")
 	}
 	blockingOpts := grpc.BalancerGetOptions{BlockingWait: true}

 	ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond*100)
 	_, _, err := hb.Get(ctx, blockingOpts)
 	cancel()
 	if err != context.DeadlineExceeded {
 		t.Errorf("Get() with no up endpoints should timeout, got %v", err)
 	}

 	downC := make(chan func(error), 1)

 	go func() {
 		// ensure hb.Up() will be called after hb.Get() to see if Up() releases blocking Get()
 		time.Sleep(time.Millisecond * 100)
 		f := hb.Up(grpc.Address{Addr: endpoints[1]})
 		if addrs := <-hb.Notify(); len(addrs) != 1 {
 			t.Errorf("first Up() should have triggered balancer to send the first connected address via Notify chan so that other connections can be closed")
 		}
 		downC <- f
 	}()
 	addrFirst, putFun, err := hb.Get(context.Background(), blockingOpts)
 	if err != nil {
 		t.Errorf("Get() with up endpoints should success, got %v", err)
 	}
 	if addrFirst.Addr != endpoints[1] {
 		t.Errorf("Get() didn't return expected address, got %v", addrFirst)
 	}
 	if putFun == nil {
 		t.Errorf("Get() returned unexpected nil put function")
 	}
 	down1 := <-downC

 	down2 := hb.Up(grpc.Address{Addr: endpoints[2]})
 	addrSecond, _, _ := hb.Get(context.Background(), blockingOpts)
 	if addrFirst.Addr != addrSecond.Addr {
 		t.Errorf("Get() didn't return the same address as previous call, got %v and %v", addrFirst, addrSecond)
 	}

 	down1(errors.New("error"))
 	if addrs := <-hb.Notify(); len(addrs) != len(endpoints)-1 { // we call down on one endpoint
 		t.Errorf("closing the only connection should triggered balancer to send the %d endpoints via Notify chan so that we can establish a connection", len(endpoints)-1)
 	}
 	down2(errors.New("error"))
 	ctx, cancel = context.WithTimeout(context.Background(), time.Millisecond*100)
 	_, _, err = hb.Get(ctx, blockingOpts)
 	cancel()
 	if err != context.DeadlineExceeded {
 		t.Errorf("Get() with no up endpoints should timeout, got %v", err)
 	}
 }

 // TestHealthBalancerGraylist checks one endpoint is tried after the other
 // due to gray listing.
 func TestHealthBalancerGraylist(t *testing.T) {
 	var wg sync.WaitGroup
 	// Use 3 endpoints so gray list doesn't fallback to all connections
 	// after failing on 2 endpoints.
 	lns, eps := make([]net.Listener, 3), make([]string, 3)
 	wg.Add(3)
 	connc := make(chan string, 2)
 	for i := range eps {
 		ln, err := net.Listen("tcp", ":0")
 		testutil.AssertNil(t, err)
 		lns[i], eps[i] = ln, ln.Addr().String()
 		go func() {
 			defer wg.Done()
 			for {
 				conn, err := ln.Accept()
 				if err != nil {
 					return
 				}
 				_, err = conn.Read(make([]byte, 512))
 				conn.Close()
 				if err == nil {
 					select {
 					case connc <- ln.Addr().String():
 						// sleep some so balancer catches up
 						// before attempted next reconnect.
 						time.Sleep(50 * time.Millisecond)
 					default:
 					}
 				}
 			}
 		}()
 	}

 	tf := func(s string) (bool, error) { return false, nil }
 	hb := newHealthBalancer(eps, 5*time.Second, tf)

 	conn, err := grpc.Dial("", grpc.WithInsecure(), grpc.WithBalancer(hb))
 	testutil.AssertNil(t, err)
 	defer conn.Close()

 	kvc := pb.NewKVClient(conn)
 	<-hb.ready()

 	kvc.Range(context.TODO(), &pb.RangeRequest{})
 	ep1 := <-connc
 	kvc.Range(context.TODO(), &pb.RangeRequest{})
 	ep2 := <-connc
 	for _, ln := range lns {
 		ln.Close()
 	}
 	wg.Wait()

 	if ep1 == ep2 {
 		t.Fatalf("expected %q != %q", ep1, ep2)
 	}
 }

 // TestBalancerDoNotBlockOnClose ensures that balancer and grpc don't deadlock each other
 // due to rapid open/close conn. The deadlock causes balancer.Close() to block forever.
 // See issue: https://github.com/coreos/etcd/issues/7283 for more detail.
 func TestBalancerDoNotBlockOnClose(t *testing.T) {
 	defer testutil.AfterTest(t)

 	kcl := newKillConnListener(t, 3)
 	defer kcl.close()

 	for i := 0; i < 5; i++ {
 		hb := newHealthBalancer(kcl.endpoints(), minHealthRetryDuration, func(string) (bool, error) { return true, nil })
 		conn, err := grpc.Dial("", grpc.WithInsecure(), grpc.WithBalancer(hb))
 		if err != nil {
 			t.Fatal(err)
 		}
 		kvc := pb.NewKVClient(conn)
 		<-hb.readyc

 		var wg sync.WaitGroup
 		wg.Add(100)
 		cctx, cancel := context.WithCancel(context.TODO())
 		for j := 0; j < 100; j++ {
 			go func() {
 				defer wg.Done()
 				kvc.Range(cctx, &pb.RangeRequest{}, grpc.FailFast(false))
 			}()
 		}
 		// balancer.Close() might block
 		// if balancer and grpc deadlock each other.
 		bclosec, cclosec := make(chan struct{}), make(chan struct{})
 		go func() {
 			defer close(bclosec)
 			hb.Close()
 		}()
 		go func() {
 			defer close(cclosec)
 			conn.Close()
 		}()
 		select {
 		case <-bclosec:
 		case <-time.After(3 * time.Second):
 			testutil.FatalStack(t, "balancer close timeout")
 		}
 		select {
 		case <-cclosec:
 		case <-time.After(3 * time.Second):
 			t.Fatal("grpc conn close timeout")
 		}

 		cancel()
 		wg.Wait()
 	}
 }

 // killConnListener listens incoming conn and kills it immediately.
 type killConnListener struct {
 	wg    sync.WaitGroup
 	eps   []string
 	stopc chan struct{}
 	t     *testing.T
 }

 func newKillConnListener(t *testing.T, size int) *killConnListener {
 	kcl := &killConnListener{stopc: make(chan struct{}), t: t}

 	for i := 0; i < size; i++ {
 		ln, err := net.Listen("tcp", ":0")
 		if err != nil {
 			t.Fatal(err)
 		}
 		kcl.eps = append(kcl.eps, ln.Addr().String())
 		kcl.wg.Add(1)
 		go kcl.listen(ln)
 	}
 	return kcl
 }

 func (kcl *killConnListener) endpoints() []string {
 	return kcl.eps
 }

 func (kcl *killConnListener) listen(l net.Listener) {
 	go func() {
 		defer kcl.wg.Done()
 		for {
 			conn, err := l.Accept()
 			select {
 			case <-kcl.stopc:
 				return
 			default:
 			}
 			if err != nil {
 				kcl.t.Fatal(err)
 			}
 			time.Sleep(1 * time.Millisecond)
 			conn.Close()
 		}
 	}()
 	<-kcl.stopc
 	l.Close()
 }

 func (kcl *killConnListener) close() {
 	close(kcl.stopc)
 	kcl.wg.Wait()
 }
	// Copyright 2017 The etcd Authors
	//
	// Licensed under the Apache License, Version 2.0 (the "License");
	// you may not use this file except in compliance with the License.
	// You may obtain a copy of the License at
	//
	// http://www.apache.org/licenses/LICENSE-2.0
	//
	// Unless required by applicable law or agreed to in writing, software
	// distributed under the License is distributed on an "AS IS" BASIS,
	// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	// See the License for the specific language governing permissions and
	// limitations under the License.

	package clientv3

	import (
	"context"
	"errors"
	"net"
	"sync"
	"testing"
	"time"

	pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
	"github.com/coreos/etcd/pkg/testutil"

	"google.golang.org/grpc"
	)

	var endpoints = []string{"localhost:2379", "localhost:22379", "localhost:32379"}

	func TestBalancerGetUnblocking(t *testing.T) {
	hb := newHealthBalancer(endpoints, minHealthRetryDuration, func(string) (bool, error) { return true, nil })
	defer hb.Close()
	if addrs := <-hb.Notify(); len(addrs) != len(endpoints) {
	t.Errorf("Initialize newHealthBalancer should have triggered Notify() chan, but it didn't")
	}
	unblockingOpts := grpc.BalancerGetOptions{BlockingWait: false}

	_, _, err := hb.Get(context.Background(), unblockingOpts)
	if err != ErrNoAddrAvilable {
	t.Errorf("Get() with no up endpoints should return ErrNoAddrAvailable, got: %v", err)
	}

	down1 := hb.Up(grpc.Address{Addr: endpoints[1]})
	if addrs := <-hb.Notify(); len(addrs) != 1 {
	t.Errorf("first Up() should have triggered balancer to send the first connected address via Notify chan so that other connections can be closed")
	}
	down2 := hb.Up(grpc.Address{Addr: endpoints[2]})
	addrFirst, putFun, err := hb.Get(context.Background(), unblockingOpts)
	if err != nil {
	t.Errorf("Get() with up endpoints should success, got %v", err)
	}
	if addrFirst.Addr != endpoints[1] {
	t.Errorf("Get() didn't return expected address, got %v", addrFirst)
	}
	if putFun == nil {
	t.Errorf("Get() returned unexpected nil put function")
	}
	addrSecond, _, _ := hb.Get(context.Background(), unblockingOpts)
	if addrFirst.Addr != addrSecond.Addr {
	t.Errorf("Get() didn't return the same address as previous call, got %v and %v", addrFirst, addrSecond)
	}

	down1(errors.New("error"))
	if addrs := <-hb.Notify(); len(addrs) != len(endpoints)-1 { // we call down on one endpoint
	t.Errorf("closing the only connection should triggered balancer to send the %d endpoints via Notify chan so that we can establish a connection", len(endpoints)-1)
	}
	down2(errors.New("error"))
	_, _, err = hb.Get(context.Background(), unblockingOpts)
	if err != ErrNoAddrAvilable {
	t.Errorf("Get() with no up endpoints should return ErrNoAddrAvailable, got: %v", err)
	}
	}

	func TestBalancerGetBlocking(t *testing.T) {
	hb := newHealthBalancer(endpoints, minHealthRetryDuration, func(string) (bool, error) { return true, nil })
	defer hb.Close()
	if addrs := <-hb.Notify(); len(addrs) != len(endpoints) {
	t.Errorf("Initialize newHealthBalancer should have triggered Notify() chan, but it didn't")
	}
	blockingOpts := grpc.BalancerGetOptions{BlockingWait: true}

	ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond*100)
	_, _, err := hb.Get(ctx, blockingOpts)
	cancel()
	if err != context.DeadlineExceeded {
	t.Errorf("Get() with no up endpoints should timeout, got %v", err)
	}

	downC := make(chan func(error), 1)

	go func() {
	// ensure hb.Up() will be called after hb.Get() to see if Up() releases blocking Get()
	time.Sleep(time.Millisecond * 100)
	f := hb.Up(grpc.Address{Addr: endpoints[1]})
	if addrs := <-hb.Notify(); len(addrs) != 1 {
	t.Errorf("first Up() should have triggered balancer to send the first connected address via Notify chan so that other connections can be closed")
	}
	downC <- f
	}()
	addrFirst, putFun, err := hb.Get(context.Background(), blockingOpts)
	if err != nil {
	t.Errorf("Get() with up endpoints should success, got %v", err)
	}
	if addrFirst.Addr != endpoints[1] {
	t.Errorf("Get() didn't return expected address, got %v", addrFirst)
	}
	if putFun == nil {
	t.Errorf("Get() returned unexpected nil put function")
	}
	down1 := <-downC

	down2 := hb.Up(grpc.Address{Addr: endpoints[2]})
	addrSecond, _, _ := hb.Get(context.Background(), blockingOpts)
	if addrFirst.Addr != addrSecond.Addr {
	t.Errorf("Get() didn't return the same address as previous call, got %v and %v", addrFirst, addrSecond)
	}

	down1(errors.New("error"))
	if addrs := <-hb.Notify(); len(addrs) != len(endpoints)-1 { // we call down on one endpoint
	t.Errorf("closing the only connection should triggered balancer to send the %d endpoints via Notify chan so that we can establish a connection", len(endpoints)-1)
	}
	down2(errors.New("error"))
	ctx, cancel = context.WithTimeout(context.Background(), time.Millisecond*100)
	_, _, err = hb.Get(ctx, blockingOpts)
	cancel()
	if err != context.DeadlineExceeded {
	t.Errorf("Get() with no up endpoints should timeout, got %v", err)
	}
	}

	// TestHealthBalancerGraylist checks one endpoint is tried after the other
	// due to gray listing.
	func TestHealthBalancerGraylist(t *testing.T) {
	var wg sync.WaitGroup
	// Use 3 endpoints so gray list doesn't fallback to all connections
	// after failing on 2 endpoints.
	lns, eps := make([]net.Listener, 3), make([]string, 3)
	wg.Add(3)
	connc := make(chan string, 2)
	for i := range eps {
	ln, err := net.Listen("tcp", ":0")
	testutil.AssertNil(t, err)
	lns[i], eps[i] = ln, ln.Addr().String()
	go func() {
	defer wg.Done()
	for {
	conn, err := ln.Accept()
	if err != nil {
	return
	}
	_, err = conn.Read(make([]byte, 512))
	conn.Close()
	if err == nil {
	select {
	case connc <- ln.Addr().String():
	// sleep some so balancer catches up
	// before attempted next reconnect.
	time.Sleep(50 * time.Millisecond)
	default:
	}
	}
	}
	}()
	}

	tf := func(s string) (bool, error) { return false, nil }
	hb := newHealthBalancer(eps, 5*time.Second, tf)

	conn, err := grpc.Dial("", grpc.WithInsecure(), grpc.WithBalancer(hb))
	testutil.AssertNil(t, err)
	defer conn.Close()

	kvc := pb.NewKVClient(conn)
	<-hb.ready()

	kvc.Range(context.TODO(), &pb.RangeRequest{})
	ep1 := <-connc
	kvc.Range(context.TODO(), &pb.RangeRequest{})
	ep2 := <-connc
	for _, ln := range lns {
	ln.Close()
	}
	wg.Wait()

	if ep1 == ep2 {
	t.Fatalf("expected %q != %q", ep1, ep2)
	}
	}

	// TestBalancerDoNotBlockOnClose ensures that balancer and grpc don't deadlock each other
	// due to rapid open/close conn. The deadlock causes balancer.Close() to block forever.
	// See issue: https://github.com/coreos/etcd/issues/7283 for more detail.
	func TestBalancerDoNotBlockOnClose(t *testing.T) {
	defer testutil.AfterTest(t)

	kcl := newKillConnListener(t, 3)
	defer kcl.close()

	for i := 0; i < 5; i++ {
	hb := newHealthBalancer(kcl.endpoints(), minHealthRetryDuration, func(string) (bool, error) { return true, nil })
	conn, err := grpc.Dial("", grpc.WithInsecure(), grpc.WithBalancer(hb))
	if err != nil {
	t.Fatal(err)
	}
	kvc := pb.NewKVClient(conn)
	<-hb.readyc

	var wg sync.WaitGroup
	wg.Add(100)
	cctx, cancel := context.WithCancel(context.TODO())
	for j := 0; j < 100; j++ {
	go func() {
	defer wg.Done()
	kvc.Range(cctx, &pb.RangeRequest{}, grpc.FailFast(false))
	}()
	}
	// balancer.Close() might block
	// if balancer and grpc deadlock each other.
	bclosec, cclosec := make(chan struct{}), make(chan struct{})
	go func() {
	defer close(bclosec)
	hb.Close()
	}()
	go func() {
	defer close(cclosec)
	conn.Close()
	}()
	select {
	case <-bclosec:
	case <-time.After(3 * time.Second):
	testutil.FatalStack(t, "balancer close timeout")
	}
	select {
	case <-cclosec:
	case <-time.After(3 * time.Second):
	t.Fatal("grpc conn close timeout")
	}

	cancel()
	wg.Wait()
	}
	}

	// killConnListener listens incoming conn and kills it immediately.
	type killConnListener struct {
	wg sync.WaitGroup
	eps []string
	stopc chan struct{}
	t *testing.T
	}

	func newKillConnListener(t testing.T, size int) killConnListener {
	kcl := &killConnListener{stopc: make(chan struct{}), t: t}

	for i := 0; i < size; i++ {
	ln, err := net.Listen("tcp", ":0")
	if err != nil {
	t.Fatal(err)
	}
	kcl.eps = append(kcl.eps, ln.Addr().String())
	kcl.wg.Add(1)
	go kcl.listen(ln)
	}
	return kcl
	}

	func (kcl *killConnListener) endpoints() []string {
	return kcl.eps
	}

	func (kcl *killConnListener) listen(l net.Listener) {
	go func() {
	defer kcl.wg.Done()
	for {
	conn, err := l.Accept()
	select {
	case <-kcl.stopc:
	return
	default:
	}
	if err != nil {
	kcl.t.Fatal(err)
	}
	time.Sleep(1 * time.Millisecond)
	conn.Close()
	}
	}()
	<-kcl.stopc
	l.Close()
	}

	func (kcl *killConnListener) close() {
	close(kcl.stopc)
	kcl.wg.Wait()
	}