Merge pull request #34 from PreetamJinka/v0.0.1-rev6

v0.0.1-rev6

api/api.go

@@ -1,30 +1,138 @@
 package api
 
 import (
+	"encoding/json"
+	"net"
 	"net/http"
 
-	"github.com/PreetamJinka/cistern/state/metrics"
+	"github.com/VividCortex/siesta"
+
+	"github.com/PreetamJinka/cistern/device"
+	"github.com/PreetamJinka/cistern/state/flows"
 	"github.com/PreetamJinka/cistern/state/series"
 )
 
-type ApiServer struct {
-	addr         string
-	hostRegistry *metrics.HostRegistry
-	engine       *series.Engine
+type APIServer struct {
+	addr           string
+	deviceRegistry *device.Registry
+	seriesEngine   *series.Engine
 }
 
-func NewApiServer(address string, reg *metrics.HostRegistry, engine *series.Engine) *ApiServer {
-	return &ApiServer{
-		addr:         address,
-		hostRegistry: reg,
-		engine:       engine,
+func NewAPIServer(address string, deviceRegistry *device.Registry, seriesEngine *series.Engine) *APIServer {
+	return &APIServer{
+		addr:           address,
+		deviceRegistry: deviceRegistry,
+		seriesEngine:   seriesEngine,
 	}
 }
 
-func (s *ApiServer) Run() {
-	http.Handle("/hosts", hostStatus(s.hostRegistry))
-	http.Handle("/metrics", hostMetrics(s.hostRegistry))
-	http.Handle("/metricstates", metricStates(s.hostRegistry))
-	http.Handle("/metricseries", metricSeries(s.engine))
+func (s *APIServer) Run() {
+	service := siesta.NewService("/")
+
+	service.AddPre(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Access-Control-Allow-Origin", "*")
+	})
+
+	service.AddPost(func(c siesta.Context, w http.ResponseWriter, r *http.Request, q func()) {
+		resp := c.Get(responseKey)
+		err, _ := c.Get(errorKey).(string)
+
+		if resp == nil && err == "" {
+			return
+		}
+
+		enc := json.NewEncoder(w)
+		enc.Encode(APIResponse{
+			Data:  resp,
+			Error: err,
+		})
+	})
+
+	service.Route("GET", "/", "Default page", func(c siesta.Context, w http.ResponseWriter, r *http.Request) {
+		c.Set(responseKey, "Welcome to the Cistern API!")
+	})
+
+	service.Route("GET", "/devices", "Lists sources", func(c siesta.Context, w http.ResponseWriter, r *http.Request) {
+		type ipHostname struct {
+			IP       string `json:"ip"`
+			Hostname string `json:"hostname,omitempty"`
+		}
+
+		devices := []ipHostname{}
+
+		for _, dev := range s.deviceRegistry.Devices() {
+			devices = append(devices, ipHostname{
+				IP:       dev.IP().String(),
+				Hostname: dev.Hostname(),
+			})
+		}
+
+		c.Set(responseKey, devices)
+	})
+
+	service.Route("GET", "/devices/:device/metrics",
+		"Lists metrics for a device",
+		func(c siesta.Context, w http.ResponseWriter, r *http.Request) {
+			var params siesta.Params
+			device := params.String("device", "", "Device name")
+			err := params.Parse(r.Form)
+			if err != nil {
+				c.Set(errorKey, err.Error())
+				return
+			}
+
+			address := net.ParseIP(*device)
+			dev, present := s.deviceRegistry.Lookup(address)
+			if !present {
+				c.Set(errorKey, "device not found")
+				return
+			}
+
+			c.Set(responseKey, dev.Metrics())
+		})
+
+	service.Route("GET", "/devices/:device/flows",
+		"Lists top flows for a device",
+		func(c siesta.Context, w http.ResponseWriter, r *http.Request) {
+			var params siesta.Params
+			device := params.String("device", "", "Device name")
+			err := params.Parse(r.Form)
+			if err != nil {
+				c.Set(errorKey, err.Error())
+				return
+			}
+
+			address := net.ParseIP(*device)
+			dev, present := s.deviceRegistry.Lookup(address)
+			if !present {
+				c.Set(errorKey, "device not found")
+				return
+			}
+
+			type flowsResponse struct {
+				ByBytes   []flows.Flow `json:"byBytes"`
+				ByPackets []flows.Flow `json:"byPackets"`
+			}
+
+			topTalkers := dev.TopTalkers()
+			if topTalkers == nil {
+				c.Set(errorKey, "No active flows")
+				return
+			}
+
+			resp := flowsResponse{
+				ByBytes:   topTalkers.ByBytes(),
+				ByPackets: topTalkers.ByPackets(),
+			}
+
+			c.Set(responseKey, resp)
+		})
+
+	service.Route("GET", "/series/query",
+		"Lists metrics for a device",
+		s.querySeriesRoute())
+
+	http.Handle("/", service)
+
 	go http.ListenAndServe(s.addr, nil)
 }

api/context_keys.go

@@ -0,0 +1,6 @@
+package api
+
+const (
+	responseKey = "response"
+	errorKey    = "error"
+)

api/response.go

@@ -0,0 +1,6 @@
+package api
+
+type APIResponse struct {
+	Data  interface{} `json:"data,omitempty"`
+	Error string      `json:"error,omitempty"`
+}

api/series_query.go

@@ -0,0 +1,82 @@
+package api
+
+import (
+	"encoding/json"
+	"net/http"
+	"time"
+
+	"github.com/PreetamJinka/catena"
+	"github.com/VividCortex/siesta"
+)
+
+func (s *APIServer) querySeriesRoute() func(siesta.Context, http.ResponseWriter, *http.Request) {
+	return func(c siesta.Context, w http.ResponseWriter, r *http.Request) {
+		var params siesta.Params
+		downsample := params.Int64("downsample", 0, "A downsample value of averages N points at a time")
+		err := params.Parse(r.Form)
+		if err != nil {
+			c.Set(errorKey, err.Error())
+			return
+		}
+
+		var descs []catena.QueryDesc
+
+		dec := json.NewDecoder(r.Body)
+		err = dec.Decode(&descs)
+		if err != nil {
+			c.Set(errorKey, err.Error())
+			return
+		}
+
+		now := time.Now().Unix()
+		for i, desc := range descs {
+			if desc.Start <= 0 {
+				desc.Start += now
+			}
+
+			if desc.End <= 0 {
+				desc.End += now
+			}
+
+			descs[i] = desc
+		}
+
+		resp := s.seriesEngine.Query(descs)
+
+		if *downsample <= 1 {
+			c.Set(responseKey, resp)
+			return
+		}
+
+		for i, series := range resp.Series {
+			pointIndex := 0
+			seenPoints := 1
+			currentPartition := series.Points[0].Timestamp / *downsample
+			for j, p := range series.Points {
+				if j == 0 {
+					continue
+				}
+
+				if p.Timestamp / *downsample == currentPartition {
+					series.Points[pointIndex].Value += p.Value
+					seenPoints++
+				} else {
+					currentPartition = p.Timestamp / *downsample
+					series.Points[pointIndex].Value /= float64(seenPoints)
+					pointIndex++
+					seenPoints = 1
+					series.Points[pointIndex] = p
+				}
+
+				if j == len(series.Points) {
+					series.Points[pointIndex].Value /= float64(seenPoints)
+				}
+			}
+
+			series.Points = series.Points[:pointIndex]
+			resp.Series[i] = series
+		}
+
+		c.Set(responseKey, resp)
+	}
+}