483 lines
12 KiB
Go
483 lines
12 KiB
Go
package monitor_test
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"expvar"
|
|
"fmt"
|
|
"os"
|
|
"reflect"
|
|
"sort"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/influxdata/influxdb/v2/models"
|
|
"github.com/influxdata/influxdb/v2/toml"
|
|
"github.com/influxdata/influxdb/v2/v1/monitor"
|
|
"github.com/influxdata/influxdb/v2/v1/services/meta"
|
|
"go.uber.org/zap"
|
|
"go.uber.org/zap/zaptest/observer"
|
|
)
|
|
|
|
func TestMonitor_Open(t *testing.T) {
|
|
s := monitor.New(nil, monitor.Config{})
|
|
if err := s.Open(); err != nil {
|
|
t.Fatalf("unexpected open error: %s", err)
|
|
}
|
|
|
|
// Verify that opening twice is fine.
|
|
if err := s.Open(); err != nil {
|
|
s.Close()
|
|
t.Fatalf("unexpected error on second open: %s", err)
|
|
}
|
|
|
|
if err := s.Close(); err != nil {
|
|
t.Fatalf("unexpected close error: %s", err)
|
|
}
|
|
|
|
// Verify that closing twice is fine.
|
|
if err := s.Close(); err != nil {
|
|
t.Fatalf("unexpected error on second close: %s", err)
|
|
}
|
|
}
|
|
|
|
func TestMonitor_SetPointsWriter_StoreEnabled(t *testing.T) {
|
|
var mc MetaClient
|
|
mc.CreateDatabaseWithRetentionPolicyFn = func(name string, spec *meta.RetentionPolicySpec) (*meta.DatabaseInfo, error) {
|
|
return &meta.DatabaseInfo{Name: name}, nil
|
|
}
|
|
|
|
config := monitor.NewConfig()
|
|
s := monitor.New(nil, config)
|
|
s.MetaClient = &mc
|
|
core, logs := observer.New(zap.DebugLevel)
|
|
s.WithLogger(zap.New(core))
|
|
|
|
// Setting the points writer should open the monitor.
|
|
var pw PointsWriter
|
|
if err := s.SetPointsWriter(&pw); err != nil {
|
|
t.Fatalf("unexpected open error: %s", err)
|
|
}
|
|
defer s.Close()
|
|
|
|
// Verify that the monitor was opened by looking at the log messages.
|
|
if logs.FilterMessage("Starting monitor service").Len() == 0 {
|
|
t.Errorf("monitor system was never started")
|
|
}
|
|
}
|
|
|
|
func TestMonitor_SetPointsWriter_StoreDisabled(t *testing.T) {
|
|
s := monitor.New(nil, monitor.Config{})
|
|
core, logs := observer.New(zap.DebugLevel)
|
|
s.WithLogger(zap.New(core))
|
|
|
|
// Setting the points writer should open the monitor.
|
|
var pw PointsWriter
|
|
if err := s.SetPointsWriter(&pw); err != nil {
|
|
t.Fatalf("unexpected open error: %s", err)
|
|
}
|
|
defer s.Close()
|
|
|
|
// Verify that the monitor was not opened by looking at the log messages.
|
|
if logs.FilterMessage("Starting monitor system").Len() > 0 {
|
|
t.Errorf("monitor system should not have been started")
|
|
}
|
|
}
|
|
|
|
func TestMonitor_StoreStatistics(t *testing.T) {
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
ch := make(chan models.Points)
|
|
|
|
var mc MetaClient
|
|
mc.CreateDatabaseWithRetentionPolicyFn = func(name string, spec *meta.RetentionPolicySpec) (*meta.DatabaseInfo, error) {
|
|
if got, want := name, monitor.DefaultStoreDatabase; got != want {
|
|
t.Errorf("unexpected database: got=%q want=%q", got, want)
|
|
}
|
|
if got, want := spec.Name, monitor.MonitorRetentionPolicy; got != want {
|
|
t.Errorf("unexpected retention policy: got=%q want=%q", got, want)
|
|
}
|
|
if spec.Duration != nil {
|
|
if got, want := *spec.Duration, monitor.MonitorRetentionPolicyDuration; got != want {
|
|
t.Errorf("unexpected duration: got=%q want=%q", got, want)
|
|
}
|
|
} else {
|
|
t.Error("expected duration in retention policy spec")
|
|
}
|
|
if spec.ReplicaN != nil {
|
|
if got, want := *spec.ReplicaN, monitor.MonitorRetentionPolicyReplicaN; got != want {
|
|
t.Errorf("unexpected replica number: got=%q want=%q", got, want)
|
|
}
|
|
} else {
|
|
t.Error("expected replica number in retention policy spec")
|
|
}
|
|
return &meta.DatabaseInfo{Name: name}, nil
|
|
}
|
|
|
|
var pw PointsWriter
|
|
pw.WritePointsFn = func(database, policy string, points models.Points) error {
|
|
// Verify that we are attempting to write to the correct database.
|
|
if got, want := database, monitor.DefaultStoreDatabase; got != want {
|
|
t.Errorf("unexpected database: got=%q want=%q", got, want)
|
|
}
|
|
if got, want := policy, monitor.MonitorRetentionPolicy; got != want {
|
|
t.Errorf("unexpected retention policy: got=%q want=%q", got, want)
|
|
}
|
|
|
|
// Attempt to write the points to the main goroutine.
|
|
select {
|
|
case <-ctx.Done():
|
|
case ch <- points:
|
|
}
|
|
return nil
|
|
}
|
|
|
|
config := monitor.NewConfig()
|
|
config.StoreInterval = toml.Duration(10 * time.Millisecond)
|
|
s := monitor.New(nil, config)
|
|
s.MetaClient = &mc
|
|
s.PointsWriter = &pw
|
|
|
|
if err := s.Open(); err != nil {
|
|
t.Fatalf("unexpected error: %s", err)
|
|
}
|
|
defer s.Close()
|
|
defer cancel()
|
|
|
|
timer := time.NewTimer(100 * time.Millisecond)
|
|
select {
|
|
case points := <-ch:
|
|
timer.Stop()
|
|
|
|
// Search for the runtime statistic.
|
|
found := false
|
|
for _, pt := range points {
|
|
if !bytes.Equal(pt.Name(), []byte("runtime")) {
|
|
continue
|
|
}
|
|
|
|
// There should be a hostname.
|
|
if got := pt.Tags().GetString("hostname"); len(got) == 0 {
|
|
t.Errorf("expected hostname tag")
|
|
}
|
|
// This should write on an exact interval of 10 milliseconds.
|
|
if got, want := pt.Time(), pt.Time().Truncate(10*time.Millisecond); got != want {
|
|
t.Errorf("unexpected time: got=%q want=%q", got, want)
|
|
}
|
|
found = true
|
|
break
|
|
}
|
|
|
|
if !found {
|
|
t.Error("unable to find runtime statistic")
|
|
}
|
|
case <-timer.C:
|
|
t.Errorf("timeout while waiting for statistics to be written")
|
|
}
|
|
}
|
|
|
|
func TestMonitor_Reporter(t *testing.T) {
|
|
reporter := ReporterFunc(func(tags map[string]string) []models.Statistic {
|
|
return []models.Statistic{
|
|
{
|
|
Name: "foo",
|
|
Tags: tags,
|
|
Values: map[string]interface{}{
|
|
"value": "bar",
|
|
},
|
|
},
|
|
}
|
|
})
|
|
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
ch := make(chan models.Points)
|
|
|
|
var mc MetaClient
|
|
mc.CreateDatabaseWithRetentionPolicyFn = func(name string, spec *meta.RetentionPolicySpec) (*meta.DatabaseInfo, error) {
|
|
return &meta.DatabaseInfo{Name: name}, nil
|
|
}
|
|
|
|
var pw PointsWriter
|
|
pw.WritePointsFn = func(database, policy string, points models.Points) error {
|
|
// Attempt to write the points to the main goroutine.
|
|
select {
|
|
case <-ctx.Done():
|
|
case ch <- points:
|
|
}
|
|
return nil
|
|
}
|
|
|
|
config := monitor.NewConfig()
|
|
config.StoreInterval = toml.Duration(10 * time.Millisecond)
|
|
s := monitor.New(reporter, config)
|
|
s.MetaClient = &mc
|
|
s.PointsWriter = &pw
|
|
|
|
if err := s.Open(); err != nil {
|
|
t.Fatalf("unexpected error: %s", err)
|
|
}
|
|
defer s.Close()
|
|
defer cancel()
|
|
|
|
timer := time.NewTimer(100 * time.Millisecond)
|
|
select {
|
|
case points := <-ch:
|
|
timer.Stop()
|
|
|
|
// Look for the statistic.
|
|
found := false
|
|
for _, pt := range points {
|
|
if !bytes.Equal(pt.Name(), []byte("foo")) {
|
|
continue
|
|
}
|
|
found = true
|
|
break
|
|
}
|
|
|
|
if !found {
|
|
t.Error("unable to find foo statistic")
|
|
}
|
|
case <-timer.C:
|
|
t.Errorf("timeout while waiting for statistics to be written")
|
|
}
|
|
}
|
|
|
|
func expvarMap(name string, tags map[string]string, fields map[string]interface{}) *expvar.Map {
|
|
m := new(expvar.Map).Init()
|
|
eName := new(expvar.String)
|
|
eName.Set(name)
|
|
m.Set("name", eName)
|
|
|
|
var eTags *expvar.Map
|
|
if len(tags) > 0 {
|
|
eTags = new(expvar.Map).Init()
|
|
for k, v := range tags {
|
|
kv := new(expvar.String)
|
|
kv.Set(v)
|
|
eTags.Set(k, kv)
|
|
}
|
|
m.Set("tags", eTags)
|
|
}
|
|
|
|
var eFields *expvar.Map
|
|
if len(fields) > 0 {
|
|
eFields = new(expvar.Map).Init()
|
|
for k, v := range fields {
|
|
switch v := v.(type) {
|
|
case float64:
|
|
kv := new(expvar.Float)
|
|
kv.Set(v)
|
|
eFields.Set(k, kv)
|
|
case int:
|
|
kv := new(expvar.Int)
|
|
kv.Set(int64(v))
|
|
eFields.Set(k, kv)
|
|
case string:
|
|
kv := new(expvar.String)
|
|
kv.Set(v)
|
|
eFields.Set(k, kv)
|
|
}
|
|
}
|
|
m.Set("values", eFields)
|
|
}
|
|
return m
|
|
}
|
|
|
|
func TestMonitor_Expvar(t *testing.T) {
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
ch := make(chan models.Points)
|
|
|
|
var mc MetaClient
|
|
mc.CreateDatabaseWithRetentionPolicyFn = func(name string, spec *meta.RetentionPolicySpec) (*meta.DatabaseInfo, error) {
|
|
return &meta.DatabaseInfo{Name: name}, nil
|
|
}
|
|
|
|
var pw PointsWriter
|
|
pw.WritePointsFn = func(database, policy string, points models.Points) error {
|
|
// Attempt to write the points to the main goroutine.
|
|
select {
|
|
case <-ctx.Done():
|
|
case ch <- points:
|
|
}
|
|
return nil
|
|
}
|
|
|
|
config := monitor.NewConfig()
|
|
config.StoreInterval = toml.Duration(10 * time.Millisecond)
|
|
s := monitor.New(nil, config)
|
|
s.MetaClient = &mc
|
|
s.PointsWriter = &pw
|
|
|
|
expvar.Publish("expvar1", expvarMap(
|
|
"expvar1",
|
|
map[string]string{
|
|
"region": "uswest2",
|
|
},
|
|
map[string]interface{}{
|
|
"value": 2.0,
|
|
},
|
|
))
|
|
expvar.Publish("expvar2", expvarMap(
|
|
"expvar2",
|
|
map[string]string{
|
|
"region": "uswest2",
|
|
},
|
|
nil,
|
|
))
|
|
expvar.Publish("expvar3", expvarMap(
|
|
"expvar3",
|
|
nil,
|
|
map[string]interface{}{
|
|
"value": 2,
|
|
},
|
|
))
|
|
|
|
bad := new(expvar.String)
|
|
bad.Set("badentry")
|
|
expvar.Publish("expvar4", bad)
|
|
|
|
if err := s.Open(); err != nil {
|
|
t.Fatalf("unexpected error: %s", err)
|
|
}
|
|
defer s.Close()
|
|
defer cancel()
|
|
|
|
hostname, _ := os.Hostname()
|
|
timer := time.NewTimer(100 * time.Millisecond)
|
|
select {
|
|
case points := <-ch:
|
|
timer.Stop()
|
|
|
|
// Look for the statistic.
|
|
var found1, found3 bool
|
|
for _, pt := range points {
|
|
if bytes.Equal(pt.Name(), []byte("expvar1")) {
|
|
if got, want := pt.Tags().HashKey(), []byte(fmt.Sprintf(",hostname=%s,region=uswest2", hostname)); !reflect.DeepEqual(got, want) {
|
|
t.Errorf("unexpected expvar1 tags: got=%v want=%v", string(got), string(want))
|
|
}
|
|
fields, _ := pt.Fields()
|
|
if got, want := fields, models.Fields(map[string]interface{}{
|
|
"value": 2.0,
|
|
}); !reflect.DeepEqual(got, want) {
|
|
t.Errorf("unexpected expvar1 fields: got=%v want=%v", got, want)
|
|
}
|
|
found1 = true
|
|
} else if bytes.Equal(pt.Name(), []byte("expvar2")) {
|
|
t.Error("found expvar2 statistic")
|
|
} else if bytes.Equal(pt.Name(), []byte("expvar3")) {
|
|
if got, want := pt.Tags().HashKey(), []byte(fmt.Sprintf(",hostname=%s", hostname)); !reflect.DeepEqual(got, want) {
|
|
t.Errorf("unexpected expvar3 tags: got=%v want=%v", string(got), string(want))
|
|
}
|
|
fields, _ := pt.Fields()
|
|
if got, want := fields, models.Fields(map[string]interface{}{
|
|
"value": int64(2),
|
|
}); !reflect.DeepEqual(got, want) {
|
|
t.Errorf("unexpected expvar3 fields: got=%v want=%v", got, want)
|
|
}
|
|
found3 = true
|
|
}
|
|
}
|
|
|
|
if !found1 {
|
|
t.Error("unable to find expvar1 statistic")
|
|
}
|
|
if !found3 {
|
|
t.Error("unable to find expvar3 statistic")
|
|
}
|
|
case <-timer.C:
|
|
t.Errorf("timeout while waiting for statistics to be written")
|
|
}
|
|
}
|
|
|
|
func TestMonitor_QuickClose(t *testing.T) {
|
|
var mc MetaClient
|
|
mc.CreateDatabaseWithRetentionPolicyFn = func(name string, spec *meta.RetentionPolicySpec) (*meta.DatabaseInfo, error) {
|
|
return &meta.DatabaseInfo{Name: name}, nil
|
|
}
|
|
|
|
var pw PointsWriter
|
|
config := monitor.NewConfig()
|
|
config.StoreInterval = toml.Duration(24 * time.Hour)
|
|
s := monitor.New(nil, config)
|
|
s.MetaClient = &mc
|
|
s.PointsWriter = &pw
|
|
|
|
if err := s.Open(); err != nil {
|
|
t.Fatalf("unexpected error: %s", err)
|
|
}
|
|
|
|
if err := s.Close(); err != nil {
|
|
t.Fatalf("unexpected error: %s", err)
|
|
}
|
|
}
|
|
|
|
func TestStatistic_ValueNames(t *testing.T) {
|
|
statistic := monitor.Statistic{
|
|
Statistic: models.Statistic{
|
|
Name: "foo",
|
|
Values: map[string]interface{}{
|
|
"abc": 1.0,
|
|
"def": 2.0,
|
|
},
|
|
},
|
|
}
|
|
|
|
names := statistic.ValueNames()
|
|
if got, want := names, []string{"abc", "def"}; !reflect.DeepEqual(got, want) {
|
|
t.Errorf("unexpected value names: got=%v want=%v", got, want)
|
|
}
|
|
}
|
|
|
|
func TestStatistics_Sort(t *testing.T) {
|
|
statistics := []*monitor.Statistic{
|
|
{Statistic: models.Statistic{Name: "b"}},
|
|
{Statistic: models.Statistic{Name: "a"}},
|
|
{Statistic: models.Statistic{Name: "c"}},
|
|
}
|
|
|
|
sort.Sort(monitor.Statistics(statistics))
|
|
names := make([]string, 0, len(statistics))
|
|
for _, stat := range statistics {
|
|
names = append(names, stat.Name)
|
|
}
|
|
|
|
if got, want := names, []string{"a", "b", "c"}; !reflect.DeepEqual(got, want) {
|
|
t.Errorf("incorrect sorting of statistics: got=%v want=%v", got, want)
|
|
}
|
|
}
|
|
|
|
type ReporterFunc func(tags map[string]string) []models.Statistic
|
|
|
|
func (f ReporterFunc) Statistics(tags map[string]string) []models.Statistic {
|
|
return f(tags)
|
|
}
|
|
|
|
type PointsWriter struct {
|
|
WritePointsFn func(database, policy string, points models.Points) error
|
|
}
|
|
|
|
func (pw *PointsWriter) WritePoints(database, policy string, points models.Points) error {
|
|
if pw.WritePointsFn != nil {
|
|
return pw.WritePointsFn(database, policy, points)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
type MetaClient struct {
|
|
CreateDatabaseWithRetentionPolicyFn func(name string, spec *meta.RetentionPolicySpec) (*meta.DatabaseInfo, error)
|
|
DatabaseFn func(name string) *meta.DatabaseInfo
|
|
}
|
|
|
|
func (m *MetaClient) CreateDatabaseWithRetentionPolicy(name string, spec *meta.RetentionPolicySpec) (*meta.DatabaseInfo, error) {
|
|
return m.CreateDatabaseWithRetentionPolicyFn(name, spec)
|
|
}
|
|
|
|
func (m *MetaClient) Database(name string) *meta.DatabaseInfo {
|
|
if m.DatabaseFn != nil {
|
|
return m.DatabaseFn(name)
|
|
}
|
|
return nil
|
|
}
|