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

Refactor registration as a service 

Registration also involves statistics and diagnostics upload, for the
purposes of remote management. This means there will be long-running
goroutines in effect. Therefore move the code to a service model.
pull/4506/head
Philip O'Toole 2015-10-19 14:35:48 -07:00
parent 90cc2cdce2
commit 667ad3342a
7 changed files with 200 additions and 94 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -22,21 +22,21 @@ 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"`
Precreator precreator.Config `toml:"shard-precreation"`
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"`
Monitor monitor.Config `toml:"monitor"`
@ -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)
}

80
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,20 @@ 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
s.Services = append(s.Services, srv)
return nil
}
func (s *Server) appendUDPService(c udp.Config) {
if !c.Enabled {
return
@ -403,11 +413,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 +524,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]

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

@ -0,0 +1,20 @@
package registration
import ()
const (
DefaultURL = "https://enterprise.influxdata.com"
)
type Config struct {
Enabled bool `toml:"enabled"`
URL string `toml:"url"`
Token string `toml:"token"`
}
func NewConfig() Config {
return Config{
Enabled: true,
URL: DefaultURL,
}
}

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

@ -0,0 +1,29 @@
package registration_test
import (
"testing"
"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"
`, &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)
}
}

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

@ -0,0 +1,118 @@
package registration
import (
"bytes"
"encoding/json"
"fmt"
"io/ioutil"
"log"
"net/http"
"net/url"
"os"
"sync"
"time"
)
// Service represents the registration service.
type Service struct {
MetaStore interface {
ClusterID() (uint64, error)
NodeID() uint64
}
enabled bool
url *url.URL
token string
version string
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,
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 {
s.logger.Println("Starting registration service")
if err := s.registerServer(); err != nil {
return err
}
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
}
// 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
}
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
}