Change the batch record/replay actions to get the cluster info from t…

CHANGELOG.md

@@ -180,6 +180,7 @@ batch
 - [#584](https://github.com/influxdata/kapacitor/issues/584): Do not block during startup to send usage stats.
 - [#553](https://github.com/influxdata/kapacitor/issues/553): Periodically check if new InfluxDB DBRPs have been created.
 - [#602](https://github.com/influxdata/kapacitor/issues/602): Fix missing To property on email alert handler.
+- [#581](https://github.com/influxdata/kapacitor/issues/581): Record/Replay batch tasks get cluster info from task not API.
 
 ## v0.13.1 [2016-05-13]
 

batch.go

@@ -94,10 +94,19 @@ func (s *BatchNode) Abort() {
 	}
 }
 
-func (s *BatchNode) Queries(start, stop time.Time) [][]string {
-	queries := make([][]string, len(s.children))
+type BatchQueries struct {
+	Queries []string
+	Cluster string
+}
+
+func (s *BatchNode) Queries(start, stop time.Time) []BatchQueries {
+	queries := make([]BatchQueries, len(s.children))
 	for i, b := range s.children {
-		queries[i] = b.(*QueryNode).Queries(start, stop)
+		qn := b.(*QueryNode)
+		queries[i] = BatchQueries{
+			Queries: qn.Queries(start, stop),
+			Cluster: qn.Cluster(),
+		}
 	}
 	return queries
 }
@@ -206,6 +215,10 @@ func (b *QueryNode) Abort() {
 	close(b.aborting)
 }
 
+func (b *QueryNode) Cluster() string {
+	return b.b.Cluster
+}
+
 func (b *QueryNode) Queries(start, stop time.Time) []string {
 	now := time.Now()
 	if stop.IsZero() {

client/API.md

@@ -791,7 +791,6 @@ A recording ID is returned to later identify the recording.
 | task      | ID of a task, records the results of the queries defined in the task.                                            |
 | start     | Earliest date for which data will be recorded. RFC3339Nano formatted.                                            |
 | stop      | Latest date for which data will be recorded. If not specified uses the current time. RFC3339Nano formatted data. |
-| cluster   | Name of a configured InfluxDB cluster. If empty uses the default cluster.                                        |
 
 ##### Query
 
@@ -1121,7 +1120,6 @@ and so is not supported.
 | task           |         | ID of a task, replays the results of the queries defined in the task against the task.                                                                                                                                                                            |
 | start          |         | Earliest date for which data will be replayed. RFC3339Nano formatted.                                                                                                                                                                            |
 | stop           | now     | Latest date for which data will be replayed. If not specified uses the current time. RFC3339Nano formatted data.                                                                                                                                 |
-| cluster        |         | Name of a configured InfluxDB cluster. If empty uses the default cluster.                                                                                                                                                                        |
 | recording-time | false   | If true, use the times in the recording, otherwise adjust times relative to the current time.                                                                                                                                                    |
 | clock          | fast    | One of `fast` or `real`. If `real` wait for real time to pass corresponding with the time in the recordings. If `fast` replay data without delay. For example, if clock is `real` then a stream recording of duration 5m will take 5m to replay. |
 

client/v1/client.go

@@ -1073,11 +1073,10 @@ func (c *Client) RecordStream(opt RecordStreamOptions) (Recording, error) {
 }
 
 type RecordBatchOptions struct {
-	ID      string    `json:"id,omitempty"`
-	Task    string    `json:"task"`
-	Start   time.Time `json:"start"`
-	Stop    time.Time `json:"stop"`
-	Cluster string    `json:"cluster,omitempty"`
+	ID    string    `json:"id,omitempty"`
+	Task  string    `json:"task"`
+	Start time.Time `json:"start"`
+	Stop  time.Time `json:"stop"`
 }
 
 // Record the batch queries for a task.
@@ -1260,7 +1259,6 @@ type ReplayBatchOptions struct {
 	Task          string    `json:"task"`
 	Start         time.Time `json:"start"`
 	Stop          time.Time `json:"stop"`
-	Cluster       string    `json:"cluster,omitempty"`
 	RecordingTime bool      `json:"recording-time"`
 	Clock         Clock     `json:"clock"`
 }

client/v1/client_test.go

@@ -1215,8 +1215,7 @@ func Test_RecordBatch(t *testing.T) {
 		if r.URL.Path == "/kapacitor/v1/recordings/batch" && r.Method == "POST" &&
 			opts.Task == "taskname" &&
 			opts.Start == start &&
-			opts.Stop == stop &&
-			opts.Cluster == "" {
+			opts.Stop == stop {
 			w.WriteHeader(http.StatusCreated)
 			fmt.Fprintf(w, `{"link": {"rel":"self", "href":"/kapacitor/v1/recordings/rid1"}, "id":"rid1"}`)
 		} else {
@@ -1663,7 +1662,6 @@ func Test_ReplayBatch(t *testing.T) {
 		Task:          "taskname",
 		Start:         start,
 		Stop:          stop,
-		Cluster:       "mycluster",
 		Clock:         client.Real,
 		RecordingTime: true,
 	})

client/v1/swagger.yml

@@ -376,8 +376,6 @@ paths:
               stop:
                 type: string
                 format: dateTime
-              cluster:
-                type: string
             required:
               - task
               - start
@@ -605,8 +603,6 @@ paths:
               stop:
                 type: string
                 format: dateTime
-              cluster:
-                type: string
               recording-time:
                 type: boolean
                 default: false

cmd/kapacitor/main.go

@@ -269,7 +269,6 @@ var (
 	rbStart          = recordBatchFlags.String("start", "", "The start time for the set of queries.")
 	rbStop           = recordBatchFlags.String("stop", "", "The stop time for the set of queries (default now).")
 	rbPast           = recordBatchFlags.String("past", "", "Set start time via 'now - past'.")
-	rbCluster        = recordBatchFlags.String("cluster", "", "Optional named InfluxDB cluster from configuration.")
 	rbNowait         = recordBatchFlags.Bool("no-wait", false, "Do not wait for the recording to finish.")
 	rbId             = recordBatchFlags.String("recording-id", "", "The ID to give to this recording. If not set an random ID is chosen.")
 
@@ -430,11 +429,10 @@ func doRecord(args []string) error {
 		}
 		noWait = *rbNowait
 		recording, err = cli.RecordBatch(client.RecordBatchOptions{
-			ID:      *rbId,
-			Task:    *rbTask,
-			Cluster: *rbCluster,
-			Start:   start,
-			Stop:    stop,
+			ID:    *rbId,
+			Task:  *rbTask,
+			Start: start,
+			Stop:  stop,
 		})
 		if err != nil {
 			return err
@@ -859,7 +857,6 @@ var (
 	rlbStart             = replayLiveBatchFlags.String("start", "", "The start time for the set of queries.")
 	rlbStop              = replayLiveBatchFlags.String("stop", "", "The stop time for the set of queries (default now).")
 	rlbPast              = replayLiveBatchFlags.String("past", "", "Set start time via 'now - past'.")
-	rlbCluster           = replayLiveBatchFlags.String("cluster", "", "Optional named InfluxDB cluster from configuration.")
 
 	replayLiveQueryFlags = flag.NewFlagSet("replay-live-query", flag.ExitOnError)
 	rlqTask              = replayLiveQueryFlags.String("task", "", "The task ID.")
@@ -980,7 +977,6 @@ func doReplayLive(args []string) error {
 			Task:          *rlbTask,
 			Start:         start,
 			Stop:          stop,
-			Cluster:       *rlbCluster,
 			RecordingTime: *rlbRec,
 			Clock:         clk,
 		})

services/influxdb/service.go

@@ -12,6 +12,7 @@ import (
 	"net/url"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/cenkalti/backoff"
@@ -178,6 +179,8 @@ type influxdb struct {
 		Open() error
 		Close() error
 	}
+
+	mu sync.Mutex
 }
 
 type subEntry struct {
@@ -192,6 +195,8 @@ type subInfo struct {
 }
 
 func (s *influxdb) Open() error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
 	if !s.disableSubs {
 		err := s.linkSubscriptions()
 		if s.subscriptionSyncInterval != 0 {
@@ -208,10 +213,14 @@ func (s *influxdb) Open() error {
 }
 
 func (s *influxdb) Close() error {
-	var lastErr error
-	if s.subscriptionSyncInterval != 0 {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if s.subSyncTicker != nil {
 		s.subSyncTicker.Stop()
 	}
+
+	var lastErr error
 	for _, service := range s.services {
 		err := service.Close()
 		if err != nil {

services/replay/service.go

@@ -637,7 +637,7 @@ func (s *Service) handleRecordBatch(w http.ResponseWriter, req *http.Request) {
 
 	go func(recording Recording) {
 		ds, _ := parseDataSourceURL(dataUrl.String())
-		err := s.doRecordBatch(ds, t, opt.Start, opt.Stop, opt.Cluster)
+		err := s.doRecordBatch(ds, t, opt.Start, opt.Stop)
 		s.updateRecordingResult(recording, ds, err)
 	}(recording)
 
@@ -977,7 +977,7 @@ func (s *Service) handleReplayBatch(w http.ResponseWriter, req *http.Request) {
 	}
 
 	go func(replay Replay) {
-		err := s.doLiveBatchReplay(opt.ID, t, clk, opt.RecordingTime, opt.Start, opt.Stop, opt.Cluster)
+		err := s.doLiveBatchReplay(opt.ID, t, clk, opt.RecordingTime, opt.Start, opt.Stop)
 		s.updateReplayResult(replay, err)
 	}(replay)
 
@@ -1080,9 +1080,9 @@ func (r *Service) doReplayFromRecording(id string, task *kapacitor.Task, recordi
 
 }
 
-func (r *Service) doLiveBatchReplay(id string, task *kapacitor.Task, clk clock.Clock, recTime bool, start, stop time.Time, cluster string) error {
+func (r *Service) doLiveBatchReplay(id string, task *kapacitor.Task, clk clock.Clock, recTime bool, start, stop time.Time) error {
 	runReplay := func(tm *kapacitor.TaskMaster) error {
-		sources, recordErrC, err := r.startRecordBatch(task, start, stop, cluster)
+		sources, recordErrC, err := r.startRecordBatch(task, start, stop)
 		if err != nil {
 			return err
 		}
@@ -1248,8 +1248,8 @@ func (b batchArchive) Close() error {
 }
 
 // Record a series of batch queries defined by a batch task
-func (s *Service) doRecordBatch(dataSource DataSource, t *kapacitor.Task, start, stop time.Time, cluster string) error {
-	sources, recordErrC, err := s.startRecordBatch(t, start, stop, cluster)
+func (s *Service) doRecordBatch(dataSource DataSource, t *kapacitor.Task, start, stop time.Time) error {
+	sources, recordErrC, err := s.startRecordBatch(t, start, stop)
 	if err != nil {
 		return err
 	}
@@ -1270,7 +1270,7 @@ func (s *Service) doRecordBatch(dataSource DataSource, t *kapacitor.Task, start,
 	return nil
 }
 
-func (s *Service) startRecordBatch(t *kapacitor.Task, start, stop time.Time, cluster string) ([]<-chan models.Batch, <-chan error, error) {
+func (s *Service) startRecordBatch(t *kapacitor.Task, start, stop time.Time) ([]<-chan models.Batch, <-chan error, error) {
 	// We do not open the task master so it does not need to be closed
 	et, err := kapacitor.NewExecutingTask(s.TaskMaster.New(""), t)
 	if err != nil {
@@ -1286,24 +1286,28 @@ func (s *Service) startRecordBatch(t *kapacitor.Task, start, stop time.Time, clu
 		return nil, nil, errors.New("InfluxDB not configured, cannot record batch query")
 	}
 
-	var con client.Client
-	if cluster != "" {
-		con, err = s.InfluxDBService.NewNamedClient(cluster)
-	} else {
-		con, err = s.InfluxDBService.NewDefaultClient()
-	}
-	if err != nil {
-		return nil, nil, err
-	}
-
 	sources := make([]<-chan models.Batch, len(batches))
 	errors := make(chan error, len(batches))
 
-	for batchIdx, queries := range batches {
+	for batchIndex, batchQueries := range batches {
 		source := make(chan models.Batch)
-		sources[batchIdx] = source
-		go func(queries []string) {
+		sources[batchIndex] = source
+		go func(cluster string, queries []string) {
 			defer close(source)
+
+			// Connect to the cluster
+			var con client.Client
+			var err error
+			if cluster != "" {
+				con, err = s.InfluxDBService.NewNamedClient(cluster)
+			} else {
+				con, err = s.InfluxDBService.NewDefaultClient()
+			}
+			if err != nil {
+				errors <- err
+				return
+			}
+			// Run queries
 			for _, q := range queries {
 				query := client.Query{
 					Command: q,
@@ -1329,7 +1333,7 @@ func (s *Service) startRecordBatch(t *kapacitor.Task, start, stop time.Time, clu
 				}
 			}
 			errors <- nil
-		}(queries)
+		}(batchQueries.Cluster, batchQueries.Queries)
 	}
 	errC := make(chan error, 1)
 	go func() {

task.go

@@ -271,7 +271,7 @@ func (et *ExecutingTask) BatchCount() (int, error) {
 }
 
 // Get the next `num` batch queries that the batcher will run starting at time `start`.
-func (et *ExecutingTask) BatchQueries(start, stop time.Time) ([][]string, error) {
+func (et *ExecutingTask) BatchQueries(start, stop time.Time) ([]BatchQueries, error) {
 	if et.Task.Type != BatchTask {
 		return nil, ErrWrongTaskType
 	}