Add log user query param

common/dynamicconfig/constants.go

@@ -1444,6 +1444,11 @@ const (
 	// Default value: true
 	// Allowed filters: DomainName
 	EnableReadVisibilityFromPinot
+	// EnableLogCustomerQueryParameter is key for enable log customer query parameters
+	// KeyName: system.enableLogCustomerQueryParameter
+	// Value type: Bool
+	// Default value: false
+	EnableLogCustomerQueryParameter
 	// EmitShardDiffLog is whether emit the shard diff log
 	// KeyName: history.emitShardDiffLog
 	// Value type: Bool
@@ -3683,9 +3688,16 @@ var BoolKeys = map[BoolKey]DynamicBool{
 	},
 	EnableReadVisibilityFromPinot: DynamicBool{
 		KeyName:      "system.enableReadVisibilityFromPinot",
+		Filters:      []Filter{DomainName},
 		Description:  "EnableReadVisibilityFromPinot is key for enable read from pinot or db visibility, usually using with AdvancedVisibilityWritingMode for seamless migration from db visibility to advanced visibility",
 		DefaultValue: true,
 	},
+	EnableLogCustomerQueryParameter: DynamicBool{
+		KeyName:      "system.enableLogCustomerQueryParameter",
+		Filters:      []Filter{DomainName},
+		Description:  "EnableLogCustomerQueryParameter is key for enable log customer query parameters",
+		DefaultValue: false,
+	},
 	EmitShardDiffLog: DynamicBool{
 		KeyName:      "history.emitShardDiffLog",
 		Description:  "EmitShardDiffLog is whether emit the shard diff log",

common/log/tag/tags.go

@@ -180,6 +180,16 @@ func WorkflowCronSchedule(schedule string) Tag {
 	return newStringTag("wf-cron-schedule", schedule)
 }
 
+// WorkflowCloseStatus returns a tag to report a workflow's close status
+func WorkflowCloseStatus(status int) Tag {
+	return newInt("close-status", status)
+}
+
+// IsWorkflowOpen returns a tag to report a workflow is open or not
+func IsWorkflowOpen(isOpen bool) Tag {
+	return newBoolTag("is-workflow-open", isOpen)
+}
+
 // domain related
 
 // WorkflowDomainID returns tag for WorkflowDomainID

common/persistence/client/factory.go

@@ -299,6 +299,7 @@ func (f *factoryImpl) NewVisibilityManager(
 			resourceConfig.EnableReadVisibilityFromPinot,
 			resourceConfig.EnableReadVisibilityFromES,
 			resourceConfig.AdvancedVisibilityWritingMode,
+			resourceConfig.EnableLogCustomerQueryParameter,
 			f.logger,
 		), nil
 	} else if params.PersistenceConfig.AdvancedVisibilityStore != "" {

common/persistence/pinotUserQueryTimeFilter.go

@@ -0,0 +1,172 @@
+// The MIT License (MIT)
+
+// Copyright (c) 2017-2020 Uber Technologies Inc.
+
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in all
+// copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+// SOFTWARE.
+
+package persistence
+
+import (
+	"errors"
+	"fmt"
+	"strings"
+
+	"github.com/xwb1989/sqlparser"
+
+	"github.com/uber/cadence/common"
+	"github.com/uber/cadence/common/definition"
+	"github.com/uber/cadence/common/types"
+)
+
+// VisibilityQueryFilter for sql query validation
+type VisibilityQueryFilter struct {
+}
+
+// NewVisibilityQueryFilter create VisibilityQueryValidator
+func NewVisibilityQueryFilter() *VisibilityQueryFilter {
+	return &VisibilityQueryFilter{}
+}
+
+// FilterQuery validates that search attributes in the query and returns modified query.
+func (qv *VisibilityQueryFilter) FilterQuery(whereClause string) (string, error) {
+	if len(whereClause) != 0 {
+		// Build a placeholder query that allows us to easily parse the contents of the where clause.
+		// IMPORTANT: This query is never executed, it is just used to parse and validate whereClause
+		var placeholderQuery string
+		whereClause := strings.TrimSpace(whereClause)
+		if common.IsJustOrderByClause(whereClause) { // just order by
+			placeholderQuery = fmt.Sprintf("SELECT * FROM dummy %s", whereClause)
+		} else {
+			placeholderQuery = fmt.Sprintf("SELECT * FROM dummy WHERE %s", whereClause)
+		}
+
+		stmt, err := sqlparser.Parse(placeholderQuery)
+		if err != nil {
+			return "", &types.BadRequestError{Message: "Invalid query."}
+		}
+
+		sel, ok := stmt.(*sqlparser.Select)
+		if !ok {
+			return "", &types.BadRequestError{Message: "Invalid select query."}
+		}
+		buf := sqlparser.NewTrackedBuffer(nil)
+		res := ""
+		// validate where expr
+		if sel.Where != nil {
+			res, err = qv.validateWhereExpr(sel.Where.Expr)
+			if err != nil {
+				return "", &types.BadRequestError{Message: err.Error()}
+			}
+		}
+
+		sel.OrderBy.Format(buf)
+		res += buf.String()
+		return res, nil
+	}
+	return whereClause, nil
+}
+
+func (qv *VisibilityQueryFilter) validateWhereExpr(expr sqlparser.Expr) (string, error) {
+	if expr == nil {
+		return "", nil
+	}
+	switch expr := expr.(type) {
+	case *sqlparser.AndExpr, *sqlparser.OrExpr:
+		return qv.filterAndOrExpr(expr)
+	case *sqlparser.ComparisonExpr:
+		return qv.filterComparisonExpr(expr)
+	case *sqlparser.RangeCond:
+		return qv.filterRangeExpr(expr)
+	case *sqlparser.ParenExpr:
+		return qv.validateWhereExpr(expr.Expr)
+	default:
+		return "", errors.New("invalid where clause")
+	}
+}
+
+// for "between...and..." only
+func (qv *VisibilityQueryFilter) filterRangeExpr(expr sqlparser.Expr) (string, error) {
+	buf := sqlparser.NewTrackedBuffer(nil)
+	rangeCond := expr.(*sqlparser.RangeCond)
+	colName, ok := rangeCond.Left.(*sqlparser.ColName)
+	if !ok {
+		return "", errors.New("invalid range expression: fail to get colname")
+	}
+	colNameStr := colName.Name.String()
+
+	if colNameStr == definition.StartTime {
+		return "", nil
+	}
+	expr.Format(buf)
+	return buf.String(), nil
+}
+
+func (qv *VisibilityQueryFilter) filterAndOrExpr(expr sqlparser.Expr) (string, error) {
+	var leftExpr sqlparser.Expr
+	var rightExpr sqlparser.Expr
+	isAnd := false
+
+	switch expr := expr.(type) {
+	case *sqlparser.AndExpr:
+		leftExpr = expr.Left
+		rightExpr = expr.Right
+		isAnd = true
+	case *sqlparser.OrExpr:
+		leftExpr = expr.Left
+		rightExpr = expr.Right
+	}
+
+	leftRes, err := qv.validateWhereExpr(leftExpr)
+	if err != nil {
+		return "", err
+	}
+
+	rightRes, err := qv.validateWhereExpr(rightExpr)
+	if err != nil {
+		return "", err
+	}
+
+	if leftRes == "" || rightRes == "" {
+		return leftRes + rightRes, nil
+	}
+
+	if isAnd {
+		return fmt.Sprintf("%s and %s", leftRes, rightRes), nil
+	}
+
+	return fmt.Sprintf("%s or %s", leftRes, rightRes), nil
+}
+
+func (qv *VisibilityQueryFilter) filterComparisonExpr(expr sqlparser.Expr) (string, error) {
+	comparisonExpr := expr.(*sqlparser.ComparisonExpr)
+	buf := sqlparser.NewTrackedBuffer(nil)
+
+	colName, ok := comparisonExpr.Left.(*sqlparser.ColName)
+	if !ok {
+		return "", errors.New("invalid comparison expression, left")
+	}
+
+	colNameStr := colName.Name.String()
+
+	if colNameStr == definition.StartTime {
+		return "", nil
+	}
+	expr.Format(buf)
+	return buf.String(), nil
+}

common/persistence/pinotUserQueryTimeFilter_test.go

@@ -0,0 +1,106 @@
+// The MIT License (MIT)
+
+// Copyright (c) 2017-2020 Uber Technologies Inc.
+
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in all
+// copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+// SOFTWARE.
+
+package persistence
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestFilterQuery(t *testing.T) {
+	tests := map[string]struct {
+		query     string
+		validated string
+		err       string
+	}{
+		"Case1: empty query": {
+			query:     "",
+			validated: "",
+		},
+		"Case2: simple query": {
+			query:     "WorkflowID = 'wid'",
+			validated: "WorkflowID = 'wid'",
+		},
+		"Case3: query with custom field": {
+			query:     "CustomStringField = 'custom'",
+			validated: "CustomStringField = 'custom'",
+		},
+		"Case4: startTime query": {
+			query:     "StartTime=123456789",
+			validated: "",
+		},
+		"Case5-1: complex query I: with parenthesis": {
+			query:     "(CustomStringField = 'custom and custom2 or custom3 order by') or StartTime between 1 and 10",
+			validated: "CustomStringField = 'custom and custom2 or custom3 order by'",
+		},
+		"Case5-2: complex query II: with startTime": {
+			query:     "DomainID = 'd-id' and (RunID = 'run-id' or StartTime < 123456778)",
+			validated: "DomainID = 'd-id' and RunID = 'run-id'",
+		},
+		"Case5-3: complex query II: without startTime": {
+			query:     "DomainID = 'd-id' or RunID = 'run-id' and WorkflowID = 'wid'",
+			validated: "DomainID = 'd-id' or RunID = 'run-id' and WorkflowID = 'wid'",
+		},
+		"Case5-4: complex query IV": {
+			query:     "WorkflowID = 'wid' and (CustomStringField = 'custom and custom2 or custom3 order by' or StartTime between 1 and 10)",
+			validated: "WorkflowID = 'wid' and CustomStringField = 'custom and custom2 or custom3 order by'",
+		},
+		"Case6: invalid sql query": {
+			query: "Invalid SQL",
+			err:   "Invalid query.",
+		},
+		"Case7: query with missing val": {
+			query:     "CloseTime = missing",
+			validated: "CloseTime = missing",
+		},
+		"Case8: invalid where expression": {
+			query: "InvalidWhereExpr",
+			err:   "invalid where clause",
+		},
+		"Case9: or clause": {
+			query:     "CustomIntField = 1 or CustomIntField = 2",
+			validated: "CustomIntField = 1 or CustomIntField = 2",
+		},
+		"Case10: range query: custom filed": {
+			query:     "CustomIntField BETWEEN 1 AND 2",
+			validated: "CustomIntField between 1 and 2",
+		},
+		"Case11: system date attribute greater than or equal to": {
+			query:     "StartTime >= 1697754674",
+			validated: "",
+		},
+	}
+
+	for name, test := range tests {
+		t.Run(name, func(t *testing.T) {
+			qv := NewVisibilityQueryFilter()
+			validated, err := qv.FilterQuery(test.query)
+			if err != nil {
+				assert.Equal(t, test.err, err.Error())
+			} else {
+				assert.Equal(t, test.validated, validated)
+			}
+		})
+	}
+}