openstack
/
ceilometer
Code Issues Proposed changes

Merge "High precision rate of change timedelta" into stable/newton

This commit is contained in:
Jenkins 2017-07-24 18:06:47 +00:00 committed by Gerrit Code Review
parent c68ab232fd 0302d470ae
commit 327648f0d2
16 changed files with 143 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
ceilometer/compute/pollsters/__init__.py
View File

@ -73,7 +73,8 @@ class BaseComputePollster(plugin_base.PollsterBase):
return duration
@staticmethod
def _get_samples_per_devices(attribute, instance, _name, _type, _unit):
def _get_samples_per_devices(attribute, instance, _name, _type, _unit,
monotonic_time=None):
samples = []
for disk, value in six.iteritems(attribute):
samples.append(util.make_sample_from_instance(
@ -84,5 +85,6 @@ class BaseComputePollster(plugin_base.PollsterBase):
volume=value,
resource_id="%s-%s" % (instance.id, disk),
additional_metadata={'disk_name': disk},
monotonic_time=monotonic_time,
))
return samples

3
ceilometer/compute/pollsters/cpu.py
View File

@ -14,6 +14,8 @@
# License for the specific language governing permissions and limitations
# under the License.
import monotonic
from oslo_log import log
import ceilometer
@ -45,6 +47,7 @@ class CPUPollster(pollsters.BaseComputePollster):
unit='ns',
volume=cpu_info.time,
additional_metadata=cpu_num,
monotonic_time=monotonic.monotonic()
)
except virt_inspector.InstanceNotFoundException as err:
# Instance was deleted while getting samples. Ignore it.

9
ceilometer/compute/pollsters/disk.py
View File

@ -17,6 +17,7 @@
import abc
import collections
import monotonic
from oslo_log import log
import six
@ -33,7 +34,7 @@ LOG = log.getLogger(__name__)
DiskIOData = collections.namedtuple(
'DiskIOData',
'r_bytes r_requests w_bytes w_requests per_disk_requests',
'r_bytes r_requests w_bytes w_requests per_disk_requests polled_time',
)
DiskRateData = collections.namedtuple('DiskRateData',
@ -103,12 +104,14 @@ class _Base(pollsters.BaseComputePollster):
'write_bytes': per_device_write_bytes,
'write_requests': per_device_write_requests,
}
polled_time = monotonic.monotonic()
i_cache[instance.id] = DiskIOData(
r_bytes=r_bytes,
r_requests=r_requests,
w_bytes=w_bytes,
w_requests=w_requests,
per_disk_requests=per_device_requests,
polled_time=polled_time
)
return i_cache[instance.id]
@ -128,13 +131,15 @@ class _Base(pollsters.BaseComputePollster):
volume=getattr(c_data, _volume),
additional_metadata={
'device': c_data.per_disk_requests[_metadata].keys()},
monotonic_time=c_data.polled_time,
)]
def _get_samples_per_device(self, c_data, _attr, instance, _name, _unit):
"""Return one or more Samples for meter 'disk.device.*'"""
return self._get_samples_per_devices(c_data.per_disk_requests[_attr],
instance, _name,
sample.TYPE_CUMULATIVE, _unit)
sample.TYPE_CUMULATIVE, _unit,
c_data.polled_time)
def get_samples(self, manager, cache, resources):
for instance in resources:

36
ceilometer/compute/pollsters/net.py
View File

