Implement smart Magic DNS copy with automatic port detection

### Magic DNS Smart Copy Features:
- **PBS servers**: Automatically append `:8007` port when copying
- **PVE servers**: Automatically append `:8006` port when copying
- **Other machines**: Copy Magic DNS address without port (existing behavior)

### UI Improvements:
- PBS servers: Blue button with `:8007` port hint
- PVE servers: Orange button with `:8006` port hint
- Enhanced tooltips with service-specific port information
- Visual distinction between different server types

### PBS Backup Server Monitoring:
- Complete PBS API integration with authentication
- Real-time backup/restore task monitoring with detailed logs
- Namespace-separated backup visualization with color coding
- Datastore usage monitoring and status tracking
- Task history with success/failure status and error details

### Technical Implementation:
- Smart port detection via JavaScript `addSmartPort()` function
- Jinja2 template logic for conditional button styling
- PBS API endpoints for comprehensive backup monitoring
- Enhanced clipboard functionality with user feedback

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
시골약사 2025-09-14 10:48:47 +09:00
parent be3795c7bf
commit 7aa08682b8
4 changed files with 1888 additions and 18 deletions

View File

@ -26,6 +26,66 @@ from utils.proxmox_client import ProxmoxClient
from utils.vnc_proxy import init_vnc_proxy, get_vnc_proxy from utils.vnc_proxy import init_vnc_proxy, get_vnc_proxy
from utils.vnc_websocket_proxy import vnc_proxy from utils.vnc_websocket_proxy import vnc_proxy
import websockets import websockets
import requests
from urllib3.exceptions import InsecureRequestWarning
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
# PBS API 설정
PBS_HOST = "100.64.0.8"
PBS_PORT = "8007"
PBS_USERNAME = "root@pam"
PBS_PASSWORD = "trajet6640"
PBS_BASE_URL = f"https://{PBS_HOST}:{PBS_PORT}/api2/json"
def pbs_get_auth_ticket():
"""PBS 인증 티켓 획득"""
try:
response = requests.post(
f"{PBS_BASE_URL}/access/ticket",
data={
'username': PBS_USERNAME,
'password': PBS_PASSWORD
},
verify=False,
timeout=10
)
if response.status_code == 200:
data = response.json()['data']
return {
'ticket': data['ticket'],
'csrf_token': data['CSRFPreventionToken']
}
return None
except Exception as e:
print(f"PBS 인증 실패: {e}")
return None
def pbs_api_call(endpoint, auth_info=None):
"""PBS API 호출"""
if not auth_info:
auth_info = pbs_get_auth_ticket()
if not auth_info:
return None
try:
headers = {
'Cookie': f"PBSAuthCookie={auth_info['ticket']}",
'CSRFPreventionToken': auth_info['csrf_token']
}
response = requests.get(
f"{PBS_BASE_URL}/{endpoint}",
headers=headers,
verify=False,
timeout=10
)
if response.status_code == 200:
return response.json()['data']
return None
except Exception as e:
print(f"PBS API 호출 실패 ({endpoint}): {e}")
return None
def create_app(config_name=None): def create_app(config_name=None):
"""Flask 애플리케이션 팩토리""" """Flask 애플리케이션 팩토리"""
@ -1586,54 +1646,491 @@ def create_app(config_name=None):
try: try:
import sqlite3 import sqlite3
from datetime import datetime, timedelta from datetime import datetime, timedelta
db_path = '/srv/headscale-setup/farmq-admin/farmq.db' db_path = '/srv/headscale-setup/farmq-admin/farmq.db'
conn = sqlite3.connect(db_path) conn = sqlite3.connect(db_path)
cursor = conn.cursor() cursor = conn.cursor()
# 최근 6개월 구독 트렌드 # 최근 6개월 구독 트렌드
trends = [] trends = []
current_date = datetime.now() current_date = datetime.now()
for i in range(6): for i in range(6):
target_date = current_date - timedelta(days=30*i) target_date = current_date - timedelta(days=30*i)
year = target_date.year year = target_date.year
month = target_date.month month = target_date.month
# 해당 월 신규 구독 수 # 해당 월 신규 구독 수
cursor.execute(''' cursor.execute('''
SELECT COUNT(*) SELECT COUNT(*)
FROM pharmacy_subscriptions FROM pharmacy_subscriptions
WHERE strftime('%Y-%m', start_date) = ? WHERE strftime('%Y-%m', start_date) = ?
''', (f'{year}-{month:02d}',)) ''', (f'{year}-{month:02d}',))
new_subscriptions = cursor.fetchone()[0] new_subscriptions = cursor.fetchone()[0]
# 해당 월 해지 구독 수 # 해당 월 해지 구독 수
cursor.execute(''' cursor.execute('''
SELECT COUNT(*) SELECT COUNT(*)
FROM pharmacy_subscriptions FROM pharmacy_subscriptions
WHERE subscription_status = 'CANCELLED' WHERE subscription_status = 'CANCELLED'
AND strftime('%Y-%m', updated_at) = ? AND strftime('%Y-%m', updated_at) = ?
''', (f'{year}-{month:02d}',)) ''', (f'{year}-{month:02d}',))
cancelled_subscriptions = cursor.fetchone()[0] cancelled_subscriptions = cursor.fetchone()[0]
trends.insert(0, { trends.insert(0, {
'month': f'{year}-{month:02d}', 'month': f'{year}-{month:02d}',
'new_subscriptions': new_subscriptions, 'new_subscriptions': new_subscriptions,
'cancelled_subscriptions': cancelled_subscriptions, 'cancelled_subscriptions': cancelled_subscriptions,
'net_growth': new_subscriptions - cancelled_subscriptions 'net_growth': new_subscriptions - cancelled_subscriptions
}) })
conn.close() conn.close()
return jsonify({ return jsonify({
'success': True, 'success': True,
'data': trends 'data': trends
}) })
except Exception as e: except Exception as e:
print(f"❌ 구독 트렌드 조회 오류: {e}") print(f"❌ 구독 트렌드 조회 오류: {e}")
return jsonify({'success': False, 'error': str(e)}), 500 return jsonify({'success': False, 'error': str(e)}), 500
# =================== PBS 백업 서버 모니터링 ===================
@app.route('/pbs')
def pbs_monitoring():
"""PBS 백업 서버 모니터링 페이지"""
return render_template('pbs/monitoring.html')
@app.route('/api/pbs/status')
def api_pbs_status():
"""PBS 서버 상태 및 기본 정보 조회"""
try:
auth_info = pbs_get_auth_ticket()
if not auth_info:
return jsonify({'success': False, 'error': 'PBS 서버 인증 실패'}), 500
# 서버 버전 정보
version_data = pbs_api_call('version', auth_info)
if not version_data:
return jsonify({'success': False, 'error': 'PBS 버전 정보 조회 실패'}), 500
return jsonify({
'success': True,
'data': {
'status': 'online',
'version': version_data.get('version', 'unknown'),
'release': version_data.get('release', 'unknown'),
'server_time': datetime.now().isoformat()
}
})
except Exception as e:
print(f"❌ PBS 상태 조회 오류: {e}")
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/pbs/datastores')
def api_pbs_datastores():
"""PBS 데이터스토어 정보 조회"""
try:
auth_info = pbs_get_auth_ticket()
if not auth_info:
return jsonify({'success': False, 'error': 'PBS 서버 인증 실패'}), 500
# 데이터스토어 목록
datastores = pbs_api_call('config/datastore', auth_info)
if not datastores:
return jsonify({'success': False, 'error': '데이터스토어 정보 조회 실패'}), 500
# 각 데이터스토어의 사용량 정보 조회
datastore_info = []
for store in datastores:
store_name = store.get('name')
if not store_name:
continue
# 데이터스토어 사용량 조회
status_data = pbs_api_call(f'admin/datastore/{store_name}/status', auth_info)
if status_data:
total_bytes = status_data.get('total', 0)
used_bytes = status_data.get('used', 0)
avail_bytes = status_data.get('avail', 0)
# 바이트를 GB로 변환
total_gb = round(total_bytes / (1024**3), 2)
used_gb = round(used_bytes / (1024**3), 2)
avail_gb = round(avail_bytes / (1024**3), 2)
usage_percent = round((used_bytes / total_bytes * 100), 1) if total_bytes > 0 else 0
datastore_info.append({
'name': store_name,
'comment': store.get('comment', ''),
'path': store.get('path', ''),
'total_gb': total_gb,
'used_gb': used_gb,
'avail_gb': avail_gb,
'usage_percent': usage_percent
})
else:
datastore_info.append({
'name': store_name,
'comment': store.get('comment', ''),
'path': store.get('path', ''),
'total_gb': 0,
'used_gb': 0,
'avail_gb': 0,
'usage_percent': 0,
'error': 'Status unavailable'
})
return jsonify({
'success': True,
'data': datastore_info
})
except Exception as e:
print(f"❌ PBS 데이터스토어 조회 오류: {e}")
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/pbs/tasks')
def api_pbs_tasks():
"""PBS 작업 상태 조회"""
try:
auth_info = pbs_get_auth_ticket()
if not auth_info:
return jsonify({'success': False, 'error': 'PBS 서버 인증 실패'}), 500
# 실행 중인 작업
running_tasks = pbs_api_call('nodes/localhost/tasks?running=true', auth_info)
if running_tasks is None:
running_tasks = []
# 최근 작업 (모든 상태)
all_tasks = pbs_api_call('nodes/localhost/tasks?limit=10', auth_info)
if all_tasks is None:
all_tasks = []
return jsonify({
'success': True,
'data': {
'running_tasks': running_tasks,
'recent_tasks': all_tasks,
'running_count': len(running_tasks)
}
})
except Exception as e:
print(f"❌ PBS 작업 조회 오류: {e}")
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/pbs/backups/<datastore_name>')
def api_pbs_backups(datastore_name):
"""PBS 백업 목록 조회 (상세)"""
try:
auth_info = pbs_get_auth_ticket()
if not auth_info:
return jsonify({'success': False, 'error': 'PBS 서버 인증 실패'}), 500
# 네임스페이스 목록 조회 (있는 경우)
namespaces = []
try:
ns_data = pbs_api_call(f'admin/datastore/{datastore_name}/namespace', auth_info)
if ns_data:
namespaces = [ns.get('ns', '') for ns in ns_data]
except:
namespaces = [''] # 기본 네임스페이스만
print(f"🔍 PBS 네임스페이스 목록: {namespaces}")
# 모든 백업 그룹 조회 (네임스페이스별)
all_backup_info = []
for ns in namespaces[:5]: # 최대 5개 네임스페이스
ns_param = f'ns={ns}' if ns else ''
# 백업 그룹 조회
groups_url = f'admin/datastore/{datastore_name}/groups'
if ns_param:
groups_url += f'?{ns_param}'
groups = pbs_api_call(groups_url, auth_info)
if not groups:
continue
print(f"🔍 네임스페이스 '{ns}' 백업 그룹 수: {len(groups)}")
# 각 그룹의 상세 정보 조회
for group in groups[:20]: # 네임스페이스당 최대 20개 그룹
backup_type = group.get('backup-type')
backup_id = group.get('backup-id')
group_ns = group.get('ns', '')
if not backup_type or not backup_id:
continue
# 해당 그룹의 스냅샷 조회
snapshot_params = f'backup-type={backup_type}&backup-id={backup_id}'
if group_ns:
snapshot_params += f'&ns={group_ns}'
snapshots = pbs_api_call(
f'admin/datastore/{datastore_name}/snapshots?{snapshot_params}',
auth_info
)
# 수동으로 최신 10개만 선택
if snapshots and len(snapshots) > 10:
snapshots = sorted(snapshots, key=lambda x: x.get('backup-time', 0), reverse=True)[:10]
if snapshots:
latest_snapshot = snapshots[0] if snapshots else None
# 스냅샷 세부 정보
snapshot_details = []
for snap in snapshots[:5]: # 최신 5개만 상세 표시
snapshot_details.append({
'backup_time': snap.get('backup-time'),
'size': snap.get('size', 0),
'protected': snap.get('protected', False),
'comment': snap.get('comment', ''),
'verification': snap.get('verification', {})
})
all_backup_info.append({
'namespace': group_ns or 'root',
'type': backup_type,
'id': backup_id,
'last_backup': latest_snapshot.get('backup-time') if latest_snapshot else None,
'snapshot_count': len(snapshots),
'total_size': sum(s.get('size', 0) for s in snapshots),
'latest_size': latest_snapshot.get('size', 0) if latest_snapshot else 0,
'snapshots': snapshot_details,
'group_comment': group.get('comment', ''),
'last_verification': latest_snapshot.get('verification', {}) if latest_snapshot else {}
})
# 크기별 정렬 (큰 것부터)
all_backup_info.sort(key=lambda x: x['total_size'], reverse=True)
return jsonify({
'success': True,
'data': {
'namespaces': namespaces,
'backups': all_backup_info[:50], # 최대 50개 백업 그룹 표시
'total_groups': len(all_backup_info),
'datastore': datastore_name
}
})
except Exception as e:
print(f"❌ PBS 백업 목록 조회 오류: {e}")
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/pbs/backup-details/<datastore_name>/<backup_type>/<backup_id>')
def api_pbs_backup_details(datastore_name, backup_type, backup_id):
"""특정 백업 그룹의 상세 정보"""
try:
auth_info = pbs_get_auth_ticket()
if not auth_info:
return jsonify({'success': False, 'error': 'PBS 서버 인증 실패'}), 500
namespace = request.args.get('ns', '')
ns_param = f'&ns={namespace}' if namespace else ''
# 스냅샷 목록 조회
snapshots = pbs_api_call(
f'admin/datastore/{datastore_name}/snapshots?backup-type={backup_type}&backup-id={backup_id}{ns_param}',
auth_info
)
if not snapshots:
return jsonify({'success': False, 'error': '스냅샷 조회 실패'}), 500
# 스냅샷 상세 정보
snapshot_list = []
for snap in snapshots:
snapshot_list.append({
'backup_time': snap.get('backup-time'),
'size': snap.get('size', 0),
'protected': snap.get('protected', False),
'comment': snap.get('comment', ''),
'verification': snap.get('verification', {}),
'encrypted': snap.get('encrypted', False),
'fingerprint': snap.get('fingerprint', '')
})
return jsonify({
'success': True,
'data': {
'backup_type': backup_type,
'backup_id': backup_id,
'namespace': namespace,
'snapshots': snapshot_list,
'total_snapshots': len(snapshot_list),
'total_size': sum(s['size'] for s in snapshot_list)
}
})
except Exception as e:
print(f"❌ PBS 백업 상세 조회 오류: {e}")
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/pbs/restore-tasks')
def api_pbs_restore_tasks():
"""PBS 복구 작업 목록 조회"""
try:
auth_info = pbs_get_auth_ticket()
if not auth_info:
return jsonify({'success': False, 'error': 'PBS 서버 인증 실패'}), 500
# 모든 작업 조회 (복구 관련 작업 필터링)
all_tasks = pbs_api_call('nodes/localhost/tasks', auth_info)
if all_tasks is None:
all_tasks = []
# 모든 작업을 카테고리별로 분류
restore_tasks = []
backup_tasks = []
other_tasks = []
for task in all_tasks:
task_type = task.get('type', '')
worker_type = task.get('worker_type', '')
task_id = task.get('upid', '')
task_info = {
'id': task_id,
'type': task_type or worker_type, # type이 비어있으면 worker_type 사용
'worker_type': worker_type,
'starttime': task.get('starttime'),
'endtime': task.get('endtime'),
'status': task.get('status'),
'exitstatus': task.get('exitstatus'),
'user': task.get('user'),
'node': task.get('node', 'localhost'),
'pid': task.get('pid'),
'pstart': task.get('pstart'),
'worker_id': task.get('worker_id')
}
# 실제 작업 타입 결정 (type이 비어있으면 worker_type 사용)
actual_type = (task_type or worker_type).lower()
# 복구 관련 작업 타입들
if any(restore_type in actual_type for restore_type in [
'restore', 'download', 'extract', 'file-restore', 'vm-restore', 'reader'
]):
restore_tasks.append(task_info)
# 백업 관련 작업들
elif any(backup_type in actual_type for backup_type in [
'backup', 'sync', 'verify', 'prune', 'gc', 'garbage-collection', 'upload'
]):
backup_tasks.append(task_info)
# 기타 작업들 (관리, 유지보수 등)
else:
other_tasks.append(task_info)
# 시작시간 기준으로 최신 순 정렬
restore_tasks.sort(key=lambda x: x.get('starttime', 0), reverse=True)
backup_tasks.sort(key=lambda x: x.get('starttime', 0), reverse=True)
return jsonify({
'success': True,
'data': {
'restore_tasks': restore_tasks[:20], # 최근 20개
'backup_tasks': backup_tasks[:20], # 최근 20개
'other_tasks': other_tasks[:10], # 기타 작업 10개
'total_restore_tasks': len(restore_tasks),
'total_backup_tasks': len(backup_tasks),
'total_other_tasks': len(other_tasks),
'total_all_tasks': len(all_tasks)
}
})
except Exception as e:
print(f"❌ PBS 복구 작업 조회 오류: {e}")
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/pbs/task-log/<task_id>')
def api_pbs_task_log(task_id):
"""PBS 작업 로그 조회"""
try:
auth_info = pbs_get_auth_ticket()
if not auth_info:
return jsonify({'success': False, 'error': 'PBS 서버 인증 실패'}), 500
# 작업 로그 조회
log_data = pbs_api_call(f'nodes/localhost/tasks/{task_id}/log', auth_info)
if log_data is None:
return jsonify({'success': False, 'error': '로그 조회 실패'}), 500
# 로그 라인들을 문자열로 변환
if isinstance(log_data, list):
log_lines = []
for line in log_data:
if isinstance(line, dict):
# 로그 라인이 객체인 경우
timestamp = line.get('t', '')
message = line.get('n', '')
log_lines.append({
'timestamp': timestamp,
'message': message,
'line': f"[{timestamp}] {message}" if timestamp else message
})
else:
# 로그 라인이 문자열인 경우
log_lines.append({
'timestamp': '',
'message': str(line),
'line': str(line)
})
else:
log_lines = [{'timestamp': '', 'message': str(log_data), 'line': str(log_data)}]
return jsonify({
'success': True,
'data': {
'task_id': task_id,
'log_lines': log_lines,
'total_lines': len(log_lines)
}
})
except Exception as e:
print(f"❌ PBS 작업 로그 조회 오류: {e}")
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/pbs/task-status/<task_id>')
def api_pbs_task_status(task_id):
"""PBS 작업 상태 상세 조회"""
try:
auth_info = pbs_get_auth_ticket()
if not auth_info:
return jsonify({'success': False, 'error': 'PBS 서버 인증 실패'}), 500
# 작업 상태 조회
task_data = pbs_api_call(f'nodes/localhost/tasks/{task_id}/status', auth_info)
if task_data is None:
# 전체 작업 목록에서 해당 작업 찾기
all_tasks = pbs_api_call('nodes/localhost/tasks', auth_info)
if all_tasks:
task_data = next((t for t in all_tasks if t.get('upid') == task_id), None)
if not task_data:
return jsonify({'success': False, 'error': '작업을 찾을 수 없습니다'}), 404
return jsonify({
'success': True,
'data': task_data
})
except Exception as e:
print(f"❌ PBS 작업 상태 조회 오류: {e}")
return jsonify({'success': False, 'error': str(e)}), 500
# 에러 핸들러 # 에러 핸들러
@app.errorhandler(404) @app.errorhandler(404)

