batch error handling

CHANGELOG.md

@@ -64,6 +64,7 @@ With #144 you can now join streams with differing group by dimensions.
 - [#270](https://github.com/influxdata/kapacitor/issues/270): The HTTP server will now gracefully stop.
 - [#300](https://github.com/influxdata/kapacitor/issues/300): Add OPTIONS method to /recording endpoint for deletes.
 - [#304](https://github.com/influxdata/kapacitor/issues/304): Fix panic if recording query but do not have an InfluxDB instance configured
+- [#289](https://github.com/influxdata/kapacitor/issues/289): Add better error handling to batch node.
 
 ## v0.10.1 [2016-02-08]
 

batch.go

@@ -262,7 +262,8 @@ func (b *BatchNode) doQuery() error {
 			// Connect
 			con, err := b.et.tm.InfluxDBService.NewClient()
 			if err != nil {
-				return err
+				b.logger.Println("E! failed to connect to InfluxDB:", err)
+				break
 			}
 			q := client.Query{
 				Command: b.query.String(),
@@ -271,22 +272,28 @@ func (b *BatchNode) doQuery() error {
 			// Execute query
 			resp, err := con.Query(q)
 			if err != nil {
-				return err
+				b.logger.Println("E! query failed:", err)
+				break
 			}
 
 			if err := resp.Error(); err != nil {
-				return err
+				b.logger.Println("E! query failed:", err)
+				break
 			}
 
 			// Collect batches
 			for _, res := range resp.Results {
 				batches, err := models.ResultToBatches(res)
 				if err != nil {
-					return err
+					b.logger.Println("E! failed to understand query result:", err)
+					continue
 				}
 				b.timer.Pause()
 				for _, bch := range batches {
-					b.ins[0].CollectBatch(bch)
+					err := b.ins[0].CollectBatch(bch)
+					if err != nil {
+						return err
+					}
 				}
 				b.timer.Resume()
 			}