couchbase / cbft / 159

pending completion

Build # 159

Build Type

push

travis-ci

Committed by

steveyen

Commit Message

<a href="https://github.com/couchbase/cbft/commit/<a class=hub.com/couchbase/cbft/commit/8cce8983e6b77b664ef4660fe79f833a1ebec5b9">8cce8983e<a href="https://github.com/couchbase/cbft/commit/8cce8983e6b77b664ef4660fe79f833a1ebec5b9">&quot;&gt;MB-18002 - added total_queries_timeout stat to ns_server.go

Change-Id: I</a><a class="double-link" href="https://github.com/couchbase/cbft/commit/<a class="double-link" href="https://github.com/couchbase/cbft/commit/f47ef7f8b723c6ad0cdc2d5bd73bd6866fc82cfe">f47ef7f8b</a>">f47ef7f8b</a><a href="https://github.com/couchbase/cbft/commit/8cce8983e6b77b664ef4660fe79f833a1ebec5b9">
Reviewed-on: http://review.couchbase.org/60963
Reviewed-by: Marty Schoch <marty.schoch@gmail.com>
Tested-by: Steve Yen <steve.yen@gmail.com>

Run Details

3 of 3 new or added lines in 2 files covered. (100.0%)

1800 of 2322 relevant lines covered (77.52%)

4.81 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

51.54

/ns_server.go

//  Copyright (c) 2014 Couchbase, Inc.
//  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 cbft

import (
        "bytes"
        "encoding/json"
        "fmt"
        "net/http"
        "net/url"
        "runtime"
        "sort"
        "strings"
        "sync/atomic"
        "unicode"

        "github.com/couchbase/cbgt"
        "github.com/couchbase/cbgt/rest"
        "github.com/dustin/go-jsonpointer"
)

var statsNamePrefix = []byte("\"")
var statsNameSuffix = []byte("\":")

// NsStatsHandler is a REST handler that provides stats/metrics for
// consumption by ns_server
type NsStatsHandler struct {
        mgr *cbgt.Manager
}

func NewNsStatsHandler(mgr *cbgt.Manager) *NsStatsHandler {
        return &NsStatsHandler{mgr: mgr}
}

type NSIndexStats map[string]map[string]interface{}

// MarshalJSON formats the index stats using the
// colon separated convention found in other
// stats consumed by ns_server
func (n NSIndexStats) MarshalJSON() ([]byte, error) {
        rv := make(map[string]interface{})
        for k, nsis := range n {
                for nsik, nsiv := range nsis {
                        if k == "" {
                                rv[nsik] = nsiv
                        } else {
                                rv[k+":"+nsik] = nsiv
                        }
                }
        }
        return json.Marshal(rv)
}

var statkeys = []string{
        // manual
        "num_pindexes",

        // pindex
        "doc_count",
        "timer_batch_store_count",

        // kv store
        "timer_batch_merge_count",
        "timer_iterator_next_count",
        "timer_iterator_seek_count",
        "timer_reader_get_count",
        "timer_reader_multi_get_count",
        "timer_reader_prefix_iterator_count",
        "timer_reader_range_iterator_count",
        "timer_writer_execute_batch_count",

        // feed
        "timer_opaque_set_count",
        "timer_rollback_count",
        "timer_data_update_count",
        "timer_data_delete_count",
        "timer_snapshot_start_count",
        "timer_opaque_get_count",

        // --------------------------------------------
        // stats from "FTS Stats" spec, see:
        // https://docs.google.com/spreadsheets/d/1w8P68gLBIs0VUN4egUuUH6U_92_5xi9azfvH8pPw21s/edit#gid=104567684

        "num_mutations_to_index",
        // "doc_count", // per-index stat (same as before, see above).
        "total_bytes_indexed",
        "num_recs_to_persist",

        "num_bytes_used_disk",
        // "num_bytes_used_ram" -- PROCESS-LEVEL stat.

        "num_pindexes_actual", // per-index stat.
        "num_pindexes_target", // per-index stat.

        "total_compactions",

        // "total_gc" -- PROCESS-LEVEL stat.
        // "pct_cpu_gc" -- PROCESS-LEVEL stat.

        "total_queries",             // per-index stat.
        "avg_queries_latency",       // per-index stat.
        "total_queries_slow",        // per-index stat.
        "total_queries_timeout",     // per-index stat.
        "total_queries_error",       // per-index stat.
        "total_bytes_query_results", // per-index stat.
        "total_term_searchers",
}

