package core import ( "bytes" "compress/gzip" "encoding/base64" "encoding/gob" "encoding/json" "fmt" "io/ioutil" "net/http" "net/http/httputil" "reflect" "runtime" "strconv" "strings" "time" notify "github.com/bitly/go-notify" log "github.com/cihub/seelog" "github.com/davecgh/go-spew/spew" "github.com/gorilla/websocket" "github.com/umbel/pilosa/db" "github.com/umbel/pilosa/index" "github.com/umbel/pilosa/util" ) const DefaultRequestLogPath = "/tmp/set_bit_log" var RequestLogPath = DefaultRequestLogPath type WebService struct { end chan bool ID util.GUID Version string Port int DefaultDB string SetBitLogEnabled bool Cluster *db.Cluster ProcessMap *ProcessMap Batcher *Batcher Executor interface { RunPQL(database_name string, pql string) (interface{}, error) } Pinger interface { Ping(process_id *util.GUID) (*time.Duration, error) } TopologyMapper interface { MakeFragments(db string, slice_int int) error } Transport interface { Push(message *db.Message) } } func NewWebService() *WebService { return &WebService{ end: make(chan bool), } } type Flusher struct { flusher chan bool } func (self *Flusher) Handle(w http.ResponseWriter, r *http.Request) { self.flusher <- true } func NewFlusher(flusher chan bool) http.HandlerFunc { f := Flusher{flusher} return f.Handle } type RequestLogger struct { handler http.HandlerFunc logger chan []byte } func NewRequestLogger(handler http.HandlerFunc, logger chan []byte) http.HandlerFunc { r := RequestLogger{handler, logger} return r.Handle } func (self *RequestLogger) Handle(w http.ResponseWriter, r *http.Request) { // Grab a dump of the incoming request dump, err := httputil.DumpRequest(r, true /*dump the body also*/) if err != nil { log.Warn("Dump Failure", err) } self.handler(w, r) self.logger <- dump } type LogRecord struct { When time.Time Data_x64 string } func NewLogRecord(t time.Time, data []byte) LogRecord { var b bytes.Buffer w := gzip.NewWriter(&b) w.Write(data) w.Flush() w.Close() e := base64.StdEncoding.EncodeToString(b.Bytes()) x := LogRecord{t, e} return x } func genFileName(id string) string { //bucket/YYYY/MM/DDHHMMSS.id.log t := time.Now() //base := "http://pilosa.umbel.com.s3.amazonaws.com/bit_log" return fmt.Sprintf("%s%s.%s.log", RequestLogPath, t.Format("/2006/01/02/15/04-05"), id) } func flush(requests []LogRecord, id string, records_to_dump int) { dest := genFileName(id) w, err := util.Create(dest) if err != nil { log.Warn("Error opening outfile ", dest) log.Warn(err) return } defer w.Close() encoder := json.NewEncoder(w) for i := 0; i < records_to_dump; i++ { encoder.Encode(requests[i]) } } func Logger(in chan []byte, end chan bool, id string, flusher chan bool) { var buffer = make([]LogRecord, 2048, 2048) i := 0 for { select { case raw := <-in: logRecord := NewLogRecord(time.Now(), raw) buffer[i] = logRecord i += 1 if i > 2047 { flush(buffer, id, i) i = 0 } case <-flusher: if i > 0 { flush(buffer, id, i) i = 0 } case <-end: flush(buffer, id, i) log.Info("Shutdown Logger") return } } } func (self *WebService) Run() { port_string := strconv.Itoa(self.Port) log.Info("Serving HTTP on port:", port_string) logger_chan := make(chan []byte, 1024) flusher := make(chan bool) mux := http.NewServeMux() 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) mux.HandleFunc("/ping", self.HandlePing) mux.HandleFunc("/batch", self.HandleBatch) mux.HandleFunc("/load", self.HandleLoad) if self.SetBitLogEnabled { mux.HandleFunc("/set_bits", NewRequestLogger(self.HandleSetBit, logger_chan)) } else { mux.HandleFunc("/set_bits", self.HandleSetBit) } mux.HandleFunc("/clear_bits", self.HandleClearBit) //mux.HandleFunc("/set_bits", self.HandleSetBit) mux.HandleFunc("/flush", NewFlusher(flusher)) s := &http.Server{ Addr: ":" + port_string, Handler: mux, } go Logger(logger_chan, self.end, self.ID.String(), flusher) s.ListenAndServe() } func (self *WebService) Shutdown() { self.end <- true } func (self *WebService) HandleMessage(w http.ResponseWriter, r *http.Request) { if r.Method != "POST" { http.Error(w, "Only POST allowed", http.StatusMethodNotAllowed) return } var message db.Message decoder := json.NewDecoder(r.Body) if decoder.Decode(&message) != nil { http.Error(w, "Invalid JSON", http.StatusBadRequest) return } } func (self *WebService) HandleLoad(w http.ResponseWriter, r *http.Request) { if r.Method != "POST" { http.Error(w, "Only POST allowed", http.StatusMethodNotAllowed) return } decoder := json.NewDecoder(r.Body) var obj JsonObject err := decoder.Decode(&obj) if err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } database_name, ok := obj["db"] if !ok { http.Error(w, "Provide a database (db)", http.StatusNotFound) return } db := database_name.(string) ms_, ok := obj["max_slice"] if ok { ms := int(ms_.(float64)) database := self.Cluster.GetOrCreateDatabase(db) ns, _ := database.NumSlices() if ns <= ms { for i := ns; i <= ms; i++ { log.Info("Load Create Slice ", i) self.TopologyMapper.MakeFragments(db, i) } http.Error(w, "Needed Slices", http.StatusNotFound) return } } _, ok = obj["id"] if !ok { http.Error(w, "Provide a bitmap id (id)", http.StatusNotFound) return } t := float64(obj["id"].(float64)) bitmap_id := uint64(t) frame, ok := obj["frame"] if !ok { http.Error(w, "Provide a frame (frame)", http.StatusNotFound) return } api_string, ok := obj["bitmap"] if !ok { http.Error(w, "Provide a compressed base64 bitmap (bitmap)", http.StatusNotFound) return } _, ok = obj["filter"] if !ok { http.Error(w, "Provide a filter for categories", http.StatusNotFound) return } t = float64(obj["filter"].(float64)) filter := uint64(t) results := FromApiString(self.Batcher, database_name.(string), frame.(string), api_string.(string), bitmap_id, filter) encoder := json.NewEncoder(w) err = encoder.Encode(results) if err != nil { log.Warn("Error Load results") log.Warn(spew.Sdump(r.Form)) err = encoder.Encode("Bad Batch Request") } } func (self *WebService) HandleBatch(w http.ResponseWriter, r *http.Request) { if r.Method != "POST" { http.Error(w, "Only POST allowed", http.StatusMethodNotAllowed) return } err := r.ParseForm() if err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } database_name := r.Form.Get("db") if database_name == "" { http.Error(w, "Provide a database (db)", http.StatusNotFound) return } bitmap_id_string := r.Form.Get("id") if bitmap_id_string == "" { http.Error(w, "Provide a bitmap id (id)", http.StatusNotFound) return } bitmap_id, _ := strconv.ParseUint(bitmap_id_string, 10, 64) slice_string := r.Form.Get("slice") if bitmap_id_string == "" { http.Error(w, "Provide a slice (slice)", http.StatusNotFound) return } slice, _ := strconv.ParseInt(slice_string, 10, 32) frame := r.Form.Get("frame") if bitmap_id_string == "" { http.Error(w, "Provide a frame (frame)", http.StatusNotFound) return } compressed_bitmap := r.Form.Get("bitmap") if bitmap_id_string == "" { http.Error(w, "Provide a compressed base64 bitmap (bitmap)", http.StatusNotFound) return } filter_string := r.Form.Get("filter") filter, _ := strconv.ParseInt(filter_string, 10, 32) if bitmap_id_string == "" { http.Error(w, "Provide a filter for categories", http.StatusNotFound) return } results := self.Batcher.Batch(database_name, frame, compressed_bitmap, bitmap_id, int(slice), uint64(filter)) encoder := json.NewEncoder(w) err = encoder.Encode(results) if err != nil { log.Warn("Error Batch results") log.Warn(spew.Sdump(r.Form)) err = encoder.Encode("Bad Batch Request") } } func (self *WebService) HandleQuery(w http.ResponseWriter, r *http.Request) { if r.Method != "POST" { http.Error(w, "Only POST allowed", http.StatusMethodNotAllowed) return } util.SendInc("webservice_Query") err := r.ParseForm() if err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } database_name := r.Form.Get("db") if database_name == "" { database_name = self.DefaultDB } if database_name == "" { http.Error(w, "Provide a database (db)", http.StatusNotFound) return } if !self.Cluster.IsValidDatabase(database_name) { http.Error(w, "Unknown Database:"+database_name, http.StatusNotFound) return } pql := r.Form.Get("pql") if pql == "" { http.Error(w, "Provide a valid query string (pql)", http.StatusNotFound) return } _, bits := r.Form["bits"] log.Debug("PQL:", database_name, pql) results, err := self.Executor.RunPQL(database_name, pql) if err != nil { log.Warn("PQL Exec Error:", err.Error(), database_name, pql) http.Error(w, "Error encoding: "+err.Error(), http.StatusInternalServerError) return } switch r := results.(type) { //a hack to handle empty sets case []index.Pair: if len(r) == 0 { results = []int{} } case []byte: //can i figure out the type of the compressed string here? if bits { reader, _ := gzip.NewReader(bytes.NewReader(r)) b, _ := ioutil.ReadAll(reader) result := index.NewBitmap() result.FromBytes(b) results = result.Bits() } } if results == nil { log.Warn("Empty results:", database_name, pql) http.Error(w, "Error encoding: "+err.Error(), http.StatusInternalServerError) return } if err != nil { http.Error(w, "Error running query: "+err.Error(), http.StatusInternalServerError) return } encoder := json.NewEncoder(w) err = encoder.Encode(results) if err != nil { log.Warn("Encode Error :", database_name, pql, err.Error()) return } } type JsonObject map[string]interface{} // post this json to the body //the frame type with extension ".t" are handled a bit differnt // it generats the timestamp based bitmap_ids //[{ "db": "3", "frame":"b.n","profile_id": 122,"filter":0, "bitmap_id":123}, // { "db": "3", "frame":"b.n","profile_id": 122,"filter":2, "bitmap_id":124}, // { "db": "3", "frame":"t.t","profile_id": 122,"filter":2, "bitmap_id":124, "timestamp":"2014-04-03 13:01:04"}]' // func bitmaps(frame string, obj JsonObject) chan uint64 { c := make(chan uint64) go func() { const shortFormT = "2006-01-02T15:04:05" const shortFormS = "2006-01-02 15:04:05" t := float64(obj["bitmap_id"].(float64)) base_id := uint64(t) if strings.HasSuffix(frame, ".t") { timestamp, present := obj["timestamp"].(string) if !present || timestamp == "2014-01-01 00:00:00" { //skip the default timestamp c <- base_id } else { quantum := index.YMDH if val, ok := obj["time_granularity"]; ok { switch val { case "Y": quantum = index.Y case "M": quantum = index.YM case "D": quantum = index.YMD } } shortForm := shortFormS if strings.Contains(timestamp, "T") { shortForm = shortFormT } atime, _ := time.Parse(shortForm, timestamp) for i, id := range index.GetTimeIds(base_id, atime, quantum) { c <- id if i > 10 { log.Warn("TO MANY TIMEIDS", base_id, atime, quantum) break } } } } else { c <- base_id } close(c) }() return c } type SBResult struct { Bitmap_id uint64 Frame string Filter uint64 Profile_id uint64 Result interface{} } func init() { gob.Register(SBResult{}) } func (self *WebService) HandleClearBit(w http.ResponseWriter, r *http.Request) { self.HandleBit(w, r, false) } func (self *WebService) HandleSetBit(w http.ResponseWriter, r *http.Request) { self.HandleBit(w, r, true) } func (self *WebService) HandleBit(w http.ResponseWriter, r *http.Request, ToSet bool) { if r.Method != "POST" { http.Error(w, "Only POST allowed", http.StatusMethodNotAllowed) return } decoder := json.NewDecoder(r.Body) var args []JsonObject err := decoder.Decode(&args) if err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } //[{ "db": "3", "frame":"brand.","profile_id": 122,"filter":0, "bitmap_id":123}] var results []SBResult if len(args) > 4096 { log.Warn("Request too large:", len(args)) http.Error(w, "Request To large", http.StatusBadRequest) return } //remoteSetBit := NewRemoteSetBit() //remoteSetBit.ID = self.ID //remoteSetBit.ProcessMap = self.ProcessMap //remoteSetBit.Hold = self.Hold //remoteSetBit.Transport = self.Transport for _, obj := range args { if obj["profile_id"] == nil { http.Error(w, "Missing Profile", http.StatusBadRequest) return } t := float64(obj["profile_id"].(float64)) profile_id := uint64(t) if obj["db"] == nil { http.Error(w, "Missing db", http.StatusBadRequest) return } dbs := obj["db"].(string) if obj["frame"] == nil { http.Error(w, "Missing Frame", http.StatusBadRequest) return } frame := obj["frame"].(string) if obj["filter"] == nil { http.Error(w, "Missing Filter", http.StatusBadRequest) return } t = float64(obj["filter"].(float64)) filter := uint64(t) for bitmap_id := range bitmaps(frame, obj) { var pql string if ToSet { pql = fmt.Sprintf("set(%d, %s, %d, %d)", bitmap_id, frame, filter, profile_id) } else { pql = fmt.Sprintf("clear(%d, %s, %d, %d)", bitmap_id, frame, filter, profile_id) } result, err := self.Executor.RunPQL(dbs, pql) bundle := SBResult{bitmap_id, frame, filter, profile_id, result} results = append(results, bundle) if err != nil { log.Warn("Error running set_bit", dbs, frame, profile_id, ToSet) http.Error(w, err.Error(), http.StatusInternalServerError) return } } } encoder := json.NewEncoder(w) err = encoder.Encode(results) if err != nil { log.Warn("JSON SetBit ERROR:", err, ToSet) return } } func (self *WebService) HandleStats(w http.ResponseWriter, r *http.Request) { if r.Method != "GET" { http.Error(w, "Only GET allowed", http.StatusMethodNotAllowed) return } encoder := json.NewEncoder(w) m := &runtime.MemStats{} runtime.ReadMemStats(m) err := encoder.Encode(m) if err != nil { log.Warn("Error encoding stats") http.Error(w, "Error econding stats", http.StatusMethodNotAllowed) } } func (self *WebService) HandleInfo(w http.ResponseWriter, r *http.Request) { if r.Method != "GET" { http.Error(w, "Only GET allowed", http.StatusMethodNotAllowed) return } spew.Fdump(w, self.Cluster) } func (self *WebService) HandleVersion(w http.ResponseWriter, r *http.Request) { if r.Method != "GET" { http.Error(w, "Only GET allowed", http.StatusMethodNotAllowed) return } fmt.Fprintf(w, "Pilosa v.("+self.Version+")\n") } func (self *WebService) HandleTest(w http.ResponseWriter, r *http.Request) { if r.Method != "GET" { http.Error(w, "Only GET allowed", http.StatusMethodNotAllowed) return } spew.Dump("TEST!") msg := new(db.Message) msg.Data = "mystring" self.Transport.Push(msg) msg2 := new(db.Message) msg2.Data = 789 self.Transport.Push(msg2) } func (self *WebService) HandleProcesses(w http.ResponseWriter, r *http.Request) { if r.Method != "GET" { http.Error(w, "Only GET allowed", http.StatusMethodNotAllowed) return } encoder := json.NewEncoder(w) processes := self.ProcessMap.GetMetadata() err := encoder.Encode(processes) if err != nil { http.Error(w, "Error Encoding", http.StatusBadRequest) } } func (self *WebService) HandlePing(w http.ResponseWriter, r *http.Request) { if r.Method != "GET" { http.Error(w, "Only GET allowed", http.StatusMethodNotAllowed) return } err := r.ParseForm() if err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } process_string := r.Form.Get("process") process_id, err := util.ParseGUID(process_string) if err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } _, err = self.ProcessMap.GetProcess(&process_id) if err != nil { http.Error(w, err.Error(), http.StatusNotFound) return } duration, err := self.Pinger.Ping(&process_id) if err != nil { spew.Fdump(w, err) return } encoder := json.NewEncoder(w) encoder.Encode(map[string]float64{"duration": duration.Seconds()}) } func (self *WebService) HandleListen(w http.ResponseWriter, r *http.Request) { w.Write([]byte(` Pilosa - streaming client `)) } 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.Info("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() out := spew.Sdump(err) log.Info(out) }() 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.Warn(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(` Pilosa - status `)) }