daniel156161/snake-python
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add more metrics data
Build and Push Docker Container / build-and-push (push) Successful in 1m2s
Details

Browse Source
This commit is contained in:
Daniel Dolezal
2026-04-04 09:58:11 +02:00
parent 316870ef7a
commit dbcf9cadaf
2 changed files with 1023 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/Server.py
+106 -1
View File
@@ -8,7 +8,7 @@ from typing import cast
from server.storage.StorageLoader import StorageLoader
from quart import Quart, request, jsonify
import logging, json, os, re
import logging, json, os, re, time
class Server:
default_snake_config = {
@@ -41,6 +41,30 @@ class Server:
'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,
'cleanup': 0,
'metrics': 0,
'metrics_prometheus': 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,
}
self.logger = build_logger('Battlesnake', debug_env_var='DEBUG_SERVER')
self.snake_version = self._get_snake_version()
@@ -52,6 +76,7 @@ class Server:
# 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}'))
@@ -60,6 +85,7 @@ class Server:
# start is called when your Battlesnake begins a game
@self.app.post('/start')
async def on_start():
self._record_http_request('start')
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']}'))
@@ -68,9 +94,24 @@ class Server:
# 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 = await self._get_game_board(game_state)
next_move = game_board.snake_neat_make_a_move()
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}'))
@@ -80,6 +121,7 @@ class Server:
# end is called when your Battlesnake finishes a game
@self.app.post('/end')
async def on_end():
self._record_http_request('end')
game_state = await request.get_json()
if self.store_game_state:
game_board = await self._get_game_board(game_state, end=True)
@@ -182,6 +224,11 @@ class Server:
self.running_games[game_id] = new_game_board
self.game_move_counts[game_id] = 0
self.metrics['games_started'] += 1
self.metrics['active_games_peak'] = max(
self.metrics['active_games_peak'],
len(self.running_games),
)
self.metrics['last_game_start_unix'] = int(time.time())
return new_game_board
def _delete_game_board(self, game_state:dict):
@@ -195,6 +242,7 @@ class Server:
game_board = self.running_games[game_id]
except KeyError:
game_board = await self._create_game_board(game_state)
self.metrics['games_autocreated'] += 1
if not end:
self.metrics['total_moves'] += 1
@@ -216,6 +264,7 @@ class Server:
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
@@ -232,8 +281,10 @@ class Server:
def _build_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
return {
**self.metrics,
@@ -241,8 +292,16 @@ class Server:
'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']),
}
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
def _build_prometheus_metrics(self) -> str:
snapshot = self._build_metrics()
lines = [
@@ -273,11 +332,57 @@ class Server:
'# 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']}',
]
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'