NETOBSERV-786 Query: Table Histogram (#270)

* histogram query

* force step as interval for count_over_time

Author: jpinsonneau

pkg/loki/topology_query.go

@@ -15,6 +15,7 @@ type Topology struct {
 	limit        string
 	rateInterval string
 	step         string
+	function     string
 	dataField    string
 	fields       string
 }
@@ -30,11 +31,15 @@ func NewTopologyQuery(cfg *Config, start, end, limit, rateInterval, step, metric
 		l = topologyDefaultLimit
 	}
 
-	var t string
+	var f, t string
 	switch metricType {
+	case "count":
+		f = "count_over_time"
 	case "packets":
+		f = "rate"
 		t = "Packets"
 	default:
+		f = "rate"
 		t = "Bytes"
 	}
 
@@ -44,6 +49,7 @@ func NewTopologyQuery(cfg *Config, start, end, limit, rateInterval, step, metric
 			rateInterval: rateInterval,
 			step:         step,
 			limit:        l,
+			function:     f,
 			dataField:    t,
 			fields:       getFields(scope, groups),
 		},
@@ -94,19 +100,21 @@ func (q *TopologyQueryBuilder) Build() string {
 	//		topk(
 	// 			<k>,
 	//			sum by(<aggregations>) (
-	//				rate(
+	//				<function>(
 	//					{<label filters>}|<line filters>|json|<json filters>
-	//						|unwrap Bytes|__error__=""[300s]
+	//						|unwrap Bytes|__error__=""[<interval>]
 	//				)
 	//			)
 	//		)
-	//		&<query params>&step=300s
+	//		&<query params>&step=<step>
 	sb := q.createStringBuilderURL()
 	sb.WriteString("topk(")
 	sb.WriteString(q.topology.limit)
 	sb.WriteString(",sum by(")
 	sb.WriteString(q.topology.fields)
-	sb.WriteString(") (rate(")
+	sb.WriteString(") (")
+	sb.WriteString(q.topology.function)
+	sb.WriteString("(")
 	q.appendLabels(sb)
 	q.appendLineFilters(sb)
 	q.appendDeduplicateFilter(sb)
@@ -117,7 +125,11 @@ func (q *TopologyQueryBuilder) Build() string {
 		sb.WriteString(`|__error__=""`)
 	}
 	sb.WriteRune('[')
-	sb.WriteString(q.topology.rateInterval)
+	if q.topology.function == "count_over_time" {
+		sb.WriteString(q.topology.step)
+	} else {
+		sb.WriteString(q.topology.rateInterval)
+	}
 	sb.WriteString("])))")
 	q.appendQueryParams(sb)
 	sb.WriteString("&step=")

pkg/server/server_test.go

@@ -267,6 +267,46 @@ func TestLokiConfigurationForTopology(t *testing.T) {
 	assert.NotNil(t, qr.Result)
 }
 
+func TestLokiConfigurationForTableHistogram(t *testing.T) {
+	// GIVEN a Loki service
+	lokiMock := httpMock{}
+	lokiMock.On("ServeHTTP", mock.Anything, mock.Anything).Run(func(args mock.Arguments) {
+		_, _ = args.Get(0).(http.ResponseWriter).Write([]byte(`{"status":"","data":{"resultType":"matrix","result":[]}}`))
+	})
+	lokiSvc := httptest.NewServer(&lokiMock)
+	defer lokiSvc.Close()
+	lokiURL, err := url.Parse(lokiSvc.URL)
+	require.NoError(t, err)
+
+	// THAT is accessed behind the NOO console plugin backend
+	backendRoutes := setupRoutes(&Config{
+		Loki: loki.Config{
+			URL:     lokiURL,
+			Timeout: time.Second,
+		},
+	})
+	backendSvc := httptest.NewServer(backendRoutes)
+	defer backendSvc.Close()
+
+	// WHEN the Loki flows endpoint is queried in the backend using count type
+	resp, err := backendSvc.Client().Get(backendSvc.URL + "/api/loki/topology?type=count")
+	require.NoError(t, err)
+
+	// THEN the query has been properly forwarded to Loki
+	req := lokiMock.Calls[0].Arguments[1].(*http.Request)
+	assert.Equal(t, `topk(100,sum by(SrcK8S_Name,SrcK8S_Type,SrcK8S_OwnerName,SrcK8S_OwnerType,SrcK8S_Namespace,SrcAddr,SrcK8S_HostName,DstK8S_Name,DstK8S_Type,DstK8S_OwnerName,DstK8S_OwnerType,DstK8S_Namespace,DstAddr,DstK8S_HostName) (count_over_time({app="netobserv-flowcollector"}|~`+"`"+`Duplicate":false`+"`"+`|json[30s])))`, req.URL.Query().Get("query"))
+	// without any multi-tenancy header
+	assert.Empty(t, req.Header.Get("X-Scope-OrgID"))
+
+	// AND the response is sent back to the client
+	body, err := ioutil.ReadAll(resp.Body)
+	require.NoError(t, err)
+	var qr model.AggregatedQueryResponse
+	err = json.Unmarshal(body, &qr)
+	require.NoError(t, err)
+	assert.NotNil(t, qr.Result)
+}
+
 func TestLokiConfiguration_MultiTenant(t *testing.T) {
 	lokiMock := httpMock{}
 	lokiMock.On("ServeHTTP", mock.Anything, mock.Anything).Run(func(args mock.Arguments) {