444 lines
14 KiB
Python
444 lines
14 KiB
Python
#!/usr/bin/env python
|
|
|
|
# Copyright 2014 Cisco Systems, Inc. All rights reserved.
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
|
# not use this file except in compliance with the License. You may obtain
|
|
# a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
|
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
|
# License for the specific language governing permissions and limitations
|
|
# under the License.
|
|
#
|
|
|
|
|
|
'''
|
|
Module for parsing statistical output from Ganglia (gmond) server
|
|
The module opens a socket connection to collect statistical data.
|
|
It parses the raw data in xml format.
|
|
|
|
The data from ganglia/gmond is in a heirarchical xml format as below:
|
|
<CLUSTER>
|
|
<HOST..>
|
|
<METRIC ../>
|
|
<METRIC ../>
|
|
:
|
|
</HOST>
|
|
:
|
|
<HOST..>
|
|
<METRIC ../>
|
|
<METRIC ../>
|
|
</HOST>
|
|
</CLUSTER>
|
|
|
|
## Usage:
|
|
Using the module is simple.
|
|
|
|
1. instantiate the Monitor with the gmond server ip and port to poll.
|
|
|
|
gmon = Monitor("172.22.191.151", 8649)
|
|
|
|
2. Start the monitoring thread
|
|
gmon.start_monitoring_thread(frequency, count)
|
|
|
|
< run tests/tasks>
|
|
|
|
gmon.stop_monitoring_thread()
|
|
|
|
3. Collecting stats:
|
|
cpu_metric = gmon.build_cpu_metric()
|
|
|
|
Returns a dictionary object with all the cpu stats for each
|
|
node
|
|
|
|
|
|
'''
|
|
|
|
import datetime
|
|
import re
|
|
import socket
|
|
import subprocess
|
|
from threading import Thread
|
|
import time
|
|
|
|
from lxml import etree
|
|
|
|
class MonitorExecutor(Thread):
|
|
'''
|
|
Thread handler class to asynchronously collect stats
|
|
'''
|
|
THREAD_STOPPED = 0
|
|
THREAD_RUNNING = 1
|
|
|
|
def __init__(self, gmond_svr, gmond_port, freq=5, count=5):
|
|
super(MonitorExecutor, self).__init__()
|
|
self.gmond_svr_ip = gmond_svr
|
|
self.gmond_port = gmond_port
|
|
|
|
self.freq = freq
|
|
self.count = count
|
|
|
|
self.force_stop = False
|
|
self.thread_status = MonitorExecutor.THREAD_STOPPED
|
|
|
|
# This dictionary always holds the latest metric.
|
|
self.gmond_parsed_tree_list = []
|
|
|
|
|
|
def run(self):
|
|
'''
|
|
The thread runnable method.
|
|
The function will periodically poll the gmond server and
|
|
collect the metrics.
|
|
'''
|
|
self.thread_status = MonitorExecutor.THREAD_RUNNING
|
|
|
|
count = self.count
|
|
while count > 0:
|
|
if self.force_stop:
|
|
self.thread_status = MonitorExecutor.THREAD_STOPPED
|
|
return
|
|
|
|
self.parse_gmond_xml_data()
|
|
count -= 1
|
|
time.sleep(self.freq)
|
|
self.thread_status = MonitorExecutor.THREAD_STOPPED
|
|
|
|
|
|
def set_force_stop(self):
|
|
'''
|
|
Setting the force stop flag to stop the thread. By default
|
|
the thread stops after the specific count/iterations is reached
|
|
'''
|
|
self.force_stop = True
|
|
|
|
def parse_gmond_xml_data(self):
|
|
'''
|
|
Parse gmond data (V2)
|
|
Retrieve the ganglia stats from the aggregation node
|
|
:return: None in case of error or a dictionary containing the stats
|
|
'''
|
|
gmond_parsed_tree = {}
|
|
raw_data = self.retrieve_stats_raw()
|
|
|
|
if raw_data is None or len(raw_data) == 0:
|
|
print("Failed to retrieve stats from server")
|
|
return
|
|
|
|
xtree = etree.XML(raw_data)
|
|
############################################
|
|
# Populate cluster information.
|
|
############################################
|
|
for elem in xtree.iter('CLUSTER'):
|
|
gmond_parsed_tree['CLUSTER-NAME'] = str(elem.get('NAME'))
|
|
gmond_parsed_tree['LOCALTIME'] = str(elem.get('LOCALTIME'))
|
|
gmond_parsed_tree['URL'] = str(elem.get('URL'))
|
|
|
|
host_list = []
|
|
for helem in elem.iterchildren():
|
|
host = {}
|
|
host['NAME'] = str(helem.get('NAME'))
|
|
host['IP'] = str(helem.get('IP'))
|
|
host['REPORTED'] = str(helem.get('REPORTED'))
|
|
host['TN'] = str(helem.get('TN'))
|
|
host['TMAX'] = str(helem.get('TMAX'))
|
|
host['DMAX'] = str(helem.get('DMAX'))
|
|
host['LOCATION'] = str(helem.get('LOCATION'))
|
|
host['GMOND_STARTED'] = str(helem.get('GMOND_STARTED'))
|
|
|
|
mlist = []
|
|
for metric in helem.iterchildren():
|
|
mdic = {}
|
|
mdic['NAME'] = str(metric.get('NAME'))
|
|
mdic['VAL'] = str(metric.get('VAL'))
|
|
mlist.append(mdic)
|
|
|
|
host['metrics'] = mlist
|
|
host_list.append(host)
|
|
|
|
gmond_parsed_tree['hosts'] = host_list
|
|
stat_dt = datetime.datetime.now()
|
|
gmond_parsed_tree['dt'] = stat_dt
|
|
self.gmond_parsed_tree_list.append(gmond_parsed_tree)
|
|
|
|
|
|
def retrieve_stats_raw(self):
|
|
'''
|
|
Retrieve stats from the gmond process.
|
|
'''
|
|
soc = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
soc.settimeout(10)
|
|
try:
|
|
soc.connect((self.gmond_svr_ip, self.gmond_port))
|
|
except socket.error as exp:
|
|
print("Connection failure host: %s [%s]" % (self.gmond_svr_ip, exp))
|
|
return None
|
|
|
|
data = ""
|
|
while True:
|
|
try:
|
|
rbytes = soc.recv(4096)
|
|
except socket.error as exp:
|
|
print("Read failed for host: ", str(exp))
|
|
return None
|
|
|
|
if len(rbytes) == 0:
|
|
break
|
|
data += rbytes
|
|
|
|
soc.close()
|
|
return data
|
|
|
|
|
|
class Monitor(object):
|
|
gmond_svr_ip = None
|
|
gmond_port = None
|
|
gmond_parsed_tree = {}
|
|
|
|
def __init__(self, gmond_svr, gmond_port=8649):
|
|
'''
|
|
The constructor simply sets the values of the gmond server and port.
|
|
'''
|
|
self.gmond_svr_ip = gmond_svr
|
|
self.gmond_port = gmond_port
|
|
# List of all stats.
|
|
self.gmond_parsed_tree_list = []
|
|
# series for all cpu loads
|
|
self.cpu_res = {}
|
|
|
|
self.mon_thread = None
|
|
|
|
def start_monitoring_thread(self, freq=10, count=10):
|
|
'''
|
|
Start the monitoring thread.
|
|
'''
|
|
self.mon_thread = MonitorExecutor(self.gmond_svr_ip,
|
|
self.gmond_port, freq, count)
|
|
self.mon_thread.start()
|
|
|
|
|
|
def stop_monitoring_thread(self):
|
|
self.mon_thread.set_force_stop()
|
|
self.gmond_parsed_tree_list = self.mon_thread.gmond_parsed_tree_list
|
|
|
|
|
|
def strip_raw_telnet_output(self, raw_data):
|
|
'''
|
|
When using the retrieve_stats_raw_telent api, the raw data
|
|
has some additional text along with the xml data. We need to
|
|
strip that before we can invoke pass it through the lxml parser.
|
|
'''
|
|
data = ""
|
|
xml_flag = False
|
|
for line in raw_data.splitlines():
|
|
if re.match(r".*<?xml version.*", line):
|
|
xml_flag = True
|
|
if xml_flag:
|
|
data += line + "\n"
|
|
|
|
return data
|
|
|
|
|
|
def retrieve_stats_raw_telnet(self):
|
|
'''
|
|
This way of retrieval is to create a subprocess and execute
|
|
the telnet command on the port to retrieve the xml raw data.
|
|
'''
|
|
cmd = "telnet " + self.gmond_svr_ip + " " + str(self.gmond_port)
|
|
print("cmd: ", cmd)
|
|
port = str(self.gmond_port)
|
|
|
|
proc = subprocess.Popen(["telnet", self.gmond_svr_ip, port],
|
|
stdout=subprocess.PIPE)
|
|
(output, _) = proc.communicate()
|
|
|
|
newout = self.strip_raw_telnet_output(output)
|
|
return newout
|
|
|
|
|
|
def get_host_list(self, gmond_parsed_tree):
|
|
'''
|
|
Function returns all the hosts {} as a list.
|
|
'''
|
|
return gmond_parsed_tree['hosts']
|
|
|
|
|
|
def get_metric_value(self, parsed_node, host_name, name):
|
|
'''
|
|
The function returns the value of a specific metric, given
|
|
the host name and the metric name to collect.
|
|
'''
|
|
for host in parsed_node['hosts']:
|
|
if host['NAME'] == host_name:
|
|
for metric in host['metrics']:
|
|
if metric['NAME'] == name:
|
|
return metric['VAL']
|
|
|
|
return 0
|
|
|
|
def get_aggregate_cpu_usage(self, parsed_node, host_name):
|
|
'''
|
|
The function returns the aggregate CPU usage for a specific host.
|
|
eqation: [user cpu + system cpu * no of cpu /100]
|
|
'''
|
|
cpu_user = float(self.get_metric_value(parsed_node, host_name, "cpu_user"))
|
|
cpu_system = float(self.get_metric_value(parsed_node, host_name, "cpu_system"))
|
|
cpu_num = int(self.get_metric_value(parsed_node, host_name, "cpu_num"))
|
|
|
|
return (cpu_user + cpu_system) * cpu_num / 100
|
|
|
|
|
|
def build_cpu_metrics(self):
|
|
'''Add a new set of cpu metrics to the results dictionary self.cpu_res
|
|
The result dest dictionary should look like this:
|
|
key = host IP, value = list of cpu load where the
|
|
the first value is the baseline value followed by 1 or more
|
|
values collected during the test
|
|
{
|
|
'10.0.0.1': [ 0.03, 1.23, 1.20 ],
|
|
'10.0.0.2': [ 0.10, 1.98, 2.72 ]
|
|
}
|
|
After another xml is decoded:
|
|
{
|
|
'10.0.0.1': [ 0.03, 1.23, 1.20, 1.41 ],
|
|
'10.0.0.2': [ 0.10, 1.98, 2.72, 2.04 ]
|
|
}
|
|
Each value in the list is the cpu load calculated as
|
|
(cpu_user + cpu_system) * num_cpu / 100
|
|
The load_five metric cannot be used as it is the average for last 5'
|
|
'''
|
|
cpu_res = {}
|
|
for parsed_node in self.gmond_parsed_tree_list:
|
|
for host in self.get_host_list(parsed_node):
|
|
host_ip = host['IP']
|
|
cpu_num = 0
|
|
cpu_user = 0.0
|
|
cpu_system = 0.0
|
|
|
|
cpu_user = float(self.get_metric_value(parsed_node, host['NAME'], "cpu_user"))
|
|
cpu_system = float(self.get_metric_value(parsed_node, host['NAME'], "cpu_system"))
|
|
cpu_num = int(self.get_metric_value(parsed_node, host['NAME'], "cpu_num"))
|
|
cpu_load = round(((cpu_user + cpu_system) * cpu_num) / 100, 2)
|
|
try:
|
|
cpu_res[host_ip].append(cpu_load)
|
|
except KeyError:
|
|
cpu_res[host_ip] = [cpu_load]
|
|
|
|
return cpu_res
|
|
|
|
def get_formatted_datetime(self, parsed_node):
|
|
'''
|
|
Returns the data in formated string. This is the
|
|
time when the last stat was collected.
|
|
'''
|
|
now = parsed_node['dt']
|
|
fmt_dt = "[" + str(now.hour) + ":" + str(now.minute) + \
|
|
":" + str(now.second) + "]"
|
|
return fmt_dt
|
|
|
|
|
|
def get_formatted_host_row(self, host_list):
|
|
'''
|
|
Returns the hosts in formated order (for printing purposes)
|
|
'''
|
|
row_str = "".ljust(10)
|
|
for host in host_list:
|
|
row_str += host['NAME'].ljust(15)
|
|
return row_str
|
|
|
|
def get_formatted_metric_row(self, parsed_node, metric, justval):
|
|
'''
|
|
Returns a specific metric for all hosts in the same row
|
|
in formated string (for printing)
|
|
'''
|
|
host_list = self.get_host_list(parsed_node)
|
|
|
|
row_str = metric.ljust(len(metric) + 2)
|
|
for host in host_list:
|
|
val = self.get_metric_value(parsed_node, host['NAME'], metric)
|
|
row_str += str(val).ljust(justval)
|
|
return row_str
|
|
|
|
|
|
def dump_cpu_stats(self):
|
|
'''
|
|
Print the CPU stats
|
|
'''
|
|
hl_len = 80
|
|
print("-" * hl_len)
|
|
print("CPU Statistics: ", end=' ')
|
|
|
|
for parsed_node in self.gmond_parsed_tree_list:
|
|
hosts = self.get_host_list(parsed_node)
|
|
|
|
print(self.get_formatted_datetime(parsed_node))
|
|
print(self.get_formatted_host_row(hosts))
|
|
print("-" * hl_len)
|
|
print(self.get_formatted_metric_row(parsed_node, "cpu_user", 18))
|
|
print(self.get_formatted_metric_row(parsed_node, "cpu_system", 18))
|
|
|
|
print("Aggregate ", end=' ')
|
|
for host in hosts:
|
|
print(str(self.get_aggregate_cpu_usage(parsed_node,
|
|
host['NAME'])).ljust(16), end=' ')
|
|
print("\n")
|
|
|
|
def dump_gmond_parsed_tree(self):
|
|
'''
|
|
Display the full tree parsed from the gmond server stats.
|
|
'''
|
|
hl_len = 60
|
|
|
|
for parsed_node in self.gmond_parsed_tree_list:
|
|
print("%-20s (%s) URL: %s " %
|
|
(parsed_node['CLUSTER-NAME'],
|
|
parsed_node['LOCALTIME'],
|
|
parsed_node['URL']))
|
|
print("-" * hl_len)
|
|
|
|
row_str = " ".ljust(9)
|
|
for host in parsed_node['hosts']:
|
|
row_str += host['NAME'].ljust(15)
|
|
row_str += "\n"
|
|
print(row_str)
|
|
print("-" * hl_len)
|
|
metric_count = len(parsed_node['hosts'][0]['metrics'])
|
|
for count in range(0, metric_count):
|
|
row_str = ""
|
|
host = parsed_node['hosts'][0]
|
|
row_str += parsed_node['hosts'][0]['metrics'][count]['NAME'].ljust(18)
|
|
for host in parsed_node['hosts']:
|
|
val = str(self.get_metric_value(parsed_node, host['NAME'],
|
|
host['metrics'][count]['NAME']))
|
|
row_str += val.ljust(12)
|
|
|
|
row_str += str(parsed_node['hosts'][0]).ljust(5)
|
|
|
|
print(row_str)
|
|
|
|
|
|
##################################################
|
|
# Only invoke the module directly for test purposes. Should be
|
|
# invoked from pns script.
|
|
##################################################
|
|
def main():
|
|
print("main: monitor")
|
|
gmon = Monitor("172.22.191.151", 8649)
|
|
gmon.start_monitoring_thread(freq=5, count=20)
|
|
print("wait for 15 seconds")
|
|
time.sleep(20)
|
|
print("Now force the thread to stop")
|
|
gmon.stop_monitoring_thread()
|
|
gmon.dump_cpu_stats()
|
|
|
|
cpu_metric = gmon.build_cpu_metrics()
|
|
print("cpu_metric: ", cpu_metric)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|