ceph_manager: add ceph-objectstore-tool wrapper

Add the CephManager.objectstore_tool method to encapsulate a call to
ceph-objectstore-tool. The wrapper can convert an object name into the
PG id and figure out the primary OSD. The designated OSD is stopped
before running the command and restarted afterwards.

Signed-off-by: Loic Dachary <ldachary@redhat.com>

diff --git a/tasks/ceph_manager.py b/tasks/ceph_manager.py
index b922c82b31b3c77bd9be8d94c8f7375d27f3a178..b67ea861d816a22dd08bc9024b0752fe05050769 100644 (file)
--- a/tasks/ceph_manager.py
+++ b/tasks/ceph_manager.py
@@ -6,6 +6,7 @@ import contextlib
 import random
 import time
 import gevent
+import base64
 import json
 import threading
 import os
@@ -595,6 +596,68 @@ class Thrasher:
         self.all_up()
 
 
+class ObjectStoreTool:
+
+    def __init__(self, manager, pool, **kwargs):
+        self.manager = manager
+        self.pool = pool
+        self.osd = kwargs.get('osd', None)
+        self.object_name = kwargs.get('object_name', None)
+        if self.osd and self.pool and self.object_name:
+            if self.osd == "primary":
+                self.osd = self.manager.get_object_primary(self.pool,
+                                                           self.object_name)
+        assert self.osd
+        if self.object_name:
+            self.pgid = self.manager.get_object_pg_with_shard(self.pool,
+                                                              self.object_name,
+                                                              self.osd)
+        self.remote = self.manager.ctx.\
+            cluster.only('osd.{o}'.format(o=self.osd)).remotes.keys()[0]
+        path = self.manager.get_filepath().format(id=self.osd)
+        self.paths = ("--data-path {path} --journal-path {path}/journal".
+                      format(path=path))
+
+    def build_cmd(self, options, args, stdin):
+        lines = []
+        if self.object_name:
+            lines.append("object=$(sudo ceph-objectstore-tool "
+                         "{paths} --pgid {pgid} --op list |"
+                         "grep '\"oid\":\"{name}\"')".
+                         format(paths=self.paths,
+                                pgid=self.pgid,
+                                name=self.object_name))
+            args = '"$object" ' + args
+            options += " --pgid {pgid}".format(pgid=self.pgid)
+        cmd = ("sudo ceph-objectstore-tool {paths} {options} {args}".
+               format(paths=self.paths,
+                      args=args,
+                      options=options))
+        if stdin:
+            cmd = ("echo {payload} | base64 --decode | {cmd}".
+                   format(payload=base64.encode(kwargs['stdin']),
+                          cmd=cmd))
+        lines.append(cmd)
+        return "\n".join(lines)
+
+    def run(self, options, args, stdin=None):
+        self.manager.kill_osd(self.osd)
+        cmd = self.build_cmd(options, args, stdin)
+        self.manager.log(cmd)
+        try:
+            proc = self.remote.run(args=['bash', '-e', '-x', '-c', cmd],
+                                   check_status=False,
+                                   stdout=StringIO(),
+                                   stderr=StringIO())
+            proc.wait()
+            if proc.exitstatus != 0:
+                self.manager.log("failed with " + str(proc.exitstatus))
+                error = proc.stdout.getvalue() + " " + proc.stderr.getvalue()
+                raise Exception(error)
+        finally:
+            self.manager.revive_osd(self.osd)
+
+
 class CephManager:
     """
     Ceph manager object.
@@ -776,6 +839,9 @@ class CephManager:
             check_status=check_status
             )
 
+    def objectstore_tool(self, pool, options, args, **kwargs):
+        return ObjectStoreTool(self, pool, **kwargs).run(options, args)
+
     def get_pgid(self, pool, pgnum):
         """
         :param pool: pool name