add gcp billing additional fields

x-pack/metricbeat/module/gcp/billing/billing.go

@@ -216,10 +216,15 @@ func getTables(ctx context.Context, client *bigquery.Client, datasetID string, t
 	return tables, nil
 }
 
+const (
+	StandardTableRowLen = 11
+	DetailedTableRowLen = 12
+)
+
 func (m *MetricSet) queryBigQuery(ctx context.Context, client *bigquery.Client, tableMeta tableMeta, month string, costType string) ([]mb.Event, error) {
 	var events []mb.Event
 
-	query := generateQuery(tableMeta.tableFullID, month, costType)
+	query := m.generateQuery(tableMeta.tableFullID, month, costType)
 	m.logger.Debug("bigquery query = ", query)
 
 	q := client.Query(query)
@@ -266,22 +271,84 @@ func (m *MetricSet) queryBigQuery(ctx context.Context, client *bigquery.Client,
 			return events, err
 		}
 
-		if len(row) == 6 {
+		// Detailed tables have 1 more field which is fetched
+		// price.effective_price
+		switch len(row) {
+		case StandardTableRowLen:
 			events = append(events, createEvents(row, m.config.ProjectID))
+		case DetailedTableRowLen:
+			events = append(events, createDetailedEvents(row, m.config.ProjectID))
 		}
 	}
 	return events, nil
 }
 
