// NewStoreStatusMonitor constructs a StoreStatusMonitor with the given ID.
func NewStoreStatusMonitor(id roachpb.StoreID, metaRegistry *metric.Registry) *StoreStatusMonitor {
	registry := metric.NewRegistry()
	// Format as `cr.store.<metric>.<id>` in output, in analogy to the time
	// series data written.
	metaRegistry.MustAdd(storeTimeSeriesPrefix+"%s."+id.String(), registry)
	return &StoreStatusMonitor{
		ID:                   id,
		registry:             registry,
		rangeCount:           registry.Counter("ranges"),
		leaderRangeCount:     registry.Gauge("ranges.leader"),
		replicatedRangeCount: registry.Gauge("ranges.replicated"),
		availableRangeCount:  registry.Gauge("ranges.available"),
		liveBytes:            registry.Gauge("livebytes"),
		keyBytes:             registry.Gauge("keybytes"),
		valBytes:             registry.Gauge("valbytes"),
		intentBytes:          registry.Gauge("intentbytes"),
		liveCount:            registry.Gauge("livecount"),
		keyCount:             registry.Gauge("keycount"),
		valCount:             registry.Gauge("valcount"),
		intentCount:          registry.Gauge("intentcount"),
		intentAge:            registry.Gauge("intentage"),
		gcBytesAge:           registry.Gauge("gcbytesage"),
		lastUpdateNanos:      registry.Gauge("lastupdatenanos"),
		capacity:             registry.Gauge("capacity"),
		available:            registry.Gauge("capacity.available"),
	}
}

// New creates an instance of a gossip node.
func New(rpcContext *rpc.Context, grpcServer *grpc.Server, resolvers []resolver.Resolver, stopper *stop.Stopper, registry *metric.Registry) *Gossip {
	g := &Gossip{
		Connected:         make(chan struct{}),
		rpcContext:        rpcContext,
		server:            newServer(stopper, registry),
		outgoing:          makeNodeSet(minPeers, registry.Gauge(ConnectionsOutgoingGaugeName)),
		bootstrapping:     map[string]struct{}{},
		disconnected:      make(chan *client, 10),
		stalledCh:         make(chan struct{}, 1),
		stallInterval:     defaultStallInterval,
		bootstrapInterval: defaultBootstrapInterval,
		cullInterval:      defaultCullInterval,
		nodeDescs:         map[roachpb.NodeID]*roachpb.NodeDescriptor{},
		resolverAddrs:     map[util.UnresolvedAddr]resolver.Resolver{},
		bootstrapAddrs:    map[util.UnresolvedAddr]struct{}{},
	}
	g.SetResolvers(resolvers)

	// Add ourselves as a SystemConfig watcher.
	g.is.registerCallback(KeySystemConfig, g.updateSystemConfig)
	// Add ourselves as a node descriptor watcher.
	g.is.registerCallback(MakePrefixPattern(KeyNodeIDPrefix), g.updateNodeAddress)

	RegisterGossipServer(grpcServer, g.server)

	return g
}

// NewExecutor creates an Executor and registers a callback on the
// system config.
func NewExecutor(db client.DB, gossip *gossip.Gossip, leaseMgr *LeaseManager, metaRegistry *metric.Registry, stopper *stop.Stopper) *Executor {
	exec := &Executor{
		db:       db,
		reCache:  parser.NewRegexpCache(512),
		leaseMgr: leaseMgr,

		latency: metaRegistry.Latency("sql.latency"),
	}
	exec.systemConfigCond = sync.NewCond(&exec.systemConfigMu)

	gossipUpdateC := gossip.RegisterSystemConfigChannel()
	stopper.RunWorker(func() {
		for {
			select {
			case <-gossipUpdateC:
				cfg := gossip.GetSystemConfig()
				exec.updateSystemConfig(cfg)
			case <-stopper.ShouldStop():
				return
			}
		}
	})

	return exec
}

