// servingConnStats returns the number of serving tablets per keyspace/shard/tablet type.
356417
func (hc *HealthCheckImpl) servingConnStats() map[string]int64 {
	res := make(map[string]int64)
	hc.mu.Lock()
	defer hc.mu.Unlock()
	for _, th := range hc.addrToHealth {
		if !th.latestTabletStats.Up || !th.latestTabletStats.Serving || th.latestTabletStats.LastError != nil {
			continue
		}
		key := fmt.Sprintf("%s.%s.%s", th.latestTabletStats.Target.Keyspace, th.latestTabletStats.Target.Shard, topoproto.TabletTypeLString(th.latestTabletStats.Target.TabletType))
		res[key]++
	}
	return res
}
1054550
func (t *T) Accept() (net.Conn, error) {
	c, err := t.l.Accept()
	if err != nil {
		return nil, err
	}

	connID := atomic.AddUint64(&t.connCount, 1)
	shard := &t.shards[int(connID)%len(t.shards)]
	wc := &conn{
		Conn:  c,
		shard: shard,
		mode:  t.mode,
	}

	if err = wc.init(); err != nil {
		return nil, err
	}
	return wc, nil
}
1311656
func ServeConn(block BlockCrypt, dataShards, parityShards int, conn net.PacketConn) (*Listener, error) {
	l := new(Listener)
	l.conn = conn
	l.sessions = make(map[string]*UDPSession)
	l.chAccepts = make(chan *UDPSession, acceptBacklog)
	l.chSessionClosed = make(chan net.Addr)
	l.die = make(chan struct{})
	l.dataShards = dataShards
	l.parityShards = parityShards
	l.block = block
	l.fecDecoder = newFECDecoder(rxFECMulti*(dataShards+parityShards), dataShards, parityShards)

	// calculate header size
	if l.block != nil {
		l.headerSize += cryptHeaderSize
	}
	if l.fecDecoder != nil {
		l.headerSize += fecHeaderSizePlus2
	}

	go l.monitor()
	return l, nil
}
7324
func (c *ServicedCli) cmdSetConnTimeout(ctx *cli.Context) {
	args := ctx.Args()
	if len(args) != 2 {
		fmt.Printf("Incorrect Usage.\n\n")
		cli.ShowCommandHelp(ctx, "set-conn-timeout")
		return
	}

	connTimeout, err := time.ParseDuration(args[1])
	if err != nil {
		fmt.Fprintf(os.Stderr, "could not parse duration: %s\n", err)
		return
	} else if connTimeout < 0 {
		fmt.Fprintln(os.Stderr, "duration cannot be negative")
		return
	}

	pool, err := c.driver.GetResourcePool(args[0])
	if err != nil {
		fmt.Fprintln(os.Stderr, err)
		return
	} else if pool == nil {
		fmt.Fprintln(os.Stderr, "pool not found")
		return
	}

	pool.ConnectionTimeout = int(connTimeout.Seconds() * 1000)
	if err := c.driver.UpdateResourcePool(*pool); err != nil {
		fmt.Fprintln(os.Stderr, err)
		return
	}
}
813552
func (h presenceHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	if h.presence == nil || !h.presence.IsEnabled() {
		w.WriteHeader(http.StatusNotFound)
		w.Write([]byte("agent is not an apiserver\n"))
		return
	}

	w.Header().Set("Content-Type", "text/plain; charset=utf-8")

	tw := output.TabWriter(w)
	wrapper := output.Wrapper{tw}

	// Could be smart here and switch on the request accept header.
	connections := h.presence.Connections()
	models := connections.Models()
	sort.Strings(models)

	for _, name := range models {
		wrapper.Println("[" + name + "]")
		wrapper.Println()
		wrapper.Println("AGENT", "SERVER", "CONN ID", "STATUS")
		values := connections.ForModel(name).Values()
		sort.Sort(ValueSort(values))
		for _, value := range values {
			agentName := value.Agent
			if value.ControllerAgent {
				agentName += " (controller)"
			}
			wrapper.Println(agentName, value.Server, value.ConnectionID, value.Status)
		}
		wrapper.Println()
	}
	tw.Flush()
}
245436
func (o *IKEGateway) IKESubnets(info *bambou.FetchingInfo) (IKESubnetsList, *bambou.Error) {

	var list IKESubnetsList
	err := bambou.CurrentSession().FetchChildren(o, IKESubnetIdentity, &list, info)
	return list, err
}
552682
func (t *trace) ServeDNS(ctx context.Context, w dns.ResponseWriter, r *dns.Msg) (int, error) {
	trace := false
	if t.every > 0 {
		queryNr := atomic.AddUint64(&t.count, 1)

		if queryNr%t.every == 0 {
			trace = true
		}
	}
	if span := ot.SpanFromContext(ctx); span == nil && trace {
		span := t.Tracer().StartSpan("servedns")
		defer span.Finish()
		ctx = ot.ContextWithSpan(ctx, span)
	}
	return plugin.NextOrFailure(t.Name(), t.Next, ctx, w, r)
}
571962
func (rc *ReplicationConn) SendStandbyStatus(k *StandbyStatus) (err error) {
	buf := rc.wbuf
	buf = append(buf, copyData)
	sp := len(buf)
	buf = pgio.AppendInt32(buf, -1)

	buf = append(buf, standbyStatusUpdate)
	buf = pgio.AppendInt64(buf, int64(k.WalWritePosition))
	buf = pgio.AppendInt64(buf, int64(k.WalFlushPosition))
	buf = pgio.AppendInt64(buf, int64(k.WalApplyPosition))
	buf = pgio.AppendInt64(buf, int64(k.ClientTime))
	buf = append(buf, k.ReplyRequested)

	pgio.SetInt32(buf[sp:], int32(len(buf[sp:])))

	_, err = rc.conn.Write(buf)
	if err != nil {
		rc.die(err)
	}

	return
}
1282277
func (s *PresenceService) GetTunnelConnections(clusterName string, opts ...services.MarshalOption) ([]services.TunnelConnection, error) {
	if clusterName == "" {
		return nil, trace.BadParameter("missing cluster name")
	}
	startKey := backend.Key(tunnelConnectionsPrefix, clusterName)
	result, err := s.GetRange(context.TODO(), startKey, backend.RangeEnd(startKey), backend.NoLimit)
	if err != nil {
		return nil, trace.Wrap(err)
	}
	conns := make([]services.TunnelConnection, len(result.Items))
	for i, item := range result.Items {
		conn, err := services.UnmarshalTunnelConnection(item.Value,
			services.AddOptions(opts, services.WithResourceID(item.ID), services.WithExpires(item.Expires))...)
		if err != nil {
			return nil, trace.Wrap(err)
		}
		conns[i] = conn
	}

	return conns, nil
}
176725
func (s *Server) connsRequest(sub *subscription, subject, reply string, msg []byte) {
	if !s.eventsRunning() {
		return
	}
	m := accNumConnsReq{}
	if err := json.Unmarshal(msg, &m); err != nil {
		s.sys.client.Errorf("Error unmarshalling account connections request message: %v", err)
		return
	}
	acc, _ := s.lookupAccount(m.Account)
	if acc == nil {
		return
	}
	if nlc := acc.NumLocalConnections(); nlc > 0 {
		s.mu.Lock()
		s.sendAccConnsUpdate(acc, reply)
		s.mu.Unlock()
	}
}
147934
func NewScatterConn(statsName string, txConn *TxConn, gw gateway.Gateway, hc discovery.HealthCheck) *ScatterConn {
	tabletCallErrorCountStatsName := ""
	if statsName != "" {
		tabletCallErrorCountStatsName = statsName + "ErrorCount"
	}
	return &ScatterConn{
		timings: stats.NewMultiTimings(
			statsName,
			"Scatter connection timings",
			[]string{"Operation", "Keyspace", "ShardName", "DbType"}),
		tabletCallErrorCount: stats.NewCountersWithMultiLabels(
			tabletCallErrorCountStatsName,
			"Error count from tablet calls in scatter conns",
			[]string{"Operation", "Keyspace", "ShardName", "DbType"}),
		txConn:      txConn,
		gateway:     gw,
		healthCheck: hc,
	}
}
730305
func Hosts(hosts string) *CassandraSource {
	return &CassandraSource{
		hosts:        hosts,
		keyspace:     "system",
		table:        "local",
		selectClause: "*",
		ShardCount:   32,
		Concurrency:  4,
		prefix:       "cassandra",
	}
}
1321812
func (a *AuthServer) GetTunnelConnections(clusterName string, opts ...services.MarshalOption) ([]services.TunnelConnection, error) {
	return a.GetCache().GetTunnelConnections(clusterName, opts...)
}
463120
func (a *api) GetServiceInstances(serviceID string) ([]service.Instance, error) {
	client, err := a.connectMaster()
	if err != nil {
		return nil, err
	}
	return client.GetServiceInstances(serviceID)
}
1002436
func WaitForTablets(gw Gateway, tabletTypesToWait []topodatapb.TabletType) error {
	log.Infof("Gateway waiting for serving tablets...")
	ctx, cancel := context.WithTimeout(context.Background(), *initialTabletTimeout)
	defer cancel()

	err := gw.WaitForTablets(ctx, tabletTypesToWait)
	switch err {
	case nil:
		// Log so we know everything is fine.
		log.Infof("Waiting for tablets completed")
	case context.DeadlineExceeded:
		// In this scenario, we were able to reach the
		// topology service, but some tablets may not be
		// ready. We just warn and keep going.
		log.Warningf("Timeout waiting for all keyspaces / shards to have healthy tablets, may be in degraded mode")
		err = nil
	default:
		// Nothing to do here, the caller will log.Fatalf.
	}
	return err
}
811029
func (server *Server) ServeConnWithContext(conn io.ReadWriteCloser, context interface{}) {
	buf := bufio.NewWriter(conn)
	srv := &gobServerCodec{conn, gob.NewDecoder(conn), gob.NewEncoder(buf), buf}
	server.ServeCodecWithContext(srv, context)
}
965766
func (a *Account) numLocalConnections() int {
	return len(a.clients) - int(a.sysclients) - int(a.nleafs)
}