Feature/add s3

api/route.go

@@ -55,6 +55,20 @@ func (server *Server) GetExecutions(resp http.ResponseWriter, req *http.Request)
 	server.JSONWrite(resp, http.StatusOK, results)
 }
 
+func (server *Server) GetAccounts(resp http.ResponseWriter, req *http.Request) {
+	queryLimit, _ := strconv.Atoi(httpparameters.QueryParamWithDefault(req, "querylimit", storage.GetExecutionsQueryLimit))
+	params := mux.Vars(req)
+	executionID := params["executionID"]
+	accounts, err := server.storage.GetAccounts(executionID, queryLimit)
+
+	if err != nil {
+		server.JSONWrite(resp, http.StatusInternalServerError, HttpErrorResponse{Error: err.Error()})
+		return
+
+	}
+	server.JSONWrite(resp, http.StatusOK, accounts)
+}
+
 // GetResourceData return resuts details by resource type
 func (server *Server) GetResourceData(resp http.ResponseWriter, req *http.Request) {
 	queryParams := req.URL.Query()

api/server.go

@@ -81,6 +81,7 @@ func (server *Server) BindEndpoints() {
 
 	server.router.HandleFunc("/api/v1/summary/{executionID}", server.GetSummary).Methods("GET")
 	server.router.HandleFunc("/api/v1/executions", server.GetExecutions).Methods("GET")
+	server.router.HandleFunc("/api/v1/accounts/{executionID}", server.GetAccounts).Methods(("GET"))
 	server.router.HandleFunc("/api/v1/resources/{type}", server.GetResourceData).Methods("GET")
 	server.router.HandleFunc("/api/v1/trends/{type}", server.GetResourceTrends).Methods("GET")
 	server.router.HandleFunc("/api/v1/tags/{executionID}", server.GetExecutionTags).Methods("GET")

api/server_test.go

@@ -235,8 +235,57 @@ func TestGetExecutions(t *testing.T) {
 
 		})
 	}
+}
+
+func TestGetAccounts(t *testing.T) {
+	ms, _ := MockServer()
+	ms.BindEndpoints()
+	ms.Serve()
 
+	testCases := []struct {
+		endpoint           string
+		expectedStatusCode int
+		Count              int
+	}{
+		{"/api/v1/accounts", http.StatusNotFound, 0},
+		{"/api/v1/accounts/1", http.StatusOK, 2},
+		{"/api/v1/accounts/err", http.StatusInternalServerError, 0},
+	}
+
+	for _, test := range testCases {
+		t.Run(test.endpoint, func(t *testing.T) {
+
+			rr := httptest.NewRecorder()
+			req, err := http.NewRequest("GET", test.endpoint, nil)
+			if err != nil {
+				t.Fatal(err)
+			}
+			ms.Router().ServeHTTP(rr, req)
+			if rr.Code != test.expectedStatusCode {
+				t.Fatalf("handler returned wrong status code: got %v want %v", rr.Code, test.expectedStatusCode)
+			}
+
+			if test.expectedStatusCode == http.StatusOK {
+				body, err := ioutil.ReadAll(rr.Body)
+				if err != nil {
+					t.Fatal(err)
+				}
+
+				var accountsData []storage.Accounts
+
+				err = json.Unmarshal(body, &accountsData)
+				if err != nil {
+					t.Fatalf("Could not parse http response")
+				}
+
+				if len(accountsData) != test.Count {
+					t.Fatalf("unexpected accounts data response, got %d expected %d", len(accountsData), test.Count)
+				}
+			}
+		})
+	}
 }
