// serve connections on this listener until it is closed.
func (s *Server) serve(ln net.Listener) {
	for {
		conn, err := ln.Accept()
		if err != nil {
			if !s.isClosing() {
				log.Error(err)
			}
			return
		}

		s.mu.Lock()
		s.conns[conn] = struct{}{}
		s.mu.Unlock()

		go func() {
			defer func() {
				s.mu.Lock()
				delete(s.conns, conn)
				s.mu.Unlock()
				conn.Close()
			}()

			if err := s.serveConn(conn); err != nil {
				if err != io.EOF && !s.isClosing() {
					log.Error(err)
				}
			}
		}()
	}
}

// handleLocalLogFile handles GET requests for a single log. If no filename is
// available, it returns 404. The log contents are returned in structured
// format as JSON.
func (s *statusServer) handleLocalLogFile(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
	log.Flush()
	file := ps.ByName("file")
	reader, err := log.GetLogReader(file, false /* !allowAbsolute */)
	if reader == nil || err != nil {
		log.Errorf("unable to open log file %s: %s", file, err)
		http.NotFound(w, r)
		return
	}
	defer reader.Close()

	entry := log.LogEntry{}
	var entries []log.LogEntry
	decoder := log.NewEntryDecoder(reader)
	for {
		if err := decoder.Decode(&entry); err != nil {
			if err == io.EOF {
				break
			}
			log.Error(err)
			w.WriteHeader(http.StatusInternalServerError)
			return
		}
		entries = append(entries, entry)
	}

	b, contentType, err := util.MarshalResponse(r, entries, []util.EncodingType{util.JSONEncoding})
	if err != nil {
		log.Error(err)
		w.WriteHeader(http.StatusInternalServerError)
		return
	}
	w.Header().Set(util.ContentTypeHeader, contentType)
	w.Write(b)
}

// handleStoreStatus handles GET requests for a single node's status. If no id
// is available, it calls handleStoresStatus to return all store's statuses.
func (s *statusServer) handleStoreStatus(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
	id, err := strconv.ParseInt(ps.ByName("id"), 10, 32)
	if err != nil {
		log.Error(err)
		w.WriteHeader(http.StatusInternalServerError)
		return
	}
	key := keys.StoreStatusKey(int32(id))

	storeStatus := &storage.StoreStatus{}
	if err := s.db.GetProto(key, storeStatus); err != nil {
		log.Error(err)
		w.WriteHeader(http.StatusInternalServerError)
		return
	}

	b, contentType, err := util.MarshalResponse(r, storeStatus, []util.EncodingType{util.JSONEncoding})
	if err != nil {
		log.Error(err)
		w.WriteHeader(http.StatusInternalServerError)
		return
	}
	w.Header().Set(util.ContentTypeHeader, contentType)
	w.Write(b)
}

// shouldQueue determines whether a range should be queued for
// splitting. This is true if the range is intersected by a zone config
// prefix or if the range's size in bytes exceeds the limit for the zone.
func (sq *splitQueue) shouldQueue(now proto.Timestamp, rng *Replica) (shouldQ bool, priority float64) {
	// Load the system config.
	cfg, err := sq.gossip.GetSystemConfig()
	if err != nil {
		log.Error(err)
		return
	}

	desc := rng.Desc()
	if len(cfg.ComputeSplitKeys(desc.StartKey, desc.EndKey)) > 0 {
		// Set priority to 1 in the event the range is split by zone configs.
		priority = 1
		shouldQ = true
	}

	// Add priority based on the size of range compared to the max
	// size for the zone it's in.
	zone, err := cfg.GetZoneConfigForKey(desc.StartKey)
	if err != nil {
		log.Error(err)
		return
	}

	if ratio := float64(rng.stats.GetSize()) / float64(zone.RangeMaxBytes); ratio > 1 {
		priority += ratio
		shouldQ = true
	}
	return
}