func makeNodeMetrics(reg *metric.Registry) nodeMetrics {
	nm := nodeMetrics{
		Latency: metric.NewLatency(metaExecLatency),
		Success: metric.NewRates(metaExecSuccess),
		Err:     metric.NewRates(metaExecError),
	}
	reg.AddMetricStruct(nm)
	return nm
}

// NewTxnMetrics returns a new instance of txnMetrics that contains metrics which have
// been registered with the provided Registry.
func NewTxnMetrics(txnRegistry *metric.Registry) *TxnMetrics {
	return &TxnMetrics{
		Aborts:    txnRegistry.Rates(abortsPrefix),
		Commits:   txnRegistry.Rates(commitsPrefix),
		Abandons:  txnRegistry.Rates(abandonsPrefix),
		Durations: txnRegistry.Latency(durationsPrefix),
		Restarts:  txnRegistry.Histogram(restartsKey, 60*time.Second, 100, 3),
	}
}

// newServer creates and returns a server struct.
func newServer(stopper *stop.Stopper, registry *metric.Registry) *server {
	return &server{
		stopper:       stopper,
		is:            newInfoStore(0, util.UnresolvedAddr{}, stopper),
		incoming:      makeNodeSet(minPeers, registry.Gauge(ConnectionsIncomingGaugeName)),
		nodeMap:       make(map[util.UnresolvedAddr]serverInfo),
		tighten:       make(chan roachpb.NodeID, 1),
		ready:         make(chan struct{}),
		nodeMetrics:   makeMetrics(registry),
		serverMetrics: makeMetrics(metric.NewRegistry()),
	}
}

// makeMetrics makes a new metrics object with rates set on the provided
// registry.
func makeMetrics(registry *metric.Registry) metrics {
	return metrics{
		bytesReceived: registry.Rates(BytesReceivedRatesName),
		bytesSent:     registry.Rates(BytesSentRatesName),
		infosReceived: registry.Rates(InfosReceivedRatesName),
		infosSent:     registry.Rates(InfosSentRatesName),
	}
}

// eachRecordableValue visits each metric in the registry, calling the supplied
// function once for each recordable value represented by that metric. This is
// useful to expand certain metric types (such as histograms) into multiple
// recordable values.
func eachRecordableValue(reg *metric.Registry, fn func(string, float64)) {
	reg.Each(func(name string, mtr interface{}) {
		if histogram, ok := mtr.(*metric.Histogram); ok {
			curr := histogram.Current()
			for _, pt := range recordHistogramQuantiles {
				fn(name+pt.suffix, float64(curr.ValueAtQuantile(pt.quantile)))
			}
		} else {
			val, err := extractValue(mtr)
			if err != nil {
				log.Warning(context.TODO(), err)
				return
			}
			fn(name, val)
		}
	})
}

// New creates an instance of a gossip node.
func New(
	ctx context.Context,
	rpcContext *rpc.Context,
	grpcServer *grpc.Server,
	resolvers []resolver.Resolver,
	stopper *stop.Stopper,
	registry *metric.Registry,
) *Gossip {
	ctx = log.WithEventLog(ctx, "gossip", "gossip")
	g := &Gossip{
		ctx:               ctx,
		Connected:         make(chan struct{}),
		rpcContext:        rpcContext,
		server:            newServer(ctx, stopper, registry),
		outgoing:          makeNodeSet(minPeers, metric.NewGauge(MetaConnectionsOutgoingGauge)),
		bootstrapping:     map[string]struct{}{},
		disconnected:      make(chan *client, 10),
		stalledCh:         make(chan struct{}, 1),
		stallInterval:     defaultStallInterval,
		bootstrapInterval: defaultBootstrapInterval,
		cullInterval:      defaultCullInterval,
		nodeDescs:         map[roachpb.NodeID]*roachpb.NodeDescriptor{},
		resolverAddrs:     map[util.UnresolvedAddr]resolver.Resolver{},
		bootstrapAddrs:    map[util.UnresolvedAddr]struct{}{},
	}
	stopper.AddCloser(stop.CloserFn(func() {
		log.FinishEventLog(ctx)
	}))

	registry.AddMetric(g.outgoing.gauge)
	g.clientsMu.breakers = map[string]*circuit.Breaker{}
	log.Infof(g.ctx, "initial resolvers: %s", resolvers)
	g.SetResolvers(resolvers)

	g.mu.Lock()
	// Add ourselves as a SystemConfig watcher.
	g.mu.is.registerCallback(KeySystemConfig, g.updateSystemConfig)
	// Add ourselves as a node descriptor watcher.
	g.mu.is.registerCallback(MakePrefixPattern(KeyNodeIDPrefix), g.updateNodeAddress)
	g.mu.Unlock()

	RegisterGossipServer(grpcServer, g.server)
	return g
}

