func (f *Factory) Make(service string, instanceId uint, data []byte) (sysconfig.Monitor, error) {
	var monitor sysconfig.Monitor
	switch service {
	case "mysql":
		// Load the MySQL instance info (DSN, name, etc.).
		mysqlIt := &proto.MySQLInstance{}
		if err := f.ir.Get(service, instanceId, mysqlIt); err != nil {
			return nil, err
		}

		// Parse the MySQL sysconfig config.
		config := &mysql.Config{}
		if err := json.Unmarshal(data, config); err != nil {
			return nil, err
		}

		// The user-friendly name of the service, e.g. sysconfig-mysql-db101:
		alias := "sysconfig-mysql-" + mysqlIt.Hostname

		// Make a MySQL sysconfig monitor.
		monitor = mysql.NewMonitor(
			alias,
			config,
			pct.NewLogger(f.logChan, alias),
			mysqlConn.NewConnection(mysqlIt.DSN),
		)
	default:
		return nil, errors.New("Unknown sysconfig monitor type: " + service)
	}
	return monitor, nil
}

// Even having a wrong user/pass, the monitor should be able to start
// It won't be able to connect to the DB, but it should start without
// blocking.
// This test if for cases where the agent starts but MySQL is down
func (s *TestSuite) TestStartWithInvalidDSN(t *C) {
	config := &mysql.Config{
		Config: mm.Config{
			ServiceInstance: proto.ServiceInstance{
				Service:    "mysql",
				InstanceId: 1,
			},
			Collect: 1,
			Report:  60,
		},
		Status: map[string]string{
			"threads_connected": "gauge",
			"threads_running":   "gauge",
		},
	}

	failDsn := "user:[email protected](127.0.0.2:3309)/"
	m := mysql.NewMonitor(s.name, config, s.logger, mysqlConn.NewConnection(failDsn), s.mrm)
	if m == nil {
		t.Fatal("Make new mysql.Monitor")
	}

	err := m.Start(s.tickChan, s.collectionChan)
	t.Assert(err, IsNil)
}

func (s *MysqlTestSuite) TestConnection(t *C) {
	conn := mysql.NewConnection(s.dsn)
	err := conn.Connect(1)
	t.Assert(err, IsNil)
	conn1 := conn.DB()

	err = conn.Connect(1)
	t.Assert(err, IsNil)
	conn2 := conn.DB()
	t.Check(conn1, Equals, conn2)

	conn.Close()
	t.Assert(conn.DB(), NotNil)

	/**
	 * we still have open connection,
	 * because we used Connect twice,
	 * so let's close it
	 */
	conn.Close()
	t.Assert(conn.DB(), IsNil)

	// lets test accidental extra closing
	conn.Close()
	t.Assert(conn.DB(), IsNil)
}

func NewSlowMySQL(dsn string) *SlowMySQL {
	n := &SlowMySQL{
		realConnection: mysql.NewConnection(dsn),
		globalDelay:    0,
	}
	return n
}

func (s *MysqlTestSuite) TestMissingSocketError(t *C) {
	// https://jira.percona.com/browse/PCT-791
	conn := mysql.NewConnection("percona:[email protected](/foo/bar/my.sock)/")
	err := conn.Connect(1)
	t.Assert(
		fmt.Sprintf("%s", err),
		Equals,
		"Cannot connect to MySQL percona:<password-hidden>@unix(/foo/bar/my.sock)/: no such file or directory: /foo/bar/my.sock",
	)
}

