120 lines
3.7 KiB
Python
120 lines
3.7 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Real-time Monitoring untuk MikroTik dengan Gemini AI
|
|
"""
|
|
|
|
import time
|
|
import json
|
|
from datetime import datetime
|
|
from gemini_mikrotik_integration import MikroTikGeminiAssistant
|
|
|
|
class RealTimeMonitor:
|
|
def __init__(self, interval=60):
|
|
"""Initialize monitor dengan interval detik"""
|
|
self.interval = interval
|
|
self.assistant = MikroTikGeminiAssistant()
|
|
self.metrics_history = []
|
|
|
|
def collect_metrics(self):
|
|
"""Kumpulkan metrics dari MikroTik"""
|
|
summary = self.assistant.get_summary()
|
|
|
|
metrics = {
|
|
"timestamp": datetime.now().isoformat(),
|
|
"cpu": summary["cpu"],
|
|
"memory": summary["memory"],
|
|
"interfaces": summary["interfaces"],
|
|
"ppp_secrets": summary["ppp_secrets"],
|
|
}
|
|
|
|
self.metrics_history.append(metrics)
|
|
# Keep only last 100 readings
|
|
if len(self.metrics_history) > 100:
|
|
self.metrics_history = self.metrics_history[-100:]
|
|
|
|
return metrics
|
|
|
|
def check_alerts(self, metrics):
|
|
"""Check untuk alert conditions"""
|
|
alerts = []
|
|
|
|
# CPU alert
|
|
try:
|
|
cpu = int(metrics["cpu"].replace("%", ""))
|
|
if cpu > 80:
|
|
alerts.append(f"CPU tinggi: {cpu}%")
|
|
except:
|
|
pass
|
|
|
|
return alerts
|
|
|
|
def generate_report(self, hours=1):
|
|
"""Generate report untuk periode tertentu"""
|
|
# Filter metrics untuk periode terakhir
|
|
cutoff = datetime.now().timestamp() - (hours * 3600)
|
|
recent_metrics = [
|
|
m for m in self.metrics_history
|
|
if datetime.fromisoformat(m["timestamp"]).timestamp() > cutoff
|
|
]
|
|
|
|
if not recent_metrics:
|
|
return "Tidak ada data untuk periode ini"
|
|
|
|
# Analisis sederhana
|
|
avg_cpu = sum(int(m["cpu"].replace("%", "")) for m in recent_metrics) / len(recent_metrics)
|
|
|
|
report = f"""📊 **Laporan Monitoring ({hours} jam)**
|
|
|
|
**Statistik:**
|
|
- Periode: {len(recent_metrics)} readings
|
|
- CPU Rata-rata: {avg_cpu:.1f}%
|
|
- Interfaces: {recent_metrics[-1]['interfaces']}
|
|
- PPP Secrets: {recent_metrics[-1]['ppp_secrets']}
|
|
|
|
**Status:** {'⚠️' if avg_cpu > 70 else '✅'}"""
|
|
|
|
return report
|
|
|
|
def start_monitoring(self):
|
|
"""Start real-time monitoring loop"""
|
|
print(f"🔍 Starting real-time monitoring (interval: {self.interval}s)")
|
|
print("Press Ctrl+C to stop\n")
|
|
|
|
try:
|
|
while True:
|
|
metrics = self.collect_metrics()
|
|
alerts = self.check_alerts(metrics)
|
|
|
|
print(f"[{metrics['timestamp'][11:19]}] ", end="")
|
|
print(f"CPU: {metrics['cpu']} ", end="")
|
|
print(f"Intf: {metrics['interfaces']} ", end="")
|
|
|
|
if alerts:
|
|
print(f" ⚠️ Alerts: {', '.join(alerts)}")
|
|
else:
|
|
print(" ✅")
|
|
|
|
time.sleep(self.interval)
|
|
|
|
except KeyboardInterrupt:
|
|
print("\n\n🛑 Monitoring stopped")
|
|
print(f"Total readings: {len(self.metrics_history)}")
|
|
|
|
# Generate final report
|
|
if self.metrics_history:
|
|
report = self.generate_report()
|
|
print("\n" + report)
|
|
|
|
if __name__ == "__main__":
|
|
import sys
|
|
|
|
interval = 60 # default 60 seconds
|
|
if len(sys.argv) > 1:
|
|
try:
|
|
interval = int(sys.argv[1])
|
|
except:
|
|
pass
|
|
|
|
monitor = RealTimeMonitor(interval=interval)
|
|
monitor.start_monitoring()
|