Merge "VMware vSphere support: Disk rates"

ceilometer/compute/pollsters/disk.py

@@ -95,6 +95,12 @@ class _Base(plugin.ComputePollster):
             except virt_inspector.InstanceNotFoundException as err:
                 # Instance was deleted while getting samples. Ignore it.
                 LOG.debug(_('Exception while getting samples %s'), err)
+            except NotImplementedError:
+                # Selected inspector does not implement this pollster.
+                LOG.debug(_('%(inspector)s does not provide data for '
+                            ' %(pollster)s'), ({
+                          'inspector': manager.inspector.__class__.__name__,
+                          'pollster': self.__class__.__name__}))
             except Exception as err:
                 LOG.warning(_('Ignoring instance %(name)s: %(error)s') % (
                             {'name': instance_name, 'error': err}))
@@ -151,3 +157,104 @@ class WriteBytesPollster(_Base):
             unit='B',
             volume=c_data.w_bytes,
         )
+
+
+@six.add_metaclass(abc.ABCMeta)
+class _DiskRatesPollsterBase(plugin.ComputePollster):
+
+    CACHE_KEY_DISK_RATE = 'diskio-rate'
+
+    def _populate_cache(self, inspector, cache, instance):
+        i_cache = cache.setdefault(self.CACHE_KEY_DISK_RATE, {})
+        if instance.id not in i_cache:
+            r_bytes_rate = 0
+            r_requests_rate = 0
+            w_bytes_rate = 0
+            w_requests_rate = 0
+            for disk, info in inspector.inspect_disk_rates(instance):
+                r_bytes_rate += info.read_bytes_rate
+                r_requests_rate += info.read_requests_rate
+                w_bytes_rate += info.write_bytes_rate
+                w_requests_rate += info.write_requests_rate
+            i_cache[instance.id] = virt_inspector.DiskRateStats(
+                r_bytes_rate,
+                r_requests_rate,
+                w_bytes_rate,
+                w_requests_rate
+            )
+        return i_cache[instance.id]
+
+    @abc.abstractmethod
+    def _get_sample(self, instance, disk_rates_info):
+        """Return one Sample."""
+
+    def get_samples(self, manager, cache, resources):
+        for instance in resources:
+            try:
+                disk_rates_info = self._populate_cache(
+                    manager.inspector,
+                    cache,
+                    instance,
+                )
+                yield self._get_sample(instance, disk_rates_info)
+            except virt_inspector.InstanceNotFoundException as err:
+                # Instance was deleted while getting samples. Ignore it.
+                LOG.debug(_('Exception while getting samples %s'), err)
+            except NotImplementedError:
+                # Selected inspector does not implement this pollster.
+                LOG.debug(_('%(inspector)s does not provide data for '
+                            ' %(pollster)s'), ({
+                          'inspector': manager.inspector.__class__.__name__,
+                          'pollster': self.__class__.__name__}))
+            except Exception as err:
+                instance_name = util.instance_name(instance)
+                LOG.error(_('Ignoring instance %(name)s: %(error)s') % (
+                    {'name': instance_name, 'error': err}))
+
+
+class ReadBytesRatePollster(_DiskRatesPollsterBase):
+
+    def _get_sample(self, instance, disk_rates_info):
+        return util.make_sample_from_instance(
+            instance,
+            name='disk.read.bytes.rate',
+            type=sample.TYPE_GAUGE,
+            unit='B/s',
+            volume=disk_rates_info.read_bytes_rate,
+        )
+
+
+class ReadRequestsRatePollster(_DiskRatesPollsterBase):
+
+    def _get_sample(self, instance, disk_rates_info):
+        return util.make_sample_from_instance(
+            instance,
+            name='disk.read.requests.rate',
+            type=sample.TYPE_GAUGE,
+            unit='requests/s',
+            volume=disk_rates_info.read_requests_rate,
+        )
+
+
+class WriteBytesRatePollster(_DiskRatesPollsterBase):
+
+    def _get_sample(self, instance, disk_rates_info):
+        return util.make_sample_from_instance(
+            instance,
+            name='disk.write.bytes.rate',
+            type=sample.TYPE_GAUGE,
+            unit='B/s',
+            volume=disk_rates_info.write_bytes_rate,
+        )
+
+
+class WriteRequestsRatePollster(_DiskRatesPollsterBase):
+
+    def _get_sample(self, instance, disk_rates_info):
+        return util.make_sample_from_instance(
+            instance,
+            name='disk.write.requests.rate',
+            type=sample.TYPE_GAUGE,
+            unit='requests/s',
+            volume=disk_rates_info.write_requests_rate,
+        )

ceilometer/compute/virt/inspector.py

@@ -118,6 +118,19 @@ DiskStats = collections.namedtuple('DiskStats',
                                     'write_bytes', 'write_requests',
                                     'errors'])
 