// handleStoresStatus handles GET requests for all store statuses.
func (s *statusServer) handleStoresStatus(w http.ResponseWriter, r *http.Request, _ httprouter.Params) {
	startKey := keys.StatusStorePrefix
	endKey := startKey.PrefixEnd()

	rows, err := s.db.Scan(startKey, endKey, 0)
	if err != nil {
		log.Error(err)
		w.WriteHeader(http.StatusInternalServerError)
		return
	}

	storeStatuses := []storage.StoreStatus{}
	for _, row := range rows {
		storeStatus := &storage.StoreStatus{}
		if err := row.ValueProto(storeStatus); err != nil {
			log.Error(err)
			w.WriteHeader(http.StatusInternalServerError)
			return
		}
		storeStatuses = append(storeStatuses, *storeStatus)
	}
	b, contentType, err := util.MarshalResponse(r, storeStatuses, []util.EncodingType{util.JSONEncoding})
	if err != nil {
		log.Error(err)
		w.WriteHeader(http.StatusInternalServerError)
		return
	}
	w.Header().Set(util.ContentTypeHeader, contentType)
	w.Write(b)
}

func (rq replicateQueue) shouldQueue(now proto.Timestamp, repl *Replica) (shouldQ bool, priority float64) {
	// Load the system config.
	cfg, err := rq.gossip.GetSystemConfig()
	if err != nil {
		log.Error(err)
		return
	}

	desc := repl.Desc()
	if len(cfg.ComputeSplitKeys(desc.StartKey, desc.EndKey)) > 0 {
		// If the replica's range needs splitting, wait until done.
		return
	}

	// Find the zone config for this range.
	zone, err := cfg.GetZoneConfigForKey(desc.StartKey)
	if err != nil {
		log.Error(err)
		return
	}

	action, priority := rq.allocator.ComputeAction(*zone, repl.Desc())
	if action == aaNoop {
		return false, 0
	}
	return true, priority
}

// runSetUser invokes the REST API with POST action and username as
// path. Prompts for the password twice on stdin.
// TODO(marc): once we have more fields in the user config, we will need
// to allow changing just some of them (eg: change email, but leave password).
func runSetUser(cmd *cobra.Command, args []string) {
	if len(args) != 1 {
		cmd.Usage()
		return
	}
	hashed, err := security.PromptForPasswordAndHash()
	if err != nil {
		log.Error(err)
		return
	}
	// Build a UserConfig object. RunSetUser expects Yaml.
	// TODO(marc): re-work admin client library to take other encodings.
	pb := &proto.UserConfig{HashedPassword: hashed}
	contents, err := yaml.Marshal(pb)
	if err != nil {
		log.Error(err)
		return
	}
	admin := client.NewAdminClient(&Context.Context, Context.Addr, client.User)
	if err := admin.SetYAML(args[0], string(contents)); err != nil {
		log.Error(err)
		return
	}
	fmt.Printf("Wrote user config for %q\n", args[0])
}

// runLog accesses creates a term log entry reader for each
// log file named in arguments.
func runLog(cmd *cobra.Command, args []string) {
	for _, arg := range args {
		reader, err := log.GetLogReader(arg, false /* !restricted */)
		if err != nil {
			log.Error(err)
			break
		}
		if _, err := io.Copy(os.Stdout, log.NewTermEntryReader(reader)); err != nil {
			log.Error(err)
			break
		}
		reader.Close()
	}
}

// newStatusServer allocates and returns a statusServer.
func newStatusServer(db *client.DB, gossip *gossip.Gossip, ctx *Context) *statusServer {
	// Create an http client with a timeout
	tlsConfig, err := ctx.GetClientTLSConfig()
	if err != nil {
		log.Error(err)
		return nil
	}
	httpClient := &http.Client{
		Transport: &http.Transport{TLSClientConfig: tlsConfig},
		Timeout:   logEntriesTimeout,
	}

	server := &statusServer{
		db:          db,
		gossip:      gossip,
		router:      httprouter.New(),
		ctx:         ctx,
		proxyClient: httpClient,
	}

	server.router.GET(statusGossipKeyPrefix, server.handleGossipStatus)
	server.router.GET(statusLocalKeyPrefix, server.handleLocalStatus)
	server.router.GET(statusLocalLogFileKeyPrefix, server.handleLocalLogFiles)
	server.router.GET(statusLocalLogFileKeyPattern, server.handleLocalLogFile)
	server.router.GET(statusLogKeyPrefix, server.handleLocalLog)
	server.router.GET(statusLogKeyPattern, server.handleLogs)
	server.router.GET(statusLocalStacksKey, server.handleLocalStacks)
	server.router.GET(statusNodeKeyPrefix, server.handleNodesStatus)
	server.router.GET(statusNodeKeyPattern, server.handleNodeStatus)
	server.router.GET(statusStoreKeyPrefix, server.handleStoresStatus)
	server.router.GET(statusStoreKeyPattern, server.handleStoreStatus)
	server.router.GET(statusTransactionsKeyPrefix, server.handleTransactionStatus)

	return server
}

