Coverage for volexport/api_volume.py: 82%

1import datetime 

2from typing import Annotated 

3from enum import Enum 

4from fastapi import APIRouter, HTTPException 

5from pydantic import BaseModel, Field, AfterValidator 

6from .config2 import config2 

7from .config import config 

8from .lvm2 import LV, VG 

9from .tgtd import Tgtd 

10 

11router = APIRouter() 

12 

13 

14def _is_volsize(value: int): 

15 if value % 512 != 0: 

16 raise ValueError(f"invalid volume size: {value} is not multiple of 512") 

17 return value 

18 

19 

20VolumeSize = Annotated[int, AfterValidator(_is_volsize)] 

21 

22 

23class VolumeCreateRequest(BaseModel): 

24 """Request type for POST /volume""" 

25 

26 name: str = Field(description="Name of the volume to create", examples=["volume1"]) 

27 size: VolumeSize = Field(description="Size of the volume in bytes", examples=[1073741824], gt=0) 

28 

29 

30class VolumeCreateResponse(BaseModel): 

31 """Response type for POST /volume""" 

32 

33 name: str = Field(description="Name of the created volume", examples=["volume1"]) 

34 size: VolumeSize = Field(description="Size of the created volume in bytes", examples=[1073741824], gt=0) 

35 

36 

37class VolumeReadResponse(BaseModel): 

38 """Response type for GET /volume/{name}, GET /volume, POST /volume/{name}""" 

39 

40 name: str = Field(description="Name of the volume", examples=["volume1"]) 

41 created: datetime.datetime = Field(description="Creation timestamp of the volume", examples=["2023-10-01T12:00:00"]) 

42 size: VolumeSize = Field(description="Size of the volume in bytes", examples=[1073741824], gt=0) 

43 used: bool = Field(description="opened or not", examples=[True, False]) 

44 readonly: bool = Field(description="true if read-only", examples=[True, False]) 

45 thin: bool = Field(description="true if thin volume", examples=[True, False]) 

46 parent: str | None = Field(description="parent volname if snapshot") 

47 

48 

49class VolumeUpdateRequest(BaseModel): 

50 """Request type for POST /volume/{name}""" 

51 

52 size: VolumeSize | None = Field( 

53 default=None, description="New size of the volume in bytes", examples=[2147483648], gt=0 

54 ) 

55 readonly: bool | None = Field(default=None, description="Set volume to read-only if true", examples=[True, False]) 

56 

57 

58class Filesystem(str, Enum): 

59 """supported filesystems""" 

60 

61 ext4 = "ext4" 

62 xfs = "xfs" 

63 btrfs = "btrfs" 

64 vfat = "vfat" 

65 ntfs = "ntfs" 

66 exfat = "exfat" 

67 nilfs2 = "nilfs2" 

68 

69 

70class VolumeFormatRequest(BaseModel): 

71 """Request type for POST /volume/{name}/mkfs""" 

72 

73 filesystem: Filesystem = Field(default=Filesystem.ext4, description="Make filesystem in the volume") 

74 label: str | None = Field(default=None, description="Label of filesystem") 

75 

76 

77class SnapshotCreateRequest(BaseModel): 

78 """Request type for POST /volume/{name}/snapshot""" 

79 

80 name: str = Field(description="Name of snapshot volume", examples=["snap001", "snap002"]) 

81 size: int | None = Field(default=None, description="Size of snapshot CoW (ignore if using thinpool)") 

82 

83 

84class PoolStats(BaseModel): 

85 """Response type for GET /stats/volume""" 

86 

87 total: int = Field(description="Total size of the pool in bytes", examples=[10737418240]) 

88 used: int = Field(description="Used size of the pool in bytes", examples=[5368709120]) 

89 free: int = Field(description="Free size of the pool in bytes", examples=[5368709120]) 

90 snapshots: int = Field(description="Number of snapshots in the pool", examples=[5]) 

91 volumes: int = Field(description="Number of volumes in the pool", examples=[10]) 

92 

93 

94@router.get("/volume", description="List all volumes") 

95def list_volume() -> list[VolumeReadResponse]: 

96 return [VolumeReadResponse.model_validate(x) for x in LV(config2.VG).volume_list()] 

97 

98 

99@router.post("/volume", description="Create a new volume") 

100def create_volume(arg: VolumeCreateRequest) -> VolumeCreateResponse: 