// newServer creates and returns a server struct.
func newServer(ctx context.Context, stopper *stop.Stopper, registry *metric.Registry) *server {
	s := &server{
		ctx:           ctx,
		stopper:       stopper,
		tighten:       make(chan roachpb.NodeID, 1),
		nodeMetrics:   makeMetrics(),
		serverMetrics: makeMetrics(),
	}

	s.mu.is = newInfoStore(ctx, 0, util.UnresolvedAddr{}, stopper)
	s.mu.incoming = makeNodeSet(minPeers, metric.NewGauge(MetaConnectionsIncomingGauge))
	s.mu.nodeMap = make(map[util.UnresolvedAddr]serverInfo)
	s.mu.ready = make(chan struct{})

	registry.AddMetric(s.mu.incoming.gauge)
	registry.AddMetricStruct(s.nodeMetrics)

	return s
}

func newServerMetrics(reg *metric.Registry) *serverMetrics {
	return &serverMetrics{
		conns:         reg.Counter("conns"),
		bytesInCount:  reg.Counter("bytesin"),
		bytesOutCount: reg.Counter("bytesout"),
	}
}

func newServerMetrics(reg *metric.Registry) *serverMetrics {
	return &serverMetrics{
		conns:         reg.Counter(MetricConnsName),
		bytesInCount:  reg.Counter(MetricBytesInName),
		bytesOutCount: reg.Counter(MetricBytesOutName),
	}
}

func makeNodeMetrics(reg *metric.Registry) nodeMetrics {
	return nodeMetrics{
		registry: reg,
		latency:  reg.Latency(execLatencyName),
		success:  reg.Rates(execSuccessName),
		err:      reg.Rates(execErrorName),
	}
}

// MakeServer creates a Server, adding network stats to the given Registry.
func MakeServer(context *base.Context, executor *sql.Executor, reg *metric.Registry) *Server {
	return &Server{
		context:  context,
		executor: executor,
		registry: reg,
		metrics: &serverMetrics{
			conns:         reg.Counter("conns"),
			bytesInCount:  reg.Counter("bytesin"),
			bytesOutCount: reg.Counter("bytesout"),
		},
	}
}

// NewExecutor creates an Executor and registers a callback on the
// system config.
func NewExecutor(ctx ExecutorContext, stopper *stop.Stopper, registry *metric.Registry) *Executor {
	exec := &Executor{
		ctx:     ctx,
		reCache: parser.NewRegexpCache(512),

		registry:         registry,
		latency:          registry.Latency("latency"),
		txnBeginCount:    registry.Counter("txn.begin.count"),
		txnCommitCount:   registry.Counter("txn.commit.count"),
		txnAbortCount:    registry.Counter("txn.abort.count"),
		txnRollbackCount: registry.Counter("txn.rollback.count"),
		selectCount:      registry.Counter("select.count"),
		updateCount:      registry.Counter("update.count"),
		insertCount:      registry.Counter("insert.count"),
		deleteCount:      registry.Counter("delete.count"),
		ddlCount:         registry.Counter("ddl.count"),
		miscCount:        registry.Counter("misc.count"),
	}
	exec.systemConfigCond = sync.NewCond(exec.systemConfigMu.RLocker())

	gossipUpdateC := ctx.Gossip.RegisterSystemConfigChannel()
	stopper.RunWorker(func() {
		for {
			select {
			case <-gossipUpdateC:
				cfg, _ := ctx.Gossip.GetSystemConfig()
				exec.updateSystemConfig(cfg)
			case <-stopper.ShouldStop():
				return
			}
		}
	})

	return exec
}

