[syslog-ng] [PATCH (3.4) 3/5] afmongodb: Added replicaset support.

[Gergely Nagy]

The servers can now be specified with a server() parameter, where the
first will be the primary (used in the stats for example), the rest
will be the seeds.

Signed-off-by: Gergely Nagy <algernon at balabit.hu>
---
 modules/afmongodb/afmongodb-grammar.ym |    4 +-
 modules/afmongodb/afmongodb-parser.c   |    4 +--
 modules/afmongodb/afmongodb.c          |   53 ++++++++++++++++++++++++--------
 modules/afmongodb/afmongodb.h          |    3 +-
 4 files changed, 44 insertions(+), 20 deletions(-)

diff --git a/modules/afmongodb/afmongodb-grammar.ym b/modules/afmongodb/afmongodb-grammar.ym
index 36d69c4..1a929ac 100644
--- a/modules/afmongodb/afmongodb-grammar.ym
+++ b/modules/afmongodb/afmongodb-grammar.ym
@@ -50,6 +50,7 @@ extern ValuePairsTransformSet *last_vp_transset;
 
 %token KW_MONGODB
 %token KW_COLLECTION
+%token KW_SERVERS
 
 %%
 
@@ -67,8 +68,7 @@ afmongodb_options
 	;
 
 afmongodb_option
-        : KW_HOST '(' string ')'		{ afmongodb_dd_set_host(last_driver, $3); free($3); }
-        | KW_PORT '(' LL_NUMBER ')'		{ afmongodb_dd_set_port(last_driver, $3); }
+	: KW_SERVERS '(' string_list ')'	{ afmongodb_dd_set_servers(last_driver, $3); }
 	| KW_DATABASE '(' string ')'		{ afmongodb_dd_set_database(last_driver, $3); free($3); }
 	| KW_COLLECTION '(' string ')'		{ afmongodb_dd_set_collection(last_driver, $3); free($3); }
 	| KW_USERNAME '(' string ')'		{ afmongodb_dd_set_user(last_driver, $3); free($3); }
diff --git a/modules/afmongodb/afmongodb-parser.c b/modules/afmongodb/afmongodb-parser.c
index 67a9b0b..b589fe8 100644
--- a/modules/afmongodb/afmongodb-parser.c
+++ b/modules/afmongodb/afmongodb-parser.c
@@ -30,13 +30,11 @@ int afmongodb_parse(CfgLexer *lexer, LogDriver **instance, gpointer arg);
 
 static CfgLexerKeyword afmongodb_keywords[] = {
   { "mongodb",			KW_MONGODB },
-  { "host",			KW_HOST },
-  { "port",			KW_PORT },
+  { "servers",                  KW_SERVERS },
   { "database",			KW_DATABASE },
   { "collection",		KW_COLLECTION },
   { "username",			KW_USERNAME },
   { "password",			KW_PASSWORD },
-  { "log_fifo_size",		KW_LOG_FIFO_SIZE  },
   { NULL }
 };
 
diff --git a/modules/afmongodb/afmongodb.c b/modules/afmongodb/afmongodb.c
index 48df68f..01d5dc8 100644
--- a/modules/afmongodb/afmongodb.c
+++ b/modules/afmongodb/afmongodb.c
@@ -49,6 +49,7 @@ typedef struct
   gchar *db;
   gchar *coll;
 
+  GList *servers;
   gchar *host;
   gint port;
 
@@ -109,20 +110,12 @@ afmongodb_dd_set_password(LogDriver *d, const gchar *password)
 }
 
 void
-afmongodb_dd_set_host(LogDriver *d, const gchar *host)
+afmongodb_dd_set_servers(LogDriver *d, GList *servers)
 {
   MongoDBDestDriver *self = (MongoDBDestDriver *)d;
 
-  g_free(self->host);
-  self->host = g_strdup (host);
-}
-
-void
-afmongodb_dd_set_port(LogDriver *d, gint port)
-{
-  MongoDBDestDriver *self = (MongoDBDestDriver *)d;
-
-  self->port = (int)port;
+  string_list_free(self->servers);
+  self->servers = servers;
 }
 
 void
