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 +}