Merge pull request #7735 from gyuho/grpc-shutdown

pkg/transport: add 'IsClosedConnError'

clientv3/integration/metrics_test.go

@@ -58,7 +58,7 @@ func TestV3ClientMetrics(t *testing.T) {
 		}
 
 		err = srv.Serve(ln)
-		if err != nil && !strings.Contains(err.Error(), "use of closed network connection") {
+		if err != nil && !transport.IsClosedConnError(err) {
 			t.Fatalf("Err serving http requests: %v", err)
 		}
 	}()

pkg/transport/listener.go

@@ -27,6 +27,7 @@ import (
 	"net"
 	"os"
 	"path/filepath"
+	"strings"
 	"time"
 
 	"github.com/coreos/etcd/pkg/tlsutil"
@@ -269,3 +270,12 @@ func ShallowCopyTLSConfig(cfg *tls.Config) *tls.Config {
 	}
 	return &ncfg
 }
+
+// IsClosedConnError returns true if the error is from closing listener, cmux.
+// copied from golang.org/x/net/http2/http2.go
+func IsClosedConnError(err error) bool {
+	// 'use of closed network connection' (Go <=1.8)
+	// 'use of closed file or network connection' (Go >1.8, internal/poll.ErrClosing)
+	// 'mux: listener closed' (cmux.ErrListenerClosed)
+	return err != nil && strings.Contains(err.Error(), "closed")
+}

pkg/transport/listener_test.go

@@ -274,3 +274,15 @@ func TestNewListenerTLSInfoSelfCert(t *testing.T) {
 	}
 	testNewListenerTLSInfoAccept(t, tlsinfo)
 }
+
+func TestIsClosedConnError(t *testing.T) {
+	l, err := NewListener("testsocket", "unix", nil)
+	if err != nil {
+		t.Errorf("error listening on unix socket (%v)", err)
+	}
+	l.Close()
+	_, err = l.Accept()
+	if !IsClosedConnError(err) {
+		t.Fatalf("expect true, got false (%v)", err)
+	}
+}

rafthttp/stream.go

@@ -19,7 +19,6 @@ import (
 	"fmt"
 	"io"
 	"io/ioutil"
-	"net"
 	"net/http"
 	"path"
 	"strings"
@@ -28,6 +27,7 @@ import (
 
 	"github.com/coreos/etcd/etcdserver/stats"
 	"github.com/coreos/etcd/pkg/httputil"
+	"github.com/coreos/etcd/pkg/transport"
 	"github.com/coreos/etcd/pkg/types"
 	"github.com/coreos/etcd/raft/raftpb"
 	"github.com/coreos/etcd/version"
@@ -317,7 +317,7 @@ func (cr *streamReader) run() {
 			// all data is read out
 			case err == io.EOF:
 			// connection is closed by the remote
-			case isClosedConnectionError(err):
+			case transport.IsClosedConnError(err):
 			default:
 				cr.status.deactivate(failureType{source: t.String(), action: "read"}, err.Error())
 			}
@@ -514,11 +514,6 @@ func (cr *streamReader) resume() {
 	cr.paused = false
 }
 
-func isClosedConnectionError(err error) bool {
-	operr, ok := err.(*net.OpError)
-	return ok && operr.Err.Error() == "use of closed network connection"
-}
-
 // checkStreamSupport checks whether the stream type is supported in the
 // given version.
 func checkStreamSupport(v *semver.Version, t streamType) bool {