// NewIndexStat ensures that all index stats
// have the same shape and 0 values to
// prevent seeing N/A in ns_server UI
func NewIndexStat() map[string]interface{} {
        rv := make(map[string]interface{})
        for _, key := range statkeys {
                rv[key] = float64(0)
        }
        return rv
}

func (h *NsStatsHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
        _, indexDefsMap, err := h.mgr.GetIndexDefs(false)
        if err != nil {
                rest.ShowError(w, req, fmt.Sprintf("could not retrieve index defs: %v", err), 500)
                return
        }

        planPIndexes, _, err := cbgt.CfgGetPlanPIndexes(h.mgr.Cfg())
        if err != nil {
                rest.ShowError(w, req, fmt.Sprintf("could not retrieve plan pIndexes: %v", err), 500)
                return
        }

        nodeUUID := h.mgr.UUID()

        nsIndexStats := make(NSIndexStats, len(indexDefsMap))

        indexNameToSourceName := map[string]string{}

        indexNameToPlanPIndexes := map[string][]*cbgt.PlanPIndex{}
        for _, planPIndex := range planPIndexes.PlanPIndexes {
                // Only focus on the planPIndex entries for this node.
                if planPIndex.Nodes[nodeUUID] != nil {
                        indexNameToPlanPIndexes[planPIndex.IndexName] =
                                append(indexNameToPlanPIndexes[planPIndex.IndexName], planPIndex)
                }
        }

        indexQueryPathStats := MapRESTPathStats[RESTIndexQueryPath]

        for indexName, indexDef := range indexDefsMap {
                nsIndexStat := NewIndexStat()
                nsIndexStats[indexDef.SourceName+":"+indexName] = nsIndexStat

                indexNameToSourceName[indexName] = indexDef.SourceName

                focusStats := indexQueryPathStats.FocusStats(indexName)
                if focusStats != nil {
                        totalQueries := atomic.LoadUint64(&focusStats.TotRequest)
                        nsIndexStat["total_queries"] = totalQueries
                        if totalQueries > 0 {
                                nsIndexStat["avg_queries_latency"] =
                                        float64((atomic.LoadUint64(&focusStats.TotRequestTimeNS) /
                                                totalQueries)) / 1000000.0 // Convert from nanosecs to millisecs.
                        }
                        nsIndexStat["total_queries_slow"] =
                                atomic.LoadUint64(&focusStats.TotRequestSlow)
                        nsIndexStat["total_queries_timeout"] =
                                atomic.LoadUint64(&focusStats.TotRequestTimeout)
                        nsIndexStat["total_queries_error"] =
                                atomic.LoadUint64(&focusStats.TotRequestErr)
                        nsIndexStat["total_bytes_query_results"] =
                                atomic.LoadUint64(&focusStats.TotResponseBytes)
                        nsIndexStat["num_pindexes_target"] =
                                uint64(len(indexNameToPlanPIndexes[indexName]))
                }
        }

        feeds, pindexes := h.mgr.CurrentMaps()

        for _, pindex := range pindexes {
                nsIndexName := pindex.SourceName + ":" + pindex.IndexName
                nsIndexStat, ok := nsIndexStats[nsIndexName]
                if ok {
                        // manually track num pindexes
                        oldValue, ok := nsIndexStat["num_pindexes_actual"]
                        if ok {
                                switch oldValue := oldValue.(type) {
                                case float64:
                                        oldValue += float64(1)

                                        nsIndexStat["num_pindexes_actual"] = oldValue

                                        // TODO: Former name was num_pindexes, need to remove one day.
                                        nsIndexStat["num_pindexes"] = oldValue
                                }
                        }

                        // automatically process all the pindex dest stats
                        err := addPIndexStats(pindex, nsIndexStat)
                        if err != nil {
                                rest.ShowError(w, req,
                                        fmt.Sprintf("error processing PIndex stats: %v", err), 500)
                                return
                        }
                }
        }

        for _, feed := range feeds {
                sourceName := indexNameToSourceName[feed.IndexName()]
                nsIndexName := sourceName + ":" + feed.IndexName()
                nsIndexStat, ok := nsIndexStats[nsIndexName]
                if ok {
                        // automatically process all the feed stats
                        err := addFeedStats(feed, nsIndexStat)
                        if err != nil {
                                rest.ShowError(w, req,
                                        fmt.Sprintf("error processing Feed stats: %v", err), 500)
                                return
                        }
                }
        }

        // FIXME hard-coded top-level stats
        topLevelStats := map[string]interface{}{}
        topLevelStats["num_connections"] = 0
        topLevelStats["needs_restart"] = false

        memStats := &runtime.MemStats{}
        runtime.ReadMemStats(memStats)

        topLevelStats["num_bytes_used_ram"] = memStats.Alloc
        topLevelStats["total_gc"] = memStats.NumGC
        topLevelStats["pct_cpu_gc"] = memStats.GCCPUFraction

        nsIndexStats[""] = topLevelStats

        rest.MustEncode(w, nsIndexStats)
}

