diff --git a/broadcast.go b/broadcast.go index a4e6403c1..de43f3b85 100644 --- a/broadcast.go +++ b/broadcast.go @@ -63,17 +63,17 @@ var NopBroadcaster Broadcaster type nopBroadcaster struct{} -// SendSync A no-op implemenetation of Broadcaster SendSync method. +// SendSync A no-op implementation of Broadcaster SendSync method. func (n *nopBroadcaster) SendSync(pb proto.Message) error { return nil } -// SendAsync A no-op implemenetation of Broadcaster SendAsync method. +// SendAsync A no-op implementation of Broadcaster SendAsync method. func (n *nopBroadcaster) SendAsync(pb proto.Message) error { return nil } -// SendTo is a no-op implemenetation of Broadcaster SendTo method. +// SendTo is a no-op implementation of Broadcaster SendTo method. func (c *nopBroadcaster) SendTo(to *Node, pb proto.Message) error { return nil } @@ -112,7 +112,7 @@ var NopGossiper Gossiper type nopGossiper struct{} -// SendAsync A no-op implemenetation of Gossiper SendAsync method. +// SendAsync A no-op implementation of Gossiper SendAsync method. func (n *nopGossiper) SendAsync(pb proto.Message) error { return nil } diff --git a/diagnostics/diagnostics.go b/diagnostics.go similarity index 73% rename from diagnostics/diagnostics.go rename to diagnostics.go index 60248d6bb..602e910e0 100644 --- a/diagnostics/diagnostics.go +++ b/diagnostics.go @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -package diagnostics +package pilosa import ( "bytes" @@ -36,7 +36,7 @@ import ( // Default version check URL. const ( - DefaultVersionCheckURL = "https://diagnostics.pilosa.com/v0/version" + defaultVersionCheckURL = "https://diagnostics.pilosa.com/v0/version" ) type versionResponse struct { @@ -44,11 +44,9 @@ type versionResponse struct { Message string `json:"message"` } -// Diagnostics represents a client to the Pilosa cluster. -type Diagnostics struct { +// DiagnosticsCollector represents a collector/sender of diagnostics data +type DiagnosticsCollector struct { mu sync.Mutex - wg sync.WaitGroup - closing chan struct{} host string VersionURL string version string @@ -65,13 +63,12 @@ type Diagnostics struct { 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 { +// New returns a pointer to a new DiagnosticsCollector Client given an addr in the format "hostname:port". +func NewDiagnosticsCollector(host string) *DiagnosticsCollector { - return &Diagnostics{ - closing: make(chan struct{}), + return &DiagnosticsCollector{ host: host, - VersionURL: DefaultVersionCheckURL, + VersionURL: defaultVersionCheckURL, startTime: time.Now().Unix(), start: time.Now(), client: http.DefaultClient, @@ -81,37 +78,21 @@ func New(host string) *Diagnostics { } // SetVersion of locally running Pilosa Cluster to check against master. -func (d *Diagnostics) SetVersion(v string) { +func (d *DiagnosticsCollector) 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) { +func (d *DiagnosticsCollector) 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 { +func (d *DiagnosticsCollector) Flush() error { d.mu.Lock() d.metrics["Uptime"] = (time.Now().Unix() - d.startTime) - buf, _ := d.Encode() + buf, _ := d.encode() d.mu.Unlock() _, err := d.cb.Execute(func() (interface{}, error) { @@ -135,7 +116,7 @@ func (d *Diagnostics) Flush() error { } // Open configures the circuit breaker used by the HTTP client. -func (d *Diagnostics) Open() { +func (d *DiagnosticsCollector) Open() { var st gobreaker.Settings if d.interval > 0 { st.Timeout = d.interval * 2 @@ -145,15 +126,8 @@ func (d *Diagnostics) Open() { 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 { +func (d *DiagnosticsCollector) CheckVersion() error { var rsp versionResponse req, err := http.NewRequest("GET", d.VersionURL, nil) resp, err := d.client.Do(req) @@ -174,15 +148,15 @@ func (d *Diagnostics) CheckVersion() error { } d.lastVersion = rsp.Version - if err := d.CompareVersion(rsp.Version); err != nil { + 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 { +// compareVersion check version strings. +func (d *DiagnosticsCollector) compareVersion(value string) error { currentVersion := VersionSegments(value) localVersion := VersionSegments(d.version) @@ -198,29 +172,29 @@ func (d *Diagnostics) CompareVersion(value string) error { } // Encode metrics maps into the json message format. -func (d *Diagnostics) Encode() ([]byte, error) { +func (d *DiagnosticsCollector) encode() ([]byte, error) { return json.Marshal(d.metrics) } // Set adds a key value metric. -func (d *Diagnostics) Set(name string, value interface{}) { +func (d *DiagnosticsCollector) 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) { +func (d *DiagnosticsCollector) SetLogger(logger io.Writer) { d.logOutput = logger } // logger returns a logger that writes to LogOutput. -func (d *Diagnostics) logger() *log.Logger { +func (d *DiagnosticsCollector) logger() *log.Logger { return log.New(d.logOutput, "", log.LstdFlags) } // EnrichWithOSInfo adds OS information to the diagnostics payload. -func (d *Diagnostics) EnrichWithOSInfo() { +func (d *DiagnosticsCollector) EnrichWithOSInfo() { osInfo, err := host.Info() if err != nil { d.logOutput.Write([]byte(err.Error())) @@ -243,7 +217,7 @@ func (d *Diagnostics) EnrichWithOSInfo() { } // EnrichWithMemoryInfo adds memory information to the diagnostics payload. -func (d *Diagnostics) EnrichWithMemoryInfo() { +func (d *DiagnosticsCollector) EnrichWithMemoryInfo() { memory, err := mem.VirtualMemory() if err != nil { d.logOutput.Write([]byte(err.Error())) @@ -254,6 +228,37 @@ func (d *Diagnostics) EnrichWithMemoryInfo() { } +// EnrichWithSchemaProperties adds schema info to the diagnostics payload. +func (d *DiagnosticsCollector) EnrichWithSchemaProperties(holder *Holder) { + var numSlices uint64 + numFrames := 0 + numIndexes := 0 + bsiFieldCount := 0 + timeQuantumEnabled := false + + for _, index := range holder.Indexes() { + numSlices += index.MaxSlice() + 1 + numIndexes += 1 + for _, frame := range index.Frames() { + numFrames += 1 + if frame.rangeEnabled { + if fields, err := frame.GetFields(); err == nil { + bsiFieldCount += len(fields) + } + } + if frame.TimeQuantum() != "" { + timeQuantumEnabled = true + } + } + } + + d.Set("NumIndexes", numIndexes) + d.Set("NumFrames", numFrames) + d.Set("NumSlices", numSlices) + d.Set("BSIFieldCount", bsiFieldCount) + d.Set("TimeQuantumEnabled", timeQuantumEnabled) +} + // VersionSegments returns the numeric segments of the version as a slice of ints. func VersionSegments(segments string) []int { segments = strings.Trim(segments, "v") diff --git a/diagnostics/diagnostics_test.go b/diagnostics_test.go similarity index 83% rename from diagnostics/diagnostics_test.go rename to diagnostics_test.go index 8f85a57db..9d8e35d76 100644 --- a/diagnostics/diagnostics_test.go +++ b/diagnostics_test.go @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -package diagnostics_test +package pilosa import ( "encoding/json" @@ -23,25 +23,21 @@ import ( "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 := NewDiagnosticsCollector(server.URL) d.SetLogger(ioutil.Discard) d.Open() - defer d.Close() d.Set("gg", 10) d.Set("ss", "ss") - data, err := d.Encode() + data, err := d.encode() if err != nil { t.Fatal(err) } @@ -58,7 +54,7 @@ func TestDiagnosticsClient(t *testing.T) { // Test the metrics after a flush. d.Flush() - data, err = d.Encode() + data, err = d.encode() if err != nil { t.Fatal(err) } @@ -74,7 +70,7 @@ func TestDiagnosticsClient(t *testing.T) { func TestDiagnosticsVersion_Parse(t *testing.T) { version := "0.1.1" - vs := diagnostics.VersionSegments(version) + vs := VersionSegments(version) output := []int{0, 1, 1} if !reflect.DeepEqual(vs, output) { @@ -83,35 +79,34 @@ func TestDiagnosticsVersion_Parse(t *testing.T) { } func TestDiagnosticsVersion_Compare(t *testing.T) { - d := diagnostics.New("localhost:10101") + d := NewDiagnosticsCollector("localhost:10101") d.Open() - defer d.Close() version := "v0.1.1" d.SetVersion(version) - err := d.CompareVersion("v1.7.0") + 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") + 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") + 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") + 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") + err = d.compareVersion("0.1.1") if err != nil { t.Fatalf("Versions should match") } d.SetVersion("v1.7.0") - err = d.CompareVersion("0.7.2") + err = d.compareVersion("0.7.2") if err != nil { t.Fatalf("Local version is greater") } @@ -125,11 +120,9 @@ func TestDiagnosticsVersion_Check(t *testing.T) { Version: "1.1.1", }) })) - defer server.Close() // Create a new client. - d := diagnostics.New("localhost:10101") - defer d.Close() + d := NewDiagnosticsCollector("localhost:10101") version := "0.1.1" d.SetVersion(version) @@ -138,10 +131,6 @@ func TestDiagnosticsVersion_Check(t *testing.T) { 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 { @@ -156,12 +145,10 @@ func compareJSON(a, b []byte) (bool, error) { func BenchmarkDiagnostics(b *testing.B) { // Mock server. server := httptest.NewServer(nil) - defer server.Close() // Create a new client. - d := diagnostics.New(server.URL) + d := NewDiagnosticsCollector(server.URL) d.SetLogger(ioutil.Discard) - defer d.Close() prev := runtime.GOMAXPROCS(4) defer runtime.GOMAXPROCS(prev) diff --git a/gc.go b/gc.go index a260cc0b5..3f036bdcd 100644 --- a/gc.go +++ b/gc.go @@ -29,10 +29,10 @@ var NopGCNotifier GCNotifier type nopGCNotifier struct{} -// Close is a no-op implemenetation of GCNotifier Close method. +// Close is a no-op implementation of GCNotifier Close method. func (n *nopGCNotifier) Close() {} -// AfterGC is a no-op implemenetation of GCNotifier AfterGC method. +// AfterGC is a no-op implementation of GCNotifier AfterGC method. func (c *nopGCNotifier) AfterGC() <-chan struct{} { return nil } diff --git a/server.go b/server.go index 531609278..82ca2db2d 100644 --- a/server.go +++ b/server.go @@ -32,7 +32,6 @@ import ( "time" "github.com/gogo/protobuf/proto" - "github.com/pilosa/pilosa/diagnostics" "github.com/pilosa/pilosa/internal" "golang.org/x/sync/errgroup" @@ -70,7 +69,7 @@ type Server struct { NodeID string URI URI Cluster *Cluster - diagnostics *diagnostics.Diagnostics + diagnostics *DiagnosticsCollector GCNotifier GCNotifier @@ -100,7 +99,7 @@ func NewServer() *Server { Handler: NewHandler(), Broadcaster: NopBroadcaster, BroadcastReceiver: NopBroadcastReceiver, - diagnostics: diagnostics.New(DefaultDiagnosticServer), + diagnostics: NewDiagnosticsCollector(DefaultDiagnosticServer), Network: "tcp", @@ -622,13 +621,13 @@ func (s *Server) monitorDiagnostics() { // Flush the diagnostics metrics at startup, then on each tick interval flush := func() { - enrichDiagnosticsWithSchemaProperties(s.diagnostics, s.Holder) openFiles, err := CountOpenFiles() if err == nil { s.diagnostics.Set("OpenFiles", openFiles) } s.diagnostics.Set("GoRoutines", runtime.NumGoroutine()) s.diagnostics.EnrichWithMemoryInfo() + s.diagnostics.EnrichWithSchemaProperties(s.Holder) s.diagnostics.CheckVersion() s.diagnostics.Flush() } @@ -725,39 +724,3 @@ type StatusHandler interface { ClusterStatus() (proto.Message, error) HandleRemoteStatus(proto.Message) error } - -type diagnosticsFrameProperties struct { - BSIFieldCount int - TimeQuantumEnabled bool -} - -func enrichDiagnosticsWithSchemaProperties(d *diagnostics.Diagnostics, holder *Holder) { - // NOTE: this function is not in the diagnostics package, since circular imports are not allowed. - var numSlices uint64 - numFrames := 0 - numIndexes := 0 - bsiFieldCount := 0 - timeQuantumEnabled := false - - for _, index := range holder.Indexes() { - numSlices += index.MaxSlice() + 1 - numIndexes += 1 - for _, frame := range index.Frames() { - numFrames += 1 - if frame.rangeEnabled { - if fields, err := frame.GetFields(); err == nil { - bsiFieldCount += len(fields) - } - } - if frame.TimeQuantum() != "" { - timeQuantumEnabled = true - } - } - } - - d.Set("NumIndexes", numIndexes) - d.Set("NumFrames", numFrames) - d.Set("NumSlices", numSlices) - d.Set("BSIFieldCount", bsiFieldCount) - d.Set("TimeQuantumEnabled", timeQuantumEnabled) -}