first pass at auto frament allocation

core/etcd.go

@@ -2,10 +2,12 @@ package core
 
 import (
 	"errors"
+	"fmt"
 	"log"
 	"pilosa/config"
 	"pilosa/db"
 	"pilosa/util"
+	"sort"
 	"strconv"
 	"strings"
 	"sync"
@@ -63,6 +65,69 @@ func NewTopologyMapper(service *Service, namespace string) *TopologyMapper {
 	return &TopologyMapper{service, namespace}
 }
 
+type Pair struct {
+	Key   string
+	Value int
+}
+
+// A slice of Pairs that implements sort.Interface to sort by Value.
+type PairList []Pair
+
+func (p PairList) Swap(i, j int)      { p[i], p[j] = p[j], p[i] }
+func (p PairList) Len() int           { return len(p) }
+func (p PairList) Less(i, j int) bool { return p[i].Value < p[j].Value }
+
+func getLightestProcess(m map[string]int) Pair {
+
+	p := make(PairList, len(m))
+	i := 0
+	for k, v := range m {
+		p[i] = Pair{k, v}
+	}
+	sort.Sort(p)
+	return p[0]
+}
+
+func (self *TopologyMapper) AllocateFragment(db, frame string, slice_int int) error {
+	//get Lock to create the fragment
+	ttl := uint64(300) //secs to hold the lock
+	lock_key := fmt.Sprintf("%s-%s-%d", db, frame, slice_int)
+	_, err := self.service.Etcd.RawCompareAndSwap(lock_key, "0", ttl, "", 0)
+
+	if err == nil {
+		//figure out least loaded process..possibly check max process
+		//to create the node, just write off the items to etcd and the watch should spawn
+		//be nice if something would notify perhaps queue
+		m := make(map[string]int)
+		for _, dbs := range self.service.Cluster.GetDatabases() {
+			for _, fsi := range dbs.GetFramSliceIntersects() {
+				for _, fragment := range fsi.GetFragments() {
+					process := fragment.GetProcess().Id().String()
+					i := m[process]
+					i++
+					m[process] = i
+
+				}
+
+			}
+
+		}
+		p := getLightestProcess(m)
+
+		// PUT -d "value=5cb315c3-6e1d-4218-89b7-943d1dba985b" http://etcd0:4001/v2/keys/pilosa/0/db/29/frame/d/slice/5/fragment/a2b632fc4001b817/proces
+		//so i need db, frame, slice , fragment_id
+		fuid := util.SUUID_to_Hex(util.Id())
+		fragment_key := fmt.Sprintf("%s/db/%d/frame/%s/slice/%d/fragment/%s/process", self.namespace, db, frame, slice_int, fuid)
+		process_guid := p.Key
+		// need to check value to see how many we have left
+		_, err = self.service.Etcd.Set(fragment_key, process_guid, 0)
+		log.Println("fragment created: %s(%s)", fragment_key, process_guid)
+
+	}
+	return err
+
+}
+
 func (self *TopologyMapper) handlenode(node *etcd.Node) error {
 	key := node.Key[len(self.namespace)+1:]
 	bits := strings.Split(key, "/")

db/topology.go

@@ -39,12 +39,6 @@ func (self *Process) Id() uuid.UUID {
 	return *self.id
 }
 
-func (self *Process) SetId(id uuid.UUID) {
-	self.mutex.Lock()
-	defer self.mutex.Unlock()
-	self.id = &id
-}
-
 func (self *Process) Host() string {
 	self.mutex.Lock()
 	defer self.mutex.Unlock()
@@ -117,6 +111,10 @@ func NewCluster() *Cluster {
 	cluster.databases = make(map[string]*Database)
 	return &cluster
 }
+func (self *Cluster) GetDatabases() map[string]*Database {
+	return self.databases
+
+}
 
 /////////// DATABASES ////////////////////////////////////////////////////////////////////
 
@@ -129,6 +127,10 @@ type Database struct {
 	mutex                  sync.Mutex
 }
 
+func (self *Database) GetFramSliceIntersects() []*FrameSliceIntersect {
+	return self.frame_slice_intersects
+}
+
 // Add a database to a cluster
 func (c *Cluster) addDatabase(name string) *Database {
 	database := Database{Name: name}
@@ -181,7 +183,6 @@ func (d *Database) getFrame(name string) (*Frame, error) {
 			return frame, nil
 		}
 	}
-	log.Println("Missing frame:", d.Name, name)
 	return nil, FrameDoesNotExistError
 }
 
@@ -271,8 +272,12 @@ func (d *Database) GetFrameSliceIntersect(frame *Frame, slice *Slice) (*FrameSli
 	return nil, FrameSliceIntersectDoesNotExistError
 }
 
-func (fsi *FrameSliceIntersect) GetFragment(fragment_id util.SUUID) (*Fragment, error) {
-	for _, fragment := range fsi.fragments {
+func (self *FrameSliceIntersect) GetFragments() []*Fragment {
+	return self.fragments
+}
+
+func (self *FrameSliceIntersect) GetFragment(fragment_id util.SUUID) (*Fragment, error) {
+	for _, fragment := range self.fragments {
 		if fragment.id == fragment_id {
 			return fragment, nil
 		}
@@ -280,9 +285,9 @@ func (fsi *FrameSliceIntersect) GetFragment(fragment_id util.SUUID) (*Fragment,
 	return nil, FragmentDoesNotExistError
 }
 
-func (fsi *FrameSliceIntersect) AddFragment(fragment *Fragment) {
-	fsi.fragments = append(fsi.fragments, fragment)
-	fsi.hashring.Add(util.SUUID_to_Hex(fragment.id))
+func (self *FrameSliceIntersect) AddFragment(fragment *Fragment) {
+	self.fragments = append(self.fragments, fragment)
+	self.hashring.Add(util.SUUID_to_Hex(fragment.id))
 }
 
 ///////// FRAGMENTS ////////////////////////////////////////////////////////////////////////
@@ -293,20 +298,20 @@ type Fragment struct {
 	process *Process
 }
 
-func (f *Fragment) GetId() util.SUUID {
-	return f.id
+func (self *Fragment) GetId() util.SUUID {
+	return self.id
 }
 
-func (f *Fragment) GetProcess() *Process {
-	return f.process
+func (self *Fragment) GetProcess() *Process {
+	return self.process
 }
 
-func (f *Fragment) GetProcessId() *uuid.UUID {
-	return f.process.id
+func (self *Fragment) GetProcessId() *uuid.UUID {
+	return self.process.id
 }
 
-func (f *Fragment) GetLocation() *Location {
-	return &Location{f.process.id, f.id}
+func (self *Fragment) GetLocation() *Location {
+	return &Location{self.process.id, self.id}
 }
 
 // rename this one
@@ -340,6 +345,7 @@ func (d *Database) GetFragmentForBitmap(slice *Slice, bitmap *Bitmap) (*Fragment
 func (d *Database) GetFragmentById(fragment_id *uuid.UUID) *Fragment {
 }
 */
+
 func (d *Database) getFragment(frame *Frame, slice *Slice, fragment_id util.SUUID) (*Fragment, error) {
 	fsi, err := d.GetFrameSliceIntersect(frame, slice)
 	if err != nil {