change config to specify timestamp/cursor per query

Author: tommyers-elastic

x-pack/filebeat/input/gcpbigquery/config.go

@@ -14,31 +14,36 @@ type config struct {
 	// The ID of the GCP project that owns the BigQuery dataset.
 	ProjectID string `config:"project_id" validate:"required"`
 
-	// The BigQuery SQL queries to execute.
-	Queries []string `config:"queries" validate:"required"`
+	// JSON file containing authentication credentials and key.
+	CredentialsFile string `config:"credentials_file"`
+
+	// The BigQuery queries to execute, along with options for each.
+	Queries []queryConfig `config:"queries" validate:"required"`
+}
+
+type queryConfig struct {
+	// The SQL query to execute.
+	Query string `config:"query" validate:"required"`
 
 	// The name of a field in the target BigQuery table that can be used to simulate cursor pagination, e.g. an incremental ID or timestamp.
 	// The following field types are supported: BIGNUMERIC, BYTES, DATE, DATETIME, FLOAT, INTEGER, NUMERIC, STRING, TIME, TIMESTAMP.
 	// If not specified, the input will run the configured queries as-is on every poll. If specified, the input will add a WHERE clause
 	// to each query to only select rows where the cursor field's value is greater than the last seen value.
 	CursorField string `config:"cursor_field"`
 
-	// JSON file containing authentication credentials and key.
-	CredentialsFile string `config:"credentials_file"`
+	// A TIMESTAMP field in the target BigQuery table to use as the event's @timestamp value.
+	TimestampField string `config:"timestamp_field"`
 
 	// Whether to attempt to parse fields of type JSON into objects/arrays instead of leaving them as strings.
 	// In the event of parsing failures, we still expand the field into a JSON object with a single field named
 	// "original" containing the original string value; this avoids mapping conflicts in Elasticsearch.
-	ExpandJsonStrings bool `config:"expand_json_strings"`
-
-	// A TIMESTAMP field in the target BigQuery table to use as the event's @timestamp value.
-	TimestampField string `config:"timestamp_field"`
+	// Defaults to true.
+	ExpandJsonStrings *bool `config:"expand_json_strings"`
 }
 
 func defaultConfig() config {
 	return config{
-		Period:            time.Minute,
-		ExpandJsonStrings: true,
+		Period: time.Minute,
 	}
 }
 

x-pack/filebeat/input/gcpbigquery/input.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"cloud.google.com/go/bigquery"
+	"github.com/cespare/xxhash/v2"
 	v2 "github.com/elastic/beats/v7/filebeat/input/v2"
 	cursor "github.com/elastic/beats/v7/filebeat/input/v2/input-cursor"
 	"github.com/elastic/beats/v7/libbeat/beat"
@@ -45,10 +46,18 @@ func configure(cfg *conf.C, logger *logp.Logger) ([]cursor.Source, cursor.Input,
 
 	var sources []cursor.Source
 	for _, query := range config.Queries {
+		// defaults to true
+		expandJSON := true
+		if query.ExpandJsonStrings != nil {
+			expandJSON = *query.ExpandJsonStrings
+		}
+
 		sources = append(sources, &bigQuerySource{
-			ProjectID:   config.ProjectID,
-			Query:       query,
-			CursorField: config.CursorField,
+			ProjectID:      config.ProjectID,
+			Query:          query.Query,
+			CursorField:    query.CursorField,
+			TimestampField: query.TimestampField,
+			ExpandJson:     expandJSON,
 		})
 	}
 
@@ -63,13 +72,19 @@ func updateStatus(ctx v2.Context, status status.Status, msg string) {
 
 // bigQuerySource defines the configuration for a single BigQuery query.
 type bigQuerySource struct {
-	ProjectID   string
-	Query       string
-	CursorField string
+	ProjectID      string
+	Query          string
+	CursorField    string
+	TimestampField string
+	ExpandJson     bool
 }
 
 func (s *bigQuerySource) Name() string {
-	return fmt.Sprintf("%s-%s-%s", s.ProjectID, s.Query, s.CursorField)
+	// this string uniquely identifies the source in the state store.
+	// configuration that doesn't affect the query/cursor itself should not be included.
+	name := fmt.Sprintf("%s-%s-%s", s.ProjectID, s.Query, s.CursorField)
+	// hash it to avoid unintentionally leaching queries into logs/files
+	return fmt.Sprintf("%d", xxhash.Sum64String(name))
 }
 
 type bigQueryInput struct {
@@ -182,7 +197,7 @@ func (bq *bigQueryInput) publishEvent(src *bigQuerySource, publisher cursor.Publ
 			}
 		}
 
-		if bq.config.TimestampField != "" && field.Name == bq.config.TimestampField {
+		if src.TimestampField != "" && field.Name == src.TimestampField {
 			bq.logger.Debugf("setting timestamp from field %s", field.Name)
 
 			ts, err := getTimestamp(field, v)
@@ -193,11 +208,10 @@ func (bq *bigQueryInput) publishEvent(src *bigQuerySource, publisher cursor.Publ
 			}
 		}
 
-		if bq.config.ExpandJsonStrings {
-			bq.logger.Debugf("expanding JSON from field %s", field.Name)
-
+		if src.ExpandJson {
 			val, err := expandJSON(field, v)
 			if err == nil {
+				bq.logger.Debugf("expanding JSON from field %s", field.Name)
 				v = val
 			} else {
 				// on error, still expand into a nested object with the original string to avoid mapping conflicts