console: add "top" widgets API

console/root.go

@@ -58,6 +58,7 @@ func (c *Component) Start() error {
 	c.d.HTTP.GinRouter.GET("/api/v0/console/widget/flow-last", c.widgetFlowLastHandlerFunc)
 	c.d.HTTP.GinRouter.GET("/api/v0/console/widget/flow-rate", c.widgetFlowRateHandlerFunc)
 	c.d.HTTP.GinRouter.GET("/api/v0/console/widget/exporters", c.widgetExportersHandlerFunc)
+	c.d.HTTP.GinRouter.GET("/api/v0/console/widget/top/:name", c.widgetTopHandlerFunc)
 
 	c.t.Go(func() error {
 		select {

console/widgets.go

@@ -1,8 +1,10 @@
 package console
 
 import (
+	"fmt"
 	"net/http"
 	"reflect"
+	"strings"
 
 	"github.com/gin-gonic/gin"
 )
@@ -83,3 +85,76 @@ func (c *Component) widgetExportersHandlerFunc(gc *gin.Context) {
 
 	gc.IndentedJSON(http.StatusOK, gin.H{"exporters": exporterList})
 }
+
+type topResult struct {
+	Name    string `json:"name"`
+	Percent uint8  `json:"percent"`
+}
+
+func (c *Component) widgetTopHandlerFunc(gc *gin.Context) {
+	ctx := c.t.Context(gc.Request.Context())
+	var (
+		selector string
+		groupby  string
+		filter   string
+	)
+
+	switch gc.Param("name") {
+	default:
+		gc.JSON(http.StatusNotFound, gin.H{"message": "Unknown top request."})
+		return
+	case "src-as":
+		selector = `concat(toString(SrcAS), ': ', dictGetOrDefault('asns', 'name', SrcAS, '???'))`
+		groupby = `SrcAS`
+	case "dst-as":
+		selector = `concat(toString(DstAS), ': ', dictGetOrDefault('asns', 'name', DstAS, '???'))`
+		groupby = `DstAS`
+	case "src-country":
+		selector = `SrcCountry`
+	case "dst-country":
+		selector = `DstCountry`
+	case "exporter":
+		selector = "ExporterName"
+	case "protocol":
+		selector = `dictGetOrDefault('protocols', 'name', Proto, '???')`
+		groupby = `Proto`
+	case "src-port":
+		selector = `concat(dictGetOrDefault('protocols', 'name', Proto, '???'), '/', toString(SrcPort))`
+		groupby = `Proto, SrcPort`
+	case "dst-port":
+		selector = `concat(dictGetOrDefault('protocols', 'name', Proto, '???'), '/', toString(DstPort))`
+		groupby = `Proto, DstPort`
+	}
+	if strings.HasPrefix(gc.Param("name"), "src-") {
+		filter = "AND InIfBoundary = 'external'"
+	} else if strings.HasPrefix(gc.Param("name"), "dst-") {
+		filter = "AND OutIfBoundary = 'internal'"
+	}
+	if groupby == "" {
+		groupby = selector
+	}
+
+	request := fmt.Sprintf(`
+WITH
+ date_sub(minute, 5, now()) AS StartTime,
+ (SELECT SUM(Bytes*SamplingRate) FROM flows WHERE TimeReceived > StartTime) AS Total
+SELECT
+ %s AS Name,
+ toUInt8(SUM(Bytes*SamplingRate) / Total * 100) AS Percent
+FROM flows
+WHERE TimeReceived > StartTime
+%s
+GROUP BY %s
+ORDER BY Percent DESC
+LIMIT 5
+`, selector, filter, groupby)
+
+	results := []topResult{}
+	err := c.d.ClickHouseDB.Conn.Select(ctx, &results, request)
+	if err != nil {
+		c.r.Err(err).Msg("unable to query database")
+		gc.JSON(http.StatusInternalServerError, gin.H{"message": "Unable to query database."})
+		return
+	}
+	gc.JSON(http.StatusOK, gin.H{"top": results})
+}

console/widgets_test.go

@@ -187,3 +187,77 @@ func TestWidgetExporters(t *testing.T) {
 	})
 
 }
+
+func TestWidgetTop(t *testing.T) {
+	r := reporter.NewMock(t)
+	ch, mockConn := clickhousedb.NewMock(t, r)
+	h := http.NewMock(t, r)
+	c, err := New(r, Configuration{}, Dependencies{
+		Daemon:       daemon.NewMock(t),
+		HTTP:         h,
+		ClickHouseDB: ch,
+	})
+	if err != nil {
+		t.Fatalf("New() error:\n%+v", err)
+	}
+	helpers.StartStop(t, c)
+
+	gomock.InOrder(
+		mockConn.EXPECT().
+			Select(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).
+			SetArg(1, []topResult{
+				{"TCP/443", uint8(51)},
+				{"UDP/443", uint8(20)},
+				{"TCP/80", uint8(18)},
+			}),
+		mockConn.EXPECT().
+			Select(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).
+			SetArg(1, []topResult{
+				{"TCP", uint8(75)},
+				{"UDP", uint8(24)},
+				{"ESP", uint8(1)},
+			}),
+		mockConn.EXPECT().
+			Select(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).
+			SetArg(1, []topResult{
+				{"exporter1", uint8(20)},
+				{"exporter3", uint8(10)},
+				{"exporter5", uint8(3)},
+			}),
+		mockConn.EXPECT().
+			Select(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).
+			SetArg(1, []topResult{
+				{"2906: Netflix", uint8(12)},
+				{"36040: Youtube", uint8(10)},
+				{"20940: Akamai", uint8(9)},
+			}),
+	)
+
+	helpers.TestHTTPEndpoints(t, h.Address, helpers.HTTPEndpointCases{
+		{
+			URL:         "/api/v0/console/widget/top/src-port",
+			ContentType: "application/json; charset=utf-8",
+			FirstLines: []string{
+				`{"top":[{"name":"TCP/443","percent":51},{"name":"UDP/443","percent":20},{"name":"TCP/80","percent":18}]}`,
+			},
+		}, {
+			URL:         "/api/v0/console/widget/top/protocol",
+			ContentType: "application/json; charset=utf-8",
+			FirstLines: []string{
+				`{"top":[{"name":"TCP","percent":75},{"name":"UDP","percent":24},{"name":"ESP","percent":1}]}`,
+			},
+		}, {
+			URL:         "/api/v0/console/widget/top/exporter",
+			ContentType: "application/json; charset=utf-8",
+			FirstLines: []string{
+				`{"top":[{"name":"exporter1","percent":20},{"name":"exporter3","percent":10},{"name":"exporter5","percent":3}]}`,
+			},
+		}, {
+			URL:         "/api/v0/console/widget/top/src-as",
+			ContentType: "application/json; charset=utf-8",
+			FirstLines: []string{
+				`{"top":[{"name":"2906: Netflix","percent":12},{"name":"36040: Youtube","percent":10},{"name":"20940: Akamai","percent":9}]}`,
+			},
+		},
+	})
+}

orchestrator/clickhouse/data/asns.csv

@@ -1010,7 +1010,7 @@ asn,name
 2902,WN-WY-AS
 2903,MORIC
 2904,"Universidad Autonoma De Ciudad Juarez"
-2906,AS-SSI
+2906,Netflix
 2907,"Research Organization of Information and Systems, National Institute of Informatics"
 2909,CITI2
 2912,CITI5