diff --git a/Gopkg.lock b/Gopkg.lock index 19483791d..09793a7e2 100644 --- a/Gopkg.lock +++ b/Gopkg.lock @@ -161,6 +161,12 @@ packages = ["."] revision = "e2103e2c35297fb7e17febb81e49b312087a2372" +[[projects]] + name = "github.com/sony/gobreaker" + packages = ["."] + revision = "e9556a45379ef1da12e54847edb2fb3d7d566f36" + version = "0.3.0" + [[projects]] branch = "master" name = "github.com/spf13/afero" @@ -230,6 +236,6 @@ [solve-meta] analyzer-name = "dep" analyzer-version = 1 - inputs-digest = "e8e78a7c61547d8f4d967c8deac9334b3151e7c10c4f7909e80758956e9c8204" + inputs-digest = "0a7eaef315413bc55acf42c5566982fb71be8c11f33162127953572738967b35" solver-name = "gps-cdcl" solver-version = 1 diff --git a/config.go b/config.go index 60c49932b..d68158f5e 100644 --- a/config.go +++ b/config.go @@ -99,6 +99,7 @@ type Config struct { Service string `toml:"service"` Host string `toml:"host"` PollInterval Duration `toml:"poll-interval"` + Diagnostics bool `toml:"diagnostics"` } `toml:"metric"` TLS TLSConfig @@ -116,6 +117,7 @@ func NewConfig() *Config { c.Cluster.Hosts = []string{} c.AntiEntropy.Interval = Duration(DefaultAntiEntropyInterval) c.Metric.Service = DefaultMetrics + c.Metric.Diagnostics = true c.TLS = TLSConfig{} return c } diff --git a/ctl/server.go b/ctl/server.go index 77361f7db..2d3b82ca1 100644 --- a/ctl/server.go +++ b/ctl/server.go @@ -44,6 +44,7 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) { flags.StringVarP(&srv.Config.Cluster.Type, "cluster.type", "", "gossip", "Determine how the cluster handles membership and state sharing. Choose from [static, gossip]") flags.StringVarP(&srv.Config.Metric.Service, "metric.service", "", "nop", "Default URI on which pilosa should listen.") flags.StringVarP(&srv.Config.Metric.Host, "metric.host", "", "", "Default URI to send metrics.") + flags.BoolVarP((&srv.Config.Metric.Diagnostics), "metric.diagnostics", "", true, "Enabled diagnostics reporting.") flags.DurationVarP((*time.Duration)(&srv.Config.Metric.PollInterval), "metric.poll-interval", "", time.Minute*0, "Polling interval metrics.") SetTLSConfig(flags, &srv.Config.TLS.CertificatePath, &srv.Config.TLS.CertificateKeyPath, &srv.Config.TLS.SkipVerify) } diff --git a/diagnostics/diagnostics.go b/diagnostics/diagnostics.go new file mode 100644 index 000000000..c653a7052 --- /dev/null +++ b/diagnostics/diagnostics.go @@ -0,0 +1,216 @@ +package diagnostics + +import ( + "bytes" + "encoding/json" + "fmt" + "io" + "io/ioutil" + "log" + "net/http" + "strconv" + "strings" + "sync" + "time" + + "github.com/sony/gobreaker" +) + +// TODO: unique Cluster ID + +// Default version check URL. +const ( + 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 + lastVersion string + startTime int64 + start time.Time + + 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 { + + return &Diagnostics{ + closing: make(chan struct{}), + host: host, + VersionURL: DefaultVersionCheckURL, + startTime: time.Now().Unix(), + start: time.Now(), + client: http.DefaultClient, + metrics: make(map[string]interface{}), + 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) +} + +// SetInterval of the diagnostic go routine and match with the circuit breaker timeout. +func (d *Diagnostics) SetInterval(i time.Duration) { + d.interval = i +} + +// 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"] = (time.Now().Unix() - d.startTime) + buf, _ := d.Encode() + d.mu.Unlock() + + _, 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 + body, err := ioutil.ReadAll(resp.Body) + if err != nil { + return nil, err + } + return body, nil + }) + + return err +} + +// Open configures the circuit breaker used by the HTTP client. +func (d *Diagnostics) Open() { + var st gobreaker.Settings + if d.interval > 0 { + st.Timeout = d.interval * 2 + } + d.cb = gobreaker.NewCircuitBreaker(st) + + d.logger().Printf("Pilosa is currently configured to send small diagnostics reports to our team every hour. More information here: https://www.pilosa.com/docs/latest/administration/#diagnostics") +} + +// 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) + } + + // Same a version as last test + if rsp.Version == d.lastVersion { + return nil + } + + d.lastVersion = rsp.Version + 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 Pilosa %s. A newer version (%s) is available: https://github.com/pilosa/pilosa/releases", d.version, value) + } else if localVersion[1] < currentVersion[1] { // Minor + return fmt.Errorf("Warning: You are running Pilosa %s. The latest Minor release is %s: https://github.com/pilosa/pilosa/releases", d.version, value) + } else if localVersion[2] < currentVersion[2] { // Patch + return fmt.Errorf("There is a new patch release of Pilosa availbale: %s: https://github.com/pilosa/pilosa/releases", value) + } + + return nil +} + +// 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 interface{}) { + d.mu.Lock() + defer d.mu.Unlock() + d.metrics[name] = value +} + +// 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..6ad0cb773 --- /dev/null +++ b/diagnostics/diagnostics_test.go @@ -0,0 +1,158 @@ +package diagnostics_test + +import ( + "encoding/json" + "io/ioutil" + "net/http" + "net/http/httptest" + "reflect" + "runtime" + "strings" + "testing" + + "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) + d.Open() + defer d.Close() + + d.Set("gg", 10) + d.Set("ss", "ss") + + 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(`{"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.Encode() + 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") + d.Open() + defer d.Close() + + version := "v0.1.1" + d.SetVersion(version) + + err := d.CompareVersion("v1.7.0") + if !strings.Contains(err.Error(), "A newer version") { + t.Fatalf("Expected a newer version is available, actual error: %s", err) + } + err = d.CompareVersion("1.7.0") + if !strings.Contains(err.Error(), "A newer version") { + t.Fatalf("Expected a newer version is available, 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 release 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 +} + +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.Set("cc", 1) + d.Set("gg", "test") + } + }) +} diff --git a/docs/administration.md b/docs/administration.md index ed4da61ee..23b5d17f0 100644 --- a/docs/administration.md +++ b/docs/administration.md @@ -112,6 +112,27 @@ Note: This will only work when the replication factor is >= 2 - Restart the cluster - Wait for the 1st sync (10 minutes) to validate Index connections +#### Diagnostics + +Each Pilosa cluster is configured by default to share anonymous usage details with Pilosa Corp. These metrics allow us to understand how Pilosa is used by the community and improve the technology to suit your needs. Diagnostics are sent to Pilosa every hour. Each of the metrics are detailed below as well as opt-out instructions. + +Version: Version string of the build. +Host: Host URI. +Cluster: List of nodes in the Cluster. +NumNodes: Number of nodes in the Cluster. +NumCPU: Number of Cores per Node +BSIEnabled: Bit Slice Index Frames in use. +TimeQuantumEnabled: Time Quantum Frames in use. +InverseEnabled: Inverse Frames in use. +NumIndexes: Number of Indexes in the Cluster. +NumFrames: Number of Frames in the Cluster. +NumSlices: Number of Slices in the Cluster. +NumViews: Number of Views in the Cluster. +OpenFiles: Open file handle count. +GoRoutines: Go routine count. + +You can opt-out of the Pilosa diagnostics reporting by setting either the command line configuration option `--metric.diagnostics=false`, use the `PILOSA_METRIC_DIAGNOSTICS` environment variable, or the TOML configuration file `[metric]` `diagnostics` option. + #### Metrics Pilosa can be configured to emit metrics pertaining to its internal processes in one of two formats: Expvar or StatsD. Metric recording is disabled by default. diff --git a/docs/configuration.md b/docs/configuration.md index 10325b09f..0396e2e95 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -219,6 +219,19 @@ Any flag that has a value that is a comma separated list on the command line bec poll-interval = "0m15s" ``` +##### Metric Diagnostics + +* Description: Enable diagnostic reporting. To disable diagnostics set to false. +* Flag: `metric.diagnostics` +* Env: `PILOSA_METRIC_DIAGNOSTICS` +* Config: + + ```toml + [metric] + diagnostics = true + ``` + + ##### TLS Certificate * Description: Path to the TLS certificate to use for serving HTTPS. Usually has one of`.crt` or `.pem` extensions. diff --git a/holder.go b/holder.go index f3b4247d0..4241dcd85 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 e94475d02..610fe6255 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" "golang.org/x/net/context" ) @@ -41,6 +42,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. @@ -59,14 +61,16 @@ type Server struct { // Cluster configuration. // Host is replaced with actual host after opening if port is ":0". - Network string - URI *URI - Cluster *Cluster + Network string + URI *URI + Cluster *Cluster + diagnostics *diagnostics.Diagnostics // Background monitoring intervals. AntiEntropyInterval time.Duration PollingInterval time.Duration MetricInterval time.Duration + DiagnosticInterval time.Duration // TLS configuration TLS *tls.Config @@ -88,12 +92,14 @@ func NewServer() *Server { Handler: NewHandler(), Broadcaster: NopBroadcaster, BroadcastReceiver: NopBroadcastReceiver, + diagnostics: diagnostics.New(DefaultDiagnosticServer), Network: "tcp", AntiEntropyInterval: DefaultAntiEntropyInterval, PollingInterval: DefaultPollingInterval, MetricInterval: 0, + DiagnosticInterval: 0, LogOutput: os.Stderr, } @@ -191,10 +197,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 } @@ -223,10 +230,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() @@ -240,7 +247,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. @@ -259,9 +266,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. @@ -490,9 +497,66 @@ func (s *Server) checkMaxSlices(scheme string, hostPort string) (map[string]uint return s.defaultClient.MaxSliceByIndex(ctx) } +// monitorDiagnostics periodically polls the the Pilosa Indexes for cluster info. +func (s *Server) monitorDiagnostics() { + if s.DiagnosticInterval <= 0 { + s.Logger().Printf("diagnostics disabled") + return + } + + s.diagnostics.SetLogger(s.LogOutput) + s.diagnostics.SetVersion(Version) + s.diagnostics.SetInterval(s.DiagnosticInterval) + s.diagnostics.Open() + s.diagnostics.Set("Host", s.URI.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 + + // Flush the diagnostics metrics at startup, then on each tick interval + flush := func() { + 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() + } + + ticker := time.NewTicker(s.DiagnosticInterval) + defer ticker.Stop() + flush() + for { + // Wait for tick or a close. + select { + case <-s.closing: + return + case <-ticker.C: + flush() + } + } +} + // monitorRuntime periodically polls the Go runtime metrics. func (s *Server) monitorRuntime() { - // Disable metrics when poll interval is zero + // Disable metrics when poll interval is zero. if s.MetricInterval <= 0 { return } @@ -512,18 +576,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) @@ -541,7 +605,7 @@ func (s *Server) createDefaultClient() { s.defaultClient = NewInternalHTTPClientFromURI(nil, &ClientOptions{TLS: s.TLS}) } -// CountOpenFiles on opperating systems that support lsof +// CountOpenFiles on operating systems that support lsof. func CountOpenFiles() int { count := 0 diff --git a/server/server.go b/server/server.go index 22e965b55..7cc8c5dec 100644 --- a/server/server.go +++ b/server/server.go @@ -30,6 +30,7 @@ import ( "time" "crypto/tls" + "github.com/pilosa/pilosa" "github.com/pilosa/pilosa/gossip" "github.com/pilosa/pilosa/statsd" @@ -43,6 +44,9 @@ func init() { const ( // DefaultDataDir is the default data directory. DefaultDataDir = "~/.pilosa" + + // DefaultDiagnosticsInterval is the default sync frequency diagnostic metrics. + DefaultDiagnosticsInterval = 1 * time.Hour ) // Command represents the state of the pilosa server command. @@ -143,6 +147,9 @@ func (m *Command) SetupServer() error { m.Server.Logger().Printf("Using data from: %s\n", m.Config.DataDir) m.Server.Holder.Path = m.Config.DataDir m.Server.MetricInterval = time.Duration(m.Config.Metric.PollInterval) + if m.Config.Metric.Diagnostics { + m.Server.DiagnosticInterval = time.Duration(DefaultDiagnosticsInterval) + } m.Server.Holder.Stats, err = NewStatsClient(m.Config.Metric.Service, m.Config.Metric.Host) if err != nil { return err 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()