140 lines
4.5 KiB
Python
140 lines
4.5 KiB
Python
import ipaddress, os, re, time, asyncio
|
|
# SNMP
|
|
from pysnmp.hlapi.v3arch.asyncio import *
|
|
# QoL
|
|
from typing import Annotated, Final
|
|
# functions
|
|
from db.collectiveWriter import ALLdbIDRACWriter
|
|
# idrac SNMP
|
|
from idrac import idrac78, idrac9
|
|
# Cisco hosts
|
|
from cisco import lob, nyater
|
|
|
|
# SNMP ENV-------------------------------------------
|
|
ROUND_PREC: Final[int] = int(os.getenv("ROUND_PREC", 2))
|
|
|
|
SNMPUSER: Final[str] = os.getenv("SNMPUSER", None)
|
|
SNMPPRIVKEY: Final[str] = os.getenv("SNMPPRIVKEY", None)
|
|
SNMPAUTHKEY: Final[str] = os.getenv("SNMPAUTHKEY", None)
|
|
# Right now I'll only use SHA
|
|
# SNMPAUTHPROTO: Final[str] = os.getenv("SNMPAUTHPROTO", "SHA")
|
|
# SNMPPRIVPROTO: Final[str] = os.getenv("SNMPPRIVPROTO", "SHA")
|
|
SNMPORT: Final[int] = int(os.getenv("SNMPORT", 161))
|
|
|
|
|
|
# Check if SNMP ENV are empty
|
|
if not SNMPUSER or not SNMPPRIVKEY or not SNMPAUTHKEY:
|
|
raise Exception("No SNMP user or/and PrivAuth passed")
|
|
|
|
# Program ENV---------------------------------------
|
|
try:
|
|
IDRAC78_HOST_LIST: Final[list] = os.getenv("IDRAC78_HOST_LIST", []).split(";")
|
|
IDRAC9_HOST_LIST: Final[list] = os.getenv("IDRAC9_HOST_LIST", []).split(";")
|
|
except:
|
|
raise Exception("No IDRAC hosts variable")
|
|
try:
|
|
CISCO_3750_HOST_LIST: Final[list] = os.getenv("CISCO_3750_HOST_LIST", []).split(";")
|
|
except:
|
|
print("No 3750-X devices passed.\nContinuing with what is available.\n")
|
|
CISCO_3750_HOST_LIST = []
|
|
try:
|
|
CISCO_NEXUS_HOST_LIST: Final[list] = os.getenv("CISCO_NEXUS_HOST_LIST", []).split(";")
|
|
except:
|
|
print("No Nexus devices passed.\nContinuing with what is available.\n")
|
|
CISCO_NEXUS_HOST_LIST = []
|
|
|
|
GET_INTERVAL: Final[int] = int(os.getenv("GET_INTERVAL", 60))
|
|
|
|
|
|
# Flightchecks-------------------------------------------
|
|
# if HOST_LIST empty, raise Exception No hosts passed
|
|
# if not IDRAC78_HOST_LIST and not IDRAC9_HOST_LIST:
|
|
# raise Exception("No IDRAC hosts passed\nExiting...")
|
|
# for host in HOST_LIST check if valid IP and create a list with strings
|
|
|
|
try:
|
|
for idrac78Host in IDRAC78_HOST_LIST:
|
|
try:
|
|
ip = str(ipaddress.IPv4Address(idrac78Host))
|
|
except ValueError:
|
|
raise Exception(f" IP {ip} for IDRAC7/8 devices is invalid.\nExiting...")
|
|
except:
|
|
IDRAC78_HOST_LIST = []
|
|
|
|
|
|
for cisco3750Host in CISCO_3750_HOST_LIST:
|
|
try:
|
|
ip = str(ipaddress.IPv4Address(cisco3750Host))
|
|
except ValueError:
|
|
raise Exception(f" IP {ip} for Cisco devices is invalid.\nExiting...")
|
|
|
|
|
|
for ciscoNexusHost in CISCO_NEXUS_HOST_LIST:
|
|
try:
|
|
ip = str(ipaddress.IPv4Address(ciscoNexusHost))
|
|
except ValueError:
|
|
raise Exception(f" IP {ip} for Cisco devices is invalid.\nExiting...")
|
|
|
|
try:
|
|
for idrac9Host in IDRAC9_HOST_LIST:
|
|
try:
|
|
ip = str(ipaddress.IPv4Address(idrac9Host))
|
|
except ValueError:
|
|
raise Exception(f" IP {ip} for IDRAC9 devices is invalid.\nExiting...")
|
|
except:
|
|
IDRAC9_HOST_LIST = []
|
|
|
|
# Done under "Program ENV" on line ~18
|
|
|
|
|
|
# Code-------------------------------------------
|
|
|
|
async def main():
|
|
|
|
mainQueue = asyncio.Queue(maxsize=225)
|
|
|
|
asyncio.create_task(ALLdbIDRACWriter(mainQueue))
|
|
|
|
print("Starting tasks")
|
|
|
|
while True:
|
|
|
|
# IDRAC part
|
|
async with asyncio.TaskGroup() as tg:
|
|
for Idrac78IP in IDRAC78_HOST_LIST:
|
|
tg.create_task(idrac78.idrac7_8PoolRemote_v3(
|
|
Idrac78IP, mainQueue,
|
|
SNMPUSER, SNMPAUTHKEY, SNMPPRIVKEY, ROUND_PREC, SNMPORT))
|
|
|
|
for Idrac9IP in IDRAC9_HOST_LIST:
|
|
tg.create_task(idrac9.idrac9PoolRemote_v3(
|
|
Idrac9IP, mainQueue,
|
|
SNMPUSER, SNMPAUTHKEY, SNMPPRIVKEY, ROUND_PREC, SNMPORT))
|
|
# FANS
|
|
# tg.create_task(idracPoolRemoteFAN_v3(IdracIP, mainQueue))
|
|
|
|
# CISCO devices
|
|
for cisco3750Xdev in CISCO_3750_HOST_LIST:
|
|
tg.create_task(lob.lobGetRemote(
|
|
cisco3750Xdev, mainQueue,
|
|
SNMPUSER, SNMPAUTHKEY, SNMPPRIVKEY, ROUND_PREC, SNMPORT))
|
|
|
|
# for ciscoNexusDev in CISCO_NEXUS_HOST_LIST:
|
|
# tg.create_task(nyater(
|
|
# ciscoNexusDev, mainQueue,
|
|
# SNMPUSER, SNMPAUTHKEY, SNMPPRIVKEY, ROUND_PREC, SNMPORT))
|
|
|
|
await asyncio.sleep(GET_INTERVAL)
|
|
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# qu = asyncio.Queue()
|
|
# asyncio.run(idracPoolRemote_v3("192.168.20.7", qu))
|
|
print("Starting")
|
|
asyncio.run(main())
|
|
|
|
# loop = asyncio.get_event_loop()
|
|
# task = loop.create_task(main())
|
|
|