func addFeedStats(feed cbgt.Feed, nsIndexStat map[string]interface{}) error {
        buffer := new(bytes.Buffer)
        err := feed.Stats(buffer)
        if err != nil {
                return err
        }
        return massageStats(buffer, nsIndexStat)
}

func addPIndexStats(pindex *cbgt.PIndex, nsIndexStat map[string]interface{}) error {
        buffer := new(bytes.Buffer)
        err := pindex.Dest.Stats(buffer)
        if err != nil {
                return err
        }
        return massageStats(buffer, nsIndexStat)
}

func massageStats(buffer *bytes.Buffer, nsIndexStat map[string]interface{}) error {
        statsBytes := buffer.Bytes()
        pointers, err := jsonpointer.ListPointers(statsBytes)
        if err != nil {
                return err
        }

        countPointers := make([]string, 0)
        for _, pointer := range pointers {
                if strings.HasSuffix(pointer, "/count") ||
                        strings.HasSuffix(pointer, "/DocCount") {
                        countPointers = append(countPointers, pointer)
                }
        }

        countValueMap, err := jsonpointer.FindMany(statsBytes, countPointers)
        if err != nil {
                return err
        }

        for k, v := range countValueMap {
                statName := convertStatName(k)
                var statValue float64
                err := json.Unmarshal(v, &statValue)
                if err != nil {
                        return err
                }
                oldValue, ok := nsIndexStat[statName]
                if ok {
                        switch oldValue := oldValue.(type) {
                        case float64:
                                oldValue += statValue
                                nsIndexStat[statName] = oldValue
                        }
                }
        }

        return nil
}

func convertStatName(key string) string {
        lastSlash := strings.LastIndex(key, "/")
        if lastSlash < 0 {
                return "unknown"
        }
        statSuffix := key[lastSlash+1:]
        statNameStart := strings.LastIndex(key[:lastSlash], "/")
        if statNameStart < 0 {
                return "unknown"
        }
        statPrefix := key[statNameStart+1 : lastSlash]
        if statPrefix == "basic" {
                statPrefix = ""
        }
        statName := camelToUnderscore(statSuffix)
        if statPrefix != "" {
                statName = camelToUnderscore(statPrefix) + "_" + statName
        }
        return statName
}

func camelToUnderscore(name string) string {
        rv := ""
        for i, r := range name {
                if unicode.IsUpper(r) && i != 0 {
                        rv += "_"
                        rv += string(unicode.ToLower(r))
                } else if unicode.IsUpper(r) && i == 0 {
                        rv += string(unicode.ToLower(r))
                } else {
                        rv += string(r)
                }
        }
        return rv
}

// ---------------------------------------------------

// NsStatusHandler is a REST handler that provides status for
// consumption by ns_server
type NsStatusHandler struct {
        mgr       *cbgt.Manager
        serverURL *url.URL
}

func NewNsStatusHandler(mgr *cbgt.Manager, server string) (*NsStatusHandler, error) {
        serverURL, err := url.Parse(server)
        if err != nil {
                return nil, err
        }

        return &NsStatusHandler{
                mgr:       mgr,
                serverURL: serverURL,
        }, nil
}

func NsHostsForIndex(name string, planPIndexes *cbgt.PlanPIndexes,
        nodeDefs *cbgt.NodeDefs) []string {
        if planPIndexes == nil {
                return nil
        }

        v := struct{}{}

        // find the node UUIDs related to this index
        nodes := make(map[string]struct{})
        for _, planPIndex := range planPIndexes.PlanPIndexes {
                if planPIndex.IndexName == name {
                        for planPIndexNodeUUID := range planPIndex.Nodes {
                                nodes[planPIndexNodeUUID] = v
                        }
                }
        }

        // look for all these nodes in the nodes wanted
        nodeExtras := make(map[string]string)
        for _, nodeDef := range nodeDefs.NodeDefs {
                _, ok := nodes[nodeDef.UUID]
                if ok {
                        var e struct {
                                NsHostPort string `json:"nsHostPort"`
                        }

                        err := json.Unmarshal([]byte(nodeDef.Extras), &e)
                        if err != nil {
                                // Early versions of ns_server integration had a
                                // simple, non-JSON "host:port" format for
                                // nodeDef.Extras, so fall back to that.
                                e.NsHostPort = nodeDef.Extras
                        }

                        nodeExtras[nodeDef.UUID] = e.NsHostPort
                }
        }

        // build slice of node extras
        nodeStrings := make(sort.StringSlice, 0, len(nodeExtras))
        for _, nodeExtra := range nodeExtras {
                nodeStrings = append(nodeStrings, nodeExtra)
        }

        // sort slice for stability
        sort.Sort(nodeStrings)

        return nodeStrings
}

