Add Shard Spaces and Retention Policies

Fixes #571. Removed the old long term and short term shards. Now if the
user doesn't manually create a shard space, a catchall called "default"
will be created the first time they write data in.

Added API endpoints to list, create, and delete shard spaces. Made a
breaking change to the old get shards API by changing the format since
shortTerm and longTerm shards aren't guaranteed to be present.

Shard spaces can be mapped to a single database. Retention policies are
enforced per server by a background routine that periodically checks
shards in spaces that have retention policies to see if they have passed
their expiration.

Author: pauldix

config.sample.toml

@@ -94,6 +94,9 @@ point-batch-size = 100
 # reduce the memory usage, but will result in slower writes.
 write-batch-size = 5000000
 
+# The server will check this often for shards that have expired that should be cleared.
+retention-sweep-period = "10m"
+
 [storage.engines.leveldb]
 
 # Maximum mmap open files, this will affect the virtual memory used by
@@ -169,47 +172,6 @@ max-response-buffer-size = 100
 # that you don't need to buffer in memory, but you won't get the best performance.
 concurrent-shard-query-limit = 10
 
-# These options specify how data is sharded across the cluster. There are two
-# shard configurations that have the same knobs: short term and long term.
-# Any series that begins with a capital letter like Exceptions will be written
-# into the long term storage. Any series beginning with a lower case letter
-# like exceptions will be written into short term. The idea being that you
-# can write high precision data into short term and drop it after a couple
-# of days. Meanwhile, continuous queries can run downsampling on the short term
-# data and write into the long term area.
-[sharding]
-  # how many servers in the cluster should have a copy of each shard.
-  # this will give you high availability and scalability on queries
-  replication-factor = 1
-
-  [sharding.short-term]
-  # each shard will have this period of time. Note that it's best to have
-  # group by time() intervals on all queries be < than this setting. If they are
-  # then the aggregate is calculated locally. Otherwise, all that data gets sent
-  # over the network when doing a query.
-  duration = "7d"
-
-  # split will determine how many shards to split each duration into. For example,
-  # if we created a shard for 2014-02-10 and split was set to 2. Then two shards
-  # would be created that have the data for 2014-02-10. By default, data will
-  # be split into those two shards deterministically by hashing the (database, serise)
-  # tuple. That means that data for a given series will be written to a single shard
-  # making querying efficient. That can be overridden with the next option.
-  split = 1
-
-  # You can override the split behavior to have the data for series that match a
-  # given regex be randomly distributed across the shards for a given interval.
-  # You can use this if you have a hot spot for a given time series writing more
-  # data than a single server can handle. Most people won't have to resort to this
-  # option. Also note that using this option means that queries will have to send
-  # all data over the network so they won't be as efficient.
-  # split-random = "/^hf.*/"
-
-  [sharding.long-term]
-  duration = "30d"
-  split = 1
-  # split-random = "/^Hf.*/"
-
 [wal]
 
 dir   = "/tmp/influxdb/development/wal"

src/api/http/api.go

