Add option to log queries only when rows returned/affected >= X

go/streamlog/streamlog.go

@@ -42,7 +42,10 @@ var (
 	QueryLogFormat = flag.String("querylog-format", "text", "format for query logs (\"text\" or \"json\")")
 
 	// QueryLogFilterTag contains an optional string that must be present in the query for it to be logged
-	QueryLogFilterTag = flag.String("querylog-filter-tag", "", "string that must be present in the query for it to be logged")
+	QueryLogFilterTag = flag.String("querylog-filter-tag", "", "string that must be present in the query for it to be logged; if using a value as the tag, you need to disable query normalization")
+
+	// QueryLogRowLimit:  only log queries returning or affecting this many rows
+	QueryLogRowLimit = flag.Uint64("querylog-row-limit", 0, "Minimum number of rows a query has to return or affect before being logged (default 0)")
 
 	sendCount      = stats.NewCountersWithSingleLabel("StreamlogSend", "stream log send count", "logger_names")
 	deliveredCount = stats.NewCountersWithMultiLabels(
@@ -208,9 +211,12 @@ func GetFormatter(logger *StreamLogger) LogFormatter {
 
 // ShouldEmitLog returns whether the log with the given SQL query
 // should be emitted or filtered
-func ShouldEmitLog(sql string) bool {
-	if *QueryLogFilterTag == "" {
-		return true
+func ShouldEmitLog(sql string, rowsAffected uint64) bool {
+	if *QueryLogRowLimit > rowsAffected && *QueryLogFilterTag == "" {
+		return false
+	}
+	if *QueryLogFilterTag != "" {
+		return strings.Contains(sql, *QueryLogFilterTag)
 	}
-	return strings.Contains(sql, *QueryLogFilterTag)
+	return true
 }

go/vt/vtgate/logstats.go

@@ -123,7 +123,7 @@ func (stats *LogStats) RemoteAddrUsername() (string, string) {
 // Logf formats the log record to the given writer, either as
 // tab-separated list of logged fields or as JSON.
 func (stats *LogStats) Logf(w io.Writer, params url.Values) error {
-	if !streamlog.ShouldEmitLog(stats.SQL) {
+	if !streamlog.ShouldEmitLog(stats.SQL, stats.RowsAffected) {
 		return nil
 	}
 

go/vt/vtgate/logstats_test.go

@@ -155,6 +155,28 @@ func TestLogStatsFilter(t *testing.T) {
 	}
 }
 
+func TestLogStatsRowLimit(t *testing.T) {
+	defer func() { *streamlog.QueryLogRowLimit = 0 }()
+
+	logStats := NewLogStats(context.Background(), "test", "sql1 /* LOG_THIS_QUERY */", map[string]*querypb.BindVariable{"intVal": sqltypes.Int64BindVariable(1)})
+	logStats.StartTime = time.Date(2017, time.January, 1, 1, 2, 3, 0, time.UTC)
+	logStats.EndTime = time.Date(2017, time.January, 1, 1, 2, 4, 1234, time.UTC)
+	params := map[string][]string{"full": {}}
+
+	got := testFormat(logStats, url.Values(params))
+	want := "test\t\t\t''\t''\t2017-01-01 01:02:03.000000\t2017-01-01 01:02:04.000001\t1.000001\t0.000000\t0.000000\t0.000000\t\t\"sql1 /* LOG_THIS_QUERY */\"\tmap[intVal:type:INT64 value:\"1\" ]\t0\t0\t\"\"\t\n"
+	if got != want {
+		t.Errorf("logstats format: got:\n%q\nwant:\n%q\n", got, want)
+	}
+
+	*streamlog.QueryLogRowLimit = 10
+	got = testFormat(logStats, url.Values(params))
+	want = ""
+	if got != want {
+		t.Errorf("logstats format: got:\n%q\nwant:\n%q\n", got, want)
+	}
+}
+
 func TestLogStatsContextHTML(t *testing.T) {
 	html := "HtmlContext"
 	callInfo := &fakecallinfo.FakeCallInfo{

go/vt/vttablet/tabletserver/query_executor.go

@@ -87,7 +87,7 @@ func (qre *QueryExecutor) Execute() (reply *sqltypes.Result, err error) {
 		}
 		qre.tsv.qe.AddStats(planName, tableName, 1, duration, mysqlTime, int64(reply.RowsAffected), 0)
 		qre.plan.AddStats(1, duration, mysqlTime, int64(reply.RowsAffected), 0)
-		qre.logStats.RowsAffected = int(reply.RowsAffected)
+		qre.logStats.RowsAffected = uint64(reply.RowsAffected)
 		qre.logStats.Rows = reply.Rows
 		tabletenv.ResultStats.Add(int64(len(reply.Rows)))
 	}(time.Now())

go/vt/vttablet/tabletserver/tabletenv/logstats.go

@@ -50,7 +50,7 @@ type LogStats struct {
 	OriginalSQL          string
 	BindVariables        map[string]*querypb.BindVariable
 	rewrittenSqls        []string
-	RowsAffected         int
+	RowsAffected         uint64
 	NumberOfQueries      int
 	StartTime            time.Time
 	EndTime              time.Time
@@ -180,7 +180,7 @@ func (stats *LogStats) CallInfo() (string, string) {
 // Logf formats the log record to the given writer, either as
 // tab-separated list of logged fields or as JSON.
 func (stats *LogStats) Logf(w io.Writer, params url.Values) error {
-	if !streamlog.ShouldEmitLog(stats.OriginalSQL) {
+	if !streamlog.ShouldEmitLog(stats.OriginalSQL, stats.RowsAffected) {
 		return nil
 	}