Add support for Kafka as a notifications target (#2869) (#3439)

cmd/bucket-notification-utils.go

@@ -147,6 +147,10 @@ func isValidQueueID(queueARN string) bool {
 		pgN := serverConfig.GetPostgreSQLNotifyByID(sqsARN.AccountID)
 		// Postgres can work with only default conn. info.
 		return pgN.Enable
+	} else if isKafkaQueue(sqsARN) {
+		kafkaN := serverConfig.GetKafkaNotifyByID(sqsARN.AccountID)
+		return (kafkaN.Enable && len(kafkaN.Brokers) > 0 &&
+			kafkaN.Topic != "")
 	}
 	return false
 }
@@ -236,6 +240,7 @@ func validateNotificationConfig(nConfig notificationConfig) APIErrorCode {
 // - elasticsearch
 // - redis
 // - postgresql
+// - kafka
 func unmarshalSqsARN(queueARN string) (mSqs arnSQS) {
 	mSqs = arnSQS{}
 	if !strings.HasPrefix(queueARN, minioSqs+serverConfig.GetRegion()+":") {
@@ -253,6 +258,8 @@ func unmarshalSqsARN(queueARN string) (mSqs arnSQS) {
 		mSqs.Type = queueTypeRedis
 	case strings.HasSuffix(sqsType, queueTypePostgreSQL):
 		mSqs.Type = queueTypePostgreSQL
+	case strings.HasSuffix(sqsType, queueTypeKafka):
+		mSqs.Type = queueTypeKafka
 	} // Add more queues here.
 	mSqs.AccountID = strings.TrimSuffix(sqsType, ":"+mSqs.Type)
 	return mSqs

cmd/config-migrate.go

@@ -62,6 +62,10 @@ func migrateConfig() error {
 	if err := migrateV9ToV10(); err != nil {
 		return err
 	}
+	// Migrate version '10' to '11'.
+	if err := migrateV10ToV11(); err != nil {
+		return err
+	}
 	return nil
 }
 
@@ -633,3 +637,91 @@ func migrateV9ToV10() error {
 	)
 	return nil
 }
+
+// Version '10' to '11' migration. Add support for Kafka
+// notifications.
+func migrateV10ToV11() error {
+	cv10, err := loadConfigV10()
+	if err != nil {
+		if os.IsNotExist(err) {
+			return nil
+		}
+		return fmt.Errorf("Unable to load config version ‘10’. %v", err)
+	}
+	if cv10.Version != "10" {
+		return nil
+	}
+
+	// Copy over fields from V10 into V11 config struct
+	srvConfig := &serverConfigV11{}
+	srvConfig.Version = "11"
+	srvConfig.Credential = cv10.Credential
+	srvConfig.Region = cv10.Region
+	if srvConfig.Region == "" {
+		// Region needs to be set for AWS Signature Version 4.
+		srvConfig.Region = "us-east-1"
+	}
+	srvConfig.Logger.Console = cv10.Logger.Console
+	srvConfig.Logger.File = cv10.Logger.File
+
+	// check and set notifiers config
+	if len(cv10.Notify.AMQP) == 0 {
+		srvConfig.Notify.AMQP = make(map[string]amqpNotify)
+		srvConfig.Notify.AMQP["1"] = amqpNotify{}
+	} else {
+		srvConfig.Notify.AMQP = cv10.Notify.AMQP
+	}
+	if len(cv10.Notify.NATS) == 0 {
+		srvConfig.Notify.NATS = make(map[string]natsNotify)
+		srvConfig.Notify.NATS["1"] = natsNotify{}
+	} else {
+		srvConfig.Notify.NATS = cv10.Notify.NATS
+	}
+	if len(cv10.Notify.ElasticSearch) == 0 {
+		srvConfig.Notify.ElasticSearch = make(map[string]elasticSearchNotify)
+		srvConfig.Notify.ElasticSearch["1"] = elasticSearchNotify{}
+	} else {
+		srvConfig.Notify.ElasticSearch = cv10.Notify.ElasticSearch
+	}
+	if len(cv10.Notify.Redis) == 0 {
+		srvConfig.Notify.Redis = make(map[string]redisNotify)
+		srvConfig.Notify.Redis["1"] = redisNotify{}
+	} else {
+		srvConfig.Notify.Redis = cv10.Notify.Redis
+	}
+	if len(cv10.Notify.PostgreSQL) == 0 {
+		srvConfig.Notify.PostgreSQL = make(map[string]postgreSQLNotify)
+		srvConfig.Notify.PostgreSQL["1"] = postgreSQLNotify{}
+	} else {
+		srvConfig.Notify.PostgreSQL = cv10.Notify.PostgreSQL
+	}
+	// V10 will not have a Kafka config. So we initialize one here.
+	srvConfig.Notify.Kafka = make(map[string]kafkaNotify)
+	srvConfig.Notify.Kafka["1"] = kafkaNotify{}
+
+	qc, err := quick.New(srvConfig)
+	if err != nil {
+		return fmt.Errorf("Unable to initialize the quick config. %v",
+			err)
+	}
+	configFile, err := getConfigFile()
+	if err != nil {
+		return fmt.Errorf("Unable to get config file. %v", err)
+	}
+
+	err = qc.Save(configFile)
+	if err != nil {
+		return fmt.Errorf(
+			"Failed to migrate config from ‘"+
+				cv10.Version+"’ to ‘"+srvConfig.Version+
+				"’ failed. %v", err,
+		)
+	}
+
+	console.Println(
+		"Migration from version ‘" +
+			cv10.Version + "’ to ‘" + srvConfig.Version +
+			"’ completed successfully.",
+	)
+	return nil
+}

cmd/config-migrate_test.go

@@ -97,10 +97,13 @@ func TestServerConfigMigrateInexistentConfig(t *testing.T) {
 	if err := migrateV9ToV10(); err != nil {
 		t.Fatal("migrate v9 to v10 should succeed when no config file is found")
 	}
+	if err := migrateV10ToV11(); err != nil {
+		t.Fatal("migrate v10 to v11 should succeed when no config file is found")
+	}
 }
 
-// Test if a config migration from v2 to v10 is successfully done
-func TestServerConfigMigrateV2toV10(t *testing.T) {
+// Test if a config migration from v2 to v11 is successfully done
+func TestServerConfigMigrateV2toV11(t *testing.T) {
 	rootPath, err := newTestConfig("us-east-1")
 	if err != nil {
 		t.Fatalf("Init Test config failed")
@@ -200,4 +203,7 @@ func TestServerConfigMigrateFaultyConfig(t *testing.T) {
 	if err := migrateV9ToV10(); err == nil {
 		t.Fatal("migrateConfigV9ToV10() should fail with a corrupted json")
 	}
+	if err := migrateV10ToV11(); err == nil {
+		t.Fatal("migrateConfigV10ToV11() should fail with a corrupted json")
+	}
 }

cmd/config-old.go

@@ -443,3 +443,40 @@ func loadConfigV9() (*serverConfigV9, error) {
 	}
 	return srvCfg, nil
 }
+
+// serverConfigV10 server configuration version '10' which is like
+// version '9' except it drops support of syslog config, and makes the
+// RWMutex global (so it does not exist in this struct).
+type serverConfigV10 struct {
+	Version string `json:"version"`
+
+	// S3 API configuration.
+	Credential credential `json:"credential"`
+	Region     string     `json:"region"`
+
+	// Additional error logging configuration.
+	Logger logger `json:"logger"`
+
+	// Notification queue configuration.
+	Notify notifier `json:"notify"`
+}
+
+func loadConfigV10() (*serverConfigV10, error) {
+	configFile, err := getConfigFile()
+	if err != nil {
+		return nil, err
+	}
+	if _, err = os.Stat(configFile); err != nil {
+		return nil, err
+	}
+	srvCfg := &serverConfigV10{}
+	srvCfg.Version = "10"
+	qc, err := quick.New(srvCfg)
+	if err != nil {
+		return nil, err
+	}
+	if err := qc.Load(configFile); err != nil {
+		return nil, err
+	}
+	return srvCfg, nil
+}

cmd/config-v11.go

@@ -26,9 +26,9 @@ import (
 // Read Write mutex for safe access to ServerConfig.
 var serverConfigMu sync.RWMutex
 
-// serverConfigV10 server configuration version '10' which is like version '9'
-// except it drops support of syslog config.
-type serverConfigV10 struct {
+// serverConfigV11 server configuration version '11' which is like
+// version '10' except it adds support for Kafka notifications.
+type serverConfigV11 struct {
 	Version string `json:"version"`
 
 	// S3 API configuration.
@@ -42,11 +42,12 @@ type serverConfigV10 struct {
 	Notify notifier `json:"notify"`
 }
 
-// initConfig - initialize server config and indicate if we are creating a new file or we are just loading
+// initConfig - initialize server config and indicate if we are
+// creating a new file or we are just loading
 func initConfig() (bool, error) {
 	if !isConfigFileExists() {
 		// Initialize server config.
-		srvCfg := &serverConfigV10{}
+		srvCfg := &serverConfigV11{}
 		srvCfg.Version = globalMinioConfigVersion
 		srvCfg.Region = "us-east-1"
 		srvCfg.Credential = mustGenAccessKeys()
@@ -68,6 +69,8 @@ func initConfig() (bool, error) {
 		srvCfg.Notify.NATS["1"] = natsNotify{}
 		srvCfg.Notify.PostgreSQL = make(map[string]postgreSQLNotify)
 		srvCfg.Notify.PostgreSQL["1"] = postgreSQLNotify{}
+		srvCfg.Notify.Kafka = make(map[string]kafkaNotify)
+		srvCfg.Notify.Kafka["1"] = kafkaNotify{}
 
 		// Create config path.
 		err := createConfigPath()
@@ -91,7 +94,7 @@ func initConfig() (bool, error) {
 	if _, err = os.Stat(configFile); err != nil {
 		return false, err
 	}
-	srvCfg := &serverConfigV10{}
+	srvCfg := &serverConfigV11{}
 	srvCfg.Version = globalMinioConfigVersion
 	qc, err := quick.New(srvCfg)
 	if err != nil {
@@ -113,10 +116,10 @@ func initConfig() (bool, error) {
 }
 
 // serverConfig server config.
-var serverConfig *serverConfigV10
+var serverConfig *serverConfigV11
 
 // GetVersion get current config version.
-func (s serverConfigV10) GetVersion() string {
+func (s serverConfigV11) GetVersion() string {
 	serverConfigMu.RLock()
 	defer serverConfigMu.RUnlock()
 
@@ -125,14 +128,14 @@ func (s serverConfigV10) GetVersion() string {
 
 /// Logger related.
 
-func (s *serverConfigV10) SetAMQPNotifyByID(accountID string, amqpn amqpNotify) {
+func (s *serverConfigV11) SetAMQPNotifyByID(accountID string, amqpn amqpNotify) {
 	serverConfigMu.Lock()
 	defer serverConfigMu.Unlock()
 
 	s.Notify.AMQP[accountID] = amqpn
 }
 
-func (s serverConfigV10) GetAMQP() map[string]amqpNotify {
+func (s serverConfigV11) GetAMQP() map[string]amqpNotify {
 	serverConfigMu.RLock()
 	defer serverConfigMu.RUnlock()
 
@@ -140,7 +143,7 @@ func (s serverConfigV10) GetAMQP() map[string]amqpNotify {
 }
 
 // GetAMQPNotify get current AMQP logger.
-func (s serverConfigV10) GetAMQPNotifyByID(accountID string) amqpNotify {
+func (s serverConfigV11) GetAMQPNotifyByID(accountID string) amqpNotify {
 	serverConfigMu.RLock()
 	defer serverConfigMu.RUnlock()
 
@@ -148,35 +151,35 @@ func (s serverConfigV10) GetAMQPNotifyByID(accountID string) amqpNotify {
 }
 
 //
-func (s *serverConfigV10) SetNATSNotifyByID(accountID string, natsn natsNotify) {
+func (s *serverConfigV11) SetNATSNotifyByID(accountID string, natsn natsNotify) {
 	serverConfigMu.Lock()
 	defer serverConfigMu.Unlock()
 
 	s.Notify.NATS[accountID] = natsn
 }
 
-func (s serverConfigV10) GetNATS() map[string]natsNotify {
+func (s serverConfigV11) GetNATS() map[string]natsNotify {
 	serverConfigMu.RLock()
 	defer serverConfigMu.RUnlock()
 	return s.Notify.NATS
 }
 
 // GetNATSNotify get current NATS logger.
-func (s serverConfigV10) GetNATSNotifyByID(accountID string) natsNotify {
+func (s serverConfigV11) GetNATSNotifyByID(accountID string) natsNotify {
 	serverConfigMu.RLock()
 	defer serverConfigMu.RUnlock()
 
 	return s.Notify.NATS[accountID]
 }
 
-func (s *serverConfigV10) SetElasticSearchNotifyByID(accountID string, esNotify elasticSearchNotify) {
+func (s *serverConfigV11) SetElasticSearchNotifyByID(accountID string, esNotify elasticSearchNotify) {
 	serverConfigMu.Lock()
 	defer serverConfigMu.Unlock()
 
 	s.Notify.ElasticSearch[accountID] = esNotify
 }
 
-func (s serverConfigV10) GetElasticSearch() map[string]elasticSearchNotify {
+func (s serverConfigV11) GetElasticSearch() map[string]elasticSearchNotify {
 	serverConfigMu.RLock()
 	defer serverConfigMu.RUnlock()
 
@@ -184,21 +187,21 @@ func (s serverConfigV10) GetElasticSearch() map[string]elasticSearchNotify {
 }
 
 // GetElasticSearchNotify get current ElasicSearch logger.
-func (s serverConfigV10) GetElasticSearchNotifyByID(accountID string) elasticSearchNotify {
+func (s serverConfigV11) GetElasticSearchNotifyByID(accountID string) elasticSearchNotify {
 	serverConfigMu.RLock()
 	defer serverConfigMu.RUnlock()
 
 	return s.Notify.ElasticSearch[accountID]
 }
 
-func (s *serverConfigV10) SetRedisNotifyByID(accountID string, rNotify redisNotify) {
+func (s *serverConfigV11) SetRedisNotifyByID(accountID string, rNotify redisNotify) {
 	serverConfigMu.Lock()
 	defer serverConfigMu.Unlock()
 
 	s.Notify.Redis[accountID] = rNotify
 }
 
-func (s serverConfigV10) GetRedis() map[string]redisNotify {
+func (s serverConfigV11) GetRedis() map[string]redisNotify {
 	serverConfigMu.RLock()
 	defer serverConfigMu.RUnlock()
 
@@ -206,36 +209,58 @@ func (s serverConfigV10) GetRedis() map[string]redisNotify {
 }
 
 // GetRedisNotify get current Redis logger.
-func (s serverConfigV10) GetRedisNotifyByID(accountID string) redisNotify {
+func (s serverConfigV11) GetRedisNotifyByID(accountID string) redisNotify {
 	serverConfigMu.RLock()
 	defer serverConfigMu.RUnlock()
 
 	return s.Notify.Redis[accountID]
 }
 
-func (s *serverConfigV10) SetPostgreSQLNotifyByID(accountID string, pgn postgreSQLNotify) {
+func (s *serverConfigV11) SetPostgreSQLNotifyByID(accountID string, pgn postgreSQLNotify) {
 	serverConfigMu.Lock()
 	defer serverConfigMu.Unlock()
 
 	s.Notify.PostgreSQL[accountID] = pgn
 }
 
-func (s serverConfigV10) GetPostgreSQL() map[string]postgreSQLNotify {
+func (s serverConfigV11) GetPostgreSQL() map[string]postgreSQLNotify {
 	serverConfigMu.RLock()
 	defer serverConfigMu.RUnlock()
 
 	return s.Notify.PostgreSQL
 }
 
-func (s serverConfigV10) GetPostgreSQLNotifyByID(accountID string) postgreSQLNotify {
+func (s serverConfigV11) GetPostgreSQLNotifyByID(accountID string) postgreSQLNotify {
 	serverConfigMu.RLock()
 	defer serverConfigMu.RUnlock()
 
 	return s.Notify.PostgreSQL[accountID]
 }
 
+// Kafka related functions
+func (s *serverConfigV11) SetKafkaNotifyByID(accountID string, kn kafkaNotify) {
+	serverConfigMu.Lock()
+	defer serverConfigMu.Unlock()
+
+	s.Notify.Kafka[accountID] = kn
+}
+
+func (s serverConfigV11) GetKafka() map[string]kafkaNotify {
+	serverConfigMu.RLock()
+	defer serverConfigMu.RUnlock()
+
+	return s.Notify.Kafka
+}
+
+func (s serverConfigV11) GetKafkaNotifyByID(accountID string) kafkaNotify {
+	serverConfigMu.RLock()
+	defer serverConfigMu.RUnlock()
+
+	return s.Notify.Kafka[accountID]
+}
+
 // SetFileLogger set new file logger.
-func (s *serverConfigV10) SetFileLogger(flogger fileLogger) {
+func (s *serverConfigV11) SetFileLogger(flogger fileLogger) {
 	serverConfigMu.Lock()
 	defer serverConfigMu.Unlock()
 
@@ -243,7 +268,7 @@ func (s *serverConfigV10) SetFileLogger(flogger fileLogger) {
 }
 
 // GetFileLogger get current file logger.
-func (s serverConfigV10) GetFileLogger() fileLogger {
+func (s serverConfigV11) GetFileLogger() fileLogger {
 	serverConfigMu.RLock()
 	defer serverConfigMu.RUnlock()
 
@@ -251,7 +276,7 @@ func (s serverConfigV10) GetFileLogger() fileLogger {
 }
 
 // SetConsoleLogger set new console logger.
-func (s *serverConfigV10) SetConsoleLogger(clogger consoleLogger) {
+func (s *serverConfigV11) SetConsoleLogger(clogger consoleLogger) {
 	serverConfigMu.Lock()
 	defer serverConfigMu.Unlock()
 
@@ -259,7 +284,7 @@ func (s *serverConfigV10) SetConsoleLogger(clogger consoleLogger) {
 }
 
 // GetConsoleLogger get current console logger.
-func (s serverConfigV10) GetConsoleLogger() consoleLogger {
+func (s serverConfigV11) GetConsoleLogger() consoleLogger {
 	serverConfigMu.RLock()
 	defer serverConfigMu.RUnlock()
 
@@ -267,7 +292,7 @@ func (s serverConfigV10) GetConsoleLogger() consoleLogger {
 }
 
 // SetRegion set new region.
-func (s *serverConfigV10) SetRegion(region string) {
+func (s *serverConfigV11) SetRegion(region string) {
 	serverConfigMu.Lock()
 	defer serverConfigMu.Unlock()
 
@@ -275,7 +300,7 @@ func (s *serverConfigV10) SetRegion(region string) {
 }
 
 // GetRegion get current region.
-func (s serverConfigV10) GetRegion() string {
+func (s serverConfigV11) GetRegion() string {
 	serverConfigMu.RLock()
 	defer serverConfigMu.RUnlock()
 
@@ -283,7 +308,7 @@ func (s serverConfigV10) GetRegion() string {
 }
 
 // SetCredentials set new credentials.
-func (s *serverConfigV10) SetCredential(creds credential) {
+func (s *serverConfigV11) SetCredential(creds credential) {
 	serverConfigMu.Lock()
 	defer serverConfigMu.Unlock()
 
@@ -291,7 +316,7 @@ func (s *serverConfigV10) SetCredential(creds credential) {
 }
 
 // GetCredentials get current credentials.
-func (s serverConfigV10) GetCredential() credential {
+func (s serverConfigV11) GetCredential() credential {
 	serverConfigMu.RLock()
 	defer serverConfigMu.RUnlock()
 
@@ -299,7 +324,7 @@ func (s serverConfigV10) GetCredential() credential {
 }
 
 // Save config.
-func (s serverConfigV10) Save() error {
+func (s serverConfigV11) Save() error {
 	serverConfigMu.RLock()
 	defer serverConfigMu.RUnlock()
 

cmd/config-v11_test.go

@@ -60,6 +60,13 @@ func TestServerConfig(t *testing.T) {
 		t.Errorf("Expecting Redis config %#v found %#v", redisNotify{}, savedNotifyCfg3)
 	}
 
+	// Set new kafka notification id.
+	serverConfig.SetKafkaNotifyByID("2", kafkaNotify{})
+	savedNotifyCfg4 := serverConfig.GetKafkaNotifyByID("2")
+	if !reflect.DeepEqual(savedNotifyCfg4, kafkaNotify{}) {
+		t.Errorf("Expecting Kafka config %#v found %#v", kafkaNotify{}, savedNotifyCfg4)
+	}
+
 	// Set new console logger.
 	serverConfig.SetConsoleLogger(consoleLogger{
 		Enable: true,

cmd/event-notifier.go

@@ -659,6 +659,31 @@ func loadAllQueueTargets() (map[string]*logrus.Logger, error) {
 		}
 		queueTargets[queueARN] = pgLog
 	}
+	// Load Kafka targets, initialize their respective loggers.
+	for accountID, kafkaN := range serverConfig.GetKafka() {
+		if !kafkaN.Enable {
+			continue
+		}
+		// Construct the queue ARN for Kafka.
+		queueARN := minioSqs + serverConfig.GetRegion() + ":" + accountID + ":" + queueTypeKafka
+		_, ok := queueTargets[queueARN]
+		if ok {
+			continue
+		}
+		// Using accountID initialize a new Kafka logrus instance.
+		kafkaLog, err := newKafkaNotify(accountID)
+		if err != nil {
+			// Encapsulate network error to be more informative.
+			if _, ok := err.(net.Error); ok {
+				return nil, &net.OpError{
+					Op: "Connecting to " + queueARN, Net: "tcp",
+					Err: err,
+				}
+			}
+			return nil, err
+		}
+		queueTargets[queueARN] = kafkaLog
+	}
 
 	// Successfully initialized queue targets.
 	return queueTargets, nil

cmd/globals.go

@@ -36,7 +36,7 @@ const (
 
 // minio configuration related constants.
 const (
-	globalMinioConfigVersion      = "10"
+	globalMinioConfigVersion      = "11"
 	globalMinioConfigDir          = ".minio"
 	globalMinioCertsDir           = "certs"
 	globalMinioCertsCADir         = "CAs"

cmd/notifiers.go

@@ -38,6 +38,8 @@ const (
 	queueTypeRedis = "redis"
 	// Static string indicating queue type 'postgresql'.
 	queueTypePostgreSQL = "postgresql"
+	// Static string indicating queue type 'kafka'.
+	queueTypeKafka = "kafka"
 )
 
 // Topic type.
@@ -58,6 +60,7 @@ type notifier struct {
 	ElasticSearch map[string]elasticSearchNotify `json:"elasticsearch"`
 	Redis         map[string]redisNotify         `json:"redis"`
 	PostgreSQL    map[string]postgreSQLNotify    `json:"postgresql"`
+	Kafka         map[string]kafkaNotify         `json:"kafka"`
 	// Add new notification queues.
 }
 
@@ -154,6 +157,24 @@ func isPostgreSQLQueue(sqsArn arnSQS) bool {
 	return true
 }
 
+// Returns true if queueArn is for Kafka.
+func isKafkaQueue(sqsArn arnSQS) bool {
+	if sqsArn.Type != queueTypeKafka {
+		return false
+	}
+	kafkaNotifyCfg := serverConfig.GetKafkaNotifyByID(sqsArn.AccountID)
+	if !kafkaNotifyCfg.Enable {
+		return false
+	}
+	kafkaC, err := dialKafka(kafkaNotifyCfg)
+	if err != nil {
+		errorIf(err, "Unable to dial Kafka server %#v", kafkaNotifyCfg)
+		return false
+	}
+	defer kafkaC.Close()
+	return true
+}
+
 // Match function matches wild cards in 'pattern' for events.
 func eventMatch(eventType string, events []string) (ok bool) {
 	for _, event := range events {

cmd/notify-kafka.go

@@ -0,0 +1,125 @@
+/*
+ * Minio Cloud Storage, (C) 2014-2016 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.
+ */
+
+package cmd
+
+import (
+	"fmt"
+	"io/ioutil"
+
+	"github.com/Sirupsen/logrus"
+
+	sarama "gopkg.in/Shopify/sarama.v1"
+)
+
+// kafkaNotify holds the configuration of the Kafka server/cluster to
+// send notifications to.
+type kafkaNotify struct {
+	// Flag to enable/disable this notification from configuration
+	// file.
+	Enable bool `json:"enable"`
+
+	// List of Kafka brokers in `addr:host` format.
+	Brokers []string `json:"brokers"`
+
+	// Topic to which event notifications should be sent.
+	Topic string `json:"topic"`
+}
+
+// kafkaConn contains the active connection to the Kafka cluster and
+// the topic to send event notifications to.
+type kafkaConn struct {
+	producer sarama.SyncProducer
+	topic    string
+}
+
+func dialKafka(kn kafkaNotify) (kafkaConn, error) {
+	if !kn.Enable {
+		return kafkaConn{}, errNotifyNotEnabled
+	}
+
+	if kn.Topic == "" {
+		return kafkaConn{}, fmt.Errorf(
+			"Kafka Notifier Error: Topic was not specified in configuration")
+	}
+
+	config := sarama.NewConfig()
+	// Wait for all in-sync replicas to ack the message
+	config.Producer.RequiredAcks = sarama.WaitForAll
+	// Retry up to 10 times to produce the message
+	config.Producer.Retry.Max = 10
+	config.Producer.Return.Successes = true
+
+	p, err := sarama.NewSyncProducer(kn.Brokers, config)
+	if err != nil {
+		return kafkaConn{}, fmt.Errorf(
+			"Kafka Notifier Error: Failed to start producer: %v",
+			err,
+		)
+	}
+
+	return kafkaConn{p, kn.Topic}, nil
+}
+
+func newKafkaNotify(accountID string) (*logrus.Logger, error) {
+	kafkaNotifyCfg := serverConfig.GetKafkaNotifyByID(accountID)
+
+	// Try connecting to the configured Kafka broker(s).
+	kc, err := dialKafka(kafkaNotifyCfg)
+	if err != nil {
+		return nil, err
+	}
+
+	// Configure kafkaConn object as a Hook in logrus.
+	kafkaLog := logrus.New()
+	kafkaLog.Out = ioutil.Discard
+	kafkaLog.Formatter = new(logrus.JSONFormatter)
+	kafkaLog.Hooks.Add(kc)
+
+	return kafkaLog, nil
+}
+
+func (kC kafkaConn) Close() {
+	_ = kC.producer.Close()
+}
+
+// Fire - to implement logrus.Hook interface
+func (kC kafkaConn) Fire(entry *logrus.Entry) error {
+	body, err := entry.Reader()
+	if err != nil {
+		return err
+	}
+
+	// Construct message to send to Kafka
+	msg := sarama.ProducerMessage{
+		Topic: kC.topic,
+		Value: sarama.ByteEncoder(body.Bytes()),
+	}
+
+	// Attempt sending the message to Kafka
+	_, _, err = kC.producer.SendMessage(&msg)
+	if err != nil {
+		return fmt.Errorf("Error sending event to Kafka - %v", err)
+	}
+	return nil
+}
+
+// Levels - to implement logrus.Hook interface
+func (kC kafkaConn) Levels() []logrus.Level {
+	return []logrus.Level{
+		logrus.InfoLevel,
+	}
+}