grpc/test/channelz_test.go

49 func czCleanupWrapper(cleanup func() error, t *testing.T) {
51 		t.Error(err)
67 func (s) TestCZServerRegistrationAndDeletion(t *testing.T) {
85 		defer czCleanupWrapper(czCleanup, t)
87 		te := newTest(t, e)
88 		te.startServers(&testServer{security: e.security}, c.total)
92 …t.Fatalf("GetServers(%d) = %+v (len of which: %d), end: %+v, want len(GetServers(%d)) = %d, end: %…
94 		te.tearDown()
97 …t.Fatalf("GetServers(0) = %+v (len of which: %d), end: %+v, want len(GetServers(0)) = 0, end: true…
102 func (s) TestCZGetServer(t *testing.T) {
104 	defer czCleanupWrapper(czCleanup, t)
106 	te := newTest(t, e)
107 	te.startServer(&testServer{security: e.security})
108 	defer te.tearDown()
112 		t.Fatalf("there should only be one server, not %d", len(ss))
118 		t.Fatalf("server %d does not exist", serverID)
121 		t.Fatalf("server want id %d, but got %d", serverID, srv.ID)
124 	te.tearDown()
134 		t.Fatal(err)
138 func (s) TestCZTopChannelRegistrationAndDeletion(t *testing.T) {
156 		defer czCleanupWrapper(czCleanup, t)
158 		te := newTest(t, e)
161 			cc := te.clientConn()
162 			te.cc = nil
164 			te.srvAddr = ""
173 			t.Fatal(err)
186 			t.Fatal(err)
188 		te.tearDown()
192 func (s) TestCZTopChannelRegistrationAndDeletionWhenDialFail(t *testing.T) {
194 	defer czCleanupWrapper(czCleanup, t)
198 		t.Fatal("expecting dial to fail")
201 		t.Fatalf("GetTopChannels(0, 0) = %v, %v, want <nil>, true", tcs, end)
205 func (s) TestCZNestedChannelRegistrationAndDeletion(t *testing.T) {
207 	defer czCleanupWrapper(czCleanup, t)
211 	te := newTest(t, e)
216 	te.resolverScheme = r.Scheme()
217 	te.clientConn()
218 	defer te.tearDown()
230 		t.Fatal(err)
246 		t.Fatal(err)
250 func (s) TestCZClientSubChannelSocketRegistrationAndDeletion(t *testing.T) {
252 	defer czCleanupWrapper(czCleanup, t)
255 	te := newTest(t, e)
257 	te.startServers(&testServer{security: e.security}, num)
260 	for _, a := range te.srvAddrs {
264 	te.resolverScheme = r.Scheme()
265 	te.clientConn()
266 	defer te.tearDown()
291 		t.Fatal(err)
318 		t.Fatal(err)
322 func (s) TestCZServerSocketRegistrationAndDeletion(t *testing.T) {
341 		defer czCleanupWrapper(czCleanup, t)
343 		te := newTest(t, e)
344 		te.startServer(&testServer{security: e.security})
347 			cc := te.clientConn()
348 			te.cc = nil
379 			t.Fatal(err)
393 			t.Fatal(err)
395 		te.tearDown()
399 func (s) TestCZServerListenSocketDeletion(t *testing.T) {
401 	defer czCleanupWrapper(czCleanup, t)
405 		t.Fatalf("failed to listen: %v", err)
418 		t.Fatal(err)
432 		t.Fatal(err)
449 func (s) TestCZRecusivelyDeletionOfEntry(t *testing.T) {
458 	defer czCleanupWrapper(czCleanup, t)
467 		t.Fatalf("There should be one TopChannel entry")
470 		t.Fatalf("There should be two SubChannel entries")
474 		t.Fatalf("There should be two Socket entries")
480 		t.Fatalf("There should be one TopChannel entry")
487 		t.Fatalf("There should be one TopChannel entry")
490 		t.Fatalf("There should be one SubChannel entry")
497 		t.Fatalf("There should be no TopChannel entry")
501 func (s) TestCZChannelMetrics(t *testing.T) {
503 	defer czCleanupWrapper(czCleanup, t)
506 	te := newTest(t, e)
507 	te.maxClientSendMsgSize = newInt(8)
509 	te.startServers(&testServer{security: e.security}, num)
512 	for _, a := range te.srvAddrs {
516 	te.resolverScheme = r.Scheme()
517 	cc := te.clientConn()
518 	defer te.tearDown()
521 		t.Fatalf("TestService/EmptyCall(_, _) = _, %v, want _, <nil>", err)
529 		t.Fatal(err)
538 …t.Fatalf("TestService/UnaryCall(_, _) = _, %v, want _, error code: %s", err, codes.ResourceExhaust…
543 		t.Fatalf("%v.FullDuplexCall(_) = _, %v, want <nil>", tc, err)
586 		t.Fatal(err)
590 func (s) TestCZServerMetrics(t *testing.T) {
592 	defer czCleanupWrapper(czCleanup, t)
594 	te := newTest(t, e)
595 	te.maxServerReceiveMsgSize = newInt(8)
596 	te.startServer(&testServer{security: e.security})
597 	defer te.tearDown()
598 	cc := te.clientConn()
601 		t.Fatalf("TestService/EmptyCall(_, _) = _, %v, want _, <nil>", err)
609 		t.Fatal(err)
617 …t.Fatalf("TestService/UnaryCall(_, _) = _, %v, want _, error code: %s", err, codes.ResourceExhaust…
622 		t.Fatalf("%v.FullDuplexCall(_) = _, %v, want <nil>", tc, err)
642 		t.Fatal(err)
652 func (t *testServiceClientWrapper) getCurrentStreamID() uint32 {
653 	t.mu.RLock()
654 	defer t.mu.RUnlock()
655 	return uint32(2*t.streamsCreated - 1)
658 func (t *testServiceClientWrapper) EmptyCall(ctx context.Context, in *testpb.Empty, opts ...grpc.Ca…
659 	t.mu.Lock()
660 	defer t.mu.Unlock()
661 	t.streamsCreated++
662 	return t.TestServiceClient.EmptyCall(ctx, in, opts...)
665 func (t *testServiceClientWrapper) UnaryCall(ctx context.Context, in *testpb.SimpleRequest, opts ..…
666 	t.mu.Lock()
667 	defer t.mu.Unlock()
668 	t.streamsCreated++
669 	return t.TestServiceClient.UnaryCall(ctx, in, opts...)
672 func (t *testServiceClientWrapper) StreamingOutputCall(ctx context.Context, in *testpb.StreamingOut…
673 	t.mu.Lock()
674 	defer t.mu.Unlock()
675 	t.streamsCreated++
676 	return t.TestServiceClient.StreamingOutputCall(ctx, in, opts...)
679 func (t *testServiceClientWrapper) StreamingInputCall(ctx context.Context, opts ...grpc.CallOption)…
680 	t.mu.Lock()
681 	defer t.mu.Unlock()
682 	t.streamsCreated++
683 	return t.TestServiceClient.StreamingInputCall(ctx, opts...)
686 func (t *testServiceClientWrapper) FullDuplexCall(ctx context.Context, opts ...grpc.CallOption) (te…
687 	t.mu.Lock()
688 	defer t.mu.Unlock()
689 	t.streamsCreated++
690 	return t.TestServiceClient.FullDuplexCall(ctx, opts...)
693 func (t *testServiceClientWrapper) HalfDuplexCall(ctx context.Context, opts ...grpc.CallOption) (te…
694 	t.mu.Lock()
695 	defer t.mu.Unlock()
696 	t.streamsCreated++
697 	return t.TestServiceClient.HalfDuplexCall(ctx, opts...)
700 func doSuccessfulUnaryCall(tc testpb.TestServiceClient, t *testing.T) {
702 		t.Fatalf("TestService/EmptyCall(_, _) = _, %v, want _, <nil>", err)
706 func doStreamingInputCallWithLargePayload(tc testpb.TestServiceClient, t *testing.T) {
709 		t.Fatalf("TestService/StreamingInputCall(_) = _, %v, want <nil>", err)
713 		t.Fatal(err)
718 func doServerSideFailedUnaryCall(tc testpb.TestServiceClient, t *testing.T) {
724 		t.Fatal(err)
732 …t.Fatalf("TestService/UnaryCall(_, _) = _, %v, want _, error code: %s", err, codes.ResourceExhaust…
736 func doClientSideInitiatedFailedStream(tc testpb.TestServiceClient, t *testing.T) {
740 		t.Fatalf("TestService/FullDuplexCall(_) = _, %v, want <nil>", err)
746 		t.Fatal(err)
758 		t.Fatalf("%v.Send(%v) = %v, want <nil>", stream, sreq, err)
761 		t.Fatalf("%v.Recv() = %v, want <nil>", stream, err)
769 func doServerSideInitiatedFailedStreamWithRSTStream(tc testpb.TestServiceClient, t *testing.T, l *l…
772 		t.Fatalf("TestService/FullDuplexCall(_) = _, %v, want <nil>", err)
778 		t.Fatal(err)
790 		t.Fatalf("%v.Send(%v) = %v, want <nil>", stream, sreq, err)
793 		t.Fatalf("%v.Recv() = %v, want <nil>", stream, err)
802 		t.Fatalf("%v.Recv() = %v, want <non-nil>", stream, err)
807 func doServerSideInitiatedFailedStreamWithGoAway(tc testpb.TestServiceClient, t *testing.T, l *list…
812 		t.Fatalf("TestService/FullDuplexCall(_) = _, %v, want <nil>", err)
819 		t.Fatalf("s.Send() failed with error: %v", err)
822 		t.Fatalf("s.Recv() failed with error: %v", err)
827 		t.Fatalf("TestService/FullDuplexCall(_) = _, %v, want <nil>", err)
834 		t.Fatalf("s.Send() failed with error: %v", err)
837 		t.Fatalf("s.Recv() failed with error: %v", err)
845 		t.Fatalf("%v.Recv() = %v, want <non-nil>", s, err)
850 func doServerSideInitiatedFailedStreamWithClientBreakFlowControl(tc testpb.TestServiceClient, t *te…
853 		t.Fatalf("TestService/FullDuplexCall(_) = _, %v, want <nil>", err)
860 		t.Fatalf("%v.Recv() = %v, want error code: %v", stream, err, codes.ResourceExhausted)
864 func doIdleCallToInvokeKeepAlive(tc testpb.TestServiceClient, t *testing.T) {
868 		t.Fatalf("TestService/FullDuplexCall(_) = _, %v, want <nil>", err)
875 func (s) TestCZClientSocketMetricsStreamsAndMessagesCount(t *testing.T) {
877 	defer czCleanupWrapper(czCleanup, t)
879 	te := newTest(t, e)
880 	te.maxServerReceiveMsgSize = newInt(20)
881 	te.maxClientReceiveMsgSize = newInt(20)
882 	rcw := te.startServerWithConnControl(&testServer{security: e.security})
883 	defer te.tearDown()
884 	cc := te.clientConn()
887 	doSuccessfulUnaryCall(tc, t)
918 		t.Fatal(err)
921 	doServerSideFailedUnaryCall(tc, t)
930 		t.Fatal(err)
933 	doClientSideInitiatedFailedStream(tc, t)
942 		t.Fatal(err)
945 	doServerSideInitiatedFailedStreamWithRSTStream(tc, t, rcw)
954 		t.Fatal(err)
957 	doServerSideInitiatedFailedStreamWithGoAway(tc, t, rcw)
966 		t.Fatal(err)
975 func (s) TestCZClientAndServerSocketMetricsStreamsCountFlowControlRSTStream(t *testing.T) {
977 	defer czCleanupWrapper(czCleanup, t)
979 	te := newTest(t, e)
980 	te.serverInitialWindowSize = 65536
983 	te.serverInitialConnWindowSize = 65536 * 2
984 	te.startServer(&testServer{security: e.security})
985 	defer te.tearDown()
986 	cc, dw := te.clientConnWithConnControl()
989 	doServerSideInitiatedFailedStreamWithClientBreakFlowControl(tc, t, dw)
1032 		t.Fatal(err)
1036 func (s) TestCZClientAndServerSocketMetricsFlowControl(t *testing.T) {
1038 	defer czCleanupWrapper(czCleanup, t)
1040 	te := newTest(t, e)
1042 	te.serverInitialWindowSize = 65536
1043 	te.serverInitialConnWindowSize = 65536
1044 	te.clientInitialWindowSize = 65536
1045 	te.clientInitialConnWindowSize = 65536
1046 	te.startServer(&testServer{security: e.security})
1047 	defer te.tearDown()
1048 	cc := te.clientConn()
1052 		doSuccessfulUnaryCall(tc, t)
1096 		t.Fatal(err)
1099 	doStreamingInputCallWithLargePayload(tc, t)
1120 		t.Fatal(err)
1125 	doStreamingInputCallWithLargePayload(tc, t)
1145 		t.Fatal(err)
1149 func (s) TestCZClientSocketMetricsKeepAlive(t *testing.T) {
1151 	defer czCleanupWrapper(czCleanup, t)
1152 	defer func(t time.Duration) { internal.KeepaliveMinPingTime = t }(internal.KeepaliveMinPingTime)
1155 	te := newTest(t, e)
1156 	te.customDialOptions = append(te.customDialOptions, grpc.WithKeepaliveParams(
1162 	te.customServerOptions = append(te.customServerOptions, grpc.KeepaliveEnforcementPolicy(
1167 	te.startServer(&testServer{security: e.security})
1168 	te.clientConn() // Dial the server
1169 	defer te.tearDown()
1198 		t.Fatal(err)
1202 func (s) TestCZServerSocketMetricsStreamsAndMessagesCount(t *testing.T) {
1204 	defer czCleanupWrapper(czCleanup, t)
1206 	te := newTest(t, e)
1207 	te.maxServerReceiveMsgSize = newInt(20)
1208 	te.maxClientReceiveMsgSize = newInt(20)
1209 	te.startServer(&testServer{security: e.security})
1210 	defer te.tearDown()
1211 	cc, _ := te.clientConnWithConnControl()
1223 		t.Fatal(err)
1226 	doSuccessfulUnaryCall(tc, t)
1235 		t.Fatal(err)
1238 	doServerSideFailedUnaryCall(tc, t)
1247 		t.Fatal(err)
1250 	doClientSideInitiatedFailedStream(tc, t)
1259 		t.Fatal(err)
1263 func (s) TestCZServerSocketMetricsKeepAlive(t *testing.T) {
1265 	defer czCleanupWrapper(czCleanup, t)
1267 	te := newTest(t, e)
1268 …te.customServerOptions = append(te.customServerOptions, grpc.KeepaliveParams(keepalive.ServerParam…
1269 	te.startServer(&testServer{security: e.security})
1270 	defer te.tearDown()
1271 	cc := te.clientConn()
1273 	doIdleCallToInvokeKeepAlive(tc, t)
1289 		t.Fatal(err)
1322 func (s) TestCZSocketGetSecurityValueTLS(t *testing.T) {
1324 	defer czCleanupWrapper(czCleanup, t)
1326 	te := newTest(t, e)
1327 	te.startServer(&testServer{security: e.security})
1328 	defer te.tearDown()
1329 	te.clientConn()
1368 		t.Fatal(err)
1372 func (s) TestCZChannelTraceCreationDeletion(t *testing.T) {
1374 	defer czCleanupWrapper(czCleanup, t)
1378 	te := newTest(t, e)
1383 	te.resolverScheme = r.Scheme()
1384 	te.clientConn()
1385 	defer te.tearDown()
1415 		t.Fatal(err)
1444 		t.Fatal(err)
1448 func (s) TestCZSubChannelTraceCreationDeletion(t *testing.T) {
1450 	defer czCleanupWrapper(czCleanup, t)
1452 	te := newTest(t, e)
1453 	te.startServer(&testServer{security: e.security})
1456 	r.InitialState(resolver.State{Addresses: []resolver.Address{{Addr: te.srvAddr}}})
1457 	te.resolverScheme = r.Scheme()
1458 	te.clientConn()
1459 	defer te.tearDown()
1494 		t.Fatal(err)
1523 		t.Fatal(err)
1527 func (s) TestCZChannelAddressResolutionChange(t *testing.T) {
1529 	defer czCleanupWrapper(czCleanup, t)
1532 	te := newTest(t, e)
1533 	te.startServer(&testServer{security: e.security})
1536 	addrs := []resolver.Address{{Addr: te.srvAddr}}
1538 	te.resolverScheme = r.Scheme()
1539 	te.clientConn()
1540 	defer te.tearDown()
1560 		t.Fatal(err)
1576 		t.Fatal(err)
1610 		t.Fatal(err)
1627 		t.Fatal(err)
1631 func (s) TestCZSubChannelPickedNewAddress(t *testing.T) {
1633 	defer czCleanupWrapper(czCleanup, t)
1636 	te := newTest(t, e)
1637 	te.startServers(&testServer{security: e.security}, 3)
1641 	for _, a := range te.srvAddrs {
1645 	te.resolverScheme = r.Scheme()
1646 	cc := te.clientConn()
1647 	defer te.tearDown()
1653 		t.Fatalf("TestService/EmptyCall(_, _) = _, %v, want _, <nil>", err)
1655 	te.srvs[0].Stop()
1656 	te.srvs[1].Stop()
1679 …if scm.Trace.Events[i].Desc == fmt.Sprintf("Subchannel picks a new address %q to connect", te.srvA…
1688 		t.Fatal(err)
1692 func (s) TestCZSubChannelConnectivityState(t *testing.T) {
1694 	defer czCleanupWrapper(czCleanup, t)
1696 	te := newTest(t, e)
1697 	te.startServer(&testServer{security: e.security})
1700 	r.InitialState(resolver.State{Addresses: []resolver.Address{{Addr: te.srvAddr}}})
1701 	te.resolverScheme = r.Scheme()
1702 	cc := te.clientConn()
1703 	defer te.tearDown()
1709 		t.Fatalf("TestService/EmptyCall(_, _) = _, %v, want _, <nil>", err)
1712 	te.srv.Stop()
1783 		t.Fatal(err)
1787 func (s) TestCZChannelConnectivityState(t *testing.T) {
1789 	defer czCleanupWrapper(czCleanup, t)
1791 	te := newTest(t, e)
1792 	te.startServer(&testServer{security: e.security})
1795 	r.InitialState(resolver.State{Addresses: []resolver.Address{{Addr: te.srvAddr}}})
1796 	te.resolverScheme = r.Scheme()
1797 	cc := te.clientConn()
1798 	defer te.tearDown()
1804 		t.Fatalf("TestService/EmptyCall(_, _) = _, %v, want _, <nil>", err)
1806 	te.srv.Stop()
1840 		t.Fatal(err)
1844 func (s) TestCZTraceOverwriteChannelDeletion(t *testing.T) {
1846 	defer czCleanupWrapper(czCleanup, t)
1850 	te := newTest(t, e)
1857 	te.resolverScheme = r.Scheme()
1858 	te.clientConn()
1859 	defer te.tearDown()
1874 		t.Fatal(err)
1890 		t.Fatal(err)
1901 		t.Fatal(err)
1905 func (s) TestCZTraceOverwriteSubChannelDeletion(t *testing.T) {
1907 	defer czCleanupWrapper(czCleanup, t)
1909 	te := newTest(t, e)
1912 	te.startServer(&testServer{security: e.security})
1915 	r.InitialState(resolver.State{Addresses: []resolver.Address{{Addr: te.srvAddr}}})
1916 	te.resolverScheme = r.Scheme()
1917 	te.clientConn()
1918 	defer te.tearDown()
1935 		t.Fatal(err)
1950 		t.Fatal(err)
1961 		t.Fatal(err)
1965 func (s) TestCZTraceTopChannelDeletionTraceClear(t *testing.T) {
1967 	defer czCleanupWrapper(czCleanup, t)
1969 	te := newTest(t, e)
1970 	te.startServer(&testServer{security: e.security})
1973 	r.InitialState(resolver.State{Addresses: []resolver.Address{{Addr: te.srvAddr}}})
1974 	te.resolverScheme = r.Scheme()
1975 	te.clientConn()
1992 		t.Fatal(err)
1994 	te.tearDown()
2003 		t.Fatal(err)