@@ -38,8 +38,10 @@ import (
38
38
"flag"
39
39
"fmt"
40
40
"io"
41
+ "log"
41
42
"math"
42
43
"net"
44
+ "os"
43
45
"reflect"
44
46
"runtime"
45
47
"sort"
@@ -54,6 +56,7 @@ import (
54
56
"google.golang.org/grpc"
55
57
"google.golang.org/grpc/codes"
56
58
"google.golang.org/grpc/credentials"
59
+ "google.golang.org/grpc/grpclog"
57
60
"google.golang.org/grpc/health"
58
61
healthpb "google.golang.org/grpc/health/grpc_health_v1alpha"
59
62
"google.golang.org/grpc/metadata"
@@ -280,6 +283,13 @@ const tlsDir = "testdata/"
280
283
281
284
func TestReconnectTimeout (t * testing.T ) {
282
285
defer leakCheck (t )()
286
+ restore := declareLogNoise (t ,
287
+ "transport: http2Client.notifyError got notified that the client transport was broken" ,
288
+ "grpc: Conn.resetTransport failed to create client transport: connection error: desc = \" transport" ,
289
+ "grpc: Conn.transportMonitor exits due to: grpc: timed out trying to connect" ,
290
+ )
291
+ defer restore ()
292
+
283
293
lis , err := net .Listen ("tcp" , ":0" )
284
294
if err != nil {
285
295
t .Fatalf ("Failed to listen: %v" , err )
@@ -520,6 +530,14 @@ func TestTimeoutOnDeadServer(t *testing.T) {
520
530
}
521
531
522
532
func testTimeoutOnDeadServer (t * testing.T , e env ) {
533
+ restore := declareLogNoise (t ,
534
+ "transport: http2Client.notifyError got notified that the client transport was broken EOF" ,
535
+ "grpc: Conn.transportMonitor exits due to: grpc: the client connection is closing" ,
536
+ "grpc: Conn.resetTransport failed to create client transport: connection error" ,
537
+ "grpc: Conn.resetTransport failed to create client transport: connection error: desc = \" transport: dial unix" ,
538
+ )
539
+ defer restore ()
540
+
523
541
s , addr := serverSetUp (t , true , nil , math .MaxUint32 , nil , nil , e )
524
542
cc := clientSetUp (t , addr , nil , nil , "" , e )
525
543
tc := testpb .NewTestServiceClient (cc )
@@ -554,6 +572,7 @@ func testTimeoutOnDeadServer(t *testing.T, e env) {
554
572
t .Fatalf ("cc.State() = %s, %v, want %s or %s, <nil>" , state , err , grpc .Connecting , grpc .TransientFailure )
555
573
}
556
574
cc .Close ()
575
+ awaitNewConnLogOutput ()
557
576
}
558
577
559
578
func healthCheck (d time.Duration , cc * grpc.ClientConn , serviceName string ) (* healthpb.HealthCheckResponse , error ) {
@@ -591,6 +610,12 @@ func TestHealthCheckOnFailure(t *testing.T) {
591
610
}
592
611
593
612
func testHealthCheckOnFailure (t * testing.T , e env ) {
613
+ defer leakCheck (t )()
614
+ restore := declareLogNoise (t ,
615
+ "Failed to dial " ,
616
+ "grpc: the client connection is closing; please retry" ,
617
+ )
618
+ defer restore ()
594
619
hs := health .NewHealthServer ()
595
620
hs .SetServingStatus ("grpc.health.v1alpha.HealthCheck" , 1 )
596
621
s , addr := serverSetUp (t , true , hs , math .MaxUint32 , nil , nil , e )
@@ -599,6 +624,7 @@ func testHealthCheckOnFailure(t *testing.T, e env) {
599
624
if _ , err := healthCheck (0 * time .Second , cc , "grpc.health.v1alpha.Health" ); err != grpc .Errorf (codes .DeadlineExceeded , "context deadline exceeded" ) {
600
625
t .Fatalf ("Health/Check(_, _) = _, %v, want _, error code %d" , err , codes .DeadlineExceeded )
601
626
}
627
+ awaitNewConnLogOutput ()
602
628
}
603
629
604
630
func TestHealthCheckOff (t * testing.T ) {
@@ -841,6 +867,10 @@ func TestRetry(t *testing.T) {
841
867
// TODO(zhaoq): Refactor to make this clearer and add more cases to test racy
842
868
// and error-prone paths.
843
869
func testRetry (t * testing.T , e env ) {
870
+ restore := declareLogNoise (t ,
871
+ "transport: http2Client.notifyError got notified that the client transport was broken" ,
872
+ )
873
+ defer restore ()
844
874
s , addr := serverSetUp (t , true , nil , math .MaxUint32 , nil , nil , e )
845
875
cc := clientSetUp (t , addr , nil , nil , "" , e )
846
876
tc := testpb .NewTestServiceClient (cc )
@@ -922,6 +952,10 @@ func TestCancel(t *testing.T) {
922
952
}
923
953
924
954
func testCancel (t * testing.T , e env ) {
955
+ restore := declareLogNoise (t ,
956
+ "grpc: the client connection is closing; please retry" ,
957
+ )
958
+ defer restore ()
925
959
s , addr := serverSetUp (t , true , nil , math .MaxUint32 , nil , nil , e )
926
960
cc := clientSetUp (t , addr , nil , nil , "" , e )
927
961
tc := testpb .NewTestServiceClient (cc )
@@ -945,6 +979,9 @@ func testCancel(t *testing.T, e env) {
945
979
if grpc .Code (err ) != codes .Canceled {
946
980
t .Fatalf (`TestService/UnaryCall(_, _) = %v, %v; want <nil>, error code: %d` , reply , err , codes .Canceled )
947
981
}
982
+ cc .Close ()
983
+
984
+ awaitNewConnLogOutput ()
948
985
}
949
986
950
987
func TestCancelNoIO (t * testing.T ) {
@@ -1585,3 +1622,126 @@ func leakCheck(t testing.TB) func() {
1585
1622
}
1586
1623
}
1587
1624
}
1625
+
1626
+ type lockingWriter struct {
1627
+ mu sync.Mutex
1628
+ w io.Writer
1629
+ }
1630
+
1631
+ func (lw * lockingWriter ) Write (p []byte ) (n int , err error ) {
1632
+ lw .mu .Lock ()
1633
+ defer lw .mu .Unlock ()
1634
+ return lw .w .Write (p )
1635
+ }
1636
+
1637
+ func (lw * lockingWriter ) setWriter (w io.Writer ) {
1638
+ lw .mu .Lock ()
1639
+ defer lw .mu .Unlock ()
1640
+ lw .w = w
1641
+ }
1642
+
1643
+ var testLogOutput = & lockingWriter {w : os .Stderr }
1644
+
1645
+ // awaitNewConnLogOutput waits for any of grpc.NewConn's goroutines to
1646
+ // terminate, if they're still running. It spams logs with this
1647
+ // message. We wait for it so our log filter is still
1648
+ // active. Otherwise the "defer restore()" at the top of various test
1649
+ // functions restores our log filter and then the goroutine spams.
1650
+ func awaitNewConnLogOutput () {
1651
+ awaitLogOutput (50 * time .Millisecond , "grpc: the client connection is closing; please retry" )
1652
+ }
1653
+
1654
+ func awaitLogOutput (maxWait time.Duration , phrase string ) {
1655
+ pb := []byte (phrase )
1656
+
1657
+ timer := time .NewTimer (maxWait )
1658
+ defer timer .Stop ()
1659
+ wakeup := make (chan bool , 1 )
1660
+ for {
1661
+ if logOutputHasContents (pb , wakeup ) {
1662
+ return
1663
+ }
1664
+ select {
1665
+ case <- timer .C :
1666
+ // Too slow. Oh well.
1667
+ return
1668
+ case <- wakeup :
1669
+ }
1670
+ }
1671
+ }
1672
+
1673
+ func logOutputHasContents (v []byte , wakeup chan <- bool ) bool {
1674
+ testLogOutput .mu .Lock ()
1675
+ defer testLogOutput .mu .Unlock ()
1676
+ fw , ok := testLogOutput .w .(* filterWriter )
1677
+ if ! ok {
1678
+ return false
1679
+ }
1680
+ fw .mu .Lock ()
1681
+ defer fw .mu .Unlock ()
1682
+ if bytes .Contains (fw .buf .Bytes (), v ) {
1683
+ return true
1684
+ }
1685
+ fw .wakeup = wakeup
1686
+ return false
1687
+ }
1688
+
1689
+ func init () {
1690
+ grpclog .SetLogger (log .New (testLogOutput , "" , log .LstdFlags ))
1691
+ }
1692
+
1693
+ var verboseLogs = flag .Bool ("verbose_logs" , false , "show all grpclog output, without filtering" )
1694
+
1695
+ func noop () {}
1696
+
1697
+ // declareLogNoise declares that t is expected to emit the following noisy phrases,
1698
+ // even on success. Those phrases will be filtered from grpclog output
1699
+ // and only be shown if *verbose_logs or t ends up failing.
1700
+ // The returned restore function should be called with defer to be run
1701
+ // before the test ends.
1702
+ func declareLogNoise (t * testing.T , phrases ... string ) (restore func ()) {
1703
+ if * verboseLogs {
1704
+ return noop
1705
+ }
1706
+ fw := & filterWriter {dst : os .Stderr , filter : phrases }
1707
+ testLogOutput .setWriter (fw )
1708
+ return func () {
1709
+ if t .Failed () {
1710
+ fw .mu .Lock ()
1711
+ defer fw .mu .Unlock ()
1712
+ if fw .buf .Len () > 0 {
1713
+ t .Logf ("Complete log output:\n %s" , fw .buf .Bytes ())
1714
+ }
1715
+ }
1716
+ testLogOutput .setWriter (os .Stderr )
1717
+ }
1718
+ }
1719
+
1720
+ type filterWriter struct {
1721
+ dst io.Writer
1722
+ filter []string
1723
+
1724
+ mu sync.Mutex
1725
+ buf bytes.Buffer
1726
+ wakeup chan <- bool // if non-nil, gets true on write
1727
+ }
1728
+
1729
+ func (fw * filterWriter ) Write (p []byte ) (n int , err error ) {
1730
+ fw .mu .Lock ()
1731
+ fw .buf .Write (p )
1732
+ if fw .wakeup != nil {
1733
+ select {
1734
+ case fw .wakeup <- true :
1735
+ default :
1736
+ }
1737
+ }
1738
+ fw .mu .Unlock ()
1739
+
1740
+ ps := string (p )
1741
+ for _ , f := range fw .filter {
1742
+ if strings .Contains (ps , f ) {
1743
+ return len (p ), nil
1744
+ }
1745
+ }
1746
+ return fw .dst .Write (p )
1747
+ }
0 commit comments