Add support for MySQL notifications (fixes #3818) (#3907)

As a new configuration parameter is added, configuration version is
bumped up from 14 to 15.

The MySQL target's behaviour is identical to the PostgreSQL: rows are
deleted from the MySQL table on delete-object events, and are
created/updated on create/over-write events.

cmd/bucket-notification-utils.go

@@ -148,6 +148,10 @@ func isValidQueueID(queueARN string) bool {
 		pgN := serverConfig.Notify.GetPostgreSQLByID(sqsARN.AccountID)
 		// Postgres can work with only default conn. info.
 		return pgN.Enable
+	} else if isMySQLQueue(sqsARN) {
+		msqlN := serverConfig.Notify.GetMySQLByID(sqsARN.AccountID)
+		// Mysql can work with only default conn. info.
+		return msqlN.Enable
 	} else if isKafkaQueue(sqsARN) {
 		kafkaN := serverConfig.Notify.GetKafkaByID(sqsARN.AccountID)
 		return (kafkaN.Enable && len(kafkaN.Brokers) > 0 &&
@@ -244,6 +248,7 @@ func validateNotificationConfig(nConfig notificationConfig) APIErrorCode {
 // - elasticsearch
 // - redis
 // - postgresql
+// - mysql
 // - kafka
 // - webhook
 func unmarshalSqsARN(queueARN string) (mSqs arnSQS) {
@@ -263,6 +268,8 @@ func unmarshalSqsARN(queueARN string) (mSqs arnSQS) {
 		mSqs.Type = queueTypeRedis
 	case hasSuffix(sqsType, queueTypePostgreSQL):
 		mSqs.Type = queueTypePostgreSQL
+	case hasSuffix(sqsType, queueTypeMySQL):
+		mSqs.Type = queueTypeMySQL
 	case hasSuffix(sqsType, queueTypeKafka):
 		mSqs.Type = queueTypeKafka
 	case hasSuffix(sqsType, queueTypeWebhook):

cmd/config-migrate.go

@@ -78,6 +78,10 @@ func migrateConfig() error {
 	if err := migrateV13ToV14(); err != nil {
 		return err
 	}
+	// Migration version '14' to '15'.
+	if err := migrateV14ToV15(); err != nil {
+		return err
+	}
 
 	return nil
 }
@@ -905,7 +909,7 @@ func migrateV12ToV13() error {
 	return nil
 }
 
-// Version '13' to '14' migration. Add support for custom webhook endpoint.
+// Version '13' to '14' migration. Add support for browser param.
 func migrateV13ToV14() error {
 	cv13, err := loadConfigV13()
 	if err != nil {
@@ -1003,3 +1007,106 @@ func migrateV13ToV14() error {
 	)
 	return nil
 }
+
+// Version '14' to '15' migration. Add support for MySQL notifications.
+func migrateV14ToV15() error {
+	cv14, err := loadConfigV14()
+	if err != nil {
+		if os.IsNotExist(err) {
+			return nil
+		}
+		return fmt.Errorf("Unable to load config version ‘14’. %v", err)
+	}
+	if cv14.Version != "14" {
+		return nil
+	}
+
+	// Copy over fields from V14 into V15 config struct
+	srvConfig := &serverConfigV15{
+		Logger: &logger{},
+		Notify: &notifier{},
+	}
+	srvConfig.Version = "15"
+	srvConfig.Credential = cv14.Credential
+	srvConfig.Region = cv14.Region
+	if srvConfig.Region == "" {
+		// Region needs to be set for AWS Signature Version 4.
+		srvConfig.Region = globalMinioDefaultRegion
+	}
+	srvConfig.Logger.Console = cv14.Logger.Console
+	srvConfig.Logger.File = cv14.Logger.File
+
+	// check and set notifiers config
+	if len(cv14.Notify.AMQP) == 0 {
+		srvConfig.Notify.AMQP = make(map[string]amqpNotify)
+		srvConfig.Notify.AMQP["1"] = amqpNotify{}
+	} else {
+		srvConfig.Notify.AMQP = cv14.Notify.AMQP
+	}
+	if len(cv14.Notify.ElasticSearch) == 0 {
+		srvConfig.Notify.ElasticSearch = make(map[string]elasticSearchNotify)
+		srvConfig.Notify.ElasticSearch["1"] = elasticSearchNotify{}
+	} else {
+		srvConfig.Notify.ElasticSearch = cv14.Notify.ElasticSearch
+	}
+	if len(cv14.Notify.Redis) == 0 {
+		srvConfig.Notify.Redis = make(map[string]redisNotify)
+		srvConfig.Notify.Redis["1"] = redisNotify{}
+	} else {
+		srvConfig.Notify.Redis = cv14.Notify.Redis
+	}
+	if len(cv14.Notify.PostgreSQL) == 0 {
+		srvConfig.Notify.PostgreSQL = make(map[string]postgreSQLNotify)
+		srvConfig.Notify.PostgreSQL["1"] = postgreSQLNotify{}
+	} else {
+		srvConfig.Notify.PostgreSQL = cv14.Notify.PostgreSQL
+	}
+	if len(cv14.Notify.Kafka) == 0 {
+		srvConfig.Notify.Kafka = make(map[string]kafkaNotify)
+		srvConfig.Notify.Kafka["1"] = kafkaNotify{}
+	} else {
+		srvConfig.Notify.Kafka = cv14.Notify.Kafka
+	}
+	if len(cv14.Notify.NATS) == 0 {
+		srvConfig.Notify.NATS = make(map[string]natsNotify)
+		srvConfig.Notify.NATS["1"] = natsNotify{}
+	} else {
+		srvConfig.Notify.NATS = cv14.Notify.NATS
+	}
+	if len(cv14.Notify.Webhook) == 0 {
+		srvConfig.Notify.Webhook = make(map[string]webhookNotify)
+		srvConfig.Notify.Webhook["1"] = webhookNotify{}
+	} else {
+		srvConfig.Notify.Webhook = cv14.Notify.Webhook
+	}
+
+	// V14 will not have mysql support, so we add that here.
+	srvConfig.Notify.MySQL = make(map[string]mySQLNotify)
+	srvConfig.Notify.MySQL["1"] = mySQLNotify{}
+
+	// Load browser config from existing config in the file.
+	srvConfig.Browser = cv14.Browser
+
+	qc, err := quick.New(srvConfig)
+	if err != nil {
+		return fmt.Errorf("Unable to initialize the quick config. %v",
+			err)
+	}
+
+	configFile := getConfigFile()
+	err = qc.Save(configFile)
+	if err != nil {
+		return fmt.Errorf(
+			"Failed to migrate config from ‘"+
+				cv14.Version+"’ to ‘"+srvConfig.Version+
+				"’ failed. %v", err,
+		)
+	}
+
+	console.Println(
+		"Migration from version ‘" +
+			cv14.Version + "’ to ‘" + srvConfig.Version +
+			"’ completed successfully.",
+	)
+	return nil
+}

cmd/config-migrate_test.go

@@ -109,10 +109,13 @@ func TestServerConfigMigrateInexistentConfig(t *testing.T) {
 	if err := migrateV13ToV14(); err != nil {
 		t.Fatal("migrate v13 to v14 should succeed when no config file is found")
 	}
+	if err := migrateV14ToV15(); err != nil {
+		t.Fatal("migrate v14 to v15 should succeed when no config file is found")
+	}
 }
 
-// Test if a config migration from v2 to v12 is successfully done
-func TestServerConfigMigrateV2toV14(t *testing.T) {
+// Test if a config migration from v2 to v15 is successfully done
+func TestServerConfigMigrateV2toV15(t *testing.T) {
 	rootPath, err := newTestConfig(globalMinioDefaultRegion)
 	if err != nil {
 		t.Fatalf("Init Test config failed")
@@ -151,7 +154,7 @@ func TestServerConfigMigrateV2toV14(t *testing.T) {
 	}
 
 	// Check the version number in the upgraded config file
-	expectedVersion := v14
+	expectedVersion := v15
 	if serverConfig.Version != expectedVersion {
 		t.Fatalf("Expect version "+expectedVersion+", found: %v", serverConfig.Version)
 	}
@@ -219,4 +222,7 @@ func TestServerConfigMigrateFaultyConfig(t *testing.T) {
 	if err := migrateV13ToV14(); err == nil {
 		t.Fatal("migrateConfigV13ToV14() should fail with a corrupted json")
 	}
+	if err := migrateV14ToV15(); err == nil {
+		t.Fatal("migrateConfigV14ToV15() should fail with a corrupted json")
+	}
 }

cmd/config-old.go

@@ -510,3 +510,29 @@ func loadConfigV13() (*serverConfigV13, error) {
 	}
 	return config.(*serverConfigV13), err
 }
+
+// serverConfigV14 server configuration version '14' which is like
+// version '13' except it adds support of browser param.
+type serverConfigV14 struct {
+	Version string `json:"version"`
+
+	// S3 API configuration.
+	Credential credential `json:"credential"`
+	Region     string     `json:"region"`
+	Browser    string     `json:"browser"`
+
+	// Additional error logging configuration.
+	Logger *logger `json:"logger"`
+
+	// Notification queue configuration.
+	Notify *notifier `json:"notify"`
+}
+
+func loadConfigV14() (*serverConfigV14, error) {
+	configFile := getConfigFile()
+	config, err := loadOldConfig(configFile, &serverConfigV14{Version: "14"})
+	if config == nil {
+		return nil, err
+	}
+	return config.(*serverConfigV14), err
+}

cmd/config-v15.go

@@ -32,11 +32,11 @@ import (
 var serverConfigMu sync.RWMutex
 
 // Config version
-var v14 = "14"
+var v15 = "15"
 
-// serverConfigV14 server configuration version '14' which is like
-// version '13' except it adds support of browser param.
-type serverConfigV14 struct {
+// serverConfigV15 server configuration version '15' which is like
+// version '14' except it adds support of MySQL notifications.
+type serverConfigV15 struct {
 	Version string `json:"version"`
 
 	// S3 API configuration.
@@ -51,9 +51,9 @@ type serverConfigV14 struct {
 	Notify *notifier `json:"notify"`
 }
 
-func newServerConfigV14() *serverConfigV14 {
-	srvCfg := &serverConfigV14{
-		Version: v14,
+func newServerConfigV15() *serverConfigV15 {
+	srvCfg := &serverConfigV15{
+		Version: v15,
 		Region:  globalMinioDefaultRegion,
 		Logger:  &logger{},
 		Notify:  &notifier{},
@@ -77,6 +77,8 @@ func newServerConfigV14() *serverConfigV14 {
 	srvCfg.Notify.NATS["1"] = natsNotify{}
 	srvCfg.Notify.PostgreSQL = make(map[string]postgreSQLNotify)
 	srvCfg.Notify.PostgreSQL["1"] = postgreSQLNotify{}
+	srvCfg.Notify.MySQL = make(map[string]mySQLNotify)
+	srvCfg.Notify.MySQL["1"] = mySQLNotify{}
 	srvCfg.Notify.Kafka = make(map[string]kafkaNotify)
 	srvCfg.Notify.Kafka["1"] = kafkaNotify{}
 	srvCfg.Notify.Webhook = make(map[string]webhookNotify)
@@ -89,7 +91,7 @@ func newServerConfigV14() *serverConfigV14 {
 // found, otherwise use default parameters
 func newConfig(envParams envParams) error {
 	// Initialize server config.
-	srvCfg := newServerConfigV14()
+	srvCfg := newServerConfigV15()
 
 	// If env is set for a fresh start, save them to config file.
 	if globalIsEnvCreds {
@@ -124,7 +126,7 @@ func loadConfig(envParams envParams) error {
 		return err
 	}
 
-	srvCfg := &serverConfigV14{}
+	srvCfg := &serverConfigV15{}
 
 	qc, err := quick.New(srvCfg)
 	if err != nil {
@@ -154,7 +156,7 @@ func loadConfig(envParams envParams) error {
 	serverConfig = srvCfg
 	serverConfigMu.Unlock()
 
-	if serverConfig.Version != v14 {
+	if serverConfig.Version != v15 {
 		return errors.New("Unsupported config version `" + serverConfig.Version + "`.")
 	}
 
@@ -218,7 +220,7 @@ func validateConfig() error {
 	// Get file config path
 	configFile := getConfigFile()
 
-	srvCfg := &serverConfigV14{}
+	srvCfg := &serverConfigV15{}
 
 	// Load config file
 	qc, err := quick.New(srvCfg)
@@ -230,8 +232,8 @@ func validateConfig() error {
 	}
 
 	// Check if config version is valid
-	if srvCfg.GetVersion() != v14 {
-		return errors.New("bad config version, expected: " + v14)
+	if srvCfg.GetVersion() != v15 {
+		return errors.New("bad config version, expected: " + v15)
 	}
 
 	// Load config file json and check for duplication json keys
@@ -272,10 +274,10 @@ func validateConfig() error {
 }
 
 // serverConfig server config.
-var serverConfig *serverConfigV14
+var serverConfig *serverConfigV15
 
 // GetVersion get current config version.
-func (s serverConfigV14) GetVersion() string {
+func (s serverConfigV15) GetVersion() string {
 	serverConfigMu.RLock()
 	defer serverConfigMu.RUnlock()
 
@@ -283,7 +285,7 @@ func (s serverConfigV14) GetVersion() string {
 }
 
 // SetRegion set new region.
-func (s *serverConfigV14) SetRegion(region string) {
+func (s *serverConfigV15) SetRegion(region string) {
 	serverConfigMu.Lock()
 	defer serverConfigMu.Unlock()
 
@@ -295,7 +297,7 @@ func (s *serverConfigV14) SetRegion(region string) {
 }
 
 // GetRegion get current region.
-func (s serverConfigV14) GetRegion() string {
+func (s serverConfigV15) GetRegion() string {
 	serverConfigMu.RLock()
 	defer serverConfigMu.RUnlock()
 
@@ -308,7 +310,7 @@ func (s serverConfigV14) GetRegion() string {
 }
 
 // SetCredentials set new credentials.
-func (s *serverConfigV14) SetCredential(creds credential) {
+func (s *serverConfigV15) SetCredential(creds credential) {
 	serverConfigMu.Lock()
 	defer serverConfigMu.Unlock()
 
@@ -317,7 +319,7 @@ func (s *serverConfigV14) SetCredential(creds credential) {
 }
 
 // GetCredentials get current credentials.
-func (s serverConfigV14) GetCredential() credential {
+func (s serverConfigV15) GetCredential() credential {
 	serverConfigMu.RLock()
 	defer serverConfigMu.RUnlock()
 
@@ -325,7 +327,7 @@ func (s serverConfigV14) GetCredential() credential {
 }
 
 // SetBrowser set if browser is enabled.
-func (s *serverConfigV14) SetBrowser(v string) {
+func (s *serverConfigV15) SetBrowser(v string) {
 	serverConfigMu.Lock()
 	defer serverConfigMu.Unlock()
 
@@ -339,7 +341,7 @@ func (s *serverConfigV14) SetBrowser(v string) {
 }
 
 // GetCredentials get current credentials.
-func (s serverConfigV14) GetBrowser() string {
+func (s serverConfigV15) GetBrowser() string {
 	serverConfigMu.RLock()
 	defer serverConfigMu.RUnlock()
 
@@ -352,7 +354,7 @@ func (s serverConfigV14) GetBrowser() string {
 }
 
 // Save config.
-func (s serverConfigV14) Save() error {
+func (s serverConfigV15) Save() error {
 	serverConfigMu.RLock()
 	defer serverConfigMu.RUnlock()
 

cmd/config-v15_test.go

@@ -76,10 +76,17 @@ func TestServerConfig(t *testing.T) {
 	serverConfig.Notify.SetWebhookByID("2", webhookNotify{})
 	savedNotifyCfg5 := serverConfig.Notify.GetWebhookByID("2")
 	if !reflect.DeepEqual(savedNotifyCfg5, webhookNotify{}) {
-		t.Errorf("Expecting Webhook config %#v found %#v", webhookNotify{}, savedNotifyCfg3)
+		t.Errorf("Expecting Webhook config %#v found %#v", webhookNotify{}, savedNotifyCfg5)
 	}
 
 	// Set new console logger.
+	// Set new Webhook notification id.
+	serverConfig.Notify.SetMySQLByID("2", mySQLNotify{})
+	savedNotifyCfg6 := serverConfig.Notify.GetMySQLByID("2")
+	if !reflect.DeepEqual(savedNotifyCfg6, mySQLNotify{}) {
+		t.Errorf("Expecting Webhook config %#v found %#v", mySQLNotify{}, savedNotifyCfg6)
+	}
+
 	serverConfig.Logger.SetConsole(consoleLogger{
 		Enable: true,
 	})
@@ -106,8 +113,8 @@ func TestServerConfig(t *testing.T) {
 	})
 
 	// Match version.
-	if serverConfig.GetVersion() != v14 {
-		t.Errorf("Expecting version %s found %s", serverConfig.GetVersion(), v14)
+	if serverConfig.GetVersion() != v15 {
+		t.Errorf("Expecting version %s found %s", serverConfig.GetVersion(), v15)
 	}
 
 	// Attempt to save.
@@ -212,7 +219,7 @@ func TestValidateConfig(t *testing.T) {
 
 	configPath := filepath.Join(rootPath, minioConfigFile)
 
-	v := v14
+	v := v15
 
 	testCases := []struct {
 		configData string
@@ -274,6 +281,9 @@ func TestValidateConfig(t *testing.T) {
 
 		// Test 19 - Test Webhook
 		{`{"version": "` + v + `", "credential": { "accessKey": "minio", "secretKey": "minio123" }, "region": "us-east-1", "browser": "on", "notify": { "webhook": { "1": { "enable": true, "endpoint": "" } }}}`, false},
+
+		// Test 19 - Test MySQL
+		{`{"version": "` + v + `", "credential": { "accessKey": "minio", "secretKey": "minio123" }, "region": "us-east-1", "browser": "on", "notify": { "mysql": { "1": { "enable": true, "dsnString": "",  "table": "", "host": "", "port": "", "user": "", "password": "", "database": "" }}}}`, false},
 	}
 
 	for i, testCase := range testCases {

cmd/event-notifier.go

@@ -674,6 +674,25 @@ func loadAllQueueTargets() (map[string]*logrus.Logger, error) {
 		}
 	}
 
+	// Load MySQL targets, initialize their respective loggers.
+	for accountID, msqlN := range serverConfig.Notify.GetMySQL() {
+		if !msqlN.Enable {
+			continue
+		}
+
+		if queueARN, err := addQueueTarget(queueTargets, accountID, queueTypeMySQL, newMySQLNotify); err != nil {
+			if _, ok := err.(net.Error); ok {
+				err = &net.OpError{
+					Op:  "Connecting to " + queueARN,
+					Net: "tcp",
+					Err: err,
+				}
+			}
+
+			return nil, err
+		}
+	}
+
 	// Load Kafka targets, initialize their respective loggers.
 	for accountID, kafkaN := range serverConfig.Notify.GetKafka() {
 		if !kafkaN.Enable {

cmd/gateway-main.go

@@ -96,7 +96,7 @@ func newGatewayLayer(backendType, accessKey, secretKey string) (GatewayLayer, er
 // only used in memory.
 func newGatewayConfig(accessKey, secretKey, region string) error {
 	// Initialize server config.
-	srvCfg := newServerConfigV14()
+	srvCfg := newServerConfigV15()
 
 	// If env is set for a fresh start, save them to config file.
 	srvCfg.SetCredential(credential{

cmd/notifier-config.go

@@ -31,6 +31,7 @@ type notifier struct {
 	PostgreSQL    postgreSQLConfigs    `json:"postgresql"`
 	Kafka         kafkaConfigs         `json:"kafka"`
 	Webhook       webhookConfigs       `json:"webhook"`
+	MySQL         mySQLConfigs         `json:"mysql"`
 	// Add new notification queues.
 }
 
@@ -167,6 +168,25 @@ func (a webhookConfigs) Validate() error {
 	return nil
 }
 
+type mySQLConfigs map[string]mySQLNotify
+
+func (a mySQLConfigs) Clone() mySQLConfigs {
+	a2 := make(mySQLConfigs, len(a))
+	for k, v := range a {
+		a2[k] = v
+	}
+	return a2
+}
+
+func (a mySQLConfigs) Validate() error {
+	for k, v := range a {
+		if err := v.Validate(); err != nil {
+			return fmt.Errorf("MySQL [%s] configuration invalid: %s", k, err.Error())
+		}
+	}
+	return nil
+}
+
 func (n *notifier) Validate() error {
 	if n == nil {
 		return nil
@@ -192,6 +212,9 @@ func (n *notifier) Validate() error {
 	if err := n.Webhook.Validate(); err != nil {
 		return err
 	}
+	if err := n.MySQL.Validate(); err != nil {
+		return err
+	}
 	return nil
 }
 
@@ -303,6 +326,24 @@ func (n *notifier) GetPostgreSQLByID(accountID string) postgreSQLNotify {
 	return n.PostgreSQL[accountID]
 }
 
+func (n *notifier) SetMySQLByID(accountID string, pgn mySQLNotify) {
+	n.Lock()
+	defer n.Unlock()
+	n.MySQL[accountID] = pgn
+}
+
+func (n *notifier) GetMySQL() map[string]mySQLNotify {
+	n.RLock()
+	defer n.RUnlock()
+	return n.MySQL.Clone()
+}
+
+func (n *notifier) GetMySQLByID(accountID string) mySQLNotify {
+	n.RLock()
+	defer n.RUnlock()
+	return n.MySQL[accountID]
+}
+
 func (n *notifier) SetKafkaByID(accountID string, kn kafkaNotify) {
 	n.Lock()
 	defer n.Unlock()

cmd/notifiers.go

@@ -37,6 +37,8 @@ const (
 	queueTypeRedis = "redis"
 	// Static string indicating queue type 'postgresql'.
 	queueTypePostgreSQL = "postgresql"
+	// Static string indicating queue type 'mysql'.
+	queueTypeMySQL = "mysql"
 	// Static string indicating queue type 'kafka'.
 	queueTypeKafka = "kafka"
 	// Static string for Webhooks
@@ -156,6 +158,24 @@ func isPostgreSQLQueue(sqsArn arnSQS) bool {
 	return true
 }
 
+// Returns true if queueArn is for MySQL.
+func isMySQLQueue(sqsArn arnSQS) bool {
+	if sqsArn.Type != queueTypeMySQL {
+		return false
+	}
+	msqlNotify := serverConfig.Notify.GetMySQLByID(sqsArn.AccountID)
+	if !msqlNotify.Enable {
+		return false
+	}
+	myC, err := dialMySQL(msqlNotify)
+	if err != nil {
+		errorIf(err, "Unable to connect to MySQL server %#v", msqlNotify)
+		return false
+	}
+	defer myC.Close()
+	return true
+}
+
 // Returns true if queueArn is for Kafka.
 func isKafkaQueue(sqsArn arnSQS) bool {
 	if sqsArn.Type != queueTypeKafka {

cmd/notify-mysql.go

@@ -0,0 +1,252 @@
+/*
+ * Minio Cloud Storage, (C) 2017 Minio, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// MySQL Notifier implementation. A table with a specific
+// structure (column names, column types, and primary key/uniqueness
+// constraint) is used. The user may set the table name in the
+// configuration. A sample SQL command that creates a command with the
+// required structure is:
+//
+//     CREATE TABLE myminio (
+//         key_name VARCHAR(2048),
+//         value JSONB,
+//         PRIMARY KEY (key_name),
+//     );
+//
+// MySQL's "INSERT ... ON DUPLICATE ..." feature (UPSERT) is used
+// here. The implementation has been tested with MySQL Ver 14.14
+// Distrib 5.7.17.
+//
+// On each create or update object event in Minio Object storage
+// server, a row is created or updated in the table in MySQL. On
+// each object removal, the corresponding row is deleted from the
+// table.
+
+package cmd
+
+import (
+	"database/sql"
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+
+	"github.com/Sirupsen/logrus"
+	"github.com/go-sql-driver/mysql"
+)
+
+const (
+	upsertRowMySQL = `INSERT INTO %s (key_name, value)
+VALUES (?, ?)
+ON DUPLICATE KEY UPDATE value=VALUES(value);
+`
+	deleteRowMySQL = ` DELETE FROM %s
+WHERE key_name = ?;`
+	createTableMySQL = `CREATE TABLE %s (
+    key_name VARCHAR(2048),
+    value JSON,
+    PRIMARY KEY (key_name)
+);`
+	tableExistsMySQL = `SELECT 1 FROM %s;`
+)
+
+type mySQLNotify struct {
+	Enable bool `json:"enable"`
+
+	// pass data-source-name connection string in config
+	// directly. This string is formatted according to
+	// https://github.com/go-sql-driver/mysql#dsn-data-source-name
+	DsnString string `json:"dsnString"`
+	// specifying a table name is required.
+	Table string `json:"table"`
+
+	// uses the values below if no connection string is specified
+	// - however the connection string method offers more
+	// flexibility.
+	Host     string `json:"host"`
+	Port     string `json:"port"`
+	User     string `json:"user"`
+	Password string `json:"password"`
+	Database string `json:"database"`
+}
+
+func (m *mySQLNotify) Validate() error {
+	if !m.Enable {
+		return nil
+	}
+	if m.DsnString == "" {
+		if _, err := checkURL(m.Host); err != nil {
+			return err
+		}
+	}
+	if m.Table == "" {
+		return fmt.Errorf(
+			"MySQL Notifier Error: Table was not specified in configuration")
+	}
+	return nil
+}
+
+type mySQLConn struct {
+	dsnStr        string
+	table         string
+	preparedStmts map[string]*sql.Stmt
+	*sql.DB
+}
+
+func dialMySQL(msql mySQLNotify) (mySQLConn, error) {
+	if !msql.Enable {
+		return mySQLConn{}, errNotifyNotEnabled
+	}
+
+	dsnStr := msql.DsnString
+	// check if connection string is specified
+	if dsnStr == "" {
+		// build from other parameters
+		config := mysql.Config{
+			User:   msql.User,
+			Passwd: msql.Password,
+			Net:    "tcp",
+			Addr:   msql.Host + ":" + msql.Port,
+			DBName: msql.Database,
+		}
+		dsnStr = config.FormatDSN()
+	}
+
+	db, err := sql.Open("mysql", dsnStr)
+	if err != nil {
+		return mySQLConn{}, fmt.Errorf(
+			"MySQL Notifier Error: Connection opening failure (dsnStr=%s): %v",
+			dsnStr, err,
+		)
+	}
+
+	// ping to check that server is actually reachable.
+	err = db.Ping()
+	if err != nil {
+		return mySQLConn{}, fmt.Errorf(
+			"MySQL Notifier Error: Ping to server failed with: %v",
+			err,
+		)
+	}
+
+	// check that table exists - if not, create it.
+	_, err = db.Exec(fmt.Sprintf(tableExistsMySQL, msql.Table))
+	if err != nil {
+		// most likely, table does not exist. try to create it:
+		_, errCreate := db.Exec(fmt.Sprintf(createTableMySQL, msql.Table))
+		if errCreate != nil {
+			// failed to create the table. error out.
+			return mySQLConn{}, fmt.Errorf(
+				"MySQL Notifier Error: 'Select' failed with %v, then 'Create Table' failed with %v",
+				err, errCreate,
+			)
+		}
+	}
+
+	// create prepared statements
+	stmts := make(map[string]*sql.Stmt)
+	// insert or update statement
+	stmts["upsertRow"], err = db.Prepare(fmt.Sprintf(upsertRowMySQL, msql.Table))
+	if err != nil {
+		return mySQLConn{},
+			fmt.Errorf("MySQL Notifier Error: create UPSERT prepared statement failed with: %v", err)
+	}
+	stmts["deleteRow"], err = db.Prepare(fmt.Sprintf(deleteRowMySQL, msql.Table))
+	if err != nil {
+		return mySQLConn{},
+			fmt.Errorf("MySQL Notifier Error: create DELETE prepared statement failed with: %v", err)
+	}
+
+	return mySQLConn{dsnStr, msql.Table, stmts, db}, nil
+}
+
+func newMySQLNotify(accountID string) (*logrus.Logger, error) {
+	mysqlNotify := serverConfig.Notify.GetMySQLByID(accountID)
+
+	// Dial mysql
+	myC, err := dialMySQL(mysqlNotify)
+	if err != nil {
+		return nil, err
+	}
+
+	mySQLLog := logrus.New()
+
+	mySQLLog.Out = ioutil.Discard
+
+	mySQLLog.Formatter = new(logrus.JSONFormatter)
+
+	mySQLLog.Hooks.Add(myC)
+
+	return mySQLLog, nil
+}
+
+func (myC mySQLConn) Close() {
+	// first close all prepared statements
+	for _, v := range myC.preparedStmts {
+		_ = v.Close()
+	}
+	// close db connection
+	_ = myC.DB.Close()
+}
+
+func (myC mySQLConn) Fire(entry *logrus.Entry) error {
+	// get event type by trying to convert to string
+	entryEventType, ok := entry.Data["EventType"].(string)
+	if !ok {
+		// ignore event if converting EventType to string
+		// fails.
+		return nil
+	}
+
+	// Check for event delete
+	if eventMatch(entryEventType, []string{"s3:ObjectRemoved:*"}) {
+		// delete row from the table
+		_, err := myC.preparedStmts["deleteRow"].Exec(entry.Data["Key"])
+		if err != nil {
+			return fmt.Errorf(
+				"Error deleting event with key = %v - got mysql error - %v",
+				entry.Data["Key"], err,
+			)
+		}
+	} else {
+		// json encode the value for the row
+		value, err := json.Marshal(map[string]interface{}{
+			"Records": entry.Data["Records"],
+		})
+		if err != nil {
+			return fmt.Errorf(
+				"Unable to encode event %v to JSON - got error - %v",
+				entry.Data["Records"], err,
+			)
+		}
+
+		// upsert row into the table
+		_, err = myC.preparedStmts["upsertRow"].Exec(entry.Data["Key"], value)
+		if err != nil {
+			return fmt.Errorf(
+				"Unable to upsert event with Key=%v and Value=%v - got mysql error - %v",
+				entry.Data["Key"], entry.Data["Records"], err,
+			)
+		}
+	}
+
+	return nil
+}
+
+func (myC mySQLConn) Levels() []logrus.Level {
+	return []logrus.Level{
+		logrus.InfoLevel,
+	}
+}