rgw: infrastructure to send requests to different region

Signed-off-by: Yehuda Sadeh <yehuda@inktank.com>

8 files changed, 234 insertions, 2 deletions

diff --git a/src/Makefile.am b/src/Makefile.am
index dc11c9a5df8..ecde5c11320 100644
--- a/src/Makefile.am
+++ b/src/Makefile.am
@@ -382,6 +382,9 @@ librgw_a_SOURCES =  \
 	rgw/rgw_bucket.cc\
         rgw/rgw_tools.cc \
 	rgw/rgw_rados.cc \
+        rgw/rgw_http_client.cc \
+        rgw/rgw_rest_client.cc \
+	rgw/rgw_rest_conn.cc \
 	rgw/rgw_op.cc \
 	rgw/rgw_common.cc \
 	rgw/rgw_cache.cc \
@@ -413,7 +416,6 @@ radosgw_SOURCES = \
         rgw/rgw_rest_user.cc \
         rgw/rgw_rest_bucket.cc \
         rgw/rgw_rest_metadata.cc \
-        rgw/rgw_http_client.cc \
         rgw/rgw_swift.cc \
 	rgw/rgw_swift_auth.cc \
 	rgw/rgw_main.cc
@@ -2077,6 +2079,8 @@ noinst_HEADERS = \
 	rgw/rgw_rest_usage.h\
 	rgw/rgw_rest_user.h\
 	rgw/rgw_rest_bucket.h\
+	rgw/rgw_rest_client.h\
+	rgw/rgw_rest_conn.h\
 	rgw/rgw_tools.h\
 	rgw/rgw_rest_metadata.h\
 	rgw/rgw_usage.h\
