实现宿主机CPU、内存、网络、磁盘的数据接收

api/host_performance.py

@@ -0,0 +1,204 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+
+from flask import Blueprint
+import json
+import jimit as ji
+
+from api.base import Base
+from models import HostCPUMemory, HostTraffic, HostDiskUsageIO, Utils, Rules
+
+
+__author__ = 'James Iter'
+__date__ = '2017/8/7'
+__contact__ = 'james.iter.cn@gmail.com'
+__copyright__ = '(c) 2017 by James Iter.'
+
+
+blueprint = Blueprint(
+    'api_host_performance',
+    __name__,
+    url_prefix='/api/host_performance'
+)
+
+blueprints = Blueprint(
+    'api_host_performances',
+    __name__,
+    url_prefix='/api/host_performances'
+)
+
+
+host_cpu_memory = Base(the_class=HostCPUMemory, the_blueprint=blueprint, the_blueprints=blueprints)
+host_traffic = Base(the_class=HostTraffic, the_blueprint=blueprint, the_blueprints=blueprints)
+host_disk_usage_io = Base(the_class=HostDiskUsageIO, the_blueprint=blueprint, the_blueprints=blueprints)
+
+
+@Utils.dumps2response
+def r_cpu_memory_get_by_filter():
+    return host_cpu_memory.get_by_filter()
+
+
+@Utils.dumps2response
+def r_traffic_get_by_filter():
+    return host_traffic.get_by_filter()
+
+
+@Utils.dumps2response
+def r_disk_usage_io_get_by_filter():
+    return host_disk_usage_io.get_by_filter()
+
+
+def get_performance_data(uuid, uuid_field, the_class=None, granularity='hour'):
+
+    args_rules = [
+        Rules.UUID.value,
+    ]
+
+    try:
+        ji.Check.previewing(args_rules, {'uuid': uuid})
+        uuids_str = ':'.join([uuid_field, 'in', uuid])
+
+        ret = dict()
+        ret['state'] = ji.Common.exchange_state(20000)
+        ret['data'] = list()
+
+        max_limit = 10080
+        ts = ji.Common.ts()
+        _boundary = ts - 60 * 60
+        if granularity == 'hour':
+            _boundary = ts - 60 * 60
+
+        elif granularity == 'six_hours':
+            _boundary = ts - 60 * 60 * 6
+
+        elif granularity == 'day':
+            _boundary = ts - 60 * 60 * 24
+
+        elif granularity == 'seven_days':
+            _boundary = ts - 60 * 60 * 24 * 7
+
+        else:
+            pass
+
+        filter_str = ';'.join([uuids_str, 'timestamp:gt:' + _boundary.__str__()])
+
+        _rows, _rows_count = the_class.get_by_filter(
+            offset=0, limit=max_limit, order_by='id', order='asc', filter_str=filter_str)
+
+        def smooth_data(boundary=0, interval=60, now_ts=ji.Common.ts(), rows=None):
+            needs = list()
+            data = list()
+
+            for t in range(boundary + interval, now_ts, interval):
+                needs.append(t - t % interval)
+
+            for row in rows:
+                if row['timestamp'] % interval != 0:
+                    continue
+
+                if needs.__len__() > 0:
+                    t = needs.pop(0)
+                else:
+                    t = now_ts
+
+                while t < row['timestamp']:
+                    data.append({
+                        'timestamp': t,
+                        'cpu_load': None,
+                        'memory_available': None,
+                        'rx_packets': None,
+                        'rx_bytes': None,
+                        'tx_packets': None,
+                        'tx_bytes': None,
+                        'rd_req': None,
+                        'rd_bytes': None,
+                        'used': None,
+                        'wr_req': None,
+                        'wr_bytes': None
+                    })
+
+                    if needs.__len__() > 0:
+                        t = needs.pop(0)
+                    else:
+                        t = now_ts
+
+                data.append(row)
+
+            return data
+
+        if granularity == 'day':
+            ret['data'] = smooth_data(boundary=_boundary, interval=600, now_ts=ts, rows=_rows)
+
+        if granularity == 'seven_days':
+            ret['data'] = smooth_data(boundary=_boundary, interval=600, now_ts=ts, rows=_rows)
+
+        else:
+            ret['data'] = smooth_data(boundary=_boundary, interval=60, now_ts=ts, rows=_rows)
+
+        return ret
+
+    except ji.PreviewingError, e:
+        return json.loads(e.message)
+
+
+@Utils.dumps2response
+def r_cpu_memory_last_hour(uuid):
+    return get_performance_data(uuid=uuid, uuid_field='host_uuid', the_class=HostCPUMemory, granularity='hour')
+
+
+@Utils.dumps2response
+def r_cpu_memory_last_six_hours(uuid):
+    return get_performance_data(uuid=uuid, uuid_field='host_uuid', the_class=HostCPUMemory, granularity='six_hours')
+
+
+@Utils.dumps2response
+def r_cpu_memory_last_day(uuid):
+    return get_performance_data(uuid=uuid, uuid_field='host_uuid', the_class=HostCPUMemory, granularity='day')
+
+
+@Utils.dumps2response
+def r_cpu_memory_last_seven_days(uuid):
+    return get_performance_data(uuid=uuid, uuid_field='host_uuid', the_class=HostCPUMemory, granularity='seven_days')
+
+
+@Utils.dumps2response
+def r_traffic_last_hour(uuid):
+    return get_performance_data(uuid=uuid, uuid_field='host_uuid', the_class=HostTraffic, granularity='hour')
+
+
+@Utils.dumps2response
+def r_traffic_last_six_hours(uuid):
+    return get_performance_data(uuid=uuid, uuid_field='host_uuid', the_class=HostTraffic, granularity='six_hours')
+
+
+@Utils.dumps2response
+def r_traffic_last_day(uuid):
+    return get_performance_data(uuid=uuid, uuid_field='host_uuid', the_class=HostTraffic, granularity='day')
+
+
+@Utils.dumps2response
+def r_traffic_last_seven_days(uuid):
+    return get_performance_data(uuid=uuid, uuid_field='host_uuid', the_class=HostTraffic, granularity='seven_days')
+
+
+@Utils.dumps2response
+def r_disk_usage_io_last_hour(uuid):
+    return get_performance_data(uuid=uuid, uuid_field='host_uuid', the_class=HostDiskUsageIO, granularity='hour')
+
+
+@Utils.dumps2response
+def r_disk_usage_io_last_six_hours(uuid):
+    return get_performance_data(uuid=uuid, uuid_field='host_uuid', the_class=HostDiskUsageIO, granularity='six_hours')
+
+
+@Utils.dumps2response
+def r_disk_usage_io_last_day(uuid):
+    return get_performance_data(uuid=uuid, uuid_field='host_uuid', the_class=HostDiskUsageIO, granularity='day')
+
+
+@Utils.dumps2response
+def r_disk_usage_io_last_seven_days(uuid):
+    return get_performance_data(uuid=uuid, uuid_field='host_uuid', the_class=HostDiskUsageIO, granularity='seven_days')
+
+

