from server.Files import read_file from server.game_board_stats import GameBoardStoreBuilder from server.GameBoard import GameBoard from snakes import SnakeBuilder from quart_common.web.logger import await_log from quart_common.web.logger import build_logger from server.metrics.MetricsManager import MetricsManager from server.storage.StorageLoader import StorageLoader from quart import Quart, request, jsonify import logging, json, os, re, time from typing import cast class Server: default_snake_config = { 'apiversion': '1', 'author': '', 'color': '#888888', 'head': 'default', 'tail': 'default', 'version': '1.0.0', } def __init__(self, data_path:str, snake_type:str, storage_type:str, debug:bool=False, check_tls_security:bool=False, game_state_backend:str='memory', game_state_redis_url:str='redis://localhost:6379/0', game_state_ttl_sec:int=900, game_state_local_cache:bool=True, metrics_backend:str='memory', metrics_redis_url:str='redis://localhost:6379/0', metrics_ttl_sec:int=None): self.debug = debug self.snake_type = snake_type self.storage_type = storage_type self.config_file = os.path.join(data_path, 'data', 'snake-config.json') self.data_path = data_path self.check_tls_security = check_tls_security self.store_game_state = False normalized_backend = (game_state_backend or 'memory').strip().lower() self.game_state_local_cache = (game_state_local_cache and normalized_backend != 'memory') self.game_state_store = GameBoardStoreBuilder.build( backend=game_state_backend, redis_url=game_state_redis_url, ttl_seconds=game_state_ttl_sec, ) self.metrics_backend = (metrics_backend or 'memory').strip().lower() self.metrics_manager = MetricsManager( backend=self.metrics_backend, redis_url=metrics_redis_url, ttl_seconds=metrics_ttl_sec, key_prefix=os.environ.get('METRICS_REDIS_KEY_PREFIX', 'snake:metrics:worker'), ) self.running_games:dict[str, GameBoard] = {} self.game_move_counts:dict[str, int] = {} self.game_last_seen_unix:dict[str, int] = {} self.stale_game_timeout_sec = self._get_stale_game_timeout_sec() self.metrics = { 'games_started': 0, 'games_ended': 0, 'wins': 0, 'losses': 0, 'total_moves': 0, 'total_turns': 0, 'max_turn': 0, 'active_games_peak': 0, 'games_autocreated': 0, 'http_requests_total': 0, 'http_requests_by_endpoint': { 'info': 0, 'start': 0, 'move': 0, 'end': 0, }, 'move_direction_counts': { 'up': 0, 'down': 0, 'left': 0, 'right': 0, 'unknown': 0, }, 'move_response_time_ms_total': 0.0, 'move_response_time_ms_max': 0.0, 'last_game_start_unix': 0, 'last_game_end_unix': 0, 'last_move_unix': 0, 'games_stuck_removed': 0, 'game_state_local_cache_enabled': bool(self.game_state_local_cache), 'metrics_backend': self.metrics_backend, } self.logger = build_logger('Battlesnake', debug_env_var='DEBUG_SERVER') self.snake_version = self._get_snake_version() self.app = Quart('Battlesnake') # info is called when you create your Battlesnake on play.battlesnake.com # and controls your Battlesnake's appearance # TIP: If you open your Battlesnake URL in a browser you should see this data @self.app.get('/') async def on_info(): self._record_http_request('info') snake_config = await self._read_json_config_or_create() await await_log(self.logger.info(f'INFO Snake: {snake_config}')) return snake_config # start is called when your Battlesnake begins a game @self.app.post('/start') async def on_start(): self._record_http_request('start') self._prune_stale_games() game_state = await request.get_json() await self._create_game_board(game_state) await await_log(self.logger.info(f'GAME START: {game_state['game']}')) return 'ok' # move is called when your Battlesnake game is running game @self.app.post('/move') async def on_move(): self._record_http_request('move') game_state = await request.get_json() move_started = time.perf_counter() game_board = cast(GameBoard, await self._get_game_board(game_state)) next_move = game_board.snake_neat_make_a_move() await self._persist_game_board(game_state['game']['id'], game_board) elapsed_ms = (time.perf_counter() - move_started) * 1000.0 self.metrics['move_response_time_ms_total'] += elapsed_ms self.metrics['move_response_time_ms_max'] = max( self.metrics['move_response_time_ms_max'], elapsed_ms, ) move_counts = self.metrics['move_direction_counts'] if next_move in move_counts: move_counts[next_move] += 1 else: move_counts['unknown'] += 1 self.metrics['last_move_unix'] = int(time.time()) if self.debug: await await_log(self.logger.debug(f'TURN: {game_state['turn']:3}, MOVE: {next_move:5}')) return {'move': next_move} # end is called when your Battlesnake finishes a game @self.app.post('/end') async def on_end(): self._record_http_request('end') self._prune_stale_games() game_state = await request.get_json() if self.store_game_state: game_board = cast(GameBoard, await self._get_game_board(game_state, end=True)) if self.check_tls_security: await game_board.save( StorageLoader.build(self.storage_type), file_path=os.path.join(self.data_path, 'data'), database=os.getenv('EDGEDB_DATABASE', None), tls_security=None, ) else: await game_board.save( StorageLoader.build(self.storage_type), file_path=os.path.join(self.data_path, 'data'), database=os.getenv('EDGEDB_DATABASE', None), ) await await_log(self.logger.info(f'GAME ENDED: Winner is {[x['name'] for x in game_state['board']['snakes']]}')) await self._delete_game_board(game_state) return 'ok' @self.app.after_request async def identify_server(response): response.headers.set('server', 'battlesnake/gitea/snake-python') return response @self.app.after_serving async def shutdown_state_storage(): await self.game_state_store.close() await self._close_metrics_store() @self.app.get('/cleanup') async def cleanup(): results = self._cleanup_database() return jsonify(data=json.loads(results), status=200) @self.app.get('/metrics') async def metrics(): return jsonify(await self._build_metrics()) @self.app.get('/metrics/prometheus') async def metrics_prometheus(): snapshot = await self._build_metrics() return ( self._build_prometheus_metrics(snapshot), 200, {'Content-Type': 'text/plain; version=0.0.4; charset=utf-8'}, ) async def run(self, host:str='0.0.0.0', port:int=8000, debug:bool=False): logging.getLogger('werkzeug').setLevel(logging.ERROR) await await_log(self.logger.info(f'Running Battlesnake at http://{host}:{port} with the {" ".join(re.findall("[A-Z][^A-Z]*", self.snake_type))}')) await self.app.run_task(host=host, port=port, debug=debug) async def _read_json_config_or_create(self) -> dict[str, str]: snake_config = cast(dict[str, str]|None, await read_file(self.config_file, json.load)) if not snake_config: return await self._override_snake_config_with_environment_variables(self.default_snake_config) return await self._override_snake_config_with_environment_variables(snake_config) async def _override_snake_config_with_environment_variables(self, config:dict[str, str]) -> dict[str, str]: config['version'] = self.snake_version for key in ('author', 'color', 'head', 'tail'): value = os.environ.get(f'SNAKE_{key.upper()}') if value is not None: config[key] = value version_override = os.environ.get('SNAKE_VERSION') if version_override is not None: config['version'] = version_override return config def _get_snake_version(self) -> str: configured_version = SnakeBuilder.get_version(self.snake_type) if configured_version: return configured_version try: snake = SnakeBuilder.build(self.snake_type) except Exception: return self.default_snake_config['version'] version = getattr(snake, 'version', None) if version is None: version = getattr(snake, 'VERSION', None) if not version: return self.default_snake_config['version'] return str(version) def _get_stale_game_timeout_sec(self) -> int: value = os.getenv('SNAKE_STUCK_GAME_TIMEOUT_SEC', '180') try: return max(30, int(value)) except ValueError: return 180 async def _create_game_board(self, game_state:dict) -> GameBoard: game_id = game_state['game']['id'] new_game_board = GameBoard( game_id=game_id, width=game_state['board']['width'], height=game_state['board']['height'], ruleset=game_state['game']['ruleset'], source=game_state['game']['source'], map=game_state['game']['map'], snake_class=SnakeBuilder.build(self.snake_type), ) await new_game_board.start_game(game_state) if self.game_state_local_cache: self.running_games[game_id] = new_game_board await self.game_state_store.save(game_id, new_game_board) self.game_move_counts[game_id] = 0 self.game_last_seen_unix[game_id] = int(time.time()) self.metrics['games_started'] += 1 self.metrics['active_games_peak'] = max( self.metrics['active_games_peak'], len(self.game_last_seen_unix), ) self.metrics['last_game_start_unix'] = int(time.time()) return new_game_board async def _persist_game_board(self, game_id:str, game_board:GameBoard): if self.game_state_local_cache: self.running_games[game_id] = game_board await self.game_state_store.save(game_id, game_board) async def _delete_game_board(self, game_state:dict): game_id = game_state['game']['id'] self.running_games.pop(game_id, None) self.game_move_counts.pop(game_id, None) self.game_last_seen_unix.pop(game_id, None) await self.game_state_store.delete(game_id) async def _get_game_board(self, game_state:dict, end:bool=False) -> GameBoard: game_id = game_state['game']['id'] game_board: GameBoard if self.game_state_local_cache and game_id in self.running_games: game_board = self.running_games[game_id] else: persisted_board = await self.game_state_store.load(game_id) if persisted_board is not None: game_board = cast(GameBoard, persisted_board) if self.game_state_local_cache: self.running_games[game_id] = game_board else: game_board = await self._create_game_board(game_state) self.metrics['games_autocreated'] += 1 if not end: self.metrics['total_moves'] += 1 self.game_move_counts[game_id] = self.game_move_counts.get(game_id, 0) + 1 self.game_last_seen_unix[game_id] = int(time.time()) game_board.read_game_data(game_state) if end: self._record_game_end(game_state) game_board.end_game(game_state) await self._persist_game_board(game_id, game_board) return game_board def enable_store_game_state(self): self.store_game_state = True def _cleanup_database(self): storage = StorageLoader.build(self.storage_type)() return storage.cleanup() def _prune_stale_games(self): if not self.game_last_seen_unix: return now = int(time.time()) stale_ids = [ game_id for game_id, last_seen in self.game_last_seen_unix.items() if now - last_seen >= self.stale_game_timeout_sec ] for game_id in stale_ids: self.running_games.pop(game_id, None) self.game_move_counts.pop(game_id, None) self.game_last_seen_unix.pop(game_id, None) self.metrics['games_stuck_removed'] += 1 def _record_game_end(self, game_state:dict): self.metrics['games_ended'] += 1 self.metrics['last_game_end_unix'] = int(time.time()) final_turn = int(game_state.get('turn', 0)) self.metrics['total_turns'] += final_turn self.metrics['max_turn'] = max(self.metrics['max_turn'], final_turn) you_id = game_state.get('you', {}).get('id') alive_snakes = game_state.get('board', {}).get('snakes', []) alive_ids = {snake.get('id') for snake in alive_snakes} if you_id and you_id in alive_ids: self.metrics['wins'] += 1 else: self.metrics['losses'] += 1 def _build_local_metrics(self) -> dict: games_ended = self.metrics['games_ended'] total_moves = self.metrics['total_moves'] avg_turns = self.metrics['total_turns'] / games_ended if games_ended else 0.0 win_rate = self.metrics['wins'] / games_ended if games_ended else 0.0 avg_move_ms = (self.metrics['move_response_time_ms_total'] / total_moves if total_moves else 0.0) now = int(time.time()) oldest_active_age = 0 if self.game_last_seen_unix: oldest_active_age = max(0, now - min(self.game_last_seen_unix.values())) stale_candidates = sum( 1 for last_seen in self.game_last_seen_unix.values() if now - last_seen >= self.stale_game_timeout_sec ) return { **self.metrics, 'active_games': len(self.game_last_seen_unix), 'tracked_games': len(self.game_move_counts), 'avg_turns_per_game': round(avg_turns, 2), 'win_rate': round(win_rate, 4), 'avg_move_response_ms': round(avg_move_ms, 2), 'http_requests_by_endpoint': dict(self.metrics['http_requests_by_endpoint']), 'move_direction_counts': dict(self.metrics['move_direction_counts']), 'oldest_active_game_age_sec': oldest_active_age, 'stale_game_timeout_sec': self.stale_game_timeout_sec, 'active_games_stale': stale_candidates, } def _record_http_request(self, endpoint: str): self.metrics['http_requests_total'] += 1 endpoint_counts = self.metrics['http_requests_by_endpoint'] endpoint_counts[endpoint] = endpoint_counts.get(endpoint, 0) + 1 async def _build_metrics(self) -> dict: local_snapshot = self._build_local_metrics() return await self.metrics_manager.snapshot(local_snapshot) def _build_prometheus_metrics(self, snapshot: dict) -> str: lines = [ '# HELP snake_games_started_total Total games started by snake server.', '# TYPE snake_games_started_total counter', f'snake_games_started_total {snapshot['games_started']}', '# HELP snake_games_ended_total Total games ended by snake server.', '# TYPE snake_games_ended_total counter', f'snake_games_ended_total {snapshot['games_ended']}', '# HELP snake_wins_total Total games won by this snake.', '# TYPE snake_wins_total counter', f'snake_wins_total {snapshot['wins']}', '# HELP snake_losses_total Total games lost by this snake.', '# TYPE snake_losses_total counter', f'snake_losses_total {snapshot['losses']}', '# HELP snake_moves_total Total move decisions served by /move.', '# TYPE snake_moves_total counter', f'snake_moves_total {snapshot['total_moves']}', '# HELP snake_turns_total Total turns across all ended games.', '# TYPE snake_turns_total counter', f'snake_turns_total {snapshot['total_turns']}', '# HELP snake_active_games Currently active games in memory.', '# TYPE snake_active_games gauge', f'snake_active_games {snapshot['active_games']}', '# HELP snake_tracked_games Currently tracked game IDs for move counters.', '# TYPE snake_tracked_games gauge', f'snake_tracked_games {snapshot['tracked_games']}', '# HELP snake_max_turn Highest final turn seen in an ended game.', '# TYPE snake_max_turn gauge', f'snake_max_turn {snapshot['max_turn']}', '# HELP snake_active_games_peak Highest active game count observed.', '# TYPE snake_active_games_peak gauge', f'snake_active_games_peak {snapshot['active_games_peak']}', '# HELP snake_games_autocreated_total Games created on /move or /end due to missing /start.', '# TYPE snake_games_autocreated_total counter', f'snake_games_autocreated_total {snapshot['games_autocreated']}', '# HELP snake_http_requests_total Total HTTP requests handled by this process.', '# TYPE snake_http_requests_total counter', f'snake_http_requests_total {snapshot['http_requests_total']}', '# HELP snake_move_response_ms_total Total move endpoint compute time in milliseconds.', '# TYPE snake_move_response_ms_total counter', f'snake_move_response_ms_total {round(snapshot['move_response_time_ms_total'], 3)}', '# HELP snake_move_response_ms_max Maximum move endpoint compute time in milliseconds.', '# TYPE snake_move_response_ms_max gauge', f'snake_move_response_ms_max {round(snapshot['move_response_time_ms_max'], 3)}', '# HELP snake_avg_turns_per_game Average final turn per ended game.', '# TYPE snake_avg_turns_per_game gauge', f'snake_avg_turns_per_game {snapshot['avg_turns_per_game']}', '# HELP snake_avg_move_response_ms Average move endpoint compute time in milliseconds.', '# TYPE snake_avg_move_response_ms gauge', f'snake_avg_move_response_ms {snapshot['avg_move_response_ms']}', '# HELP snake_win_rate Win ratio from ended games (0.0 - 1.0).', '# TYPE snake_win_rate gauge', f'snake_win_rate {snapshot['win_rate']}', '# HELP snake_last_game_start_unix Unix timestamp of most recent /start request.', '# TYPE snake_last_game_start_unix gauge', f'snake_last_game_start_unix {snapshot['last_game_start_unix']}', '# HELP snake_last_game_end_unix Unix timestamp of most recent /end request.', '# TYPE snake_last_game_end_unix gauge', f'snake_last_game_end_unix {snapshot['last_game_end_unix']}', '# HELP snake_last_move_unix Unix timestamp of most recent /move response.', '# TYPE snake_last_move_unix gauge', f'snake_last_move_unix {snapshot['last_move_unix']}', '# HELP snake_games_stuck_removed_total Active games auto-removed due to inactivity timeout.', '# TYPE snake_games_stuck_removed_total counter', f'snake_games_stuck_removed_total {snapshot['games_stuck_removed']}', '# HELP snake_oldest_active_game_age_sec Age in seconds of the oldest active game.', '# TYPE snake_oldest_active_game_age_sec gauge', f'snake_oldest_active_game_age_sec {snapshot['oldest_active_game_age_sec']}', '# HELP snake_stale_game_timeout_sec Configured inactivity timeout for stale games.', '# TYPE snake_stale_game_timeout_sec gauge', f'snake_stale_game_timeout_sec {snapshot['stale_game_timeout_sec']}', '# HELP snake_active_games_stale Active games currently beyond stale timeout.', '# TYPE snake_active_games_stale gauge', f'snake_active_games_stale {snapshot['active_games_stale']}', ] lines.extend([ '# HELP snake_http_requests_by_endpoint_total Requests served grouped by endpoint.', '# TYPE snake_http_requests_by_endpoint_total counter', ]) for endpoint, count in snapshot['http_requests_by_endpoint'].items(): lines.append(f'snake_http_requests_by_endpoint_total{{endpoint="{endpoint}"}} {count}') lines.extend([ '# HELP snake_moves_by_direction_total Move responses grouped by direction.', '# TYPE snake_moves_by_direction_total counter', ]) for direction, count in snapshot['move_direction_counts'].items(): lines.append(f'snake_moves_by_direction_total{{direction="{direction}"}} {count}') return '\n'.join(lines) + '\n' async def _close_metrics_store(self) -> None: await self.metrics_manager.close()