+
 func TestSave(t *testing.T) {
 	ms, mockStorage := MockServer()
 	ms.BindEndpoints()

api/storage/elasticsearch/elasticsearch.go

@@ -10,6 +10,7 @@ import (
 	"fmt"
 	"reflect"
 	"strconv"
+	"strings"
 	"time"
 
 	elastic "github.com/olivere/elastic/v7"
@@ -110,14 +111,24 @@ func (sm *StorageManager) getDynamicMatchQuery(filters map[string]string, operat
 	dynamicMatchQuery := []elastic.Query{}
 	var mq *elastic.MatchQuery
 	for name, value := range filters {
-		mq = elastic.NewMatchQuery(name, value)
-		// Minimum number of clauses that must match for a document to be returned
-		mq.MinimumShouldMatch("100%")
-		if operator == "and" {
-			mq = mq.Operator("and")
+		if name == "Data.AccountID" {
+			var accountIds = strings.Split(value, ",")
+			var accountBoolQuery = elastic.NewBoolQuery()
+			for _, accountId := range accountIds {
+				accountBoolQuery.Should(elastic.NewMatchQuery(name, accountId))
+			}
+			accountBoolQuery.MinimumShouldMatch("1")
+			dynamicMatchQuery = append(dynamicMatchQuery, accountBoolQuery)
+		} else {
+			mq = elastic.NewMatchQuery(name, value)
+			// Minimum number of clauses that must match for a document to be returned
+			mq.MinimumShouldMatch("100%")
+			if operator == "and" {
+				mq = mq.Operator("and")
+			}
+			log.Info("Query ", mq)
+			dynamicMatchQuery = append(dynamicMatchQuery, mq)
 		}
-
-		dynamicMatchQuery = append(dynamicMatchQuery, mq)
 	}
 	return dynamicMatchQuery
 }
@@ -183,60 +194,69 @@ func (sm *StorageManager) GetSummary(executionID string, filters map[string]stri
 	for resourceName, resourceData := range summary {
 		filters["ResourceName"] = resourceName
 		log.WithField("filters", filters).Debug("Going to get resources summary details with the following filters")
-		totalSpent, resourceCount, err := sm.getResourceSummaryDetails(executionID, filters)
+		totalSpent, resourceCount, spentAccounts, err := sm.getResourceSummaryDetails(executionID, filters)
 
 		if err != nil {
 			continue
 		}
 		newResourceData := resourceData
 		newResourceData.TotalSpent = totalSpent
 		newResourceData.ResourceCount = resourceCount
+		newResourceData.SpentAccounts = spentAccounts
 		summary[resourceName] = newResourceData
-
 	}
 
 	return summary, nil
 
 }
 
 // getResourceSummaryDetails returns total resource spent and total resources detected
-func (sm *StorageManager) getResourceSummaryDetails(executionID string, filters map[string]string) (float64, int64, error) {
+func (sm *StorageManager) getResourceSummaryDetails(executionID string, filters map[string]string) (float64, int64, map[string]float64, error) {
 
 	var totalSpent float64
 	var resourceCount int64
+	var spentAccounts = make(map[string]float64)
 
 	dynamicMatchQuery := sm.getDynamicMatchQuery(filters, "or")
 	dynamicMatchQuery = append(dynamicMatchQuery, elastic.NewTermQuery("ExecutionID", executionID))
 	dynamicMatchQuery = append(dynamicMatchQuery, elastic.NewTermQuery("EventType", "resource_detected"))
 
-	searchResult, err := sm.client.Search().
+	searchResultAccount, err := sm.client.Search().
 		Query(elastic.NewBoolQuery().Must(dynamicMatchQuery...)).
-		Aggregation("sum", elastic.NewSumAggregation().Field("Data.PricePerMonth")).
+		Aggregation("accounts", elastic.NewTermsAggregation().Field("Data.AccountID.keyword").
+			SubAggregation("accountSum", elastic.NewSumAggregation().Field("Data.PricePerMonth"))).
 		Size(0).Do(context.Background())
 
 	if err != nil {
-		log.WithError(err).WithFields(log.Fields{
-			"filters": filters,
-		}).Error("error when trying to get summary details")
+		log.WithError(err).Error("error when trying to get executions collectors")
+		return totalSpent, resourceCount, spentAccounts, ErrInvalidQuery
+	}
 
-		return totalSpent, resourceCount, err
+	respAccount, ok := searchResultAccount.Aggregations.Terms("accounts")
+	if !ok {
+		log.Error("accounts field term does not exist")
+		return totalSpent, resourceCount, spentAccounts, ErrAggregationTermNotFound
 	}
 
-	log.WithFields(log.Fields{
-		"filters":      filters,
-		"milliseconds": searchResult.TookInMillis,
-	}).Debug("get execution details")
+	for _, AccountIdBucket := range respAccount.Buckets {
 
-	resp, ok := searchResult.Aggregations.Terms("sum")
-	if ok {
-		if val, ok := resp.Aggregations["value"]; ok {
+		spent, ok := AccountIdBucket.Aggregations.Terms("accountSum")
+		if ok {
+			if val, ok := spent.Aggregations["value"]; ok {
+				accountID, ok := AccountIdBucket.Key.(string)
+				if !ok {
+					log.Error("type assertion to string failed")
+					continue
+				}
+				spentAccounts[accountID], _ = strconv.ParseFloat(string(val), 64)
 
-			totalSpent, _ = strconv.ParseFloat(string(val), 64)
-			resourceCount = searchResult.Hits.TotalHits.Value
+				totalSpent += spentAccounts[accountID]
+				resourceCount += AccountIdBucket.DocCount
+			}
 		}
 	}
 
-	return totalSpent, resourceCount, nil
+	return totalSpent, resourceCount, spentAccounts, nil
 }
 
 // GetExecutions returns collector executions
@@ -298,6 +318,44 @@ func (sm *StorageManager) GetExecutions(queryLimit int) ([]storage.Executions, e
 	return executions, nil
 }
 
+func (sm *StorageManager) GetAccounts(executionID string, querylimit int) ([]storage.Accounts, error) {
+	accounts := []storage.Accounts{}
+
+	searchResult, err := sm.client.Search().Query(elastic.NewMatchQuery("ExecutionID", executionID)).
+		Aggregation("Accounts", elastic.NewTermsAggregation().
+			Field("Data.AccountInformation.keyword").Size(querylimit)).
+		Do(context.Background())
+
+	if err != nil {
+		log.WithError(err).Error("error when trying to get AccountIDs")
+		return accounts, ErrInvalidQuery
+	}
+
+	resp, ok := searchResult.Aggregations.Terms("Accounts")
+	if !ok {
+		log.Error("accounts field term does not exist")
+		return accounts, ErrAggregationTermNotFound
+	}
+
+	for _, accountsBucket := range resp.Buckets {
+		account, ok := accountsBucket.Key.(string)
+		if !ok {
+			log.Error("type assertion to string failed")
+			continue
+		}
+		name, id, err := interpolation.ExtractAccountInformation(account)
+		if err != nil {
+			log.WithError(err).WithField("account", account).Error("could not extract account information")
+			continue
+		}
+		accounts = append(accounts, storage.Accounts{
+			ID:   id,
+			Name: name,
+		})
+	}
+	return accounts, nil
+}
+
 // GetResources return resource data
 func (sm *StorageManager) GetResources(resourceType string, executionID string, filters map[string]string) ([]map[string]interface{}, error) {
 

api/storage/elasticsearch/elasticsearch_test.go

@@ -217,6 +217,64 @@ func TestGetExecutions(t *testing.T) {
 	}
 }
 
+func TestGetAccounts(t *testing.T) {
+
+	// each different queryLimit will result in a different elasticsearch response.
+	// 1 - returns valid account data response
+	// 2 - returns invalid aggregation term query response
+	// 4 - returns invalid statuscode response
+	testCases := []struct {
+		name          string
+		queryLimit    int
+		responseCount int
+		ErrorMessage  error
+	}{
+		{"valid response", 1, 2, nil},
+		{"invalid terms", 2, 0, ErrAggregationTermNotFound},
+		{"invalid es response", 3, 0, ErrInvalidQuery},
+	}
+
+	mockClient, config := testutils.NewESMock(prefixIndexName, true)
+
+	mockClient.Router.HandleFunc("/_search", func(resp http.ResponseWriter, req *http.Request) {
+		switch testutils.GetPostParams(req) {
+		case `{"aggregations":{"Accounts":{"terms":{"field":"Data.AccountInformation.keyword","size":1}}},"query":{"match":{"ExecutionID":{"query":"1"}}}}`:
+			testutils.JSONResponse(resp, http.StatusOK, elastic.SearchResult{Aggregations: map[string]json.RawMessage{
+				"Accounts": testutils.LoadResponse("accounts/aggregations/default"),
+			}})
+		case `{"aggregations":{"Accounts":{"terms":{"field":"Data.AccountInformation.keyword","size":2}}},"query":{"match":{"ExecutionID":{"query":"1"}}}}`:
+			testutils.JSONResponse(resp, http.StatusOK, elastic.SearchResult{Aggregations: map[string]json.RawMessage{
+				"invalid-key": testutils.LoadResponse("accounts/aggregations/default"),
+			}})
+		case `{"aggregations":{"Accounts":{"terms":{"field":"Data.AccountInformation.keyword","size":3}}},"query":{"match":{"ExecutionID":{"query":"1"}}}}`:
+			testutils.JSONResponse(resp, http.StatusBadRequest, elastic.SearchResult{Aggregations: map[string]json.RawMessage{}})
+		default:
+			t.Fatalf("unexpected request params")
+		}
+	})
+
+	es, err := NewStorageManager(config)
+	if err != nil {
+		t.Fatalf("unexpected error, got %v expected nil", err)
+	}
+
+	for _, test := range testCases {
+		t.Run(test.name, func(t *testing.T) {
+
+			response, err := es.GetAccounts("1", test.queryLimit)
+
+			if err != test.ErrorMessage {
+				t.Fatalf("unexpected error, got %v expected %v", err, test.ErrorMessage)
+			}
+
+			if len(response) != test.responseCount {
+				t.Fatalf("handler query response: got %d want %d", len(response), test.responseCount)
+			}
+
+		})
+	}
+}
+
 func TestGetSummary(t *testing.T) {
 
 	mockClient, config := testutils.NewESMock(prefixIndexName, true)
@@ -225,7 +283,9 @@ func TestGetSummary(t *testing.T) {
 
 		response := elastic.SearchResult{}
 
-		switch testutils.GetPostParams(req) {
+		var s = testutils.GetPostParams(req)
+		fmt.Println(s)
+		switch s {
 		case `{"query":{"bool":{"must":[{"term":{"EventType":"service_status"}},{"term":{"ExecutionID":""}}]}},"size":0}`:
 			response.Hits = &elastic.SearchHits{TotalHits: &elastic.TotalHits{Value: 1}}
 		case `{"query":{"bool":{"must":[{"term":{"EventType":"service_status"}},{"term":{"ExecutionID":""}}]}},"size":1}`:
@@ -238,6 +298,9 @@ func TestGetSummary(t *testing.T) {
 		case `{"aggregations":{"sum":{"sum":{"field":"Data.PricePerMonth"}}},"query":{"bool":{"must":[{"match":{"ResourceName":{"minimum_should_match":"100%","query":"aws_resource_name"}}},{"term":{"ExecutionID":""}},{"term":{"EventType":"resource_detected"}}]}},"size":0}`:
 			response.Aggregations = map[string]json.RawMessage{"sum": []byte(`{"value": 36.5}`)}
 			response.Hits = &elastic.SearchHits{TotalHits: &elastic.TotalHits{Value: 1}}
+		case `{"aggregations":{"accounts":{"aggregations":{"accountSum":{"sum":{"field":"Data.PricePerMonth"}}},"terms":{"field":"Data.AccountID.keyword"}}},"query":{"bool":{"must":[{"match":{"ResourceName":{"minimum_should_match":"100%","query":"aws_resource_name"}}},{"term":{"ExecutionID":""}},{"term":{"EventType":"resource_detected"}}]}},"size":0}`:
+			response.Aggregations = map[string]json.RawMessage{"accounts": testutils.LoadResponse("summary/aggregations/default")}
+			response.Hits = &elastic.SearchHits{TotalHits: &elastic.TotalHits{Value: 1}}
 		default:
 			t.Fatalf("unexpected request params")
 		}

api/storage/elasticsearch/testutils/responses/accounts/aggregations/default.json

@@ -0,0 +1,12 @@
+{
+  "buckets" : [
+    {
+      "key" : "Test_123456789",
+      "doc_count" : 66
+    },
+    {
+      "key" : "Test2_12345675",
+      "doc_count" : 6
+    }
+  ]
+}

api/storage/elasticsearch/testutils/responses/summary/aggregations/default.json

@@ -0,0 +1,11 @@
+{
+  "buckets": [
+    {
+      "key": "123456789012",
+      "doc_count": 1,
+      "accountSum": {
+        "value": 36.5
+      }
+    }
+  ]
+}