func (i *Installer) createNewMySQLUser() (dsn mysql.DSN, err error) {
	// Auto-detect the root MySQL user connection options.
	superUserDSN := i.defaultDSN
	if i.flags.Bool["auto-detect-mysql"] {
		if err := i.autodetectDSN(&superUserDSN); err != nil {
			if i.flags.Bool["debug"] {
				log.Println(err)
			}
		}
	}
	fmt.Printf("MySQL root DSN: %s\n", superUserDSN)

	// Try to connect as root automatically.  If this fails and interactive is true,
	// start prompting user to enter valid root MySQL connection info.
	if err = i.verifyMySQLConnection(superUserDSN); err != nil {
		fmt.Printf("Error connecting to MySQL %s: %s\n", superUserDSN, err)
		if i.flags.Bool["interactive"] {
			if again, err := i.term.PromptBool("Try again?", "Y"); err != nil {
				return superUserDSN, err
			} else if !again {
				return superUserDSN, fmt.Errorf("Failed to connect to MySQL")
			}
			fmt.Println("Specify a root/super MySQL user to create a user for the agent")
			if err := i.getDSNFromUser(&superUserDSN); err != nil {
				return dsn, err
			}
		} else {
			// Can't auto-detect MySQL root user and not interactive, fail.
			return dsn, err
		}
	}

	// Check MySQL Version
	dsnString, err := superUserDSN.DSN()
	if err != nil {
		return dsn, err
	}
	tmpConn := mysql.NewConnection(dsnString)
	isVersionSupported, err := i.IsVersionSupported(tmpConn)
	if err != nil {
		return dsn, err
	}

	if !isVersionSupported {
		return dsn, fmt.Errorf("MySQL version not supported. It should be > %s", agent.MIN_SUPPORTED_MYSQL_VERSION)
	}

	dsn, err = i.createMySQLUser(superUserDSN)
	if err != nil {
		return dsn, err
	}

	return dsn, nil
}

func (s *TestSuite) SetUpSuite(t *C) {
	s.dsn = os.Getenv("PCT_TEST_MYSQL_DSN")
	if s.dsn == "" {
		t.Fatal("PCT_TEST_MYSQL_DSN is not set")
	}

	s.conn = mysql.NewConnection(s.dsn)
	if err := s.conn.Connect(1); err != nil {
		t.Fatal(err)
	}
}

func (s *ManagerTestSuite) SetUpSuite(t *C) {
	s.dsn = os.Getenv("PCT_TEST_MYSQL_DSN")
	if s.dsn == "" {
		t.Fatal("PCT_TEST_MYSQL_DSN is not set")
	}
	s.realmysql = mysql.NewConnection(s.dsn)
	if err := s.realmysql.Connect(1); err != nil {
		t.Fatal(err)
	}
	s.reset = []mysql.Query{
		mysql.Query{Set: "SET GLOBAL slow_query_log=OFF"},
		mysql.Query{Set: "SET GLOBAL long_query_time=10"},
	}

	s.nullmysql = mock.NewNullMySQL()

	s.logChan = make(chan *proto.LogEntry, 1000)
	s.logger = pct.NewLogger(s.logChan, "qan-test")

	s.intervalChan = make(chan *qan.Interval, 1)
	s.iter = mock.NewMockIntervalIter(s.intervalChan)
	s.iterFactory = &mock.IntervalIterFactory{
		Iters:     []qan.IntervalIter{s.iter},
		TickChans: make(map[qan.IntervalIter]chan time.Time),
	}

	s.dataChan = make(chan interface{}, 2)
	s.spool = mock.NewSpooler(s.dataChan)
	s.workerFactory = &qan.SlowLogWorkerFactory{}

	var err error
	s.tmpDir, err = ioutil.TempDir("/tmp", "agent-test")
	t.Assert(err, IsNil)

	if err := pct.Basedir.Init(s.tmpDir); err != nil {
		t.Fatal(err)
	}
	s.configDir = pct.Basedir.Dir("config")

	s.im = instance.NewRepo(pct.NewLogger(s.logChan, "im-test"), s.configDir, s.api)
	data, err := json.Marshal(&proto.MySQLInstance{
		Hostname: "db1",
		DSN:      s.dsn,
	})
	t.Assert(err, IsNil)
	s.im.Add("mysql", 1, data, false)
	s.mysqlInstance = proto.ServiceInstance{Service: "mysql", InstanceId: 1}

	links := map[string]string{
		"agent":     "http://localhost/agent",
		"instances": "http://localhost/instances",
	}
	s.api = mock.NewAPI("http://localhost", "http://localhost", "123", "abc-123-def", links)
}

