Add store support for ProxySQL (#129)

* Add store support for ProxySQL

go.mod

@@ -3,10 +3,11 @@ module github.com/github/freno
 go 1.14
 
 require (
+	github.com/DATA-DOG/go-sqlmock v1.4.1
 	github.com/armon/go-metrics v0.0.0-20190430140413-ec5e00d3c878
 	github.com/boltdb/bolt v1.3.1
 	github.com/bradfitz/gomemcache v0.0.0-20190913173617-a41fca850d0b
-	github.com/go-sql-driver/mysql v1.5.0 // indirect
+	github.com/go-sql-driver/mysql v1.5.0
 	github.com/hashicorp/go-msgpack v0.5.5
 	github.com/julienschmidt/httprouter v1.3.0
 	github.com/outbrain/golib v0.0.0-20180830062331-ab954725f502

go.sum

@@ -1,3 +1,5 @@
+github.com/DATA-DOG/go-sqlmock v1.4.1 h1:ThlnYciV1iM/V0OSF/dtkqWb6xo5qITT1TJBG1MRDJM=
+github.com/DATA-DOG/go-sqlmock v1.4.1/go.mod h1:f/Ixk793poVmq4qj/V1dPUg2JEAKC73Q5eFN3EC/SaM=
 github.com/DataDog/datadog-go v2.2.0+incompatible/go.mod h1:LButxg5PwREeZtORoXG3tL4fMGNddJ+vMq1mwgfaqoQ=
 github.com/armon/go-metrics v0.0.0-20190430140413-ec5e00d3c878 h1:EFSB7Zo9Eg91v7MJPVsifUysc/wPdN+NOnVe6bWbdBM=
 github.com/armon/go-metrics v0.0.0-20190430140413-ec5e00d3c878/go.mod h1:3AMJUQhVx52RsWOnlkpikZr01T/yAVN2gn0861vByNg=

pkg/config/haproxy_config_test.go

@@ -16,7 +16,7 @@ func init() {
 	log.SetLevel(log.ERROR)
 }
 
-func TestParseAddresses(t *testing.T) {
+func TestHAProxyParseAddresses(t *testing.T) {
 	{
 		c := &HAProxyConfigurationSettings{Addresses: ""}
 		addresses, err := c.parseAddresses()
@@ -90,7 +90,7 @@ func TestParseAddresses(t *testing.T) {
 	}
 }
 
-func TestGetProxyAddresses(t *testing.T) {
+func TestHAProxyGetProxyAddresses(t *testing.T) {
 	{
 		c := &HAProxyConfigurationSettings{Addresses: ""}
 		addresses, err := c.GetProxyAddresses()
@@ -127,7 +127,7 @@ func TestGetProxyAddresses(t *testing.T) {
 	}
 }
 
-func TestIsEmpty(t *testing.T) {
+func TestHAProxyIsEmpty(t *testing.T) {
 	{
 		c := &HAProxyConfigurationSettings{}
 		isEmpty := c.IsEmpty()

pkg/config/mysql_config.go

@@ -23,8 +23,9 @@ type MySQLClusterConfigurationSettings struct {
 	HttpCheckPath        string   // Specify if different than specified by MySQLConfigurationSettings
 	IgnoreHosts          []string // override MySQLConfigurationSettings's, or leave empty to inherit those settings
 
-	HAProxySettings     HAProxyConfigurationSettings // If list of servers is to be acquired via HAProxy, provide this field
-	VitessSettings      VitessConfigurationSettings  // If list of servers is to be acquired via Vitess, provide this field
+	HAProxySettings     HAProxyConfigurationSettings  // If list of servers is to be acquired via HAProxy, provide this field
+	ProxySQLSettings    ProxySQLConfigurationSettings // If list of servers is to be acquired via ProxySQL, provide this field
+	VitessSettings      VitessConfigurationSettings   // If list of servers is to be acquired via Vitess, provide this field
 	StaticHostsSettings StaticHostsConfigurationSettings
 }
 
@@ -58,6 +59,9 @@ type MySQLConfigurationSettings struct {
 	HttpCheckPort        int      // port for HTTP check. -1 to disable.
 	HttpCheckPath        string   // If non-empty, requires HttpCheckPort
 	IgnoreHosts          []string // If non empty, substrings to indicate hosts to be ignored/skipped
+	ProxySQLAddresses    []string // A list of ProxySQL instances to query for hosts
+	ProxySQLUser         string   // ProxySQL stats username
+	ProxySQLPassword     string   // ProxySQL stats password
 	VitessCells          []string // Name of the Vitess cells for polling tablet hosts
 
 	Clusters map[string](*MySQLClusterConfigurationSettings) // cluster name -> cluster config
@@ -115,6 +119,17 @@ func (settings *MySQLConfigurationSettings) postReadAdjustments() error {
 		if len(clusterSettings.IgnoreHosts) == 0 {
 			clusterSettings.IgnoreHosts = settings.IgnoreHosts
 		}
+		if !clusterSettings.ProxySQLSettings.IsEmpty() {
+			if len(clusterSettings.ProxySQLSettings.Addresses) < 1 {
+				clusterSettings.ProxySQLSettings.Addresses = settings.ProxySQLAddresses
+			}
+			if clusterSettings.ProxySQLSettings.User == "" {
+				clusterSettings.ProxySQLSettings.User = settings.ProxySQLUser
+			}
+			if clusterSettings.ProxySQLSettings.Password == "" {
+				clusterSettings.ProxySQLSettings.Password = settings.ProxySQLPassword
+			}
+		}
 		if !clusterSettings.VitessSettings.IsEmpty() && len(clusterSettings.VitessSettings.Cells) < 1 {
 			clusterSettings.VitessSettings.Cells = settings.VitessCells
 		}

pkg/config/proxysql_config.go

@@ -0,0 +1,34 @@
+package config
+
+import "fmt"
+
+//
+// ProxySQL-specific configuration
+//
+
+const ProxySQLDefaultDatabase = "stats"
+
+type ProxySQLConfigurationSettings struct {
+	Addresses           []string
+	User                string
+	Password            string
+	HostgroupID         uint
+	IgnoreServerTTLSecs uint
+}
+
+func (settings ProxySQLConfigurationSettings) AddressToDSN(address string) string {
+	return fmt.Sprintf("mysql://%s:*****@%s/%s", settings.User, address, ProxySQLDefaultDatabase)
+}
+
+func (settings *ProxySQLConfigurationSettings) IsEmpty() bool {
+	if len(settings.Addresses) == 0 {
+		return true
+	}
+	if settings.User == "" || settings.Password == "" {
+		return true
+	}
+	if settings.HostgroupID < 1 {
+		return true
+	}
+	return false
+}

pkg/config/proxysql_config_test.go

@@ -0,0 +1,42 @@
+package config
+
+import (
+	"testing"
+
+	test "github.com/outbrain/golib/tests"
+)
+
+func TestProxySQLAddressToDSN(t *testing.T) {
+	{
+		c := &ProxySQLConfigurationSettings{User: "freno"}
+		test.S(t).ExpectEquals(c.AddressToDSN("proxysql-123abcd.test:6032"), "mysql://freno:*****@proxysql-123abcd.test:6032/"+ProxySQLDefaultDatabase)
+	}
+}
+
+func TestProxySQLIsEmpty(t *testing.T) {
+	{
+		c := &ProxySQLConfigurationSettings{}
+		isEmpty := c.IsEmpty()
+		test.S(t).ExpectTrue(isEmpty)
+	}
+	{
+		c := &ProxySQLConfigurationSettings{Addresses: []string{"localhost:6032"}}
+		isEmpty := c.IsEmpty()
+		test.S(t).ExpectTrue(isEmpty)
+	}
+	{
+		c := &ProxySQLConfigurationSettings{Addresses: []string{"localhost:6032"}, User: "freno"}
+		isEmpty := c.IsEmpty()
+		test.S(t).ExpectTrue(isEmpty)
+	}
+	{
+		c := &ProxySQLConfigurationSettings{Addresses: []string{"localhost:6032"}, User: "freno", Password: "freno"}
+		isEmpty := c.IsEmpty()
+		test.S(t).ExpectTrue(isEmpty)
+	}
+	{
+		c := &ProxySQLConfigurationSettings{Addresses: []string{"localhost:6032"}, User: "freno", Password: "freno", HostgroupID: 20}
+		isEmpty := c.IsEmpty()
+		test.S(t).ExpectFalse(isEmpty)
+	}
+}

pkg/proxysql/README.md

@@ -0,0 +1,21 @@
+# ProxySQL
+
+This package implements freno store support for [ProxySQL](https://proxysql.com/)
+
+## Logic
+
+Freno will probe servers found in the `stats.stats_mysql_connection_pool` ProxySQL admin table that have either status:
+1. `ONLINE` - connect, ping and replication checks pass
+1. `SHUNNED_REPLICATION_LAG` - connect and ping checks pass, but replication is lagging
+
+All other statuses are considered unhealthy and therefore are ignored by freno, eg:
+1. `SHUNNED` - proxysql connot connect and/or ping a backend
+1. `OFFLINE_SOFT` - a server that is draining, usually for maintenance, etc
+1. `OFFLINE_HARD` - a server that is completely offline
+
+## Requirements
+1. The ProxySQL `--no-monitor` flag is not set
+1. The [ProxySQL monitor module](https://github.com/sysown/proxysql/wiki/Monitor-Module) is enabled, eg: [`mysql-monitor_enabled`](https://github.com/sysown/proxysql/wiki/Global-variables#mysql-monitor_enabled) is `true`
+1. The `max_replication_lag` column is defined for backend servers in [the `mysql_servers` admin table](https://github.com/sysown/proxysql/wiki/Main-(runtime)#mysql_servers)
+    - This ensures servers with lag do not receive reads but are still probed by freno
+

pkg/proxysql/client.go

@@ -0,0 +1,121 @@
+package proxysql
+
+import (
+	"database/sql"
+	"errors"
+	"fmt"
+	"sort"
+	"time"
+
+	"github.com/github/freno/pkg/config"
+	_ "github.com/go-sql-driver/mysql"
+	"github.com/outbrain/golib/log"
+	"github.com/patrickmn/go-cache"
+)
+
+const ignoreServerCacheCleanupTTL = time.Duration(500) * time.Millisecond
+
+// MySQLConnectionPoolServer represents a row in the stats_mysql_connection_pool table
+type MySQLConnectionPoolServer struct {
+	Host   string
+	Port   int32
+	Status string
+}
+
+// Address returns a string of the hostname/port of a server
+func (ms *MySQLConnectionPoolServer) Address() string {
+	return fmt.Sprintf("%s:%d", ms.Host, ms.Port)
+}
+
+// Client is the ProxySQL admin client
+type Client struct {
+	dbs                    map[string]*sql.DB
+	defaultIgnoreServerTTL time.Duration
+	ignoreServerCache      *cache.Cache
+}
+
+// NewClient returns a new ProxySQL admin client
+func NewClient(defaultIgnoreServerTTL time.Duration) *Client {
+	return &Client{
+		dbs:                    make(map[string]*sql.DB, 0),
+		defaultIgnoreServerTTL: defaultIgnoreServerTTL,
+		ignoreServerCache:      cache.New(cache.NoExpiration, ignoreServerCacheCleanupTTL),
+	}
+}
+
+// GetDB returns a configured ProxySQL admin connection
+func (c *Client) GetDB(settings config.ProxySQLConfigurationSettings) (*sql.DB, string, error) {
+	addrs := settings.Addresses
+	sort.Strings(addrs)
+
+	var lastErr error
+	for _, addr := range addrs {
+		if db, found := c.dbs[addr]; found {
+			return db, addr, nil
+		}
+		db, err := sql.Open("mysql", fmt.Sprintf("%s:%s@tcp(%s)/%s?interpolateParams=true&timeout=500ms",
+			settings.User, settings.Password, addr, config.ProxySQLDefaultDatabase,
+		))
+		if err != nil {
+			lastErr = err
+			continue
+		}
+		if err = db.Ping(); err != nil {
+			lastErr = err
+			continue
+		}
+		log.Debugf("connected to ProxySQL at %s", settings.AddressToDSN(addr))
+		c.dbs[addr] = db
+		return c.dbs[addr], addr, nil
+	}
+	if lastErr != nil {
+		return nil, "", lastErr
+	}
+	return nil, "", errors.New("failed to get connection")
+}
+
+// CloseDB closes a ProxySQL admin connection based on an address string
+func (c *Client) CloseDB(addr string) {
+	if db, found := c.dbs[addr]; found {
+		db.Close()
+		delete(c.dbs, addr)
+	}
+}
+
+// GetServers returns a list of MySQLConnectionPoolServers with 'ONLINE' or 'SHUNNED_REPLICATION_LAG' status, based on hostgroup ID
+func (c *Client) GetServers(db *sql.DB, settings config.ProxySQLConfigurationSettings) (servers []*MySQLConnectionPoolServer, err error) {
+	ignoreServerTTL := c.defaultIgnoreServerTTL
+	if settings.IgnoreServerTTLSecs > 0 {
+		ignoreServerTTL = time.Duration(settings.IgnoreServerTTLSecs) * time.Second
+	}
+
+	rows, err := db.Query(fmt.Sprintf(`SELECT srv_host, srv_port, status FROM stats_mysql_connection_pool WHERE hostgroup=%d`, settings.HostgroupID))
+	if err != nil {
+		return servers, err
+	}
+	defer rows.Close()
+
+	var server *MySQLConnectionPoolServer
+	for rows.Next() {
+		server = &MySQLConnectionPoolServer{}
+		if err = rows.Scan(&server.Host, &server.Port, &server.Status); err != nil {
+			return nil, err
+		}
+
+		switch server.Status {
+		case "ONLINE":
+			if _, ignore := c.ignoreServerCache.Get(server.Address()); ignore {
+				log.Debugf("found %q in the proxysql ignore-server cache, ignoring ONLINE state for %s", server.Address(), ignoreServerTTL)
+				continue
+			}
+			servers = append(servers, server)
+		case "SHUNNED_REPLICATION_LAG":
+			defer c.ignoreServerCache.Delete(server.Address())
+			servers = append(servers, server)
+		default:
+			c.ignoreServerCache.Set(server.Address(), true, ignoreServerTTL)
+		}
+	}
+
+	return servers, rows.Err()
+}