sBubshait/PDG
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: smarter date choosing and fixing nfs mount path

Browse Source
This commit is contained in:
sBubshait 2025-08-05 15:56:42 +03:00
parent bcc44e8089
commit eb8e078a29
1 changed files with 147 additions and 85 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

232
activationDrill/drillServer.py
View File

@ -63,7 +63,10 @@ class ActivationDrillServer:
return [line.strip() for line in output.split('\n') if line.strip()] return [line.strip() for line in output.split('\n') if line.strip()]
def get_dataset_snapshots(self, dataset: str) -> List[Dict]: def get_dataset_snapshots(self, dataset: str) -> List[Dict]:
"""Get all snapshots for a dataset with timestamps""" """Get snapshots for a dataset that match PDG naming convention"""
dataset_basename = dataset.split('/')[-1]
expected_prefix = f"{dataset.replace('/', '-')}-"
success, output = self.run_command(f"zfs list -H -o name,creation -t snapshot | grep '^{dataset}@'") success, output = self.run_command(f"zfs list -H -o name,creation -t snapshot | grep '^{dataset}@'")
if not success: if not success:
return [] return []
@ -76,13 +79,16 @@ class ActivationDrillServer:
snap_name = parts[0].strip() snap_name = parts[0].strip()
creation = parts[1].strip() creation = parts[1].strip()
# Extract timestamp from snapshot name (assuming default format) # Extract snapshot suffix (after @)
snap_suffix = snap_name.split('@')[1] snap_suffix = snap_name.split('@')[1]
snapshots.append({
'name': snap_name, # Only include snapshots that match PDG naming convention
'suffix': snap_suffix, if snap_suffix.startswith(expected_prefix):
'creation': creation snapshots.append({
}) 'name': snap_name,
'suffix': snap_suffix,
'creation': creation
})
# Sort by creation time (newest first) # Sort by creation time (newest first)
snapshots.sort(key=lambda x: x['creation'], reverse=True) snapshots.sort(key=lambda x: x['creation'], reverse=True)
@ -144,121 +150,175 @@ class ActivationDrillServer:
return None return None
def select_snapshot_by_date(self, dataset: str) -> Optional[str]: def select_snapshot_by_date(self, dataset: str) -> Optional[str]:
"""Interactive snapshot selection by date""" """Hierarchical snapshot selection by year/month/day"""
snapshots = self.get_dataset_snapshots(dataset) snapshots = self.get_dataset_snapshots(dataset)
if not snapshots: if not snapshots:
print(f"{Colors.RED}✗ No snapshots found for {dataset}{Colors.ENDC}") print(f"{Colors.RED}✗ No PDG snapshots found for {dataset}{Colors.ENDC}")
return None return None
# Group snapshots by date # Parse all snapshots and group hierarchically
date_groups = {} parsed_snapshots = []
for snap in snapshots: for snap in snapshots:
# Parse snapshot suffix: format is [dataset-name]-YYYY-MM-DD_HH-MM-SS
# Example: pool1-hanaStandby_nfs-2025-08-05_15-36-10
try: try:
suffix = snap['suffix'] suffix = snap['suffix']
# Find the last occurrence of a date pattern (YYYY-MM-DD)
# This handles cases where dataset names contain dashes
import re import re
date_pattern = r'(\d{4}-\d{2}-\d{2})_(\d{2}-\d{2}-\d{2})' date_pattern = r'(\d{4})-(\d{2})-(\d{2})_(\d{2})-(\d{2})-(\d{2})'
match = re.search(date_pattern, suffix) match = re.search(date_pattern, suffix)
if match: if match:
date_part = match.group(1) # YYYY-MM-DD year, month, day, hour, minute, second = match.groups()
time_part = match.group(2) # HH-MM-SS parsed_snapshots.append({
**snap,
if date_part not in date_groups: 'year': year,
date_groups[date_part] = [] 'month': month,
'day': day,
# Store both date and time info 'time': f"{hour}:{minute}:{second}",
snap['parsed_date'] = date_part 'full_date': f"{year}-{month}-{day}"
snap['parsed_time'] = time_part.replace('-', ':') })
date_groups[date_part].append(snap) except Exception:
except Exception as e:
print(f"{Colors.DIM}Warning: Could not parse snapshot {snap['suffix']}: {e}{Colors.ENDC}")
continue continue
if not date_groups: if not parsed_snapshots:
print(f"{Colors.RED}✗ No properly formatted snapshots found{Colors.ENDC}") print(f"{Colors.RED}✗ No properly formatted PDG snapshots found{Colors.ENDC}")
print(f"{Colors.DIM}Expected format: [dataset-name]-YYYY-MM-DD_HH-MM-SS{Colors.ENDC}")
return None return None
dates = sorted(date_groups.keys(), reverse=True) # Group by year, month, day hierarchically
earliest_date = min(dates) years = {}
latest_date = max(dates) for snap in parsed_snapshots:
year = snap['year']
month = snap['month']
day = snap['day']
if year not in years:
years[year] = {}
if month not in years[year]:
years[year][month] = {}
if day not in years[year][month]:
years[year][month][day] = []
years[year][month][day].append(snap)
print(f"\n{Colors.BOLD}📅 Snapshot Date Selection{Colors.ENDC}") # Hierarchical selection
print(f"{Colors.DIM}({len(snapshots)}) Snapshots available from {earliest_date} to {latest_date}{Colors.ENDC}") selected_year = self._select_year(years) if len(years) > 1 else list(years.keys())[0]
if not selected_year:
return None
selected_month = self._select_month(years[selected_year], selected_year) if len(years[selected_year]) > 1 else list(years[selected_year].keys())[0]
if not selected_month:
return None
selected_day = self._select_day(years[selected_year][selected_month], selected_year, selected_month) if len(years[selected_year][selected_month]) > 1 else list(years[selected_year][selected_month].keys())[0]
if not selected_day:
return None
# Show available dates # Select specific snapshot from the chosen day
print(f"\n{Colors.BOLD}Available Dates:{Colors.ENDC}") day_snapshots = years[selected_year][selected_month][selected_day]
for i, date in enumerate(dates, 1): return self._select_time(day_snapshots, selected_year, selected_month, selected_day)
count = len(date_groups[date])
# Format date nicely def _select_year(self, years: Dict) -> Optional[str]:
try: """Select year from available years"""
from datetime import datetime year_list = sorted(years.keys(), reverse=True)
date_obj = datetime.strptime(date, '%Y-%m-%d')
formatted_date = date_obj.strftime('%B %d, %Y (%A)') print(f"\n{Colors.BOLD}📅 Year Selection{Colors.ENDC}")
print(f"{Colors.CYAN}{i:2d}.{Colors.ENDC} {formatted_date} ({count} snapshot{'s' if count > 1 else ''})") for i, year in enumerate(year_list, 1):
except: total_snapshots = sum(len(day_snaps) for month in years[year].values() for day_snaps in month.values())
print(f"{Colors.CYAN}{i:2d}.{Colors.ENDC} {date} ({count} snapshot{'s' if count > 1 else ''})") print(f"{Colors.CYAN}{i:2d}.{Colors.ENDC} {year} ({total_snapshots} snapshots)")
# Select date
while True: while True:
try: try:
choice = input(f"\n{Colors.BOLD}Select date (1-{len(dates)}): {Colors.ENDC}") choice = input(f"\n{Colors.BOLD}Select year (1-{len(year_list)}): {Colors.ENDC}")
idx = int(choice) - 1 idx = int(choice) - 1
if 0 <= idx < len(dates): if 0 <= idx < len(year_list):
selected_date = dates[idx] return year_list[idx]
break
print(f"{Colors.RED}Invalid selection{Colors.ENDC}") print(f"{Colors.RED}Invalid selection{Colors.ENDC}")
except (ValueError, KeyboardInterrupt): except (ValueError, KeyboardInterrupt):
return None return None
def _select_month(self, months: Dict, year: str) -> Optional[str]:
"""Select month from available months"""
month_list = sorted(months.keys(), reverse=True)
month_names = ["", "January", "February", "March", "April", "May", "June",
"July", "August", "September", "October", "November", "December"]
# Show snapshots for selected date print(f"\n{Colors.BOLD}📅 Month Selection for {year}{Colors.ENDC}")
day_snapshots = date_groups[selected_date] for i, month in enumerate(month_list, 1):
total_snapshots = sum(len(day_snaps) for day_snaps in months[month].values())
month_name = month_names[int(month)] if 1 <= int(month) <= 12 else month
print(f"{Colors.CYAN}{i:2d}.{Colors.ENDC} {month_name} ({total_snapshots} snapshots)")
# Format the selected date nicely while True:
try: try:
from datetime import datetime choice = input(f"\n{Colors.BOLD}Select month (1-{len(month_list)}): {Colors.ENDC}")
date_obj = datetime.strptime(selected_date, '%Y-%m-%d') idx = int(choice) - 1
formatted_date = date_obj.strftime('%B %d, %Y') if 0 <= idx < len(month_list):
except: return month_list[idx]
formatted_date = selected_date print(f"{Colors.RED}Invalid selection{Colors.ENDC}")
except (ValueError, KeyboardInterrupt):
print(f"\n{Colors.BOLD}Snapshots for {formatted_date}:{Colors.ENDC}") return None
def _select_day(self, days: Dict, year: str, month: str) -> Optional[str]:
"""Select day from available days"""
day_list = sorted(days.keys(), reverse=True)
month_names = ["", "January", "February", "March", "April", "May", "June",
"July", "August", "September", "October", "November", "December"]
month_name = month_names[int(month)] if 1 <= int(month) <= 12 else month
for i, snap in enumerate(day_snapshots, 1): print(f"\n{Colors.BOLD}📅 Day Selection for {month_name} {year}{Colors.ENDC}")
time_display = snap.get('parsed_time', 'Unknown time') for i, day in enumerate(day_list, 1):
# Make creation time more readable snapshot_count = len(days[day])
try:
from datetime import datetime
date_obj = datetime(int(year), int(month), int(day))
day_name = date_obj.strftime('%A')
print(f"{Colors.CYAN}{i:2d}.{Colors.ENDC} {day_name}, {month_name} {day} ({snapshot_count} snapshots)")
except:
print(f"{Colors.CYAN}{i:2d}.{Colors.ENDC} Day {day} ({snapshot_count} snapshots)")
while True:
try:
choice = input(f"\n{Colors.BOLD}Select day (1-{len(day_list)}): {Colors.ENDC}")
idx = int(choice) - 1
if 0 <= idx < len(day_list):
return day_list[idx]
print(f"{Colors.RED}Invalid selection{Colors.ENDC}")
except (ValueError, KeyboardInterrupt):
return None
def _select_time(self, snapshots: List[Dict], year: str, month: str, day: str) -> Optional[str]:
"""Select specific snapshot by time"""
month_names = ["", "January", "February", "March", "April", "May", "June",
"July", "August", "September", "October", "November", "December"]
month_name = month_names[int(month)] if 1 <= int(month) <= 12 else month
# Sort by time (newest first)
snapshots.sort(key=lambda x: x['time'], reverse=True)
print(f"\n{Colors.BOLD}🕐 Time Selection for {month_name} {day}, {year}{Colors.ENDC}")
for i, snap in enumerate(snapshots, 1):
creation_parts = snap['creation'].split() creation_parts = snap['creation'].split()
if len(creation_parts) >= 4: if len(creation_parts) >= 4:
# Show just date and time, skip timezone and year if same
creation_display = f"{creation_parts[1]} {creation_parts[2]} {creation_parts[3]}" creation_display = f"{creation_parts[1]} {creation_parts[2]} {creation_parts[3]}"
else: else:
creation_display = snap['creation'] creation_display = snap['creation']
print(f"{Colors.CYAN}{i:2d}.{Colors.ENDC} {time_display} {Colors.DIM}(created: {creation_display}){Colors.ENDC}") print(f"{Colors.CYAN}{i:2d}.{Colors.ENDC} {snap['time']} {Colors.DIM}(created: {creation_display}){Colors.ENDC}")
# Select specific snapshot
while True: while True:
try: try:
choice = input(f"\n{Colors.BOLD}Select snapshot (1-{len(day_snapshots)}): {Colors.ENDC}") choice = input(f"\n{Colors.BOLD}Select snapshot (1-{len(snapshots)}): {Colors.ENDC}")
idx = int(choice) - 1 idx = int(choice) - 1
if 0 <= idx < len(day_snapshots): if 0 <= idx < len(snapshots):
selected_snap = day_snapshots[idx]['name'] selected_snap = snapshots[idx]
selected_time = day_snapshots[idx].get('parsed_time', 'Unknown')
# Confirm selection # Confirm selection
print(f"\n{Colors.YELLOW}Selected: {formatted_date} at {selected_time}{Colors.ENDC}") print(f"\n{Colors.YELLOW}Selected: {month_name} {day}, {year} at {selected_snap['time']}{Colors.ENDC}")
print(f"{Colors.DIM}Snapshot: {selected_snap}{Colors.ENDC}") print(f"{Colors.DIM}Snapshot: {selected_snap['name']}{Colors.ENDC}")
confirm = input(f"{Colors.BOLD}Proceed with this snapshot? (y/N): {Colors.ENDC}") confirm = input(f"{Colors.BOLD}Proceed with this snapshot? (y/N): {Colors.ENDC}")
if confirm.lower() == 'y': if confirm.lower() == 'y':
return selected_snap return selected_snap['name']
else: else:
return self.select_snapshot_by_date(dataset) # Start over return self.select_snapshot_by_date(selected_snap['name'].split('@')[0]) # Start over
print(f"{Colors.RED}Invalid selection{Colors.ENDC}") print(f"{Colors.RED}Invalid selection{Colors.ENDC}")
except (ValueError, KeyboardInterrupt): except (ValueError, KeyboardInterrupt):
return None return None
@ -285,16 +345,18 @@ class ActivationDrillServer:
def update_nfs_exports(self, clone_name: str) -> Tuple[bool, str]: def update_nfs_exports(self, clone_name: str) -> Tuple[bool, str]:
"""Add clone to NFS exports if not already present""" """Add clone to NFS exports if not already present"""
# Use full pool path for NFS export
export_path = f"/{clone_name}"
# Check if clone is already in exports # Check if clone is already in exports
if os.path.exists(self.nfs_exports_file): if os.path.exists(self.nfs_exports_file):
with open(self.nfs_exports_file, 'r') as f: with open(self.nfs_exports_file, 'r') as f:
content = f.read() content = f.read()
if clone_name in content: if export_path in content:
return True, "NFS export already exists" return True, "NFS export already exists"
# Add to exports # Add to exports with full path
mount_point = f"/{clone_name.split('/')[-1]}" export_line = f"{export_path} *(rw,sync,no_root_squash,no_subtree_check)\n"
export_line = f"{mount_point} *(rw,sync,no_root_squash,no_subtree_check)\n"
try: try:
with open(self.nfs_exports_file, 'a') as f: with open(self.nfs_exports_file, 'a') as f:
@ -303,7 +365,7 @@ class ActivationDrillServer:
# Reload NFS exports # Reload NFS exports
success, output = self.run_command("exportfs -ra") success, output = self.run_command("exportfs -ra")
if success: if success:
return True, f"NFS export added: {mount_point}" return True, f"NFS export added: {export_path}"
else: else:
return False, f"Failed to reload NFS exports: {output}" return False, f"Failed to reload NFS exports: {output}"
except Exception as e: except Exception as e: