import errno
import json
import sqlite3
+from typing import Sequence, Optional
from .fs.schedule_client import SnapSchedClient
from mgr_module import MgrModule, CLIReadCommand, CLIWriteCommand, Option
from mgr_util import CephfsConnectionException
self._initialized.wait()
return -errno.EINVAL, "", "Unknown command"
- @CLIReadCommand('fs snap-schedule status',
- 'name=path,type=CephString,req=false '
- 'name=subvol,type=CephString,req=false '
- 'name=fs,type=CephString,req=false '
- 'name=format,type=CephString,req=false')
- def snap_schedule_get(self, path='/', subvol=None, fs=None, format='plain'):
+ @CLIReadCommand('fs snap-schedule status')
+ def snap_schedule_get(self,
+ path: Optional[str] = '/',
+ subvol: Optional[str] = None,
+ fs: Optional[str] = None,
+ format: Optional[str] = 'plain'):
'''
List current snapshot schedules
'''
return 0, f'{json_report}', ''
return 0, '\n===\n'.join([ret_sched.report() for ret_sched in ret_scheds]), ''
- @CLIReadCommand('fs snap-schedule list',
- 'name=path,type=CephString '
- 'name=recursive,type=CephString,req=false '
- 'name=subvol,type=CephString,req=false '
- 'name=fs,type=CephString,req=false '
- 'name=format,type=CephString,req=false')
- def snap_schedule_list(self, path, subvol=None, recursive=False, fs=None,
- format='plain'):
+ @CLIReadCommand('fs snap-schedule list')
+ def snap_schedule_list(self, path: str,
+ subvol: Optional[str] = None,
+ recursive: Optional[bool] = False,
+ fs: Optional[str] = None,
+ format: Optional[str] = 'plain'):
'''
Get current snapshot schedule for <path>
'''
return 0, json.dumps(out), ''
return 0, '\n'.join([str(sched) for sched in scheds]), ''
- @CLIWriteCommand('fs snap-schedule add',
- 'name=path,type=CephString '
- 'name=snap-schedule,type=CephString '
- 'name=start,type=CephString,req=false '
- 'name=fs,type=CephString,req=false '
- 'name=subvol,type=CephString,req=false')
+ @CLIWriteCommand('fs snap-schedule add')
def snap_schedule_add(self,
- path,
- snap_schedule,
- start=None,
- fs=None,
- subvol=None):
+ path: str,
+ snap_schedule: Optional[str],
+ start: Optional[str] = None,
+ fs: Optional[str] = None,
+ subvol: Optional[str] = None):
'''
Set a snapshot schedule for <path>
'''
return e.to_tuple()
return 0, suc_msg, ''
- @CLIWriteCommand('fs snap-schedule remove',
- 'name=path,type=CephString '
- 'name=repeat,type=CephString,req=false '
- 'name=start,type=CephString,req=false '
- 'name=subvol,type=CephString,req=false '
- 'name=fs,type=CephString,req=false')
+ @CLIWriteCommand('fs snap-schedule remove')
def snap_schedule_rm(self,
- path,
- repeat=None,
- start=None,
- subvol=None,
- fs=None):
+ path: str,
+ repeat: Optional[str] = None,
+ start: Optional[str] = None,
+ subvol: Optional[str] = None,
+ fs: Optional[str] = None):
'''
Remove a snapshot schedule for <path>
'''
return -errno.ENOENT, '', str(e)
return 0, 'Schedule removed for path {}'.format(path), ''
- @CLIWriteCommand('fs snap-schedule retention add',
- 'name=path,type=CephString '
- 'name=retention-spec-or-period,type=CephString '
- 'name=retention-count,type=CephString,req=false '
- 'name=fs,type=CephString,req=false '
- 'name=subvol,type=CephString,req=false')
+ @CLIWriteCommand('fs snap-schedule retention add')
def snap_schedule_retention_add(self,
- path,
- retention_spec_or_period,
- retention_count=None,
- fs=None,
- subvol=None):
+ path: str,
+ retention_spec_or_period: str,
+ retention_count: Optional[str] = None,
+ fs: Optional[str] = None,
+ subvol: Optional[str] = None):
'''
Set a retention specification for <path>
'''
return -errno.ENOENT, '', str(e)
return 0, 'Retention added to path {}'.format(path), ''
- @CLIWriteCommand('fs snap-schedule retention remove',
- 'name=path,type=CephString '
- 'name=retention-spec-or-period,type=CephString '
- 'name=retention-count,type=CephString,req=false '
- 'name=fs,type=CephString,req=false '
- 'name=subvol,type=CephString,req=false')
+ @CLIWriteCommand('fs snap-schedule retention remove')
def snap_schedule_retention_rm(self,
- path,
- retention_spec_or_period,
- retention_count=None,
- fs=None,
- subvol=None):
+ path: str,
+ retention_spec_or_period: str,
+ retention_count: Optional[str] = None,
+ fs: Optional[str] = None,
+ subvol: Optional[str] = None):
'''
Remove a retention specification for <path>
'''
return -errno.ENOENT, '', str(e)
return 0, 'Retention removed from path {}'.format(path), ''
- @CLIWriteCommand('fs snap-schedule activate',
- 'name=path,type=CephString '
- 'name=repeat,type=CephString,req=false '
- 'name=start,type=CephString,req=false '
- 'name=subvol,type=CephString,req=false '
- 'name=fs,type=CephString,req=false')
+ @CLIWriteCommand('fs snap-schedule activate')
def snap_schedule_activate(self,
- path,
- repeat=None,
- start=None,
- subvol=None,
- fs=None):
+ path: str,
+ repeat: Optional[str] = None,
+ start: Optional[str] = None,
+ subvol: Optional[str] = None,
+ fs: Optional[str] = None):
'''
Activate a snapshot schedule for <path>
'''
return -errno.ENOENT, '', str(e)
return 0, 'Schedule activated for path {}'.format(path), ''
- @CLIWriteCommand('fs snap-schedule deactivate',
- 'name=path,type=CephString '
- 'name=repeat,type=CephString,req=false '
- 'name=start,type=CephString,req=false '
- 'name=subvol,type=CephString,req=false '
- 'name=fs,type=CephString,req=false')
+ @CLIWriteCommand('fs snap-schedule deactivate')
def snap_schedule_deactivate(self,
- path,
- repeat=None,
- start=None,
- subvol=None,
- fs=None):
+ path: str,
+ repeat: Optional[str] = None,
+ start: Optional[str] = None,
+ subvol: Optional[str] = None,
+ fs: Optional[str] = None):
'''
Deactivate a snapshot schedule for <path>
'''