View File

@ -212,6 +212,11 @@
<i class="fas fa-chart-pie text-warning"></i> 매출 대시보드 <i class="fas fa-chart-pie text-warning"></i> 매출 대시보드
</a> </a>
</li> </li>
<li class="nav-item">
<a class="nav-link {% if request.endpoint and 'pbs' in request.endpoint %}active{% endif %}" href="{{ url_for('pbs_monitoring') }}">
<i class="fas fa-server text-info"></i> PBS 백업 서버
</a>
</li>
<li class="nav-item"> <li class="nav-item">
<a class="nav-link" href="#"> <a class="nav-link" href="#">
<i class="fas fa-chart-line"></i> 모니터링 <i class="fas fa-chart-line"></i> 모니터링

View File

@ -122,9 +122,20 @@
</div> </div>
<div class="small text-success"> <div class="small text-success">
<i class="fas fa-link"></i> <code id="magicDns-{{ machine_data.id }}">{{ machine_data.machine_name or machine_data.hostname }}.headscale.local</code> <i class="fas fa-link"></i> <code id="magicDns-{{ machine_data.id }}">{{ machine_data.machine_name or machine_data.hostname }}.headscale.local</code>
<button class="btn btn-sm btn-outline-secondary ms-1" onclick="copyToClipboard('{{ machine_data.machine_name or machine_data.hostname }}.headscale.local')" title="Magic DNS 주소 복사"> {% set machine_name = (machine_data.machine_name or machine_data.hostname).lower() %}
<i class="fas fa-copy"></i> {% if 'pbs' in machine_name %}
</button> <button class="btn btn-sm btn-outline-info ms-1" onclick="copyToClipboard('{{ machine_data.machine_name or machine_data.hostname }}.headscale.local')" title="Magic DNS 주소 복사 (PBS 서버 - 포트 8007 자동 추가)">
<i class="fas fa-copy"></i><small class="ms-1">:8007</small>
</button>
{% elif 'pve' in machine_name %}
<button class="btn btn-sm btn-outline-warning ms-1" onclick="copyToClipboard('{{ machine_data.machine_name or machine_data.hostname }}.headscale.local')" title="Magic DNS 주소 복사 (Proxmox VE - 포트 8006 자동 추가)">
<i class="fas fa-copy"></i><small class="ms-1">:8006</small>
</button>
{% else %}
<button class="btn btn-sm btn-outline-secondary ms-1" onclick="copyToClipboard('{{ machine_data.machine_name or machine_data.hostname }}.headscale.local')" title="Magic DNS 주소 복사">
<i class="fas fa-copy"></i>
</button>
{% endif %}
</div> </div>
{% if machine_data.hostname != (machine_data.machine_name or machine_data.hostname) %} {% if machine_data.hostname != (machine_data.machine_name or machine_data.hostname) %}
<div class="small text-muted"> <div class="small text-muted">
@ -457,10 +468,37 @@ function deleteNode(nodeId, nodeName) {
}); });
} }
// Magic DNS 주소 클립보드 복사 기능 // 스마트 포트 추가 기능
function addSmartPort(address) {
const lowerAddress = address.toLowerCase();
// PBS가 포함된 경우 :8007 포트 추가
if (lowerAddress.includes('pbs')) {
return address + ':8007';
}
// PVE가 포함된 경우 :8006 포트 추가
if (lowerAddress.includes('pve')) {
return address + ':8006';
}
// 기타 경우는 그대로 반환
return address;
}
// Magic DNS 주소 클립보드 복사 기능 (스마트 포트 지원)
function copyToClipboard(text) { function copyToClipboard(text) {
navigator.clipboard.writeText(text).then(() => { // 스마트 포트 추가 로직 적용
showToast(`Magic DNS 주소가 복사되었습니다: ${text}`, 'success'); const enhancedText = addSmartPort(text);
navigator.clipboard.writeText(enhancedText).then(() => {
// 포트가 추가되었는지 확인하여 메시지 표시
if (enhancedText !== text) {
const port = enhancedText.split(':')[1];
showToast(`Magic DNS 주소가 복사되었습니다 (포트 ${port} 자동 추가): ${enhancedText}`, 'success');
} else {
showToast(`Magic DNS 주소가 복사되었습니다: ${enhancedText}`, 'success');
}
}).catch(err => { }).catch(err => {
console.error('복사 실패:', err); console.error('복사 실패:', err);
showToast('복사에 실패했습니다.', 'danger'); showToast('복사에 실패했습니다.', 'danger');

File diff suppressed because it is too large Load Diff