@@ -196,6 +189,8 @@ afmongodb_dd_disconnect(MongoDBDestDriver *self)
 static gboolean
 afmongodb_dd_connect(MongoDBDestDriver *self, gboolean reconnect)
 {
+  GList *l;
+
   if (reconnect && self->conn)
     return TRUE;
 
@@ -207,6 +202,27 @@ afmongodb_dd_connect(MongoDBDestDriver *self, gboolean reconnect)
       return FALSE;
     }
 
+  l = self->servers;
+  while ((l = g_list_next(l)) != NULL)
+    {
+      gchar *host = NULL;
+      gint port = 27017;
+
+      if (!mongo_util_parse_addr(l->data, &host, &port))
+	{
+	  msg_warning("Cannot parse MongoDB server address, ignoring",
+		      evt_tag_str("address", l->data),
+		      NULL);
+	  continue;
+	}
+      mongo_sync_conn_seed_add (self->conn, host, port);
+      msg_verbose("Added MongoDB server seed",
+		  evt_tag_str("host", host),
+		  evt_tag_int("port", port),
+		  NULL);
+      g_free(host);
+    }
+
   /*
   if (self->user || self->password)
     {
@@ -421,6 +437,17 @@ afmongodb_dd_init(LogPipe *s)
       value_pairs_add_scope(self->vp, "nv-pairs");
     }
 
+  self->host = NULL;
+  self->port = 27017;
+  if (!mongo_util_parse_addr(g_list_nth_data(self->servers, 0), &self->host,
+			     &self->port))
+    {
+      msg_error("Cannot parse the primary host",
+		evt_tag_str("primary", g_list_nth_data(self->servers, 0)),
+		NULL);
+      return FALSE;
+    }
+
   msg_verbose("Initializing MongoDB destination",
 	      evt_tag_str("host", self->host),
 	      evt_tag_int("port", self->port),
@@ -484,6 +511,7 @@ afmongodb_dd_free(LogPipe *d)
   g_free(self->user);
   g_free(self->password);
   g_free(self->host);
+  string_list_free(self->servers);
   value_pairs_free(self->vp);
   log_dest_driver_free(d);
 }
@@ -541,8 +569,7 @@ afmongodb_dd_new(void)
   self->super.super.super.queue = afmongodb_dd_queue;
   self->super.super.super.free_fn = afmongodb_dd_free;
 
-  afmongodb_dd_set_host((LogDriver *)self, "127.0.0.1");
-  afmongodb_dd_set_port((LogDriver *)self, 27017);
+  afmongodb_dd_set_servers((LogDriver *)self, g_list_append (NULL, g_strdup ("127.0.0.1:27017")));
   afmongodb_dd_set_database((LogDriver *)self, "syslog");
   afmongodb_dd_set_collection((LogDriver *)self, "messages");
 
diff --git a/modules/afmongodb/afmongodb.h b/modules/afmongodb/afmongodb.h
index bd7b258..4aa47ef 100644
--- a/modules/afmongodb/afmongodb.h
+++ b/modules/afmongodb/afmongodb.h
@@ -29,8 +29,7 @@
 
 LogDriver *afmongodb_dd_new(void);
 
-void afmongodb_dd_set_host(LogDriver *d, const gchar *host);
-void afmongodb_dd_set_port(LogDriver *d, gint port);
+void afmongodb_dd_set_servers(LogDriver *d, GList *servers);
 void afmongodb_dd_set_database(LogDriver *d, const gchar *database);
 void afmongodb_dd_set_collection(LogDriver *d, const gchar *collection);
 void afmongodb_dd_set_user(LogDriver *d, const gchar *user);
-- 
1.7.7.3