101 if config.LVM_THINPOOL: 101 ↛ 102line 101 didn't jump to line 102 because the condition on line 101 was never true

102 return VolumeCreateResponse.model_validate( 

103 LV(config2.VG, arg.name).create_thin(size=arg.size, thinpool=config.LVM_THINPOOL) 

104 ) 

105 return VolumeCreateResponse.model_validate(LV(config2.VG, arg.name).create(size=arg.size)) 

106 

107 

108@router.get("/volume/{name}", description="Read volume details by name") 

109def read_volume(name) -> VolumeReadResponse: 

110 res = LV(config2.VG, name).volume_read() 

111 if res is None: 

112 raise HTTPException(status_code=404, detail="volume not found") 

113 return VolumeReadResponse.model_validate(res) 

114 

115 

116@router.delete("/volume/{name}", description="Delete a volume by name") 

117def delete_volume(name) -> dict: 

118 LV(config2.VG, name).delete() 

119 return {} 

120 

121 

122@router.post("/volume/{name}/snapshot", description="Create snapshot") 

123def create_snapshot(name, arg: SnapshotCreateRequest) -> VolumeReadResponse: 

124 if config.LVM_THINPOOL: 124 ↛ 125line 124 didn't jump to line 125 because the condition on line 124 was never true

125 res = LV(config2.VG, arg.name).create_thinsnap(parent=name) 

126 return VolumeReadResponse.model_validate(res) 

127 assert arg.size 

128 res = LV(config2.VG, arg.name).create_snapshot(size=arg.size, parent=name) 

129 return VolumeReadResponse.model_validate(res) 

130 

131 

132@router.get("/volume/{name}/snapshot", description="List snapshot") 

133def list_snapshot(name) -> list[VolumeReadResponse]: 

134 return [VolumeReadResponse.model_validate(x) for x in LV(config2.VG).volume_list() if x.get("parent") == name] 

135 

136 

137@router.get("/volume/{name}/snapshot/{snapname}", description="Read snapshot") 

138def read_snapshot(name, snapname) -> VolumeReadResponse: 

139 # check if name is parent 

140 lv = LV(config2.VG, snapname) 

141 if lv.get_parent() != name: 

142 raise HTTPException(status_code=404, detail="volume not found") 

143 res = LV(config2.VG, snapname).volume_read() 

144 if res is None: 

145 raise HTTPException(status_code=404, detail="volume not found") 

146 return VolumeReadResponse.model_validate(res) 

147 

148 

149@router.delete("/volume/{name}/snapshot/{snapname}", description="Delete snapshot") 

150def delete_snapshot(name, snapname) -> dict: 

151 # check if name is parent 

152 lv = LV(config2.VG, snapname) 

153 if lv.get_parent() != name: 

154 raise HTTPException(status_code=404, detail="volume not found") 

155 lv.delete() 

156 return {} 

157 

158 

159@router.post("/volume/{name}", description="Update a volume by name") 

160def update_volume(name, arg: VolumeUpdateRequest) -> VolumeReadResponse: 

161 lv = LV(config2.VG, name) 

162 if arg.readonly is not None: 

163 lv.read_only(arg.readonly) 

164 if arg.size is not None: 

165 lv.resize(arg.size) 

166 try: 

167 Tgtd().refresh_volume_bypath(lv.volume_vol2path()) 

168 except FileNotFoundError: 

169 # not exported 

170 pass 

171 return VolumeReadResponse.model_validate(lv.volume_read()) 

172 

173 

174@router.post("/volume/{name}/mkfs", description="Format a volume, make filesystem") 

175def format_volume(name, arg: VolumeFormatRequest) -> VolumeReadResponse: 

176 lv = LV(config2.VG, name) 

177 lv.format_volume(arg.filesystem.value, arg.label) 

178 return VolumeReadResponse.model_validate(lv.volume_read()) 

179 

180 

181@router.get("/stats/volume", description="Get statistics of the volume pool") 

182def stats_volume() -> PoolStats: 

183 info = VG(config2.VG).get() 

184 if info is None: 

185 raise HTTPException(status_code=404, detail="pool not found") 

186 vols = int(info["lv_count"]) 

187 total = int(info["vg_size"]) 

188 free = int(info["vg_free"]) 

189 snaps = int(info["snap_count"]) 

190 return PoolStats(total=total, used=total - free, free=free, snapshots=snaps, volumes=vols)