<commit_before>// Package mysql parses the mysql slow query log
package mysql

import (
	"database/sql"
	"fmt"
	"log"
	"regexp"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/honeycombio/honeytail/event"

	"github.com/Sirupsen/logrus"
	_ "github.com/go-sql-driver/mysql"
	"github.com/percona/go-mysql/query"
)

// 3 sample log entries
//
// # Time: 2016-04-01T00:31:09.817887Z
// # User@Host: root[root] @ localhost []  Id:   233
// # Query_time: 0.008393  Lock_time: 0.000154 Rows_sent: 1  Rows_examined: 357
// SET timestamp=1459470669;
// show status like 'Uptime';
//
// # Time: 2016-04-01T00:31:09.853523Z
// # User@Host: root[root] @ localhost []  Id:   233
// # Query_time: 0.020424  Lock_time: 0.000147 Rows_sent: 494  Rows_examined: 494
// SET timestamp=1459470669;
// SHOW /*innotop*/ GLOBAL VARIABLES;
//
// # Time: 2016-04-01T00:31:09.856726Z
// # User@Host: root[root] @ localhost []  Id:   233
// # Query_time: 0.000021  Lock_time: 0.000000 Rows_sent: 494  Rows_examined: 494
// SET timestamp=1459470669;
// # administrator command: Ping;
//
// We should ignore the administrator command entry; the stats it presents (eg rows_sent)
// are actually for the previous command

var (
	reTime       = myRegexp{regexp.MustCompile("^# Time: (?P<time>[^ ]+)Z *$")}
	reAdminPing  = myRegexp{regexp.MustCompile("^# administrator command: Ping; *$")}
	reUser       = myRegexp{regexp.MustCompile("^# User@Host: (?P<user>[^ ]+) @ (?P<host>[^ ]+).*$")}
	reQueryStats = myRegexp{regexp.MustCompile("^# Query_time: (?P<queryTime>[0-9.]+) *Lock_time: (?P<lockTime>[0-9.]+) *Rows_sent: (?P<rowsSent>[0-9]+) *Rows_examined: (?P<rowsExamined>[0-9]+) *$")}
	reSetTime    = myRegexp{regexp.MustCompile("^SET timestamp=(?P<unixTime>[0-9]+);$")}
	reQuery      = myRegexp{regexp.MustCompile("^(?P<query>[^#]*).*$")}
	reUse        = myRegexp{regexp.MustCompile("^(?i)use ")}

	rdsStr  = "rds"
	ec2Str  = "ec2"
	selfStr = "self"
)

const timeFormat = "2006-01-02T15:04:05.000000"

type Options struct {
	Host          string `long:"host" description:"MySQL host in the format (address:port)"`
	User          string `long:"user" description:"MySQL username"`
	Pass          string `long:"pass" description:"MySQL password"`
	QueryInterval uint   `long:"interval" description:"interval for querying the MySQL DB in seconds" default:"30"`
}

type Parser struct {
	conf     Options
	wg       sync.WaitGroup
	nower    Nower
	hostedOn string
	readOnly bool
}

type Nower interface {
	Now() time.Time
}

type RealNower struct{}

func (n *RealNower) Now() time.Time {
	return time.Now().UTC()
}

// rawEvent is the unparsed multi-line entry representing a single slow query
type rawEvent struct {
	lines []string
}

// slowQuery represents the structured form of a query from the slow query log
type SlowQuery struct {
	Timestamp       time.Time `json:"time"`
	UnixTime        int       `json:"unixtime"`
	User            string    `json:"user"`
	Host            string    `json:"host"`
	QueryTime       float64   `json:"query_time"`
	LockTime        float64   `json:"lock_time"`
	RowsSent        int       `json:"rows_sent"`
	RowsExamined    int       `json:"rows_examined"`
	Query           string    `json:"query,omitempty"`
	NormalizedQuery string    `json:"normalized_query,omitempty"`
	DB              string    `json:"db,omitempty"`
	HostedOn        string    `json:"hosted_on,omitempty"`
	ReadOnly        bool      `json:"read_only,omitempty"`
	skipQuery       bool
}