// NewExecutor creates an Executor and registers a callback on the
// system config.
func NewExecutor(ctx ExecutorContext, stopper *stop.Stopper, registry *metric.Registry) *Executor {
	exec := &Executor{
		ctx:     ctx,
		reCache: parser.NewRegexpCache(512),

		registry:         registry,
		latency:          registry.Latency(MetricLatencyName),
		txnBeginCount:    registry.Counter(MetricTxnBeginName),
		txnCommitCount:   registry.Counter(MetricTxnCommitName),
		txnAbortCount:    registry.Counter(MetricTxnAbortName),
		txnRollbackCount: registry.Counter(MetricTxnRollbackName),
		selectCount:      registry.Counter(MetricSelectName),
		updateCount:      registry.Counter(MetricUpdateName),
		insertCount:      registry.Counter(MetricInsertName),
		deleteCount:      registry.Counter(MetricDeleteName),
		ddlCount:         registry.Counter(MetricDdlName),
		miscCount:        registry.Counter(MetricMiscName),
		queryCount:       registry.Counter(MetricQueryName),
	}
	exec.systemConfigCond = sync.NewCond(exec.systemConfigMu.RLocker())

	gossipUpdateC := ctx.Gossip.RegisterSystemConfigChannel()
	stopper.RunWorker(func() {
		for {
			select {
			case <-gossipUpdateC:
				cfg, _ := ctx.Gossip.GetSystemConfig()
				exec.updateSystemConfig(cfg)
			case <-stopper.ShouldStop():
				return
			}
		}
	})

	return exec
}

// MakeRuntimeStatSampler constructs a new RuntimeStatSampler object.
func MakeRuntimeStatSampler(clock *hlc.Clock, reg *metric.Registry) RuntimeStatSampler {
	return RuntimeStatSampler{
		registry:       reg,
		clock:          clock,
		cgoCalls:       reg.Gauge(nameCgoCalls),
		goroutines:     reg.Gauge(nameGoroutines),
		goAllocBytes:   reg.Gauge(nameGoAllocBytes),
		goTotalBytes:   reg.Gauge(nameGoTotalBytes),
		cgoAllocBytes:  reg.Gauge(nameCgoAllocBytes),
		cgoTotalBytes:  reg.Gauge(nameCgoTotalBytes),
		gcCount:        reg.Gauge(nameGCCount),
		gcPauseNS:      reg.Gauge(nameGCPauseNS),
		gcPausePercent: reg.GaugeFloat64(nameGCPausePercent),
		cpuUserNS:      reg.Gauge(nameCPUUserNS),
		cpuUserPercent: reg.GaugeFloat64(nameCPUUserPercent),
		cpuSysNS:       reg.Gauge(nameCPUSysNS),
		cpuSysPercent:  reg.GaugeFloat64(nameCPUSysPercent),
		rss:            reg.Gauge(nameRSS),
	}
}

// RegisterMetrics adds the local metrics to a registry.
// TODO(marc): this pattern deviates from other users of the registry
// that take it as an argument at metric construction time.
func (r *RemoteClockMonitor) RegisterMetrics(reg *metric.Registry) {
	reg.MustAdd(clusterOffsetLowerBoundName, r.metrics.clusterOffsetLowerBound)
	reg.MustAdd(clusterOffsetUpperBoundName, r.metrics.clusterOffsetUpperBound)
}