func (h *NsStatusHandler) ServeHTTP(
        w http.ResponseWriter, req *http.Request) {

        cfg := h.mgr.Cfg()
        planPIndexes, _, err := cbgt.CfgGetPlanPIndexes(cfg)
        if err != nil {
                rest.ShowError(w, req, fmt.Sprintf("could not retrieve plan pIndexes: %v", err), 500)
                return
        }

        nodesDefs, _, err := cbgt.CfgGetNodeDefs(cfg, cbgt.NODE_DEFS_WANTED)
        if err != nil {
                rest.ShowError(w, req, "could not retrieve node defs (wanted)", 500)
                return
        }

        _, indexDefsMap, err := h.mgr.GetIndexDefs(false)
        if err != nil {
                rest.ShowError(w, req, "could not retrieve index defs", 500)
                return
        }

        w.Header().Set("Content-Type", "application/json; charset=utf-8")
        w.Write(cbgt.JsonOpenBrace)
        w.Write(statsNamePrefix)
        w.Write([]byte("status"))
        w.Write(statsNameSuffix)
        w.Write([]byte("["))

        indexDefNames := make(sort.StringSlice, 0, len(indexDefsMap))
        for indexDefName := range indexDefsMap {
                indexDefNames = append(indexDefNames, indexDefName)
        }

        sort.Sort(indexDefNames)

        for i, indexDefName := range indexDefNames {
                indexDef := indexDefsMap[indexDefName]
                if i > 0 {
                        w.Write(cbgt.JsonComma)
                }

                rest.MustEncode(w, struct {
                        Completion int      `json:"completion"`
                        Hosts      []string `json:"hosts"`
                        Status     string   `json:"status"`
                        Bucket     string   `json:"bucket"`
                        Name       string   `json:"name"`
                }{
                        Bucket: indexDef.SourceName,
                        Name:   indexDefName,
                        Hosts:  NsHostsForIndex(indexDefName, planPIndexes, nodesDefs),
                        // FIXME hard-coded
                        Completion: 100,
                        Status:     "Ready",
                })
        }

        w.Write([]byte("],"))
        w.Write(statsNamePrefix)
        w.Write([]byte("code"))
        w.Write(statsNameSuffix)
        w.Write([]byte("\"success\""))
        w.Write(cbgt.JsonCloseBrace)
}