func (p *Parser) Init(options interface{}) error {
	p.conf = *options.(*Options)
	p.nower = &RealNower{}
	if p.conf.Host != "" {
		url := fmt.Sprintf("%s:%s@tcp(%s)/", p.conf.User, p.conf.Pass, p.conf.Host)
		db, err := sql.Open("mysql", url)
		if err != nil {
			log.Fatal(err)
		}
		// update hostedOn and readOnly every <n> seconds
		go func() {
			defer db.Close()
			ticker := time.NewTicker(time.Second * time.Duration(p.conf.QueryInterval))
			for _ = range ticker.C {
				p.hostedOn = getHostedOn(db)
				p.readOnly = getReadOnly(db)
			}
		}()
	}
	return nil
}

func getReadOnly(db *sql.DB) bool {
	rows, err := db.Query("SELECT @@global.read_only;")
	if err != nil {
		log.Fatal(err)
	}
	defer rows.Close()

	var value bool

	rows.Next()
	if err := rows.Scan(&value); err != nil {
		log.Fatal(err)
	}
	if err := rows.Err(); err != nil {
		log.Fatal(err)
	}
	return value
}

func getHostedOn(db *sql.DB) string {
	rows, err := db.Query("SHOW GLOBAL VARIABLES WHERE Variable_name = 'basedir';")
	if err != nil {
		log.Fatal(err)
	}
	defer rows.Close()

	var varName, value string

	for rows.Next() {
		if err := rows.Scan(&varName, &value); err != nil {
			log.Fatal(err)
		}
		if strings.HasPrefix(value, "/rdsdbbin/") {
			return rdsStr
		}
	}

	// TODO: implement ec2 detection

	if err = rows.Err(); err != nil {
		log.Fatal(err)
	}
	return selfStr
}

