From 40f1457eef45c00eaf129a4ab636a2f98950c0bb Mon Sep 17 00:00:00 2001 From: Michael Baird Date: Tue, 26 Sep 2017 22:39:57 -0500 Subject: [PATCH 1/6] Diagnostics package --- diagnostics/diagnostics.go | 282 ++++++++++++++++++++++++++++++++ diagnostics/diagnostics_test.go | 143 ++++++++++++++++ 2 files changed, 425 insertions(+) create mode 100644 diagnostics/diagnostics.go create mode 100644 diagnostics/diagnostics_test.go diff --git a/diagnostics/diagnostics.go b/diagnostics/diagnostics.go new file mode 100644 index 000000000..fc78b94cb --- /dev/null +++ b/diagnostics/diagnostics.go @@ -0,0 +1,282 @@ +package diagnostics + +import ( + "bytes" + "encoding/json" + "fmt" + "io" + "io/ioutil" + "log" + "net/http" + "strconv" + "strings" + "sync" + "time" + + "github.com/pilosa/pilosa" +) + +// TODO: white list of statsd metrics to use +// TODO: unique Cluster ID +// TODO: how should this be disabled, config + +// Default interval to sync diagnostics metrics. +const ( + DefaultDiagnosticsInterval = 10 * time.Second + DefaultVersionCheckURL = "https://diagnostics.pilosa.com/v0/version" +) + +type versionResponse struct { + Version string `json:"version"` + Message string `json:"message"` +} + +// Diagnostics represents a client to the Pilosa cluster. +type Diagnostics struct { + mu sync.Mutex + wg sync.WaitGroup + closing chan struct{} + host string + VersionURL string + version string + startTime int64 + start time.Time + + counts map[string]int64 + metrics map[string]string + + client *http.Client + interval time.Duration + + logOutput io.Writer +} + +// New returns a pointer to a new Diagnostics Client given an addr in the format "hostname:port". +func New(host string) *Diagnostics { + return &Diagnostics{ + closing: make(chan struct{}), + host: host, + VersionURL: DefaultVersionCheckURL, + startTime: time.Now().Unix(), + start: time.Now(), + client: http.DefaultClient, + counts: make(map[string]int64), + metrics: make(map[string]string), + interval: DefaultDiagnosticsInterval, + logOutput: ioutil.Discard, + } +} + +// SetVersion of locally running Pilosa Cluster to check against master. +func (d *Diagnostics) SetVersion(v string) { + d.version = v + d.Set("Version", v, 1.0) +} + +// schedule start the diagnostics service ticker. +func (d *Diagnostics) schedule() { + ticker := time.NewTicker(d.interval) + defer ticker.Stop() + + for { + select { + case <-d.closing: + return + case <-ticker.C: + d.CheckVersion() + d.Flush() + } + } +} + +// Flush sends the current metrics. +func (d *Diagnostics) Flush() error { + d.mu.Lock() + d.metrics["uptime"] = strconv.FormatInt((time.Now().Unix() - d.startTime), 10) + + buf, _ := d.MarshalJSON() + d.Reset() + d.mu.Unlock() + + // d.logger().Println(string(buf)) + req, err := http.NewRequest("POST", d.host, bytes.NewReader(buf)) + req.Header.Set("Content-Type", "application/json") + resp, err := d.client.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + + // TODO verify response + // Read response into buffer. + // body, err := ioutil.ReadAll(resp.Body) + // if err != nil { + // return err + // } + + // TODO circuit breaker + return nil +} + +// Reset clears the incremented metrics. +func (d *Diagnostics) Reset() { + d.counts = make(map[string]int64) +} + +// Open starts the diagnostics metric go routine. +func (d *Diagnostics) Open() { + d.wg.Add(1) + go func() { defer d.wg.Done(); d.schedule() }() +} + +// Close notify goroutine to stop. +func (d *Diagnostics) Close() error { + close(d.closing) + d.wg.Wait() + return nil +} + +// CheckVersion of the local build against Pilosa master. +func (d *Diagnostics) CheckVersion() error { + var rsp versionResponse + req, err := http.NewRequest("GET", d.VersionURL, nil) + resp, err := d.client.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("http: status=%d", resp.StatusCode) + } else if err := json.NewDecoder(resp.Body).Decode(&rsp); err != nil { + return fmt.Errorf("json decode: %s", err) + } + + if err := d.CompareVersion(rsp.Version); err != nil { + d.logger().Printf("%s\n", err.Error()) + } + + return nil +} + +// CompareVersion check version strings. +func (d *Diagnostics) CompareVersion(value string) error { + currentVersion := VersionSegments(value) + localVersion := VersionSegments(d.version) + + if localVersion[0] < currentVersion[0] { //Major + return fmt.Errorf("Warning: You are running an older version of Pilosa %s. The latest Major release is %s", d.version, value) + } else if localVersion[1] < currentVersion[1] { // Minor + return fmt.Errorf("Warning: You are running an older version of Pilosa %s. The latest Minor release is %s", d.version, value) + } else if localVersion[2] < currentVersion[2] { // Patch + return fmt.Errorf("There is a new patch relese of Pilosa availbale: %s", value) + } + + return nil +} + +// MarshalJSON custom marshall string and int maps together. +func (d *Diagnostics) MarshalJSON() ([]byte, error) { + buffer := bytes.NewBufferString("{") + length := len(d.counts) + count := 0 + + for key, value := range d.counts { + jsonValue, err := json.Marshal(value) + if err != nil { + return nil, err + } + buffer.WriteString(fmt.Sprintf("\"%s\":%s", key, string(jsonValue))) + count++ + if count < length { + buffer.WriteString(",") + } + } + if length > 0 { + buffer.WriteString(",") + } + length = len(d.metrics) + count = 0 + for key, value := range d.metrics { + jsonValue, err := json.Marshal(value) + if err != nil { + return nil, err + } + buffer.WriteString(fmt.Sprintf("\"%s\":%s", key, string(jsonValue))) + count++ + if count < length { + buffer.WriteString(",") + } + } + + buffer.WriteString("}") + return buffer.Bytes(), nil +} + +// Stats interface implementation. + +// Tags no-op. +func (d *Diagnostics) Tags() []string { + return nil +} + +// WithTags no-op. +func (d *Diagnostics) WithTags(tags ...string) pilosa.StatsClient { + return d +} + +// Count tracks the number of times something occurs per diagnostic period. +func (d *Diagnostics) Count(name string, value int64, rate float64) { + d.mu.Lock() + defer d.mu.Unlock() + d.counts[name] += value +} + +// CountWithCustomTags Tracks the number of times something occurs per diagnostic period. +func (d *Diagnostics) CountWithCustomTags(name string, value int64, rate float64, tags []string) { + d.mu.Lock() + defer d.mu.Unlock() + d.counts[name] += value +} + +// Gauge records the value of a metric. +func (d *Diagnostics) Gauge(name string, value float64, rate float64) { + d.Set(name, strconv.FormatFloat(value, 'f', -1, 64), rate) +} + +// Histogram is a no-op. +func (d *Diagnostics) Histogram(name string, value float64, rate float64) { +} + +// Set adds a key value metric. +func (d *Diagnostics) Set(name string, value string, rate float64) { + d.mu.Lock() + defer d.mu.Unlock() + d.metrics[name] = value +} + +// Timing no-op. +func (d *Diagnostics) Timing(name string, value time.Duration, rate float64) { +} + +// SetLogger Set the logger output type. +func (d *Diagnostics) SetLogger(logger io.Writer) { + d.logOutput = logger +} + +// logger returns a logger that writes to LogOutput. +func (d *Diagnostics) logger() *log.Logger { + return log.New(d.logOutput, "", log.LstdFlags) +} + +// VersionSegments returns the numeric segments of the version as a slice of ints. +func VersionSegments(segments string) []int { + segments = strings.Trim(segments, "v") + segments = strings.Split(segments, "-")[0] + s := strings.Split(segments, ".") + segmentSlice := make([]int, len(s)) + for i, v := range s { + segmentSlice[i], _ = strconv.Atoi(v) + } + return segmentSlice +} diff --git a/diagnostics/diagnostics_test.go b/diagnostics/diagnostics_test.go new file mode 100644 index 000000000..85cdc56d6 --- /dev/null +++ b/diagnostics/diagnostics_test.go @@ -0,0 +1,143 @@ +package diagnostics_test + +import ( + "encoding/json" + "io/ioutil" + "net/http" + "net/http/httptest" + "reflect" + "strings" + "testing" + "time" + + "github.com/pilosa/pilosa/diagnostics" +) + +func TestDiagnosticsClient(t *testing.T) { + // Mock server. + server := httptest.NewServer(nil) + defer server.Close() + + // Create a new client. + d := diagnostics.New(server.URL) + d.SetLogger(ioutil.Discard) + defer d.Close() + + dur, _ := time.ParseDuration("123us") + d.CountWithCustomTags("ct", 1, 1.0, []string{"foo:bar"}) + d.Count("cc", 1, 1.0) + d.Gauge("gg", 10, 1.0) + d.Histogram("hh", 1, 1.0) + d.Timing("tt", dur, 1.0) + d.Set("ss", "ss", 1.0) + + d1 := d.WithTags("test") + if !reflect.DeepEqual(d, d1) { + t.Fatalf("Diagnostics is a singleton") + } + + if s := d.Tags(); s != nil { + t.Fatalf("No Diagnostics Tags") + } + + data, err := d.MarshalJSON() + if err != nil { + t.Fatal(err) + } + + // Test the recorded metrics, note that some types are skipped. + var eq bool + output1 := []byte(`{"ct":1,"cc":1,"gg":"10","ss":"ss"}`) + if eq, err = compareJSON(data, output1); err != nil { + t.Fatal(err) + } + if !eq { + t.Fatalf("unexpected diagnostics: %+v", string(data)) + } + + // Test the metrics after a flush. + d.Flush() + data, err = d.MarshalJSON() + if err != nil { + t.Fatal(err) + } + + output2 := []byte(`{"gg":"10","ss":"ss","uptime":"0"}`) + if eq, err = compareJSON(data, output2); err != nil { + t.Fatal(err) + } + if !eq { + t.Fatalf("unexpected diagnostics after flush: %+v", string(data)) + } +} + +func TestDiagnosticsVersion_Parse(t *testing.T) { + version := "0.1.1" + vs := diagnostics.VersionSegments(version) + + output := []int{0, 1, 1} + if !reflect.DeepEqual(vs, output) { + t.Fatalf("unexpected version: %+v", vs) + } +} + +func TestDiagnosticsVersion_Compare(t *testing.T) { + d := diagnostics.New("localhost:10101") + defer d.Close() + + version := "0.1.1" + d.SetVersion(version) + + err := d.CompareVersion("1.7.0") + if !strings.Contains(err.Error(), "The latest Major release is") { + t.Fatalf("Expected Major Version Missmatch, actual error: %s", err) + } + err = d.CompareVersion("0.7.0") + if !strings.Contains(err.Error(), "The latest Minor release is") { + t.Fatalf("Expected Minor Version Missmatch, actual error: %s", err) + } + err = d.CompareVersion("0.1.2") + if !strings.Contains(err.Error(), "There is a new patch relese of Pilosa") { + t.Fatalf("Expected Patch Version Missmatch, actual error: %s", err) + } + err = d.CompareVersion("0.1.1") + if err != nil { + t.Fatalf("Versions should match") + } +} + +func TestDiagnosticsVersion_Check(t *testing.T) { + // Mock server. + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + json.NewEncoder(w).Encode(versionResponse{ + Version: "1.1.1", + }) + })) + defer server.Close() + + // Create a new client. + d := diagnostics.New("localhost:10101") + defer d.Close() + + version := "0.1.1" + d.SetVersion(version) + d.VersionURL = server.URL + + d.CheckVersion() +} + +type versionResponse struct { + Version string `json:"version"` +} + +func compareJSON(a, b []byte) (bool, error) { + var j1, j2 interface{} + if err := json.Unmarshal(a, &j1); err != nil { + return false, err + } + if err := json.Unmarshal(b, &j2); err != nil { + return false, err + } + return reflect.DeepEqual(j1, j2), nil +} From 68ce07b45d281fd3d65ff7a737daafda45c4756a Mon Sep 17 00:00:00 2001 From: Michael Baird Date: Tue, 26 Sep 2017 22:40:55 -0500 Subject: [PATCH 2/6] add Open/Close interface to Stats packages --- stats.go | 36 ++++++++++++++++++++++++++++++++++-- stats_test.go | 2 ++ statsd/statsd.go | 3 +++ 3 files changed, 39 insertions(+), 2 deletions(-) diff --git a/stats.go b/stats.go index 71f0d56b1..998b84511 100644 --- a/stats.go +++ b/stats.go @@ -58,6 +58,12 @@ type StatsClient interface { // SetLogger Set the logger output type SetLogger(logger io.Writer) + + // Starts the service + Open() + + // Closes the client + Close() error } // NopStatsClient represents a client that doesn't do anything. @@ -74,6 +80,8 @@ func (c *nopStatsClient) Histogram(name string, value float64, rate float64) func (c *nopStatsClient) Set(name string, value string, rate float64) {} func (c *nopStatsClient) Timing(name string, value time.Duration, rate float64) {} func (c *nopStatsClient) SetLogger(logger io.Writer) {} +func (c *nopStatsClient) Open() {} +func (c *nopStatsClient) Close() error { return nil } // ExpvarStatsClient writes stats out to expvars. type ExpvarStatsClient struct { @@ -145,10 +153,16 @@ func (c *ExpvarStatsClient) Timing(name string, value time.Duration, rate float6 c.mu.Unlock() } -// SetLogger has no logger +// SetLogger has no logger. func (c *ExpvarStatsClient) SetLogger(logger io.Writer) { } +// Open no-op. +func (c *ExpvarStatsClient) Open() {} + +// Close no-op. +func (c *ExpvarStatsClient) Close() error { return nil } + // MultiStatsClient joins multiple stats clients together. type MultiStatsClient []StatsClient @@ -211,13 +225,31 @@ func (a MultiStatsClient) Timing(name string, value time.Duration, rate float64) } } -// SetLogger Sets the StatsD logger output type +// SetLogger Sets the StatsD logger output type. func (a MultiStatsClient) SetLogger(logger io.Writer) { for _, c := range a { c.SetLogger(logger) } } +// Open starts the stat service. +func (a MultiStatsClient) Open() { + for _, c := range a { + c.Open() + } +} + +// Close shuts down the stats clients. +func (a MultiStatsClient) Close() error { + for _, c := range a { + err := c.Close() + if err != nil { + return err + } + } + return nil +} + // UnionStringSlice returns a sorted set of tags which combine a & b. func UnionStringSlice(a, b []string) []string { // Sort both sets first. diff --git a/stats_test.go b/stats_test.go index 7ed679f21..07254a702 100644 --- a/stats_test.go +++ b/stats_test.go @@ -344,3 +344,5 @@ func (c *MockStats) Histogram(name string, value float64, rate float64) {} func (c *MockStats) Set(name string, value string, rate float64) {} func (c *MockStats) Timing(name string, value time.Duration, rate float64) {} func (c *MockStats) SetLogger(logger io.Writer) {} +func (c *MockStats) Open() {} +func (c *MockStats) Close() error { return nil } diff --git a/statsd/statsd.go b/statsd/statsd.go index e55b190ad..d46dd637f 100644 --- a/statsd/statsd.go +++ b/statsd/statsd.go @@ -58,6 +58,9 @@ func NewStatsClient(host string) (*StatsClient, error) { }, nil } +// Open no-op +func (c *StatsClient) Open() {} + // Close closes the connection to the agent. func (c *StatsClient) Close() error { return c.client.Close() From 566f4116f17e5c223ef02ec90d4cc2077646f184 Mon Sep 17 00:00:00 2001 From: Michael Baird Date: Tue, 26 Sep 2017 22:42:21 -0500 Subject: [PATCH 3/6] Using MultiStatsClient add Diagnostics --- server/server.go | 20 +++++++++++++++----- 1 file changed, 15 insertions(+), 5 deletions(-) diff --git a/server/server.go b/server/server.go index 157fc69f3..ba60e8923 100644 --- a/server/server.go +++ b/server/server.go @@ -31,6 +31,7 @@ import ( "time" "github.com/pilosa/pilosa" + "github.com/pilosa/pilosa/diagnostics" "github.com/pilosa/pilosa/gossip" "github.com/pilosa/pilosa/statsd" ) @@ -41,7 +42,8 @@ func init() { const ( // DefaultDataDir is the default data directory. - DefaultDataDir = "~/.pilosa" + DefaultDataDir = "~/.pilosa" + DefaultDiagnosticServer = "https://requestb.in/w3uukzw3" ) // Command represents the state of the pilosa server command. @@ -222,12 +224,20 @@ func (m *Command) Close() error { // NewStatsClient creates a stats client from the config func NewStatsClient(name string, host string) (pilosa.StatsClient, error) { + ms := make(pilosa.MultiStatsClient, 1) + d := diagnostics.New(DefaultDiagnosticServer) + d.SetVersion(pilosa.Version) + ms[0] = d + switch name { case "expvar": - return pilosa.NewExpvarStatsClient(), nil + ms = append(ms, pilosa.NewExpvarStatsClient()) case "statsd": - return statsd.NewStatsClient(host) - default: - return pilosa.NopStatsClient, nil + r, err := statsd.NewStatsClient(host) + if err != nil { + return nil, err + } + ms = append(ms, r) } + return ms, nil } From e87e7fda68b995148eef060e9419bcc3ebfb3bf9 Mon Sep 17 00:00:00 2001 From: Michael Baird Date: Tue, 26 Sep 2017 22:42:35 -0500 Subject: [PATCH 4/6] Add some new Diagnostics metrics --- holder.go | 3 +++ server.go | 26 ++++++++++++++++---------- 2 files changed, 19 insertions(+), 10 deletions(-) diff --git a/holder.go b/holder.go index 0cf5f6901..e358739f5 100644 --- a/holder.go +++ b/holder.go @@ -123,11 +123,14 @@ func (h *Holder) Open() error { h.wg.Add(1) go func() { defer h.wg.Done(); h.monitorCacheFlush() }() + h.Stats.Open() return nil } // Close closes all open fragments. func (h *Holder) Close() error { + h.Stats.Close() + // Notify goroutines of closing and wait for completion. close(h.closing) h.wg.Wait() diff --git a/server.go b/server.go index 8194278ba..e4d772349 100644 --- a/server.go +++ b/server.go @@ -202,10 +202,10 @@ func (s *Server) Addr() net.Addr { return s.ln.Addr() } +// Logger returns a logger that writes to LogOutput func (s *Server) Logger() *log.Logger { return log.New(s.LogOutput, "", log.LstdFlags) } func (s *Server) monitorAntiEntropy() { - t := time.Now() ticker := time.NewTicker(s.AntiEntropyInterval) defer ticker.Stop() @@ -219,7 +219,7 @@ func (s *Server) monitorAntiEntropy() { case <-ticker.C: s.Holder.Stats.Count("AntiEntropy", 1, 1.0) } - + t := time.Now() s.Logger().Printf("holder sync beginning") // Initialize syncer with local holder and remote client. @@ -237,9 +237,9 @@ func (s *Server) monitorAntiEntropy() { // Record successful sync in log. s.Logger().Printf("holder sync complete") + dif := time.Since(t) + s.Holder.Stats.Histogram("AntiEntropyDuration", float64(dif), 1.0) } - dif := time.Since(t) - s.Holder.Stats.Histogram("AntiEntropyDuration", float64(dif), 1.0) } // monitorMaxSlices periodically pulls the highest slice from each node in the cluster. @@ -486,7 +486,13 @@ func checkMaxSlices(hostport string) (map[string]uint64, error) { // monitorRuntime periodically polls the Go runtime metrics. func (s *Server) monitorRuntime() { - // Disable metrics when poll interval is zero + s.Holder.Stats.Set("Host", s.Host, 1.0) + s.Holder.Stats.Set("Cluster", strings.Join(s.Cluster.NodeSetHosts(), ","), 1.0) + s.Holder.Stats.Set("NumNodes", strconv.Itoa(len(s.Cluster.Nodes)), 1.0) + s.Holder.Stats.Set("NumCPU", strconv.Itoa(runtime.NumCPU()), 1.0) + // TODO should we force this to run for diagnostics? + + // Disable metrics when poll interval is zero. if s.MetricInterval <= 0 { return } @@ -506,18 +512,18 @@ func (s *Server) monitorRuntime() { case <-s.closing: return case <-gcn.AfterGC(): - // GC just ran + // GC just ran. s.Holder.Stats.Count("garbage_collection", 1, 1.0) case <-ticker.C: } - // Record the number of go routines + // Record the number of go routines. s.Holder.Stats.Gauge("goroutines", float64(runtime.NumGoroutine()), 1.0) - // Open File handles + // Open File handles. s.Holder.Stats.Gauge("OpenFiles", float64(CountOpenFiles()), 1.0) - // Runtime memory metrics + // Runtime memory metrics. runtime.ReadMemStats(&m) s.Holder.Stats.Gauge("HeapAlloc", float64(m.HeapAlloc), 1.0) s.Holder.Stats.Gauge("HeapInuse", float64(m.HeapInuse), 1.0) @@ -527,7 +533,7 @@ func (s *Server) monitorRuntime() { } } -// CountOpenFiles on opperating systems that support lsof +// CountOpenFiles on operating systems that support lsof. func CountOpenFiles() int { count := 0 From e91817701862b7c9a3f92f2529c4003a4ba50ed3 Mon Sep 17 00:00:00 2001 From: Cody Soyland Date: Thu, 28 Sep 2017 17:58:13 -0500 Subject: [PATCH 5/6] Add basic diagnostics benchmark --- diagnostics/diagnostics_test.go | 24 ++++++++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/diagnostics/diagnostics_test.go b/diagnostics/diagnostics_test.go index 85cdc56d6..1a811df9c 100644 --- a/diagnostics/diagnostics_test.go +++ b/diagnostics/diagnostics_test.go @@ -6,6 +6,7 @@ import ( "net/http" "net/http/httptest" "reflect" + "runtime" "strings" "testing" "time" @@ -141,3 +142,26 @@ func compareJSON(a, b []byte) (bool, error) { } return reflect.DeepEqual(j1, j2), nil } + +func BenchmarkDiagnostics(b *testing.B) { + // Mock server. + server := httptest.NewServer(nil) + defer server.Close() + + // Create a new client. + d := diagnostics.New(server.URL) + d.SetLogger(ioutil.Discard) + defer d.Close() + + prev := runtime.GOMAXPROCS(4) + defer runtime.GOMAXPROCS(prev) + + b.ResetTimer() + + b.RunParallel(func(pb *testing.PB) { + for pb.Next() { + d.Count("cc", 1, 1.0) + d.Gauge("gg", 10, 1.0) + } + }) +} From 9876b51fc49230c923578ad1b542825309737b40 Mon Sep 17 00:00:00 2001 From: Michael Baird Date: Mon, 16 Oct 2017 12:08:48 -0500 Subject: [PATCH 6/6] simplifying the diagnostics client. Using circuit breaker to manage the diagnostics http connection. --- diagnostics/diagnostics.go | 142 +++++++------------------------- diagnostics/diagnostics_test.go | 31 ++----- server.go | 70 +++++++++++++--- server/server.go | 20 ++--- 4 files changed, 105 insertions(+), 158 deletions(-) diff --git a/diagnostics/diagnostics.go b/diagnostics/diagnostics.go index fc78b94cb..761e7adca 100644 --- a/diagnostics/diagnostics.go +++ b/diagnostics/diagnostics.go @@ -13,16 +13,14 @@ import ( "sync" "time" - "github.com/pilosa/pilosa" + "github.com/sony/gobreaker" ) -// TODO: white list of statsd metrics to use // TODO: unique Cluster ID -// TODO: how should this be disabled, config // Default interval to sync diagnostics metrics. const ( - DefaultDiagnosticsInterval = 10 * time.Second + DefaultDiagnosticsInterval = 1 * time.Hour DefaultVersionCheckURL = "https://diagnostics.pilosa.com/v0/version" ) @@ -42,17 +40,20 @@ type Diagnostics struct { startTime int64 start time.Time - counts map[string]int64 - metrics map[string]string + metrics map[string]interface{} client *http.Client interval time.Duration + cb *gobreaker.CircuitBreaker logOutput io.Writer } // New returns a pointer to a new Diagnostics Client given an addr in the format "hostname:port". func New(host string) *Diagnostics { + var st gobreaker.Settings + st.Timeout = DefaultDiagnosticsInterval * 2 + return &Diagnostics{ closing: make(chan struct{}), host: host, @@ -60,17 +61,17 @@ func New(host string) *Diagnostics { startTime: time.Now().Unix(), start: time.Now(), client: http.DefaultClient, - counts: make(map[string]int64), - metrics: make(map[string]string), + metrics: make(map[string]interface{}), interval: DefaultDiagnosticsInterval, logOutput: ioutil.Discard, + cb: gobreaker.NewCircuitBreaker(st), } } // SetVersion of locally running Pilosa Cluster to check against master. func (d *Diagnostics) SetVersion(v string) { d.version = v - d.Set("Version", v, 1.0) + d.Set("Version", v) } // schedule start the diagnostics service ticker. @@ -92,35 +93,28 @@ func (d *Diagnostics) schedule() { // Flush sends the current metrics. func (d *Diagnostics) Flush() error { d.mu.Lock() - d.metrics["uptime"] = strconv.FormatInt((time.Now().Unix() - d.startTime), 10) - - buf, _ := d.MarshalJSON() - d.Reset() + d.metrics["uptime"] = (time.Now().Unix() - d.startTime) + buf, _ := d.Encode() d.mu.Unlock() - // d.logger().Println(string(buf)) - req, err := http.NewRequest("POST", d.host, bytes.NewReader(buf)) - req.Header.Set("Content-Type", "application/json") - resp, err := d.client.Do(req) - if err != nil { - return err - } - defer resp.Body.Close() + _, err := d.cb.Execute(func() (interface{}, error) { + req, err := http.NewRequest("POST", d.host, bytes.NewReader(buf)) + req.Header.Set("Content-Type", "application/json") + resp, err := d.client.Do(req) + if err != nil { + return nil, err + } + defer resp.Body.Close() - // TODO verify response - // Read response into buffer. - // body, err := ioutil.ReadAll(resp.Body) - // if err != nil { - // return err - // } + // TODO verify response + body, err := ioutil.ReadAll(resp.Body) + if err != nil { + return nil, err + } + return body, nil + }) - // TODO circuit breaker - return nil -} - -// Reset clears the incremented metrics. -func (d *Diagnostics) Reset() { - d.counts = make(map[string]int64) + return err } // Open starts the diagnostics metric go routine. @@ -175,90 +169,18 @@ func (d *Diagnostics) CompareVersion(value string) error { return nil } -// MarshalJSON custom marshall string and int maps together. -func (d *Diagnostics) MarshalJSON() ([]byte, error) { - buffer := bytes.NewBufferString("{") - length := len(d.counts) - count := 0 - - for key, value := range d.counts { - jsonValue, err := json.Marshal(value) - if err != nil { - return nil, err - } - buffer.WriteString(fmt.Sprintf("\"%s\":%s", key, string(jsonValue))) - count++ - if count < length { - buffer.WriteString(",") - } - } - if length > 0 { - buffer.WriteString(",") - } - length = len(d.metrics) - count = 0 - for key, value := range d.metrics { - jsonValue, err := json.Marshal(value) - if err != nil { - return nil, err - } - buffer.WriteString(fmt.Sprintf("\"%s\":%s", key, string(jsonValue))) - count++ - if count < length { - buffer.WriteString(",") - } - } - - buffer.WriteString("}") - return buffer.Bytes(), nil -} - -// Stats interface implementation. - -// Tags no-op. -func (d *Diagnostics) Tags() []string { - return nil -} - -// WithTags no-op. -func (d *Diagnostics) WithTags(tags ...string) pilosa.StatsClient { - return d -} - -// Count tracks the number of times something occurs per diagnostic period. -func (d *Diagnostics) Count(name string, value int64, rate float64) { - d.mu.Lock() - defer d.mu.Unlock() - d.counts[name] += value -} - -// CountWithCustomTags Tracks the number of times something occurs per diagnostic period. -func (d *Diagnostics) CountWithCustomTags(name string, value int64, rate float64, tags []string) { - d.mu.Lock() - defer d.mu.Unlock() - d.counts[name] += value -} - -// Gauge records the value of a metric. -func (d *Diagnostics) Gauge(name string, value float64, rate float64) { - d.Set(name, strconv.FormatFloat(value, 'f', -1, 64), rate) -} - -// Histogram is a no-op. -func (d *Diagnostics) Histogram(name string, value float64, rate float64) { +// Encode metrics maps into the json message format +func (d *Diagnostics) Encode() ([]byte, error) { + return json.Marshal(d.metrics) } // Set adds a key value metric. -func (d *Diagnostics) Set(name string, value string, rate float64) { +func (d *Diagnostics) Set(name string, value interface{}) { d.mu.Lock() defer d.mu.Unlock() d.metrics[name] = value } -// Timing no-op. -func (d *Diagnostics) Timing(name string, value time.Duration, rate float64) { -} - // SetLogger Set the logger output type. func (d *Diagnostics) SetLogger(logger io.Writer) { d.logOutput = logger diff --git a/diagnostics/diagnostics_test.go b/diagnostics/diagnostics_test.go index 1a811df9c..d780a276f 100644 --- a/diagnostics/diagnostics_test.go +++ b/diagnostics/diagnostics_test.go @@ -9,7 +9,6 @@ import ( "runtime" "strings" "testing" - "time" "github.com/pilosa/pilosa/diagnostics" ) @@ -24,31 +23,17 @@ func TestDiagnosticsClient(t *testing.T) { d.SetLogger(ioutil.Discard) defer d.Close() - dur, _ := time.ParseDuration("123us") - d.CountWithCustomTags("ct", 1, 1.0, []string{"foo:bar"}) - d.Count("cc", 1, 1.0) - d.Gauge("gg", 10, 1.0) - d.Histogram("hh", 1, 1.0) - d.Timing("tt", dur, 1.0) - d.Set("ss", "ss", 1.0) + d.Set("gg", 10) + d.Set("ss", "ss") - d1 := d.WithTags("test") - if !reflect.DeepEqual(d, d1) { - t.Fatalf("Diagnostics is a singleton") - } - - if s := d.Tags(); s != nil { - t.Fatalf("No Diagnostics Tags") - } - - data, err := d.MarshalJSON() + data, err := d.Encode() if err != nil { t.Fatal(err) } // Test the recorded metrics, note that some types are skipped. var eq bool - output1 := []byte(`{"ct":1,"cc":1,"gg":"10","ss":"ss"}`) + output1 := []byte(`{"gg":10,"ss":"ss"}`) if eq, err = compareJSON(data, output1); err != nil { t.Fatal(err) } @@ -58,12 +43,12 @@ func TestDiagnosticsClient(t *testing.T) { // Test the metrics after a flush. d.Flush() - data, err = d.MarshalJSON() + data, err = d.Encode() if err != nil { t.Fatal(err) } - output2 := []byte(`{"gg":"10","ss":"ss","uptime":"0"}`) + output2 := []byte(`{"gg":10,"ss":"ss","uptime":0}`) if eq, err = compareJSON(data, output2); err != nil { t.Fatal(err) } @@ -160,8 +145,8 @@ func BenchmarkDiagnostics(b *testing.B) { b.RunParallel(func(pb *testing.PB) { for pb.Next() { - d.Count("cc", 1, 1.0) - d.Gauge("gg", 10, 1.0) + d.Set("cc", 1) + d.Set("gg", "test") } }) } diff --git a/server.go b/server.go index e4d772349..3813a6822 100644 --- a/server.go +++ b/server.go @@ -33,6 +33,7 @@ import ( "github.com/CAFxX/gcnotifier" "github.com/gogo/protobuf/proto" + "github.com/pilosa/pilosa/diagnostics" "github.com/pilosa/pilosa/internal" ) @@ -40,6 +41,7 @@ import ( const ( DefaultAntiEntropyInterval = 10 * time.Minute DefaultPollingInterval = 60 * time.Second + DefaultDiagnosticServer = "https://diagnostics.pilosa.com/v0/diagnostics" ) // Server represents a holder wrapped by a running HTTP server. @@ -58,14 +60,16 @@ type Server struct { // Cluster configuration. // Host is replaced with actual host after opening if port is ":0". - Network string - Host string - Cluster *Cluster + Network string + Host string + Cluster *Cluster + diagnostics *diagnostics.Diagnostics // Background monitoring intervals. AntiEntropyInterval time.Duration PollingInterval time.Duration MetricInterval time.Duration + DiagnosticInterval time.Duration // Misc options. MaxWritesPerRequest int @@ -82,12 +86,14 @@ func NewServer() *Server { Handler: NewHandler(), Broadcaster: NopBroadcaster, BroadcastReceiver: NopBroadcastReceiver, + diagnostics: diagnostics.New(DefaultDiagnosticServer), Network: "tcp", AntiEntropyInterval: DefaultAntiEntropyInterval, PollingInterval: DefaultPollingInterval, MetricInterval: 0, + DiagnosticInterval: diagnostics.DefaultDiagnosticsInterval LogOutput: os.Stderr, } @@ -170,10 +176,11 @@ func (s *Server) Open() error { }() // Start background monitoring. - s.wg.Add(3) + s.wg.Add(4) go func() { defer s.wg.Done(); s.monitorAntiEntropy() }() go func() { defer s.wg.Done(); s.monitorMaxSlices() }() go func() { defer s.wg.Done(); s.monitorRuntime() }() + go func() { defer s.wg.Done(); s.monitorDiagnostics() }() return nil } @@ -484,14 +491,57 @@ func checkMaxSlices(hostport string) (map[string]uint64, error) { return pb.MaxSlices, nil } +// monitorDiagnostics periodically polls the the Pilosa Indexes for cluster info. +func (s *Server) monitorDiagnostics() { + if s.DiagnosticInterval <= 0 { + return + } + + s.diagnostics.SetLogger(s.LogOutput) + s.diagnostics.SetVersion(Version) + s.diagnostics.Set("Host", s.Host) + s.diagnostics.Set("Cluster", strings.Join(s.Cluster.NodeSetHosts(), ",")) + s.diagnostics.Set("NumNodes", len(s.Cluster.Nodes)) + s.diagnostics.Set("NumCPU", runtime.NumCPU()) + // TODO: unique cluster ID + + ticker := time.NewTicker(s.DiagnosticInterval) + defer ticker.Stop() + + for { + // Wait for tick or a close. + select { + case <-s.closing: + return + case <-ticker.C: + numFrames := 0 + numSlices := uint64(0) + for _, index := range s.Holder.Indexes() { + numSlices += index.MaxSlice() + 1 + for _, f := range index.Frames() { + numFrames++ + if f.rangeEnabled { + s.diagnostics.Set("BSIEnabled", true) + } + if f.timeQuantum != "" { + s.diagnostics.Set("TimeQuantumEnabled", true) + } + } + } + + s.diagnostics.Set("NumIndexes", len(s.Holder.Indexes())) + s.diagnostics.Set("NumFrames", numFrames) + s.diagnostics.Set("NumSlices", numSlices) + s.diagnostics.Set("OpenFiles", CountOpenFiles()) + s.diagnostics.Set("GoRoutines", runtime.NumGoroutine()) + s.diagnostics.CheckVersion() + s.diagnostics.Flush() + } + } +} + // monitorRuntime periodically polls the Go runtime metrics. func (s *Server) monitorRuntime() { - s.Holder.Stats.Set("Host", s.Host, 1.0) - s.Holder.Stats.Set("Cluster", strings.Join(s.Cluster.NodeSetHosts(), ","), 1.0) - s.Holder.Stats.Set("NumNodes", strconv.Itoa(len(s.Cluster.Nodes)), 1.0) - s.Holder.Stats.Set("NumCPU", strconv.Itoa(runtime.NumCPU()), 1.0) - // TODO should we force this to run for diagnostics? - // Disable metrics when poll interval is zero. if s.MetricInterval <= 0 { return diff --git a/server/server.go b/server/server.go index ba60e8923..157fc69f3 100644 --- a/server/server.go +++ b/server/server.go @@ -31,7 +31,6 @@ import ( "time" "github.com/pilosa/pilosa" - "github.com/pilosa/pilosa/diagnostics" "github.com/pilosa/pilosa/gossip" "github.com/pilosa/pilosa/statsd" ) @@ -42,8 +41,7 @@ func init() { const ( // DefaultDataDir is the default data directory. - DefaultDataDir = "~/.pilosa" - DefaultDiagnosticServer = "https://requestb.in/w3uukzw3" + DefaultDataDir = "~/.pilosa" ) // Command represents the state of the pilosa server command. @@ -224,20 +222,12 @@ func (m *Command) Close() error { // NewStatsClient creates a stats client from the config func NewStatsClient(name string, host string) (pilosa.StatsClient, error) { - ms := make(pilosa.MultiStatsClient, 1) - d := diagnostics.New(DefaultDiagnosticServer) - d.SetVersion(pilosa.Version) - ms[0] = d - switch name { case "expvar": - ms = append(ms, pilosa.NewExpvarStatsClient()) + return pilosa.NewExpvarStatsClient(), nil case "statsd": - r, err := statsd.NewStatsClient(host) - if err != nil { - return nil, err - } - ms = append(ms, r) + return statsd.NewStatsClient(host) + default: + return pilosa.NopStatsClient, nil } - return ms, nil }