#!/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: ",
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 ",
for host in hosts:
print str(self.get_aggregate_cpu_usage(parsed_node,
host['NAME'])).ljust(16),
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()