diff --git a/src/rgw/rgw_http_client.h b/src/rgw/rgw_http_client.h
index 944ea89e3f3..e11704fdf9e 100644
--- a/src/rgw/rgw_http_client.h
+++ b/src/rgw/rgw_http_client.h
@@ -5,6 +5,7 @@
 
 class RGWHTTPClient
 {
+protected:
   list<pair<string, string> > headers;
 public:
   virtual ~RGWHTTPClient() {}
diff --git a/src/rgw/rgw_rados.cc b/src/rgw/rgw_rados.cc
index f48923d2681..15408818825 100644
--- a/src/rgw/rgw_rados.cc
+++ b/src/rgw/rgw_rados.cc
@@ -548,6 +548,19 @@ int RGWRados::init_complete()
   if (ret < 0)
     return ret;
 
+  ret = region_map.read(cct, this);
+  if (ret < 0) {
+    ldout(cct, 0) << "WARNING: cannot read region map" << dendl;
+  } else {
+    string master_region = region_map.master_region;
+    map<string, RGWRegion>::iterator iter = region_map.regions.find(master_region);
+    if (iter == region_map.regions.end()) {
+      lderr(cct) << "ERROR: bad region map: inconsistent master region" << dendl;
+      return -EINVAL;
+    }
+    rest_conn = new RGWRegionConnection(cct, this, iter->second);
+  }
+
   ret = open_root_pool_ctx();
   if (ret < 0)
     return ret;
diff --git a/src/rgw/rgw_rados.h b/src/rgw/rgw_rados.h
index d4733aef842..50e450f63e8 100644
--- a/src/rgw/rgw_rados.h
+++ b/src/rgw/rgw_rados.h
@@ -10,6 +10,7 @@
 #include "cls/log/cls_log_types.h"
 #include "rgw_log.h"
 #include "rgw_metadata.h"
+#include "rgw_rest_conn.h"
 
 class RGWWatcher;
 class SafeTimer;
@@ -537,7 +538,9 @@ public:
                num_watchers(0), watchers(NULL), watch_handles(NULL),
                bucket_id_lock("rados_bucket_id"), max_bucket_id(0),
                cct(NULL), rados(NULL),
-               pools_initialized(false), meta_mgr(NULL), data_log(NULL) {}
+               pools_initialized(false),
+               rest_conn(NULL),
+               meta_mgr(NULL), data_log(NULL) {}
 
   void set_context(CephContext *_cct) {
     cct = _cct;
@@ -553,6 +556,8 @@ public:
 
   RGWRegion region;
   RGWZoneParams zone;
+  RGWRegionMap region_map;
+  RGWRegionConnection *rest_conn;
 
   RGWMetadataManager *meta_mgr;
 
diff --git a/src/rgw/rgw_rest_client.cc b/src/rgw/rgw_rest_client.cc
new file mode 100644
index 00000000000..0d896652277
--- /dev/null
+++ b/src/rgw/rgw_rest_client.cc
@@ -0,0 +1,109 @@
+#include "rgw_common.h"
+#include "rgw_rest_client.h"
+
+#include "common/ceph_crypto_cms.h"
+#include "common/armor.h"
+
+#define dout_subsys ceph_subsys_rgw
+
+int RGWRESTClient::read_header(void *ptr, size_t len)
+{
+  char line[len + 1];
+
+  char *s = (char *)ptr, *end = (char *)ptr + len;
+  char *p = line;
+  ldout(cct, 10) << "read_http_header" << dendl;
+
+  while (s != end) {
+    if (*s == '\r') {
+      s++;
+      continue;
+    }
+    if (*s == '\n') {
+      *p = '\0';
+      ldout(cct, 10) << "os_auth:" << line << dendl;
+      // TODO: fill whatever data required here
+      char *l = line;
+      char *tok = strsep(&l, " \t:");
+      if (tok) {
+        while (l && *l == ' ')
+          l++;
+ 
+        if (strcmp(tok, "HTTP") == 0) {
+          status = atoi(l);
+        } else {
+          /* convert header field name to upper case  */
+          char *src = tok;
+          char buf[len + 1];
+          size_t i;
+          for (i = 0; i < len && *src; ++i, ++src) {
+            buf[i] = toupper(*src);
+          }
+          buf[i] = '\0';
+          out_headers[buf] = l;
+        }
+      }
+    }
+    if (s != end)
+      *p++ = *s++;
+  }
+  return 0;
+}
+
+int RGWRESTClient::execute(RGWAccessKey& key, const string& method, const string& resource)
+{
+  string new_url = url;
+  string new_resource = resource;
+
+  if (new_url[new_url.size() - 1] == '/' && resource[0] == '/') {
+    new_url = new_url.substr(0, new_url.size() - 1);
+  } else if (resource[0] != '/') {
+    new_resource = "/";
+    new_resource.append(resource);
+  }
+  new_url.append(new_resource);
+
+  if (params.size()) {
+    new_url.append("?");
+
+    list<pair<string, string> >::iterator iter;
+    for (iter = params.begin(); iter != params.end(); ++iter) {
+      new_url.append(iter->first + "=" + iter->second);
+    }
+  }
+
+  utime_t tm = ceph_clock_now(cct);
+  stringstream s;
+  tm.gmtime(s);
+  string date_str = s.str();
+  headers.push_back(make_pair<string, string>("HTTP_DATE", date_str));
+
+  string canonical_header = method + " " +
+                            "\n" + /* CONTENT_MD5 */
+                            "\n" + /* CONTENT_TYPE */
+                            date_str + "\n" +
+                            "\n" + /* amz headers */
+                            new_resource;
+
+  string& k = key.key;
+  
+  char hmac_sha1[CEPH_CRYPTO_HMACSHA1_DIGESTSIZE];
+  calc_hmac_sha1(k.c_str(), k.size(), canonical_header.c_str(), canonical_header.size(), hmac_sha1);
+
+#define ARMOR_LEN 64
+  char b64[ARMOR_LEN]; /* 64 is really enough */
+  int ret = ceph_armor(b64, b64 + ARMOR_LEN, hmac_sha1,
+		       hmac_sha1 + CEPH_CRYPTO_HMACSHA1_DIGESTSIZE);
+  if (ret < 0) {
+    dout(10) << "ceph_armor failed" << dendl;
+    return -EPERM;
+  }
+  b64[ret] = '\0';
+
+  string auth_hdr = "AWS " + key.id + ":" + b64;
+
+  headers.push_back(make_pair<string, string>("AUTHORIZATION", auth_hdr));
+  return process(new_url);
+}
+
+
diff --git a/src/rgw/rgw_rest_client.h b/src/rgw/rgw_rest_client.h
new file mode 100644
index 00000000000..f885a223d84
--- /dev/null
+++ b/src/rgw/rgw_rest_client.h
@@ -0,0 +1,37 @@
+#ifndef CEPH_RGW_REST_CLIENT_H
+#define CEPH_RGW_REST_CLIENT_H
+
+#include <list>
+
+#include "rgw_http_client.h"
+
+class RGWRESTClient : public RGWHTTPClient {
+  CephContext *cct;
+
+protected:
+  int status;
+
+  string url;
+
+  map<string, string> out_headers;
+  list<pair<string, string> > params;
+
+  RGWRESTClient() : cct(NULL), status(0) {}
+public:
+  RGWRESTClient(CephContext *_cct, string& _url,
+                list<pair<string, string> > *_headers, list<pair<string, string> > *_params) : cct(_cct), url(_url) {
+    if (_headers)
+      headers = *_headers;
+
+    if (_params)
+      params = *_params;
+  }
+
+  int read_header(void *ptr, size_t len);
+
+  int execute(RGWAccessKey& key, const string& method, const string& resource);
+};
+
+
+#endif
+
diff --git a/src/rgw/rgw_rest_conn.cc b/src/rgw/rgw_rest_conn.cc
new file mode 100644
index 00000000000..274939a38b8
--- /dev/null
+++ b/src/rgw/rgw_rest_conn.cc
@@ -0,0 +1,38 @@
+#include "rgw_rados.h"
+#include "rgw_rest_conn.h"
+
+#define dout_subsys ceph_subsys_rgw
+
+RGWRegionConnection::RGWRegionConnection(CephContext *_cct, RGWRados *store, RGWRegion& upstream) : cct(_cct) {
+  list<string>::iterator iter;
+  int i;
+  for (i = 0, iter = upstream.endpoints.begin(); iter != upstream.endpoints.end(); ++iter, ++i) {
+    endpoints[i] = *iter;
+  }
+  key = store->zone.system_key;
+}
+
+int RGWRegionConnection::get_url(string& endpoint)
+{
+  if (endpoints.empty()) {
+    ldout(cct, 0) << "ERROR: endpoints not configured for upstream zone" << dendl;
+    return -EIO;
+  }
+
+  int i = counter.inc();
+  endpoint = endpoints[i % endpoints.size()];
+
+  return 0;
+}
+
+int RGWRegionConnection::create_bucket(const string& uid, const string& bucket) {
+  list<pair<string, string> > params;
+  params.push_back(make_pair<string, string>("uid", uid));
+  params.push_back(make_pair<string, string>("bucket", bucket));
+  string url;
+  int ret = get_url(url);
+  if (ret < 0)
+    return ret;
+  RGWRESTClient client(cct, url, NULL, &params);
+  return client.execute(key, "PUT", "/admin/bucket");
+}
diff --git a/src/rgw/rgw_rest_conn.h b/src/rgw/rgw_rest_conn.h
new file mode 100644
index 00000000000..0ddf4a15973
--- /dev/null
+++ b/src/rgw/rgw_rest_conn.h
@@ -0,0 +1,25 @@
+#ifndef CEPH_RGW_REST_REQ_H
+#define CEPH_RGW_REST_REQ_H
+
+#include "rgw_rest_client.h"
+
+class CephContext;
+class RGWRados;
+class RGWRegion;
+
+class RGWRegionConnection
+{
+  CephContext *cct;
+  map<int, string> endpoints;
+  RGWAccessKey key;
+  atomic_t counter;
+public:
+
+  RGWRegionConnection(CephContext *_cct, RGWRados *store, RGWRegion& upstream);
+  int get_url(string& endpoint);
+
+  int create_bucket(const string& uid, const string& bucket);
+
+};
+
+#endif