d8dfun/virtual-kubelet
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

CRI Provider implementation (#195)

Browse Source 
* First commit of CRI provider. Vendor deps not included

* First commit of CRI provider. Vendor deps not included

* Tidy up comments and format code

* vendor grpc, CRI APIs, update protobuf and tidy logging

* First commit of CRI provider. Vendor deps not included

* Tidy up comments and format code

* vendor grpc, CRI APIs, update protobuf and tidy logging

* Add README

* Fix errors in make test
This commit is contained in:
Ben Corrie
2018-05-15 08:50:52 -07:00
committed by Ria Bhatia
parent 4b61932ac1
commit c6c89f062f
702 changed files with 456968 additions and 11190 deletions
Download Patch File Download Diff File Expand all files Collapse all files

573
vendor/google.golang.org/grpc/channelz/funcs.go generated vendored Normal file
View File

@@ -0,0 +1,573 @@
/*
*
* Copyright 2018 gRPC authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
// Package channelz defines APIs for enabling channelz service, entry
// registration/deletion, and accessing channelz data. It also defines channelz
// metric struct formats.
//
// All APIs in this package are experimental.
package channelz
import (
"sort"
"sync"
"sync/atomic"
"google.golang.org/grpc/grpclog"
)
var (
db dbWrapper
idGen idGenerator
// EntryPerPage defines the number of channelz entries to be shown on a web page.
EntryPerPage = 50
curState int32
)
// TurnOn turns on channelz data collection.
func TurnOn() {
if !IsOn() {
NewChannelzStorage()
atomic.StoreInt32(&curState, 1)
}
}
// IsOn returns whether channelz data collection is on.
func IsOn() bool {
return atomic.CompareAndSwapInt32(&curState, 1, 1)
}
// dbWarpper wraps around a reference to internal channelz data storage, and
// provide synchronized functionality to set and get the reference.
type dbWrapper struct {
mu sync.RWMutex
DB *channelMap
}
func (d *dbWrapper) set(db *channelMap) {
d.mu.Lock()
d.DB = db
d.mu.Unlock()
}
func (d *dbWrapper) get() *channelMap {
d.mu.RLock()
defer d.mu.RUnlock()
return d.DB
}
// NewChannelzStorage initializes channelz data storage and id generator.
//
// Note: This function is exported for testing purpose only. User should not call
// it in most cases.
func NewChannelzStorage() {
db.set(&channelMap{
topLevelChannels: make(map[int64]struct{}),
channels: make(map[int64]*channel),
listenSockets: make(map[int64]*listenSocket),
normalSockets: make(map[int64]*normalSocket),
servers: make(map[int64]*server),
subChannels: make(map[int64]*subChannel),
})
idGen.reset()
}
// GetTopChannels returns a slice of top channel's ChannelMetric, along with a
// boolean indicating whether there's more top channels to be queried for.
//
// The arg id specifies that only top channel with id at or above it will be included
// in the result. The returned slice is up to a length of EntryPerPage, and is
// sorted in ascending id order.
func GetTopChannels(id int64) ([]*ChannelMetric, bool) {
return db.get().GetTopChannels(id)
}
// GetServers returns a slice of server's ServerMetric, along with a
// boolean indicating whether there's more servers to be queried for.
//
// The arg id specifies that only server with id at or above it will be included
// in the result. The returned slice is up to a length of EntryPerPage, and is
// sorted in ascending id order.
func GetServers(id int64) ([]*ServerMetric, bool) {
return db.get().GetServers(id)
}
// GetServerSockets returns a slice of server's (identified by id) normal socket's
// SocketMetric, along with a boolean indicating whether there's more sockets to
// be queried for.
//
// The arg startID specifies that only sockets with id at or above it will be
// included in the result. The returned slice is up to a length of EntryPerPage,
// and is sorted in ascending id order.
func GetServerSockets(id int64, startID int64) ([]*SocketMetric, bool) {
return db.get().GetServerSockets(id, startID)
}
// GetChannel returns the ChannelMetric for the channel (identified by id).
func GetChannel(id int64) *ChannelMetric {
return db.get().GetChannel(id)
}
// GetSubChannel returns the SubChannelMetric for the subchannel (identified by id).
func GetSubChannel(id int64) *SubChannelMetric {
return db.get().GetSubChannel(id)
}
// GetSocket returns the SocketInternalMetric for the socket (identified by id).
func GetSocket(id int64) *SocketMetric {
return db.get().GetSocket(id)
}
// RegisterChannel registers the given channel c in channelz database with ref
// as its reference name, and add it to the child list of its parent (identified
// by pid). pid = 0 means no parent. It returns the unique channelz tracking id
// assigned to this channel.
func RegisterChannel(c Channel, pid int64, ref string) int64 {
id := idGen.genID()
cn := &channel{
refName: ref,
c: c,
subChans: make(map[int64]string),
nestedChans: make(map[int64]string),
id: id,
pid: pid,
}
if pid == 0 {
db.get().addChannel(id, cn, true, pid, ref)
} else {
db.get().addChannel(id, cn, false, pid, ref)
}
return id
}
// RegisterSubChannel registers the given channel c in channelz database with ref
// as its reference name, and add it to the child list of its parent (identified
// by pid). It returns the unique channelz tracking id assigned to this subchannel.
func RegisterSubChannel(c Channel, pid int64, ref string) int64 {
if pid == 0 {
grpclog.Error("a SubChannel's parent id cannot be 0")
return 0
}
id := idGen.genID()
sc := &subChannel{
refName: ref,
c: c,
sockets: make(map[int64]string),
id: id,
pid: pid,
}
db.get().addSubChannel(id, sc, pid, ref)
return id
}
// RegisterServer registers the given server s in channelz database. It returns
// the unique channelz tracking id assigned to this server.
func RegisterServer(s Server, ref string) int64 {
id := idGen.genID()
svr := &server{
refName: ref,
s: s,
sockets: make(map[int64]string),
listenSockets: make(map[int64]string),
id: id,
}
db.get().addServer(id, svr)
return id
}
// RegisterListenSocket registers the given listen socket s in channelz database
// with ref as its reference name, and add it to the child list of its parent
// (identified by pid). It returns the unique channelz tracking id assigned to
// this listen socket.
func RegisterListenSocket(s Socket, pid int64, ref string) int64 {
if pid == 0 {
grpclog.Error("a ListenSocket's parent id cannot be 0")
return 0
}
id := idGen.genID()
ls := &listenSocket{refName: ref, s: s, id: id, pid: pid}
db.get().addListenSocket(id, ls, pid, ref)
return id
}
// RegisterNormalSocket registers the given normal socket s in channelz database
// with ref as its reference name, and add it to the child list of its parent
// (identified by pid). It returns the unique channelz tracking id assigned to
// this normal socket.
func RegisterNormalSocket(s Socket, pid int64, ref string) int64 {
if pid == 0 {
grpclog.Error("a NormalSocket's parent id cannot be 0")
return 0
}
id := idGen.genID()
ns := &normalSocket{refName: ref, s: s, id: id, pid: pid}
db.get().addNormalSocket(id, ns, pid, ref)
return id
}
// RemoveEntry removes an entry with unique channelz trakcing id to be id from
// channelz database.
func RemoveEntry(id int64) {
db.get().removeEntry(id)
}
// channelMap is the storage data structure for channelz.
// Methods of channelMap can be divided in two two categories with respect to locking.
// 1. Methods acquire the global lock.
// 2. Methods that can only be called when global lock is held.
// A second type of method need always to be called inside a first type of method.
type channelMap struct {
mu sync.RWMutex
topLevelChannels map[int64]struct{}
servers map[int64]*server
channels map[int64]*channel
subChannels map[int64]*subChannel
listenSockets map[int64]*listenSocket
normalSockets map[int64]*normalSocket
}
func (c *channelMap) addServer(id int64, s *server) {
c.mu.Lock()
s.cm = c
c.servers[id] = s
c.mu.Unlock()
}
func (c *channelMap) addChannel(id int64, cn *channel, isTopChannel bool, pid int64, ref string) {
c.mu.Lock()
cn.cm = c
c.channels[id] = cn
if isTopChannel {
c.topLevelChannels[id] = struct{}{}
} else {
c.findEntry(pid).addChild(id, cn)
}
c.mu.Unlock()
}
func (c *channelMap) addSubChannel(id int64, sc *subChannel, pid int64, ref string) {
c.mu.Lock()
sc.cm = c
c.subChannels[id] = sc
c.findEntry(pid).addChild(id, sc)
c.mu.Unlock()
}
func (c *channelMap) addListenSocket(id int64, ls *listenSocket, pid int64, ref string) {
c.mu.Lock()
ls.cm = c
c.listenSockets[id] = ls
c.findEntry(pid).addChild(id, ls)
c.mu.Unlock()
}
func (c *channelMap) addNormalSocket(id int64, ns *normalSocket, pid int64, ref string) {
c.mu.Lock()
ns.cm = c
c.normalSockets[id] = ns
c.findEntry(pid).addChild(id, ns)
c.mu.Unlock()
}
// removeEntry triggers the removal of an entry, which may not indeed delete the
// entry, if it has to wait on the deletion of its children, or may lead to a chain
// of entry deletion. For example, deleting the last socket of a gracefully shutting
// down server will lead to the server being also deleted.
func (c *channelMap) removeEntry(id int64) {
c.mu.Lock()
c.findEntry(id).triggerDelete()
c.mu.Unlock()
}
// c.mu must be held by the caller.
func (c *channelMap) findEntry(id int64) entry {
var v entry
var ok bool
if v, ok = c.channels[id]; ok {
return v
}
if v, ok = c.subChannels[id]; ok {
return v
}
if v, ok = c.servers[id]; ok {
return v
}
if v, ok = c.listenSockets[id]; ok {
return v
}
if v, ok = c.normalSockets[id]; ok {
return v
}
return &dummyEntry{idNotFound: id}
}
// c.mu must be held by the caller
// deleteEntry simply deletes an entry from the channelMap. Before calling this
// method, caller must check this entry is ready to be deleted, i.e removeEntry()
// has been called on it, and no children still exist.
// Conditionals are ordered by the expected frequency of deletion of each entity
// type, in order to optimize performance.
func (c *channelMap) deleteEntry(id int64) {
var ok bool
if _, ok = c.normalSockets[id]; ok {
delete(c.normalSockets, id)
return
}
if _, ok = c.subChannels[id]; ok {
delete(c.subChannels, id)
return
}
if _, ok = c.channels[id]; ok {
delete(c.channels, id)
delete(c.topLevelChannels, id)
return
}
if _, ok = c.listenSockets[id]; ok {
delete(c.listenSockets, id)
return
}
if _, ok = c.servers[id]; ok {
delete(c.servers, id)
return
}
}
type int64Slice []int64
func (s int64Slice) Len() int { return len(s) }
func (s int64Slice) Swap(i, j int) { s[i], s[j] = s[j], s[i] }
func (s int64Slice) Less(i, j int) bool { return s[i] < s[j] }
func copyMap(m map[int64]string) map[int64]string {
n := make(map[int64]string)
for k, v := range m {
n[k] = v
}
return n
}
func min(a, b int) int {
if a < b {
return a
}
return b
}
func (c *channelMap) GetTopChannels(id int64) ([]*ChannelMetric, bool) {
c.mu.RLock()
l := len(c.topLevelChannels)
ids := make([]int64, 0, l)
cns := make([]*channel, 0, min(l, EntryPerPage))
for k := range c.topLevelChannels {
ids = append(ids, k)
}
sort.Sort(int64Slice(ids))
idx := sort.Search(len(ids), func(i int) bool { return ids[i] >= id })
count := 0
var end bool
var t []*ChannelMetric
for i, v := range ids[idx:] {
if count == EntryPerPage {
break
}
if cn, ok := c.channels[v]; ok {
cns = append(cns, cn)
t = append(t, &ChannelMetric{
NestedChans: copyMap(cn.nestedChans),
SubChans: copyMap(cn.subChans),
})
count++
}
if i == len(ids[idx:])-1 {
end = true
break
}
}
c.mu.RUnlock()
if count == 0 {
end = true
}
for i, cn := range cns {
t[i].ChannelData = cn.c.ChannelzMetric()
t[i].ID = cn.id
t[i].RefName = cn.refName
}
return t, end
}
func (c *channelMap) GetServers(id int64) ([]*ServerMetric, bool) {
c.mu.RLock()
l := len(c.servers)
ids := make([]int64, 0, l)
ss := make([]*server, 0, min(l, EntryPerPage))
for k := range c.servers {
ids = append(ids, k)
}
sort.Sort(int64Slice(ids))
idx := sort.Search(len(ids), func(i int) bool { return ids[i] >= id })
count := 0
var end bool
var s []*ServerMetric
for i, v := range ids[idx:] {
if count == EntryPerPage {
break
}
if svr, ok := c.servers[v]; ok {
ss = append(ss, svr)
s = append(s, &ServerMetric{
ListenSockets: copyMap(svr.listenSockets),
})
count++
}
if i == len(ids[idx:])-1 {
end = true
break
}
}
c.mu.RUnlock()
if count == 0 {
end = true
}
for i, svr := range ss {
s[i].ServerData = svr.s.ChannelzMetric()
s[i].ID = svr.id
s[i].RefName = svr.refName
}
return s, end
}
func (c *channelMap) GetServerSockets(id int64, startID int64) ([]*SocketMetric, bool) {
var svr *server
var ok bool
c.mu.RLock()
if svr, ok = c.servers[id]; !ok {
// server with id doesn't exist.
c.mu.RUnlock()
return nil, true
}
svrskts := svr.sockets
l := len(svrskts)
ids := make([]int64, 0, l)
sks := make([]*normalSocket, 0, min(l, EntryPerPage))
for k := range svrskts {
ids = append(ids, k)
}
sort.Sort((int64Slice(ids)))
idx := sort.Search(len(ids), func(i int) bool { return ids[i] >= id })
count := 0
var end bool
for i, v := range ids[idx:] {
if count == EntryPerPage {
break
}
if ns, ok := c.normalSockets[v]; ok {
sks = append(sks, ns)
count++
}
if i == len(ids[idx:])-1 {
end = true
break
}
}
c.mu.RUnlock()
if count == 0 {
end = true
}
var s []*SocketMetric
for _, ns := range sks {
sm := &SocketMetric{}
sm.SocketData = ns.s.ChannelzMetric()
sm.ID = ns.id
sm.RefName = ns.refName
s = append(s, sm)
}
return s, end
}
func (c *channelMap) GetChannel(id int64) *ChannelMetric {
cm := &ChannelMetric{}
var cn *channel
var ok bool
c.mu.RLock()
if cn, ok = c.channels[id]; !ok {
// channel with id doesn't exist.
c.mu.RUnlock()
return nil
}
cm.NestedChans = copyMap(cn.nestedChans)
cm.SubChans = copyMap(cn.subChans)
c.mu.RUnlock()
cm.ChannelData = cn.c.ChannelzMetric()
cm.ID = cn.id
cm.RefName = cn.refName
return cm
}
func (c *channelMap) GetSubChannel(id int64) *SubChannelMetric {
cm := &SubChannelMetric{}
var sc *subChannel
var ok bool
c.mu.RLock()
if sc, ok = c.subChannels[id]; !ok {
// subchannel with id doesn't exist.
c.mu.RUnlock()
return nil
}
cm.Sockets = copyMap(sc.sockets)
c.mu.RUnlock()
cm.ChannelData = sc.c.ChannelzMetric()
cm.ID = sc.id
cm.RefName = sc.refName
return cm
}
func (c *channelMap) GetSocket(id int64) *SocketMetric {
sm := &SocketMetric{}
c.mu.RLock()
if ls, ok := c.listenSockets[id]; ok {
c.mu.RUnlock()
sm.SocketData = ls.s.ChannelzMetric()
sm.ID = ls.id
sm.RefName = ls.refName
return sm
}
if ns, ok := c.normalSockets[id]; ok {
c.mu.RUnlock()
sm.SocketData = ns.s.ChannelzMetric()
sm.ID = ns.id
sm.RefName = ns.refName
return sm
}
c.mu.RUnlock()
return nil
}
type idGenerator struct {
id int64
}
func (i *idGenerator) reset() {
atomic.StoreInt64(&i.id, 0)
}
func (i *idGenerator) genID() int64 {
return atomic.AddInt64(&i.id, 1)
}

3004
vendor/google.golang.org/grpc/channelz/grpc_channelz_v1/channelz.pb.go generated vendored Normal file
View File

File diff suppressed because it is too large Load Diff

485
vendor/google.golang.org/grpc/channelz/grpc_channelz_v1/channelz.proto generated vendored Normal file
View File

@@ -0,0 +1,485 @@
// Copyright 2018 gRPC authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// This file defines an interface for exporting monitoring information
// out of gRPC servers. See the full design at
// https://github.com/grpc/proposal/blob/master/A14-channelz.md
//
// The canonical version of this proto can be found at
// https://github.com/grpc/grpc-proto/blob/master/grpc/channelz/v1/channelz.proto
syntax = "proto3";
package grpc.channelz.v1;
import "google/protobuf/any.proto";
import "google/protobuf/duration.proto";
import "google/protobuf/timestamp.proto";
import "google/protobuf/wrappers.proto";
option go_package = "google.golang.org/grpc/channelz/grpc_channelz_v1";
option java_multiple_files = true;
option java_package = "io.grpc.channelz.v1";
option java_outer_classname = "ChannelzProto";
// Channel is a logical grouping of channels, subchannels, and sockets.
message Channel {
// The identifier for this channel. This should bet set.
ChannelRef ref = 1;
// Data specific to this channel.
ChannelData data = 2;
// At most one of 'channel_ref+subchannel_ref' and 'socket' is set.
// There are no ordering guarantees on the order of channel refs.
// There may not be cycles in the ref graph.
// A channel ref may be present in more than one channel or subchannel.
repeated ChannelRef channel_ref = 3;
// At most one of 'channel_ref+subchannel_ref' and 'socket' is set.
// There are no ordering guarantees on the order of subchannel refs.
// There may not be cycles in the ref graph.
// A sub channel ref may be present in more than one channel or subchannel.
repeated SubchannelRef subchannel_ref = 4;
// There are no ordering guarantees on the order of sockets.
repeated SocketRef socket_ref = 5;
}
// Subchannel is a logical grouping of channels, subchannels, and sockets.
// A subchannel is load balanced over by it's ancestor
message Subchannel {
// The identifier for this channel.
SubchannelRef ref = 1;
// Data specific to this channel.
ChannelData data = 2;
// At most one of 'channel_ref+subchannel_ref' and 'socket' is set.
// There are no ordering guarantees on the order of channel refs.
// There may not be cycles in the ref graph.
// A channel ref may be present in more than one channel or subchannel.
repeated ChannelRef channel_ref = 3;
// At most one of 'channel_ref+subchannel_ref' and 'socket' is set.
// There are no ordering guarantees on the order of subchannel refs.
// There may not be cycles in the ref graph.
// A sub channel ref may be present in more than one channel or subchannel.
repeated SubchannelRef subchannel_ref = 4;
// There are no ordering guarantees on the order of sockets.
repeated SocketRef socket_ref = 5;
}
// These come from the specified states in this document:
// https://github.com/grpc/grpc/blob/master/doc/connectivity-semantics-and-api.md
message ChannelConnectivityState {
enum State {
UNKNOWN = 0;
IDLE = 1;
CONNECTING = 2;
READY = 3;
TRANSIENT_FAILURE = 4;
SHUTDOWN = 5;
}
State state = 1;
}
// Channel data is data related to a specific Channel or Subchannel.
message ChannelData {
// The connectivity state of the channel or subchannel. Implementations
// should always set this.
ChannelConnectivityState state = 1;
// The target this channel originally tried to connect to. May be absent
string target = 2;
// A trace of recent events on the channel. May be absent.
ChannelTrace trace = 3;
// The number of calls started on the channel
int64 calls_started = 4;
// The number of calls that have completed with an OK status
int64 calls_succeeded = 5;
// The number of calls that have completed with a non-OK status
int64 calls_failed = 6;
// The last time a call was started on the channel.
google.protobuf.Timestamp last_call_started_timestamp = 7;
}
// ChannelTrace represents the recent events that have occurred on the channel.
message ChannelTrace {
// see the proto in the gRFC for channel tracing:
// A3-channel-tracing.md
}
// ChannelRef is a reference to a Channel.
message ChannelRef {
// The globally unique id for this channel. Must be a positive number.
int64 channel_id = 1;
// An optional name associated with the channel.
string name = 2;
// Intentionally don't use field numbers from other refs.
reserved 3, 4, 5, 6, 7, 8;
}
// ChannelRef is a reference to a Subchannel.
message SubchannelRef {
// The globally unique id for this subchannel. Must be a positive number.
int64 subchannel_id = 7;
// An optional name associated with the subchannel.
string name = 8;
// Intentionally don't use field numbers from other refs.
reserved 1, 2, 3, 4, 5, 6;
}
// SocketRef is a reference to a Socket.
message SocketRef {
int64 socket_id = 3;
// An optional name associated with the socket.
string name = 4;
// Intentionally don't use field numbers from other refs.
reserved 1, 2, 5, 6, 7, 8;
}
// ServerRef is a reference to a Server.
message ServerRef {
// A globally unique identifier for this server. Must be a positive number.
int64 server_id = 5;
// An optional name associated with the server.
string name = 6;
// Intentionally don't use field numbers from other refs.
reserved 1, 2, 3, 4, 7, 8;
}
// Server represents a single server. There may be multiple servers in a single
// program.
message Server {
// The identifier for a Server. This should be set.
ServerRef ref = 1;
// The associated data of the Server.
ServerData data = 2;
// The sockets that the server is listening on. There are no ordering
// guarantees. This may be absent.
repeated SocketRef listen_socket = 3;
}
// ServerData is data for a specific Server.
message ServerData {
// A trace of recent events on the server. May be absent.
ChannelTrace trace = 1;
// The number of incoming calls started on the server
int64 calls_started = 2;
// The number of incoming calls that have completed with an OK status
int64 calls_succeeded = 3;
// The number of incoming calls that have a completed with a non-OK status
int64 calls_failed = 4;
// The last time a call was started on the server.
google.protobuf.Timestamp last_call_started_timestamp = 5;
}
// Information about an actual connection. Pronounced "sock-ay".
message Socket {
// The identifier for the Socket.
SocketRef ref = 1;
// Data specific to this Socket.
SocketData data = 2;
// The locally bound address.
Address local = 3;
// The remote bound address. May be absent.
Address remote = 4;
// Security details for this socket. May be absent if not available, or
// there is no security on the socket.
Security security = 5;
// Optional, represents the name of the remote endpoint, if different than
// the original target name.
string remote_name = 6;
}
// SocketData is data associated for a specific Socket. The fields present
// are specific to the implementation, so there may be minor differences in
// the semantics. (e.g. flow control windows)
message SocketData {
// The number of streams that have been started.
int64 streams_started = 1;
// The number of streams that have ended successfully:
// On client side, received frame with eos bit set;
// On server side, sent frame with eos bit set.
int64 streams_succeeded = 2;
// The number of streams that have ended unsuccessfully:
// On client side, ended without receiving frame with eos bit set;
// On server side, ended without sending frame with eos bit set.
int64 streams_failed = 3;
// The number of grpc messages successfully sent on this socket.
int64 messages_sent = 4;
// The number of grpc messages received on this socket.
int64 messages_received = 5;
// The number of keep alives sent. This is typically implemented with HTTP/2
// ping messages.
int64 keep_alives_sent = 6;
// The last time a stream was created by this endpoint. Usually unset for
// servers.
google.protobuf.Timestamp last_local_stream_created_timestamp = 7;
// The last time a stream was created by the remote endpoint. Usually unset
// for clients.
google.protobuf.Timestamp last_remote_stream_created_timestamp = 8;
// The last time a message was sent by this endpoint.
google.protobuf.Timestamp last_message_sent_timestamp = 9;
// The last time a message was received by this endpoint.
google.protobuf.Timestamp last_message_received_timestamp = 10;
// The amount of window, granted to the local endpoint by the remote endpoint.
// This may be slightly out of date due to network latency. This does NOT
// include stream level or TCP level flow control info.
google.protobuf.Int64Value local_flow_control_window = 11;
// The amount of window, granted to the remote endpoint by the local endpoint.
// This may be slightly out of date due to network latency. This does NOT
// include stream level or TCP level flow control info.
google.protobuf.Int64Value remote_flow_control_window = 12;
// Socket options set on this socket. May be absent.
repeated SocketOption option = 13;
}
// Address represents the address used to create the socket.
message Address {
message TcpIpAddress {
// Either the IPv4 or IPv6 address in bytes. Will be either 4 bytes or 16
// bytes in length.
bytes ip_address = 1;
// 0-64k, or -1 if not appropriate.
int32 port = 2;
}
// A Unix Domain Socket address.
message UdsAddress {
string filename = 1;
}
// An address type not included above.
message OtherAddress {
// The human readable version of the value. This value should be set.
string name = 1;
// The actual address message.
google.protobuf.Any value = 2;
}
oneof address {
TcpIpAddress tcpip_address = 1;
UdsAddress uds_address = 2;
OtherAddress other_address = 3;
}
}
// Security represents details about how secure the socket is.
message Security {
message Tls {
oneof cipher_suite {
// The cipher suite name in the RFC 4346 format:
// https://tools.ietf.org/html/rfc4346#appendix-C
string standard_name = 1;
// Some other way to describe the cipher suite if
// the RFC 4346 name is not available.
string other_name = 2;
}
// the certificate used by this endpoint.
bytes local_certificate = 3;
// the certificate used by the remote endpoint.
bytes remote_certificate = 4;
}
message OtherSecurity {
// The human readable version of the value.
string name = 1;
// The actual security details message.
google.protobuf.Any value = 2;
}
oneof model {
Tls tls = 1;
OtherSecurity other = 2;
}
}
// SocketOption represents socket options for a socket. Specifically, these
// are the options returned by getsockopt().
message SocketOption {
// The full name of the socket option. Typically this will be the upper case
// name, such as "SO_REUSEPORT".
string name = 1;
// The human readable value of this socket option. At least one of value or
// additional will be set.
string value = 2;
// Additional data associated with the socket option. At least one of value
// or additional will be set.
google.protobuf.Any additional = 3;
}
// For use with SocketOption's additional field. This is primarily used for
// SO_RCVTIMEO and SO_SNDTIMEO
message SocketOptionTimeout {
google.protobuf.Duration duration = 1;
}
// For use with SocketOption's additional field. This is primarily used for
// SO_LINGER.
message SocketOptionLinger {
// active maps to `struct linger.l_onoff`
bool active = 1;
// duration maps to `struct linger.l_linger`
google.protobuf.Duration duration = 2;
}
// For use with SocketOption's additional field. Tcp info for
// SOL_TCP and TCP_INFO.
message SocketOptionTcpInfo {
uint32 tcpi_state = 1;
uint32 tcpi_ca_state = 2;
uint32 tcpi_retransmits = 3;
uint32 tcpi_probes = 4;
uint32 tcpi_backoff = 5;
uint32 tcpi_options = 6;
uint32 tcpi_snd_wscale = 7;
uint32 tcpi_rcv_wscale = 8;
uint32 tcpi_rto = 9;
uint32 tcpi_ato = 10;
uint32 tcpi_snd_mss = 11;
uint32 tcpi_rcv_mss = 12;
uint32 tcpi_unacked = 13;
uint32 tcpi_sacked = 14;
uint32 tcpi_lost = 15;
uint32 tcpi_retrans = 16;
uint32 tcpi_fackets = 17;
uint32 tcpi_last_data_sent = 18;
uint32 tcpi_last_ack_sent = 19;
uint32 tcpi_last_data_recv = 20;
uint32 tcpi_last_ack_recv = 21;
uint32 tcpi_pmtu = 22;
uint32 tcpi_rcv_ssthresh = 23;
uint32 tcpi_rtt = 24;
uint32 tcpi_rttvar = 25;
uint32 tcpi_snd_ssthresh = 26;
uint32 tcpi_snd_cwnd = 27;
uint32 tcpi_advmss = 28;
uint32 tcpi_reordering = 29;
}
// Channelz is a service exposed by gRPC servers that provides detailed debug
// information.
service Channelz {
// Gets all root channels (i.e. channels the application has directly
// created). This does not include subchannels nor non-top level channels.
rpc GetTopChannels(GetTopChannelsRequest) returns (GetTopChannelsResponse);
// Gets all servers that exist in the process.
rpc GetServers(GetServersRequest) returns (GetServersResponse);
// Gets all server sockets that exist in the process.
rpc GetServerSockets(GetServerSocketsRequest) returns (GetServerSocketsResponse);
// Returns a single Channel, or else a NOT_FOUND code.
rpc GetChannel(GetChannelRequest) returns (GetChannelResponse);
// Returns a single Subchannel, or else a NOT_FOUND code.
rpc GetSubchannel(GetSubchannelRequest) returns (GetSubchannelResponse);
// Returns a single Socket or else a NOT_FOUND code.
rpc GetSocket(GetSocketRequest) returns (GetSocketResponse);
}
message GetTopChannelsRequest {
// start_channel_id indicates that only channels at or above this id should be
// included in the results.
int64 start_channel_id = 1;
}
message GetTopChannelsResponse {
// list of channels that the connection detail service knows about. Sorted in
// ascending channel_id order.
repeated Channel channel = 1;
// If set, indicates that the list of channels is the final list. Requesting
// more channels can only return more if they are created after this RPC
// completes.
bool end = 2;
}
message GetServersRequest {
// start_server_id indicates that only servers at or above this id should be
// included in the results.
int64 start_server_id = 1;
}
message GetServersResponse {
// list of servers that the connection detail service knows about. Sorted in
// ascending server_id order.
repeated Server server = 1;
// If set, indicates that the list of servers is the final list. Requesting
// more servers will only return more if they are created after this RPC
// completes.
bool end = 2;
}
message GetServerSocketsRequest {
int64 server_id = 1;
// start_socket_id indicates that only sockets at or above this id should be
// included in the results.
int64 start_socket_id = 2;
}
message GetServerSocketsResponse {
// list of socket refs that the connection detail service knows about. Sorted in
// ascending socket_id order.
repeated SocketRef socket_ref = 1;
// If set, indicates that the list of sockets is the final list. Requesting
// more sockets will only return more if they are created after this RPC
// completes.
bool end = 2;
}
message GetChannelRequest {
// channel_id is the identifier of the specific channel to get.
int64 channel_id = 1;
}
message GetChannelResponse {
// The Channel that corresponds to the requested channel_id. This field
// should be set.
Channel channel = 1;
}
message GetSubchannelRequest {
// subchannel_id is the identifier of the specific subchannel to get.
int64 subchannel_id = 1;
}
message GetSubchannelResponse {
// The Subchannel that corresponds to the requested subchannel_id. This
// field should be set.
Subchannel subchannel = 1;
}
message GetSocketRequest {
// socket_id is the identifier of the specific socket to get.
int64 socket_id = 1;
}
message GetSocketResponse {
// The Socket that corresponds to the requested socket_id. This field
// should be set.
Socket socket = 1;
}

265
vendor/google.golang.org/grpc/channelz/service/service.go generated vendored Normal file
View File

@@ -0,0 +1,265 @@
/*
*
* Copyright 2018 gRPC authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
//go:generate protoc -I ../grpc_channelz_v1 --go_out=plugins=grpc,paths=source_relative:../grpc_channelz_v1 ../grpc_channelz_v1/channelz.proto
// Package service provides an implementation for channelz service server.
package service
import (
"net"
"github.com/golang/protobuf/ptypes"
wrpb "github.com/golang/protobuf/ptypes/wrappers"
"golang.org/x/net/context"
"google.golang.org/grpc"
"google.golang.org/grpc/channelz"
pb "google.golang.org/grpc/channelz/grpc_channelz_v1"
"google.golang.org/grpc/connectivity"
)
// RegisterChannelzServiceToServer registers the channelz service to the given server.
func RegisterChannelzServiceToServer(s *grpc.Server) {
pb.RegisterChannelzServer(s, &serverImpl{})
}
func newCZServer() pb.ChannelzServer {
return &serverImpl{}
}
type serverImpl struct{}
func connectivityStateToProto(s connectivity.State) *pb.ChannelConnectivityState {
switch s {
case connectivity.Idle:
return &pb.ChannelConnectivityState{State: pb.ChannelConnectivityState_IDLE}
case connectivity.Connecting:
return &pb.ChannelConnectivityState{State: pb.ChannelConnectivityState_CONNECTING}
case connectivity.Ready:
return &pb.ChannelConnectivityState{State: pb.ChannelConnectivityState_READY}
case connectivity.TransientFailure:
return &pb.ChannelConnectivityState{State: pb.ChannelConnectivityState_TRANSIENT_FAILURE}
case connectivity.Shutdown:
return &pb.ChannelConnectivityState{State: pb.ChannelConnectivityState_SHUTDOWN}
default:
return &pb.ChannelConnectivityState{State: pb.ChannelConnectivityState_UNKNOWN}
}
}
func channelMetricToProto(cm *channelz.ChannelMetric) *pb.Channel {
c := &pb.Channel{}
c.Ref = &pb.ChannelRef{ChannelId: cm.ID, Name: cm.RefName}
c.Data = &pb.ChannelData{
State: connectivityStateToProto(cm.ChannelData.State),
Target: cm.ChannelData.Target,
CallsStarted: cm.ChannelData.CallsStarted,
CallsSucceeded: cm.ChannelData.CallsSucceeded,
CallsFailed: cm.ChannelData.CallsFailed,
}
if ts, err := ptypes.TimestampProto(cm.ChannelData.LastCallStartedTimestamp); err == nil {
c.Data.LastCallStartedTimestamp = ts
}
nestedChans := make([]*pb.ChannelRef, 0, len(cm.NestedChans))
for id, ref := range cm.NestedChans {
nestedChans = append(nestedChans, &pb.ChannelRef{ChannelId: id, Name: ref})
}
c.ChannelRef = nestedChans
subChans := make([]*pb.SubchannelRef, 0, len(cm.SubChans))
for id, ref := range cm.SubChans {
subChans = append(subChans, &pb.SubchannelRef{SubchannelId: id, Name: ref})
}
c.SubchannelRef = subChans
sockets := make([]*pb.SocketRef, 0, len(cm.Sockets))
for id, ref := range cm.Sockets {
sockets = append(sockets, &pb.SocketRef{SocketId: id, Name: ref})
}
c.SocketRef = sockets
return c
}
func subChannelMetricToProto(cm *channelz.SubChannelMetric) *pb.Subchannel {
sc := &pb.Subchannel{}
sc.Ref = &pb.SubchannelRef{SubchannelId: cm.ID, Name: cm.RefName}
sc.Data = &pb.ChannelData{
State: connectivityStateToProto(cm.ChannelData.State),
Target: cm.ChannelData.Target,
CallsStarted: cm.ChannelData.CallsStarted,
CallsSucceeded: cm.ChannelData.CallsSucceeded,
CallsFailed: cm.ChannelData.CallsFailed,
}
if ts, err := ptypes.TimestampProto(cm.ChannelData.LastCallStartedTimestamp); err == nil {
sc.Data.LastCallStartedTimestamp = ts
}
nestedChans := make([]*pb.ChannelRef, 0, len(cm.NestedChans))
for id, ref := range cm.NestedChans {
nestedChans = append(nestedChans, &pb.ChannelRef{ChannelId: id, Name: ref})
}
sc.ChannelRef = nestedChans
subChans := make([]*pb.SubchannelRef, 0, len(cm.SubChans))
for id, ref := range cm.SubChans {
subChans = append(subChans, &pb.SubchannelRef{SubchannelId: id, Name: ref})
}
sc.SubchannelRef = subChans
sockets := make([]*pb.SocketRef, 0, len(cm.Sockets))
for id, ref := range cm.Sockets {
sockets = append(sockets, &pb.SocketRef{SocketId: id, Name: ref})
}
sc.SocketRef = sockets
return sc
}
func addrToProto(a net.Addr) *pb.Address {
switch a.Network() {
case "udp":
// TODO: Address_OtherAddress{}. Need proto def for Value.
case "ip":
// Note zone info is discarded through the conversion.
return &pb.Address{Address: &pb.Address_TcpipAddress{TcpipAddress: &pb.Address_TcpIpAddress{IpAddress: a.(*net.IPAddr).IP}}}
case "ip+net":
// Note mask info is discarded through the conversion.
return &pb.Address{Address: &pb.Address_TcpipAddress{TcpipAddress: &pb.Address_TcpIpAddress{IpAddress: a.(*net.IPNet).IP}}}
case "tcp":
// Note zone info is discarded through the conversion.
return &pb.Address{Address: &pb.Address_TcpipAddress{TcpipAddress: &pb.Address_TcpIpAddress{IpAddress: a.(*net.TCPAddr).IP, Port: int32(a.(*net.TCPAddr).Port)}}}
case "unix", "unixgram", "unixpacket":
return &pb.Address{Address: &pb.Address_UdsAddress_{UdsAddress: &pb.Address_UdsAddress{Filename: a.String()}}}
default:
}
return &pb.Address{}
}
func socketMetricToProto(sm *channelz.SocketMetric) *pb.Socket {
s := &pb.Socket{}
s.Ref = &pb.SocketRef{SocketId: sm.ID, Name: sm.RefName}
s.Data = &pb.SocketData{
StreamsStarted: sm.SocketData.StreamsStarted,
StreamsSucceeded: sm.SocketData.StreamsSucceeded,
StreamsFailed: sm.SocketData.StreamsFailed,
MessagesSent: sm.SocketData.MessagesSent,
MessagesReceived: sm.SocketData.MessagesReceived,
KeepAlivesSent: sm.SocketData.KeepAlivesSent,
}
if ts, err := ptypes.TimestampProto(sm.SocketData.LastLocalStreamCreatedTimestamp); err == nil {
s.Data.LastLocalStreamCreatedTimestamp = ts
}
if ts, err := ptypes.TimestampProto(sm.SocketData.LastRemoteStreamCreatedTimestamp); err == nil {
s.Data.LastRemoteStreamCreatedTimestamp = ts
}
if ts, err := ptypes.TimestampProto(sm.SocketData.LastMessageSentTimestamp); err == nil {
s.Data.LastMessageSentTimestamp = ts
}
if ts, err := ptypes.TimestampProto(sm.SocketData.LastMessageReceivedTimestamp); err == nil {
s.Data.LastMessageReceivedTimestamp = ts
}
s.Data.LocalFlowControlWindow = &wrpb.Int64Value{Value: sm.SocketData.LocalFlowControlWindow}
s.Data.RemoteFlowControlWindow = &wrpb.Int64Value{Value: sm.SocketData.RemoteFlowControlWindow}
if sm.SocketData.LocalAddr != nil {
s.Local = addrToProto(sm.SocketData.LocalAddr)
}
if sm.SocketData.RemoteAddr != nil {
s.Remote = addrToProto(sm.SocketData.RemoteAddr)
}
s.RemoteName = sm.SocketData.RemoteName
return s
}
func (s *serverImpl) GetTopChannels(ctx context.Context, req *pb.GetTopChannelsRequest) (*pb.GetTopChannelsResponse, error) {
metrics, end := channelz.GetTopChannels(req.GetStartChannelId())
resp := &pb.GetTopChannelsResponse{}
for _, m := range metrics {
resp.Channel = append(resp.Channel, channelMetricToProto(m))
}
resp.End = end
return resp, nil
}
func serverMetricToProto(sm *channelz.ServerMetric) *pb.Server {
s := &pb.Server{}
s.Ref = &pb.ServerRef{ServerId: sm.ID, Name: sm.RefName}
s.Data = &pb.ServerData{
CallsStarted: sm.ServerData.CallsStarted,
CallsSucceeded: sm.ServerData.CallsSucceeded,
CallsFailed: sm.ServerData.CallsFailed,
}
if ts, err := ptypes.TimestampProto(sm.ServerData.LastCallStartedTimestamp); err == nil {
s.Data.LastCallStartedTimestamp = ts
}
sockets := make([]*pb.SocketRef, 0, len(sm.ListenSockets))
for id, ref := range sm.ListenSockets {
sockets = append(sockets, &pb.SocketRef{SocketId: id, Name: ref})
}
s.ListenSocket = sockets
return s
}
func (s *serverImpl) GetServers(ctx context.Context, req *pb.GetServersRequest) (*pb.GetServersResponse, error) {
metrics, end := channelz.GetServers(req.GetStartServerId())
resp := &pb.GetServersResponse{}
for _, m := range metrics {
resp.Server = append(resp.Server, serverMetricToProto(m))
}
resp.End = end
return resp, nil
}
func (s *serverImpl) GetServerSockets(ctx context.Context, req *pb.GetServerSocketsRequest) (*pb.GetServerSocketsResponse, error) {
metrics, end := channelz.GetServerSockets(req.GetServerId(), req.GetStartSocketId())
resp := &pb.GetServerSocketsResponse{}
for _, m := range metrics {
resp.SocketRef = append(resp.SocketRef, &pb.SocketRef{SocketId: m.ID, Name: m.RefName})
}
resp.End = end
return resp, nil
}
func (s *serverImpl) GetChannel(ctx context.Context, req *pb.GetChannelRequest) (*pb.GetChannelResponse, error) {
var metric *channelz.ChannelMetric
if metric = channelz.GetChannel(req.GetChannelId()); metric == nil {
return &pb.GetChannelResponse{}, nil
}
resp := &pb.GetChannelResponse{Channel: channelMetricToProto(metric)}
return resp, nil
}
func (s *serverImpl) GetSubchannel(ctx context.Context, req *pb.GetSubchannelRequest) (*pb.GetSubchannelResponse, error) {
var metric *channelz.SubChannelMetric
if metric = channelz.GetSubChannel(req.GetSubchannelId()); metric == nil {
return &pb.GetSubchannelResponse{}, nil
}
resp := &pb.GetSubchannelResponse{Subchannel: subChannelMetricToProto(metric)}
return resp, nil
}
func (s *serverImpl) GetSocket(ctx context.Context, req *pb.GetSocketRequest) (*pb.GetSocketResponse, error) {
var metric *channelz.SocketMetric
if metric = channelz.GetSocket(req.GetSocketId()); metric == nil {
return &pb.GetSocketResponse{}, nil
}
resp := &pb.GetSocketResponse{Socket: socketMetricToProto(metric)}
return resp, nil
}

477
vendor/google.golang.org/grpc/channelz/service/service_test.go generated vendored Normal file
View File

@@ -0,0 +1,477 @@
/*
*
* Copyright 2018 gRPC authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package service
import (
"net"
"reflect"
"strconv"
"testing"
"time"
"github.com/golang/protobuf/ptypes"
"golang.org/x/net/context"
"google.golang.org/grpc/channelz"
pb "google.golang.org/grpc/channelz/grpc_channelz_v1"
"google.golang.org/grpc/connectivity"
)
func init() {
channelz.TurnOn()
}
// emptyTime is used for detecting unset value of time.Time type.
// For go1.7 and earlier, ptypes.Timestamp will fill in the loc field of time.Time
// with &utcLoc. However zero value of a time.Time type value loc field is nil.
// This behavior will make reflect.DeepEqual fail upon unset time.Time field,
// and cause false positive fatal error.
var emptyTime time.Time
type dummyChannel struct {
state connectivity.State
target string
callsStarted int64
callsSucceeded int64
callsFailed int64
lastCallStartedTimestamp time.Time
}
func (d *dummyChannel) ChannelzMetric() *channelz.ChannelInternalMetric {
return &channelz.ChannelInternalMetric{
State: d.state,
Target: d.target,
CallsStarted: d.callsStarted,
CallsSucceeded: d.callsSucceeded,
CallsFailed: d.callsFailed,
LastCallStartedTimestamp: d.lastCallStartedTimestamp,
}
}
type dummyServer struct {
callsStarted int64
callsSucceeded int64
callsFailed int64
lastCallStartedTimestamp time.Time
}
func (d *dummyServer) ChannelzMetric() *channelz.ServerInternalMetric {
return &channelz.ServerInternalMetric{
CallsStarted: d.callsStarted,
CallsSucceeded: d.callsSucceeded,
CallsFailed: d.callsFailed,
LastCallStartedTimestamp: d.lastCallStartedTimestamp,
}
}
type dummySocket struct {
streamsStarted int64
streamsSucceeded int64
streamsFailed int64
messagesSent int64
messagesReceived int64
keepAlivesSent int64
lastLocalStreamCreatedTimestamp time.Time
lastRemoteStreamCreatedTimestamp time.Time
lastMessageSentTimestamp time.Time
lastMessageReceivedTimestamp time.Time
localFlowControlWindow int64
remoteFlowControlWindow int64
//socket options
localAddr net.Addr
remoteAddr net.Addr
// Security
remoteName string
}
func (d *dummySocket) ChannelzMetric() *channelz.SocketInternalMetric {
return &channelz.SocketInternalMetric{
StreamsStarted: d.streamsStarted,
StreamsSucceeded: d.streamsSucceeded,
StreamsFailed: d.streamsFailed,
MessagesSent: d.messagesSent,
MessagesReceived: d.messagesReceived,
KeepAlivesSent: d.keepAlivesSent,
LastLocalStreamCreatedTimestamp: d.lastLocalStreamCreatedTimestamp,
LastRemoteStreamCreatedTimestamp: d.lastRemoteStreamCreatedTimestamp,
LastMessageSentTimestamp: d.lastMessageSentTimestamp,
LastMessageReceivedTimestamp: d.lastMessageReceivedTimestamp,
LocalFlowControlWindow: d.localFlowControlWindow,
RemoteFlowControlWindow: d.remoteFlowControlWindow,
//socket options
LocalAddr: d.localAddr,
RemoteAddr: d.remoteAddr,
// Security
RemoteName: d.remoteName,
}
}
func channelProtoToStruct(c *pb.Channel) *dummyChannel {
dc := &dummyChannel{}
pdata := c.GetData()
switch pdata.GetState().GetState() {
case pb.ChannelConnectivityState_UNKNOWN:
// TODO: what should we set here?
case pb.ChannelConnectivityState_IDLE:
dc.state = connectivity.Idle
case pb.ChannelConnectivityState_CONNECTING:
dc.state = connectivity.Connecting
case pb.ChannelConnectivityState_READY:
dc.state = connectivity.Ready
case pb.ChannelConnectivityState_TRANSIENT_FAILURE:
dc.state = connectivity.TransientFailure
case pb.ChannelConnectivityState_SHUTDOWN:
dc.state = connectivity.Shutdown
}
dc.target = pdata.GetTarget()
dc.callsStarted = pdata.CallsStarted
dc.callsSucceeded = pdata.CallsSucceeded
dc.callsFailed = pdata.CallsFailed
if t, err := ptypes.Timestamp(pdata.GetLastCallStartedTimestamp()); err == nil {
if !t.Equal(emptyTime) {
dc.lastCallStartedTimestamp = t
}
}
return dc
}
func serverProtoToStruct(s *pb.Server) *dummyServer {
ds := &dummyServer{}
pdata := s.GetData()
ds.callsStarted = pdata.CallsStarted
ds.callsSucceeded = pdata.CallsSucceeded
ds.callsFailed = pdata.CallsFailed
if t, err := ptypes.Timestamp(pdata.GetLastCallStartedTimestamp()); err == nil {
if !t.Equal(emptyTime) {
ds.lastCallStartedTimestamp = t
}
}
return ds
}
func protoToAddr(a *pb.Address) net.Addr {
switch v := a.Address.(type) {
case *pb.Address_TcpipAddress:
if port := v.TcpipAddress.GetPort(); port != 0 {
return &net.TCPAddr{IP: v.TcpipAddress.GetIpAddress(), Port: int(port)}
}
return &net.IPAddr{IP: v.TcpipAddress.GetIpAddress()}
case *pb.Address_UdsAddress_:
return &net.UnixAddr{Name: v.UdsAddress.GetFilename(), Net: "unix"}
case *pb.Address_OtherAddress_:
// TODO:
}
return nil
}
func socketProtoToStruct(s *pb.Socket) *dummySocket {
ds := &dummySocket{}
pdata := s.GetData()
ds.streamsStarted = pdata.GetStreamsStarted()
ds.streamsSucceeded = pdata.GetStreamsSucceeded()
ds.streamsFailed = pdata.GetStreamsFailed()
ds.messagesSent = pdata.GetMessagesSent()
ds.messagesReceived = pdata.GetMessagesReceived()
ds.keepAlivesSent = pdata.GetKeepAlivesSent()
if t, err := ptypes.Timestamp(pdata.GetLastLocalStreamCreatedTimestamp()); err == nil {
if !t.Equal(emptyTime) {
ds.lastLocalStreamCreatedTimestamp = t
}
}
if t, err := ptypes.Timestamp(pdata.GetLastRemoteStreamCreatedTimestamp()); err == nil {
if !t.Equal(emptyTime) {
ds.lastRemoteStreamCreatedTimestamp = t
}
}
if t, err := ptypes.Timestamp(pdata.GetLastMessageSentTimestamp()); err == nil {
if !t.Equal(emptyTime) {
ds.lastMessageSentTimestamp = t
}
}
if t, err := ptypes.Timestamp(pdata.GetLastMessageReceivedTimestamp()); err == nil {
if !t.Equal(emptyTime) {
ds.lastMessageReceivedTimestamp = t
}
}
if v := pdata.GetLocalFlowControlWindow(); v != nil {
ds.localFlowControlWindow = v.Value
}
if v := pdata.GetRemoteFlowControlWindow(); v != nil {
ds.remoteFlowControlWindow = v.Value
}
if local := s.GetLocal(); local != nil {
ds.localAddr = protoToAddr(local)
}
if remote := s.GetRemote(); remote != nil {
ds.remoteAddr = protoToAddr(remote)
}
ds.remoteName = s.GetRemoteName()
return ds
}
func convertSocketRefSliceToMap(sktRefs []*pb.SocketRef) map[int64]string {
m := make(map[int64]string)
for _, sr := range sktRefs {
m[sr.SocketId] = sr.Name
}
return m
}
func TestGetTopChannels(t *testing.T) {
tcs := []*dummyChannel{
{
state: connectivity.Connecting,
target: "test.channelz:1234",
callsStarted: 6,
callsSucceeded: 2,
callsFailed: 3,
lastCallStartedTimestamp: time.Now().UTC(),
},
{
state: connectivity.Connecting,
target: "test.channelz:1234",
callsStarted: 1,
callsSucceeded: 2,
callsFailed: 3,
lastCallStartedTimestamp: time.Now().UTC(),
},
{
state: connectivity.Shutdown,
target: "test.channelz:8888",
callsStarted: 0,
callsSucceeded: 0,
callsFailed: 0,
},
{},
}
channelz.NewChannelzStorage()
for _, c := range tcs {
channelz.RegisterChannel(c, 0, "")
}
s := newCZServer()
resp, _ := s.GetTopChannels(context.Background(), &pb.GetTopChannelsRequest{StartChannelId: 0})
if !resp.GetEnd() {
t.Fatalf("resp.GetEnd() want true, got %v", resp.GetEnd())
}
for i, c := range resp.GetChannel() {
if !reflect.DeepEqual(channelProtoToStruct(c), tcs[i]) {
t.Fatalf("dummyChannel: %d, want: %#v, got: %#v", i, tcs[i], channelProtoToStruct(c))
}
}
for i := 0; i < 50; i++ {
channelz.RegisterChannel(tcs[0], 0, "")
}
resp, _ = s.GetTopChannels(context.Background(), &pb.GetTopChannelsRequest{StartChannelId: 0})
if resp.GetEnd() {
t.Fatalf("resp.GetEnd() want false, got %v", resp.GetEnd())
}
}
func TestGetServers(t *testing.T) {
ss := []*dummyServer{
{
callsStarted: 6,
callsSucceeded: 2,
callsFailed: 3,
lastCallStartedTimestamp: time.Now().UTC(),
},
{
callsStarted: 1,
callsSucceeded: 2,
callsFailed: 3,
lastCallStartedTimestamp: time.Now().UTC(),
},
{
callsStarted: 1,
callsSucceeded: 0,
callsFailed: 0,
lastCallStartedTimestamp: time.Now().UTC(),
},
}
channelz.NewChannelzStorage()
for _, s := range ss {
channelz.RegisterServer(s, "")
}
svr := newCZServer()
resp, _ := svr.GetServers(context.Background(), &pb.GetServersRequest{StartServerId: 0})
if !resp.GetEnd() {
t.Fatalf("resp.GetEnd() want true, got %v", resp.GetEnd())
}
for i, s := range resp.GetServer() {
if !reflect.DeepEqual(serverProtoToStruct(s), ss[i]) {
t.Fatalf("dummyServer: %d, want: %#v, got: %#v", i, ss[i], serverProtoToStruct(s))
}
}
for i := 0; i < 50; i++ {
channelz.RegisterServer(ss[0], "")
}
resp, _ = svr.GetServers(context.Background(), &pb.GetServersRequest{StartServerId: 0})
if resp.GetEnd() {
t.Fatalf("resp.GetEnd() want false, got %v", resp.GetEnd())
}
}
func TestGetServerSockets(t *testing.T) {
channelz.NewChannelzStorage()
svrID := channelz.RegisterServer(&dummyServer{}, "")
refNames := []string{"listen socket 1", "normal socket 1", "normal socket 2"}
ids := make([]int64, 3)
ids[0] = channelz.RegisterListenSocket(&dummySocket{}, svrID, refNames[0])
ids[1] = channelz.RegisterNormalSocket(&dummySocket{}, svrID, refNames[1])
ids[2] = channelz.RegisterNormalSocket(&dummySocket{}, svrID, refNames[2])
svr := newCZServer()
resp, _ := svr.GetServerSockets(context.Background(), &pb.GetServerSocketsRequest{ServerId: svrID, StartSocketId: 0})
if !resp.GetEnd() {
t.Fatalf("resp.GetEnd() want: true, got: %v", resp.GetEnd())
}
// GetServerSockets only return normal sockets.
want := map[int64]string{
ids[1]: refNames[1],
ids[2]: refNames[2],
}
if !reflect.DeepEqual(convertSocketRefSliceToMap(resp.GetSocketRef()), want) {
t.Fatalf("GetServerSockets want: %#v, got: %#v", want, resp.GetSocketRef())
}
for i := 0; i < 50; i++ {
channelz.RegisterNormalSocket(&dummySocket{}, svrID, "")
}
resp, _ = svr.GetServerSockets(context.Background(), &pb.GetServerSocketsRequest{ServerId: svrID, StartSocketId: 0})
if resp.GetEnd() {
t.Fatalf("resp.GetEnd() want false, got %v", resp.GetEnd())
}
}
func TestGetChannel(t *testing.T) {
channelz.NewChannelzStorage()
refNames := []string{"top channel 1", "nested channel 1", "nested channel 2", "nested channel 3"}
ids := make([]int64, 4)
ids[0] = channelz.RegisterChannel(&dummyChannel{}, 0, refNames[0])
ids[1] = channelz.RegisterChannel(&dummyChannel{}, ids[0], refNames[1])
ids[2] = channelz.RegisterSubChannel(&dummyChannel{}, ids[0], refNames[2])
ids[3] = channelz.RegisterChannel(&dummyChannel{}, ids[1], refNames[3])
svr := newCZServer()
resp, _ := svr.GetChannel(context.Background(), &pb.GetChannelRequest{ChannelId: ids[0]})
metrics := resp.GetChannel()
subChans := metrics.GetSubchannelRef()
if len(subChans) != 1 || subChans[0].GetName() != refNames[2] || subChans[0].GetSubchannelId() != ids[2] {
t.Fatalf("GetSubChannelRef() want %#v, got %#v", []*pb.SubchannelRef{{SubchannelId: ids[2], Name: refNames[2]}}, subChans)
}
nestedChans := metrics.GetChannelRef()
if len(nestedChans) != 1 || nestedChans[0].GetName() != refNames[1] || nestedChans[0].GetChannelId() != ids[1] {
t.Fatalf("GetChannelRef() want %#v, got %#v", []*pb.ChannelRef{{ChannelId: ids[1], Name: refNames[1]}}, nestedChans)
}
resp, _ = svr.GetChannel(context.Background(), &pb.GetChannelRequest{ChannelId: ids[1]})
metrics = resp.GetChannel()
nestedChans = metrics.GetChannelRef()
if len(nestedChans) != 1 || nestedChans[0].GetName() != refNames[3] || nestedChans[0].GetChannelId() != ids[3] {
t.Fatalf("GetChannelRef() want %#v, got %#v", []*pb.ChannelRef{{ChannelId: ids[3], Name: refNames[3]}}, nestedChans)
}
}
func TestGetSubChannel(t *testing.T) {
channelz.NewChannelzStorage()
refNames := []string{"top channel 1", "sub channel 1", "socket 1", "socket 2"}
ids := make([]int64, 4)
ids[0] = channelz.RegisterChannel(&dummyChannel{}, 0, refNames[0])
ids[1] = channelz.RegisterSubChannel(&dummyChannel{}, ids[0], refNames[1])
ids[2] = channelz.RegisterNormalSocket(&dummySocket{}, ids[1], refNames[2])
ids[3] = channelz.RegisterNormalSocket(&dummySocket{}, ids[1], refNames[3])
svr := newCZServer()
resp, _ := svr.GetSubchannel(context.Background(), &pb.GetSubchannelRequest{SubchannelId: ids[1]})
metrics := resp.GetSubchannel()
want := map[int64]string{
ids[2]: refNames[2],
ids[3]: refNames[3],
}
if !reflect.DeepEqual(convertSocketRefSliceToMap(metrics.GetSocketRef()), want) {
t.Fatalf("GetSocketRef() want %#v: got: %#v", want, metrics.GetSocketRef())
}
}
func TestGetSocket(t *testing.T) {
channelz.NewChannelzStorage()
ss := []*dummySocket{
{
streamsStarted: 10,
streamsSucceeded: 2,
streamsFailed: 3,
messagesSent: 20,
messagesReceived: 10,
keepAlivesSent: 2,
lastLocalStreamCreatedTimestamp: time.Now().UTC(),
lastRemoteStreamCreatedTimestamp: time.Now().UTC(),
lastMessageSentTimestamp: time.Now().UTC(),
lastMessageReceivedTimestamp: time.Now().UTC(),
localFlowControlWindow: 65536,
remoteFlowControlWindow: 1024,
localAddr: &net.TCPAddr{IP: net.ParseIP("1.0.0.1"), Port: 10001},
remoteAddr: &net.TCPAddr{IP: net.ParseIP("12.0.0.1"), Port: 10002},
remoteName: "remote.remote",
},
{
streamsStarted: 10,
streamsSucceeded: 2,
streamsFailed: 3,
messagesSent: 20,
messagesReceived: 10,
keepAlivesSent: 2,
lastRemoteStreamCreatedTimestamp: time.Now().UTC(),
lastMessageSentTimestamp: time.Now().UTC(),
lastMessageReceivedTimestamp: time.Now().UTC(),
localFlowControlWindow: 65536,
remoteFlowControlWindow: 1024,
localAddr: &net.UnixAddr{Name: "file.path", Net: "unix"},
remoteAddr: &net.UnixAddr{Name: "another.path", Net: "unix"},
remoteName: "remote.remote",
},
{
streamsStarted: 5,
streamsSucceeded: 2,
streamsFailed: 3,
messagesSent: 20,
messagesReceived: 10,
keepAlivesSent: 2,
lastLocalStreamCreatedTimestamp: time.Now().UTC(),
lastMessageSentTimestamp: time.Now().UTC(),
lastMessageReceivedTimestamp: time.Now().UTC(),
localFlowControlWindow: 65536,
remoteFlowControlWindow: 10240,
localAddr: &net.IPAddr{IP: net.ParseIP("1.0.0.1")},
remoteAddr: &net.IPAddr{IP: net.ParseIP("9.0.0.1")},
remoteName: "",
},
{
localAddr: &net.TCPAddr{IP: net.ParseIP("127.0.0.1"), Port: 10001},
},
}
svr := newCZServer()
ids := make([]int64, len(ss))
svrID := channelz.RegisterServer(&dummyServer{}, "")
for i, s := range ss {
ids[i] = channelz.RegisterNormalSocket(s, svrID, strconv.Itoa(i))
}
for i, s := range ss {
resp, _ := svr.GetSocket(context.Background(), &pb.GetSocketRequest{SocketId: ids[i]})
metrics := resp.GetSocket()
if !reflect.DeepEqual(metrics.GetRef(), &pb.SocketRef{SocketId: ids[i], Name: strconv.Itoa(i)}) || !reflect.DeepEqual(socketProtoToStruct(metrics), s) {
t.Fatalf("resp.GetSocket() want: metrics.GetRef() = %#v and %#v, got: metrics.GetRef() = %#v and %#v", &pb.SocketRef{SocketId: ids[i], Name: strconv.Itoa(i)}, s, metrics.GetRef(), socketProtoToStruct(metrics))
}
}
}

418
vendor/google.golang.org/grpc/channelz/types.go generated vendored Normal file
View File

@@ -0,0 +1,418 @@
/*
*
* Copyright 2018 gRPC authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package channelz
import (
"net"
"time"
"google.golang.org/grpc/connectivity"
"google.golang.org/grpc/grpclog"
)
// entry represents a node in the channelz database.
type entry interface {
// addChild adds a child e, whose channelz id is id to child list
addChild(id int64, e entry)
// deleteChild deletes a child with channelz id to be id from child list
deleteChild(id int64)
// triggerDelete tries to delete self from channelz database. However, if child
// list is not empty, then deletion from the database is on hold until the last
// child is deleted from database.
triggerDelete()
// deleteSelfIfReady check whether triggerDelete() has been called before, and whether child
// list is now empty. If both conditions are met, then delete self from database.
deleteSelfIfReady()
}
// dummyEntry is a fake entry to handle entry not found case.
type dummyEntry struct {
idNotFound int64
}
func (d *dummyEntry) addChild(id int64, e entry) {
// Note: It is possible for a normal program to reach here under race condition.
// For example, there could be a race between ClientConn.Close() info being propagated
// to addrConn and http2Client. ClientConn.Close() cancel the context and result
// in http2Client to error. The error info is then caught by transport monitor
// and before addrConn.tearDown() is called in side ClientConn.Close(). Therefore,
// the addrConn will create a new transport. And when registering the new transport in
// channelz, its parent addrConn could have already been torn down and deleted
// from channelz tracking, and thus reach the code here.
grpclog.Infof("attempt to add child of type %T with id %d to a parent (id=%d) that doesn't currently exist", e, id, d.idNotFound)
}
func (d *dummyEntry) deleteChild(id int64) {
// It is possible for a normal program to reach here under race condition.
// Refer to the example described in addChild().
grpclog.Infof("attempt to delete child with id %d from a parent (id=%d) that doesn't currently exist", id, d.idNotFound)
}
func (d *dummyEntry) triggerDelete() {
grpclog.Warningf("attempt to delete an entry (id=%d) that doesn't currently exist", d.idNotFound)
}
func (*dummyEntry) deleteSelfIfReady() {
// code should not reach here. deleteSelfIfReady is always called on an existing entry.
}
// ChannelMetric defines the info channelz provides for a specific Channel, which
// includes ChannelInternalMetric and channelz-specific data, such as channelz id,
// child list, etc.
type ChannelMetric struct {
// ID is the channelz id of this channel.
ID int64
// RefName is the human readable reference string of this channel.
RefName string
// ChannelData contains channel internal metric reported by the channel through
// ChannelzMetric().
ChannelData *ChannelInternalMetric
// NestedChans tracks the nested channel type children of this channel in the format of
// a map from nested channel channelz id to corresponding reference string.
NestedChans map[int64]string
// SubChans tracks the subchannel type children of this channel in the format of a
// map from subchannel channelz id to corresponding reference string.
SubChans map[int64]string
// Sockets tracks the socket type children of this channel in the format of a map
// from socket channelz id to corresponding reference string.
// Note current grpc implementation doesn't allow channel having sockets directly,
// therefore, this is field is unused.
Sockets map[int64]string
}
// SubChannelMetric defines the info channelz provides for a specific SubChannel,
// which includes ChannelInternalMetric and channelz-specific data, such as
// channelz id, child list, etc.
type SubChannelMetric struct {
// ID is the channelz id of this subchannel.
ID int64
// RefName is the human readable reference string of this subchannel.
RefName string
// ChannelData contains subchannel internal metric reported by the subchannel
// through ChannelzMetric().
ChannelData *ChannelInternalMetric
// NestedChans tracks the nested channel type children of this subchannel in the format of
// a map from nested channel channelz id to corresponding reference string.
// Note current grpc implementation doesn't allow subchannel to have nested channels
// as children, therefore, this field is unused.
NestedChans map[int64]string
// SubChans tracks the subchannel type children of this subchannel in the format of a
// map from subchannel channelz id to corresponding reference string.
// Note current grpc implementation doesn't allow subchannel to have subchannels
// as children, therefore, this field is unused.
SubChans map[int64]string
// Sockets tracks the socket type children of this subchannel in the format of a map
// from socket channelz id to corresponding reference string.
Sockets map[int64]string
}
// ChannelInternalMetric defines the struct that the implementor of Channel interface
// should return from ChannelzMetric().
type ChannelInternalMetric struct {
// current connectivity state of the channel.
State connectivity.State
// The target this channel originally tried to connect to. May be absent
Target string
// The number of calls started on the channel.
CallsStarted int64
// The number of calls that have completed with an OK status.
CallsSucceeded int64
// The number of calls that have a completed with a non-OK status.
CallsFailed int64
// The last time a call was started on the channel.
LastCallStartedTimestamp time.Time
//TODO: trace
}
// Channel is the interface that should be satisfied in order to be tracked by
// channelz as Channel or SubChannel.
type Channel interface {
ChannelzMetric() *ChannelInternalMetric
}
type channel struct {
refName string
c Channel
closeCalled bool
nestedChans map[int64]string
subChans map[int64]string
id int64
pid int64
cm *channelMap
}
func (c *channel) addChild(id int64, e entry) {
switch v := e.(type) {
case *subChannel:
c.subChans[id] = v.refName
case *channel:
c.nestedChans[id] = v.refName
default:
grpclog.Errorf("cannot add a child (id = %d) of type %T to a channel", id, e)
}
}
func (c *channel) deleteChild(id int64) {
delete(c.subChans, id)
delete(c.nestedChans, id)
c.deleteSelfIfReady()
}
func (c *channel) triggerDelete() {
c.closeCalled = true
c.deleteSelfIfReady()
}
func (c *channel) deleteSelfIfReady() {
if !c.closeCalled || len(c.subChans)+len(c.nestedChans) != 0 {
return
}
c.cm.deleteEntry(c.id)
// not top channel
if c.pid != 0 {
c.cm.findEntry(c.pid).deleteChild(c.id)
}
}
type subChannel struct {
refName string
c Channel
closeCalled bool
sockets map[int64]string
id int64
pid int64
cm *channelMap
}
func (sc *subChannel) addChild(id int64, e entry) {
if v, ok := e.(*normalSocket); ok {
sc.sockets[id] = v.refName
} else {
grpclog.Errorf("cannot add a child (id = %d) of type %T to a subChannel", id, e)
}
}
func (sc *subChannel) deleteChild(id int64) {
delete(sc.sockets, id)
sc.deleteSelfIfReady()
}
func (sc *subChannel) triggerDelete() {
sc.closeCalled = true
sc.deleteSelfIfReady()
}
func (sc *subChannel) deleteSelfIfReady() {
if !sc.closeCalled || len(sc.sockets) != 0 {
return
}
sc.cm.deleteEntry(sc.id)
sc.cm.findEntry(sc.pid).deleteChild(sc.id)
}
// SocketMetric defines the info channelz provides for a specific Socket, which
// includes SocketInternalMetric and channelz-specific data, such as channelz id, etc.
type SocketMetric struct {
// ID is the channelz id of this socket.
ID int64
// RefName is the human readable reference string of this socket.
RefName string
// SocketData contains socket internal metric reported by the socket through
// ChannelzMetric().
SocketData *SocketInternalMetric
}
// SocketInternalMetric defines the struct that the implementor of Socket interface
// should return from ChannelzMetric().
type SocketInternalMetric struct {
// The number of streams that have been started.
StreamsStarted int64
// The number of streams that have ended successfully:
// On client side, receiving frame with eos bit set.
// On server side, sending frame with eos bit set.
StreamsSucceeded int64
// The number of streams that have ended unsuccessfully:
// On client side, termination without receiving frame with eos bit set.
// On server side, termination without sending frame with eos bit set.
StreamsFailed int64
// The number of messages successfully sent on this socket.
MessagesSent int64
MessagesReceived int64
// The number of keep alives sent. This is typically implemented with HTTP/2
// ping messages.
KeepAlivesSent int64
// The last time a stream was created by this endpoint. Usually unset for
// servers.
LastLocalStreamCreatedTimestamp time.Time
// The last time a stream was created by the remote endpoint. Usually unset
// for clients.
LastRemoteStreamCreatedTimestamp time.Time
// The last time a message was sent by this endpoint.
LastMessageSentTimestamp time.Time
// The last time a message was received by this endpoint.
LastMessageReceivedTimestamp time.Time
// The amount of window, granted to the local endpoint by the remote endpoint.
// This may be slightly out of date due to network latency. This does NOT
// include stream level or TCP level flow control info.
LocalFlowControlWindow int64
// The amount of window, granted to the remote endpoint by the local endpoint.
// This may be slightly out of date due to network latency. This does NOT
// include stream level or TCP level flow control info.
RemoteFlowControlWindow int64
// The locally bound address.
LocalAddr net.Addr
// The remote bound address. May be absent.
RemoteAddr net.Addr
// Optional, represents the name of the remote endpoint, if different than
// the original target name.
RemoteName string
//TODO: socket options
//TODO: Security
}
// Socket is the interface that should be satisfied in order to be tracked by
// channelz as Socket.
type Socket interface {
ChannelzMetric() *SocketInternalMetric
}
type listenSocket struct {
refName string
s Socket
id int64
pid int64
cm *channelMap
}
func (ls *listenSocket) addChild(id int64, e entry) {
grpclog.Errorf("cannot add a child (id = %d) of type %T to a listen socket", id, e)
}
func (ls *listenSocket) deleteChild(id int64) {
grpclog.Errorf("cannot delete a child (id = %d) from a listen socket", id)
}
func (ls *listenSocket) triggerDelete() {
ls.cm.deleteEntry(ls.id)
ls.cm.findEntry(ls.pid).deleteChild(ls.id)
}
func (ls *listenSocket) deleteSelfIfReady() {
grpclog.Errorf("cannot call deleteSelfIfReady on a listen socket")
}
type normalSocket struct {
refName string
s Socket
id int64
pid int64
cm *channelMap
}
func (ns *normalSocket) addChild(id int64, e entry) {
grpclog.Errorf("cannot add a child (id = %d) of type %T to a normal socket", id, e)
}
func (ns *normalSocket) deleteChild(id int64) {
grpclog.Errorf("cannot delete a child (id = %d) from a normal socket", id)
}
func (ns *normalSocket) triggerDelete() {
ns.cm.deleteEntry(ns.id)
ns.cm.findEntry(ns.pid).deleteChild(ns.id)
}
func (ns *normalSocket) deleteSelfIfReady() {
grpclog.Errorf("cannot call deleteSelfIfReady on a normal socket")
}
// ServerMetric defines the info channelz provides for a specific Server, which
// includes ServerInternalMetric and channelz-specific data, such as channelz id,
// child list, etc.
type ServerMetric struct {
// ID is the channelz id of this server.
ID int64
// RefName is the human readable reference string of this server.
RefName string
// ServerData contains server internal metric reported by the server through
// ChannelzMetric().
ServerData *ServerInternalMetric
// ListenSockets tracks the listener socket type children of this server in the
// format of a map from socket channelz id to corresponding reference string.
ListenSockets map[int64]string
}
// ServerInternalMetric defines the struct that the implementor of Server interface
// should return from ChannelzMetric().
type ServerInternalMetric struct {
// The number of incoming calls started on the server.
CallsStarted int64
// The number of incoming calls that have completed with an OK status.
CallsSucceeded int64
// The number of incoming calls that have a completed with a non-OK status.
CallsFailed int64
// The last time a call was started on the server.
LastCallStartedTimestamp time.Time
//TODO: trace
}
// Server is the interface to be satisfied in order to be tracked by channelz as
// Server.
type Server interface {
ChannelzMetric() *ServerInternalMetric
}
type server struct {
refName string
s Server
closeCalled bool
sockets map[int64]string
listenSockets map[int64]string
id int64
cm *channelMap
}
func (s *server) addChild(id int64, e entry) {
switch v := e.(type) {
case *normalSocket:
s.sockets[id] = v.refName
case *listenSocket:
s.listenSockets[id] = v.refName
default:
grpclog.Errorf("cannot add a child (id = %d) of type %T to a server", id, e)
}
}
func (s *server) deleteChild(id int64) {
delete(s.sockets, id)
delete(s.listenSockets, id)
s.deleteSelfIfReady()
}
func (s *server) triggerDelete() {
s.closeCalled = true
s.deleteSelfIfReady()
}
func (s *server) deleteSelfIfReady() {
if !s.closeCalled || len(s.sockets)+len(s.listenSockets) != 0 {
return
}
s.cm.deleteEntry(s.id)
}