@@ -151,6 +151,9 @@ func (self *HttpServer) Serve(listener net.Listener) {
 	self.registerEndpoint(p, "post", "/cluster/shards", self.createShard)
 	self.registerEndpoint(p, "get", "/cluster/shards", self.getShards)
 	self.registerEndpoint(p, "del", "/cluster/shards/:id", self.dropShard)
+	self.registerEndpoint(p, "get", "/cluster/shard_spaces", self.getShardSpaces)
+	self.registerEndpoint(p, "post", "/cluster/shard_spaces", self.createShardSpace)
+	self.registerEndpoint(p, "del", "/cluster/shard_spaces/:db/:name", self.dropShardSpace)
 
 	// return whether the cluster is in sync or not
 	self.registerEndpoint(p, "get", "/sync", self.isInSync)
@@ -1018,7 +1021,8 @@ type newShardInfo struct {
 	StartTime int64               `json:"startTime"`
 	EndTime   int64               `json:"endTime"`
 	Shards    []newShardServerIds `json:"shards"`
-	LongTerm  bool                `json:"longTerm"`
+	SpaceName string              `json:"spaceName"`
+	Database  string              `json:"database"`
 }
 
 type newShardServerIds struct {
@@ -1038,16 +1042,13 @@ func (self *HttpServer) createShard(w libhttp.ResponseWriter, r *libhttp.Request
 		}
 		shards := make([]*cluster.NewShardData, 0)
 
-		shardType := cluster.SHORT_TERM
-		if newShards.LongTerm {
-			shardType = cluster.LONG_TERM
-		}
 		for _, s := range newShards.Shards {
 			newShardData := &cluster.NewShardData{
 				StartTime: time.Unix(newShards.StartTime, 0),
 				EndTime:   time.Unix(newShards.EndTime, 0),
 				ServerIds: s.ServerIds,
-				Type:      shardType,
+				SpaceName: newShards.SpaceName,
+				Database:  newShards.Database,
 			}
 			shards = append(shards, newShardData)
 		}
@@ -1061,10 +1062,18 @@ func (self *HttpServer) createShard(w libhttp.ResponseWriter, r *libhttp.Request
 
 func (self *HttpServer) getShards(w libhttp.ResponseWriter, r *libhttp.Request) {
 	self.tryAsClusterAdmin(w, r, func(u User) (int, interface{}) {
-		result := make(map[string]interface{})
-		result["shortTerm"] = self.convertShardsToMap(self.clusterConfig.GetShortTermShards())
-		result["longTerm"] = self.convertShardsToMap(self.clusterConfig.GetLongTermShards())
-		return libhttp.StatusOK, result
+		shards := self.clusterConfig.GetShards()
+		shardMaps := make([]map[string]interface{}, 0, len(shards))
+		for _, s := range shards {
+			shardMaps = append(shardMaps, map[string]interface{}{
+				"id":        s.Id(),
+				"endTime":   s.EndTime().Unix(),
+				"startTime": s.StartTime().Unix(),
+				"serverIds": s.ServerIds(),
+				"spaceName": s.SpaceName,
+				"database":  s.Database})
+		}
+		return libhttp.StatusOK, shardMaps
 	})
 }
 
@@ -1123,3 +1132,43 @@ func (self *HttpServer) convertShardsToMap(shards []*cluster.ShardData) []interf
 	}
 	return result
 }
+
+func (self *HttpServer) getShardSpaces(w libhttp.ResponseWriter, r *libhttp.Request) {
+	self.tryAsClusterAdmin(w, r, func(u User) (int, interface{}) {
+		return libhttp.StatusOK, self.clusterConfig.GetShardSpaces()
+	})
+}
+
+func (self *HttpServer) createShardSpace(w libhttp.ResponseWriter, r *libhttp.Request) {
+	self.tryAsClusterAdmin(w, r, func(u User) (int, interface{}) {
+		space := &cluster.ShardSpace{}
+		body, err := ioutil.ReadAll(r.Body)
+		if err != nil {
+			return libhttp.StatusInternalServerError, err.Error()
+		}
+		err = json.Unmarshal(body, space)
+		if err != nil {
+			return libhttp.StatusInternalServerError, err.Error()
+		}
+		err = space.Validate(self.clusterConfig)
+		if err != nil {
+			return libhttp.StatusBadRequest, err.Error()
+		}
+		err = self.raftServer.CreateShardSpace(space)
+		if err != nil {
+			return libhttp.StatusInternalServerError, err.Error()
+		}
+		return libhttp.StatusOK, nil
+	})
+}
+
+func (self *HttpServer) dropShardSpace(w libhttp.ResponseWriter, r *libhttp.Request) {
+	self.tryAsClusterAdmin(w, r, func(u User) (int, interface{}) {
+		name := r.URL.Query().Get(":name")
+		db := r.URL.Query().Get(":db")
+		if err := self.raftServer.DropShardSpace(db, name); err != nil {
+			return libhttp.StatusInternalServerError, err.Error()
+		}
+		return libhttp.StatusOK, nil
+	})
+}