diff --git a/CHANGELOG.md b/CHANGELOG.md
index 2af9ad1710..a611ef6c0d 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -9,6 +9,7 @@
1. [#1645](https://github.com/influxdata/chronograf/pull/1645): Add Auth0 as a supported OAuth2 provider
1. [#1660](https://github.com/influxdata/chronograf/pull/1660): Add ability to add custom links to User menu via server CLI or ENV vars
1. [#1674](https://github.com/influxdata/chronograf/pull/1674): Add support for organizations in Auth0
+1. [#1695](https://github.com/influxdata/chronograf/pull/1695): Add server flag for adding new InfluxDb sources with Kapacitor servers
### UI Improvements
1. [#1644](https://github.com/influxdata/chronograf/pull/1644): Redesign Alerts History table to have sticky headers
diff --git a/chronograf.go b/chronograf.go
index 0525482a70..d31cb7b85c 100644
--- a/chronograf.go
+++ b/chronograf.go
@@ -625,3 +625,44 @@ type LayoutStore interface {
// Update the dashboard in the store.
Update(context.Context, Layout) error
}
+
+// SourceAndKapacitor is used to parse any NewSources server flag arguments
+type SourceAndKapacitor struct {
+ Source Source `json:"influxdb"`
+ Kapacitor Server `json:"kapacitor"`
+}
+
+// NewSources adds sources to BoltDb idempotently by name, as well as respective kapacitors
+func NewSources(ctx context.Context, sourcesStore SourcesStore, serversStore ServersStore, srcsKaps []SourceAndKapacitor, logger Logger) error {
+ srcs, err := sourcesStore.All(ctx)
+ if err != nil {
+ return err
+ }
+
+SourceLoop:
+ for _, srcKap := range srcsKaps {
+ for _, src := range srcs {
+ // If source already exists, do nothing
+ if src.Name == srcKap.Source.Name {
+ logger.
+ WithField("component", "server").
+ WithField("NewSources", src.Name).
+ Info("Source already exists")
+ continue SourceLoop
+ }
+ }
+
+ src, err := sourcesStore.Add(ctx, srcKap.Source)
+ if err != nil {
+ return err
+ }
+
+ srcKap.Kapacitor.SrcID = src.ID
+ _, err = serversStore.Add(ctx, srcKap.Kapacitor)
+ if err != nil {
+ return err
+ }
+ }
+
+ return nil
+}
diff --git a/chronograf_test.go b/chronograf_test.go
new file mode 100644
index 0000000000..bed0c4f40d
--- /dev/null
+++ b/chronograf_test.go
@@ -0,0 +1,88 @@
+package chronograf_test
+
+import (
+ "context"
+ "reflect"
+ "testing"
+
+ "github.com/influxdata/chronograf"
+ "github.com/influxdata/chronograf/mocks"
+)
+
+func Test_NewSources(t *testing.T) {
+ t.Parallel()
+
+ srcsKaps := []chronograf.SourceAndKapacitor{
+ {
+ Source: chronograf.Source{
+ Default: true,
+ InsecureSkipVerify: false,
+ MetaURL: "http://metaurl.com",
+ Name: "Influx 1",
+ Password: "pass1",
+ Telegraf: "telegraf",
+ URL: "http://localhost:8086",
+ Username: "user1",
+ },
+ Kapacitor: chronograf.Server{
+ Active: true,
+ Name: "Kapa 1",
+ URL: "http://localhost:9092",
+ },
+ },
+ }
+ saboteurSrcsKaps := []chronograf.SourceAndKapacitor{
+ {
+ Source: chronograf.Source{
+ Name: "Influx 1",
+ },
+ Kapacitor: chronograf.Server{
+ Name: "Kapa Aspiring Saboteur",
+ },
+ },
+ }
+
+ ctx := context.Background()
+ srcs := []chronograf.Source{}
+ srcsStore := mocks.SourcesStore{
+ AllF: func(ctx context.Context) ([]chronograf.Source, error) {
+ return srcs, nil
+ },
+ AddF: func(ctx context.Context, src chronograf.Source) (chronograf.Source, error) {
+ srcs = append(srcs, src)
+ return src, nil
+ },
+ }
+ srvs := []chronograf.Server{}
+ srvsStore := mocks.ServersStore{
+ AddF: func(ctx context.Context, srv chronograf.Server) (chronograf.Server, error) {
+ srvs = append(srvs, srv)
+ return srv, nil
+ },
+ }
+
+ err := chronograf.NewSources(ctx, &srcsStore, &srvsStore, srcsKaps, &mocks.TestLogger{})
+ if err != nil {
+ t.Fatal("Expected no error when creating New Sources. Error:", err)
+ }
+ if len(srcs) != 1 {
+ t.Error("Expected one source in sourcesStore")
+ }
+ if len(srvs) != 1 {
+ t.Error("Expected one source in serversStore")
+ }
+
+ err = chronograf.NewSources(ctx, &srcsStore, &srvsStore, saboteurSrcsKaps, &mocks.TestLogger{})
+ if err != nil {
+ t.Fatal("Expected no error when creating New Sources. Error:", err)
+ }
+ if len(srcs) != 1 {
+ t.Error("Expected one source in sourcesStore")
+ }
+ if len(srvs) != 1 {
+ t.Error("Expected one source in serversStore")
+ }
+ if !reflect.DeepEqual(srcs[0], srcsKaps[0].Source) {
+ t.Error("Expected source in sourceStore to remain unchanged")
+ }
+}
diff --git a/server/server_test.go b/mocks/logger.go
similarity index 98%
rename from server/server_test.go
rename to mocks/logger.go
index 22330e851e..46c7bae9f6 100644
--- a/server/server_test.go
+++ b/mocks/logger.go
@@ -1,4 +1,4 @@
-package server_test
+package mocks
import (
"fmt"
diff --git a/mocks/servers.go b/mocks/servers.go
new file mode 100644
index 0000000000..837c0a3b1a
--- /dev/null
+++ b/mocks/servers.go
@@ -0,0 +1,38 @@
+package mocks
+
+import (
+ "context"
+
+ "github.com/influxdata/chronograf"
+)
+
+var _ chronograf.ServersStore = &ServersStore{}
+
+// ServersStore mock allows all functions to be set for testing
+type ServersStore struct {
+ AllF func(context.Context) ([]chronograf.Server, error)
+ AddF func(context.Context, chronograf.Server) (chronograf.Server, error)
+ DeleteF func(context.Context, chronograf.Server) error
+ GetF func(ctx context.Context, ID int) (chronograf.Server, error)
+ UpdateF func(context.Context, chronograf.Server) error
+}
+
+func (s *ServersStore) All(ctx context.Context) ([]chronograf.Server, error) {
+ return s.AllF(ctx)
+}
+
+func (s *ServersStore) Add(ctx context.Context, srv chronograf.Server) (chronograf.Server, error) {
+ return s.AddF(ctx, srv)
+}
+
+func (s *ServersStore) Delete(ctx context.Context, srv chronograf.Server) error {
+ return s.DeleteF(ctx, srv)
+}
+
+func (s *ServersStore) Get(ctx context.Context, id int) (chronograf.Server, error) {
+ return s.GetF(ctx, id)
+}
+
+func (s *ServersStore) Update(ctx context.Context, srv chronograf.Server) error {
+ return s.UpdateF(ctx, srv)
+}
diff --git a/server/server.go b/server/server.go
index 3b3097078d..39e1fcfc07 100644
--- a/server/server.go
+++ b/server/server.go
@@ -3,6 +3,7 @@ package server
import (
"context"
"crypto/tls"
+ "encoding/json"
"log"
"math/rand"
"net"
@@ -50,6 +51,8 @@ type Server struct {
KapacitorUsername string `long:"kapacitor-username" description:"Username of your Kapacitor instance" env:"KAPACITOR_USERNAME"`
KapacitorPassword string `long:"kapacitor-password" description:"Password of your Kapacitor instance" env:"KAPACITOR_PASSWORD"`
+ NewSources string `long:"new-sources" description:"Config for adding a new InfluxDb source and Kapacitor server, in JSON as an array of objects, and surrounded by single quotes. E.g. --new-sources='[{\"influxdb\":{\"name\":\"Influx 1\",\"username\":\"user1\",\"password\":\"pass1\",\"url\":\"http://localhost:8086\",\"metaUrl\":\"http://metaurl.com\",\"insecureSkipVerify\":false,\"default\":true,\"telegraf\":\"telegraf\"},\"kapacitor\":{\"name\":\"Kapa 1\",\"url\":\"http://localhost:9092\",\"active\":true}}]'" env:"NEW_SOURCES"`
+
Develop bool `short:"d" long:"develop" description:"Run server in develop mode."`
BoltPath string `short:"b" long:"bolt-path" description:"Full path to boltDB file (/var/lib/chronograf/chronograf-v1.db)" env:"BOLT_PATH" default:"chronograf-v1.db"`
CannedPath string `short:"c" long:"canned-path" description:"Path to directory of pre-canned application layouts (/usr/share/chronograf/canned)" env:"CANNED_PATH" default:"canned"`
@@ -298,6 +301,9 @@ func (s *Server) Serve(ctx context.Context) error {
return err
}
service := openService(ctx, s.BoltPath, layoutBuilder, sourcesBuilder, kapacitorBuilder, logger, s.useAuth())
+
+ go processNewSources(ctx, service, s.NewSources, logger)
+
basepath = s.Basepath
if basepath != "" && s.PrefixRoutes == false {
logger.
@@ -431,6 +437,33 @@ func openService(ctx context.Context, boltPath string, lBuilder LayoutBuilder, s
}
}
+// processNewSources parses and persists new sources passed in via server flag
+func processNewSources(ctx context.Context, service Service, newSources string, logger chronograf.Logger) error {
+ if newSources == "" {
+ return nil
+ }
+
+ var srcsKaps []chronograf.SourceAndKapacitor
+ // On JSON unmarshal error, continue server process without new source and write error to log
+ if err := json.Unmarshal([]byte(newSources), &srcsKaps); err != nil {
+ logger.
+ WithField("component", "server").
+ WithField("NewSources", "invalid").
+ Error(err)
+ }
+
+ // Add any new sources and kapacitors as specified via server flag
+ if err := chronograf.NewSources(ctx, service.SourcesStore, service.ServersStore, srcsKaps, logger); err != nil {
+ // Continue with server run even if adding NewSources fails
+ logger.
+ WithField("component", "server").
+ WithField("NewSources", "invalid").
+ Error(err)
+ }
+
+ return nil
+}
+
// reportUsageStats starts periodic server reporting.
func reportUsageStats(bi BuildInfo, logger chronograf.Logger) {
rand.Seed(time.Now().UTC().UnixNano())
diff --git a/server/url_prefixer_test.go b/server/url_prefixer_test.go
index 619b304d2c..d1cf66678b 100644
--- a/server/url_prefixer_test.go
+++ b/server/url_prefixer_test.go
@@ -7,6 +7,7 @@ import (
"net/http/httptest"
"testing"
+ "github.com/influxdata/chronograf/mocks"
"github.com/influxdata/chronograf/server"
)
@@ -138,7 +139,7 @@ func Test_Server_Prefixer_NoPrefixingWithoutFlusther(t *testing.T) {
})
}
- tl := &TestLogger{}
+ tl := &mocks.TestLogger{}
pfx := &server.URLPrefixer{
Prefix: "/hill",
Next: backend,