@ -16,6 +16,7 @@
import copy
import monotonic
from oslo_log import log
import ceilometer
@ -35,7 +36,8 @@ class _Base(pollsters.BaseComputePollster):
"write-bytes=%d"])
@staticmethod
def make_vnic_sample(instance, name, type, unit, volume, vnic_data):
def make_vnic_sample(instance, name, type, unit, volume, vnic_data,
monotonic_time):
metadata = copy.copy(vnic_data)
additional_metadata = dict(zip(metadata._fields, metadata))
if vnic_data.fref is not None:
@ -53,7 +55,8 @@ class _Base(pollsters.BaseComputePollster):
unit=unit,
volume=volume,
resource_id=rid,
additional_metadata=additional_metadata
additional_metadata=additional_metadata,
monotonic_time=monotonic_time,
)
CACHE_KEY_VNIC = 'vnics'
@ -72,9 +75,10 @@ class _Base(pollsters.BaseComputePollster):
def _get_vnics_for_instance(self, cache, inspector, instance):
i_cache = cache.setdefault(self.CACHE_KEY_VNIC, {})
if instance.id not in i_cache:
i_cache[instance.id] = list(
self._get_vnic_info(inspector, instance)
)
data = list(self._get_vnic_info(inspector, instance))
polled_time = monotonic.monotonic()
i_cache[instance.id] = [(vnic, info, polled_time)
for vnic, info in data]
return i_cache[instance.id]
def get_samples(self, manager, cache, resources):
@ -88,11 +92,11 @@ class _Base(pollsters.BaseComputePollster):
self.inspector,
instance,
)
for vnic, info in vnics:
for vnic, info, polled_time in vnics:
LOG.debug(self.NET_USAGE_MESSAGE, instance_name,
vnic.name, self._get_rx_info(info),
self._get_tx_info(info))
yield self._get_sample(instance, vnic, info)
yield self._get_sample(instance, vnic, info, polled_time)
except virt_inspector.InstanceNotFoundException as err:
# Instance was deleted while getting samples. Ignore it.
LOG.debug('Exception while getting samples %s', err)
@ -151,7 +155,7 @@ class _PacketsBase(_Base):
class IncomingBytesPollster(_Base):
def _get_sample(self, instance, vnic, info):
def _get_sample(self, instance, vnic, info, polled_time):
return self.make_vnic_sample(
instance,
name='network.incoming.bytes',
@ -159,12 +163,13 @@ class IncomingBytesPollster(_Base):
unit='B',
volume=info.rx_bytes,
vnic_data=vnic,
monotonic_time=polled_time,
)
class IncomingPacketsPollster(_PacketsBase):
def _get_sample(self, instance, vnic, info):
def _get_sample(self, instance, vnic, info, polled_time):
return self.make_vnic_sample(
instance,
name='network.incoming.packets',
@ -172,12 +177,13 @@ class IncomingPacketsPollster(_PacketsBase):
unit='packet',
volume=info.rx_packets,
vnic_data=vnic,
monotonic_time=polled_time,
)
class OutgoingBytesPollster(_Base):
def _get_sample(self, instance, vnic, info):
def _get_sample(self, instance, vnic, info, polled_time):
return self.make_vnic_sample(
instance,
name='network.outgoing.bytes',
@ -185,12 +191,13 @@ class OutgoingBytesPollster(_Base):
unit='B',
volume=info.tx_bytes,
vnic_data=vnic,
monotonic_time=polled_time,
)
class OutgoingPacketsPollster(_PacketsBase):
def _get_sample(self, instance, vnic, info):
def _get_sample(self, instance, vnic, info, polled_time):
return self.make_vnic_sample(
instance,
name='network.outgoing.packets',
@ -198,12 +205,13 @@ class OutgoingPacketsPollster(_PacketsBase):
unit='packet',
volume=info.tx_packets,
vnic_data=vnic,
monotonic_time=polled_time,
)
class IncomingBytesRatePollster(_RateBase):
def _get_sample(self, instance, vnic, info):
def _get_sample(self, instance, vnic, info, polled_time):
return self.make_vnic_sample(
instance,
name='network.incoming.bytes.rate',
@ -211,12 +219,13 @@ class IncomingBytesRatePollster(_RateBase):
unit='B/s',
volume=info.rx_bytes_rate,
vnic_data=vnic,
monotonic_time=polled_time,
)
class OutgoingBytesRatePollster(_RateBase):
def _get_sample(self, instance, vnic, info):
def _get_sample(self, instance, vnic, info, polled_time):
return self.make_vnic_sample(
instance,
name='network.outgoing.bytes.rate',
@ -224,4 +233,5 @@ class OutgoingBytesRatePollster(_RateBase):
unit='B/s',
volume=info.tx_bytes_rate,
vnic_data=vnic,
monotonic_time=polled_time,
)

