1/* 2 * 3 * Copyright 2018 gRPC authors. 4 * 5 * Licensed under the Apache License, Version 2.0 (the "License"); 6 * you may not use this file except in compliance with the License. 7 * You may obtain a copy of the License at 8 * 9 * http://www.apache.org/licenses/LICENSE-2.0 10 * 11 * Unless required by applicable law or agreed to in writing, software 12 * distributed under the License is distributed on an "AS IS" BASIS, 13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 14 * See the License for the specific language governing permissions and 15 * limitations under the License. 16 * 17 */ 18 19package service 20 21import ( 22 "context" 23 "fmt" 24 "net" 25 "reflect" 26 "strconv" 27 "testing" 28 "time" 29 30 "github.com/golang/protobuf/proto" 31 "github.com/golang/protobuf/ptypes" 32 channelzpb "google.golang.org/grpc/channelz/grpc_channelz_v1" 33 "google.golang.org/grpc/connectivity" 34 "google.golang.org/grpc/credentials" 35 "google.golang.org/grpc/internal/channelz" 36 "google.golang.org/grpc/internal/grpctest" 37) 38 39func init() { 40 channelz.TurnOn() 41} 42 43type s struct { 44 grpctest.Tester 45} 46 47func Test(t *testing.T) { 48 grpctest.RunSubTests(t, s{}) 49} 50 51func cleanupWrapper(cleanup func() error, t *testing.T) { 52 if err := cleanup(); err != nil { 53 t.Error(err) 54 } 55} 56 57type protoToSocketOptFunc func([]*channelzpb.SocketOption) *channelz.SocketOptionData 58 59// protoToSocketOpt is used in function socketProtoToStruct to extract socket option 60// data from unmarshaled proto message. 61// It is only defined under linux environment on x86 architecture. 62var protoToSocketOpt protoToSocketOptFunc 63 64// emptyTime is used for detecting unset value of time.Time type. 65// For go1.7 and earlier, ptypes.Timestamp will fill in the loc field of time.Time 66// with &utcLoc. However zero value of a time.Time type value loc field is nil. 67// This behavior will make reflect.DeepEqual fail upon unset time.Time field, 68// and cause false positive fatal error. 69// TODO: Go1.7 is no longer supported - does this need a change? 70var emptyTime time.Time 71 72const defaultTestTimeout = 10 * time.Second 73 74type dummyChannel struct { 75 state connectivity.State 76 target string 77 callsStarted int64 78 callsSucceeded int64 79 callsFailed int64 80 lastCallStartedTimestamp time.Time 81} 82 83func (d *dummyChannel) ChannelzMetric() *channelz.ChannelInternalMetric { 84 return &channelz.ChannelInternalMetric{ 85 State: d.state, 86 Target: d.target, 87 CallsStarted: d.callsStarted, 88 CallsSucceeded: d.callsSucceeded, 89 CallsFailed: d.callsFailed, 90 LastCallStartedTimestamp: d.lastCallStartedTimestamp, 91 } 92} 93 94type dummyServer struct { 95 callsStarted int64 96 callsSucceeded int64 97 callsFailed int64 98 lastCallStartedTimestamp time.Time 99} 100 101func (d *dummyServer) ChannelzMetric() *channelz.ServerInternalMetric { 102 return &channelz.ServerInternalMetric{ 103 CallsStarted: d.callsStarted, 104 CallsSucceeded: d.callsSucceeded, 105 CallsFailed: d.callsFailed, 106 LastCallStartedTimestamp: d.lastCallStartedTimestamp, 107 } 108} 109 110type dummySocket struct { 111 streamsStarted int64 112 streamsSucceeded int64 113 streamsFailed int64 114 messagesSent int64 115 messagesReceived int64 116 keepAlivesSent int64 117 lastLocalStreamCreatedTimestamp time.Time 118 lastRemoteStreamCreatedTimestamp time.Time 119 lastMessageSentTimestamp time.Time 120 lastMessageReceivedTimestamp time.Time 121 localFlowControlWindow int64 122 remoteFlowControlWindow int64 123 socketOptions *channelz.SocketOptionData 124 localAddr net.Addr 125 remoteAddr net.Addr 126 security credentials.ChannelzSecurityValue 127 remoteName string 128} 129 130func (d *dummySocket) ChannelzMetric() *channelz.SocketInternalMetric { 131 return &channelz.SocketInternalMetric{ 132 StreamsStarted: d.streamsStarted, 133 StreamsSucceeded: d.streamsSucceeded, 134 StreamsFailed: d.streamsFailed, 135 MessagesSent: d.messagesSent, 136 MessagesReceived: d.messagesReceived, 137 KeepAlivesSent: d.keepAlivesSent, 138 LastLocalStreamCreatedTimestamp: d.lastLocalStreamCreatedTimestamp, 139 LastRemoteStreamCreatedTimestamp: d.lastRemoteStreamCreatedTimestamp, 140 LastMessageSentTimestamp: d.lastMessageSentTimestamp, 141 LastMessageReceivedTimestamp: d.lastMessageReceivedTimestamp, 142 LocalFlowControlWindow: d.localFlowControlWindow, 143 RemoteFlowControlWindow: d.remoteFlowControlWindow, 144 SocketOptions: d.socketOptions, 145 LocalAddr: d.localAddr, 146 RemoteAddr: d.remoteAddr, 147 Security: d.security, 148 RemoteName: d.remoteName, 149 } 150} 151 152func channelProtoToStruct(c *channelzpb.Channel) *dummyChannel { 153 dc := &dummyChannel{} 154 pdata := c.GetData() 155 switch pdata.GetState().GetState() { 156 case channelzpb.ChannelConnectivityState_UNKNOWN: 157 // TODO: what should we set here? 158 case channelzpb.ChannelConnectivityState_IDLE: 159 dc.state = connectivity.Idle 160 case channelzpb.ChannelConnectivityState_CONNECTING: 161 dc.state = connectivity.Connecting 162 case channelzpb.ChannelConnectivityState_READY: 163 dc.state = connectivity.Ready 164 case channelzpb.ChannelConnectivityState_TRANSIENT_FAILURE: 165 dc.state = connectivity.TransientFailure 166 case channelzpb.ChannelConnectivityState_SHUTDOWN: 167 dc.state = connectivity.Shutdown 168 } 169 dc.target = pdata.GetTarget() 170 dc.callsStarted = pdata.CallsStarted 171 dc.callsSucceeded = pdata.CallsSucceeded 172 dc.callsFailed = pdata.CallsFailed 173 if t, err := ptypes.Timestamp(pdata.GetLastCallStartedTimestamp()); err == nil { 174 if !t.Equal(emptyTime) { 175 dc.lastCallStartedTimestamp = t 176 } 177 } 178 return dc 179} 180 181func serverProtoToStruct(s *channelzpb.Server) *dummyServer { 182 ds := &dummyServer{} 183 pdata := s.GetData() 184 ds.callsStarted = pdata.CallsStarted 185 ds.callsSucceeded = pdata.CallsSucceeded 186 ds.callsFailed = pdata.CallsFailed 187 if t, err := ptypes.Timestamp(pdata.GetLastCallStartedTimestamp()); err == nil { 188 if !t.Equal(emptyTime) { 189 ds.lastCallStartedTimestamp = t 190 } 191 } 192 return ds 193} 194 195func socketProtoToStruct(s *channelzpb.Socket) *dummySocket { 196 ds := &dummySocket{} 197 pdata := s.GetData() 198 ds.streamsStarted = pdata.GetStreamsStarted() 199 ds.streamsSucceeded = pdata.GetStreamsSucceeded() 200 ds.streamsFailed = pdata.GetStreamsFailed() 201 ds.messagesSent = pdata.GetMessagesSent() 202 ds.messagesReceived = pdata.GetMessagesReceived() 203 ds.keepAlivesSent = pdata.GetKeepAlivesSent() 204 if t, err := ptypes.Timestamp(pdata.GetLastLocalStreamCreatedTimestamp()); err == nil { 205 if !t.Equal(emptyTime) { 206 ds.lastLocalStreamCreatedTimestamp = t 207 } 208 } 209 if t, err := ptypes.Timestamp(pdata.GetLastRemoteStreamCreatedTimestamp()); err == nil { 210 if !t.Equal(emptyTime) { 211 ds.lastRemoteStreamCreatedTimestamp = t 212 } 213 } 214 if t, err := ptypes.Timestamp(pdata.GetLastMessageSentTimestamp()); err == nil { 215 if !t.Equal(emptyTime) { 216 ds.lastMessageSentTimestamp = t 217 } 218 } 219 if t, err := ptypes.Timestamp(pdata.GetLastMessageReceivedTimestamp()); err == nil { 220 if !t.Equal(emptyTime) { 221 ds.lastMessageReceivedTimestamp = t 222 } 223 } 224 if v := pdata.GetLocalFlowControlWindow(); v != nil { 225 ds.localFlowControlWindow = v.Value 226 } 227 if v := pdata.GetRemoteFlowControlWindow(); v != nil { 228 ds.remoteFlowControlWindow = v.Value 229 } 230 if v := pdata.GetOption(); v != nil && protoToSocketOpt != nil { 231 ds.socketOptions = protoToSocketOpt(v) 232 } 233 if v := s.GetSecurity(); v != nil { 234 ds.security = protoToSecurity(v) 235 } 236 if local := s.GetLocal(); local != nil { 237 ds.localAddr = protoToAddr(local) 238 } 239 if remote := s.GetRemote(); remote != nil { 240 ds.remoteAddr = protoToAddr(remote) 241 } 242 ds.remoteName = s.GetRemoteName() 243 return ds 244} 245 246func protoToSecurity(protoSecurity *channelzpb.Security) credentials.ChannelzSecurityValue { 247 switch v := protoSecurity.Model.(type) { 248 case *channelzpb.Security_Tls_: 249 return &credentials.TLSChannelzSecurityValue{StandardName: v.Tls.GetStandardName(), LocalCertificate: v.Tls.GetLocalCertificate(), RemoteCertificate: v.Tls.GetRemoteCertificate()} 250 case *channelzpb.Security_Other: 251 sv := &credentials.OtherChannelzSecurityValue{Name: v.Other.GetName()} 252 var x ptypes.DynamicAny 253 if err := ptypes.UnmarshalAny(v.Other.GetValue(), &x); err == nil { 254 sv.Value = x.Message 255 } 256 return sv 257 } 258 return nil 259} 260 261func protoToAddr(a *channelzpb.Address) net.Addr { 262 switch v := a.Address.(type) { 263 case *channelzpb.Address_TcpipAddress: 264 if port := v.TcpipAddress.GetPort(); port != 0 { 265 return &net.TCPAddr{IP: v.TcpipAddress.GetIpAddress(), Port: int(port)} 266 } 267 return &net.IPAddr{IP: v.TcpipAddress.GetIpAddress()} 268 case *channelzpb.Address_UdsAddress_: 269 return &net.UnixAddr{Name: v.UdsAddress.GetFilename(), Net: "unix"} 270 case *channelzpb.Address_OtherAddress_: 271 // TODO: 272 } 273 return nil 274} 275 276func convertSocketRefSliceToMap(sktRefs []*channelzpb.SocketRef) map[int64]string { 277 m := make(map[int64]string) 278 for _, sr := range sktRefs { 279 m[sr.SocketId] = sr.Name 280 } 281 return m 282} 283 284type OtherSecurityValue struct { 285 LocalCertificate []byte `protobuf:"bytes,1,opt,name=local_certificate,json=localCertificate,proto3" json:"local_certificate,omitempty"` 286 RemoteCertificate []byte `protobuf:"bytes,2,opt,name=remote_certificate,json=remoteCertificate,proto3" json:"remote_certificate,omitempty"` 287} 288 289func (m *OtherSecurityValue) Reset() { *m = OtherSecurityValue{} } 290func (m *OtherSecurityValue) String() string { return proto.CompactTextString(m) } 291func (*OtherSecurityValue) ProtoMessage() {} 292 293func init() { 294 // Ad-hoc registering the proto type here to facilitate UnmarshalAny of OtherSecurityValue. 295 proto.RegisterType((*OtherSecurityValue)(nil), "grpc.credentials.OtherChannelzSecurityValue") 296} 297 298func (s) TestGetTopChannels(t *testing.T) { 299 tcs := []*dummyChannel{ 300 { 301 state: connectivity.Connecting, 302 target: "test.channelz:1234", 303 callsStarted: 6, 304 callsSucceeded: 2, 305 callsFailed: 3, 306 lastCallStartedTimestamp: time.Now().UTC(), 307 }, 308 { 309 state: connectivity.Connecting, 310 target: "test.channelz:1234", 311 callsStarted: 1, 312 callsSucceeded: 2, 313 callsFailed: 3, 314 lastCallStartedTimestamp: time.Now().UTC(), 315 }, 316 { 317 state: connectivity.Shutdown, 318 target: "test.channelz:8888", 319 callsStarted: 0, 320 callsSucceeded: 0, 321 callsFailed: 0, 322 }, 323 {}, 324 } 325 czCleanup := channelz.NewChannelzStorage() 326 defer cleanupWrapper(czCleanup, t) 327 for _, c := range tcs { 328 id := channelz.RegisterChannel(c, 0, "") 329 defer channelz.RemoveEntry(id) 330 } 331 s := newCZServer() 332 ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout) 333 defer cancel() 334 resp, _ := s.GetTopChannels(ctx, &channelzpb.GetTopChannelsRequest{StartChannelId: 0}) 335 if !resp.GetEnd() { 336 t.Fatalf("resp.GetEnd() want true, got %v", resp.GetEnd()) 337 } 338 for i, c := range resp.GetChannel() { 339 if !reflect.DeepEqual(channelProtoToStruct(c), tcs[i]) { 340 t.Fatalf("dummyChannel: %d, want: %#v, got: %#v", i, tcs[i], channelProtoToStruct(c)) 341 } 342 } 343 for i := 0; i < 50; i++ { 344 id := channelz.RegisterChannel(tcs[0], 0, "") 345 defer channelz.RemoveEntry(id) 346 } 347 resp, _ = s.GetTopChannels(ctx, &channelzpb.GetTopChannelsRequest{StartChannelId: 0}) 348 if resp.GetEnd() { 349 t.Fatalf("resp.GetEnd() want false, got %v", resp.GetEnd()) 350 } 351} 352 353func (s) TestGetServers(t *testing.T) { 354 ss := []*dummyServer{ 355 { 356 callsStarted: 6, 357 callsSucceeded: 2, 358 callsFailed: 3, 359 lastCallStartedTimestamp: time.Now().UTC(), 360 }, 361 { 362 callsStarted: 1, 363 callsSucceeded: 2, 364 callsFailed: 3, 365 lastCallStartedTimestamp: time.Now().UTC(), 366 }, 367 { 368 callsStarted: 1, 369 callsSucceeded: 0, 370 callsFailed: 0, 371 lastCallStartedTimestamp: time.Now().UTC(), 372 }, 373 } 374 czCleanup := channelz.NewChannelzStorage() 375 defer cleanupWrapper(czCleanup, t) 376 for _, s := range ss { 377 id := channelz.RegisterServer(s, "") 378 defer channelz.RemoveEntry(id) 379 } 380 svr := newCZServer() 381 ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout) 382 defer cancel() 383 resp, _ := svr.GetServers(ctx, &channelzpb.GetServersRequest{StartServerId: 0}) 384 if !resp.GetEnd() { 385 t.Fatalf("resp.GetEnd() want true, got %v", resp.GetEnd()) 386 } 387 for i, s := range resp.GetServer() { 388 if !reflect.DeepEqual(serverProtoToStruct(s), ss[i]) { 389 t.Fatalf("dummyServer: %d, want: %#v, got: %#v", i, ss[i], serverProtoToStruct(s)) 390 } 391 } 392 for i := 0; i < 50; i++ { 393 id := channelz.RegisterServer(ss[0], "") 394 defer channelz.RemoveEntry(id) 395 } 396 resp, _ = svr.GetServers(ctx, &channelzpb.GetServersRequest{StartServerId: 0}) 397 if resp.GetEnd() { 398 t.Fatalf("resp.GetEnd() want false, got %v", resp.GetEnd()) 399 } 400} 401 402func (s) TestGetServerSockets(t *testing.T) { 403 czCleanup := channelz.NewChannelzStorage() 404 defer cleanupWrapper(czCleanup, t) 405 svrID := channelz.RegisterServer(&dummyServer{}, "") 406 defer channelz.RemoveEntry(svrID) 407 refNames := []string{"listen socket 1", "normal socket 1", "normal socket 2"} 408 ids := make([]int64, 3) 409 ids[0] = channelz.RegisterListenSocket(&dummySocket{}, svrID, refNames[0]) 410 ids[1] = channelz.RegisterNormalSocket(&dummySocket{}, svrID, refNames[1]) 411 ids[2] = channelz.RegisterNormalSocket(&dummySocket{}, svrID, refNames[2]) 412 for _, id := range ids { 413 defer channelz.RemoveEntry(id) 414 } 415 svr := newCZServer() 416 ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout) 417 defer cancel() 418 resp, _ := svr.GetServerSockets(ctx, &channelzpb.GetServerSocketsRequest{ServerId: svrID, StartSocketId: 0}) 419 if !resp.GetEnd() { 420 t.Fatalf("resp.GetEnd() want: true, got: %v", resp.GetEnd()) 421 } 422 // GetServerSockets only return normal sockets. 423 want := map[int64]string{ 424 ids[1]: refNames[1], 425 ids[2]: refNames[2], 426 } 427 if !reflect.DeepEqual(convertSocketRefSliceToMap(resp.GetSocketRef()), want) { 428 t.Fatalf("GetServerSockets want: %#v, got: %#v", want, resp.GetSocketRef()) 429 } 430 431 for i := 0; i < 50; i++ { 432 id := channelz.RegisterNormalSocket(&dummySocket{}, svrID, "") 433 defer channelz.RemoveEntry(id) 434 } 435 resp, _ = svr.GetServerSockets(ctx, &channelzpb.GetServerSocketsRequest{ServerId: svrID, StartSocketId: 0}) 436 if resp.GetEnd() { 437 t.Fatalf("resp.GetEnd() want false, got %v", resp.GetEnd()) 438 } 439} 440 441// This test makes a GetServerSockets with a non-zero start ID, and expect only 442// sockets with ID >= the given start ID. 443func (s) TestGetServerSocketsNonZeroStartID(t *testing.T) { 444 czCleanup := channelz.NewChannelzStorage() 445 defer cleanupWrapper(czCleanup, t) 446 svrID := channelz.RegisterServer(&dummyServer{}, "") 447 defer channelz.RemoveEntry(svrID) 448 refNames := []string{"listen socket 1", "normal socket 1", "normal socket 2"} 449 ids := make([]int64, 3) 450 ids[0] = channelz.RegisterListenSocket(&dummySocket{}, svrID, refNames[0]) 451 ids[1] = channelz.RegisterNormalSocket(&dummySocket{}, svrID, refNames[1]) 452 ids[2] = channelz.RegisterNormalSocket(&dummySocket{}, svrID, refNames[2]) 453 for _, id := range ids { 454 defer channelz.RemoveEntry(id) 455 } 456 svr := newCZServer() 457 ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout) 458 defer cancel() 459 // Make GetServerSockets with startID = ids[1]+1, so socket-1 won't be 460 // included in the response. 461 resp, _ := svr.GetServerSockets(ctx, &channelzpb.GetServerSocketsRequest{ServerId: svrID, StartSocketId: ids[1] + 1}) 462 if !resp.GetEnd() { 463 t.Fatalf("resp.GetEnd() want: true, got: %v", resp.GetEnd()) 464 } 465 // GetServerSockets only return normal socket-2, socket-1 should be 466 // filtered by start ID. 467 want := map[int64]string{ 468 ids[2]: refNames[2], 469 } 470 if !reflect.DeepEqual(convertSocketRefSliceToMap(resp.GetSocketRef()), want) { 471 t.Fatalf("GetServerSockets want: %#v, got: %#v", want, resp.GetSocketRef()) 472 } 473} 474 475func (s) TestGetChannel(t *testing.T) { 476 czCleanup := channelz.NewChannelzStorage() 477 defer cleanupWrapper(czCleanup, t) 478 refNames := []string{"top channel 1", "nested channel 1", "sub channel 2", "nested channel 3"} 479 ids := make([]int64, 4) 480 ids[0] = channelz.RegisterChannel(&dummyChannel{}, 0, refNames[0]) 481 channelz.AddTraceEvent(logger, ids[0], 0, &channelz.TraceEventDesc{ 482 Desc: "Channel Created", 483 Severity: channelz.CtInfo, 484 }) 485 ids[1] = channelz.RegisterChannel(&dummyChannel{}, ids[0], refNames[1]) 486 channelz.AddTraceEvent(logger, ids[1], 0, &channelz.TraceEventDesc{ 487 Desc: "Channel Created", 488 Severity: channelz.CtInfo, 489 Parent: &channelz.TraceEventDesc{ 490 Desc: fmt.Sprintf("Nested Channel(id:%d) created", ids[1]), 491 Severity: channelz.CtInfo, 492 }, 493 }) 494 495 ids[2] = channelz.RegisterSubChannel(&dummyChannel{}, ids[0], refNames[2]) 496 channelz.AddTraceEvent(logger, ids[2], 0, &channelz.TraceEventDesc{ 497 Desc: "SubChannel Created", 498 Severity: channelz.CtInfo, 499 Parent: &channelz.TraceEventDesc{ 500 Desc: fmt.Sprintf("SubChannel(id:%d) created", ids[2]), 501 Severity: channelz.CtInfo, 502 }, 503 }) 504 ids[3] = channelz.RegisterChannel(&dummyChannel{}, ids[1], refNames[3]) 505 channelz.AddTraceEvent(logger, ids[3], 0, &channelz.TraceEventDesc{ 506 Desc: "Channel Created", 507 Severity: channelz.CtInfo, 508 Parent: &channelz.TraceEventDesc{ 509 Desc: fmt.Sprintf("Nested Channel(id:%d) created", ids[3]), 510 Severity: channelz.CtInfo, 511 }, 512 }) 513 channelz.AddTraceEvent(logger, ids[0], 0, &channelz.TraceEventDesc{ 514 Desc: fmt.Sprintf("Channel Connectivity change to %v", connectivity.Ready), 515 Severity: channelz.CtInfo, 516 }) 517 channelz.AddTraceEvent(logger, ids[0], 0, &channelz.TraceEventDesc{ 518 Desc: "Resolver returns an empty address list", 519 Severity: channelz.CtWarning, 520 }) 521 for _, id := range ids { 522 defer channelz.RemoveEntry(id) 523 } 524 svr := newCZServer() 525 ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout) 526 defer cancel() 527 resp, _ := svr.GetChannel(ctx, &channelzpb.GetChannelRequest{ChannelId: ids[0]}) 528 metrics := resp.GetChannel() 529 subChans := metrics.GetSubchannelRef() 530 if len(subChans) != 1 || subChans[0].GetName() != refNames[2] || subChans[0].GetSubchannelId() != ids[2] { 531 t.Fatalf("metrics.GetSubChannelRef() want %#v, got %#v", []*channelzpb.SubchannelRef{{SubchannelId: ids[2], Name: refNames[2]}}, subChans) 532 } 533 nestedChans := metrics.GetChannelRef() 534 if len(nestedChans) != 1 || nestedChans[0].GetName() != refNames[1] || nestedChans[0].GetChannelId() != ids[1] { 535 t.Fatalf("metrics.GetChannelRef() want %#v, got %#v", []*channelzpb.ChannelRef{{ChannelId: ids[1], Name: refNames[1]}}, nestedChans) 536 } 537 trace := metrics.GetData().GetTrace() 538 want := []struct { 539 desc string 540 severity channelzpb.ChannelTraceEvent_Severity 541 childID int64 542 childRef string 543 }{ 544 {desc: "Channel Created", severity: channelzpb.ChannelTraceEvent_CT_INFO}, 545 {desc: fmt.Sprintf("Nested Channel(id:%d) created", ids[1]), severity: channelzpb.ChannelTraceEvent_CT_INFO, childID: ids[1], childRef: refNames[1]}, 546 {desc: fmt.Sprintf("SubChannel(id:%d) created", ids[2]), severity: channelzpb.ChannelTraceEvent_CT_INFO, childID: ids[2], childRef: refNames[2]}, 547 {desc: fmt.Sprintf("Channel Connectivity change to %v", connectivity.Ready), severity: channelzpb.ChannelTraceEvent_CT_INFO}, 548 {desc: "Resolver returns an empty address list", severity: channelzpb.ChannelTraceEvent_CT_WARNING}, 549 } 550 551 for i, e := range trace.Events { 552 if e.GetDescription() != want[i].desc { 553 t.Fatalf("trace: GetDescription want %#v, got %#v", want[i].desc, e.GetDescription()) 554 } 555 if e.GetSeverity() != want[i].severity { 556 t.Fatalf("trace: GetSeverity want %#v, got %#v", want[i].severity, e.GetSeverity()) 557 } 558 if want[i].childID == 0 && (e.GetChannelRef() != nil || e.GetSubchannelRef() != nil) { 559 t.Fatalf("trace: GetChannelRef() should return nil, as there is no reference") 560 } 561 if e.GetChannelRef().GetChannelId() != want[i].childID || e.GetChannelRef().GetName() != want[i].childRef { 562 if e.GetSubchannelRef().GetSubchannelId() != want[i].childID || e.GetSubchannelRef().GetName() != want[i].childRef { 563 t.Fatalf("trace: GetChannelRef/GetSubchannelRef want (child ID: %d, child name: %q), got %#v and %#v", want[i].childID, want[i].childRef, e.GetChannelRef(), e.GetSubchannelRef()) 564 } 565 } 566 } 567 resp, _ = svr.GetChannel(ctx, &channelzpb.GetChannelRequest{ChannelId: ids[1]}) 568 metrics = resp.GetChannel() 569 nestedChans = metrics.GetChannelRef() 570 if len(nestedChans) != 1 || nestedChans[0].GetName() != refNames[3] || nestedChans[0].GetChannelId() != ids[3] { 571 t.Fatalf("metrics.GetChannelRef() want %#v, got %#v", []*channelzpb.ChannelRef{{ChannelId: ids[3], Name: refNames[3]}}, nestedChans) 572 } 573} 574 575func (s) TestGetSubChannel(t *testing.T) { 576 var ( 577 subchanCreated = "SubChannel Created" 578 subchanConnectivityChange = fmt.Sprintf("Subchannel Connectivity change to %v", connectivity.Ready) 579 subChanPickNewAddress = fmt.Sprintf("Subchannel picks a new address %q to connect", "0.0.0.0") 580 ) 581 czCleanup := channelz.NewChannelzStorage() 582 defer cleanupWrapper(czCleanup, t) 583 refNames := []string{"top channel 1", "sub channel 1", "socket 1", "socket 2"} 584 ids := make([]int64, 4) 585 ids[0] = channelz.RegisterChannel(&dummyChannel{}, 0, refNames[0]) 586 channelz.AddTraceEvent(logger, ids[0], 0, &channelz.TraceEventDesc{ 587 Desc: "Channel Created", 588 Severity: channelz.CtInfo, 589 }) 590 ids[1] = channelz.RegisterSubChannel(&dummyChannel{}, ids[0], refNames[1]) 591 channelz.AddTraceEvent(logger, ids[1], 0, &channelz.TraceEventDesc{ 592 Desc: subchanCreated, 593 Severity: channelz.CtInfo, 594 Parent: &channelz.TraceEventDesc{ 595 Desc: fmt.Sprintf("Nested Channel(id:%d) created", ids[0]), 596 Severity: channelz.CtInfo, 597 }, 598 }) 599 ids[2] = channelz.RegisterNormalSocket(&dummySocket{}, ids[1], refNames[2]) 600 ids[3] = channelz.RegisterNormalSocket(&dummySocket{}, ids[1], refNames[3]) 601 channelz.AddTraceEvent(logger, ids[1], 0, &channelz.TraceEventDesc{ 602 Desc: subchanConnectivityChange, 603 Severity: channelz.CtInfo, 604 }) 605 channelz.AddTraceEvent(logger, ids[1], 0, &channelz.TraceEventDesc{ 606 Desc: subChanPickNewAddress, 607 Severity: channelz.CtInfo, 608 }) 609 for _, id := range ids { 610 defer channelz.RemoveEntry(id) 611 } 612 svr := newCZServer() 613 ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout) 614 defer cancel() 615 resp, _ := svr.GetSubchannel(ctx, &channelzpb.GetSubchannelRequest{SubchannelId: ids[1]}) 616 metrics := resp.GetSubchannel() 617 want := map[int64]string{ 618 ids[2]: refNames[2], 619 ids[3]: refNames[3], 620 } 621 if !reflect.DeepEqual(convertSocketRefSliceToMap(metrics.GetSocketRef()), want) { 622 t.Fatalf("metrics.GetSocketRef() want %#v: got: %#v", want, metrics.GetSocketRef()) 623 } 624 625 trace := metrics.GetData().GetTrace() 626 wantTrace := []struct { 627 desc string 628 severity channelzpb.ChannelTraceEvent_Severity 629 childID int64 630 childRef string 631 }{ 632 {desc: subchanCreated, severity: channelzpb.ChannelTraceEvent_CT_INFO}, 633 {desc: subchanConnectivityChange, severity: channelzpb.ChannelTraceEvent_CT_INFO}, 634 {desc: subChanPickNewAddress, severity: channelzpb.ChannelTraceEvent_CT_INFO}, 635 } 636 for i, e := range trace.Events { 637 if e.GetDescription() != wantTrace[i].desc { 638 t.Fatalf("trace: GetDescription want %#v, got %#v", wantTrace[i].desc, e.GetDescription()) 639 } 640 if e.GetSeverity() != wantTrace[i].severity { 641 t.Fatalf("trace: GetSeverity want %#v, got %#v", wantTrace[i].severity, e.GetSeverity()) 642 } 643 if wantTrace[i].childID == 0 && (e.GetChannelRef() != nil || e.GetSubchannelRef() != nil) { 644 t.Fatalf("trace: GetChannelRef() should return nil, as there is no reference") 645 } 646 if e.GetChannelRef().GetChannelId() != wantTrace[i].childID || e.GetChannelRef().GetName() != wantTrace[i].childRef { 647 if e.GetSubchannelRef().GetSubchannelId() != wantTrace[i].childID || e.GetSubchannelRef().GetName() != wantTrace[i].childRef { 648 t.Fatalf("trace: GetChannelRef/GetSubchannelRef want (child ID: %d, child name: %q), got %#v and %#v", wantTrace[i].childID, wantTrace[i].childRef, e.GetChannelRef(), e.GetSubchannelRef()) 649 } 650 } 651 } 652} 653 654func (s) TestGetSocket(t *testing.T) { 655 czCleanup := channelz.NewChannelzStorage() 656 defer cleanupWrapper(czCleanup, t) 657 ss := []*dummySocket{ 658 { 659 streamsStarted: 10, 660 streamsSucceeded: 2, 661 streamsFailed: 3, 662 messagesSent: 20, 663 messagesReceived: 10, 664 keepAlivesSent: 2, 665 lastLocalStreamCreatedTimestamp: time.Now().UTC(), 666 lastRemoteStreamCreatedTimestamp: time.Now().UTC(), 667 lastMessageSentTimestamp: time.Now().UTC(), 668 lastMessageReceivedTimestamp: time.Now().UTC(), 669 localFlowControlWindow: 65536, 670 remoteFlowControlWindow: 1024, 671 localAddr: &net.TCPAddr{IP: net.ParseIP("1.0.0.1"), Port: 10001}, 672 remoteAddr: &net.TCPAddr{IP: net.ParseIP("12.0.0.1"), Port: 10002}, 673 remoteName: "remote.remote", 674 }, 675 { 676 streamsStarted: 10, 677 streamsSucceeded: 2, 678 streamsFailed: 3, 679 messagesSent: 20, 680 messagesReceived: 10, 681 keepAlivesSent: 2, 682 lastRemoteStreamCreatedTimestamp: time.Now().UTC(), 683 lastMessageSentTimestamp: time.Now().UTC(), 684 lastMessageReceivedTimestamp: time.Now().UTC(), 685 localFlowControlWindow: 65536, 686 remoteFlowControlWindow: 1024, 687 localAddr: &net.UnixAddr{Name: "file.path", Net: "unix"}, 688 remoteAddr: &net.UnixAddr{Name: "another.path", Net: "unix"}, 689 remoteName: "remote.remote", 690 }, 691 { 692 streamsStarted: 5, 693 streamsSucceeded: 2, 694 streamsFailed: 3, 695 messagesSent: 20, 696 messagesReceived: 10, 697 keepAlivesSent: 2, 698 lastLocalStreamCreatedTimestamp: time.Now().UTC(), 699 lastMessageSentTimestamp: time.Now().UTC(), 700 lastMessageReceivedTimestamp: time.Now().UTC(), 701 localFlowControlWindow: 65536, 702 remoteFlowControlWindow: 10240, 703 localAddr: &net.IPAddr{IP: net.ParseIP("1.0.0.1")}, 704 remoteAddr: &net.IPAddr{IP: net.ParseIP("9.0.0.1")}, 705 remoteName: "", 706 }, 707 { 708 localAddr: &net.TCPAddr{IP: net.ParseIP("127.0.0.1"), Port: 10001}, 709 }, 710 { 711 security: &credentials.TLSChannelzSecurityValue{ 712 StandardName: "TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256", 713 RemoteCertificate: []byte{48, 130, 2, 156, 48, 130, 2, 5, 160}, 714 }, 715 }, 716 { 717 security: &credentials.OtherChannelzSecurityValue{ 718 Name: "XXXX", 719 }, 720 }, 721 { 722 security: &credentials.OtherChannelzSecurityValue{ 723 Name: "YYYY", 724 Value: &OtherSecurityValue{LocalCertificate: []byte{1, 2, 3}, RemoteCertificate: []byte{4, 5, 6}}, 725 }, 726 }, 727 } 728 svr := newCZServer() 729 ids := make([]int64, len(ss)) 730 svrID := channelz.RegisterServer(&dummyServer{}, "") 731 defer channelz.RemoveEntry(svrID) 732 for i, s := range ss { 733 ids[i] = channelz.RegisterNormalSocket(s, svrID, strconv.Itoa(i)) 734 defer channelz.RemoveEntry(ids[i]) 735 } 736 ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout) 737 defer cancel() 738 for i, s := range ss { 739 resp, _ := svr.GetSocket(ctx, &channelzpb.GetSocketRequest{SocketId: ids[i]}) 740 metrics := resp.GetSocket() 741 if !reflect.DeepEqual(metrics.GetRef(), &channelzpb.SocketRef{SocketId: ids[i], Name: strconv.Itoa(i)}) || !reflect.DeepEqual(socketProtoToStruct(metrics), s) { 742 t.Fatalf("resp.GetSocket() want: metrics.GetRef() = %#v and %#v, got: metrics.GetRef() = %#v and %#v", &channelzpb.SocketRef{SocketId: ids[i], Name: strconv.Itoa(i)}, s, metrics.GetRef(), socketProtoToStruct(metrics)) 743 } 744 } 745} 746