Skip to content

CIRCLE / monitor-client

Go to a project
Commit 99be2c47 by Gregory Nagy
Options

Debug messages and exception handling added.

parent ba7f4804
Showing with 158 additions and 121 deletions
src/client.py
...@@ -12,132 +12,169 @@ logging.basicConfig() ...@@ -12,132 +12,169 @@ logging.basicConfig()
class Client: class Client:
def __init__(self, config): def __init__(self, config):
""" """
Constructor of the client requires a configuration provided by cnfparse Constructor of the client requires a configuration provided by cnfparse
modul. It is a dictionary: {debugMode} modul. It is a dictionary: {debugMode}
""" """
hostname = socket.gethostname().split('.') hostname = socket.gethostname().split('.')
hostname.reverse() hostname.reverse()
self.name = "circle." + ".".join(hostname) self.name = "circle." + ".".join(hostname)
self.server_address = str(os.getenv("GRAPHITE_SERVER_ADDRESS")) self.server_address = str(os.getenv("GRAPHITE_SERVER_ADDRESS"))
self.server_port = int(os.getenv("GRAPHITE_SERVER_PORT").__str__()) self.server_port = int(os.getenv("GRAPHITE_SERVER_PORT"))
self.debugMode = config["debugMode"] self.debugMode = config["debugMode"]
self.amqp_user = str(os.getenv("GRAPHITE_AMQP_USER")) self.amqp_user = str(os.getenv("GRAPHITE_AMQP_USER"))
self.amqp_pass = str(os.getenv("GRAPHITE_AMQP_PASSWORD")) self.amqp_pass = str(os.getenv("GRAPHITE_AMQP_PASSWORD"))
self.amqp_queue = str(os.getenv("GRAPHITE_AMQP_QUEUE")) self.amqp_queue = str(os.getenv("GRAPHITE_AMQP_QUEUE"))
self.amqp_vhost = str(os.getenv("GRAPHITE_AMQP_VHOST")) self.amqp_vhost = str(os.getenv("GRAPHITE_AMQP_VHOST"))
self.beat = 1 self.beat = 1
if self.server_address is None:
print("GRAPHITE_SERVER_ADDRESS cannot be found in environmental "
"variables"
)
if self.server_port is None:
print("GRAPHITE_SERVER_PORT cannot be found in environmental "
"variables. (AMQP standard is: 5672"
)
if self.amqp_user is None or self.amqp_user is None:
print("GRAPHITE_AMQP_USER or GRAPHITE_AMQP_PASSWORD cannot be "
"found in environmental variables. (AMQP standard is: "
"guest-guest)"
)
if self.amqp_queue is None or self.amqp_vhost is None:
print("GRAPHITE_AMQP_QUEUE or GRAPHITE_AMQP_VHOST cannot be "
"found in environmental variables."
)
def __connect(self):
"""
This method creates the connection to the queue of the graphite server.
Returns true if the connection was successful.
"""
try:
credentials = pika.PlainCredentials(self.amqp_user, self.amqp_pass)
params = pika.ConnectionParameters(host=self.server_address,
port=self.server_port,
virtual_host=self.amqp_vhost,
credentials=credentials)
self.connection = pika.BlockingConnection(params)
self.channel = self.connection.channel()
return True
except:
raise
def __disconnect(self): def __connect(self):
""" """
Break up the connection to the graphite server. This method creates the connection to the queue of the graphite server.
""" Returns true if the connection was successful.
self.channel.close() """
self.connection.close() try:
credentials = pika.PlainCredentials(self.amqp_user, self.amqp_pass)
params = pika.ConnectionParameters(host=self.server_address,
port=self.server_port,
virtual_host=self.amqp_vhost,
credentials=credentials)
self.connection = pika.BlockingConnection(params)
self.channel = self.connection.channel()
return True
except RuntimeError:
print ("[ERROR] Cannot connect to the server. "
"Parameters could be wrong."
)
def __send(self, message): def __disconnect(self):
""" """
Send the message given in the parameters. Break up the connection to the graphite server.
""" """
self.channel.basic_publish(exchange=self.amqp_queue, try:
routing_key='', body="\n".join(message)) self.channel.close()
self.connection.close()
except RuntimeError:
print("[ERROR] An error has occured while disconnecting from the "
"server."
)
def __collectFromNode(self, metricCollectors): def __send(self, message):
""" """
It harvests the given metrics in the metricCollectors list. This list Send the message given in the parameters.
should be provided by the collectables modul. """
""" try:
metrics = [] self.channel.basic_publish(exchange=self.amqp_queue,
for collector in metricCollectors: routing_key='', body="\n".join(message))
if (self.beat % collector[1]) is 0: return True
stat = collector[0]() except:
metrics.append((self.name + "." + print("[ERROR] An error has occured while sending metrics to the "
stat.name + " %d" % (stat.value) "server."
+ " %d" % (time.time()) )
)) return False
return metrics
def __collectFromVMs(self): def __collectFromNode(self, metricCollectors):
metrics = [] """
running_vms = [] It harvests the given metrics in the metricCollectors list. This list
for entry in psutil.get_process_list(): should be provided by the collectables modul.
if entry.name in "kvm": """
search = [cmd_param_index for cmd_param_index, cmd_param in metrics = []
enumerate(entry.as_dict()["cmdline"]) for collector in metricCollectors:
if cmd_param == "-name"] if (self.beat % collector[1]) is 0:
memory = [cmd_param_index for cmd_param_index, cmd_param in stat = collector[0]()
enumerate(entry.as_dict()["cmdline"]) metrics.append((self.name + "." +
if cmd_param == "-m"] stat.name + " %d" % (stat.value)
running_vms.append([entry.as_dict()["cmdline"][search[0] + 1], + " %d" % (time.time())
entry.pid, ))
int(entry.as_dict()["cmdline"][memory[0] + 1])]) return metrics
for vm in running_vms:
vm_proc = psutil.Process(vm[1])
metrics.append((self.name + "." + "kvm." +
vm[0] + "." + "memory.usage." +
" %d" % (vm_proc.get_memory_percent() / 100 * vm[2])
+ " %d" % (time.time())
))
metrics.append((self.name + "." + "kvm." +
vm[0] + "." + "cpu.usage" +
" %d" % (vm_proc.get_cpu_times().system +
vm_proc.get_cpu_times().user)
+ " %d" % (time.time())
))
return metrics
def getMaxFrequency(self, metricCollectors=[]): def __collectFromVMs(self):
max = metricCollectors[0][1] metrics = []
for item in metricCollectors: running_vms = []
if max < item[1]: for entry in psutil.get_process_list():
max = item[1] if entry.name in "kvm":
return max search = [cmd_param_index for cmd_param_index, cmd_param in
enumerate(entry.as_dict()["cmdline"])
if cmd_param == "-name"]
memory = [cmd_param_index for cmd_param_index, cmd_param in
enumerate(entry.as_dict()["cmdline"])
if cmd_param == "-m"]
running_vms.append([entry.as_dict()["cmdline"][search[0] + 1],
entry.pid,
int(entry.as_dict()["cmdline"][
memory[0] + 1])])
for vm in running_vms:
vm_proc = psutil.Process(vm[1])
metrics.append((self.name + "." + "kvm." +
vm[0] + "." + "memory.usage." +
" %d" % (vm_proc.get_memory_percent() / 100 * vm[2])
+ " %d" % (time.time())
))
metrics.append((self.name + "." + "kvm." +
vm[0] + "." + "cpu.usage" +
" %d" % (vm_proc.get_cpu_times().system +
vm_proc.get_cpu_times().user)
+ " %d" % (time.time())
))
return metrics
def startReporting(self, metricCollectors=[], debugMode=False): def getMaxFrequency(self, metricCollectors=[]):
""" max = metricCollectors[0][1]
Call this method to start reporting to the server, it needs the for item in metricCollectors:
metricCollectors parameter that should be provided by the collectables if max < item[1]:
modul to work properly. max = item[1]
""" return max
if self.__connect() is False:
print("An error has occured while connecting to the server on %s" def startReporting(self, metricCollectors=[], debugMode=False):
% (self.server_address + ":" + str(self.server_port))) """
else: Call this method to start reporting to the server, it needs the
print("Connection established to %s on port %s. \ metricCollectors parameter that should be provided by the collectables
modul to work properly.
"""
if self.__connect() is False:
print("[ERROR] An error has occured while connecting to the "
"server on %s."
% (self.server_address + ":" + str(self.server_port)))
else:
print("[SUCCESS] Connection established to %s on port %s. \
Clientname: %s" Clientname: %s"
% (self.server_address, self.server_port, % (self.server_address, self.server_port,
self.name)) self.name))
try: try:
maxFrequency = self.getMaxFrequency(metricCollectors) maxFrequency = self.getMaxFrequency(metricCollectors)
while True: while True:
metrics = self.__collectFromNode(metricCollectors) metrics = self.__collectFromNode(metricCollectors)
metrics.append(self.__collectFromVMs()) vmMetrics = self.__collectFromVMs()
if self.debugMode == "True": if len(vmMetrics) is not 0:
print(metrics) metrics.append(vmMetrics)
self.__send(metrics) if self.debugMode == "True" and len(metrics) is not 0:
time.sleep(1) print("[*" + self.beat + "] " + metrics)
self.beat = self.beat + 1 if self.__send(metrics) is False:
if ((self.beat % (maxFrequency + 1)) is 0): raise RuntimeError
self.beat = 1 time.sleep(1)
except KeyboardInterrupt: self.beat = self.beat + 1
print("Reporting has stopped by the user. Exiting...") if ((self.beat % (maxFrequency + 1)) is 0):
finally: self.beat = 1
self.__disconnect() except KeyboardInterrupt:
print("[x] Reporting has stopped by the user. Exiting...")
finally:
self.__disconnect()
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or sign in to comment