rgw: push log list|show|rm into RGWStore

Signed-off-by: Sage Weil <sage.weil@dreamhost.com>

src/rgw/rgw_access.h

@@ -40,6 +40,9 @@ public:
   
   virtual int log_list_init(const string& prefix, RGWAccessHandle *handle) { return -ENOENT; }
   virtual int log_list_next(RGWAccessHandle handle, string *name) { return -ENOENT; }
+  virtual int log_remove(const string& name) { return -ENOENT; }
+  virtual int log_show_init(const string& name, RGWAccessHandle *handle) { return -ENOENT; }
+  virtual int log_show_next(RGWAccessHandle handle, class rgw_log_entry *entry) { return -ENOENT; }
 
   /** 
    * get listing of the objects in a bucket.

src/rgw/rgw_admin.cc

@@ -952,7 +952,7 @@ int main(int argc, char **argv)
     formatter->reset();
     formatter->open_array_section("logs");
     RGWAccessHandle h;
-    int r = store->list_logs_init(date, &h);
+    int r = store->log_list_init(date, &h);
     if (r == -ENOENT) {
       // no logs.
     } else {
@@ -962,7 +962,7 @@ int main(int argc, char **argv)
       }
       while (true) {
 	string name;
-	int r = store->list_logs_next(h, &name);
+	int r = store->log_list_next(h, &name);
 	if (r == -ENOENT)
 	  break;
 	if (r < 0) {
@@ -982,7 +982,6 @@ int main(int argc, char **argv)
       return usage();
     }
 
-    rgw_bucket log_bucket(RGW_LOG_POOL_NAME);
     string oid;
     if (!object.empty()) {
       oid = object;
@@ -995,45 +994,33 @@ int main(int argc, char **argv)
       oid += "-";
       oid += string(bucket.name);
     }
-    rgw_obj obj(log_bucket, oid);
 
     if (opt_cmd == OPT_LOG_SHOW) {
-      uint64_t size;
-      int r = store->obj_stat(NULL, obj, &size, NULL);
+      RGWAccessHandle h;
+
+      int r = store->log_show_init(oid, &h);
       if (r < 0) {
-	cerr << "error while doing stat on " <<  log_bucket << ":" << oid
-	     << " " << cpp_strerror(-r) << std::endl;
+	cerr << "error opening log " << oid << ": " << cpp_strerror(-r) << std::endl;
 	return -r;
       }
 
-      bufferlist bl;
-      r = store->read(NULL, obj, 0, size, bl);
-      if (r < 0) {
-	cerr << "error while reading from " <<  log_bucket << ":" << oid
-	     << " " << cpp_strerror(-r) << std::endl;
-	return -r;
-      }
-      
-      bufferlist::iterator iter = bl.begin();
-      
-      struct rgw_log_entry entry;
-      
       formatter->reset();
       formatter->open_object_section("log");
 
-      // peek at first entry to get bucket metadata
-      bufferlist::iterator first_iter = iter;
-      if (!first_iter.end()) {
-	::decode(entry, first_iter);
-	formatter->dump_int("bucket_id", entry.bucket_id);
-	formatter->dump_string("bucket_owner", entry.bucket_owner);
-	formatter->dump_string("bucket", entry.bucket);
-      }
-      formatter->open_array_section("log_entries");
+      struct rgw_log_entry entry;
       
-      while (!iter.end()) {
-	::decode(entry, iter);
-	
+      // peek at first entry to get bucket metadata
+      r = store->log_show_next(h, &entry);
+      if (r < 0) {
+	cerr << "error reading log " << oid << ": " << cpp_strerror(-r) << std::endl;
+	return -r;
+      }
+      formatter->dump_int("bucket_id", entry.bucket_id);
+      formatter->dump_string("bucket_owner", entry.bucket_owner);
+      formatter->dump_string("bucket", entry.bucket);
+
+      formatter->open_array_section("log_entries");
+      do {
 	uint64_t total_time =  entry.total_time.sec() * 1000000LL * entry.total_time.usec();
 	
 	formatter->open_object_section("log_entry");
@@ -1056,17 +1043,23 @@ int main(int argc, char **argv)
 	formatter->dump_string("referrer",  entry.referrer.c_str());
 	formatter->close_section();
 	formatter->flush(cout);
+
+	r = store->log_show_next(h, &entry);
+      } while (r > 0);
+
+      if (r < 0) {
+      	cerr << "error reading log " << oid << ": " << cpp_strerror(-r) << std::endl;
+	return -r;
       }
+
       formatter->close_section();
       formatter->close_section();
       formatter->flush(cout);
     }
     if (opt_cmd == OPT_LOG_RM) {
-      std::string id;
-      int r = store->delete_obj(NULL, id, obj);
+      int r = store->log_remove(oid);
       if (r < 0) {
-	cerr << "error removing " <<  log_bucket << ":" << oid
-	     << " " << cpp_strerror(-r) << std::endl;
+	cerr << "error removing log " << oid << ": " << cpp_strerror(-r) << std::endl;
 	return -r;
       }
     }

src/rgw/rgw_rados.cc

@@ -209,6 +209,9 @@ int RGWRados::list_buckets_next(std::string& id, RGWObjEnt& obj, RGWAccessHandle
   return 0;
 }
 
+
+/**** logs ****/
+
 struct log_list_state {
   string prefix;
   librados::IoCtx io_ctx;
@@ -218,13 +221,12 @@ struct log_list_state {
 int RGWRados::log_list_init(const string& prefix, RGWAccessHandle *handle)
 {
   log_list_state *state = new log_list_state;
-  rgw_bucket log_bucket(RGW_LOG_POOL_NAME);
-  int r = open_bucket_ctx(log_bucket, state->io_ctx);
+  int r = rados->ioctx_create(RGW_LOG_POOL_NAME, state->io_ctx);
   if (r < 0)
     return r;
   state->prefix = prefix;
   state->obit = state->io_ctx.objects_begin();
-  *handle = (RGWAccessHandle*)state;
+  *handle = (RGWAccessHandle)state;
   return 0;
 }
 
@@ -248,6 +250,72 @@ int RGWRados::log_list_next(RGWAccessHandle handle, string *name)
   return 0;
 }
 
+int RGWRados::log_remove(const string& name)
+{
+  librados::IoCtx io_ctx;
+  int r = rados->ioctx_create(RGW_LOG_POOL_NAME, io_ctx);
+  if (r < 0)
+    return r;
+  return io_ctx.remove(name);
+}
+
+struct log_show_state {
+  librados::IoCtx io_ctx;
+  bufferlist bl;
+  bufferlist::iterator p;
+  string name;
+  uint64_t pos;
+  bool eof;
+  log_show_state() : pos(0), eof(false) {}
+};
+
+int RGWRados::log_show_init(const string& name, RGWAccessHandle *handle)
+{
+  log_show_state *state = new log_show_state;
+  int r = rados->ioctx_create(RGW_LOG_POOL_NAME, state->io_ctx);
+  if (r < 0)
+    return r;
+  state->name = name;
+  *handle = (RGWAccessHandle)state;
+  return 0;
+}
+
+int RGWRados::log_show_next(RGWAccessHandle handle, rgw_log_entry *entry)
+{
+  log_show_state *state = (log_show_state *)handle;
+
+  dout(10) << "log_show_next pos " << state->pos << " bl " << state->bl.length()
+	   << " off " << state->p.get_off()
+	   << " eof " << (int)state->eof
+	   << dendl;
+  // read some?
+  unsigned chunk = 1024*1024;
+  if (state->bl.length() < chunk/2 && !state->eof) {
+    bufferlist more;
+    int r = state->io_ctx.read(state->name, more, chunk, state->pos);
+    if (r < 0)
+      return r;
+    bufferlist old;
+    old.substr_of(state->bl, state->p.get_off(), state->bl.length() - state->p.get_off());
+    state->bl.clear();
+    state->bl.claim(old);
+    state->bl.claim_append(more);
+    state->p = state->bl.begin();
+    if ((unsigned)r < chunk)
+      state->eof = true;
+    dout(10) << " read " << r << dendl;
+  }
+
+  if (state->p.end())
+    return 0;  // end of file
+  try {
+    ::decode(*entry, state->p);
+  }
+  catch (const buffer::error &e) {
+    return -EINVAL;
+  }
+  return 1;
+}
 
 int RGWRados::decode_policy(bufferlist& bl, ACLOwner *owner)
 {

src/rgw/rgw_rados.h

@@ -145,6 +145,14 @@ public:
   int log_list_init(const string& prefix, RGWAccessHandle *handle);
   int log_list_next(RGWAccessHandle handle, string *name);
 
+  /// remove log
+  int log_remove(const string& name);
+
+  /// show log
+  int log_show_init(const string& name, RGWAccessHandle *handle);
+  int log_show_next(RGWAccessHandle handle, rgw_log_entry *entry);
+
+
   /** get listing of the objects in a bucket */
   virtual int list_objects(std::string& id, rgw_bucket& bucket, int max, std::string& prefix, std::string& delim,
                    std::string& marker, std::vector<RGWObjEnt>& result, map<string, bool>& common_prefixes,