sync
This commit is contained in:
350
checker.py
Normal file
350
checker.py
Normal file
@@ -0,0 +1,350 @@
|
||||
import requests
|
||||
import time
|
||||
import logging
|
||||
import sys
|
||||
from typing import Dict, Any, Optional
|
||||
from pprint import pprint
|
||||
|
||||
# Configure logging
|
||||
logging.basicConfig(
|
||||
level=logging.DEBUG,
|
||||
format='%(asctime)s - %(levelname)s: %(message)s',
|
||||
datefmt='%Y-%m-%d %H:%M:%S',
|
||||
handlers=[
|
||||
logging.StreamHandler(sys.stdout),
|
||||
logging.FileHandler('connection_monitor.log')
|
||||
]
|
||||
)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
class ConnectionMonitor:
|
||||
def __init__(self,
|
||||
qbittorrent_url: str = 'http://127.0.0.1:8080',
|
||||
nomad_url: str = 'http://127.0.0.1:4646',
|
||||
tracker_name: str = 'myanon'):
|
||||
"""
|
||||
Initialize connection monitoring with configurable parameters
|
||||
"""
|
||||
self.api_url = f'{qbittorrent_url}/api/v2/transfer/info'
|
||||
self.qbittorrent_base_url = qbittorrent_url
|
||||
self.nomad_url = nomad_url
|
||||
self.tracker_name = tracker_name
|
||||
|
||||
# Tracking variables
|
||||
self.consecutive_failures = 0
|
||||
self.max_consecutive_failures = 2 # 10 minutes (30s * 20)
|
||||
self.stability_wait_time = 1800 # 30 minutes
|
||||
self.check_interval = 30 # seconds
|
||||
|
||||
# Authentication (update with your credentials)
|
||||
self.qbittorrent_username = 'admin'
|
||||
self.qbittorrent_password = 'adminpass'
|
||||
|
||||
def qbittorrent_login(self) -> requests.Session:
|
||||
"""
|
||||
Authenticate with qBittorrent
|
||||
"""
|
||||
try:
|
||||
session = requests.Session()
|
||||
login_url = f'{self.qbittorrent_base_url}/api/v2/auth/login'
|
||||
login_data = {
|
||||
'username': self.qbittorrent_username,
|
||||
'password': self.qbittorrent_password
|
||||
}
|
||||
response = session.post(login_url, data=login_data)
|
||||
response.raise_for_status()
|
||||
logger.info("Successfully logged into qBittorrent")
|
||||
return session
|
||||
except requests.RequestException as e:
|
||||
logger.error(f"qBittorrent login failed: {e}")
|
||||
return None
|
||||
|
||||
def get_connection_status(self) -> Dict[str, Any]:
|
||||
"""
|
||||
Retrieve connection status from qBittorrent API
|
||||
"""
|
||||
try:
|
||||
response = requests.get(self.api_url, timeout=10)
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
except requests.RequestException as e:
|
||||
logger.error(f"API request failed: {e}")
|
||||
return {}
|
||||
|
||||
def stop_tracker_torrents(self, session: requests.Session):
|
||||
"""
|
||||
Stop torrents matching specific tracker
|
||||
"""
|
||||
try:
|
||||
# Get list of torrents
|
||||
torrents_url = f'{self.qbittorrent_base_url}/api/v2/torrents/info'
|
||||
torrents = session.get(torrents_url).json()
|
||||
|
||||
# Find and stop torrents with matching tracker
|
||||
tracker_torrents = [
|
||||
torrent['hash'] for torrent in torrents
|
||||
if self.tracker_name.lower() in str(torrent).lower()
|
||||
]
|
||||
|
||||
if tracker_torrents:
|
||||
hashes_str = '|'.join(tracker_torrents)
|
||||
pprint(hashes_str)
|
||||
pause_url = f'{self.qbittorrent_base_url}/api/v2/torrents/stop'
|
||||
response = session.post(pause_url, data={'hashes': hashes_str})
|
||||
response.raise_for_status()
|
||||
logger.info(f"Stopped {len(tracker_torrents)} torrents for tracker {self.tracker_name}")
|
||||
else:
|
||||
logger.info(f"No torrents found for tracker {self.tracker_name}")
|
||||
|
||||
except requests.RequestException as e:
|
||||
logger.error(f"Failed to stop torrents: {e}")
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Unexpected error stopping torrents: {e}")
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
|
||||
def restart_nomad_task_via_allocation(self, job_id: str, task_name: str, namespace: str = "default", wait_time: int = 60) -> bool:
|
||||
"""
|
||||
Restart a specific task in a Nomad job by restarting just that task.
|
||||
|
||||
Args:
|
||||
job_id: The ID of the job containing the task
|
||||
task_name: The name of the task to restart
|
||||
namespace: The namespace of the job (default: 'default')
|
||||
wait_time: Seconds to wait after restart (default: 60)
|
||||
|
||||
Returns:
|
||||
bool: True if restart succeeded, False otherwise
|
||||
"""
|
||||
headers = {}
|
||||
if hasattr(self, 'token') and self.token:
|
||||
headers['X-Nomad-Token'] = self.token
|
||||
|
||||
try:
|
||||
# Get allocations for the job
|
||||
allocs_url = f"{self.nomad_url}/v1/job/{job_id}/allocations"
|
||||
params = {'namespace': namespace}
|
||||
|
||||
logger.info(f"Fetching allocations for job '{job_id}'...")
|
||||
response = requests.get(allocs_url, headers=headers, params=params, timeout=10)
|
||||
response.raise_for_status()
|
||||
allocations = response.json()
|
||||
|
||||
# Find allocation containing the task
|
||||
target_alloc = None
|
||||
for alloc in allocations:
|
||||
if alloc['ClientStatus'] == 'running':
|
||||
task_states = alloc.get('TaskStates', {})
|
||||
if task_name in task_states:
|
||||
target_alloc = alloc
|
||||
break
|
||||
|
||||
if not target_alloc:
|
||||
logger.error(f"No running allocation found for task '{task_name}' in job '{job_id}'")
|
||||
return False
|
||||
|
||||
# Restart just the specific task
|
||||
alloc_id = target_alloc['ID']
|
||||
restart_url = f"{self.nomad_url}/v1/client/allocation/{alloc_id}/restart"
|
||||
payload = {"TaskName": task_name}
|
||||
|
||||
logger.info(f"Restarting task '{task_name}' in job '{job_id}'...")
|
||||
response = requests.post(restart_url, headers=headers, params=params, json=payload, timeout=10)
|
||||
|
||||
if response.status_code in [200, 204]:
|
||||
logger.info(f"Successfully restarted task '{task_name}' in job '{job_id}'")
|
||||
time.sleep(wait_time)
|
||||
return True
|
||||
else:
|
||||
logger.error(f"Failed: {response.status_code} - {response.text}")
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Request failed: {e}")
|
||||
return False
|
||||
|
||||
|
||||
|
||||
def restart_tracker_torrents(self, session: requests.Session):
|
||||
"""
|
||||
Restart torrents for specific tracker after stability period
|
||||
"""
|
||||
try:
|
||||
# Get list of previously stopped torrents
|
||||
torrents_url = f'{self.qbittorrent_base_url}/api/v2/torrents/info'
|
||||
torrents = session.get(torrents_url).json()
|
||||
|
||||
# Find and resume torrents with matching tracker
|
||||
tracker_torrents = [
|
||||
torrent['hash'] for torrent in torrents
|
||||
if (self.tracker_name.lower() in str(torrent).lower()
|
||||
and torrent.get('state') == 'paused')
|
||||
]
|
||||
|
||||
if tracker_torrents:
|
||||
# resume_url = f'{self.qbittorrent_base_url}/api/v2/torrents/start'
|
||||
# response = session.post(resume_url, data={'hashes': ','.join(tracker_torrents)})
|
||||
# response.raise_for_status()
|
||||
logger.info(f"Restarted {len(tracker_torrents)} torrents for tracker {self.tracker_name}")
|
||||
else:
|
||||
logger.info(f"No paused torrents found for tracker {self.tracker_name}")
|
||||
|
||||
except requests.RequestException as e:
|
||||
logger.error(f"Failed to restart torrents: {e}")
|
||||
|
||||
def remediate_connection(self):
|
||||
"""
|
||||
Execute full remediation process
|
||||
"""
|
||||
logger.warning("Connection instability detected. Starting remediation...")
|
||||
|
||||
# Login to qBittorrent
|
||||
qbt_session = self.qbittorrent_login()
|
||||
if not qbt_session:
|
||||
logger.error("Could not log in to qBittorrent. Aborting remediation.")
|
||||
return False
|
||||
|
||||
# Stop torrents for specific tracker
|
||||
self.stop_tracker_torrents(qbt_session)
|
||||
sleep(30)
|
||||
self.stop_tracker_torrents(qbt_session)
|
||||
sleep(120) #wait 2 mins to make sure everything stopped
|
||||
# Restart Nomad task
|
||||
if not self.restart_nomad_task_via_allocation(
|
||||
job_id="qbittorrent",
|
||||
task_name="qbittorrent"
|
||||
):
|
||||
logger.error("Nomad task restart failed")
|
||||
return False
|
||||
|
||||
# Wait for connection stability
|
||||
logger.info(f"Waiting {self.stability_wait_time/60} minutes for connection stabilization...")
|
||||
time.sleep(self.stability_wait_time)
|
||||
|
||||
# Verify connection
|
||||
for _ in range(6): # 3 minutes of checks
|
||||
status = self.get_connection_status()
|
||||
is_connected = (
|
||||
status.get('connection_status') == 'connected' and
|
||||
status.get('dht_nodes', 0) > 0
|
||||
)
|
||||
|
||||
if is_connected:
|
||||
# Restart torrents
|
||||
self.restart_tracker_torrents(qbt_session)
|
||||
logger.info("Remediation completed successfully")
|
||||
return True
|
||||
|
||||
time.sleep(30)
|
||||
|
||||
logger.error("Could not stabilize connection after remediation")
|
||||
return False
|
||||
|
||||
def monitor_connection(self):
|
||||
"""
|
||||
Main connection monitoring loop
|
||||
"""
|
||||
logger.info("Starting connection monitoring...")
|
||||
logger.info(f"Monitoring parameters:")
|
||||
logger.info(f"Monitoring parameters:")
|
||||
logger.info(f"- API URL: {self.api_url}")
|
||||
logger.info(f"- Tracker: {self.tracker_name}")
|
||||
logger.info(f"- Check Interval: {self.check_interval} seconds")
|
||||
logger.info(f"- Max Consecutive Failures: {self.max_consecutive_failures}")
|
||||
|
||||
while True:
|
||||
try:
|
||||
# Get current connection status
|
||||
status = self.get_connection_status()
|
||||
|
||||
# Check connection criteria
|
||||
is_connected = (
|
||||
status.get('connection_status') == 'xconnected' and
|
||||
status.get('dht_nodes', 0) > 0
|
||||
)
|
||||
|
||||
if is_connected:
|
||||
# Reset failure counter if connected
|
||||
self.consecutive_failures = 0
|
||||
logger.debug(f"Connected. DHT Nodes: {status.get('dht_nodes', 0)}")
|
||||
else:
|
||||
# Increment failure counter
|
||||
self.consecutive_failures += 1
|
||||
logger.warning(f"Connection unstable. Failures: {self.consecutive_failures}")
|
||||
|
||||
# Check if remediation is needed
|
||||
if self.consecutive_failures >= self.max_consecutive_failures:
|
||||
logger.error("Persistent connection failure. Initiating remediation.")
|
||||
remediation_result = self.remediate_connection()
|
||||
|
||||
# Reset failure counter based on remediation result
|
||||
self.consecutive_failures = 0 if remediation_result else self.max_consecutive_failures
|
||||
|
||||
# Wait before next check
|
||||
time.sleep(self.check_interval)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Unexpected error in monitoring loop: {e}")
|
||||
time.sleep(self.check_interval)
|
||||
|
||||
except KeyboardInterrupt:
|
||||
logger.info("Connection monitoring stopped by user.")
|
||||
break
|
||||
|
||||
def main():
|
||||
"""
|
||||
Main entry point for the connection monitoring script
|
||||
"""
|
||||
# Create monitor instance with optional custom parameters
|
||||
monitor = ConnectionMonitor(
|
||||
qbittorrent_url='http://sp.service.dc1.consul:8080', # Customize as needed
|
||||
nomad_url='http://192.168.4.36:4646', # Customize as needed
|
||||
tracker_name='https://t.myanonamouse.net/tracker.php/xxxx/announce' # Customize tracker name
|
||||
)
|
||||
|
||||
try:
|
||||
# Start connection monitoring
|
||||
monitor.monitor_connection()
|
||||
|
||||
except Exception as e:
|
||||
logger.critical(f"Critical error in main execution: {e}")
|
||||
|
||||
def debug_system_info():
|
||||
"""
|
||||
Log system and environment information for troubleshooting
|
||||
"""
|
||||
import platform
|
||||
import socket
|
||||
|
||||
logger.info("System Diagnostic Information:")
|
||||
logger.info(f"Python Version: {platform.python_version()}")
|
||||
logger.info(f"Operating System: {platform.platform()}")
|
||||
logger.info(f"Hostname: {socket.gethostname()}")
|
||||
|
||||
# Network connectivity check
|
||||
try:
|
||||
import requests
|
||||
response = requests.get('https://www.google.com', timeout=5)
|
||||
logger.info(f"Internet Connectivity: OK (Status Code: {response.status_code})")
|
||||
except Exception as e:
|
||||
logger.warning(f"Internet Connectivity Check Failed: {e}")
|
||||
|
||||
if __name__ == '__main__':
|
||||
# Run system diagnostics before main script
|
||||
debug_system_info()
|
||||
|
||||
# Configure additional logging
|
||||
try:
|
||||
# Attempt to set up more detailed logging
|
||||
import logging
|
||||
logging.getLogger('urllib3').setLevel(logging.WARNING)
|
||||
logging.getLogger('requests').setLevel(logging.WARNING)
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to configure additional logging: {e}")
|
||||
|
||||
# Run the main monitoring script
|
||||
main()
|
||||
Reference in New Issue
Block a user