Merge branch 'master' of https://github.com/percona/go-mysql into thread_id_support

Author: Rolf Martin-Hoster

event/aggregator.go

@@ -89,7 +89,7 @@ func (a *Aggregator) Finalize() Result {
 		class.Finalize(a.rateLimit)
 		class.UniqueQueries = 1
 		if class.Example != nil && class.Example.Ts != "" {
-			if t, err := time.Parse("060102 15:04:05", class.Example.Ts); err != nil {
+			if t, err := time.Parse("2006-01-02 15:04:05", class.Example.Ts); err != nil {
 				class.Example.Ts = ""
 			} else {
 				class.Example.Ts = t.Add(a.utcOffset).Format("2006-01-02 15:04:05")

event/aggregator_test.go

@@ -46,7 +46,9 @@ func aggregateSlowLog(input, output string, utcOffset time.Duration, examples bo
 	if err != nil {
 		l.Fatal(err)
 	}
-	p := parser.NewSlowLogParser(file, log.Options{})
+	opt := log.Options{}
+	opt.DefaultLocation = time.UTC
+	p := parser.NewSlowLogParser(file, opt)
 	go p.Start()
 	a := event.NewAggregator(examples, utcOffset, 10)
 	for e := range p.EventChan() {

event/class.go

@@ -100,7 +100,10 @@ func (c *Class) AddEvent(e *log.Event, outlier bool) {
 				} else {
 					c.Example.Query = e.Query
 				}
-				c.Example.Ts = e.Ts
+				if !e.Ts.IsZero() {
+					// todo use time.RFC3339Nano instead
+					c.Example.Ts = e.Ts.UTC().Format("2006-01-02 15:04:05")
+				}
 			}
 		}
 	}

log/log.go

@@ -20,17 +20,22 @@
 // like max Query_time. See also percona.com/go-mysql/event/.
 package log
 
+import (
+	"time"
+)
+
 // An event is a query like "SELECT col FROM t WHERE id = 1", some metrics like
 // Query_time (slow log) or SUM_TIMER_WAIT (Performance Schema), and other
 // metadata like default database, timestamp, etc. Metrics and metadata are not
 // guaranteed to be defined--and frequently they are not--but at minimum an
 // event is expected to define the query and Query_time metric. Other metrics
 // and metadata vary according to MySQL version, distro, and configuration.
 type Event struct {
-	Offset        uint64 // byte offset in file at which event starts
-	Ts            string // raw timestamp of event
-	Admin         bool   // true if Query is admin command
-	Query         string // SQL query or admin command
+	Offset        uint64    // byte offset in file at which event starts
+	OffsetEnd     uint64    // byte offset in file at which event ends
+	Ts            time.Time // timestamp of event
+	Admin         bool      // true if Query is admin command
+	Query         string    // SQL query or admin command
 	User          string
 	Host          string
 	Db            string
@@ -56,6 +61,7 @@ type Options struct {
 	StartOffset        uint64          // byte offset in file at which to start parsing
 	FilterAdminCommand map[string]bool // admin commands to ignore
 	Debug              bool            // print trace info to STDOUT
+	DefaultLocation    *time.Location  // DefaultLocation to assume for logs in MySQL < 5.7 format.
 }
 
 // A LogParser sends events to a channel.

log/slow/parser.go

@@ -27,20 +27,24 @@ import (
 	"regexp"
 	"strconv"
 	"strings"
+	"time"
 
 	"github.com/percona/go-mysql/log"
 )
 
 // Regular expressions to match important lines in slow log.
-var timeRe = regexp.MustCompile(`Time: (\S+\s{1,2}\S+)`)
-var userRe = regexp.MustCompile(`User@Host: ([^\[]+|\[[^[]+\]).*?@ (\S*) \[(.*)\]`)
-var idRe = regexp.MustCompile(`(Id|Thread_id): +([0-9]*)`)
-var schema = regexp.MustCompile(`Schema: +(.*?) +Last_errno:`)
-var headerRe = regexp.MustCompile(`^#\s+[A-Z]`)
-var metricsRe = regexp.MustCompile(`(\w+): (\S+|\z)`)
-var adminRe = regexp.MustCompile(`command: (.+)`)
-var setRe = regexp.MustCompile(`^SET (?:last_insert_id|insert_id|timestamp)`)
-var useRe = regexp.MustCompile(`^(?i)use `)
+var (
+	timeRe    = regexp.MustCompile(`Time: (\S+\s{1,2}\S+)`)
+	timeNewRe = regexp.MustCompile(`Time:\s+(\d{4}-\d{2}-\d{2}\S+)`)
+	userRe    = regexp.MustCompile(`User@Host: ([^\[]+|\[[^[]+\]).*?@ (\S*) \[(.*)\]`)
+	idRe      = regexp.MustCompile(`(Id|Thread_id): +([0-9]*)`)
+	schema    = regexp.MustCompile(`Schema: +(.*?) +Last_errno:`)
+	headerRe  = regexp.MustCompile(`^#\s+[A-Z]`)
+	metricsRe = regexp.MustCompile(`(\w+): (\S+|\z)`)
+	adminRe   = regexp.MustCompile(`command: (.+)`)
+	setRe     = regexp.MustCompile(`^SET (?:last_insert_id|insert_id|timestamp)`)
+	useRe     = regexp.MustCompile(`^(?i)use `)
+)
 
 // A SlowLogParser parses a MySQL slow log. It implements the LogParser interface.
 type SlowLogParser struct {
@@ -55,12 +59,17 @@ type SlowLogParser struct {
 	queryLines  uint64
 	bytesRead   uint64
 	lineOffset  uint64
+	endOffset   uint64
 	stopped     bool
 	event       *log.Event
 }
 
 // NewSlowLogParser returns a new SlowLogParser that reads from the open file.
 func NewSlowLogParser(file *os.File, opt log.Options) *SlowLogParser {
+	if opt.DefaultLocation == nil {
+		// Old MySQL format assumes time is taken from SYSTEM.
+		opt.DefaultLocation = time.Local
+	}
 	p := &SlowLogParser{
 		file: file,
 		opt:  opt,
@@ -136,12 +145,6 @@ SCANNER_LOOP:
 		lineLen := uint64(len(line))
 		p.bytesRead += lineLen
 		p.lineOffset = p.bytesRead - lineLen
-		if p.lineOffset != 0 {
-			// @todo Need to get clear on why this is needed;
-			// it does make the value correct; an off-by-one issue
-			p.lineOffset += 1
-		}
-
 		if p.opt.Debug {
 			fmt.Println()
 			l.Printf("+%d line: %s", p.lineOffset, line)
@@ -181,6 +184,7 @@ SCANNER_LOOP:
 	}
 
 	if !p.stopped && p.queryLines > 0 {
+		p.endOffset = p.bytesRead
 		p.sendEvent(false, false)
 	}
 
@@ -214,10 +218,16 @@ func (p *SlowLogParser) parseHeader(line string) {
 			l.Println("time")
 		}
 		m := timeRe.FindStringSubmatch(line)
-		if len(m) < 2 {
-			return
+		if len(m) == 2 {
+			p.event.Ts, _ = time.ParseInLocation("060102 15:04:05", m[1], p.opt.DefaultLocation)
+		} else {
+			m = timeNewRe.FindStringSubmatch(line)
+			if len(m) == 2 {
+				p.event.Ts, _ = time.ParseInLocation(time.RFC3339Nano, m[1], p.opt.DefaultLocation)
+			} else {
+				return
+			}
 		}
-		p.event.Ts = m[1]
 		if userRe.MatchString(line) {
 			if p.opt.Debug {
 				l.Println("user (bad format)")
@@ -300,6 +310,7 @@ func (p *SlowLogParser) parseQuery(line string) {
 		}
 		p.inHeader = true
 		p.inQuery = false
+		p.endOffset = p.lineOffset
 		p.sendEvent(true, false)
 		p.parseHeader(line)
 		return
@@ -352,6 +363,7 @@ func (p *SlowLogParser) parseAdmin(line string) {
 		if p.opt.Debug {
 			l.Println("not filtered")
 		}
+		p.endOffset = p.bytesRead
 		p.sendEvent(false, false)
 	} else {
 		p.inHeader = false
@@ -364,6 +376,8 @@ func (p *SlowLogParser) sendEvent(inHeader bool, inQuery bool) {
 		l.Println("send event")
 	}
 
+	p.event.OffsetEnd = p.endOffset
+
 	// Make a new event and reset our metadata.
 	defer func() {
 		p.event = log.NewEvent()