+# Named tuple representing disk rate statistics.
+#
+# read_bytes_rate: number of bytes read per second
+# read_requests_rate: number of read operations per second
+# write_bytes_rate: number of bytes written per second
+# write_requests_rate: number of write operations per second
+#
+DiskRateStats = collections.namedtuple('DiskRateStats',
+                                       ['read_bytes_rate',
+                                        'read_requests_rate',
+                                        'write_bytes_rate',
+                                        'write_requests_rate'])
+
 
 # Exception types
 #
@@ -189,6 +202,15 @@ class Inspector(object):
         """
         raise NotImplementedError()
 
+    def inspect_disk_rates(self, instance):
+        """Inspect the disk statistics as rates for an instance.
+
+        :param instance: the target instance
+        :return: for each disk, the number of bytes & operations
+                 read and written per second, with the error count
+        """
+        raise NotImplementedError()
+
 
 def get_hypervisor_inspector():
     try:

ceilometer/compute/virt/vmware/inspector.py

@@ -53,6 +53,10 @@ VC_AVERAGE_MEMORY_CONSUMED_CNTR = 'mem:consumed:average'
 VC_AVERAGE_CPU_CONSUMED_CNTR = 'cpu:usage:average'
 VC_NETWORK_RX_BYTES_COUNTER = 'net:bytesRx:average'
 VC_NETWORK_TX_BYTES_COUNTER = 'net:bytesTx:average'
+VC_DISK_READ_RATE_CNTR = "disk:read:average"
+VC_DISK_READ_REQUESTS_RATE_CNTR = "disk:numberReadAveraged:average"
+VC_DISK_WRITE_RATE_CNTR = "disk:write:average"
+VC_DISK_WRITE_REQUESTS_RATE_CNTR = "disk:numberWriteAveraged:average"
 
 
 def get_api_session():
@@ -138,6 +142,42 @@ class VsphereInspector(virt_inspector.Inspector):
         mem_counter_id = self._ops.get_perf_counter_id(
             VC_AVERAGE_MEMORY_CONSUMED_CNTR)
         memory = self._ops.query_vm_aggregate_stats(vm_moid, mem_counter_id)
-        #Stat provided from VMware Vsphere is in Bytes, converting it to MB.
+        # Stat provided from VMware Vsphere is in Bytes, converting it to MB.
         memory = memory / (units.Mi)
         return virt_inspector.MemoryUsageStats(usage=memory)
+
+    def inspect_disk_rates(self, instance):
+        vm_moid = self._ops.get_vm_moid(instance.id)
+        if not vm_moid:
+            raise virt_inspector.InstanceNotFoundException(
+                _('VM %s not found in VMware Vsphere') % instance.id)
+
+        disk_stats = {}
+        disk_ids = set()
+        disk_counters = [
+            VC_DISK_READ_RATE_CNTR,
+            VC_DISK_READ_REQUESTS_RATE_CNTR,
+            VC_DISK_WRITE_RATE_CNTR,
+            VC_DISK_WRITE_REQUESTS_RATE_CNTR
+        ]
+
+        for disk_counter in disk_counters:
+            disk_counter_id = self._ops.get_perf_counter_id(disk_counter)
+            disk_id_to_stat_map = self._ops.query_vm_device_stats(
+                vm_moid, disk_counter_id)
+            disk_stats[disk_counter] = disk_id_to_stat_map
+            disk_ids.update(disk_id_to_stat_map.iterkeys())
+
+        for disk_id in disk_ids:
+
+            def stat_val(counter_name):
+                return disk_stats[counter_name].get(disk_id, 0)
+
+            disk = virt_inspector.Disk(device=disk_id)
+            disk_rate_info = virt_inspector.DiskRateStats(
+                read_bytes_rate=stat_val(VC_DISK_READ_RATE_CNTR) * units.Ki,
+                read_requests_rate=stat_val(VC_DISK_READ_REQUESTS_RATE_CNTR),
+                write_bytes_rate=stat_val(VC_DISK_WRITE_RATE_CNTR) * units.Ki,
+                write_requests_rate=stat_val(VC_DISK_WRITE_REQUESTS_RATE_CNTR)
+            )
+            yield(disk, disk_rate_info)

ceilometer/tests/compute/pollsters/test_diskio.py

@@ -73,3 +73,54 @@ class TestDiskPollsters(base.TestPollsterBase):
     def test_disk_write_bytes(self):
         self._check_get_samples(disk.WriteBytesPollster,
                                 'disk.write.bytes', 3L)
+
+
+class TestDiskRatePollsters(base.TestPollsterBase):
+
+    DISKS = [
+        (virt_inspector.Disk(device='disk1'),
+         virt_inspector.DiskRateStats(1024, 300, 5120, 700)),
+
+        (virt_inspector.Disk(device='disk2'),
+         virt_inspector.DiskRateStats(2048, 400, 6144, 800))
+    ]
+
+    def setUp(self):
+        super(TestDiskRatePollsters, self).setUp()
+        self.inspector.inspect_disk_rates = \
+            mock.Mock(return_value=self.DISKS)
+
+    @mock.patch('ceilometer.pipeline.setup_pipeline', mock.MagicMock())
+    def _check_get_samples(self, factory, sample_name, expected_volume):
+        pollster = factory()
+
+        mgr = manager.AgentManager()
+        cache = {}
+        samples = list(pollster.get_samples(mgr, cache, [self.instance]))
+        assert samples
+        self.assertIsNotNone(samples)
+        self.assertIn(pollster.CACHE_KEY_DISK_RATE, cache)
+        self.assertIn(self.instance.id, cache[pollster.CACHE_KEY_DISK_RATE])
+
+        self.assertEqual(set([sample_name]), set([s.name for s in samples]))
+
+        match = [s for s in samples if s.name == sample_name]
+        self.assertEqual(1, len(match), 'missing counter %s' % sample_name)
+        self.assertEqual(expected_volume, match[0].volume)
+        self.assertEqual('gauge', match[0].type)
+
+    def test_disk_read_bytes_rate(self):
+        self._check_get_samples(disk.ReadBytesRatePollster,
+                                'disk.read.bytes.rate', 3072L)
+
+    def test_disk_read_requests_rate(self):
+        self._check_get_samples(disk.ReadRequestsRatePollster,
+                                'disk.read.requests.rate', 700L)
+
+    def test_disk_write_bytes_rate(self):
+        self._check_get_samples(disk.WriteBytesRatePollster,
+                                'disk.write.bytes.rate', 11264L)
+
+    def test_disk_write_requests_rate(self):
+        self._check_get_samples(disk.WriteRequestsRatePollster,
+                                'disk.write.requests.rate', 1500L)

ceilometer/tests/compute/virt/vmware/test_inspector.py

@@ -121,3 +121,48 @@ class TestVsphereInspection(test.BaseTestCase):
 
         for vnic, rates_info in result:
             self.assertEqual(expected_stats[vnic.name], rates_info)
+
+    def test_inspect_disk_rates(self):
+
+        # construct test data
+        test_vm_moid = "vm-21"
+        disk1 = "disk-1"
+        disk2 = "disk-2"
+        counter_name_to_id_map = {
+            vsphere_inspector.VC_DISK_READ_RATE_CNTR: 1,
+            vsphere_inspector.VC_DISK_READ_REQUESTS_RATE_CNTR: 2,
+            vsphere_inspector.VC_DISK_WRITE_RATE_CNTR: 3,
+            vsphere_inspector.VC_DISK_WRITE_REQUESTS_RATE_CNTR: 4
+        }
+        counter_id_to_stats_map = {
+            1: {disk1: 1, disk2: 2},
+            2: {disk1: 300, disk2: 400},
+            3: {disk1: 5, disk2: 6},
+            4: {disk1: 700},
+        }
+
+        def get_counter_id_side_effect(counter_full_name):
+            return counter_name_to_id_map[counter_full_name]
+
+        def query_stat_side_effect(vm_moid, counter_id):
+            # assert inputs
+            self.assertEqual(test_vm_moid, vm_moid)
+            self.assertTrue(counter_id in counter_id_to_stats_map)
+            return counter_id_to_stats_map[counter_id]
+
+        # configure vsphere operations mock with the test data
+        ops_mock = self._inspector._ops
+        ops_mock.get_vm_moid.return_value = test_vm_moid
+        ops_mock.get_perf_counter_id.side_effect = get_counter_id_side_effect
+        ops_mock.query_vm_device_stats.side_effect = query_stat_side_effect
+
+        result = self._inspector.inspect_disk_rates(mock.MagicMock())
+
+        # validate result
+        expected_stats = {
+            disk1: virt_inspector.DiskRateStats(1024, 300, 5120, 700),
+            disk2: virt_inspector.DiskRateStats(2048, 400, 6144, 0)
+        }
+
+        actual_stats = dict((disk.device, rates) for (disk, rates) in result)
+        self.assertEqual(expected_stats, actual_stats)

setup.cfg

@@ -72,6 +72,10 @@ ceilometer.poll.compute =
     disk.write.requests = ceilometer.compute.pollsters.disk:WriteRequestsPollster
     disk.read.bytes = ceilometer.compute.pollsters.disk:ReadBytesPollster
     disk.write.bytes = ceilometer.compute.pollsters.disk:WriteBytesPollster
+    disk.read.requests.rate = ceilometer.compute.pollsters.disk:ReadRequestsRatePollster
+    disk.write.requests.rate = ceilometer.compute.pollsters.disk:WriteRequestsRatePollster
+    disk.read.bytes.rate = ceilometer.compute.pollsters.disk:ReadBytesRatePollster
+    disk.write.bytes.rate = ceilometer.compute.pollsters.disk:WriteBytesRatePollster
     cpu = ceilometer.compute.pollsters.cpu:CPUPollster
     cpu_util = ceilometer.compute.pollsters.cpu:CPUUtilPollster
     network.incoming.bytes = ceilometer.compute.pollsters.net:IncomingBytesPollster