service_test.go 27 KB

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