func (rq *replicateQueue) shouldQueue(now roachpb.Timestamp, repl *Replica,
	sysCfg config.SystemConfig) (shouldQ bool, priority float64) {

	if repl.needsSplitBySize() {
		// If the range exceeds the split threshold, let that finish
		// first. Ranges must fit in memory on both sender and receiver
		// nodes while being replicated. This supplements the check
		// provided by acceptsUnsplitRanges, which looks at zone config
		// boundaries rather than data size.
		return
	}

	// Find the zone config for this range.
	desc := repl.Desc()
	zone, err := sysCfg.GetZoneConfigForKey(desc.StartKey)
	if err != nil {
		log.Error(err)
		return
	}

	action, priority := rq.allocator.ComputeAction(*zone, desc)
	if action != AllocatorNoop {
		return true, priority
	}
	// See if there is a rebalancing opportunity present.
	shouldRebalance := rq.allocator.ShouldRebalance(repl.store.StoreID())
	return shouldRebalance, 0
}

// serve connections on this listener until it is closed.
func (s *Server) serve(ln net.Listener) {
	for {
		conn, err := ln.Accept()
		if err != nil {
			log.Error(err)
			return
		}

		s.conns = append(s.conns, conn)
		go func() {
			if err := s.serveConn(conn); err != nil {
				log.Error(err)
			}
		}()
	}
}

// runGetZone retrieves the zone config for a given object id,
// and if present, outputs its YAML representation.
// TODO(marc): accept db/table names rather than IDs.
func runGetZone(cmd *cobra.Command, args []string) {
	if len(args) != 1 {
		mustUsage(cmd)
		return
	}
	id, err := strconv.Atoi(args[0])
	if err != nil {
		log.Errorf("could not parse object ID %s", args[0])
		return
	}

	db, _ := makeSQLClient()
	defer func() { _ = db.Close() }()
	// TODO(marc): switch to placeholders once they work with pgwire.
	_, rows, err := runQueryWithFormat(db, fmtMap{"config": formatZone},
		fmt.Sprintf(`SELECT * FROM system.zones WHERE id=%d`, id))
	if err != nil {
		log.Error(err)
		return
	}

	if len(rows) == 0 {
		log.Errorf("Object %d: no zone config found", id)
		return
	}
	fmt.Println(rows[0][1])
}

func (z *zeroSum) maybeLogError(err error) {
	if strings.Contains(err.Error(), "range is frozen") {
		return
	}
	log.Error(context.Background(), err)
	atomic.AddUint64(&z.stats.errors, 1)
}

func (r *Replica) verifyChecksumTrigger(
	ctx context.Context, args roachpb.VerifyChecksumRequest,
) {
	id := args.ChecksumID
	c, ok := r.getChecksum(ctx, id)
	if !ok {
		log.Errorf(ctx, "consistency check skipped: checksum for id = %v doesn't exist", id)
		// Return success because a checksum might be missing only on
		// this replica. A checksum might be missing because of a
		// number of reasons: GC-ed, server restart, and ComputeChecksum
		// version incompatibility.
		return
	}
	if c.checksum != nil && !bytes.Equal(c.checksum, args.Checksum) {
		// Replication consistency problem!
		logFunc := log.Errorf

		// Collect some more debug information.
		if args.Snapshot == nil {
			// No debug information; run another consistency check to deliver
			// more debug information.
			if err := r.store.stopper.RunAsyncTask(func() {
				log.Errorf(ctx, "%s: consistency check failed; fetching details", r)
				desc := r.Desc()
				startKey := desc.StartKey.AsRawKey()
				// Can't use a start key less than LocalMax.
				if bytes.Compare(startKey, keys.LocalMax) < 0 {
					startKey = keys.LocalMax
				}
				if err := r.store.db.CheckConsistency(startKey, desc.EndKey.AsRawKey(), true /* withDiff */); err != nil {
					log.Errorf(ctx, "couldn't rerun consistency check: %s", err)
				}
			}); err != nil {
				log.Error(ctx, errors.Wrap(err, "could not rerun consistency check"))
			}
		} else {
			// Compute diff.
			diff := diffRange(args.Snapshot, c.snapshot)
			if diff != nil {
				for _, d := range diff {
					l := "leader"
					if d.LeaseHolder {
						l = "replica"
					}
					log.Errorf(ctx, "consistency check failed: k:v = (%s (%x), %s, %x) not present on %s",
						d.Key, d.Key, d.Timestamp, d.Value, l)
				}
			}
			if r.store.ctx.ConsistencyCheckPanicOnFailure {
				if p := r.store.ctx.TestingKnobs.BadChecksumPanic; p != nil {
					p(diff)
				} else {
					logFunc = log.Fatalf
				}
			}
		}

		logFunc(ctx, "consistency check failed on replica: %s, checksum mismatch: e = %x, v = %x", args.Checksum, c.checksum)
	}
}