+func createTagsMap(tagsRowItem bigquery.Value) mapstr.M {
+	tagsMap := mapstr.M{}
+
+	if tags, ok := tagsRowItem.(string); ok {
+		pairs := strings.Split(tags, ",")
+		for _, pair := range pairs {
+			kv := strings.Split(pair, ":")
+			if len(kv) == 2 {
+				_, _ = tagsMap.Put(kv[0], kv[1])
+			}
+		}
+	}
+
+	return tagsMap.Flatten()
+}
+
 func createEvents(rowItems []bigquery.Value, projectID string) mb.Event {
 	event := mb.Event{}
+
+	tagsMap := createTagsMap(rowItems[9])
+
 	event.MetricSetFields = mapstr.M{
-		"invoice_month":      rowItems[0],
-		"project_id":         rowItems[1],
-		"project_name":       rowItems[2],
-		"billing_account_id": rowItems[3],
-		"cost_type":          rowItems[4],
-		"total":              rowItems[5],
+		"invoice_month":       rowItems[0],
+		"project_id":          rowItems[1],
+		"project_name":        rowItems[2],
+		"billing_account_id":  rowItems[3],
+		"cost_type":           rowItems[4],
+		"total":               rowItems[10],
+		"sku_id":              rowItems[5],
+		"sku_description":     rowItems[6],
+		"service_id":          rowItems[7],
+		"service_description": rowItems[8],
+		"tags":                tagsMap,
+	}
+
+	event.RootFields = mapstr.M{
+		"cloud.provider":     "gcp",
+		"cloud.project.id":   projectID,
+		"cloud.project.name": rowItems[2],
+		"cloud.account.id":   rowItems[3],
+	}
+
+	// create eventID for each current_date + invoice_month + project_id + cost_type
+	currentDate := getCurrentDate()
+	event.ID = generateEventID(currentDate, rowItems)
+	return event
+}
+
+func createDetailedEvents(rowItems []bigquery.Value, projectID string) mb.Event {
+	event := mb.Event{}
+
+	tagsMap := createTagsMap(rowItems[10])
+
+	event.MetricSetFields = mapstr.M{
+		"invoice_month":       rowItems[0],
+		"project_id":          rowItems[1],
+		"project_name":        rowItems[2],
+		"billing_account_id":  rowItems[3],
+		"cost_type":           rowItems[4],
+		"total":               rowItems[11],
+		"sku_id":              rowItems[5],
+		"sku_description":     rowItems[6],
+		"service_id":          rowItems[7],
+		"service_description": rowItems[8],
+		"effective_price":     rowItems[9],
+		"tags":                tagsMap,
 	}
 
 	event.RootFields = mapstr.M{
@@ -314,27 +381,135 @@ func generateEventID(currentDate string, rowItems []bigquery.Value) string {
 
 // generateQuery returns the query to be used by the BigQuery client to retrieve monthly
 // cost types breakdown.
-func generateQuery(tableName, month, costType string) string {
+func (m *MetricSet) generateQuery(tableName, month, costType string) string {
+	/*
+		Standard tables have the following format:
+		- gcp_billing_export_v1_011702_58A742_BEB4E7
+
+		Detailed tables have the following format:
+		- gcp_billing_export_resource_v1_011702_58A742_BEB4E7
+	*/
+
+	if strings.Contains(strings.ToLower(m.config.TablePattern), "resource") {
+		return createDetailedQuery(tableName, month, costType)
+	}
+
+	return createStandardQuery(tableName, month, costType)
+}
+
+func createStandardQuery(tableName, month, costType string) string {
+	// The table name is user provided, so it may contains special characters.
+	// In order to allow any character in the table identifier, use the Quoted identifier format.
+	// See https://github.com/elastic/beats/issues/26855
+	// NOTE: is not possible to escape backtics (`) in a multiline string
+	escapedTableName := fmt.Sprintf("`%s`", tableName)
+
+	query := fmt.Sprintf(`
+SELECT
+	invoice.month AS invoice_month,
+	project.id AS project_id,
+	project.name AS project_name,
+	billing_account_id,
+	cost_type,
+	IFNULL(sku.id, '') AS sku_id,
+	IFNULL(sku.description, '') AS sku_description,
+	IFNULL(service.id, '') AS service_id,
+	IFNULL(service.description, '') AS service_description,
+	ARRAY_TO_STRING(ARRAY(SELECT CONCAT(t.key, ':', t.value) FROM UNNEST(system_labels) AS t), ',') AS tags_string,
+	(SUM(CAST(cost * 1000000 AS int64)) + SUM(IFNULL((
+			SELECT
+				SUM(CAST(c.amount * 1000000 AS int64))
+			FROM
+				UNNEST(credits) c), 0))) / 1000000 AS total_exact
+FROM
+	%s
+WHERE
+	project.id IS NOT NULL
+	AND invoice.month = '%s'
+	AND cost_type = '%s'
+GROUP BY
+	invoice_month,
+	project_id,
+	project_name,
+	billing_account_id,
+	cost_type,
+	sku_id,
+	sku_description,
+	service_id,
+	service_description,
+	tags_string
+ORDER BY
+	invoice_month ASC,
+	project_id ASC,
+	project_name ASC,
+	billing_account_id ASC,
+	cost_type ASC,
+	sku_id ASC,
+	sku_description ASC,
+	service_id ASC,
+	service_description ASC,
+	tags_string ASC;`,
+		escapedTableName, month, costType)
+
+	return query
+}
+
+func createDetailedQuery(tableName, month, costType string) string {
 	// The table name is user provided, so it may contains special characters.
 	// In order to allow any character in the table identifier, use the Quoted identifier format.
 	// See https://github.com/elastic/beats/issues/26855
 	// NOTE: is not possible to escape backtics (`) in a multiline string
 	escapedTableName := fmt.Sprintf("`%s`", tableName)
+
 	query := fmt.Sprintf(`
 SELECT
-	invoice.month,
-	project.id,
-	project.name,
+	invoice.month AS invoice_month,
+	project.id AS project_id,
+	project.name AS project_name,
 	billing_account_id,
 	cost_type,
-	(SUM(CAST(cost * 1000000 AS int64))
-	+ SUM(IFNULL((SELECT SUM(CAST(c.amount * 1000000 as int64)) FROM UNNEST(credits) c), 0))) / 1000000
-	AS total_exact
-FROM %s
-WHERE project.id IS NOT NULL
-AND invoice.month = '%s'
-AND cost_type = '%s'
-GROUP BY 1, 2, 3, 4, 5
-ORDER BY 1 ASC, 2 ASC, 3 ASC, 4 ASC, 5 ASC;`, escapedTableName, month, costType)
+	IFNULL(sku.id, '') AS sku_id,
+	IFNULL(sku.description, '') AS sku_description,
+	IFNULL(service.id, '') AS service_id,
+	IFNULL(service.description, '') AS service_description,
+	IFNULL(SAFE_CAST(price.effective_price AS float64), 0) AS effective_price,
+	ARRAY_TO_STRING(ARRAY(SELECT CONCAT(t.key, ':', t.value) FROM UNNEST(system_labels) AS t), ',') AS tags_string,
+	(SUM(CAST(cost * 1000000 AS int64)) + SUM(IFNULL((
+			SELECT
+				SUM(CAST(c.amount * 1000000 AS int64))
+			FROM
+				UNNEST(credits) c), 0))) / 1000000 AS total_exact
+FROM
+	%s
+WHERE
+	project.id IS NOT NULL
+	AND invoice.month = '%s'
+	AND cost_type = '%s'
+GROUP BY
+	invoice_month,
+	project_id,
+	project_name,
+	billing_account_id,
+	cost_type,
+	sku_id,
+	sku_description,
+	service_id,
+	service_description,
+	effective_price,
+	tags_string
+ORDER BY
+	invoice_month ASC,
+	project_id ASC,
+	project_name ASC,
+	billing_account_id ASC,
+	cost_type ASC,
+	sku_id ASC,
+	sku_description ASC,
+	service_id ASC,
+	service_description ASC,
+	effective_price ASC,
+	tags_string ASC;`,
+		escapedTableName, month, costType)
+
 	return query
 }

x-pack/metricbeat/module/gcp/billing/billing_test.go

@@ -5,7 +5,8 @@
 package billing
 
 import (
-	"io/ioutil"
+	"github.com/elastic/beats/v7/metricbeat/mb"
+	"io"
 	"log"
 	"strconv"
 	"testing"
@@ -20,15 +21,21 @@ func TestGetCurrentMonth(t *testing.T) {
 }
 
 func TestGenerateQuery(t *testing.T) {
-	log.SetOutput(ioutil.Discard)
+	log.SetOutput(io.Discard)
 
-	query := generateQuery("my-table", "jan", "cost")
+	m := MetricSet{
+		BaseMetricSet: mb.BaseMetricSet{},
+		config:        config{},
+		logger:        nil,
+	}
+
+	query := m.generateQuery("my-table", "jan", "cost")
 	log.Println(query)
 
-	// verify that table name quoting is in effect
+	//verify that table name quoting is in effect
 	assert.Contains(t, query, "`my-table`")
-	// verify the group by is preserved
-	assert.Contains(t, query, "GROUP BY 1, 2, 3, 4, 5")
-	// verify the order by is preserved
-	assert.Contains(t, query, "ORDER BY 1 ASC, 2 ASC, 3 ASC, 4 ASC, 5 ASC")
+	//verify the group by is preserved
+	assert.Contains(t, query, "GROUP BY\n\tinvoice_month,\n\tproject_id,\n\tproject_name,\n\tbilling_account_id,\n\tcost_type")
+	//verify the order by is preserved
+	assert.Contains(t, query, "ORDER BY\n\tinvoice_month ASC,\n\tproject_id ASC,\n\tproject_name ASC,\n\tbilling_account_id ASC,\n\tcost_type ASC")
 }