""" Cloud Resource Allocation RL Environment. This module implements the core environment class for simulating cloud infrastructure resource allocation. The environment maintains system state (CPU, memory, request rate, allocated resources) and processes actions to dynamically adjust resource allocation. """ import numpy as np import torch from typing import Tuple, Dict, List, Optional from env.config import EnvConfig from env.reward import RewardCalculator class CloudResourceEnv: """ Simulates a cloud infrastructure environment for RL-based resource allocation. The environment models a cloud system where an agent must dynamically allocate server instances based on observed system metrics. The state includes CPU utilization, memory utilization, incoming request rate, and currently allocated resources. State space: [cpu_util, memory_util, request_rate, allocated_resources] Action space: {0: decrease, 1: maintain, 2: increase} The environment incorporates stochastic dynamics to simulate real-world variability in request patterns and system behavior. """ # Action constants ACTION_DECREASE = 0 ACTION_MAINTAIN = 1 ACTION_INCREASE = 2 def __init__( self, config: Optional[EnvConfig] = None, reward_calculator: Optional[RewardCalculator] = None, task_name: str = "medium" ): """ Initialize the cloud resource allocation environment. Args: config: Environment configuration parameters. If None, uses default EnvConfig. reward_calculator: Reward calculator instance. If None, creates default RewardCalculator. task_name: Task difficulty level ("easy", "medium", or "hard"). Defaults to "medium". Raises: ValueError: If task_name is not in TASKS dictionary. """ # Import TASKS here to avoid circular imports from env.config import TASKS # Validate task_name if task_name not in TASKS: raise ValueError( f"Invalid task_name '{task_name}'. Must be one of: {list(TASKS.keys())}" ) # Use default config if not provided self.config = config if config is not None else EnvConfig() # Override config with task-specific parameters task_config = TASKS[task_name] self.config.base_request_rate = task_config["base_request_rate"] self.config.request_rate_std = task_config["request_rate_std"] self.task_name = task_name # Initialize reward calculator self.reward_calculator = reward_calculator if reward_calculator is not None else RewardCalculator() # Initialize state variables self.cpu_util: float = 0.0 self.memory_util: float = 0.0 self.request_rate: int = 0 self.allocated_resources: int = self.config.initial_resources self.latency: float = 0.0 # Latency in ms # Initialize episode tracking self.current_step: int = 0 self.episode_states: List[np.ndarray] = [] self.episode_actions: List[int] = [] self.episode_rewards: List[float] = [] self.cumulative_reward: float = 0.0 # Episode termination flag self.done: bool = False # Random number generator for reproducibility self.rng = np.random.RandomState() def reset(self) -> np.ndarray: """ Reset environment to initial state. This method reinitializes the environment for a new episode. It resets all state variables to their initial values, clears episode history, and introduces stochastic variation in the initial request rate to ensure diverse starting conditions across episodes. Returns: Initial observation as numpy array [cpu_util, memory_util, request_rate, allocated_resources] """ # Reset episode tracking self.current_step = 0 self.episode_states = [] self.episode_actions = [] self.episode_rewards = [] self.cumulative_reward = 0.0 self.done = False # Reset allocated resources to initial value self.allocated_resources = self.config.initial_resources # Initialize request rate with stochastic variation # This ensures different starting conditions across episodes self.request_rate = int( self.config.base_request_rate + self.rng.normal(0, self.config.request_rate_std) ) # Ensure request rate is non-negative self.request_rate = max(0, self.request_rate) # Calculate initial CPU and memory utilization # Utilization depends on request rate and allocated resources self.cpu_util = (self.request_rate * self.config.cpu_per_request) / ( self.allocated_resources * self.config.resource_capacity ) * 100.0 self.memory_util = (self.request_rate * self.config.memory_per_request) / ( self.allocated_resources * self.config.resource_capacity ) * 100.0 # Clamp utilization values to valid range [0, 100] self.cpu_util = max(0.0, min(100.0, self.cpu_util)) self.memory_util = max(0.0, min(100.0, self.memory_util)) # Calculate latency: higher request rate and fewer resources increase latency self.latency = self.request_rate / max(self.allocated_resources, 1) # Return initial observation return self._get_observation() def _get_observation(self) -> np.ndarray: """ Get current state as observation array. Returns: Numpy array [cpu_util, memory_util, request_rate, allocated_resources] """ return np.array([ self.cpu_util, self.memory_util, float(self.request_rate), float(self.allocated_resources) ], dtype=np.float32) def get_state_tensor(self) -> torch.Tensor: """ Get current state as PyTorch tensor. This method provides a PyTorch tensor representation of the current state, enabling seamless integration with PyTorch-based RL algorithms and neural networks. The tensor maintains the same structure as the NumPy observation array. Returns: PyTorch tensor [cpu_util, memory_util, request_rate, allocated_resources] """ observation = self._get_observation() return torch.from_numpy(observation) def state(self): """ Get current state as EnvState instance for OpenEnv compliance. This method provides typed, validated access to the environment state using the Pydantic EnvState model. It can be called after reset() or step() to retrieve the current state in a structured format. Returns: EnvState: Pydantic model with cpu, memory, request_rate, and resources fields """ from env.config import EnvState return EnvState( cpu=self.cpu_util, memory=self.memory_util, request_rate=float(self.request_rate), resources=self.allocated_resources ) def _update_state(self, action: int) -> None: """ Update system state based on action and stochastic dynamics. This method applies the selected action to modify allocated resources, updates the request rate with stochastic fluctuation, and recalculates CPU and memory utilization based on the new system configuration. Args: action: Action to execute (0: decrease, 1: maintain, 2: increase) """ # Apply action to allocated resources if action == self.ACTION_INCREASE: self.allocated_resources += 1 elif action == self.ACTION_DECREASE: # Maintain minimum of 1 resource self.allocated_resources = max(1, self.allocated_resources - 1) elif action == self.ACTION_MAINTAIN: # Keep resources unchanged pass else: raise ValueError(f"Invalid action: {action}. Must be 0, 1, or 2.") # Update request rate with stochastic fluctuation # This simulates real-world variability in incoming traffic fluctuation = self.rng.normal(0, self.config.request_rate_std) self.request_rate = int(self.config.base_request_rate + fluctuation) # Ensure request rate is non-negative self.request_rate = max(0, self.request_rate) # Calculate CPU utilization based on request rate and allocated resources # Higher request rate increases utilization # More allocated resources decrease utilization self.cpu_util = (self.request_rate * self.config.cpu_per_request) / ( self.allocated_resources * self.config.resource_capacity ) * 100.0 # Calculate memory utilization based on request rate and allocated resources self.memory_util = (self.request_rate * self.config.memory_per_request) / ( self.allocated_resources * self.config.resource_capacity ) * 100.0 # Clamp all values to valid ranges self.cpu_util = max(0.0, min(100.0, self.cpu_util)) self.memory_util = max(0.0, min(100.0, self.memory_util)) # Calculate latency: higher request rate and fewer resources increase latency self.latency = self.request_rate / max(self.allocated_resources, 1) def _check_termination(self) -> bool: """ Check if episode should terminate based on state or step count. Termination conditions: 1. Step count exceeds max_steps 2. CPU utilization exceeds termination threshold (default 95%) 3. Memory utilization exceeds termination threshold (default 95%) Returns: True if episode should terminate, False otherwise """ # Check if max steps exceeded if self.current_step >= self.config.max_steps: return True # Check if CPU utilization exceeds threshold if self.cpu_util > self.config.termination_threshold: return True # Check if memory utilization exceeds threshold if self.memory_util > self.config.termination_threshold: return True return False def step(self, action: int) -> Tuple[np.ndarray, float, bool, Dict]: """ Execute one time step with the given action. This method processes the action, updates the environment state, calculates the reward, checks for termination, and stores the transition in episode history. It returns the standard RL tuple (observation, reward, done, info). Args: action: Action to execute (0: decrease, 1: maintain, 2: increase) Returns: observation: Current state after action as numpy array reward: Reward signal from reward calculator done: Whether episode has terminated info: Additional diagnostic information dictionary Raises: ValueError: If action is not in valid range {0, 1, 2} RuntimeError: If step is called after episode termination """ # Check if episode already terminated if self.done: raise RuntimeError( "Cannot call step() after episode termination. Call reset() to start a new episode." ) # Validate action is in valid range if action not in {self.ACTION_DECREASE, self.ACTION_MAINTAIN, self.ACTION_INCREASE}: raise ValueError( f"Invalid action: {action}. Must be 0 (decrease), 1 (maintain), or 2 (increase)." ) # Update state based on action self._update_state(action) # Increment step counter self.current_step += 1 # Calculate reward reward = self.reward_calculator.calculate_reward( self.cpu_util, self.memory_util, self.allocated_resources ) # Check termination status self.done = self._check_termination() # Get current observation observation = self._get_observation() # Store transition in episode history self.episode_states.append(observation.copy()) self.episode_actions.append(action) self.episode_rewards.append(reward) self.cumulative_reward += reward # Prepare info dictionary info = { 'step': self.current_step, 'cumulative_reward': self.cumulative_reward, 'cpu_util': self.cpu_util, 'memory_util': self.memory_util, 'request_rate': self.request_rate, 'allocated_resources': self.allocated_resources, 'latency': self.latency } return observation, reward, self.done, info