func (i *Installer) useExistingMySQLUser() (mysql.DSN, error) {
	userDSN := i.defaultDSN
	userDSN.Username = "percona-agent"
	userDSN.Password = ""
	if i.flags.Bool["auto-detect-mysql"] {
		if err := i.autodetectDSN(&userDSN); err != nil {
			if i.flags.Bool["debug"] {
				log.Println(err)
			}
		}
	}
	for {
		// Let user specify the MySQL account to use for the agent.
		fmt.Println("Specify the existing MySQL user to use for the agent")
		if err := i.getDSNFromUser(&userDSN); err != nil {
			return userDSN, nil
		}

		// Verify DSN provided by user
		if err := i.verifyMySQLConnection(userDSN); err != nil {
			fmt.Printf("Error connecting to MySQL %s: %s\n", userDSN, err)
			if i.flags.Bool["interactive"] {
				if again, err := i.term.PromptBool("Try again?", "Y"); err != nil {
					return userDSN, err
				} else if !again {
					return userDSN, fmt.Errorf("Failed to connect to MySQL")
				}
				continue // again
			} else {
				// Can't auto-detect MySQL root user and not interactive, fail.
				return userDSN, err
			}
		}
		break
	}

	// Check MySQL Version
	dsnString, err := userDSN.DSN()
	if err != nil {
		return userDSN, err
	}
	tmpConn := mysql.NewConnection(dsnString)
	isVersionSupported, err := i.IsVersionSupported(tmpConn)
	if err != nil {
		return userDSN, err
	}
	if !isVersionSupported {
		return userDSN, fmt.Errorf("MySQL version not supported. It should be > %s", agent.MIN_SUPPORTED_MYSQL_VERSION)
	}
	return userDSN, nil // success
}

func TestMySQLConnection(dsn mysql.DSN) error {
	dsnString, err := dsn.DSN()
	if err != nil {
		return err
	}

	fmt.Printf("Testing MySQL connection %s...\n", dsn)
	conn := mysql.NewConnection(dsnString)
	if err := conn.Connect(1); err != nil {
		return err
	}
	defer conn.Close()
	return nil
}

func (i *Installer) verifyMySQLConnection(dsn mysql.DSN) (err error) {
	dsnString, err := dsn.DSN()
	if err != nil {
		return err
	}
	if i.flags.Bool["debug"] {
		log.Printf("verifyMySQLConnection: %#v %s\n", dsn, dsnString)
	}
	conn := mysql.NewConnection(dsnString)
	if err := conn.Connect(1); err != nil {
		return err
	}
	conn.Close()
	return nil
}

func (f *Factory) Make(service string, instanceId uint, data []byte) (mm.Monitor, error) {
	var monitor mm.Monitor
	switch service {
	case "mysql":
		// Load the MySQL instance info (DSN, name, etc.).
		mysqlIt := &proto.MySQLInstance{}
		if err := f.ir.Get(service, instanceId, mysqlIt); err != nil {
			return nil, err
		}

		// Parse the MySQL sysconfig config.
		config := &mysql.Config{}
		if err := json.Unmarshal(data, config); err != nil {
			return nil, err
		}

		// The user-friendly name of the service, e.g. sysconfig-mysql-db101:
		alias := "mm-mysql-" + mysqlIt.Hostname

		// Make a MySQL metrics monitor.
		monitor = mysql.NewMonitor(
			alias,
			config,
			pct.NewLogger(f.logChan, alias),
			mysqlConn.NewConnection(mysqlIt.DSN),
			f.mrm,
		)
	case "server":
		// Parse the system mm config.
		config := &system.Config{}
		if err := json.Unmarshal(data, config); err != nil {
			return nil, err
		}

		// Only one system for now, so no SystemInstance and no  "-instanceName" suffix.
		alias := "mm-system"

		// Make a MySQL metrics monitor.
		monitor = system.NewMonitor(
			alias,
			config,
			pct.NewLogger(f.logChan, alias),
		)
	default:
		return nil, errors.New("Unknown metrics monitor type: " + service)
	}
	return monitor, nil
}