func (p *Parser) ProcessLines(lines <-chan string, send chan<- event.Event) {
	// start up a goroutine to handle grouped sets of lines
	rawEvents := make(chan rawEvent)
	var wg sync.WaitGroup
	p.wg = wg
	defer p.wg.Wait()
	p.wg.Add(1)
	go p.handleEvents(rawEvents, send)
<commit_msg>Get rid of non fatal logging in mysql<commit_after>// Package mysql parses the mysql slow query log
package mysql

import (
	"database/sql"
	"fmt"
	"log"
	"regexp"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/honeycombio/honeytail/event"

	"github.com/Sirupsen/logrus"
	_ "github.com/go-sql-driver/mysql"
	"github.com/percona/go-mysql/query"
)

// 3 sample log entries
//
// # Time: 2016-04-01T00:31:09.817887Z
// # User@Host: root[root] @ localhost []  Id:   233
// # Query_time: 0.008393  Lock_time: 0.000154 Rows_sent: 1  Rows_examined: 357
// SET timestamp=1459470669;
// show status like 'Uptime';
//
// # Time: 2016-04-01T00:31:09.853523Z
// # User@Host: root[root] @ localhost []  Id:   233
// # Query_time: 0.020424  Lock_time: 0.000147 Rows_sent: 494  Rows_examined: 494
// SET timestamp=1459470669;
// SHOW /*innotop*/ GLOBAL VARIABLES;
//
// # Time: 2016-04-01T00:31:09.856726Z
// # User@Host: root[root] @ localhost []  Id:   233
// # Query_time: 0.000021  Lock_time: 0.000000 Rows_sent: 494  Rows_examined: 494
// SET timestamp=1459470669;
// # administrator command: Ping;
//
// We should ignore the administrator command entry; the stats it presents (eg rows_sent)
// are actually for the previous command

var (
	reTime       = myRegexp{regexp.MustCompile("^# Time: (?P<time>[^ ]+)Z *$")}
	reAdminPing  = myRegexp{regexp.MustCompile("^# administrator command: Ping; *$")}
	reUser       = myRegexp{regexp.MustCompile("^# User@Host: (?P<user>[^ ]+) @ (?P<host>[^ ]+).*$")}
	reQueryStats = myRegexp{regexp.MustCompile("^# Query_time: (?P<queryTime>[0-9.]+) *Lock_time: (?P<lockTime>[0-9.]+) *Rows_sent: (?P<rowsSent>[0-9]+) *Rows_examined: (?P<rowsExamined>[0-9]+) *$")}
	reSetTime    = myRegexp{regexp.MustCompile("^SET timestamp=(?P<unixTime>[0-9]+);$")}
	reQuery      = myRegexp{regexp.MustCompile("^(?P<query>[^#]*).*$")}
	reUse        = myRegexp{regexp.MustCompile("^(?i)use ")}

	rdsStr  = "rds"
	ec2Str  = "ec2"
	selfStr = "self"
)

const timeFormat = "2006-01-02T15:04:05.000000"

type Options struct {
	Host          string `long:"host" description:"MySQL host in the format (address:port)"`
	User          string `long:"user" description:"MySQL username"`
	Pass          string `long:"pass" description:"MySQL password"`
	QueryInterval uint   `long:"interval" description:"interval for querying the MySQL DB in seconds" default:"30"`
}

type Parser struct {
	conf     Options
	wg       sync.WaitGroup
	nower    Nower
	hostedOn string
	readOnly bool
}

type Nower interface {
	Now() time.Time
}

type RealNower struct{}

func (n *RealNower) Now() time.Time {
	return time.Now().UTC()
}

// rawEvent is the unparsed multi-line entry representing a single slow query
type rawEvent struct {
	lines []string
}

// slowQuery represents the structured form of a query from the slow query log
type SlowQuery struct {
	Timestamp       time.Time `json:"time"`
	UnixTime        int       `json:"unixtime"`
	User            string    `json:"user"`
	Host            string    `json:"host"`
	QueryTime       float64   `json:"query_time"`
	LockTime        float64   `json:"lock_time"`
	RowsSent        int       `json:"rows_sent"`
	RowsExamined    int       `json:"rows_examined"`
	Query           string    `json:"query,omitempty"`
	NormalizedQuery string    `json:"normalized_query,omitempty"`
	DB              string    `json:"db,omitempty"`
	HostedOn        string    `json:"hosted_on,omitempty"`
	ReadOnly        bool      `json:"read_only,omitempty"`
	skipQuery       bool
}

func (p *Parser) Init(options interface{}) error {
	p.conf = *options.(*Options)
	p.nower = &RealNower{}
	if p.conf.Host != "" {
		url := fmt.Sprintf("%s:%s@tcp(%s)/", p.conf.User, p.conf.Pass, p.conf.Host)
		db, err := sql.Open("mysql", url)
		if err != nil {
			log.Fatal(err)
		}
		// update hostedOn and readOnly every <n> seconds
		go func() {
			defer db.Close()
			ticker := time.NewTicker(time.Second * time.Duration(p.conf.QueryInterval))
			for _ = range ticker.C {
				p.hostedOn = getHostedOn(db)
				p.readOnly = getReadOnly(db)
			}
		}()
	}
	return nil
}

func getReadOnly(db *sql.DB) bool {
	rows, err := db.Query("SELECT @@global.read_only;")
	if err != nil {
		log.Println(err)
		return false
	}
	defer rows.Close()

	var value bool

	rows.Next()
	if err := rows.Scan(&value); err != nil {
		log.Println(err)
		return false
	}
	if err := rows.Err(); err != nil {
		log.Println(err)
		return false
	}
	return value
}

func getHostedOn(db *sql.DB) string {
	rows, err := db.Query("SHOW GLOBAL VARIABLES WHERE Variable_name = 'basedir';")
	if err != nil {
		log.Println(err)
		return ""
	}
	defer rows.Close()

	var varName, value string

	for rows.Next() {
		if err := rows.Scan(&varName, &value); err != nil {
			log.Println(err)
			return ""
		}
		if strings.HasPrefix(value, "/rdsdbbin/") {
			return rdsStr
		}
	}

	// TODO: implement ec2 detection

	if err := rows.Err(); err != nil {
		log.Println(err)
		return ""
	}
	return selfStr
}

func (p *Parser) ProcessLines(lines <-chan string, send chan<- event.Event) {
	// start up a goroutine to handle grouped sets of lines
	rawEvents := make(chan rawEvent)
	var wg sync.WaitGroup
	p.wg = wg
	defer p.wg.Wait()
	p.wg.Add(1)
	go p.handleEvents(rawEvents, send)

{'fraction_non_alphanumeric': 0.1556845670053217, 'fraction_numerical': 0.038316400580551524, 'mean_word_length': 3.403919897741798, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 23, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6850171', 'n_tokens_mistral': 4235, 'n_tokens_neox': 3795, 'n_words': 910}