4
ceilometer/compute/pollsters/util.py
View File

@ -76,7 +76,8 @@ def _get_metadata_from_object(instance):
def make_sample_from_instance(instance, name, type, unit, volume,
resource_id=None, additional_metadata=None):
resource_id=None, additional_metadata=None,
monotonic_time=None):
additional_metadata = additional_metadata or {}
resource_metadata = _get_metadata_from_object(instance)
resource_metadata.update(additional_metadata)
@ -89,6 +90,7 @@ def make_sample_from_instance(instance, name, type, unit, volume,
project_id=instance.tenant_id,
resource_id=resource_id or instance.id,
resource_metadata=resource_metadata,
monotonic_time=monotonic_time,
)

5
ceilometer/pipeline.py
View File

@ -101,7 +101,10 @@ class SamplePipelineEndpoint(PipelineEndpoint):
resource_id=s['resource_id'],
timestamp=s['timestamp'],
resource_metadata=s['resource_metadata'],
source=s.get('source'))
source=s.get('source'),
# NOTE(sileht): May come from an older node,
# Put None in this case.
monotonic_time=s.get('monotonic_time'))
for s in samples if publisher_utils.verify_signature(
s, cfg.CONF.publisher.telemetry_secret)
]

1
ceilometer/publisher/utils.py
View File

@ -128,6 +128,7 @@ def meter_message_from_counter(sample, secret):
'timestamp': sample.timestamp,
'resource_metadata': sample.resource_metadata,
'message_id': sample.id,
'monotonic_time': sample.monotonic_time,
}
msg['message_signature'] = compute_signature(msg, secret)
return msg

3
ceilometer/sample.py
View File

@ -55,7 +55,7 @@ class Sample(object):
def __init__(self, name, type, unit, volume, user_id, project_id,
resource_id, timestamp=None, resource_metadata=None,
source=None, id=None):
source=None, id=None, monotonic_time=None):
self.name = name
self.type = type
self.unit = unit
@ -67,6 +67,7 @@ class Sample(object):
self.resource_metadata = resource_metadata or {}
self.source = source or cfg.CONF.sample_source
self.id = id or str(uuid.uuid1())
self.monotonic_time = monotonic_time
def as_dict(self):
return copy.copy(self.__dict__)

4
ceilometer/storage/impl_hbase.py
View File

