1. import os
  2. import time
  3. import json
  4. import requests
  5. from collections import defaultdict
  6. from urllib.parse import urlparse
  8. class ConfigFileMetrics:
  9. def __init__(self, base_url, rate_limit=10):
  10. self.base_url = base_url
  11. self.rate_limit = rate_limit
  12. self.request_counts = defaultdict(int) # Track requests per config
  13. self.last_request_times = defaultdict(float) # Track last request time
  14. self.lock = threading.Lock() # thread safety
  16. def get_config_metrics(self, config_path):
  17. """
  18. Collects metrics from a configuration file.
  19. """
  20. config_name = os.path.basename(config_path)
  22. with self.lock:
  23. if self.request_counts[config_name] >= self.rate_limit:
  24. wait_time = self.calculate_wait_time(config_name)
  25. print(f"Rate limit exceeded for {config_name}. Waiting {wait_time:.2f} seconds.")
  26. time.sleep(wait_time)
  27. with self.lock:
  28. self.request_counts[config_name] -= 1
  30. current_time = time.time()
  31. self.last_request_times[config_name] = current_time
  32. self.request_counts[config_name] += 1
  34. try:
  35. # Simulate fetching config data. Replace with actual config loading.
  36. config_data = self._fetch_config_data(config_path)
  37. return config_data #Return the config data
  38. except Exception as e:
  39. print(f"Error processing {config_path}: {e}")
  40. return None
  42. def _fetch_config_data(self, config_path):
  43. """Simulates fetching config data from a URL or file."""
  44. # Replace with your actual config fetching logic
  45. try:
  46. with open(config_path, 'r') as f:
  47. config_data = json.load(f)
  49. #Example of more complex data retrieval
  50. #parsed_url = urlparse(config_path)
  51. #response = requests.get(f"{self.base_url}{config_path}")
  52. #response.raise_for_status()
  53. #config_data = response.json()
  55. return config_data
  56. except FileNotFoundError:
  57. print(f"Config file not found: {config_path}")
  58. return None
  59. except json.JSONDecodeError:
  60. print(f"Invalid JSON in config file: {config_path}")
  61. return None
  62. except requests.exceptions.RequestException as e:
  63. print(f"Error fetching config from {config_path}: {e}")
  64. return None
  67. def calculate_wait_time(self, config_name):
  68. """Calculates the wait time based on last request time."""
  69. if not self.last_request_times[config_name]:
  70. return 0.0 # No previous request, no wait
  72. time_since_last_request = time.time() - self.last_request_times[config_name]
  73. return max(0.0, self.rate_limit - time_since_last_request)
  75. def get_metrics(self):
  76. """Returns the collected metrics."""
  77. return {
  78. "request_counts": self.request_counts,
  79. "last_request_times": self.last_request_times
  80. }
  83. import threading
  85. if __name__ == '__main__':
  86. # Example Usage
  87. base_url = "http://example.com/configs" #Replace with your base URL
  88. metrics = ConfigFileMetrics(base_url, rate_limit=5)
  90. # Simulate processing multiple configuration files
  91. config_files = ["config1.json", "config2.json", "config3.json", "config1.json", "config2.json", "config1.json"] #Example files
  92. for file in config_files:
  93. metrics.get_config_metrics(file)
  95. metrics_data = metrics.get_metrics()
  96. print(json.dumps(metrics_data, indent=2))

Add your comment