ping_mon/pingmon.py

86 lines
2.5 KiB
Python

from termios import TAB3
import threading
from types import NoneType
from prometheus_client import Counter
import json
from telnetlib import theNULL
import pingparsing
from prometheus_client import CollectorRegistry, Gauge, push_to_gateway
tests={
"1": {
'dest': "66.29.128.140",
'name': 'African webserver',
'packet_loss_rate_permitted': 50,
'rtt_max_permitted': 150,
},
"2": {
'dest': "8.8.8.8",
'name': 'Google DNS',
'packet_loss_rate_permitted': 0,
'rtt_max_permitted': 150,
},
"3": {
'dest': "1.1.1.1",
'name': 'Cloudflare DNS',
'packet_loss_rate_permitted': 0,
'rtt_max_permitted': 150,
},
}
def pinger(test_id):
print("Testing {} on IP {} with RTT threshold of {} and packet loss max of {}".format(
tests[test_id]['name'],tests[test_id]['dest'],
tests[test_id]['rtt_max_permitted'],tests[test_id]['packet_loss_rate_permitted']))
dest=tests[test_id]['dest']
name=tests[test_id]['name']
rtt_max_permitted=tests[test_id]['rtt_max_permitted']
packet_loss_rate_permitted=tests[test_id]['packet_loss_rate_permitted']
ping_parser = pingparsing.PingParsing()
transmitter = pingparsing.PingTransmitter()
transmitter.destination = dest
transmitter.count = 10
transmitter.timeout=2
# while 1:
result = transmitter.ping()
data=ping_parser.parse(result).as_dict()
packet_loss_rate=data["packet_loss_rate"]
rtt_max=data["rtt_max"]
notify=0
print("Dest: {} Loss: {}% RTT: {}ms".format(dest,packet_loss_rate,rtt_max))
if rtt_max>rtt_max_permitted:
print("ERROR: rtt_max_permitted exceeded!")
notify=1
if packet_loss_rate>packet_loss_rate_permitted:
print("ERROR: packet_loss_rate_permitted exceeded!")
notify=1
if notify:
print(json.dumps(ping_parser.parse(result).as_dict(), indent=4))
registry = CollectorRegistry()
g_pl = Gauge('packet_loss_rate', 'Amt of packet loss', ['destination_ip'],registry=registry )
g_pl.labels(dest).set(packet_loss_rate)
if not type(rtt_max)==NoneType:
g_rtt = Gauge('rtt_max', 'Round trip time', ['destination_ip'],registry=registry )
g_rtt.labels(dest).set(rtt_max)
push_to_gateway('10.10.110.250:9091', job='cory_test_job1', registry=registry)
if __name__ == "__main__":
while 1:
for _id,_item in tests.items():
pinger(_id)
# t = threading.Thread(target=pinger, args=(_id,))
# t.start()