api_route_table.py

@@ -12,6 +12,7 @@ from api import os_template
 from api import log
 from api import host
 from api import performance
+from api import host_performance
 
 
 __author__ = 'James Iter'
@@ -95,7 +96,7 @@ add_rule_api(host.blueprints, '/<ids>', api_func='host.r_get', methods=['GET'])
 add_rule_api(host.blueprints, '', api_func='host.r_get_by_filter', methods=['GET'])
 add_rule_api(host.blueprints, '/_search', api_func='host.r_content_search', methods=['GET'])
 
-# 性能查询
+# Guest 性能查询
 add_rule_api(performance.blueprint, '/cpu_memory', api_func='performance.r_cpu_memory_get_by_filter', methods=['GET'])
 add_rule_api(performance.blueprint, '/traffic', api_func='performance.r_traffic_get_by_filter', methods=['GET'])
 add_rule_api(performance.blueprint, '/disk_io', api_func='performance.r_disk_io_get_by_filter', methods=['GET'])
@@ -135,4 +136,47 @@ add_rule_api(performance.blueprint, '/disk_io/last_day/<uuid>',
 add_rule_api(performance.blueprint, '/disk_io/last_seven_days/<uuid>',
              api_func='performance.r_disk_io_last_seven_days', methods=['GET'])
 
+# Host 性能查询
+add_rule_api(host_performance.blueprint, '/cpu_memory', api_func='host_performance.r_cpu_memory_get_by_filter',
+             methods=['GET'])
+add_rule_api(host_performance.blueprint, '/traffic', api_func='host_performance.r_traffic_get_by_filter',
+             methods=['GET'])
+add_rule_api(host_performance.blueprint, '/disk_io', api_func='host_performance.r_disk_usage_io_get_by_filter',
+             methods=['GET'])
+add_rule_api(host_performance.blueprint, '/cpu_memory/last_hour/<uuid>',
+             api_func='host_performance.r_cpu_memory_last_hour', methods=['GET'])
+
+add_rule_api(host_performance.blueprint, '/cpu_memory/last_six_hours/<uuid>',
+             api_func='host_performance.r_cpu_memory_last_six_hours', methods=['GET'])
+
+add_rule_api(host_performance.blueprint, '/cpu_memory/last_day/<uuid>',
+             api_func='host_performance.r_cpu_memory_last_day', methods=['GET'])
+
+add_rule_api(host_performance.blueprint, '/cpu_memory/last_seven_days/<uuid>',
+             api_func='host_performance.r_cpu_memory_last_seven_days', methods=['GET'])
+
+add_rule_api(host_performance.blueprint, '/traffic/last_hour/<uuid>',
+             api_func='host_performance.r_traffic_last_hour', methods=['GET'])
+
+add_rule_api(host_performance.blueprint, '/traffic/last_six_hours/<uuid>',
+             api_func='host_performance.r_traffic_last_six_hours', methods=['GET'])
+
+add_rule_api(host_performance.blueprint, '/traffic/last_day/<uuid>',
+             api_func='host_performance.r_traffic_last_day', methods=['GET'])
+
+add_rule_api(host_performance.blueprint, '/traffic/last_seven_days/<uuid>',
+             api_func='host_performance.r_traffic_last_seven_days', methods=['GET'])
+
+add_rule_api(host_performance.blueprint, '/disk_io/last_hour/<uuid>',
+             api_func='host_performance.r_disk_usage_io_last_hour', methods=['GET'])
+
+add_rule_api(host_performance.blueprint, '/disk_io/last_six_hours/<uuid>',
+             api_func='host_performance.r_disk_usage_io_last_six_hours', methods=['GET'])
+
+add_rule_api(host_performance.blueprint, '/disk_io/last_day/<uuid>',
+             api_func='host_performance.r_disk_usage_io_last_day', methods=['GET'])
+
+add_rule_api(host_performance.blueprint, '/disk_io/last_seven_days/<uuid>',
+             api_func='host_performance.r_disk_usage_io_last_seven_days', methods=['GET'])
+
 

misc/init.sql

@@ -210,7 +210,7 @@ ALTER TABLE disk_io ADD INDEX (disk_uuid, timestamp);
 
 CREATE TABLE IF NOT EXISTS host_cpu_memory(
     id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
-    host_uuid CHAR(36) NOT NULL,
+    node_id BIGINT UNSIGNED NOT NULL,
     cpu_load INT UNSIGNED NOT NULL,
     memory_available BIGINT UNSIGNED NOT NULL,
     timestamp BIGINT UNSIGNED NOT NULL,
@@ -218,14 +218,14 @@ CREATE TABLE IF NOT EXISTS host_cpu_memory(
     ENGINE=Innodb
     DEFAULT CHARSET=utf8;
 
-ALTER TABLE host_cpu_memory ADD INDEX (host_uuid);
+ALTER TABLE host_cpu_memory ADD INDEX (node_id);
 ALTER TABLE host_cpu_memory ADD INDEX (timestamp);
-ALTER TABLE host_cpu_memory ADD INDEX (host_uuid, timestamp);
+ALTER TABLE host_cpu_memory ADD INDEX (node_id, timestamp);
 
 
 CREATE TABLE IF NOT EXISTS host_traffic(
     id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
-    host_uuid CHAR(36) NOT NULL,
+    node_id BIGINT UNSIGNED NOT NULL,
     name VARCHAR(36) NOT NULL,
     rx_bytes BIGINT UNSIGNED NOT NULL,
     rx_packets BIGINT UNSIGNED NOT NULL,
@@ -240,18 +240,18 @@ CREATE TABLE IF NOT EXISTS host_traffic(
     ENGINE=Innodb
     DEFAULT CHARSET=utf8;
 
-ALTER TABLE host_traffic ADD INDEX (host_uuid);
+ALTER TABLE host_traffic ADD INDEX (node_id);
 ALTER TABLE host_traffic ADD INDEX (rx_bytes);
 ALTER TABLE host_traffic ADD INDEX (rx_packets);
 ALTER TABLE host_traffic ADD INDEX (tx_bytes);
 ALTER TABLE host_traffic ADD INDEX (tx_packets);
 ALTER TABLE host_traffic ADD INDEX (timestamp);
-ALTER TABLE host_traffic ADD INDEX (host_uuid, timestamp);
+ALTER TABLE host_traffic ADD INDEX (node_id, timestamp);
 
 
 CREATE TABLE IF NOT EXISTS host_disk_usage_io(
     id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
-    host_uuid CHAR(36) NOT NULL,
+    node_id BIGINT UNSIGNED NOT NULL,
     mountpoint VARCHAR(255) NOT NULL,
     used BIGINT UNSIGNED NOT NULL,
     rd_req BIGINT UNSIGNED NOT NULL,
@@ -263,7 +263,7 @@ CREATE TABLE IF NOT EXISTS host_disk_usage_io(
     ENGINE=Innodb
     DEFAULT CHARSET=utf8;
 
-ALTER TABLE host_disk_usage_io ADD INDEX (host_uuid);
+ALTER TABLE host_disk_usage_io ADD INDEX (node_id);
 ALTER TABLE host_disk_usage_io ADD INDEX (mountpoint);
 ALTER TABLE host_disk_usage_io ADD INDEX (used);
 ALTER TABLE host_disk_usage_io ADD INDEX (rd_req);
@@ -271,6 +271,6 @@ ALTER TABLE host_disk_usage_io ADD INDEX (rd_bytes);
 ALTER TABLE host_disk_usage_io ADD INDEX (wr_req);
 ALTER TABLE host_disk_usage_io ADD INDEX (wr_bytes);
 ALTER TABLE host_disk_usage_io ADD INDEX (timestamp);
-ALTER TABLE host_disk_usage_io ADD INDEX (host_uuid, mountpoint, timestamp);
+ALTER TABLE host_disk_usage_io ADD INDEX (node_id, mountpoint, timestamp);
 
 

models/__init__.py

@@ -69,6 +69,12 @@ from event_processor import (
     EventProcessor
 )
 
+from host_performance import (
+    HostCPUMemory,
+    HostTraffic,
+    HostDiskUsageIO
+)
+
 
 __author__ = 'James Iter'
 __date__ = '2017/3/21'
@@ -79,7 +85,8 @@ __copyright__ = '(c) 2017 by James Iter.'
 __all__ = [
     'Rules', 'Utils', 'Init', 'Database', 'FilterFieldType', 'Filter', 'EmitKind', 'GuestState', 'DiskState',
     'LogLevel', 'ORM', 'Config', 'Guest', 'Disk', 'BootJob', 'OperateRule', 'OSTemplate', 'GuestXML', 'Log',
-    'EventProcessor', 'ResponseState', 'CPUMemory', 'Traffic', 'DiskIO'
+    'EventProcessor', 'ResponseState', 'CPUMemory', 'Traffic', 'DiskIO', 'HostCPUMemory', 'HostTraffic',
+    'HostDiskUsageIO'
 ]
 
 

models/event_processor.py

@@ -16,7 +16,8 @@ from models import EmitKind
 from models import ResponseState, GuestState, DiskState
 from models.guest import GuestMigrateInfo
 from models.initialize import app, logger
-from models.status import CollectionPerformanceDataKind
+from models.status import CollectionPerformanceDataKind, HostCollectionPerformanceDataKind
+from models import HostCPUMemory, HostTraffic, HostDiskUsageIO
 
 
 __author__ = 'James Iter'
@@ -35,6 +36,9 @@ class EventProcessor(object):
     cpu_memory = CPUMemory()
     traffic = Traffic()
     disk_io = DiskIO()
+    host_cpu_memory = HostCPUMemory()
+    host_traffic = HostTraffic()
+    host_disk_usage_io = HostDiskUsageIO()
 
     @classmethod
     def log_processor(cls):
@@ -255,6 +259,49 @@ class EventProcessor(object):
         else:
             pass
 
+    @classmethod
+    def host_collection_performance_processor(cls):
+        data_kind = cls.message['type']
+        timestamp = ji.Common.ts()
+        timestamp -= (timestamp % 60)
+        data = cls.message['message']['data']
+
+        if data_kind == HostCollectionPerformanceDataKind.cpu_memory.value:
+            cls.host_cpu_memory.node_id = data['node_id']
+            cls.host_cpu_memory.cpu_load = data['cpu_load']
+            cls.host_cpu_memory.memory_available = data['memory_available']
+            cls.host_cpu_memory.timestamp = timestamp
+            cls.host_cpu_memory.create()
+
+        if data_kind == HostCollectionPerformanceDataKind.traffic.value:
+            for item in data:
+                cls.host_traffic.node_id = item['node_id']
+                cls.host_traffic.name = item['name']
+                cls.host_traffic.rx_bytes = item['rx_bytes']
+                cls.host_traffic.rx_packets = item['rx_packets']
+                cls.host_traffic.rx_errs = item['rx_errs']
+                cls.host_traffic.rx_drop = item['rx_drop']
+                cls.host_traffic.tx_bytes = item['tx_bytes']
+                cls.host_traffic.tx_packets = item['tx_packets']
+                cls.host_traffic.tx_errs = item['tx_errs']
+                cls.host_traffic.tx_drop = item['tx_drop']
+                cls.host_traffic.timestamp = timestamp
+                cls.host_traffic.create()
+
+        if data_kind == HostCollectionPerformanceDataKind.disk_usage_io.value:
+            cls.host_disk_usage_io.node_id = data['node_id']
+            cls.host_disk_usage_io.mountpoint = data['mountpoint']
+            cls.host_disk_usage_io.used = data['used']
+            cls.host_disk_usage_io.rd_req = data['rd_req']
+            cls.host_disk_usage_io.rd_bytes = data['rd_bytes']
+            cls.host_disk_usage_io.wr_req = data['wr_req']
+            cls.host_disk_usage_io.wr_bytes = data['wr_bytes']
+            cls.host_disk_usage_io.timestamp = timestamp
+            cls.host_disk_usage_io.create()
+
+        else:
+            pass
+
     @classmethod
     def launch(cls):
         while True:
@@ -286,6 +333,9 @@ class EventProcessor(object):
                 elif cls.message['kind'] == EmitKind.collection_performance.value:
                     cls.collection_performance_processor()
 
+                elif cls.message['kind'] == EmitKind.host_collection_performance.value:
+                    cls.host_collection_performance_processor()
+
                 else:
                     pass
 

models/host_performance.py

@@ -0,0 +1,126 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+
+from filter import FilterFieldType
+from orm import ORM
+
+
+__author__ = 'James Iter'
+__date__ = '2017/8/7'
+__contact__ = 'james.iter.cn@gmail.com'
+__copyright__ = '(c) 2017 by James Iter.'
+
+
+class HostCPUMemory(ORM):
+
+    _table_name = 'host_cpu_memory'
+    _primary_key = 'id'
+
+    def __init__(self):
+        super(HostCPUMemory, self).__init__()
+        self.id = 0
+        self.node_id = None
+        self.cpu_load = None
+        self.memory_available = None
+        self.timestamp = None
+
+    @staticmethod
+    def get_filter_keywords():
+        return {
+            'id': FilterFieldType.INT.value,
+            'node_id': FilterFieldType.STR.value,
+            'cpu_load': FilterFieldType.INT.value,
+            'timestamp': FilterFieldType.INT.value
+        }
+
+    @staticmethod
+    def get_allow_update_keywords():
+        return []
+
+    @staticmethod
+    def get_allow_content_search_keywords():
+        return []
+
+
+class HostTraffic(ORM):
+
+    _table_name = 'host_traffic'
+    _primary_key = 'id'
+
+    def __init__(self):
+        super(HostTraffic, self).__init__()
+        self.id = 0
+        self.node_id = None
+        self.name = None
+        self.rx_bytes = None
+        self.rx_packets = None
+        self.rx_errs = None
+        self.rx_drop = None
+        self.tx_bytes = None
+        self.tx_packets = None
+        self.tx_errs = None
+        self.tx_drop = None
+        self.timestamp = None
+
+    @staticmethod
+    def get_filter_keywords():
+        return {
+            'id': FilterFieldType.INT.value,
+            'node_id': FilterFieldType.STR.value,
+            'name': FilterFieldType.STR.value,
+            'rx_bytes': FilterFieldType.INT.value,
+            'rx_packets': FilterFieldType.INT.value,
+            'tx_bytes': FilterFieldType.INT.value,
+            'tx_packets': FilterFieldType.INT.value,
+            'timestamp': FilterFieldType.INT.value
+        }
+
+    @staticmethod
+    def get_allow_update_keywords():
+        return []
+
+    @staticmethod
+    def get_allow_content_search_keywords():
+        return []
+
+
+class HostDiskUsageIO(ORM):
+
+    _table_name = 'host_disk_usage_io'
+    _primary_key = 'id'
+
+    def __init__(self):
+        super(HostDiskUsageIO, self).__init__()
+        self.id = 0
+        self.node_id = None
+        self.mountpoint = None
+        self.used = None
+        self.rd_req = None
+        self.rd_bytes = None
+        self.wr_req = None
+        self.wr_bytes = None
+        self.timestamp = None
+
+    @staticmethod
+    def get_filter_keywords():
+        return {
+            'id': FilterFieldType.INT.value,
+            'node_id': FilterFieldType.STR.value,
+            'mountpoint': FilterFieldType.STR.value,
+            'used': FilterFieldType.INT.value,
+            'rd_req': FilterFieldType.INT.value,
+            'rd_bytes': FilterFieldType.INT.value,
+            'wr_req': FilterFieldType.INT.value,
+            'wr_bytes': FilterFieldType.INT.value,
+            'timestamp': FilterFieldType.INT.value
+        }
+
+    @staticmethod
+    def get_allow_update_keywords():
+        return []
+
+    @staticmethod
+    def get_allow_content_search_keywords():
+        return []
+

models/status.py

@@ -17,6 +17,7 @@ class EmitKind(IntEnum):
     host_event = 2
     response = 3
     collection_performance = 4
+    host_collection_performance = 5
 
 
 class GuestState(IntEnum):
@@ -82,3 +83,9 @@ class CollectionPerformanceDataKind(IntEnum):
     traffic = 1
     disk_io = 2
 
+
+class HostCollectionPerformanceDataKind(IntEnum):
+    cpu_memory = 0
+    traffic = 1
+    disk_usage_io = 2
+