mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-08-28 10:54:59 +00:00
Merge pull request #1166 from codysoyland/cluster-resize-refactor-diagnostics
Refactor Diagnostics: Move to main package, remove gobreaker, add SystemInfo abstraction for gopsutil dependency injection
This commit is contained in:
Cody Soyland
GitHub
commit
70ec216acf
10 changed files with 527 additions and 352 deletions
6
Gopkg.lock
generated
6
Gopkg.lock
generated
|
|
@ -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"
|
||||
|
|
|
|||
|
|
@ -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
|
||||
}
|
||||
|
|
|
|||
324
diagnostics.go
Normal file
324
diagnostics.go
Normal file
|
|
@ -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
|
||||
}
|
||||
|
|
@ -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
|
||||
}
|
||||
|
|
@ -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)
|
||||
7
gc.go
7
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
|
||||
}
|
||||
|
|
|
|||
101
gopsutil/systeminfo.go
Normal file
101
gopsutil/systeminfo.go
Normal file
|
|
@ -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{}
|
||||
}
|
||||
63
gopsutil/systeminfo_test.go
Normal file
63
gopsutil/systeminfo_test.go
Normal file
|
|
@ -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)
|
||||
}
|
||||
}
|
||||
58
server.go
58
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)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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 {
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue