This is the beginning of the clustering stuff. Client/server communication, etc.

config.json.sample

@@ -5,5 +5,6 @@
   "RaftServerPort": 8090,
   "SeedServers":    [],
   "DataDir":        "/tmp/influxdb/development/db",
-  "RaftDir":        "/tmp/influxdb/development/raft"
+  "RaftDir":        "/tmp/influxdb/development/raft",
+  "ProtobufPort":   8099
 }

package.sh

@@ -27,7 +27,7 @@ function package_admin_interface {
 }
 
 function packae_source {
-    rm -f server
+    rm -f influxd
     git ls-files --others  | egrep -v 'github|launchpad|code.google' > /tmp/influxdb.ignored
     echo "pkg/*" >> /tmp/influxdb.ignored
     echo "packages/*" >> /tmp/influxdb.ignored
@@ -55,7 +55,7 @@ function package_files {
 
     package_admin_interface
 
-    mv server build/influxdb
+    mv influxd build/influxdb
 
     # cp -R src/admin/site/ build/admin/
     mkdir build/admin
@@ -110,14 +110,14 @@ function build_packages {
 function setup_version {
     echo "Changing version from dev to $influxdb_version"
     sha1=`git rev-list --max-count=1 HEAD`
-    sed -i.bak -e "s/version = \"dev\"/version = \"$influxdb_version\"/" -e "s/gitSha\s*=\s*\"HEAD\"/gitSha = \"$sha1\"/" src/server/server.go
+    sed -i.bak -e "s/version = \"dev\"/version = \"$influxdb_version\"/" -e "s/gitSha\s*=\s*\"HEAD\"/gitSha = \"$sha1\"/" src/daemon/influxd.go
     sed -i.bak -e "s/REPLACE_VERSION/$influxdb_version/" scripts/post_install.sh
 }
 
 function revert_version {
-    if [ -e src/server/server.go.bak ]; then
-        rm src/server/server.go
-        mv src/server/server.go.bak src/server/server.go
+    if [ -e src/daemon/influxd.go.bak ]; then
+        rm src/daemon/influxd.go
+        mv src/daemon/influxd.go.bak src/daemon/influxd.go
     fi
 
     if [ -e scripts/post_install.sh ]; then

src/api/http/api.go

@@ -344,23 +344,16 @@ func (self *HttpServer) writePoints(w libhttp.ResponseWriter, r *libhttp.Request
 }
 
 type createDatabaseRequest struct {
-	Name string `json:"name"`
-}
-
-type Database struct {
-	Name string `json:"name"`
+	Name              string `json:"name"`
+	ReplicationFactor uint8  `json:"replicationFactor"`
 }
 
 func (self *HttpServer) listDatabases(w libhttp.ResponseWriter, r *libhttp.Request) {
 	self.tryAsClusterAdmin(w, r, func(u common.User) (int, interface{}) {
-		dbNames, err := self.coordinator.ListDatabases(u)
+		databases, err := self.coordinator.ListDatabases(u)
 		if err != nil {
 			return errorToStatusCode(err), err.Error()
 		}
-		databases := make([]*Database, 0, len(dbNames))
-		for _, db := range dbNames {
-			databases = append(databases, &Database{db})
-		}
 		return libhttp.StatusOK, databases
 	})
 }
@@ -376,7 +369,7 @@ func (self *HttpServer) createDatabase(w libhttp.ResponseWriter, r *libhttp.Requ
 		if err != nil {
 			return libhttp.StatusBadRequest, err.Error()
 		}
-		err = self.coordinator.CreateDatabase(user, createRequest.Name)
+		err = self.coordinator.CreateDatabase(user, createRequest.Name, createRequest.ReplicationFactor)
 		if err != nil {
 			return errorToStatusCode(err), err.Error()
 		}

src/api/http/api_test.go

@@ -3,6 +3,7 @@ package http
 import (
 	"bytes"
 	"common"
+	"coordinator"
 	"encoding/base64"
 	"encoding/json"
 	"fmt"
@@ -109,13 +110,13 @@ func (self *MockCoordinator) WriteSeriesData(_ common.User, db string, series *p
 	return nil
 }
 
-func (self *MockCoordinator) CreateDatabase(_ common.User, db string) error {
+func (self *MockCoordinator) CreateDatabase(_ common.User, db string, _ uint8) error {
 	self.db = db
 	return nil
 }
 
-func (self *MockCoordinator) ListDatabases(_ common.User) ([]string, error) {
-	return []string{"db1", "db2"}, nil
+func (self *MockCoordinator) ListDatabases(_ common.User) ([]*coordinator.Database, error) {
+	return []*coordinator.Database{&coordinator.Database{"db1", 1}, &coordinator.Database{"db2", 1}}, nil
 }
 
 func (self *MockCoordinator) DropDatabase(_ common.User, db string) error {
@@ -602,10 +603,10 @@ func (self *ApiSuite) TestDatabasesIndex(c *C) {
 		defer resp.Body.Close()
 		body, err := ioutil.ReadAll(resp.Body)
 		c.Assert(err, IsNil)
-		users := []*Database{}
+		users := []*coordinator.Database{}
 		err = json.Unmarshal(body, &users)
 		c.Assert(err, IsNil)
-		c.Assert(users, DeepEquals, []*Database{&Database{"db1"}, &Database{"db2"}})
+		c.Assert(users, DeepEquals, []*coordinator.Database{&coordinator.Database{"db1", uint8(1)}, &coordinator.Database{"db2", uint8(1)}})
 	}
 }
 
@@ -620,8 +621,8 @@ func (self *ApiSuite) TestBasicAuthentication(c *C) {
 	body, err := ioutil.ReadAll(resp.Body)
 	c.Assert(err, IsNil)
 	c.Assert(resp.StatusCode, Equals, libhttp.StatusOK)
-	users := []*Database{}
+	users := []*coordinator.Database{}
 	err = json.Unmarshal(body, &users)
 	c.Assert(err, IsNil)
-	c.Assert(users, DeepEquals, []*Database{&Database{"db1"}, &Database{"db2"}})
+	c.Assert(users, DeepEquals, []*coordinator.Database{&coordinator.Database{"db1", 1}, &coordinator.Database{"db2", 1}})
 }

src/common/helpers.go

@@ -1,7 +1,11 @@
 package common
 
 import (
+	"bytes"
+	"crypto/sha1"
+	"encoding/binary"
 	"encoding/json"
+	"fmt"
 	"protocol"
 	"time"
 )
@@ -15,3 +19,16 @@ func StringToSeriesArray(seriesString string) ([]*protocol.Series, error) {
 func CurrentTime() int64 {
 	return time.Now().UnixNano() / int64(1000)
 }
+
+func RingLocation(database *string, timeSeries *string, time *int64) int {
+	hasher := sha1.New()
+	hasher.Write([]byte(fmt.Sprintf("%s%s%d", *database, *timeSeries, *time)))
+	buf := bytes.NewBuffer(hasher.Sum(nil))
+	var n int64
+	binary.Read(buf, binary.LittleEndian, &n)
+	nInt := int(n)
+	if nInt < 0 {
+		nInt = nInt * -1
+	}
+	return nInt
+}

src/configuration/configuration.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"io/ioutil"
 	"log"
+	"os"
 )
 
 type Configuration struct {
@@ -15,6 +16,8 @@ type Configuration struct {
 	SeedServers    []string
 	DataDir        string
 	RaftDir        string
+	ProtobufPort   int
+	Hostname       string
 }
 
 func LoadConfiguration(fileName string) *Configuration {
@@ -43,3 +46,24 @@ func (self *Configuration) AdminHttpPortString() string {
 func (self *Configuration) ApiHttpPortString() string {
 	return fmt.Sprintf(":%d", self.ApiHttpPort)
 }
+
+func (self *Configuration) ProtobufPortString() string {
+	return fmt.Sprintf(":%d", self.ProtobufPort)
+}
+
+func (self *Configuration) HostnameOrDetect() string {
+	if self.Hostname != "" {
+		return self.Hostname
+	} else {
+		n, err := os.Hostname()
+		if err == nil {
+			return n
+		} else {
+			return "localhost"
+		}
+	}
+}
+
+func (self *Configuration) ProtobufConnectionString() string {
+	return fmt.Sprintf("%s:%d", self.HostnameOrDetect(), self.ProtobufPort)
+}

src/coordinator/client_server_test.go

@@ -0,0 +1,79 @@
+package coordinator
+
+import (
+	"datastore"
+	"fmt"
+	. "launchpad.net/gocheck"
+	"os"
+	"protocol"
+	"time"
+)
+
+type ClientServerSuite struct{}
+
+var _ = Suite(&ClientServerSuite{})
+
+const DB_DIR = "/tmp/influxdb/datastore_test"
+
+func newDatastore(c *C) datastore.Datastore {
+	os.MkdirAll(DB_DIR, 0744)
+	db, err := datastore.NewLevelDbDatastore(DB_DIR)
+	c.Assert(err, Equals, nil)
+	return db
+}
+
+func cleanDb(db datastore.Datastore) {
+	if db != nil {
+		db.Close()
+	}
+	os.RemoveAll(DB_DIR)
+}
+
+func (self *ClientServerSuite) TestClientCanMakeRequests(c *C) {
+	server := startAndVerifyCluster(1, c)[0]
+	defer clean(server)
+	db := newDatastore(c)
+	requestHandler := NewProtobufRequestHandler(db, server)
+	protobufServer := NewProtobufServer(":8091", requestHandler)
+	go protobufServer.ListenAndServe()
+	c.Assert(protobufServer, Not(IsNil))
+	protobufClient := NewProtobufClient("localhost:8091")
+	responseStream := make(chan *protocol.Response, 1)
+
+	mock := `
+  {
+    "points": [
+      { "values": [{"int64_value": 3}]},
+      { "values": [{"int64_value": 23}]}
+    ],
+    "name": "foo",
+    "fields": ["val"]
+  }`
+	fmt.Println("creating series")
+	series := stringToSeries(mock, c)
+	id := uint32(1)
+	database := "pauldb"
+	proxyWrite := protocol.Request_PROXY_WRITE
+	request := &protocol.Request{Id: &id, Type: &proxyWrite, Database: &database, Series: series}
+
+	time.Sleep(time.Second * 1)
+	err := protobufClient.MakeRequest(request, responseStream)
+	c.Assert(err, IsNil)
+	timer := time.NewTimer(time.Second)
+	select {
+	case <-timer.C:
+		c.Error("Timed out waiting for response")
+	case response := <-responseStream:
+		c.Assert(*response.Type, Equals, protocol.Response_WRITE_OK)
+	}
+}
+
+func (self *ClientServerSuite) TestClientReconnectsIfDisconnected(c *C) {
+}
+
+func (self *ClientServerSuite) TestServerExecutesReplayRequestIfWriteIsOutOfSequence(c *C) {
+}
+
+func (self *ClientServerSuite) TestServerKillsOldHandlerWhenClientReconnects(c *C) {
+
+}