"""
import json
from ceph_volume import process
-from ceph_volume.exceptions import MultipleLVsError
+from ceph_volume.exceptions import MultipleLVsError, MultipleVGsError
def parse_tags(lv_tags):
return tag_mapping
+def get_api_vgs():
+ """
+ Return the list of group volumes available in the system using flags to include common
+ metadata associated with them
+
+ Command and sample JSON output, should look like::
+
+ $ sudo vgs --reportformat=json
+ {
+ "report": [
+ {
+ "vg": [
+ {
+ "vg_name":"VolGroup00",
+ "pv_count":"1",
+ "lv_count":"2",
+ "snap_count":"0",
+ "vg_attr":"wz--n-",
+ "vg_size":"38.97g",
+ "vg_free":"0 "},
+ {
+ "vg_name":"osd_vg",
+ "pv_count":"3",
+ "lv_count":"1",
+ "snap_count":"0",
+ "vg_attr":"wz--n-",
+ "vg_size":"32.21g",
+ "vg_free":"9.21g"
+ }
+ ]
+ }
+ ]
+ }
+
+ """
+ stdout, stderr, returncode = process.call(
+ [
+ 'sudo', 'vgs', '--reportformat=json'
+ ]
+ )
+ report = json.loads(b''.join(stdout).decode('utf-8'))
+ for report_item in report.get('report', []):
+ # is it possible to get more than one item in "report" ?
+ return report_item['vg']
+ return []
+
+
def get_api_lvs():
"""
Return the list of logical volumes available in the system using flags to include common
return matched_lvs[0]
+class VolumeGroups(list):
+ """
+ A list of all known volume groups for the current system, with the ability
+ to filter them via keyword arguments.
+ """
+
+ def __init__(self):
+ self._populate()
+
+ def _populate(self):
+ # get all the vgs in the current system
+ for vg_item in get_api_vgs():
+ self.append(VolumeGroup(**vg_item))
+
+ def _purge(self):
+ """
+ Deplete all the items in the list, used internally only so that we can
+ dynamically allocate the items when filtering without the concern of
+ messing up the contents
+ """
+ self[:] = []
+
+ def _filter(self, vg_name=None, vg_tags=None):
+ """
+ The actual method that filters using a new list. Useful so that other
+ methods that do not want to alter the contents of the list (e.g.
+ ``self.find``) can operate safely.
+
+ .. note:: ``vg_tags`` is not yet implemented
+ """
+ filtered = [i for i in self]
+ if vg_name:
+ filtered = [i for i in filtered if i.vg_name == vg_name]
+
+ # at this point, `filtered` has either all the volumes in self or is an
+ # actual filtered list if any filters were applied
+ if vg_tags:
+ tag_filtered = []
+ for k, v in vg_tags.items():
+ for volume in filtered:
+ if volume.tags.get(k) == str(v):
+ if volume not in tag_filtered:
+ tag_filtered.append(volume)
+ # return the tag_filtered volumes here, the `filtered` list is no
+ # longer useable
+ return tag_filtered
+
+ return filtered
+
+ def filter(self, vg_name=None, vg_tags=None):
+ """
+ Filter out groups on top level attributes like ``vg_name`` or by
+ ``vg_tags`` where a dict is required. For example, to find a Ceph group
+ with dmcache as the type, the filter would look like::
+
+ vg_tags={'ceph.type': 'dmcache'}
+
+ .. warning:: These tags are not documented because they are currently
+ unused, but are here to maintain API consistency
+ """
+ if not any([vg_name, vg_tags]):
+ raise TypeError('.filter() requires vg_name or vg_tags (none given)')
+ # first find the filtered volumes with the values in self
+ filtered_groups = self._filter(
+ vg_name=vg_name,
+ vg_tags=vg_tags
+ )
+ # then purge everything
+ self._purge()
+ # and add the filtered items
+ self.extend(filtered_groups)
+
+ def get(self, vg_name=None, vg_tags=None):
+ """
+ This is a bit expensive, since it will try to filter out all the
+ matching items in the list, filter them out applying anything that was
+ added and return the matching item.
+
+ This method does *not* alter the list, and it will raise an error if
+ multiple VGs are matched
+
+ It is useful to use ``tags`` when trying to find a specific volume group,
+ but it can also lead to multiple vgs being found (although unlikely)
+ """
+ if not any([vg_name, vg_tags]):
+ raise TypeError('.get() requires vg_name or vg_tags (none given)')
+ vgs = self._filter(
+ vg_name=vg_name,
+ vg_tags=vg_tags
+ )
+ if len(vgs) > 1:
+ # this is probably never going to happen, but it is here to keep
+ # the API code consistent
+ raise MultipleVGsError(vg_name)
+ try:
+ return vgs[0]
+ except IndexError:
+ return None
+
+
class Volumes(list):
"""
A list of all known (logical) volumes for the current system, with the ability
return None
+class VolumeGroup(object):
+ """
+ Represents an LVM group, with some top-level attributes like ``vg_name``
+ """
+
+ def __init__(self, **kw):
+ for k, v in kw.items():
+ setattr(self, k, v)
+ self.name = kw['vg_name']
+
+ def __str__(self):
+ return '<%s>' % self.name
+
+ def __repr__(self):
+ return self.__str__()
+
+
class Volume(object):
"""
Represents a Logical Volume from LVM, with some top-level attributes like
projects / ceph.git / commitdiff