func (i *Installer) createMySQLUser(dsn mysql.DSN) (mysql.DSN, error) {
	// Same host:port or socket, but different user and pass.
	userDSN := dsn
	userDSN.Username = "percona-agent"
	userDSN.Password = fmt.Sprintf("%p%d", &dsn, rand.Uint32())
	userDSN.OldPasswords = i.flags.Bool["old-passwords"]

	dsnString, _ := dsn.DSN()
	conn := mysql.NewConnection(dsnString)
	if err := conn.Connect(1); err != nil {
		return userDSN, err
	}
	defer conn.Close()
	grants := MakeGrant(dsn, userDSN.Username, userDSN.Password, i.flags.Int64["mysql-max-user-connections"])
	for _, grant := range grants {
		if i.flags.Bool["debug"] {
			log.Println(grant)
		}
		_, err := conn.DB().Exec(grant)
		if err != nil {
			return userDSN, fmt.Errorf("Error executing %s: %s", grant, err)
		}
	}

	// Go MySQL driver resolves localhost to 127.0.0.1 but localhost is a special
	// value for MySQL, so 127.0.0.1 may not work with a grant @localhost, so we
	// add a 2nd grant @127.0.0.1 to be sure.
	if dsn.Hostname == "localhost" {
		dsn2 := dsn
		dsn2.Hostname = "127.0.0.1"
		grants := MakeGrant(dsn2, userDSN.Username, userDSN.Password, i.flags.Int64["mysql-max-user-connections"])
		for _, grant := range grants {
			if i.flags.Bool["debug"] {
				log.Println(grant)
			}
			_, err := conn.DB().Exec(grant)
			if err != nil {
				return userDSN, fmt.Errorf("Error executing %s: %s", grant, err)
			}
		}
	}

	return userDSN, nil
}

func GetMySQLInfo(it *proto.MySQLInstance) error {
	conn := mysql.NewConnection(it.DSN)
	if err := conn.Connect(1); err != nil {
		return err
	}
	defer conn.Close()
	sql := "SELECT /* percona-agent */" +
		" CONCAT_WS('.', @@hostname, IF(@@port='3306',NULL,@@port)) AS Hostname," +
		" @@version_comment AS Distro," +
		" @@version AS Version"
	err := conn.DB().QueryRow(sql).Scan(
		&it.Hostname,
		&it.Distro,
		&it.Version,
	)
	if err != nil {
		return err
	}
	return nil
}

func (s *WorkerTestSuite) TestWorkerWithAnotherTZ(t *C) {
	if s.dsn == "" {
		t.Fatal("PCT_TEST_MYSQL_DSN is not set")
	}
	mysqlConn := mysql.NewConnection(s.dsn)
	err := mysqlConn.Connect(1)
	t.Assert(err, IsNil)
	defer mysqlConn.Close()
	mysqlConn.DB().Exec(`SET @@global.time_zone="-1:00"`)
	mysqlConn.DB().Exec(`SET time_zone="-1:00"`)
	// Reset the timezone or other tests could fail
	mysqlConn.DB().Exec(`SET @@global.time_zone="+00:00"`)
	defer mysqlConn.DB().Exec(`SET time_zone="+00:00"`)

	i := &qan.Interval{
		Number:      1,
		StartTime:   s.now,
		StopTime:    s.now.Add(1 * time.Minute),
		Filename:    inputDir + "slow001.log",
		StartOffset: 0,
		EndOffset:   524,
	}
	got, err := s.RunWorker(s.config, mysqlConn, i)
	t.Check(err, IsNil)
	expect := &qan.Result{}
	test.LoadMmReport(outputDir+"slow001.json", expect)

	// I'm going to use the same results for TestWorkerSlow001
	// But simulating tz = UTC -1

	expect.Class[0].Example.Ts = "2007-10-15 22:45:10"
	expect.Class[1].Example.Ts = "2007-10-15 22:43:52"

	sort.Sort(ByQueryId(got.Class))
	sort.Sort(ByQueryId(expect.Class))
	if ok, diff := IsDeeply(got, expect); !ok {
		Dump(got)
		t.Error(diff)
	}
}