func (l *LocalCluster) runDockerSpy() {
	l.panicOnStop()

	create := func() (*Container, error) {
		return createContainer(l,
			container.Config{
				Image: dockerspyImage + ":" + dockerspyTag,
				Cmd:   strslice.New("--dns-domain=" + domain),
			}, container.HostConfig{
				Binds:           []string{"/var/run/docker.sock:/var/run/docker.sock"},
				PublishAllPorts: true,
			},
			"docker-spy",
		)
	}
	c, err := create()
	if dockerclient.IsErrImageNotFound(err) {
		if err := pullImage(l, types.ImagePullOptions{ImageID: dockerspyImage, Tag: dockerspyTag}); err != nil {
			log.Fatal(err)
		}

		c, err = create()
	}
	maybePanic(err)
	maybePanic(c.Start())
	l.dns = c
	if ci, err := c.Inspect(); err != nil {
		log.Error(err)
	} else {
		log.Infof("started %s: %s", c.Name(), ci.NetworkSettings.IPAddress)
	}
}

func (rq replicateQueue) shouldQueue(now proto.Timestamp, repl *Replica) (shouldQ bool, priority float64) {
	// If the replica's range spans multiple zones, ignore it until the split
	// queue has processed it.
	if len(computeSplitKeys(rq.gossip, repl)) > 0 {
		return
	}

	// Load the zone config to find the desired replica attributes.
	zone, err := lookupZoneConfig(rq.gossip, repl)
	if err != nil {
		log.Error(err)
		return
	}

	delta := rq.replicaDelta(zone, repl, repl.Desc())
	if delta == 0 {
		if log.V(1) {
			log.Infof("%s has the correct number of nodes", repl)
		}
		return false, 0
	}
	if delta > 0 {
		if log.V(1) {
			log.Infof("%s needs to add %d nodes", repl, delta)
		}
		// For ranges which need additional replicas, increase the priority
		return true, float64(delta + 10)
	}
	if log.V(1) {
		log.Infof("%s needs to remove %d nodes", repl, 0-delta)
	}
	// For ranges which have too many replicas, priority is absolute value of
	// the delta.
	return true, float64(0 - delta)
}

// SetStorage provides an instance of the Storage interface
// for reading and writing gossip bootstrap data from persistent
// storage. This should be invoked as early in the lifecycle of a
// gossip instance as possible, but can be called at any time.
func (g *Gossip) SetStorage(storage Storage) error {
	g.mu.Lock()
	defer g.mu.Unlock()
	g.storage = storage

	// Read the bootstrap info from the persistent store.
	var storedBI BootstrapInfo
	err := storage.ReadBootstrapInfo(&storedBI)
	if err != nil {
		log.Warningf("failed to read gossip bootstrap info: %s", err)
	}

	// Merge the stored bootstrap info addresses with any we've become
	// aware of through gossip.
	if len(g.bootstrapInfo.Addresses) > 0 {
		existing := map[string]struct{}{}
		makeKey := func(a util.UnresolvedAddr) string { return fmt.Sprintf("%s,%s", a.Network(), a.String()) }
		for _, addr := range g.bootstrapInfo.Addresses {
			existing[makeKey(addr)] = struct{}{}
		}
		for _, addr := range storedBI.Addresses {
			// If the address is new, and isn't our own address, add it.
			if _, ok := existing[makeKey(addr)]; !ok && addr != g.is.NodeAddr {
				g.maybeAddBootstrapAddress(addr)
			}
		}
		// Persist merged addresses.
		if numAddrs := len(g.bootstrapInfo.Addresses); numAddrs > len(storedBI.Addresses) {
			if err := g.storage.WriteBootstrapInfo(&g.bootstrapInfo); err != nil {
				log.Error(err)
			}
		}
	} else {
		g.bootstrapInfo = storedBI
	}

	// Cycle through all persisted bootstrap hosts and add resolvers for
	// any which haven't already been added.
	newResolverFound := false
	for _, addr := range g.bootstrapInfo.Addresses {
		if !g.maybeAddResolver(addr) {
			continue
		}
		// If we find a new resolver, reset the resolver index so that the
		// next resolver we try is the first of the new resolvers.
		if !newResolverFound {
			newResolverFound = true
			g.resolverIdx = len(g.resolvers) - 1
		}
	}

	// If a new resolver was found, immediately signal bootstrap.
	if newResolverFound {
		if log.V(1) {
			log.Infof("found new resolvers from storage; signalling bootstrap")
		}
		g.signalStalled()
	}
	return nil
}

