data = self._get("/api/summary")
self.assertStatus(200)
- self.assertIn('filesystems', data)
self.assertIn('health_status', data)
- self.assertIn('rbd_pools', data)
self.assertIn('mgr_id', data)
self.assertIn('have_mon_connection', data)
self.assertIn('rbd_mirroring', data)
self.assertIn('executing_tasks', data)
self.assertIn('finished_tasks', data)
- self.assertIsNotNone(data['filesystems'])
self.assertIsNotNone(data['health_status'])
- self.assertIsNotNone(data['rbd_pools'])
self.assertIsNotNone(data['mgr_id'])
self.assertIsNotNone(data['have_mon_connection'])
self.assertEqual(data['rbd_mirroring'], {'errors': 0, 'warnings': 0})
from . import AuthRequired, ApiController, Endpoint, BaseController
from ..controllers.rbd_mirroring import get_daemons_and_pools
from ..tools import ViewCacheNoDataException
-from ..services.ceph_service import CephService
from ..tools import TaskManager
@ApiController('/summary')
@AuthRequired()
class Summary(BaseController):
- def _rbd_pool_data(self):
- pool_names = [pool['pool_name'] for pool in CephService.get_pool_list('rbd')]
- return sorted(pool_names)
-
def _health_status(self):
health_data = mgr.get("health")
return json.loads(health_data["json"])['status']
- def _filesystems(self):
- fsmap = mgr.get("fs_map")
- return [
- {
- "id": f['id'],
- "name": f['mdsmap']['fs_name']
- }
- for f in fsmap['filesystems']
- ]
-
def _rbd_mirroring(self):
try:
_, data = get_daemons_and_pools()
def __call__(self):
executing_t, finished_t = TaskManager.list_serializable()
return {
- 'rbd_pools': self._rbd_pool_data(),
'health_status': self._health_status(),
- 'filesystems': self._filesystems(),
'rbd_mirroring': self._rbd_mirroring(),
'mgr_id': mgr.get_mgr_id(),
'have_mon_connection': mgr.have_mon_connection(),