Support joining nodes to an existing cluster

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ### Features
 - [#3376](https://github.com/influxdb/influxdb/pull/3376): Support for remote shard query mapping
+- [#3372](https://github.com/influxdb/influxdb/pull/3372): Support joining nodes to existing cluster
 
 ### Bugfixes
 - [#3405](https://github.com/influxdb/influxdb/pull/3405): Prevent database panic when fields are missing. Thanks @jhorwit2

cmd/influxd/restore/restore.go

@@ -229,8 +229,8 @@ restore uses a snapshot of a data node to rebuild a cluster.
 
 // Config represents a partial config for rebuilding the server.
 type Config struct {
-	Meta meta.Config `toml:"meta"`
-	Data tsdb.Config `toml:"data"`
+	Meta *meta.Config `toml:"meta"`
+	Data tsdb.Config  `toml:"data"`
 }
 
 type nopListener struct {

cmd/influxd/run/command.go

@@ -82,6 +82,10 @@ func (cmd *Command) Run(args ...string) error {
 		config.Meta.Hostname = options.Hostname
 	}
 
+	if options.Join != "" {
+		config.Meta.Join = options.Join
+	}
+
 	// Validate the configuration.
 	if err := config.Validate(); err != nil {
 		return fmt.Errorf("%s. To generate a valid configuration file run `influxd config > influxdb.generated.conf`.", err)

cmd/influxd/run/config.go

@@ -24,7 +24,7 @@ import (
 
 // Config represents the configuration format for the influxd binary.
 type Config struct {
-	Meta       meta.Config       `toml:"meta"`
+	Meta       *meta.Config      `toml:"meta"`
 	Data       tsdb.Config       `toml:"data"`
 	Cluster    cluster.Config    `toml:"cluster"`
 	Retention  retention.Config  `toml:"retention"`

cmd/influxd/run/server.go

@@ -9,6 +9,7 @@ import (
 	"os"
 	"runtime"
 	"runtime/pprof"
+	"strings"
 	"time"
 
 	"github.com/influxdb/influxdb/cluster"
@@ -290,10 +291,17 @@ func (s *Server) Open() error {
 		}
 		s.Listener = ln
 
+		// The port 0 is used, we need to retrieve the port assigned by the kernel
+		if strings.HasSuffix(s.BindAddress, ":0") {
+			s.MetaStore.Addr = ln.Addr()
+		}
+
 		// Multiplex listener.
 		mux := tcp.NewMux()
 		s.MetaStore.RaftListener = mux.Listen(meta.MuxRaftHeader)
 		s.MetaStore.ExecListener = mux.Listen(meta.MuxExecHeader)
+		s.MetaStore.RPCListener = mux.Listen(meta.MuxRPCHeader)
+
 		s.ClusterService.Listener = mux.Listen(cluster.MuxHeader)
 		s.SnapshotterService.Listener = mux.Listen(snapshotter.MuxHeader)
 		go mux.Serve(ln)

meta/config.go

@@ -37,10 +37,14 @@ type Config struct {
 	HeartbeatTimeout    toml.Duration `toml:"heartbeat-timeout"`
 	LeaderLeaseTimeout  toml.Duration `toml:"leader-lease-timeout"`
 	CommitTimeout       toml.Duration `toml:"commit-timeout"`
+	ClusterTracing      bool          `toml:"cluster-tracing"`
+
+	// The join command-line argument
+	Join string `toml:"-"`
 }
 
-func NewConfig() Config {
-	return Config{
+func NewConfig() *Config {
+	return &Config{
 		Hostname:            DefaultHostname,
 		BindAddress:         DefaultBindAddress,
 		RetentionAutoCreate: true,

meta/data.go

@@ -706,14 +706,18 @@ func (di *DatabaseInfo) unmarshal(pb *internal.DatabaseInfo) {
 	di.Name = pb.GetName()
 	di.DefaultRetentionPolicy = pb.GetDefaultRetentionPolicy()
 
-	di.RetentionPolicies = make([]RetentionPolicyInfo, len(pb.GetRetentionPolicies()))
-	for i, x := range pb.GetRetentionPolicies() {
-		di.RetentionPolicies[i].unmarshal(x)
+	if len(pb.GetRetentionPolicies()) > 0 {
+		di.RetentionPolicies = make([]RetentionPolicyInfo, len(pb.GetRetentionPolicies()))
+		for i, x := range pb.GetRetentionPolicies() {
+			di.RetentionPolicies[i].unmarshal(x)
+		}
 	}
 
-	di.ContinuousQueries = make([]ContinuousQueryInfo, len(pb.GetContinuousQueries()))
-	for i, x := range pb.GetContinuousQueries() {
-		di.ContinuousQueries[i].unmarshal(x)
+	if len(pb.GetContinuousQueries()) > 0 {
+		di.ContinuousQueries = make([]ContinuousQueryInfo, len(pb.GetContinuousQueries()))
+		for i, x := range pb.GetContinuousQueries() {
+			di.ContinuousQueries[i].unmarshal(x)
+		}
 	}
 }
 
@@ -794,9 +798,11 @@ func (rpi *RetentionPolicyInfo) unmarshal(pb *internal.RetentionPolicyInfo) {
 	rpi.Duration = time.Duration(pb.GetDuration())
 	rpi.ShardGroupDuration = time.Duration(pb.GetShardGroupDuration())
 
-	rpi.ShardGroups = make([]ShardGroupInfo, len(pb.GetShardGroups()))
-	for i, x := range pb.GetShardGroups() {
-		rpi.ShardGroups[i].unmarshal(x)
+	if len(pb.GetShardGroups()) > 0 {
+		rpi.ShardGroups = make([]ShardGroupInfo, len(pb.GetShardGroups()))
+		for i, x := range pb.GetShardGroups() {
+			rpi.ShardGroups[i].unmarshal(x)
+		}
 	}
 }
 
@@ -900,9 +906,11 @@ func (sgi *ShardGroupInfo) unmarshal(pb *internal.ShardGroupInfo) {
 	sgi.EndTime = UnmarshalTime(pb.GetEndTime())
 	sgi.DeletedAt = UnmarshalTime(pb.GetDeletedAt())
 
-	sgi.Shards = make([]ShardInfo, len(pb.GetShards()))
-	for i, x := range pb.GetShards() {
-		sgi.Shards[i].unmarshal(x)
+	if len(pb.GetShards()) > 0 {
+		sgi.Shards = make([]ShardInfo, len(pb.GetShards()))
+		for i, x := range pb.GetShards() {
+			sgi.Shards[i].unmarshal(x)
+		}
 	}
 }