func (s *TestSuite) TestCollectUserstats(t *C) {
	/**
	 * Disable and reset user stats.
	 */
	if _, err := s.db.Exec("set global userstat = off"); err != nil {
		t.Fatal(err)
	}
	if _, err := s.db.Exec("flush user_statistics"); err != nil {
		t.Fatal(err)
	}
	if _, err := s.db.Exec("flush index_statistics"); err != nil {
		t.Fatal(err)
	}

	config := &mysql.Config{
		Config: mm.Config{
			ServiceInstance: proto.ServiceInstance{
				Service:    "mysql",
				InstanceId: 1,
			},
			Collect: 1,
			Report:  60,
		},
		UserStats: true,
	}

	m := mysql.NewMonitor(s.name, config, s.logger, mysqlConn.NewConnection(dsn), s.mrm)
	if m == nil {
		t.Fatal("Make new mysql.Monitor")
	}

	err := m.Start(s.tickChan, s.collectionChan)
	if err != nil {
		t.Fatalf("Start monitor without error, got %s", err)
	}

	if ok := test.WaitStatus(5, m, s.name+"-mysql", "Connected"); !ok {
		t.Fatal("Monitor is ready")
	}

	var user, host string
	err = s.db.QueryRow("SELECT SUBSTRING_INDEX(CURRENT_USER(),'@',1) AS 'user', SUBSTRING_INDEX(CURRENT_USER(),'@',-1) AS host").Scan(&user, &host)
	if err != nil {
		t.Fatal(err)
	}

	// To get index stats, we need to use an index: mysq.user PK <host, user>
	rows, err := s.db.Query("select * from mysql.user where host=? and user=?", host, user)
	if err != nil {
		t.Fatal(err)
	}
	defer rows.Close()

	s.tickChan <- time.Now()
	got := test.WaitCollection(s.collectionChan, 1)
	if len(got) == 0 {
		t.Fatal("Got a collection after tick")
	}
	c := got[0]

	/**
	 * Monitor should have collected the user stats: just table and index.
	 * Values vary a little, but there should be a table metric for mysql.user
	 * because login uses this table.
	 */
	if len(c.Metrics) < 1 {
		t.Fatalf("Collect at least 1 user stat metric; got %+v", c.Metrics)
	}

	var tblStat mm.Metric
	var idxStat mm.Metric
	for _, m := range c.Metrics {
		switch m.Name {
		case "mysql/db.mysql/t.user/rows_read":
			tblStat = m
		case "mysql/db.mysql/t.user/idx.PRIMARY/rows_read":
			idxStat = m
		}
	}

	// At least 2 rows should have been read from mysql.user:
	//   1: our db connection
	//   2: the monitor's db connection
	if tblStat.Number < 2 {
		t.Errorf("mysql/db.mysql/t.user/rows_read >= 2, got %+v", tblStat)
	}

	// At least 1 index read on mysql.user PK due to our SELECT ^.
	if idxStat.Number < 1 {
		t.Errorf("mysql/db.mysql/t.user/idx.PRIMARY/rows_read >= 1, got %+v", idxStat)
	}

	// Stop montior, clean up.
	m.Stop()
}

func (s *TestSuite) TestStartCollectStop(t *C) {
	// Create the monitor.
	config := &mysql.Config{
		Config: sysconfig.Config{
			ServiceInstance: proto.ServiceInstance{
				Service:    "mysql",
				InstanceId: 1,
			},
		},
	}
	m := mysql.NewMonitor(s.name, config, s.logger, mysqlConn.NewConnection(dsn))
	if m == nil {
		t.Fatal("Make new mysql.Monitor")
	}

	// Start the monitor.
	err := m.Start(s.tickChan, s.reportChan)
	if err != nil {
		t.Fatalf("Start monitor without error, got %s", err)
	}

	// monitor=Ready once it has successfully connected to MySQL.  This may
	// take a few seconds (hopefully < 5) on a slow test machine.
	if ok := test.WaitStatusPrefix(5, m, s.name, "Idle"); !ok {
		t.Fatal("Monitor is ready")
	}

	// Send tick to make the monitor collect.
	now := time.Now().UTC()
	s.tickChan <- now
	got := test.WaitSystemConfig(s.reportChan, 1)
	if len(got) == 0 {
		t.Fatal("Got a sysconfig after tick")
	}
	c := got[0]

	if c.Ts != now.Unix() {
		t.Error("Report.Ts set to %s; got %s", now.Unix(), c.Ts)
	}

	if len(c.Settings) < 100 {
		t.Fatal("Collect > 100 vars; got %+v", c.Settings)
	}

	haveWaitTimeout := false
	val := ""
	for _, s := range c.Settings {
		if s[0] == "wait_timeout" {
			haveWaitTimeout = true
			val = s[1]
		}
	}
	if !haveWaitTimeout {
		t.Logf("%+v\n", c)
		t.Error("Got wait_timeout")
	}
	if val == "" {
		t.Error("wait_timeout has value")
	}

	/**
	 * Stop the monitor.
	 */

	m.Stop()

	if ok := test.WaitStatus(5, m, s.name, "Stopped"); !ok {
		t.Fatal("Monitor has stopped")
	}
}

