diff --git a/Gopkg.lock b/Gopkg.lock index c77dc951a..af6174c08 100644 --- a/Gopkg.lock +++ b/Gopkg.lock @@ -205,12 +205,6 @@ packages = ["."] revision = "bb4de0191aa41b5507caa14b0650cdbddcd9280b" -[[projects]] - name = "github.com/sony/gobreaker" - packages = ["."] - revision = "e9556a45379ef1da12e54847edb2fb3d7d566f36" - version = "0.3.0" - [[projects]] branch = "master" name = "github.com/spf13/afero" 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.go b/diagnostics.go new file mode 100644 index 000000000..3ad07ed42 --- /dev/null +++ b/diagnostics.go @@ -0,0 +1,324 @@ +// Copyright 2017 Pilosa Corp. +// +// 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 pilosa + +import ( + "bytes" + "encoding/json" + "fmt" + "io" + "io/ioutil" + "log" + "net/http" + "strconv" + "strings" + "sync" + "time" +) + +// Default version check URL. +const ( + defaultVersionCheckURL = "https://diagnostics.pilosa.com/v0/version" +) + +type versionResponse struct { + Version string `json:"version"` + Message string `json:"message"` +} + +// DiagnosticsCollector represents a collector/sender of diagnostics data. +type DiagnosticsCollector struct { + mu sync.Mutex + host string + VersionURL string + version string + lastVersion string + startTime int64 + start time.Time + + metrics map[string]interface{} + + client *http.Client + + logOutput io.Writer + + server *Server +} + +// NewDiagnosticsCollector returns a new DiagnosticsCollector given an addr in the format "hostname:port". +func NewDiagnosticsCollector(host string) *DiagnosticsCollector { + return &DiagnosticsCollector{ + host: host, + VersionURL: defaultVersionCheckURL, + startTime: time.Now().Unix(), + start: time.Now(), + client: &http.Client{Timeout: 10 * time.Second}, + metrics: make(map[string]interface{}), + logOutput: ioutil.Discard, + } +} + +// SetVersion of locally running Pilosa Cluster to check against master. +func (d *DiagnosticsCollector) SetVersion(v string) { + d.version = v + d.Set("Version", v) +} + +// Flush sends the current metrics. +func (d *DiagnosticsCollector) Flush() error { + d.mu.Lock() + defer d.mu.Unlock() + d.metrics["Uptime"] = (time.Now().Unix() - d.startTime) + buf, err := d.encode() + if err != nil { + return err + } + 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 + } + // Intentionally ignoring response body, as user does not need to be notified of error. + defer resp.Body.Close() + return nil +} + +// CheckVersion of the local build against Pilosa master. +func (d *DiagnosticsCollector) 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 version has not changed since the last check, return + 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 *DiagnosticsCollector) 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] && localVersion[0] == currentVersion[0] { // 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] && localVersion[0] == currentVersion[0] && localVersion[1] == currentVersion[1] { // Patch + return fmt.Errorf("There is a new patch release of Pilosa available: %s: https://github.com/pilosa/pilosa/releases", value) + } + + return nil +} + +// Encode metrics maps into the json message format. +func (d *DiagnosticsCollector) encode() ([]byte, error) { + return json.Marshal(d.metrics) +} + +// Set adds a key value metric. +func (d *DiagnosticsCollector) Set(name string, value interface{}) { + switch v := value.(type) { + case string: + if v == "" { + // Do not set empty string + return + } + } + d.mu.Lock() + defer d.mu.Unlock() + d.metrics[name] = value +} + +// SetLogger Set the logger output type. +func (d *DiagnosticsCollector) SetLogger(logger io.Writer) { + d.logOutput = logger +} + +// logger returns a logger that writes to LogOutput. +func (d *DiagnosticsCollector) logger() *log.Logger { + return log.New(d.logOutput, "", log.LstdFlags) +} + +// logErr logs the error and returns true if an error exists +func (d *DiagnosticsCollector) logErr(err error) bool { + if err != nil { + d.logOutput.Write([]byte(err.Error())) + return true + } + return false +} + +// EnrichWithOSInfo adds OS information to the diagnostics payload. +func (d *DiagnosticsCollector) EnrichWithOSInfo() { + uptime, err := d.server.SystemInfo.Uptime() + if !d.logErr(err) { + d.Set("HostUptime", uptime) + } + platform, err := d.server.SystemInfo.Platform() + if !d.logErr(err) { + d.Set("OSPlatform", platform) + } + family, err := d.server.SystemInfo.Family() + if !d.logErr(err) { + d.Set("OSFamily", family) + } + version, err := d.server.SystemInfo.OSVersion() + if !d.logErr(err) { + d.Set("OSVersion", version) + } + kernelVersion, err := d.server.SystemInfo.KernelVersion() + if !d.logErr(err) { + d.Set("OSKernelVersion", kernelVersion) + } +} + +// EnrichWithMemoryInfo adds memory information to the diagnostics payload. +func (d *DiagnosticsCollector) EnrichWithMemoryInfo() { + memFree, err := d.server.SystemInfo.MemFree() + if !d.logErr(err) { + d.Set("MemFree", memFree) + } + memTotal, err := d.server.SystemInfo.MemTotal() + if !d.logErr(err) { + d.Set("MemTotal", memTotal) + } + memUsed, err := d.server.SystemInfo.MemUsed() + if !d.logErr(err) { + d.Set("MemUsed", memUsed) + } +} + +// EnrichWithSchemaProperties adds schema info to the diagnostics payload. +func (d *DiagnosticsCollector) EnrichWithSchemaProperties() { + var numSlices uint64 + numFrames := 0 + numIndexes := 0 + bsiFieldCount := 0 + timeQuantumEnabled := false + + for _, index := range d.server.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") + 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 +} + +// SystemInfo collects information about the host OS. +type SystemInfo interface { + Uptime() (uint64, error) + Platform() (string, error) + Family() (string, error) + OSVersion() (string, error) + KernelVersion() (string, error) + MemFree() (uint64, error) + MemTotal() (uint64, error) + MemUsed() (uint64, error) +} + +// NewNopSystemInfo creates a no-op implementation of SystemInfo. +func NewNopSystemInfo() *NopSystemInfo { + return &NopSystemInfo{} +} + +// NopSystemInfo is a no-op implementation of SystemInfo. +type NopSystemInfo struct { +} + +// Uptime is a no-op implementation of SystemInfo.Uptime. +func (n *NopSystemInfo) Uptime() (uint64, error) { + return 0, nil +} + +// Platform is a no-op implementation of SystemInfo.Platform. +func (n *NopSystemInfo) Platform() (string, error) { + return "", nil +} + +// Family is a no-op implementation of SystemInfo.Family. +func (n *NopSystemInfo) Family() (string, error) { + return "", nil +} + +// OSVersion is a no-op implementation of SystemInfo.OSVersion. +func (n *NopSystemInfo) OSVersion() (string, error) { + return "", nil +} + +// KernelVersion is a no-op implementation of SystemInfo.KernelVersion. +func (n *NopSystemInfo) KernelVersion() (string, error) { + return "", nil +} + +// MemFree is a no-op implementation of SystemInfo.MemFree. +func (n *NopSystemInfo) MemFree() (uint64, error) { + return 0, nil +} + +// MemTotal is a no-op implementation of SystemInfo.MemTotal. +func (n *NopSystemInfo) MemTotal() (uint64, error) { + return 0, nil +} + +// MemUsed is a no-op implementation of SystemInfo.MemUsed. +func (n *NopSystemInfo) MemUsed() (uint64, error) { + return 0, nil +} diff --git a/diagnostics/diagnostics.go b/diagnostics/diagnostics.go deleted file mode 100644 index 60248d6bb..000000000 --- a/diagnostics/diagnostics.go +++ /dev/null @@ -1,267 +0,0 @@ -// Copyright 2017 Pilosa Corp. -// -// 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 diagnostics - -import ( - "bytes" - "encoding/json" - "fmt" - "io" - "io/ioutil" - "log" - "net/http" - "strconv" - "strings" - "sync" - "time" - - "github.com/shirou/gopsutil/host" - "github.com/shirou/gopsutil/mem" - "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] && localVersion[0] == currentVersion[0] { // 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] && localVersion[0] == currentVersion[0] && localVersion[1] == currentVersion[1] { // Patch - return fmt.Errorf("There is a new patch release of Pilosa available: %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) -} - -// EnrichWithOSInfo adds OS information to the diagnostics payload. -func (d *Diagnostics) EnrichWithOSInfo() { - osInfo, err := host.Info() - if err != nil { - d.logOutput.Write([]byte(err.Error())) - } - d.Set("HostUptime", osInfo.Uptime) - - platform, family, version, err := host.PlatformInformation() - if err != nil { - d.logOutput.Write([]byte(err.Error())) - } - d.Set("OSPlatform", platform) - d.Set("OSFamily", family) - d.Set("OSVersion", version) - - kernelVersion, err := host.KernelVersion() - if err != nil { - d.logOutput.Write([]byte(err.Error())) - } - d.Set("OSKernelVersion", kernelVersion) -} - -// EnrichWithMemoryInfo adds memory information to the diagnostics payload. -func (d *Diagnostics) EnrichWithMemoryInfo() { - memory, err := mem.VirtualMemory() - if err != nil { - d.logOutput.Write([]byte(err.Error())) - } - d.Set("MemFree", memory.Free) - d.Set("MemTotal", memory.Total) - d.Set("MemUsed", memory.Used) - -} - -// 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_internal_test.go similarity index 82% rename from diagnostics/diagnostics_test.go rename to diagnostics_internal_test.go index 8f85a57db..eb2498297 100644 --- a/diagnostics/diagnostics_test.go +++ b/diagnostics_internal_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,20 @@ 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 +53,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 +69,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 +78,33 @@ func TestDiagnosticsVersion_Parse(t *testing.T) { } func TestDiagnosticsVersion_Compare(t *testing.T) { - d := diagnostics.New("localhost:10101") - d.Open() - defer d.Close() + d := NewDiagnosticsCollector("localhost:10101") 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 +118,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 +129,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 +143,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..23dd0f0d0 100644 --- a/gc.go +++ b/gc.go @@ -14,6 +14,9 @@ package pilosa +// Ensure nopGCNotifier implements interface. +var _ GCNotifier = &nopGCNotifier{} + // GCNotifier represents an interface for garbage collection notificationss. type GCNotifier interface { Close() @@ -29,10 +32,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/gopsutil/systeminfo.go b/gopsutil/systeminfo.go new file mode 100644 index 000000000..e6285ade8 --- /dev/null +++ b/gopsutil/systeminfo.go @@ -0,0 +1,101 @@ +package gopsutil + +import ( + "github.com/pilosa/pilosa" + "github.com/shirou/gopsutil/host" + "github.com/shirou/gopsutil/mem" +) + +var _ pilosa.SystemInfo = NewSystemInfo() + +// SystemInfo is an implementation of pilosa.SystemInfo that uses gopsutil to collect information about the host OS. +type SystemInfo struct { + platform string + family string + osVersion string +} + +// Uptime returns the system uptime in seconds. +func (s *SystemInfo) Uptime() (uptime uint64, err error) { + hostInfo, err := host.Info() + if err != nil { + return 0, err + } + return hostInfo.Uptime, nil +} + +// collectPlatformInfo fetches and caches system platform information. +func (s *SystemInfo) collectPlatformInfo() error { + var err error + if s.platform == "" { + s.platform, s.family, s.osVersion, err = host.PlatformInformation() + if err != nil { + return err + } + } + return nil +} + +// Platform returns the system platform. +func (s *SystemInfo) Platform() (string, error) { + err := s.collectPlatformInfo() + if err != nil { + return "", err + } + return s.platform, nil +} + +// Family returns the system family. +func (s *SystemInfo) Family() (string, error) { + err := s.collectPlatformInfo() + if err != nil { + return "", err + } + return s.family, err +} + +// OSVersion returns the OS Version. +func (s *SystemInfo) OSVersion() (string, error) { + err := s.collectPlatformInfo() + if err != nil { + return "", err + } + return s.osVersion, err +} + +// MemFree returns the amount of free memory in bytes. +func (s *SystemInfo) MemFree() (uint64, error) { + memInfo, err := mem.VirtualMemory() + if err != nil { + return 0, err + } + return memInfo.Free, err +} + +// MemTotal returns the amount of total memory in bytes. +func (s *SystemInfo) MemTotal() (uint64, error) { + memInfo, err := mem.VirtualMemory() + if err != nil { + return 0, err + } + return memInfo.Total, err +} + +// MemUsed returns the amount of used memory in bytes. +func (s *SystemInfo) MemUsed() (uint64, error) { + memInfo, err := mem.VirtualMemory() + if err != nil { + return 0, err + } + return memInfo.Used, err +} + +// KernelVersion returns the kernel version as a string. +func (s *SystemInfo) KernelVersion() (string, error) { + return host.KernelVersion() +} + +// NewSystemInfo is a constructor for the gopsutil implementation of SystemInfo. +func NewSystemInfo() *SystemInfo { + return &SystemInfo{} +} diff --git a/gopsutil/systeminfo_test.go b/gopsutil/systeminfo_test.go new file mode 100644 index 000000000..5d96a499c --- /dev/null +++ b/gopsutil/systeminfo_test.go @@ -0,0 +1,63 @@ +package gopsutil_test + +import ( + "log" + "testing" + + "github.com/pilosa/pilosa" + "github.com/pilosa/pilosa/gopsutil" +) + +func TestSystemInfo(t *testing.T) { + var systemInfo pilosa.SystemInfo = gopsutil.NewSystemInfo() + + // Uptime()(uint64, error) + // Platform()(string, error) + // Family()(string, error) + // OSVersion()(string, error) + // KernelVersion()(string, error) + // MemFree()(uint64, error) + // MemTotal()(uint64, error) + // MemUsed()(uint64, error) + // + uptime, err := systemInfo.Uptime() + if err != nil || uptime == 0 { + t.Fatalf("Error collecting uptime (error: %v)", err) + } + + platform, err := systemInfo.Platform() + if err != nil { + t.Fatalf("Error getting platform. (platform: %v, error: %v)", platform, err) + } + + family, err := systemInfo.Family() + if err != nil { + t.Fatalf("Error getting OS family. (family: %v, error: %v)", family, err) + } + + osversion, err := systemInfo.OSVersion() + if err != nil { + t.Fatalf("Error getting OS version. (osversion: %v, error: %v)", osversion, err) + } + + kernelversion, err := systemInfo.KernelVersion() + if err != nil { + t.Fatalf("Error getting kernel version. (kernelversion: %v, error: %v)", kernelversion, err) + } + + memfree, err := systemInfo.MemFree() + if err != nil { + t.Fatalf("Error getting memfree. (memfree: %v, error: %v)", memfree, err) + } + + memused, err := systemInfo.MemUsed() + if err != nil { + t.Fatalf("Error getting memused. (memused: %v, error: %v)", memused, err) + } + + memtotal, err := systemInfo.MemTotal() + log.Println(memtotal) + if err != nil { + t.Fatalf("Error getting memtotal. (memtotal: %v, error: %v)", memtotal, err) + } +} diff --git a/server.go b/server.go index 531609278..54bacda1d 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,8 @@ type Server struct { NodeID string URI URI Cluster *Cluster - diagnostics *diagnostics.Diagnostics + diagnostics *DiagnosticsCollector + SystemInfo SystemInfo GCNotifier GCNotifier @@ -100,7 +100,8 @@ func NewServer() *Server { Handler: NewHandler(), Broadcaster: NopBroadcaster, BroadcastReceiver: NopBroadcastReceiver, - diagnostics: diagnostics.New(DefaultDiagnosticServer), + diagnostics: NewDiagnosticsCollector(DefaultDiagnosticServer), + SystemInfo: NewNopSystemInfo(), Network: "tcp", @@ -115,6 +116,7 @@ func NewServer() *Server { s.logger = log.New(s.LogOutput, "", log.LstdFlags) s.Handler.Holder = s.Holder + s.diagnostics.server = s return s } @@ -603,15 +605,16 @@ func (s *Server) mergeRemoteStatus(ns *internal.NodeStatus) error { // monitorDiagnostics periodically polls the Pilosa Indexes for cluster info. func (s *Server) monitorDiagnostics() { - if s.DiagnosticInterval <= 0 { + // Do not send more than once a minute + if s.DiagnosticInterval < time.Minute { s.Logger().Printf("diagnostics disabled") return + } else { + s.Logger().Printf("Pilosa is currently configured to send small diagnostics reports to our team every %v. More information here: https://www.pilosa.com/docs/latest/administration/#diagnostics", s.DiagnosticInterval) } 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.NodeIDs(), ",")) s.diagnostics.Set("NumNodes", len(s.Cluster.Nodes)) @@ -622,15 +625,18 @@ 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.diagnostics.CheckVersion() - s.diagnostics.Flush() + err = s.diagnostics.Flush() + if err != nil { + s.Logger().Printf("Diagnostics error: %s", err) + } } ticker := time.NewTicker(s.DiagnosticInterval) @@ -725,39 +731,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) -} diff --git a/server/server.go b/server/server.go index 240b8b16b..eac98121d 100644 --- a/server/server.go +++ b/server/server.go @@ -34,6 +34,7 @@ import ( "github.com/pilosa/pilosa" "github.com/pilosa/pilosa/gcnotify" + "github.com/pilosa/pilosa/gopsutil" "github.com/pilosa/pilosa/gossip" "github.com/pilosa/pilosa/statsd" ) @@ -152,6 +153,7 @@ func (m *Command) SetupServer() error { if m.Config.Metric.Diagnostics { m.Server.DiagnosticInterval = time.Duration(DefaultDiagnosticsInterval) } + m.Server.SystemInfo = gopsutil.NewSystemInfo() m.Server.GCNotifier = gcnotify.NewActiveGCNotifier() m.Server.Holder.Stats, err = NewStatsClient(m.Config.Metric.Service, m.Config.Metric.Host) if err != nil {