@ -146,6 +146,10 @@ class Connection(hbase_base.Connection, base.Connection):
:param data: a dictionary such as returned by
ceilometer.publisher.utils.meter_message_from_counter
"""
# We must not record thing.
data.pop("monotonic_time", None)
with self.conn_pool.connection() as conn:
resource_table = conn.table(self.RESOURCE_TABLE)
meter_table = conn.table(self.METER_TABLE)

7
ceilometer/storage/impl_mongodb.py
View File

@ -259,6 +259,13 @@ class Connection(pymongo_base.Connection):
# unconditionally insert sample timestamps and resource metadata
# (in the update case, this must be conditional on the sample not
# being out-of-order)
# We must not store this
samples = copy.deepcopy(samples)
for sample in samples:
sample.pop("monotonic_time", None)
sorted_samples = sorted(
copy.deepcopy(samples),
key=lambda s: (s['resource_id'], s['timestamp']))

7
ceilometer/tests/functional/api/v2/test_post_samples_scenarios.py
View File

@ -80,6 +80,7 @@ class TestPostSamples(v2.FunctionalTest):
s1[0]['source'] = '%s:openstack' % s1[0]['project_id']
self.assertEqual(s1, data.json)
s1[0]["monotonic_time"] = None
self.assertEqual(s1[0], self.published[0][0])
def test_nested_metadata(self):
@ -107,6 +108,7 @@ class TestPostSamples(v2.FunctionalTest):
unwound['resource_metadata'] = {'nest': {'name1': 'value1',
'name2': 'value3'},
'name2': 'value2'}
unwound["monotonic_time"] = None
# only the published sample should be unwound, not the representation
# in the API response
self.assertEqual(s1[0], data.json[0])
@ -218,6 +220,7 @@ class TestPostSamples(v2.FunctionalTest):
msg['timestamp'] = timestamp.replace(tzinfo=None).isoformat()
self.assertEqual(s, c)
s["monotonic_time"] = None
self.assertEqual(s, self.published[0][x])
def test_missing_mandatory_fields(self):
@ -278,6 +281,7 @@ class TestPostSamples(v2.FunctionalTest):
s['timestamp'] = data.json[x]['timestamp']
s.setdefault('resource_metadata', dict())
self.assertEqual(s, data.json[x])
s['monotonic_time'] = None
self.assertEqual(s, self.published[0][x])
def test_multiple_samples_multiple_sources(self):
@ -328,6 +332,7 @@ class TestPostSamples(v2.FunctionalTest):
s['timestamp'] = data.json[x]['timestamp']
s.setdefault('resource_metadata', dict())
self.assertEqual(s, data.json[x])
s['monotonic_time'] = None
self.assertEqual(s, self.published[0][x])
def test_missing_project_user_id(self):
@ -364,4 +369,6 @@ class TestPostSamples(v2.FunctionalTest):
s['project_id'] = project_id
self.assertEqual(s, data.json[x])
s['monotonic_time'] = None
self.assertEqual(s, self.published[0][x])

6
ceilometer/tests/functional/storage/test_storage_scenarios.py
View File

@ -435,6 +435,7 @@ class RawSampleTest(DBTestBase):
d = meter.as_dict()
self.assertTimestampEqual(timeutils.utcnow(), d['recorded_at'])
del d['recorded_at']
d['monotonic_time'] = None
self.assertIn(d, self.msgs[:3])
def test_get_samples_by_user_limit(self):
@ -455,6 +456,7 @@ class RawSampleTest(DBTestBase):
d = meter.as_dict()
self.assertTimestampEqual(timeutils.utcnow(), d['recorded_at'])
del d['recorded_at']
d['monotonic_time'] = None
self.assertIn(d, self.msgs[:4])
def test_get_samples_by_resource(self):
@ -464,6 +466,7 @@ class RawSampleTest(DBTestBase):
d = results[1].as_dict()
self.assertEqual(timeutils.utcnow(), d['recorded_at'])
del d['recorded_at']
d['monotonic_time'] = None
self.assertEqual(self.msgs[0], d)
def test_get_samples_by_metaquery(self):
@ -475,6 +478,7 @@ class RawSampleTest(DBTestBase):
d = meter.as_dict()
self.assertTimestampEqual(timeutils.utcnow(), d['recorded_at'])
del d['recorded_at']
d['monotonic_time'] = None
self.assertIn(d, self.msgs)
def test_get_samples_by_metaquery_key_with_dot_in_metadata(self):
@ -726,6 +730,7 @@ class ComplexSampleQueryTest(DBTestBase):
for sample_item in results:
d = sample_item.as_dict()
del d['recorded_at']
d['monotonic_time'] = None
self.assertIn(d, self.msgs)
def test_query_complex_filter_with_regexp(self):
@ -3224,6 +3229,7 @@ class TestBatchRecordingMetering(tests_db.TestBase):
for sample_item in results:
d = sample_item.as_dict()
del d['recorded_at']
d['monotonic_time'] = None
self.assertIn(d, self.sample_dicts)
resources = list(self.conn.get_resources())

61
ceilometer/tests/pipeline_base.py
View File

@ -22,6 +22,7 @@ import tempfile
import traceback
import mock
import monotonic
from oslo_utils import timeutils
from oslotest import base
from oslotest import mockpatch
@ -1092,6 +1093,66 @@ class BasePipelineTestCase(base.BaseTestCase):
pipe.flush()
self.assertEqual(0, len(publisher.samples))
def test_rate_of_change_precision(self):
s = "100.0 / (10**9 * (resource_metadata.cpu_number or 1))"
transformer_cfg = [
{
'name': 'rate_of_change',
'parameters': {
'source': {},
'target': {'name': 'cpu_util',
'unit': '%',
'type': sample.TYPE_GAUGE,
'scale': s}
}
},
]
self._set_pipeline_cfg('transformers', transformer_cfg)
self._set_pipeline_cfg('counters', ['cpu'])
pipeline_manager = pipeline.PipelineManager(
self.cfg2file(self.pipeline_cfg), self.transformer_manager)
pipe = pipeline_manager.pipelines[0]
now = timeutils.utcnow()
now_time = monotonic.monotonic()
# Simulate a laggy poller
later = now + datetime.timedelta(seconds=12345)
later_time = now_time + 10
counters = [
sample.Sample(
name='cpu',
type=sample.TYPE_CUMULATIVE,
volume=125000000000,
unit='ns',
user_id='test_user',
project_id='test_proj',
resource_id='test_resource',
timestamp=now.isoformat(),
monotonic_time=now_time,
resource_metadata={'cpu_number': 4}
),
sample.Sample(
name='cpu',
type=sample.TYPE_CUMULATIVE,
volume=165000000000,
unit='ns',
user_id='test_user',
project_id='test_proj',
resource_id='test_resource',
timestamp=later.isoformat(),
monotonic_time=later_time,
resource_metadata={'cpu_number': 4}
),
]
pipe.publish_data(counters)
publisher = pipe.publishers[0]
self.assertEqual(1, len(publisher.samples))
cpu_util_sample = publisher.samples[0]
self.assertEqual(100, cpu_util_sample.volume)
def test_rate_of_change_max(self):
s = "100.0 / (10**9 * (resource_metadata.cpu_number or 1))"
transformer_cfg = [

5
ceilometer/tests/unit/compute/pollsters/test_net.py
View File

@ -184,7 +184,8 @@ class TestNetPollster(base.TestPollsterBase):
type=sample.TYPE_CUMULATIVE,
unit='B',
volume=100,
vnic_data=self.vnic0)
vnic_data=self.vnic0,
monotonic_time=123.123)
user_metadata = sm.resource_metadata['user_metadata']
expected = self.INSTANCE_PROPERTIES[
@ -207,7 +208,7 @@ class TestNetPollsterCache(base.TestPollsterBase):
dhcp_server='10.0.0.1'))
stats0 = virt_inspector.InterfaceStats(rx_bytes=1, rx_packets=2,
tx_bytes=3, tx_packets=4)
vnics = [(vnic0, stats0)]
vnics = [(vnic0, stats0, 123.123)]
mgr = manager.AgentManager()
pollster = factory()

10
ceilometer/transformer/conversions.py
View File

@ -187,12 +187,18 @@ class RateOfChangeTransformer(ScalingTransformer):
key = s.name + s.resource_id
prev = self.cache.get(key)
timestamp = timeutils.parse_isotime(s.timestamp)
self.cache[key] = (s.volume, timestamp)
self.cache[key] = (s.volume, timestamp, s.monotonic_time)
if prev:
prev_volume = prev[0]
prev_timestamp = prev[1]
time_delta = timeutils.delta_seconds(prev_timestamp, timestamp)
prev_monotonic_time = prev[2]
if (prev_monotonic_time is not None and
s.monotonic_time is not None):
# NOTE(sileht): Prefer high precision timer
time_delta = s.monotonic_time - prev_monotonic_time
else:
time_delta = timeutils.delta_seconds(prev_timestamp, timestamp)
# disallow violations of the arrow of time
if time_delta < 0:
LOG.warning(_('dropping out of time order sample: %s'), (s,))

1
requirements.txt
View File

@ -12,6 +12,7 @@ jsonschema!=2.5.0,<3.0.0,>=2.0.0 # MIT
kafka-python<1.0.0,>=0.9.5 # Apache-2.0
keystonemiddleware!=4.1.0,>=4.0.0 # Apache-2.0
lxml>=2.3 # BSD
monotonic
msgpack-python>=0.4.0 # Apache-2.0
oslo.concurrency>=3.5.0 # Apache-2.0
oslo.config>=3.9.0 # Apache-2.0