Merge branch 'master' of ops:nuevo-pilosa

commands/pilosa-nexter/nexter.go

@@ -2,15 +2,37 @@ package main
 
 import (
 	"encoding/json"
-	"github.com/coreos/go-etcd/etcd"
+	"flag"
+	"fmt"
 	"log"
 	"net/http"
 	"strconv"
 	"strings"
-	"time"
+
+	"github.com/coreos/go-etcd/etcd"
 )
 
-const blocksize = 64
+type stringslice []string
+
+func (self *stringslice) String() string {
+	return fmt.Sprintf("%d", *self)
+}
+
+func (self *stringslice) Set(value string) error {
+	*self = append(*self, value)
+	return nil
+}
+
+var port uint
+var blocksize uint64
+var etcd_nodes stringslice
+
+func init() {
+	flag.UintVar(&port, "port", 9000, "Port to run HTTP server on")
+	flag.Uint64Var(&blocksize, "blocksize", 64, "Block size")
+	flag.Var(&etcd_nodes, "etcd", "Etcd server")
+	flag.Parse()
+}
 
 type Req interface{}
 
@@ -36,40 +58,32 @@ func (self *Nexter) countloop(ch chan uint64, id int, client *etcd.Client) {
 	for {
 		node, err := client.Get(path, false, false)
 		if err != nil {
-			ee, ok := err.(etcd.EtcdError)
+			ee, ok := err.(*etcd.EtcdError)
 			if ok && ee.ErrorCode == 100 { // node does not exist
-				/*  start = 0
-				    end = blocksize - 1
-				    client.Set(path, "0", 0) // TODO: error check
-				    _ , err := client.CompareAndSwap(path, "0", 0, "0", 0)
-
-				*/
-				//_,_  := c.put(path, "0", 0, options)
-				client.RawCreate(path, "0", 0)
+				_, err := client.Create(path, "0", 0)
+				if err != nil {
+					ee, ok := err.(*etcd.EtcdError)
+					// Catch race condition where another node did the same client.Create()
+					if ok && ee.ErrorCode == 105 { // Node has been created
+						continue
+					} else {
+						log.Fatal(err)
+					}
+				}
 				continue
 			} else {
 				log.Fatal(err)
 			}
 		} else { // No error, get start of series from etcd node
 			start, err = strconv.ParseUint(node.Node.Value, 10, 0)
-			end = start + blocksize
 			if err != nil {
 				log.Fatal(err)
 			}
+			end = start + blocksize
 		}
-		for {
-			newval, err := client.CompareAndSwap(path, strconv.FormatUint(end, 10), 0, strconv.FormatUint(start, 10), 0)
-			if err != nil {
-				break
-			} else {
-				log.Println("Error with CompareAndSet! Trying again in 1 second...")
-				time.Sleep(time.Second)
-				start, err = strconv.ParseUint(newval.Node.Value, 10, 0)
-				if err != nil {
-					log.Fatal(err)
-				}
-				end = start + blocksize
-			}
+		_, err = client.CompareAndSwap(path, strconv.FormatUint(end, 10), 0, strconv.FormatUint(start, 10), 0)
+		if err != nil {
+			continue
 		}
 		for c := start; c < end; c += 1 {
 			ch <- c
@@ -79,12 +93,12 @@ func (self *Nexter) countloop(ch chan uint64, id int, client *etcd.Client) {
 
 func (self *Nexter) loop() {
 	counters := make(map[int]chan uint64)
-	client := etcd.NewClient(nil)
+	client := etcd.NewClient(etcd_nodes)
 	for {
 		select {
 		case req := <-self.reqchan:
 			switch req.(type) {
-			case IncReq:
+			case *IncReq:
 				countreq := req.(*IncReq)
 				counter, ok := counters[countreq.id]
 				if !ok {
@@ -93,12 +107,12 @@ func (self *Nexter) loop() {
 					go self.countloop(counter, countreq.id, client)
 				}
 				go func() { countreq.ret <- <-counter }()
-			case DelReq:
+			case *DelReq:
 				delreq := req.(*DelReq)
 				delete(counters, delreq.id)
 				path := "nexter/" + strconv.Itoa(delreq.id)
 				client.Delete(path, true)
-
+			default:
 			}
 		case <-self.done:
 			break
@@ -145,5 +159,6 @@ func main() {
 		dec := json.NewEncoder(w)
 		dec.Encode(num)
 	})
-	http.ListenAndServe(":9000", nil)
+	port_string := strconv.FormatUint(uint64(port), 10)
+	http.ListenAndServe(":"+port_string, nil)
 }

core/http.go

@@ -7,10 +7,13 @@ import (
 	"net/http"
 	"pilosa/config"
 	"pilosa/db"
+	"reflect"
 	"runtime"
 	"strconv"
 
+	notify "github.com/bitly/go-notify"
 	"github.com/davecgh/go-spew/spew"
+	"github.com/gorilla/websocket"
 	"tux21b.org/v1/gocql/uuid"
 )
 
@@ -29,8 +32,11 @@ func (self *WebService) Run() {
 	mux.HandleFunc("/message", self.HandleMessage)
 	mux.HandleFunc("/query", self.HandleQuery)
 	mux.HandleFunc("/stats", self.HandleStats)
+	mux.HandleFunc("/status", self.HandleStatus)
 	mux.HandleFunc("/info", self.HandleInfo)
 	mux.HandleFunc("/processes", self.HandleProcesses)
+	mux.HandleFunc("/listen/ws", self.HandleListenWS)
+	mux.HandleFunc("/listen/stream", self.HandleListenStream)
 	mux.HandleFunc("/listen", self.HandleListen)
 	mux.HandleFunc("/test", self.HandleTest)
 	mux.HandleFunc("/version", self.HandleVersion)
@@ -249,16 +255,149 @@ func (self *WebService) HandlePing(w http.ResponseWriter, r *http.Request) {
 }
 
 func (self *WebService) HandleListen(w http.ResponseWriter, r *http.Request) {
-	//listener := service.NewListener()
-	//encoder := json.NewEncoder(w)
-	//for {
-	//	select {
-	//	case message := <-listener:
-	//		err := encoder.Encode(message)
-	//		if err != nil {
-	//			log.Println("Error sending message")
-	//			return
-	//		}
-	//	}
-	//}
+	w.Write([]byte(`
+	<html><head><title>Pilosa - streaming client</title></head><body>
+	<style type="text/css">
+		dt.inbox { background-color: #efe; }
+		dt.outbox { background-color: #eef; }
+	</style>
+	<script src="//ajax.googleapis.com/ajax/libs/jquery/1.10.2/jquery.min.js"></script>
+	<script type="text/javascript">
+		$dl = $("<dl/>")
+		$dl.on('click', 'dt', function(e) {
+			$(this).next().toggle()
+		})
+		$("body").append($dl)
+		ws = new WebSocket("ws://" + window.location.host + "/listen/ws")
+		ws.onmessage = function(mes) {
+			obj = JSON.parse(mes.data)
+			if (obj.host) {
+				$dl.append('<dt class="outbox">&rarr;' + obj.type + ' (to: ' + obj.host + ')</dt>')
+			} else {
+				$dl.append('<dt class="inbox">&larr;' + obj.type + '</dt>')
+			}
+			$dl.append('<dd style="display:none;"><pre>' + obj.dump + obj.host + '</pre></dd>')
+		}
+	</script>
+	</body></html>
+	`))
+}
+
+func (self *WebService) streamer(writer func(map[string]interface{}) error) {
+	inbox := make(chan interface{}, 10)
+	outbox := make(chan interface{}, 10)
+	notify.Start("inbox", inbox)
+	notify.Start("outbox", outbox)
+	var obj interface{}
+	var data interface{}
+	var inmessage *db.Message
+	var outmessage *db.Envelope
+	var host string
+	for {
+		select {
+		case obj = <-outbox:
+			outmessage = obj.(*db.Envelope)
+			data = outmessage.Message.Data
+			host = outmessage.Host.String()
+		case obj = <-inbox:
+			inmessage = obj.(*db.Message)
+			data = inmessage.Data
+			host = ""
+		}
+
+		typ := reflect.TypeOf(data)
+
+		err := writer(map[string]interface{}{
+			"type": typ.String(),
+			"dump": spew.Sdump(data),
+			"host": host,
+		})
+		if err != nil {
+			log.Println("stopping")
+			notify.Stop("inbox", inbox)
+			notify.Stop("outbox", outbox)
+			drain(inbox)
+			drain(outbox)
+			return
+		}
+	}
+}
+
+func (self *WebService) HandleListenWS(w http.ResponseWriter, r *http.Request) {
+	defer func() {
+		err := recover()
+		spew.Dump(err)
+	}()
+	ws, err := websocket.Upgrade(w, r, nil, 1024, 1024)
+	if _, ok := err.(websocket.HandshakeError); ok {
+		http.Error(w, "Not a websocket handshake", 400)
+		return
+	} else if err != nil {
+		log.Println(err)
+		return
+	}
+	self.streamer(func(data map[string]interface{}) error {
+		return ws.WriteJSON(data)
+	})
+}
+
+func drain(ch chan interface{}) {
+	for {
+		switch {
+		case <-ch:
+		default:
+			return
+		}
+	}
+}
+
+func (self *WebService) HandleListenStream(w http.ResponseWriter, r *http.Request) {
+	defer func() {
+		err := recover()
+		spew.Dump(err)
+	}()
+	writer := json.NewEncoder(w)
+	flusher := w.(http.Flusher)
+	self.streamer(func(data map[string]interface{}) error {
+		err := writer.Encode(data)
+		flusher.Flush()
+		return err
+	})
+}
+
+func (self *WebService) HandleStatus(w http.ResponseWriter, r *http.Request) {
+	w.Write([]byte(`
+	<html><head><title>Pilosa - status</title></head><body>
+	<style type="text/css">
+	td {
+		background: #eee;
+	}
+	</style>
+	<script src="//ajax.googleapis.com/ajax/libs/jquery/1.10.2/jquery.min.js"></script>
+	<script type="text/javascript">
+		$table = $("<table><tr><th>process id</th><th>host</th><th>tcp port</th><th>http port</th><th>latency</th></tr></table>")
+		$("body").append($table)
+		$.ajax('/processes', {
+			dataType: "json",
+			success: function(resp) {
+				$.each(resp, function(index, value) {
+					$table.append('<tr><td>' + index + '</td><td>' + value.host + '</td><td>' + value.port_tcp + '</td><td>' + value.port_http + '</td><td><button class="pinger"/></td></tr>')
+				})
+			}
+		})
+		$table.on('click', 'button.pinger', function(e) {
+			var $td = $(this).parent()
+			var $tr = $td.parent()
+			var id = $tr.children('td').eq(0).text()
+
+			$.ajax('/ping?process=' +  id, {
+				dataType: 'json',
+				success: function(resp) {
+					$td.html(resp.duration)
+				}
+			})
+		})
+	</script>
+	</body></html>
+	`))
 }

db/db.go

@@ -10,21 +10,11 @@ type Message struct {
 	Data interface{} `json:data`
 }
 
-/*
-import "pilosa/core"
-
-type Message interface {
-	Handle(*core.Service)
+type Envelope struct {
+	Message *Message
+	Host    *uuid.UUID
 }
 
-
-type Message struct {
-	Key         string      `json:key`
-	Data        interface{} `json:data`
-	Destination Location
-}
-*/
-
 type HoldResult interface {
 	ResultId() *uuid.UUID
 	ResultData() interface{}

deps.json

@@ -63,5 +63,15 @@
         "repo": "github.com/davecgh/go-spew/spew",
         "version": "e762b3d1320b76030bd7f6cc2bfc3d9acce874c0",
         "type": "git"
+    },
+    "notify": {
+        "repo": "github.com/bitly/go-notify",
+        "version": "0a148b8111d688ba7550fc7119fe0d5d8e650838",
+        "type": "git"
+    },
+    "websocket": {
+        "repo": "github.com/gorilla/websocket",
+        "version": "92334662baa9cbebc2e6e68b8d56bc1233f85a4c",
+        "type": "git"
     }
 }

transport/tcp.go

@@ -10,14 +10,10 @@ import (
 	"pilosa/db"
 
 	"time"
+	notify "github.com/bitly/go-notify"
 	"tux21b.org/v1/gocql/uuid"
 )
 
-type envelope struct {
-	message *db.Message
-	host    *uuid.UUID
-}
-
 type connection struct {
 	transport *TcpTransport
 	inbox     chan *db.Message
@@ -106,7 +102,7 @@ type TcpTransport struct {
 	service     *core.Service
 	port        int
 	inbox       chan *db.Message
-	outbox      chan envelope
+	outbox      chan db.Envelope
 	connections map[uuid.UUID]*connection
 	reg         chan *newconnection
 }
@@ -117,13 +113,13 @@ func (self *TcpTransport) Run() {
 	for {
 		select {
 		case env := <-self.outbox:
-			con, ok := self.connections[*(env.host)]
+			con, ok := self.connections[*(env.Host)]
 			if !ok {
-				con = &connection{self, make(chan *db.Message, 100), make(chan *db.Message, 100), nil, env.host}
+				con = &connection{self, make(chan *db.Message, 100), make(chan *db.Message, 100), nil, env.Host}
 				go con.manage()
-				self.connections[*env.host] = con
+				self.connections[*env.Host] = con
 			}
-			con.outbox <- env.message
+			con.outbox <- env.Message
 		case nc := <-self.reg:
 			self.connections[*nc.id] = nc.connection
 		}
@@ -157,11 +153,15 @@ func (self *TcpTransport) Close() {
 }
 
 func (self *TcpTransport) Send(message *db.Message, host *uuid.UUID) {
-	self.outbox <- envelope{message, host}
+	envelope := db.Envelope{message, host}
+	notify.Post("outbox", &envelope)
+	self.outbox <- envelope
 }
 
 func (self *TcpTransport) Receive() *db.Message {
-	return <-self.inbox
+	message := <-self.inbox
+	notify.Post("inbox", message)
+	return message
 }
 
 func (self *TcpTransport) Push(message *db.Message) {
@@ -169,5 +169,5 @@ func (self *TcpTransport) Push(message *db.Message) {
 }
 
 func NewTcpTransport(service *core.Service) *TcpTransport {
-	return &TcpTransport{service, config.GetInt("port_tcp"), make(chan *db.Message, 100), make(chan envelope, 100), make(map[uuid.UUID]*connection), make(chan *newconnection)}
+	return &TcpTransport{service, config.GetInt("port_tcp"), make(chan *db.Message, 100), make(chan db.Envelope, 100), make(map[uuid.UUID]*connection), make(chan *newconnection)}
 }