Influxdata
/
influxdb
mirror of https://github.com/influxdata/influxdb.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Merge pull request #4506 from influxdb/enterprise_stats 

Enterprise registration as a service
pull/4612/head
Philip O'Toole 2015-10-20 08:57:19 -07:00
parent ddca29a87b 3cfe4e8da5
commit 956efaeb94
10 changed files with 326 additions and 108 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
CHANGELOG.md
View File

@ -19,7 +19,7 @@
- [#4409](https://github.com/influxdb/influxdb/pull/4409): wire up INTO queries.
- [#4379](https://github.com/influxdb/influxdb/pull/4379): Auto-create database for UDP input.
- [#4375](https://github.com/influxdb/influxdb/pull/4375): Add Subscriptions so data can be 'forked' out of InfluxDB to another third party.
- [#4459](https://github.com/influxdb/influxdb/pull/4459): Register with Enterprise service if token available.
- [#4506](https://github.com/influxdb/influxdb/pull/4506): Register with Enterprise service and upload stats, if token is available.
- [#4501](https://github.com/influxdb/influxdb/pull/4501): Allow filtering SHOW MEASUREMENTS by regex.
### Bugfixes

10
cmd/influxd/run/config.go
View File

@ -22,20 +22,20 @@ import (
"github.com/influxdb/influxdb/services/httpd"
"github.com/influxdb/influxdb/services/opentsdb"
"github.com/influxdb/influxdb/services/precreator"
"github.com/influxdb/influxdb/services/registration"
"github.com/influxdb/influxdb/services/retention"
"github.com/influxdb/influxdb/services/subscriber"
"github.com/influxdb/influxdb/services/udp"
"github.com/influxdb/influxdb/tsdb"
)
const DefaultEnterpriseURL = "https://enterprise.influxdata.com"
// Config represents the configuration format for the influxd binary.
type Config struct {
Meta *meta.Config `toml:"meta"`
Data tsdb.Config `toml:"data"`
Cluster cluster.Config `toml:"cluster"`
Retention retention.Config `toml:"retention"`
Registration registration.Config `toml:"registration"`
Precreator precreator.Config `toml:"shard-precreation"`
Admin admin.Config `toml:"admin"`
@ -54,19 +54,15 @@ type Config struct {
// Server reporting
ReportingDisabled bool `toml:"reporting-disabled"`
// Server registration
EnterpriseURL string `toml:"enterprise-url"`
EnterpriseToken string `toml:"enterprise-token"`
}
// NewConfig returns an instance of Config with reasonable defaults.
func NewConfig() *Config {
c := &Config{}
c.EnterpriseURL = DefaultEnterpriseURL
c.Meta = meta.NewConfig()
c.Data = tsdb.NewConfig()
c.Cluster = cluster.NewConfig()
c.Registration = registration.NewConfig()
c.Precreator = precreator.NewConfig()
c.Admin = admin.NewConfig()

16
cmd/influxd/run/config_test.go
View File

@ -13,8 +13,6 @@ func TestConfig_Parse(t *testing.T) {
// Parse configuration.
var c run.Config
if _, err := toml.Decode(`
enterprise-token = "deadbeef"
[meta]
dir = "/tmp/meta"
@ -57,9 +55,7 @@ enabled = true
}
// Validate configuration.
if c.EnterpriseToken != "deadbeef" {
t.Fatalf("unexpected Enterprise token: %s", c.EnterpriseToken)
} else if c.Meta.Dir != "/tmp/meta" {
if c.Meta.Dir != "/tmp/meta" {
t.Fatalf("unexpected meta dir: %s", c.Meta.Dir)
} else if c.Data.Dir != "/tmp/data" {
t.Fatalf("unexpected data dir: %s", c.Data.Dir)
@ -91,8 +87,6 @@ func TestConfig_Parse_EnvOverride(t *testing.T) {
// Parse configuration.
var c run.Config
if _, err := toml.Decode(`
enterprise-token = "deadbeef"
[meta]
dir = "/tmp/meta"
@ -131,10 +125,6 @@ enabled = true
t.Fatal(err)
}
if err := os.Setenv("INFLUXDB_ENTERPRISE_TOKEN", "wheresthebeef"); err != nil {
t.Fatalf("failed to set env var: %v", err)
}
if err := os.Setenv("INFLUXDB_UDP_BIND_ADDRESS", ":1234"); err != nil {
t.Fatalf("failed to set env var: %v", err)
}
@ -147,10 +137,6 @@ enabled = true
t.Fatalf("failed to apply env overrides: %v", err)
}
if c.EnterpriseToken != "wheresthebeef" {
t.Fatalf("unexpected Enterprise token: %s", c.EnterpriseToken)
}
if c.UDPs[0].BindAddress != ":4444" {
t.Fatalf("unexpected udp bind address: %s", c.UDPs[0].BindAddress)
}

81
cmd/influxd/run/server.go
View File

@ -2,9 +2,7 @@ package run
import (
"bytes"
"encoding/json"
"fmt"
"io/ioutil"
"log"
"net"
"net/http"
@ -26,6 +24,7 @@ import (
"github.com/influxdb/influxdb/services/httpd"
"github.com/influxdb/influxdb/services/opentsdb"
"github.com/influxdb/influxdb/services/precreator"
"github.com/influxdb/influxdb/services/registration"
"github.com/influxdb/influxdb/services/retention"
"github.com/influxdb/influxdb/services/snapshotter"
"github.com/influxdb/influxdb/services/subscriber"
@ -76,8 +75,6 @@ type Server struct {
// Server reporting and registration
reportingDisabled bool
enterpriseURL string
enterpriseToken string
// Profiling
CPUProfile string
@ -104,8 +101,6 @@ func NewServer(c *Config, buildInfo *BuildInfo) (*Server, error) {
Monitor: monitor.New(c.Monitor),
reportingDisabled: c.ReportingDisabled,
enterpriseURL: c.EnterpriseURL,
enterpriseToken: c.EnterpriseToken,
}
// Copy TSDB configuration.
@ -162,6 +157,7 @@ func NewServer(c *Config, buildInfo *BuildInfo) (*Server, error) {
// Append services.
s.appendClusterService(c.Cluster)
s.appendPrecreatorService(c.Precreator)
s.appendRegistrationService(c.Registration)
s.appendSnapshotterService()
s.appendCopierService()
s.appendAdminService(c.Admin)
@ -299,6 +295,21 @@ func (s *Server) appendPrecreatorService(c precreator.Config) error {
return nil
}
func (s *Server) appendRegistrationService(c registration.Config) error {
if !c.Enabled {
return nil
}
srv, err := registration.NewService(c, s.buildInfo.Version)
if err != nil {
return err
}
srv.MetaStore = s.MetaStore
srv.Monitor = s.Monitor
s.Services = append(s.Services, srv)
return nil
}
func (s *Server) appendUDPService(c udp.Config) {
if !c.Enabled {
return
@ -403,11 +414,6 @@ func (s *Server) Open() error {
go s.startServerReporting()
}
// Register server
if err := s.registerServer(); err != nil {
log.Printf("failed to register server: %s", err.Error())
}
return nil
}(); err != nil {
@ -519,59 +525,6 @@ func (s *Server) reportServer() {
go client.Post("http://m.influxdb.com:8086/db/reporting/series?u=reporter&p=influxdb", "application/json", data)
}
// registerServer registers the server on start-up.
func (s *Server) registerServer() error {
if s.enterpriseToken == "" {
return nil
}
clusterID, err := s.MetaStore.ClusterID()
if err != nil {
log.Printf("failed to retrieve cluster ID for registration: %s", err.Error())
return err
}
hostname, err := os.Hostname()
if err != nil {
return err
}
j := map[string]interface{}{
"cluster_id": fmt.Sprintf("%d", clusterID),
"server_id": fmt.Sprintf("%d", s.MetaStore.NodeID()),
"host": hostname,
"product": "influxdb",
"version": s.buildInfo.Version,
}
b, err := json.Marshal(j)
if err != nil {
return err
}
url := fmt.Sprintf("%s/api/v1/servers?token=%s", s.enterpriseURL, s.enterpriseToken)
go func() {
client := http.Client{Timeout: time.Duration(5 * time.Second)}
resp, err := client.Post(url, "application/json", bytes.NewBuffer(b))
if err != nil {
log.Printf("failed to register server with %s: %s", s.enterpriseURL, err.Error())
return
}
defer resp.Body.Close()
if resp.StatusCode == http.StatusCreated {
return
}
body, err := ioutil.ReadAll(resp.Body)
if err != nil {
log.Printf("failed to read response from registration server: %s", err.Error())
return
}
log.Printf("failed to register server with %s: received code %s, body: %s", s.enterpriseURL, resp.Status, string(body))
}()
return nil
}
// monitorErrorChan reads an error channel and resends it through the server.
func (s *Server) monitorErrorChan(ch <-chan error) {
for {

11
etc/config.sample.toml
View File

@ -8,9 +8,14 @@
# Change this option to true to disable reporting.
reporting-disabled = false
# Enterprise registration control
# enterprise-url = "https://enterprise.influxdata.com" # The Enterprise server URL
# enterprise-token = "" # Registration token for Enterprise server
###
### Enterprise registration control
###
[registration]
# enabled = true
# url = "https://enterprise.influxdata.com" # The Enterprise server URL
# token = "" # Registration token for Enterprise server
###
### [meta]

24
monitor/service.go
View File

@ -171,8 +171,8 @@ func (m *Monitor) DeregisterDiagnosticsClient(name string) {
// Statistics returns the combined statistics for all expvar data. The given
// tags are added to each of the returned statistics.
func (m *Monitor) Statistics(tags map[string]string) ([]*statistic, error) {
statistics := make([]*statistic, 0)
func (m *Monitor) Statistics(tags map[string]string) ([]*Statistic, error) {
statistics := make([]*Statistic, 0)
expvar.Do(func(kv expvar.KeyValue) {
// Skip built-in expvar stats.
@ -180,7 +180,7 @@ func (m *Monitor) Statistics(tags map[string]string) ([]*statistic, error) {
return
}
statistic := &statistic{
statistic := &Statistic{
Tags: make(map[string]string),
Values: make(map[string]interface{}),
}
@ -246,7 +246,7 @@ func (m *Monitor) Statistics(tags map[string]string) ([]*statistic, error) {
})
// Add Go memstats.
statistic := &statistic{
statistic := &Statistic{
Name: "runtime",
Tags: make(map[string]string),
Values: make(map[string]interface{}),
@ -388,16 +388,16 @@ func (m *Monitor) storeStatistics() {
}
}
// statistic represents the information returned by a single monitor client.
type statistic struct {
Name string
Tags map[string]string
Values map[string]interface{}
// Statistic represents the information returned by a single monitor client.
type Statistic struct {
Name string `json:"name"`
Tags map[string]string `json:"tags"`
Values map[string]interface{} `json:"values"`
}
// newStatistic returns a new statistic object.
func newStatistic(name string, tags map[string]string, values map[string]interface{}) *statistic {
return &statistic{
func newStatistic(name string, tags map[string]string, values map[string]interface{}) *Statistic {
return &Statistic{
Name: name,
Tags: tags,
Values: values,
@ -405,7 +405,7 @@ func newStatistic(name string, tags map[string]string, values map[string]interfa
}
// valueNames returns a sorted list of the value names, if any.
func (s *statistic) valueNames() []string {
func (s *Statistic) valueNames() []string {
a := make([]string, 0, len(s.Values))
for k, _ := range s.Values {
a = append(a, k)

2
monitor/statement_executor.go
View File

@ -11,7 +11,7 @@ import (
// StatementExecutor translates InfluxQL queries to Monitor methods.
type StatementExecutor struct {
Monitor interface {
Statistics(map[string]string) ([]*statistic, error)
Statistics(map[string]string) ([]*Statistic, error)
Diagnostics() (map[string]*Diagnostic, error)
}
}

27
services/registration/config.go Normal file
View File

@ -0,0 +1,27 @@
package registration
import (
"time"
"github.com/influxdb/influxdb/toml"
)
const (
DefaultURL = "https://enterprise.influxdata.com"
DefaultStatsInterval = time.Minute
)
type Config struct {
Enabled bool `toml:"enabled"`
URL string `toml:"url"`
Token string `toml:"token"`
StatsInterval toml.Duration `toml:"stats-interval"`
}
func NewConfig() Config {
return Config{
Enabled: true,
URL: DefaultURL,
StatsInterval: toml.Duration(DefaultStatsInterval),
}
}

33
services/registration/config_test.go Normal file
View File

@ -0,0 +1,33 @@
package registration_test
import (
"testing"
"time"
"github.com/BurntSushi/toml"
"github.com/influxdb/influxdb/services/registration"
)
func TestConfig_Parse(t *testing.T) {
// Parse configuration.
var c registration.Config
if _, err := toml.Decode(`
enabled = true
url = "a.b.c"
token = "1234"
stats-interval = "1s"
`, &c); err != nil {
t.Fatal(err)
}
// Validate configuration.
if c.Enabled != true {
t.Fatalf("unexpected enabled state: %v", c.Enabled)
} else if c.URL != "a.b.c" {
t.Fatalf("unexpected Enterprise URL: %s", c.URL)
} else if c.Token != "1234" {
t.Fatalf("unexpected Enterprise URL: %s", c.URL)
} else if time.Duration(c.StatsInterval) != time.Second {
t.Fatalf("unexpected stats interval: %v", c.StatsInterval)
}
}

218
services/registration/service.go Normal file
View File

@ -0,0 +1,218 @@
package registration
import (
"bytes"
"encoding/json"
"fmt"
"io/ioutil"
"log"
"net/http"
"net/url"
"os"
"sync"
"time"
"github.com/influxdb/influxdb/monitor"
)
// Service represents the registration service.
type Service struct {
MetaStore interface {
ClusterID() (uint64, error)
NodeID() uint64
}
Monitor interface {
Statistics(tags map[string]string) ([]*monitor.Statistic, error)
RegisterDiagnosticsClient(name string, client monitor.DiagsClient)
}
enabled bool
url *url.URL
token string
statsInterval time.Duration
version string
mu sync.Mutex
lastContact time.Time
wg sync.WaitGroup
done chan struct{}
logger *log.Logger
}
// NewService returns a configured registration service.
func NewService(c Config, version string) (*Service, error) {
url, err := url.Parse(c.URL)
if err != nil {
return nil, err
}
return &Service{
enabled: c.Enabled,
url: url,
token: c.Token,
statsInterval: time.Duration(c.StatsInterval),
version: version,
done: make(chan struct{}),
logger: log.New(os.Stderr, "[registration] ", log.LstdFlags),
}, nil
}
// Open starts retention policy enforcement.
func (s *Service) Open() error {
if !s.enabled {
return nil
}
s.logger.Println("Starting registration service")
if err := s.registerServer(); err != nil {
return err
}
// Register diagnostics if a Monitor service is available.
if s.Monitor != nil {
s.Monitor.RegisterDiagnosticsClient("registration", s)
}
s.wg.Add(1)
go s.reportStats()
return nil
}
// Close stops retention policy enforcement.
func (s *Service) Close() error {
s.logger.Println("registration service terminating")
close(s.done)
s.wg.Wait()
return nil
}
func (s *Service) Diagnostics() (*monitor.Diagnostic, error) {
diagnostics := map[string]interface{}{
"URL": s.url.String(),
"token": s.token,
"last_contact": s.getLastContact().String(),
}
return monitor.DiagnosticFromMap(diagnostics), nil
}
// registerServer registers the server.
func (s *Service) registerServer() error {
if !s.enabled || s.token == "" {
return nil
}
clusterID, err := s.MetaStore.ClusterID()
if err != nil {
s.logger.Printf("failed to retrieve cluster ID for registration: %s", err.Error())
return err
}
hostname, err := os.Hostname()
if err != nil {
return err
}
j := map[string]interface{}{
"cluster_id": fmt.Sprintf("%d", clusterID),
"server_id": fmt.Sprintf("%d", s.MetaStore.NodeID()),
"host": hostname,
"product": "influxdb",
"version": s.version,
}
b, err := json.Marshal(j)
if err != nil {
return err
}
url := fmt.Sprintf("%s/api/v1/servers?token=%s", s.url.String(), s.token)
s.wg.Add(1)
go func() {
defer s.wg.Done()
client := http.Client{Timeout: time.Duration(5 * time.Second)}
resp, err := client.Post(url, "application/json", bytes.NewBuffer(b))
if err != nil {
s.logger.Printf("failed to register server with %s: %s", s.url.String(), err.Error())
return
}
s.updateLastContact(time.Now().UTC())
defer resp.Body.Close()
if resp.StatusCode == http.StatusCreated {
return
}
body, err := ioutil.ReadAll(resp.Body)
if err != nil {
s.logger.Printf("failed to read response from registration server: %s", err.Error())
return
}
s.logger.Printf("failed to register server with %s: received code %s, body: %s", s.url.String(), resp.Status, string(body))
}()
return nil
}
func (s *Service) reportStats() {
defer s.wg.Done()
if s.token == "" {
// No reporting, for now, without token.
return
}
statsURL := fmt.Sprintf("%s/api/v1/stats/influxdb?token=%s", s.url.String(), s.token)
clusterID, err := s.MetaStore.ClusterID()
if err != nil {
s.logger.Printf("failed to retrieve cluster ID for registration -- aborting stats upload: %s", err.Error())
return
}
t := time.NewTicker(s.statsInterval)
for {
select {
case <-t.C:
stats, err := s.Monitor.Statistics(nil)
if err != nil {
s.logger.Printf("failed to retrieve statistics: %s", err.Error())
continue
}
o := map[string]interface{}{
"cluster_id": fmt.Sprintf("%d", clusterID),
"server_id": fmt.Sprintf("%d", s.MetaStore.NodeID()),
"stats": stats,
}
b, err := json.Marshal(o)
if err != nil {
s.logger.Printf("failed to JSON-encode stats: %s", err.Error())
continue
}
client := http.Client{Timeout: time.Duration(5 * time.Second)}
resp, err := client.Post(statsURL, "application/json", bytes.NewBuffer(b))
if err != nil {
s.logger.Printf("failed to post statistics to %s: %s", statsURL, err.Error())
continue
}
s.updateLastContact(time.Now().UTC())
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
s.logger.Printf("failed to post statistics to %s: repsonse code: %d", statsURL, resp.StatusCode)
continue
}
case <-s.done:
return
}
}
}
func (s *Service) updateLastContact(t time.Time) {
s.mu.Lock()
defer s.mu.Unlock()
s.lastContact = t
}
func (s *Service) getLastContact() time.Time {
s.mu.Lock()
defer s.mu.Unlock()
return s.lastContact
}