func (s *ManagerTestSuite) TestHandleGetInfoMySQL(t *C) {
	if dsn == "" {
		t.Fatal("PCT_TEST_MYSQL_DSN is not set")
	}

	/**
	 * First get MySQL info manually.  This is what GetInfo should do, too.
	 */

	conn := mysql.NewConnection(dsn)
	if err := conn.Connect(1); err != nil {
		t.Fatal(err)
	}
	var hostname, distro, version string
	sql := "SELECT" +
		" CONCAT_WS('.', @@hostname, IF(@@port='3306',NULL,@@port)) AS Hostname," +
		" @@version_comment AS Distro," +
		" @@version AS Version"
	if err := conn.DB().QueryRow(sql).Scan(&hostname, &distro, &version); err != nil {
		t.Fatal(err)
	}

	/**
	 * Now use the instance manager and GetInfo to get MySQL info like API would.
	 */

	// Create an instance manager.
	m := instance.NewManager(s.logger, s.configDir, s.api)
	t.Assert(m, NotNil)

	// API sends Cmd[Service:"instance", Cmd:"GetInfo",
	//               Data:proto.ServiceInstance[Service:"mysql",
	//                                          Data:proto.MySQLInstance[]]]
	// Only DSN is needed.  We set Id just to test that it's not changed.
	mysqlIt := &proto.MySQLInstance{
		Id:  9,
		DSN: dsn,
	}
	mysqlData, err := json.Marshal(mysqlIt)
	t.Assert(err, IsNil)

	serviceIt := &proto.ServiceInstance{
		Service:  "mysql",
		Instance: mysqlData,
	}
	serviceData, err := json.Marshal(serviceIt)
	t.Assert(err, IsNil)

	cmd := &proto.Cmd{
		Cmd:     "GetInfo",
		Service: "instance",
		Data:    serviceData,
	}

	reply := m.Handle(cmd)

	got := &proto.MySQLInstance{}
	err = json.Unmarshal(reply.Data, got)
	t.Assert(err, IsNil)

	t.Check(got.Id, Equals, uint(9))        // not changed
	t.Check(got.DSN, Equals, mysqlIt.DSN)   // not changed
	t.Check(got.Hostname, Equals, hostname) // new
	t.Check(got.Distro, Equals, distro)     // new
	t.Check(got.Version, Equals, version)   // new
}

func (s *AnalyzerTestSuite) TestRealSlowLogWorker(t *C) {
	dsn := os.Getenv("PCT_TEST_MYSQL_DSN")
	if dsn == "" {
		t.Fatal("PCT_TEST_MYSQL_DSN is not set")
	}
	realmysql := mysql.NewConnection(dsn)
	if err := realmysql.Connect(1); err != nil {
		t.Fatal(err)
	}
	// Don't release all resources immediately because the worker needs the connection
	defer realmysql.Close()
	defer test.DrainRecvData(s.dataChan)

	config := s.config
	config.Start = []mysql.Query{
		mysql.Query{Set: "SET GLOBAL slow_query_log=OFF"},
		mysql.Query{Set: "SET GLOBAL long_query_time=0"},
		mysql.Query{Set: "SET GLOBAL slow_query_log=ON"},
	}
	config.Stop = []mysql.Query{
		mysql.Query{Set: "SET GLOBAL slow_query_log=OFF"},
		mysql.Query{Set: "SET GLOBAL long_query_time=10"},
	}

	worker := slowlog.NewWorker(pct.NewLogger(s.logChan, "qan-worker"), config, realmysql)
	//intervalChan := make(chan *qan.Interval, 1)
	//iter := mock.NewIter(intervalChan)

	a := qan.NewRealAnalyzer(
		pct.NewLogger(s.logChan, "qan-analyzer"),
		config,
		s.iter,
		realmysql,
		s.restartChan,
		worker,
		s.clock,
		s.spool,
	)
	err := a.Start()
	t.Assert(err, IsNil)
	if !test.WaitStatus(3, a, "qan-analyzer", "Idle") {
		t.Fatal("Timeout waiting for qan-analyzer=Idle")
	}

	now := time.Now().UTC()
	i := &qan.Interval{
		Number:      1,
		StartTime:   now,
		StopTime:    now.Add(1 * time.Minute),
		Filename:    inputDir + "slow001.log",
		StartOffset: 0,
		EndOffset:   524,
	}
	s.intervalChan <- i
	data := test.WaitData(s.dataChan)
	t.Assert(data, HasLen, 1)
	res := data[0].(*qan.Report)
	t.Check(res.Global.TotalQueries, Equals, uint64(2))

	err = a.Stop()
	t.Assert(err, IsNil)
}

