integration: add Blackhole to bridgeConn

Signed-off-by: Gyu-Ho Lee <[email protected]>

integration/bridge.go

@@ -17,6 +17,7 @@ package integration
 import (
 	"fmt"
 	"io"
+	"io/ioutil"
 	"net"
 	"sync"
 
@@ -31,21 +32,23 @@ type bridge struct {
 	l       net.Listener
 	conns   map[*bridgeConn]struct{}
 
-	stopc  chan struct{}
-	pausec chan struct{}
-	wg     sync.WaitGroup
+	stopc      chan struct{}
+	pausec     chan struct{}
+	blackholec chan struct{}
+	wg         sync.WaitGroup
 
 	mu sync.Mutex
 }
 
 func newBridge(addr string) (*bridge, error) {
 	b := &bridge{
 		// bridge "port" is ("%05d%05d0", port, pid) since go1.8 expects the port to be a number
-		inaddr:  addr + "0",
-		outaddr: addr,
-		conns:   make(map[*bridgeConn]struct{}),
-		stopc:   make(chan struct{}),
-		pausec:  make(chan struct{}),
+		inaddr:     addr + "0",
+		outaddr:    addr,
+		conns:      make(map[*bridgeConn]struct{}),
+		stopc:      make(chan struct{}),
+		pausec:     make(chan struct{}),
+		blackholec: make(chan struct{}),
 	}
 	close(b.pausec)
 
@@ -152,12 +155,12 @@ func (b *bridge) serveConn(bc *bridgeConn) {
 	var wg sync.WaitGroup
 	wg.Add(2)
 	go func() {
-		io.Copy(bc.out, bc.in)
+		b.ioCopy(bc, bc.out, bc.in)
 		bc.close()
 		wg.Done()
 	}()
 	go func() {
-		io.Copy(bc.in, bc.out)
+		b.ioCopy(bc, bc.in, bc.out)
 		bc.close()
 		wg.Done()
 	}()
@@ -179,3 +182,47 @@ func (bc *bridgeConn) close() {
 	bc.in.Close()
 	bc.out.Close()
 }
+
+func (b *bridge) Blackhole() {
+	b.mu.Lock()
+	close(b.blackholec)
+	b.mu.Unlock()
+}
+
+func (b *bridge) Unblackhole() {
+	b.mu.Lock()
+	for bc := range b.conns {
+		bc.Close()
+	}
+	b.conns = make(map[*bridgeConn]struct{})
+	b.blackholec = make(chan struct{})
+	b.mu.Unlock()
+}
+
+// ref. https://github.com/golang/go/blob/master/src/io/io.go copyBuffer
+func (b *bridge) ioCopy(bc *bridgeConn, dst io.Writer, src io.Reader) (err error) {
+	buf := make([]byte, 32*1024)
+	for {
+		select {
+		case <-b.blackholec:
+			io.Copy(ioutil.Discard, src)
+			return nil
+		default:
+		}
+		nr, er := src.Read(buf)
+		if nr > 0 {
+			nw, ew := dst.Write(buf[0:nr])
+			if ew != nil {
+				return ew
+			}
+			if nr != nw {
+				return io.ErrShortWrite
+			}
+		}
+		if er != nil {
+			err = er
+			break
+		}
+	}
+	return
+}

integration/bridge_test.go

@@ -0,0 +1,41 @@
+// 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 integration
+
+import (
+	"testing"
+	"time"
+
+	"github.com/coreos/etcd/pkg/testutil"
+
+	"golang.org/x/net/context"
+)
+
+func TestBlackhole(t *testing.T) {
+	defer testutil.AfterTest(t)
+
+	clus := NewClusterV3(t, &ClusterConfig{Size: 1})
+	defer clus.Terminate(t)
+
+	clus.Members[0].Blackhole()
+	time.Sleep(time.Second)
+
+	clus.Members[0].Unblackhole()
+	time.Sleep(time.Second)
+
+	if _, err := clus.Client(0).Put(context.Background(), "foo", "bar"); err != nil {
+		t.Fatal(err)
+	}
+}

integration/cluster.go

@@ -588,6 +588,8 @@ func (m *member) ID() types.ID { return m.s.ID() }
 func (m *member) DropConnections()    { m.grpcBridge.Reset() }
 func (m *member) PauseConnections()   { m.grpcBridge.Pause() }
 func (m *member) UnpauseConnections() { m.grpcBridge.Unpause() }
+func (m *member) Blackhole()          { m.grpcBridge.Blackhole() }
+func (m *member) Unblackhole()        { m.grpcBridge.Unblackhole() }
 
 // NewClientV3 creates a new grpc client connection to the member
 func NewClientV3(m *member) (*clientv3.Client, error) {