#250: set ceph client options. define client settings with prefix: rb…

…ox_ceph_client, e.g. rbox_ceph_client_client_oc=true in 90-plugin.conf

src/librmb/rados-cluster-impl.cpp

@@ -60,6 +60,7 @@ int RadosClusterImpl::init(const std::string &clustername, const std::string &ra
   int ret = 0;
   if (RadosClusterImpl::cluster_ref_count == 0) {
     RadosClusterImpl::cluster = new librados::Rados();
+
     ret = RadosClusterImpl::cluster->init2(rados_username.c_str(), clustername.c_str(), 0);
     if (ret == 0) {
       ret = initialize();
@@ -94,6 +95,9 @@ int RadosClusterImpl::initialize() {
     RadosClusterImpl::cluster->conf_set(RADOS_OSD_OP_TIMEOUT, RADOS_OSD_OP_TIMEOUT_DEFAULT);
   }
 
+  for (std::map<const char *, const char *>::iterator it = client_options.begin(); it != client_options.end(); ++it) {
+    RadosClusterImpl::cluster->conf_set(it->first, it->second);
+  }
   return ret;
 }
 
@@ -115,6 +119,7 @@ void RadosClusterImpl::deinit() {
         RadosClusterImpl::cluster->shutdown();
         RadosClusterImpl::connected = false;
         delete RadosClusterImpl::cluster;
+        RadosClusterImpl::cluster = nullptr;
       }
     }
   }
@@ -175,3 +180,5 @@ int RadosClusterImpl::io_ctx_create(const string &pool, librados::IoCtx *io_ctx)
 int RadosClusterImpl::get_config_option(const char *option, string *value) {
   return RadosClusterImpl::cluster->conf_get(option, *value);
 }
+
+void RadosClusterImpl::set_config_option(const char *option, const char *value) { client_options[option] = value; }

src/librmb/rados-cluster-impl.h

@@ -16,7 +16,7 @@
 #include <string>
 
 #include <rados/librados.hpp>
-
+#include <map>
 #include "rados-cluster.h"
 namespace librmb {
 
@@ -38,6 +38,7 @@ class RadosClusterImpl : public RadosCluster {
                         RadosDictionary **dictionary);
   bool is_connected() override;
   librados::Rados &get_cluster() { return *cluster; }
+  void set_config_option(const char *option, const char *value);
 
  private:
   int initialize();
@@ -46,6 +47,7 @@ class RadosClusterImpl : public RadosCluster {
   static librados::Rados *cluster;
   static int cluster_ref_count;
   static bool connected;
+  std::map<const char *, const char *> client_options;
 
   static const char *CLIENT_MOUNT_TIMEOUT;
   static const char *RADOS_MON_OP_TIMEOUT;

src/librmb/rados-cluster.h

@@ -61,6 +61,16 @@ class RadosCluster {
    *
    */
   virtual int get_config_option(const char *option, std::string *value) = 0;
+
+  /*!
+     * set ceph configuration
+     * @param[in] option option name as described in the ceph documentation
+     * @param[out] value valid ptr to a string buffer.
+     * @return linux error code or 0 if successful
+     *
+     */
+  virtual void set_config_option(const char *option, const char *value) = 0;
+
   /*!
    * Check if cluster connection does exist and is working-
    * @return true if connected

src/storage-rbox/rbox-storage.cpp

@@ -385,6 +385,30 @@ static int rbox_open_mailbox(struct mailbox *box) {
   FUNC_END();
   return 0;
 }
+static void read_plugin_ceph_client_settings(struct mailbox *box, const char *prefix) {
+  struct rbox_storage *r_storage = (struct rbox_storage *)box->storage;
+  const char *const *envs;
+  unsigned int i, count;
+  char errstr[512];
+
+  if (!array_is_created(&r_storage->storage.user->set->plugin_envs)) {
+    return;
+  }
+
+  if (prefix == NULL) {
+    return;
+  }
+
+  envs = array_get(&r_storage->storage.user->set->plugin_envs, &count);
+  for (i = 0; i < count; i += 2) {
+    if (strlen(envs[i]) > strlen(prefix)) {
+      if (strncmp(envs[i], prefix, strlen(prefix)) == 0) {
+        const char *s = envs[i] + strlen(prefix) + 1;
+        r_storage->cluster->set_config_option(s, envs[i + 1]);
+      }
+    }
+  }
+}
 
 void read_plugin_configuration(struct mailbox *box) {
   FUNC_START();
@@ -410,8 +434,9 @@ void read_plugin_configuration(struct mailbox *box) {
 
   FUNC_END();
 }
-
-bool is_alternate_storage_set(uint8_t flags) { return (flags & RBOX_INDEX_FLAG_ALT) != 0; }
+read_plugin_ceph_client_settings bool is_alternate_storage_set(uint8_t flags) {
+  return (flags & RBOX_INDEX_FLAG_ALT) != 0;
+}
 
 bool is_alternate_pool_valid(struct mailbox *_box) {
   return _box->list->set.alt_dir != NULL && strlen(_box->list->set.alt_dir) > 0;
@@ -425,6 +450,10 @@ int rbox_open_rados_connection(struct mailbox *box, bool alt_storage) {
 
   // initialize storage with plugin configuration
   read_plugin_configuration(box);
+
+  // set the ceph client options!
+  read_plugin_ceph_client_settings(box, "rbox_ceph_client");
+
   int ret = 0;
   try {
     rados_storage->set_ceph_wait_method(rbox->storage->config->is_ceph_aio_wait_for_safe_and_cb()

src/tests/mocks/mock_test.h

@@ -130,6 +130,7 @@ class RadosClusterMock : public RadosCluster {
   MOCK_METHOD2(io_ctx_create, int(const std::string &pool, librados::IoCtx *io_ctx));
   MOCK_METHOD2(get_config_option, int(const char *option, std::string *value));
   MOCK_METHOD0(is_connected, bool());
+  MOCK_METHOD2(set_config_option, void(const char *option, const char *value));
 };
 
 using librmb::RadosDovecotCephCfg;