func (s *WorkerTestSuite) TestRealWorker(t *C) {
	if s.dsn == "" {
		t.Fatal("PCT_TEST_MYSQL_DSN is not set")
	}
	mysqlConn := mysql.NewConnection(s.dsn)
	err := mysqlConn.Connect(1)
	t.Assert(err, IsNil)
	defer mysqlConn.Close()

	f := perfschema.NewRealWorkerFactory(s.logChan)
	w := f.Make("qan-worker", mysqlConn)

	start := []mysql.Query{
		mysql.Query{Verify: "performance_schema", Expect: "1"},
		mysql.Query{Set: "UPDATE performance_schema.setup_consumers SET ENABLED = 'YES' WHERE NAME = 'statements_digest'"},
		mysql.Query{Set: "UPDATE performance_schema.setup_instruments SET ENABLED = 'YES', TIMED = 'YES' WHERE NAME LIKE 'statement/sql/%'"},
		mysql.Query{Set: "TRUNCATE performance_schema.events_statements_summary_by_digest"},
	}
	if err := mysqlConn.Set(start); err != nil {
		t.Fatal(err)
	}
	stop := []mysql.Query{
		mysql.Query{Set: "UPDATE performance_schema.setup_consumers SET ENABLED = 'NO' WHERE NAME = 'statements_digest'"},
		mysql.Query{Set: "UPDATE performance_schema.setup_instruments SET ENABLED = 'NO', TIMED = 'NO' WHERE NAME LIKE 'statement/sql/%'"},
	}
	defer func() {
		if err := mysqlConn.Set(stop); err != nil {
			t.Fatal(err)
		}
	}()

	// SCHEMA_NAME: NULL
	//      DIGEST: fbe070dfb47e4a2401c5be6b5201254e
	// DIGEST_TEXT: SELECT ? FROM DUAL
	_, err = mysqlConn.DB().Exec("SELECT 'teapot' FROM DUAL")

	// First interval.
	err = w.Setup(&qan.Interval{Number: 1, StartTime: time.Now().UTC()})
	t.Assert(err, IsNil)

	res, err := w.Run()
	t.Assert(err, IsNil)
	t.Check(res, IsNil)

	err = w.Cleanup()
	t.Assert(err, IsNil)

	// Some query activity between intervals.
	_, err = mysqlConn.DB().Exec("SELECT 'teapot' FROM DUAL")
	time.Sleep(1 * time.Second)

	// Second interval and a result.
	err = w.Setup(&qan.Interval{Number: 2, StartTime: time.Now().UTC()})
	t.Assert(err, IsNil)

	res, err = w.Run()
	t.Assert(res, NotNil)
	if len(res.Class) == 0 {
		t.Fatal("Expected len(res.Class) > 0")
	}
	var class *event.QueryClass
	for _, c := range res.Class {
		if c.Fingerprint == "SELECT ? FROM DUAL " {
			class = c
			break
		}
	}
	t.Assert(class, NotNil)
	// Digests on different versions or distros of MySQL don't match
	//t.Check(class.Id, Equals, "01C5BE6B5201254E")
	//t.Check(class.Fingerprint, Equals, "SELECT ? FROM DUAL ")
	queryTime := class.Metrics.TimeMetrics["Query_time"]
	if queryTime.Min == 0 {
		t.Error("Expected Query_time_min > 0")
	}
	if queryTime.Max == 0 {
		t.Error("Expected Query_time_max > 0")
	}
	if queryTime.Avg == 0 {
		t.Error("Expected Query_time_avg > 0")
	}
	if queryTime.Min > queryTime.Max {
		t.Error("Expected Query_time_min >= Query_time_max")
	}
	t.Check(class.Metrics.NumberMetrics["Rows_affected"].Sum, Equals, uint64(0))
	t.Check(class.Metrics.NumberMetrics["Rows_examined"].Sum, Equals, uint64(0))
	t.Check(class.Metrics.NumberMetrics["Rows_sent"].Sum, Equals, uint64(1))

	err = w.Cleanup()
	t.Assert(err, IsNil)
}