// newStatusServer allocates and returns a statusServer.
func newStatusServer(
	db *client.DB,
	gossip *gossip.Gossip,
	metricSource json.Marshaler,
	ctx *base.Context,
	rpcCtx *rpc.Context,
	stores *storage.Stores,
) *statusServer {
	// Create an http client with a timeout
	httpClient, err := ctx.GetHTTPClient()
	if err != nil {
		log.Error(err)
		return nil
	}

	server := &statusServer{
		db:           db,
		gossip:       gossip,
		metricSource: metricSource,
		router:       httprouter.New(),
		rpcCtx:       rpcCtx,
		proxyClient:  httpClient,
		stores:       stores,
	}

	server.router.GET(statusLogFilesListPattern, server.handleLogFilesList)
	server.router.GET(statusLogFilePattern, server.handleLogFile)
	server.router.GET(statusLogsPattern, server.handleLogs)
	// TODO(tschottdorf): significant overlap with /debug/pprof/goroutine,
	// except that this one allows querying by NodeID.
	server.router.GET(statusStacksPattern, server.handleStacks)
	server.router.GET(statusMetricsPattern, server.handleMetrics)

	return server
}

func (rq *replicateQueue) shouldQueue(
	now hlc.Timestamp,
	repl *Replica,
	sysCfg config.SystemConfig,
) (shouldQ bool, priority float64) {
	if !repl.store.splitQueue.Disabled() && repl.needsSplitBySize() {
		// If the range exceeds the split threshold, let that finish first.
		// Ranges must fit in memory on both sender and receiver nodes while
		// being replicated. This supplements the check provided by
		// acceptsUnsplitRanges, which looks at zone config boundaries rather
		// than data size.
		//
		// This check is ignored if the split queue is disabled, since in that
		// case, the split will never come.
		return
	}

	// Find the zone config for this range.
	desc := repl.Desc()
	zone, err := sysCfg.GetZoneConfigForKey(desc.StartKey)
	if err != nil {
		log.Error(err)
		return
	}

	action, priority := rq.allocator.ComputeAction(*zone, desc)
	if action != AllocatorNoop {
		return true, priority
	}
	// See if there is a rebalancing opportunity present.
	shouldRebalance := rq.allocator.ShouldRebalance(repl.store.StoreID())
	return shouldRebalance, 0
}

// SendNext invokes the specified RPC on the supplied client when the
// client is ready. On success, the reply is sent on the channel;
// otherwise an error is sent.
func (gt *grpcTransport) SendNext(done chan BatchCall) {
	client := gt.orderedClients[0]
	gt.orderedClients = gt.orderedClients[1:]

	addr := client.remoteAddr
	if log.V(2) {
		log.Infof(gt.opts.Context, "sending request to %s: %+v", addr, client.args)
	}

	if localServer := gt.rpcContext.GetLocalInternalServerForAddr(addr); enableLocalCalls && localServer != nil {
		ctx, cancel := gt.opts.contextWithTimeout()
		defer cancel()

		reply, err := localServer.Batch(ctx, &client.args)
		done <- BatchCall{Reply: reply, Err: err}
		return
	}

	go func() {
		ctx, cancel := gt.opts.contextWithTimeout()
		defer cancel()

		reply, err := client.client.Batch(ctx, &client.args)
		if reply != nil {
			for i := range reply.Responses {
				if err := reply.Responses[i].GetInner().Verify(client.args.Requests[i].GetInner()); err != nil {
					log.Error(ctx, err)
				}
			}
		}
		done <- BatchCall{Reply: reply, Err: err}
	}()
}