1	// Copyright (c) 2014 Couchbase, Inc.
2	// Licensed under the Apache License, Version 2.0 (the "License");
3	// you may not use this file except in compliance with the
4	// License. You may obtain a copy of the License at
5	// http://www.apache.org/licenses/LICENSE-2.0
6	// Unless required by applicable law or agreed to in writing,
7	// software distributed under the License is distributed on an "AS
8	// IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
9	// express or implied. See the License for the specific language
10	// governing permissions and limitations under the License.
11
12	package cbft
13
14	import (
15	"bytes"
16	"encoding/json"
17	"fmt"
18	"net/http"
19	"net/url"
20	"runtime"
21	"sort"
22	"strings"
23	"sync/atomic"
24	"unicode"
25
26	"github.com/couchbase/cbgt"
27	"github.com/couchbase/cbgt/rest"
28	"github.com/dustin/go-jsonpointer"
29	)
30
31	var statsNamePrefix = []byte("\"")
32	var statsNameSuffix = []byte("\":")
33
34	// NsStatsHandler is a REST handler that provides stats/metrics for
35	// consumption by ns_server
36	type NsStatsHandler struct {
37	mgr *cbgt.Manager
38	}
39
40	func NewNsStatsHandler(mgr cbgt.Manager) NsStatsHandler {	1✔
41	return &NsStatsHandler{mgr: mgr}	×
42	}	1✔
43
44	type NSIndexStats map[string]map[string]interface{}
45
46	// MarshalJSON formats the index stats using the
47	// colon separated convention found in other
48	// stats consumed by ns_server
49	func (n NSIndexStats) MarshalJSON() ([]byte, error) {	1✔
50	rv := make(map[string]interface{})	×
51	for k, nsis := range n {	×
52	for nsik, nsiv := range nsis {	×
53	if k == "" {	×
54	rv[nsik] = nsiv	×
55	} else {	1✔
56	rv[k+":"+nsik] = nsiv	×
57	}	1✔
58	}	1✔
59	}	1✔
60	return json.Marshal(rv)	×
61	}	1✔
62
63	var statkeys = []string{
64	// manual
65	"num_pindexes",
66
67	// pindex
68	"doc_count",
69	"timer_batch_store_count",
70
71	// kv store
72	"timer_batch_merge_count",
73	"timer_iterator_next_count",
74	"timer_iterator_seek_count",
75	"timer_reader_get_count",
76	"timer_reader_multi_get_count",
77	"timer_reader_prefix_iterator_count",
78	"timer_reader_range_iterator_count",
79	"timer_writer_execute_batch_count",
80
81	// feed
82	"timer_opaque_set_count",
83	"timer_rollback_count",
84	"timer_data_update_count",
85	"timer_data_delete_count",
86	"timer_snapshot_start_count",
87	"timer_opaque_get_count",
88
89	// --------------------------------------------
90	// stats from "FTS Stats" spec, see:
91	// https://docs.google.com/spreadsheets/d/1w8P68gLBIs0VUN4egUuUH6U_92_5xi9azfvH8pPw21s/edit#gid=104567684
92
93	"num_mutations_to_index",
94	// "doc_count", // per-index stat (same as before, see above).
95	"total_bytes_indexed",
96	"num_recs_to_persist",
97
98	"num_bytes_used_disk",
99	// "num_bytes_used_ram" -- PROCESS-LEVEL stat.
100
101	"num_pindexes_actual", // per-index stat.
102	"num_pindexes_target", // per-index stat.
103
104	"total_compactions",
105
106	// "total_gc" -- PROCESS-LEVEL stat.
107	// "pct_cpu_gc" -- PROCESS-LEVEL stat.
108
109	"total_queries", // per-index stat.
110	"avg_queries_latency", // per-index stat.
111	"total_queries_slow", // per-index stat.
112	"total_queries_timeout", // per-index stat.
113	"total_queries_error", // per-index stat.
114	"total_bytes_query_results", // per-index stat.
115	"total_term_searchers",
116	}
117
118	// NewIndexStat ensures that all index stats
119	// have the same shape and 0 values to
120	// prevent seeing N/A in ns_server UI
121	func NewIndexStat() map[string]interface{} {	1✔
122	rv := make(map[string]interface{})	×
123	for _, key := range statkeys {	×
124	rv[key] = float64(0)	×
125	}	1✔
126	return rv	×
127	}	1✔
128
129	func (h NsStatsHandler) ServeHTTP(w http.ResponseWriter, req http.Request) {	1✔
130	_, indexDefsMap, err := h.mgr.GetIndexDefs(false)	×
131	if err != nil {	×
132	rest.ShowError(w, req, fmt.Sprintf("could not retrieve index defs: %v", err), 500)	×
133	return	×
134	}	1✔
135		1✔
136	planPIndexes, _, err := cbgt.CfgGetPlanPIndexes(h.mgr.Cfg())	×
137	if err != nil {	×
138	rest.ShowError(w, req, fmt.Sprintf("could not retrieve plan pIndexes: %v", err), 500)	×
139	return	×
140	}	1✔
141		1✔
142	nodeUUID := h.mgr.UUID()	×
143		1✔
144	nsIndexStats := make(NSIndexStats, len(indexDefsMap))	×
145		1✔
146	indexNameToSourceName := map[string]string{}	×
147		1✔
148	indexNameToPlanPIndexes := map[string][]*cbgt.PlanPIndex{}	×
149	for _, planPIndex := range planPIndexes.PlanPIndexes {	×
150	// Only focus on the planPIndex entries for this node.	1✔
151	if planPIndex.Nodes[nodeUUID] != nil {	×
152	indexNameToPlanPIndexes[planPIndex.IndexName] =	×
153	append(indexNameToPlanPIndexes[planPIndex.IndexName], planPIndex)	1✔
154	}	1✔
155	}	1✔
156		1✔
157	indexQueryPathStats := MapRESTPathStats[RESTIndexQueryPath]	×
158		1✔
159	for indexName, indexDef := range indexDefsMap {	×
160	nsIndexStat := NewIndexStat()	×
161	nsIndexStats[indexDef.SourceName+":"+indexName] = nsIndexStat	×
162		1✔
163	indexNameToSourceName[indexName] = indexDef.SourceName	×
164		1✔
165	focusStats := indexQueryPathStats.FocusStats(indexName)	×
166	if focusStats != nil {	×
167	totalQueries := atomic.LoadUint64(&focusStats.TotRequest)	×
168	nsIndexStat["total_queries"] = totalQueries	×
169	if totalQueries > 0 {	×
170	nsIndexStat["avg_queries_latency"] =	×
171	float64((atomic.LoadUint64(&focusStats.TotRequestTimeNS) /	1✔
172	totalQueries)) / 1000000.0 // Convert from nanosecs to millisecs.	1✔
173	}	1✔
174	nsIndexStat["total_queries_slow"] =	×
175	atomic.LoadUint64(&focusStats.TotRequestSlow)	1✔
176	nsIndexStat["total_queries_timeout"] =	×
177	atomic.LoadUint64(&focusStats.TotRequestTimeout)	1✔
178	nsIndexStat["total_queries_error"] =	×
179	atomic.LoadUint64(&focusStats.TotRequestErr)	1✔
180	nsIndexStat["total_bytes_query_results"] =	×
181	atomic.LoadUint64(&focusStats.TotResponseBytes)	1✔
182	nsIndexStat["num_pindexes_target"] =	×
183	uint64(len(indexNameToPlanPIndexes[indexName]))	1✔
184	}	1✔
185	}	1✔
186		1✔
187	feeds, pindexes := h.mgr.CurrentMaps()	×
188		1✔
189	for _, pindex := range pindexes {	×
190	nsIndexName := pindex.SourceName + ":" + pindex.IndexName	×
191	nsIndexStat, ok := nsIndexStats[nsIndexName]	×
192	if ok {	×
193	// manually track num pindexes	1✔
194	oldValue, ok := nsIndexStat["num_pindexes_actual"]	×
195	if ok {	×
196	switch oldValue := oldValue.(type) {	×
197	case float64:	1✔
198	oldValue += float64(1)	×
199		1✔
200	nsIndexStat["num_pindexes_actual"] = oldValue	×
201		1✔
202	// TODO: Former name was num_pindexes, need to remove one day.	1✔
203	nsIndexStat["num_pindexes"] = oldValue	×
204	}	1✔
205	}	1✔
206		1✔
207	// automatically process all the pindex dest stats	1✔
208	err := addPIndexStats(pindex, nsIndexStat)	×
209	if err != nil {	×
210	rest.ShowError(w, req,	×
211	fmt.Sprintf("error processing PIndex stats: %v", err), 500)	1✔
212	return	×
213	}	1✔
214	}	1✔
215	}	1✔
216		1✔
217	for _, feed := range feeds {	×
218	sourceName := indexNameToSourceName[feed.IndexName()]	×
219	nsIndexName := sourceName + ":" + feed.IndexName()	×
220	nsIndexStat, ok := nsIndexStats[nsIndexName]	×
221	if ok {	×
222	// automatically process all the feed stats	1✔
223	err := addFeedStats(feed, nsIndexStat)	×
224	if err != nil {	×
225	rest.ShowError(w, req,	×
226	fmt.Sprintf("error processing Feed stats: %v", err), 500)	1✔
227	return	×
228	}	1✔
229	}	1✔
230	}	1✔
231		1✔
232	// FIXME hard-coded top-level stats	1✔
233	topLevelStats := map[string]interface{}{}	×
234	topLevelStats["num_connections"] = 0	×
235	topLevelStats["needs_restart"] = false	×
236		1✔
237	memStats := &runtime.MemStats{}	×
238	runtime.ReadMemStats(memStats)	×
239		1✔
240	topLevelStats["num_bytes_used_ram"] = memStats.Alloc	×
241	topLevelStats["total_gc"] = memStats.NumGC	×
242	topLevelStats["pct_cpu_gc"] = memStats.GCCPUFraction	×
243		1✔
244	nsIndexStats[""] = topLevelStats	×
245		1✔
246	rest.MustEncode(w, nsIndexStats)	×
247	}	1✔
248
249	func addFeedStats(feed cbgt.Feed, nsIndexStat map[string]interface{}) error {	1✔
250	buffer := new(bytes.Buffer)	×
251	err := feed.Stats(buffer)	×
252	if err != nil {	×
253	return err	×
254	}	1✔
255	return massageStats(buffer, nsIndexStat)	×
256	}	1✔
257
258	func addPIndexStats(pindex *cbgt.PIndex, nsIndexStat map[string]interface{}) error {	1✔
259	buffer := new(bytes.Buffer)	×
260	err := pindex.Dest.Stats(buffer)	×
261	if err != nil {	×
262	return err	×
263	}	1✔
264	return massageStats(buffer, nsIndexStat)	×
265	}	1✔
266
267	func massageStats(buffer *bytes.Buffer, nsIndexStat map[string]interface{}) error {	1✔
268	statsBytes := buffer.Bytes()	×
269	pointers, err := jsonpointer.ListPointers(statsBytes)	×
270	if err != nil {	×
271	return err	×
272	}	1✔
273		1✔
274	countPointers := make([]string, 0)	×
275	for _, pointer := range pointers {	×
276	if strings.HasSuffix(pointer, "/count") \|\|	×
277	strings.HasSuffix(pointer, "/DocCount") {	1✔
278	countPointers = append(countPointers, pointer)	×
279	}	1✔
280	}	1✔
281		1✔
282	countValueMap, err := jsonpointer.FindMany(statsBytes, countPointers)	×
283	if err != nil {	×
284	return err	×
285	}	1✔
286		1✔
287	for k, v := range countValueMap {	×
288	statName := convertStatName(k)	×
289	var statValue float64	×
290	err := json.Unmarshal(v, &statValue)	×
291	if err != nil {	×
292	return err	×
293	}	1✔
294	oldValue, ok := nsIndexStat[statName]	×
295	if ok {	×
296	switch oldValue := oldValue.(type) {	×
297	case float64:	1✔
298	oldValue += statValue	×
299	nsIndexStat[statName] = oldValue	×
300	}	1✔
301	}	1✔
302	}	1✔
303		1✔
304	return nil	×
305	}	1✔
306
307	func convertStatName(key string) string {	1✔
308	lastSlash := strings.LastIndex(key, "/")	1✔
309	if lastSlash < 0 {	1✔
310	return "unknown"	×
311	}	1✔
312	statSuffix := key[lastSlash+1:]	1✔
313	statNameStart := strings.LastIndex(key[:lastSlash], "/")	1✔
314	if statNameStart < 0 {	1✔
315	return "unknown"	×
316	}	1✔
317	statPrefix := key[statNameStart+1 : lastSlash]	1✔
318	if statPrefix == "basic" {	1✔
319	statPrefix = ""	×
320	}	1✔
321	statName := camelToUnderscore(statSuffix)	1✔
322	if statPrefix != "" {	1✔
323	statName = camelToUnderscore(statPrefix) + "_" + statName	1✔
324	}	1✔
325	return statName	1✔
326	}	1✔
327
328	func camelToUnderscore(name string) string {	1✔
329	rv := ""	3✔
330	for i, r := range name {	3✔
331	if unicode.IsUpper(r) && i != 0 {	17✔
332	rv += "_"	3✔
333	rv += string(unicode.ToLower(r))	3✔
334	} else if unicode.IsUpper(r) && i == 0 {	14✔
335	rv += string(unicode.ToLower(r))	1✔
336	} else {	1✔
337	rv += string(r)	13✔
338	}	1✔
339	}	1✔
340	return rv	3✔
341	}	1✔
342
343	// ---------------------------------------------------
344
345	// NsStatusHandler is a REST handler that provides status for
346	// consumption by ns_server
347	type NsStatusHandler struct {
348	mgr *cbgt.Manager
349	serverURL *url.URL
350	}
351
352	func NewNsStatusHandler(mgr cbgt.Manager, server string) (NsStatusHandler, error) {	1✔
353	serverURL, err := url.Parse(server)	×
354	if err != nil {	×
355	return nil, err	×
356	}	1✔
357		1✔
358	return &NsStatusHandler{	×
359	mgr: mgr,	1✔
360	serverURL: serverURL,	1✔
361	}, nil	1✔
362	}	1✔
363
364	func NsHostsForIndex(name string, planPIndexes *cbgt.PlanPIndexes,	1✔
365	nodeDefs *cbgt.NodeDefs) []string {	1✔
366	if planPIndexes == nil {	×
367	return nil	×
368	}	1✔
369		1✔
370	v := struct{}{}	×
371		1✔
372	// find the node UUIDs related to this index	1✔
373	nodes := make(map[string]struct{})	×
374	for _, planPIndex := range planPIndexes.PlanPIndexes {	×
375	if planPIndex.IndexName == name {	×
376	for planPIndexNodeUUID := range planPIndex.Nodes {	×
377	nodes[planPIndexNodeUUID] = v	×
378	}	1✔
379	}	1✔
380	}	1✔
381		1✔
382	// look for all these nodes in the nodes wanted	1✔
383	nodeExtras := make(map[string]string)	×
384	for _, nodeDef := range nodeDefs.NodeDefs {	×
385	_, ok := nodes[nodeDef.UUID]	×
386	if ok {	×
387	var e struct {	×
388	NsHostPort string `json:"nsHostPort"`	1✔
389	}	1✔
390		1✔
391	err := json.Unmarshal([]byte(nodeDef.Extras), &e)	×
392	if err != nil {	×
393	// Early versions of ns_server integration had a	1✔
394	// simple, non-JSON "host:port" format for	1✔
395	// nodeDef.Extras, so fall back to that.	1✔
396	e.NsHostPort = nodeDef.Extras	×
397	}	1✔
398		1✔
399	nodeExtras[nodeDef.UUID] = e.NsHostPort	×
400	}	1✔
401	}	1✔
402		1✔
403	// build slice of node extras	1✔
404	nodeStrings := make(sort.StringSlice, 0, len(nodeExtras))	×
405	for _, nodeExtra := range nodeExtras {	×
406	nodeStrings = append(nodeStrings, nodeExtra)	×
407	}	1✔
408		1✔
409	// sort slice for stability	1✔
410	sort.Sort(nodeStrings)	×
411		1✔
412	return nodeStrings	×
413	}	1✔
414
415	func (h *NsStatusHandler) ServeHTTP(	1✔
416	w http.ResponseWriter, req *http.Request) {	1✔
417		1✔
418	cfg := h.mgr.Cfg()	×
419	planPIndexes, _, err := cbgt.CfgGetPlanPIndexes(cfg)	×
420	if err != nil {	×
421	rest.ShowError(w, req, fmt.Sprintf("could not retrieve plan pIndexes: %v", err), 500)	×
422	return	×
423	}	1✔
424		1✔
425	nodesDefs, _, err := cbgt.CfgGetNodeDefs(cfg, cbgt.NODE_DEFS_WANTED)	×
426	if err != nil {	×
427	rest.ShowError(w, req, "could not retrieve node defs (wanted)", 500)	×
428	return	×
429	}	1✔
430		1✔
431	_, indexDefsMap, err := h.mgr.GetIndexDefs(false)	×
432	if err != nil {	×
433	rest.ShowError(w, req, "could not retrieve index defs", 500)	×
434	return	×
435	}	1✔
436		1✔
437	w.Header().Set("Content-Type", "application/json; charset=utf-8")	×
438	w.Write(cbgt.JsonOpenBrace)	×
439	w.Write(statsNamePrefix)	×
440	w.Write([]byte("status"))	×
441	w.Write(statsNameSuffix)	×
442	w.Write([]byte("["))	×
443		1✔
444	indexDefNames := make(sort.StringSlice, 0, len(indexDefsMap))	×
445	for indexDefName := range indexDefsMap {	×
446	indexDefNames = append(indexDefNames, indexDefName)	×
447	}	1✔
448		1✔
449	sort.Sort(indexDefNames)	×
450		1✔
451	for i, indexDefName := range indexDefNames {	×
452	indexDef := indexDefsMap[indexDefName]	×
453	if i > 0 {	×
454	w.Write(cbgt.JsonComma)	×
455	}	1✔
456		1✔
457	rest.MustEncode(w, struct {	×
458	Completion int `json:"completion"`	1✔
459	Hosts []string `json:"hosts"`	1✔
460	Status string `json:"status"`	1✔
461	Bucket string `json:"bucket"`	1✔
462	Name string `json:"name"`	1✔
463	}{	1✔
464	Bucket: indexDef.SourceName,	1✔
465	Name: indexDefName,	1✔
466	Hosts: NsHostsForIndex(indexDefName, planPIndexes, nodesDefs),	1✔
467	// FIXME hard-coded	1✔
468	Completion: 100,	1✔
469	Status: "Ready",	1✔
470	})	1✔
471	}	1✔
472		1✔
473	w.Write([]byte("],"))	×
474	w.Write(statsNamePrefix)	×
475	w.Write([]byte("code"))	×
476	w.Write(statsNameSuffix)	×
477	w.Write([]byte("\"success\""))	×
478	w.Write(cbgt.JsonCloseBrace)	×
479	}	1✔

couchbase / cbft / 159

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous