Coverage for dlabel/compose.py: 75%

1import fnmatch

2import io

3import tarfile

4from logging import getLogger

5from pathlib import Path

6from typing import Any

8import docker

9import yaml

11from .util import download_files

13_log = getLogger(__name__)

16def envlist2map(env: list[str], sep: str = "=") -> dict[str, str]:

17 res = {}

18 for i in env:

19 kv = i.split(sep, 1)

20 if len(kv) == 2: 20 ↛ 18line 20 didn't jump to line 18 because the condition on line 20 was always true

21 res[kv[0]] = kv[1]

22 return res

25def portmap2compose(pmap: dict) -> list[str | dict]:

26 res: list[str | dict] = []

27 for k, v in pmap.items():

28 ctport = k

29 if ctport.endswith("/tcp") and len(v) == 1:

30 ctport = k.split("/")[0]

31 hostip = v[0].get("HostIp")

32 hostport = v[0].get("HostPort")

33 if hostip:

34 res.append(f"{hostip}:{hostport}:{ctport}")

35 else:

36 res.append(f"{hostport}:{ctport}")

37 else:

38 target, protocol = k.split("/", 1)

39 res.append(

40 {

41 "target": int(target),

42 "published": int(v[0].get("HostPort")),

43 "protocol": protocol,

44 "mode": "host",

45 }

46 )

47 return res

50def convdict(convmap: dict[str, str], fromdict: dict[str, Any], todict: dict[str, Any]):

51 for k, v in convmap.items():

52 if fromdict.get(k): 52 ↛ 53line 52 didn't jump to line 53 because the condition on line 52 was never true

53 todict[v] = fromdict.get(k)

56def convdict_differ(

57 convmap: dict[str, str],

58 dict_img: dict[str, Any],

59 dict_ctn: dict[str, Any],

60 todict: dict[str, Any],

61):

62 for k, v in convmap.items():

63 if k in dict_ctn and dict_img.get(k) != dict_ctn.get(k): 63 ↛ 64line 63 didn't jump to line 64 because the condition on line 63 was never true

64 todict[v] = dict_ctn[k]

67def copy_files(

68 ctn: docker.models.containers.Container, src: str | Path, dst: str | Path

69):

70 def tfilter(member, path):

71 res = tarfile.data_filter(member, path)

72 if res and "/" in res.name:

73 _, res.name = res.name.split("/", 1)

74 return res

75 return None

77 _log.info("copy %s:%s -> %s", ctn.name, src, dst)

78 odir = Path(dst)

79 bin, arc = ctn.get_archive(str(src))

80 _log.debug("arc=%s", arc)

81 bio = io.BytesIO()

82 for x in bin:

83 bio.write(x)

84 bio.seek(0)

85 tf = tarfile.TarFile(fileobj=bio)

86 members = tf.getmembers()

87 if len(members) == 1 and members[0].isreg():

88 _log.info("single file: %s", members[0])

89 tf.extractall(odir.parent, filter="data")

90 else:

91 odir.mkdir(exist_ok=True, parents=True)

92 tf.extractall(odir, filter=tfilter)

93 tf.close()

94 bio.close()

97def compose(client: docker.DockerClient, project, volume):

98 """generate docker-compose.yml from running containers"""

99 svcs = {}

100 vols = {}

101 nets: dict[str, Any] = {}

102 for ctn in client.containers.list():

103 config = ctn.attrs.get("Config", {})

104 hostconfig = ctn.attrs.get("HostConfig", {})

105 labels: dict[str, str] = config.get("Labels", {})

106 proj = labels.get("com.docker.compose.project")

107 wdir = Path(labels.get("com.docker.compose.project.working_dir", "/"))

108 if project and not proj: 108 ↛ 109line 108 didn't jump to line 109 because the condition on line 108 was never true

109 _log.debug("skip: no project: %s", ctn.name)

110 continue

111 if project and proj and not fnmatch.fnmatch(proj, project):

112 _log.debug("skip by project (%s)", proj)

113 continue

114 name = labels.get("com.docker.compose.service", ctn.name)

115 _log.info("processing %s, service=%s", ctn.name, name)

116 img = ctn.image

117 imglabel = img.labels

118 imgconfig = img.attrs.get("Config", {})

119 for k, v in imglabel.items():

120 if labels.get(k) == v:

121 labels.pop(k)

122 labels = {

123 k: v for k, v in labels.items() if not k.startswith("com.docker.compose.")

124 }

125 envs = envlist2map(config.get("Env", []))

126 imgenv = envlist2map(imgconfig.get("Env", []))

127 for k, v in imgenv.items():

128 if envs.get(k) == v:

129 envs.pop(k)

130 imgvol = imgconfig.get("Volumes", {})

131 cvols = []

132 for i in hostconfig.get("Binds") or []:

133 v = i.split(":", 2)

134 if imgvol and v[1] in imgvol: 134 ↛ 135line 134 didn't jump to line 135 because the condition on line 134 was never true

135 continue

136 src = Path(v[0])

137 dest = v[1]

138 if src.is_relative_to(wdir):

139 srcstr = "./" + str(src.relative_to(wdir))

140 else:

141 srcstr = str(src)

142 if len(v) == 2 or v[2] == "rw":

143 cvols.append(f"{srcstr}:{dest}")

144 elif len(v) == 3: 144 ↛ 146line 144 didn't jump to line 146 because the condition on line 144 was always true

145 cvols.append(f"{srcstr}:{dest}:{v[2]}")

146 if volume and srcstr.startswith("./"): 146 ↛ 147line 146 didn't jump to line 147 because the condition on line 146 was never true

147 for is_dir, tinfo, bin in download_files(ctn, dest):

148 _log.debug(

149 "read from volume: src=%s, is_dir=%s, name=%s, %s bytes",

150 srcstr,

151 is_dir,

152 tinfo.name,

153 len(bin),

154 )

155 yield Path(srcstr) / ".." / tinfo.name, bin

156 elif volume: 156 ↛ 157line 156 didn't jump to line 157 because the condition on line 156 was never true

157 _log.info("skip copy: %s:%s -> %s", name, dest, srcstr)

158 for m in hostconfig.get("Mounts", []):

159 if imgvol and m.get("Target") in imgvol: 159 ↛ 160line 159 didn't jump to line 160 because the condition on line 159 was never true

160 continue

161 volname = m.get("Source")

162 if proj and volname.startswith(proj + "_"): 162 ↛ 164line 162 didn't jump to line 164 because the condition on line 162 was always true

163 volname = volname[len(proj) + 1 :]

164 if m.get("Type") == "volume": 164 ↛ 166line 164 didn't jump to line 166 because the condition on line 164 was always true

165 vols[volname] = m.get("VolumeOptions", {})

166 if m.get("Target"): 166 ↛ 158line 166 didn't jump to line 158 because the condition on line 166 was always true

167 cvols.append(f"{volname}:{m['Target']}")

168 nwmode: str | None = None

169 cnws = []

170 if not proj or hostconfig.get("NetworkMode") != f"{proj}_default": 170 ↛ 172line 170 didn't jump to line 172 because the condition on line 170 was always true

171 nwmode = hostconfig.get("NetworkMode")

172 if isinstance(nwmode, str) and nwmode not in ("host", "none"): 172 ↛ 173line 172 didn't jump to line 173 because the condition on line 172 was never true

173 nets[nwmode] = {}

174 cnws.append(nwmode)

175 nwmode = None

176 svc = {

177 "image": config.get("Image"),

178 }

179 if proj and not ctn.name.startswith(proj + "_"):

180 svc["container_name"] = ctn.name

181 if nwmode: 181 ↛ 182line 181 didn't jump to line 182 because the condition on line 181 was never true

182 svc["network_mode"] = nwmode

183 if cvols:

184 svc["volumes"] = cvols

185 if cnws: 185 ↛ 186line 185 didn't jump to line 186 because the condition on line 185 was never true

186 svc["networks"] = cnws

187 if hostconfig.get("PortBindings"):

188 svc["ports"] = portmap2compose(hostconfig.get("PortBindings", {}))

189 if hostconfig.get("RestartPolicy", {}).get("Name") not in ("no", None):

190 svc["restart"] = hostconfig.get("RestartPolicy", {}).get("Name")

191 if labels:

192 svc["labels"] = labels

193 if envs:

194 svc["environment"] = envs

195 convmap_hostconfig = {

196 "ExtraHosts": "extra_hosts",

197 "CpuShares": "cpu_shares",

198 "CpuPeriod": "cpu_period",

199 "CpuPercent": "cpu_percent",

200 "CpuCount": "cpu_count",

201 "CpuQuota": "cpu_quota",

202 "CpuRealtimeRuntime": "cpu_rt_runtime",

203 "CpuRealtimePeriod": "cpu_rt_period",

204 "CpusetCpus": "cpuset",

205 "CapAdd": "cap_add",

206 "CapDrop": "cap_drop",

207 "CgroupParent": "cgroup_parent",

208 "GroupAdd": "group_add",

209 "Privileged": "privileged",

210 }

211 convmap_label = {

212 "com.docker.compose.depends_on": "depends_on",

213 }

214 convdict(convmap_hostconfig, hostconfig, svc)

215 convdict(convmap_label, config.get("Labels", {}), svc)

216 diffcopy_config = {

217 "Cmd": "command",

218 "Entrypoint": "entrypoint",

219 }

220 convdict_differ(diffcopy_config, imgconfig, config, svc)

221 svcs[name] = svc

222 res = {}

223 if svcs:

224 res["services"] = svcs

225 if vols:

226 res["volumes"] = vols

227 if nets: 227 ↛ 228line 227 didn't jump to line 228 because the condition on line 227 was never true

228 res["networks"] = nets

229 yield (

230 Path("compose.yml"),

231 yaml.dump(res, allow_unicode=True, encoding="utf-8", sort_keys=False),

232 )

233 return res