2 Star 2 Fork 1

cockroachdb/cockroach

加入 Gitee
与超过 1200万 开发者一起发现、参与优秀开源项目,私有仓库也完全免费 :)
免费加入
克隆/下载
status.go 21.97 KB
一键复制 编辑 原始数据 按行查看 历史
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774
// Copyright 2014 The Cockroach 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.
//
// Author: Bram Gruneir (bram+code@cockroachlabs.com)
package server
import (
"encoding/json"
"fmt"
"io"
"net/http"
"reflect"
"regexp"
"runtime"
"strconv"
"sync"
"github.com/coreos/etcd/raft"
gwruntime "github.com/grpc-ecosystem/grpc-gateway/runtime"
"github.com/pkg/errors"
"golang.org/x/net/context"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"github.com/cockroachdb/cockroach/pkg/base"
"github.com/cockroachdb/cockroach/pkg/build"
"github.com/cockroachdb/cockroach/pkg/gossip"
"github.com/cockroachdb/cockroach/pkg/internal/client"
"github.com/cockroachdb/cockroach/pkg/keys"
"github.com/cockroachdb/cockroach/pkg/roachpb"
"github.com/cockroachdb/cockroach/pkg/rpc"
"github.com/cockroachdb/cockroach/pkg/server/serverpb"
"github.com/cockroachdb/cockroach/pkg/server/status"
"github.com/cockroachdb/cockroach/pkg/storage"
"github.com/cockroachdb/cockroach/pkg/util/httputil"
"github.com/cockroachdb/cockroach/pkg/util/log"
"github.com/cockroachdb/cockroach/pkg/util/stop"
"github.com/cockroachdb/cockroach/pkg/util/syncutil"
"github.com/cockroachdb/cockroach/pkg/util/timeutil"
)
const (
// Default Maximum number of log entries returned.
defaultMaxLogEntries = 1000
// stackTraceApproxSize is the approximate size of a goroutine stack trace.
stackTraceApproxSize = 1024
// statusPrefix is the root of the cluster statistics and metrics API.
statusPrefix = "/_status/"
// statusVars exposes prometheus metrics for monitoring consumption.
statusVars = statusPrefix + "vars"
// rangeDebugEndpoint exposes an html page with information about a specific range.
rangeDebugEndpoint = "/debug/range"
// problemRangesDebugEndpoint exposes an html page with a list of all ranges
// that are experiencing problems.
problemRangesDebugEndpoint = "/debug/problemranges"
// raftStateDormant is used when there is no known raft state.
raftStateDormant = "StateDormant"
)
// Pattern for local used when determining the node ID.
var localRE = regexp.MustCompile(`(?i)local`)
type metricMarshaler interface {
json.Marshaler
PrintAsText(io.Writer) error
}
// A statusServer provides a RESTful status API.
type statusServer struct {
log.AmbientContext
db *client.DB
gossip *gossip.Gossip
metricSource metricMarshaler
nodeLiveness *storage.NodeLiveness
rpcCtx *rpc.Context
stores *storage.Stores
stopper *stop.Stopper
}
// newStatusServer allocates and returns a statusServer.
func newStatusServer(
ambient log.AmbientContext,
db *client.DB,
gossip *gossip.Gossip,
metricSource metricMarshaler,
nodeLiveness *storage.NodeLiveness,
rpcCtx *rpc.Context,
stores *storage.Stores,
stopper *stop.Stopper,
) *statusServer {
ambient.AddLogTag("status", nil)
server := &statusServer{
AmbientContext: ambient,
db: db,
gossip: gossip,
metricSource: metricSource,
nodeLiveness: nodeLiveness,
rpcCtx: rpcCtx,
stores: stores,
stopper: stopper,
}
return server
}
// RegisterService registers the GRPC service.
func (s *statusServer) RegisterService(g *grpc.Server) {
serverpb.RegisterStatusServer(g, s)
}
// RegisterGateway starts the gateway (i.e. reverse
// proxy) that proxies HTTP requests to the appropriate gRPC endpoints.
func (s *statusServer) RegisterGateway(
ctx context.Context, mux *gwruntime.ServeMux, conn *grpc.ClientConn,
) error {
ctx = s.AnnotateCtx(ctx)
return serverpb.RegisterStatusHandler(ctx, mux, conn)
}
func (s *statusServer) parseNodeID(nodeIDParam string) (roachpb.NodeID, bool, error) {
// No parameter provided or set to local.
if len(nodeIDParam) == 0 || localRE.MatchString(nodeIDParam) {
return s.gossip.NodeID.Get(), true, nil
}
id, err := strconv.ParseInt(nodeIDParam, 10, 64)
if err != nil {
return 0, false, errors.Wrap(err, "node id could not be parsed")
}
nodeID := roachpb.NodeID(id)
return nodeID, nodeID == s.gossip.NodeID.Get(), nil
}
func (s *statusServer) dialNode(nodeID roachpb.NodeID) (serverpb.StatusClient, error) {
addr, err := s.gossip.GetNodeIDAddress(nodeID)
if err != nil {
return nil, err
}
conn, err := s.rpcCtx.GRPCDial(addr.String())
if err != nil {
return nil, err
}
return serverpb.NewStatusClient(conn), nil
}
// Gossip returns gossip network status.
func (s *statusServer) Gossip(
ctx context.Context, req *serverpb.GossipRequest,
) (*gossip.InfoStatus, error) {
ctx = s.AnnotateCtx(ctx)
nodeID, local, err := s.parseNodeID(req.NodeId)
if err != nil {
return nil, grpc.Errorf(codes.InvalidArgument, err.Error())
}
if local {
infoStatus := s.gossip.GetInfoStatus()
return &infoStatus, nil
}
status, err := s.dialNode(nodeID)
if err != nil {
return nil, err
}
return status.Gossip(ctx, req)
}
// Details returns node details.
func (s *statusServer) Details(
ctx context.Context, req *serverpb.DetailsRequest,
) (*serverpb.DetailsResponse, error) {
ctx = s.AnnotateCtx(ctx)
nodeID, local, err := s.parseNodeID(req.NodeId)
if err != nil {
return nil, grpc.Errorf(codes.InvalidArgument, err.Error())
}
if local {
resp := &serverpb.DetailsResponse{
NodeID: s.gossip.NodeID.Get(),
BuildInfo: build.GetInfo(),
}
if addr, err := s.gossip.GetNodeIDAddress(s.gossip.NodeID.Get()); err == nil {
resp.Address = *addr
}
return resp, nil
}
status, err := s.dialNode(nodeID)
if err != nil {
return nil, err
}
return status.Details(ctx, req)
}
// LogFilesList returns a list of available log files.
func (s *statusServer) LogFilesList(
ctx context.Context, req *serverpb.LogFilesListRequest,
) (*serverpb.LogFilesListResponse, error) {
ctx = s.AnnotateCtx(ctx)
nodeID, local, err := s.parseNodeID(req.NodeId)
if err != nil {
return nil, grpc.Errorf(codes.InvalidArgument, err.Error())
}
if !local {
status, err := s.dialNode(nodeID)
if err != nil {
return nil, err
}
return status.LogFilesList(ctx, req)
}
log.Flush()
logFiles, err := log.ListLogFiles()
if err != nil {
return nil, err
}
return &serverpb.LogFilesListResponse{Files: logFiles}, err
}
// LogFile returns a single log file.
func (s *statusServer) LogFile(
ctx context.Context, req *serverpb.LogFileRequest,
) (*serverpb.LogEntriesResponse, error) {
ctx = s.AnnotateCtx(ctx)
nodeID, local, err := s.parseNodeID(req.NodeId)
if err != nil {
return nil, grpc.Errorf(codes.InvalidArgument, err.Error())
}
if !local {
status, err := s.dialNode(nodeID)
if err != nil {
return nil, err
}
return status.LogFile(ctx, req)
}
log.Flush()
reader, err := log.GetLogReader(req.File, true /* restricted */)
if reader == nil || err != nil {
return nil, fmt.Errorf("log file %s could not be opened: %s", req.File, err)
}
defer reader.Close()
var entry log.Entry
var resp serverpb.LogEntriesResponse
decoder := log.NewEntryDecoder(reader)
for {
if err := decoder.Decode(&entry); err != nil {
if err == io.EOF {
break
}
return nil, err
}
resp.Entries = append(resp.Entries, entry)
}
return &resp, nil
}
// parseInt64WithDefault attempts to parse the passed in string. If an empty
// string is supplied or parsing results in an error the default value is
// returned. If an error does occur during parsing, the error is returned as
// well.
func parseInt64WithDefault(s string, defaultValue int64) (int64, error) {
if len(s) == 0 {
return defaultValue, nil
}
result, err := strconv.ParseInt(s, 10, 0)
if err != nil {
return defaultValue, err
}
return result, nil
}
// Logs returns the log entries parsed from the log files stored on
// the server. Log entries are returned in reverse chronological order. The
// following options are available:
// * "starttime" query parameter filters the log entries to only ones that
// occurred on or after the "starttime". Defaults to a day ago.
// * "endtime" query parameter filters the log entries to only ones that
// occurred before on on the "endtime". Defaults to the current time.
// * "pattern" query parameter filters the log entries by the provided regexp
// pattern if it exists. Defaults to nil.
// * "max" query parameter is the hard limit of the number of returned log
// entries. Defaults to defaultMaxLogEntries.
// To filter the log messages to only retrieve messages from a given level,
// use a pattern that excludes all messages at the undesired levels.
// (e.g. "^[^IW]" to only get errors, fatals and panics). An exclusive
// pattern is better because panics and some other errors do not use
// a prefix character.
func (s *statusServer) Logs(
ctx context.Context, req *serverpb.LogsRequest,
) (*serverpb.LogEntriesResponse, error) {
ctx = s.AnnotateCtx(ctx)
nodeID, local, err := s.parseNodeID(req.NodeId)
if err != nil {
return nil, grpc.Errorf(codes.InvalidArgument, err.Error())
}
if !local {
status, err := s.dialNode(nodeID)
if err != nil {
return nil, err
}
return status.Logs(ctx, req)
}
log.Flush()
startTimestamp, err := parseInt64WithDefault(
req.StartTime,
timeutil.Now().AddDate(0, 0, -1).UnixNano())
if err != nil {
return nil, grpc.Errorf(codes.InvalidArgument, "StartTime could not be parsed: %s", err)
}
endTimestamp, err := parseInt64WithDefault(req.EndTime, timeutil.Now().UnixNano())
if err != nil {
return nil, grpc.Errorf(codes.InvalidArgument, "EndTime could not be parsed: %s", err)
}
if startTimestamp > endTimestamp {
return nil, grpc.Errorf(codes.InvalidArgument, "StartTime: %d should not be greater than endtime: %d", startTimestamp, endTimestamp)
}
maxEntries, err := parseInt64WithDefault(req.Max, defaultMaxLogEntries)
if err != nil {
return nil, grpc.Errorf(codes.InvalidArgument, "Max could not be parsed: %s", err)
}
if maxEntries < 1 {
return nil, grpc.Errorf(codes.InvalidArgument, "Max: %d should be set to a value greater than 0", maxEntries)
}
var regex *regexp.Regexp
if len(req.Pattern) > 0 {
if regex, err = regexp.Compile(req.Pattern); err != nil {
return nil, grpc.Errorf(codes.InvalidArgument, "regex pattern could not be compiled: %s", err)
}
}
entries, err := log.FetchEntriesFromFiles(startTimestamp, endTimestamp, int(maxEntries), regex)
if err != nil {
return nil, err
}
return &serverpb.LogEntriesResponse{Entries: entries}, nil
}
// TODO(tschottdorf): significant overlap with /debug/pprof/goroutine, except
// that this one allows querying by NodeID.
//
// Stacks handles returns goroutine stack traces.
func (s *statusServer) Stacks(
ctx context.Context, req *serverpb.StacksRequest,
) (*serverpb.JSONResponse, error) {
ctx = s.AnnotateCtx(ctx)
nodeID, local, err := s.parseNodeID(req.NodeId)
if err != nil {
return nil, grpc.Errorf(codes.InvalidArgument, err.Error())
}
if !local {
status, err := s.dialNode(nodeID)
if err != nil {
return nil, err
}
return status.Stacks(ctx, req)
}
bufSize := runtime.NumGoroutine() * stackTraceApproxSize
for {
buf := make([]byte, bufSize)
length := runtime.Stack(buf, true)
// If this wasn't large enough to accommodate the full set of
// stack traces, increase by 2 and try again.
if length == bufSize {
bufSize = bufSize * 2
continue
}
return &serverpb.JSONResponse{Data: buf[:length]}, nil
}
}
// Nodes returns all node statuses.
func (s *statusServer) Nodes(
ctx context.Context, req *serverpb.NodesRequest,
) (*serverpb.NodesResponse, error) {
ctx = s.AnnotateCtx(ctx)
startKey := keys.StatusNodePrefix
endKey := startKey.PrefixEnd()
b := &client.Batch{}
b.Scan(startKey, endKey)
if err := s.db.Run(ctx, b); err != nil {
log.Error(ctx, err)
return nil, grpc.Errorf(codes.Internal, err.Error())
}
rows := b.Results[0].Rows
resp := serverpb.NodesResponse{
Nodes: make([]status.NodeStatus, len(rows)),
}
for i, row := range rows {
if err := row.ValueProto(&resp.Nodes[i]); err != nil {
log.Error(ctx, err)
return nil, grpc.Errorf(codes.Internal, err.Error())
}
}
return &resp, nil
}
// handleNodeStatus handles GET requests for a single node's status.
func (s *statusServer) Node(
ctx context.Context, req *serverpb.NodeRequest,
) (*status.NodeStatus, error) {
ctx = s.AnnotateCtx(ctx)
nodeID, _, err := s.parseNodeID(req.NodeId)
if err != nil {
return nil, grpc.Errorf(codes.InvalidArgument, err.Error())
}
key := keys.NodeStatusKey(nodeID)
b := &client.Batch{}
b.Get(key)
if err := s.db.Run(ctx, b); err != nil {
log.Error(ctx, err)
return nil, grpc.Errorf(codes.Internal, err.Error())
}
var nodeStatus status.NodeStatus
if err := b.Results[0].Rows[0].ValueProto(&nodeStatus); err != nil {
err = errors.Errorf("could not unmarshal NodeStatus from %s: %s", key, err)
log.Error(ctx, err)
return nil, grpc.Errorf(codes.Internal, err.Error())
}
return &nodeStatus, nil
}
// Metrics return metrics information for the server specified.
func (s *statusServer) Metrics(
ctx context.Context, req *serverpb.MetricsRequest,
) (*serverpb.JSONResponse, error) {
ctx = s.AnnotateCtx(ctx)
nodeID, local, err := s.parseNodeID(req.NodeId)
if err != nil {
return nil, grpc.Errorf(codes.InvalidArgument, err.Error())
}
if !local {
status, err := s.dialNode(nodeID)
if err != nil {
return nil, err
}
return status.Metrics(ctx, req)
}
return marshalJSONResponse(s.metricSource)
}
// RaftDebug returns raft debug information for all known nodes.
func (s *statusServer) RaftDebug(
ctx context.Context, req *serverpb.RaftDebugRequest,
) (*serverpb.RaftDebugResponse, error) {
ctx = s.AnnotateCtx(ctx)
nodes, err := s.Nodes(ctx, nil)
if err != nil {
return nil, err
}
mu := struct {
syncutil.Mutex
resp serverpb.RaftDebugResponse
}{
resp: serverpb.RaftDebugResponse{
Ranges: make(map[roachpb.RangeID]serverpb.RaftRangeStatus),
},
}
// Subtract base.NetworkTimeout from the deadline so we have time to process
// the results and return them.
if deadline, ok := ctx.Deadline(); ok {
var cancel context.CancelFunc
ctx, cancel = context.WithDeadline(ctx, deadline.Add(-base.NetworkTimeout))
defer cancel()
}
// Parallelize fetching of ranges to minimize total time.
var wg sync.WaitGroup
for _, node := range nodes.Nodes {
wg.Add(1)
nodeID := node.Desc.NodeID
go func() {
defer wg.Done()
ranges, err := s.Ranges(ctx, &serverpb.RangesRequest{NodeId: nodeID.String(), RangeIDs: req.RangeIDs})
mu.Lock()
defer mu.Unlock()
if err != nil {
err := errors.Wrapf(err, "failed to get ranges from %d", nodeID)
mu.resp.Errors = append(mu.resp.Errors, serverpb.RaftRangeError{Message: err.Error()})
return
}
for _, rng := range ranges.Ranges {
rangeID := rng.State.Desc.RangeID
status, ok := mu.resp.Ranges[rangeID]
if !ok {
status = serverpb.RaftRangeStatus{
RangeID: rangeID,
}
}
status.Nodes = append(status.Nodes, serverpb.RaftRangeNode{
NodeID: nodeID,
Range: rng,
})
mu.resp.Ranges[rangeID] = status
}
}()
}
wg.Wait()
mu.Lock()
defer mu.Unlock()
// Check for errors.
for i, rng := range mu.resp.Ranges {
for j, node := range rng.Nodes {
desc := node.Range.State.Desc
// Check for whether replica should be GCed.
containsNode := false
for _, replica := range desc.Replicas {
if replica.NodeID == node.NodeID {
containsNode = true
}
}
if !containsNode {
rng.Errors = append(rng.Errors, serverpb.RaftRangeError{
Message: fmt.Sprintf("node %d not in range descriptor and should be GCed", node.NodeID),
})
}
// Check for replica descs not matching.
if j > 0 {
prevDesc := rng.Nodes[j-1].Range.State.Desc
if !reflect.DeepEqual(&desc, &prevDesc) {
prevNodeID := rng.Nodes[j-1].NodeID
rng.Errors = append(rng.Errors, serverpb.RaftRangeError{
Message: fmt.Sprintf("node %d range descriptor does not match node %d", node.NodeID, prevNodeID),
})
}
}
mu.resp.Ranges[i] = rng
}
}
return &mu.resp, nil
}
func (s *statusServer) handleVars(w http.ResponseWriter, r *http.Request) {
w.Header().Set(httputil.ContentTypeHeader, httputil.PlaintextContentType)
err := s.metricSource.PrintAsText(w)
if err != nil {
log.Error(r.Context(), err)
http.Error(w, err.Error(), http.StatusInternalServerError)
}
}
// Ranges returns range info for the specified node.
func (s *statusServer) Ranges(
ctx context.Context, req *serverpb.RangesRequest,
) (*serverpb.RangesResponse, error) {
ctx = s.AnnotateCtx(ctx)
nodeID, local, err := s.parseNodeID(req.NodeId)
if err != nil {
return nil, grpc.Errorf(codes.InvalidArgument, err.Error())
}
if !local {
status, err := s.dialNode(nodeID)
if err != nil {
return nil, err
}
return status.Ranges(ctx, req)
}
output := serverpb.RangesResponse{
Ranges: make([]serverpb.RangeInfo, 0, s.stores.GetStoreCount()),
}
convertRaftStatus := func(raftStatus *raft.Status) serverpb.RaftState {
var state serverpb.RaftState
if raftStatus == nil {
state.State = raftStateDormant
return state
}
state.ReplicaID = raftStatus.ID
state.HardState = raftStatus.HardState
state.Applied = raftStatus.Applied
// Grab Lead and State, which together form the SoftState.
state.Lead = raftStatus.Lead
state.State = raftStatus.RaftState.String()
state.Progress = make(map[uint64]serverpb.RaftState_Progress)
for id, progress := range raftStatus.Progress {
state.Progress[id] = serverpb.RaftState_Progress{
Match: progress.Match,
Next: progress.Next,
Paused: progress.Paused,
PendingSnapshot: progress.PendingSnapshot,
State: progress.State.String(),
}
}
return state
}
constructRangeInfo := func(
desc roachpb.RangeDescriptor, rep *storage.Replica, storeID roachpb.StoreID, leaseHistory []roachpb.Lease, metrics storage.ReplicaMetrics,
) serverpb.RangeInfo {
return serverpb.RangeInfo{
Span: serverpb.PrettySpan{
StartKey: desc.StartKey.String(),
EndKey: desc.EndKey.String(),
},
RaftState: convertRaftStatus(rep.RaftStatus()),
State: rep.State(),
SourceNodeID: nodeID,
SourceStoreID: storeID,
LeaseHistory: leaseHistory,
Problems: serverpb.RangeProblems{
Unavailable: metrics.RangeCounter && metrics.Unavailable,
LeaderNotLeaseHolder: metrics.Leader && metrics.LeaseValid && !metrics.Leaseholder,
},
}
}
cfg, ok := s.gossip.GetSystemConfig()
if !ok {
return nil, grpc.Errorf(codes.Internal, "system config not yet available")
}
isLiveMap := s.nodeLiveness.GetIsLiveMap()
err = s.stores.VisitStores(func(store *storage.Store) error {
timestamp := store.Clock().Now()
if len(req.RangeIDs) == 0 {
// All ranges requested.
// Use IterateRangeDescriptors to read from the engine only
// because it's already exported.
err := storage.IterateRangeDescriptors(ctx, store.Engine(),
func(desc roachpb.RangeDescriptor) (bool, error) {
rep, err := store.GetReplica(desc.RangeID)
if err != nil {
return true, err
}
output.Ranges = append(output.Ranges,
constructRangeInfo(
desc,
rep,
store.Ident.StoreID,
rep.GetLeaseHistory(),
rep.Metrics(ctx, timestamp, cfg, isLiveMap),
))
return false, nil
})
return err
}
// Specific ranges requested:
for _, rid := range req.RangeIDs {
rep, err := store.GetReplica(rid)
if err != nil {
// Not found: continue.
continue
}
desc := rep.Desc()
output.Ranges = append(output.Ranges,
constructRangeInfo(
*desc,
rep,
store.Ident.StoreID,
rep.GetLeaseHistory(),
rep.Metrics(ctx, timestamp, cfg, isLiveMap),
))
}
return nil
})
if err != nil {
return nil, grpc.Errorf(codes.Internal, err.Error())
}
return &output, nil
}
// SpanStats requests the total statistics stored on a node for a given key
// span, which may include multiple ranges.
func (s *statusServer) SpanStats(
ctx context.Context, req *serverpb.SpanStatsRequest,
) (*serverpb.SpanStatsResponse, error) {
ctx = s.AnnotateCtx(ctx)
nodeID, local, err := s.parseNodeID(req.NodeID)
if err != nil {
return nil, grpc.Errorf(codes.InvalidArgument, err.Error())
}
if !local {
status, err := s.dialNode(nodeID)
if err != nil {
return nil, err
}
return status.SpanStats(ctx, req)
}
output := &serverpb.SpanStatsResponse{}
err = s.stores.VisitStores(func(store *storage.Store) error {
stats, count := store.ComputeStatsForKeySpan(req.StartKey.Next(), req.EndKey)
output.TotalStats.Add(stats)
output.RangeCount += int32(count)
return nil
})
if err != nil {
return nil, err
}
return output, nil
}
// jsonWrapper provides a wrapper on any slice data type being
// marshaled to JSON. This prevents a security vulnerability
// where a phishing attack can trick a user's browser into
// requesting a document from Cockroach as an executable script,
// allowing the contents of the fetched document to be treated
// as executable javascript. More details here:
// http://haacked.com/archive/2009/06/25/json-hijacking.aspx/
type jsonWrapper struct {
Data interface{} `json:"d"`
}
// marshalToJSON marshals the given value into nicely indented JSON. If the
// value is an array or slice it is wrapped in jsonWrapper and then marshalled.
func marshalToJSON(value interface{}) ([]byte, error) {
switch reflect.ValueOf(value).Kind() {
case reflect.Array, reflect.Slice:
value = jsonWrapper{Data: value}
}
body, err := json.MarshalIndent(value, "", " ")
if err != nil {
return nil, errors.Errorf("unable to marshal %+v to json: %s", value, err)
}
return body, nil
}
// marshalJSONResponse converts an arbitrary value into a JSONResponse protobuf
// that can be sent via grpc.
func marshalJSONResponse(value interface{}) (*serverpb.JSONResponse, error) {
data, err := marshalToJSON(value)
if err != nil {
return nil, err
}
return &serverpb.JSONResponse{Data: data}, nil
}
马建仓 AI 助手
尝试更多
代码解读
代码找茬
代码优化
1
https://gitee.com/mirrors_cockroachdb/cockroach.git
git@gitee.com:mirrors_cockroachdb/cockroach.git
mirrors_cockroachdb
cockroach
cockroach
v1.0.6

搜索帮助