source file: /home/buildslave/tahoe/edgy/build/src/allmydata/dirnode.py
file stats: 458 lines, 454 executed: 99.1% covered
1.
2. import os, time, math
3.
4. from zope.interface import implements
5. from twisted.internet import defer
6. import simplejson
7. from allmydata.mutable.common import NotMutableError
8. from allmydata.mutable.node import MutableFileNode
9. from allmydata.interfaces import IMutableFileNode, IDirectoryNode,\
10. IURI, IFileNode, IMutableFileURI, IVerifierURI, IFilesystemNode, \
11. ExistingChildError, ICheckable
12. from allmydata.immutable.checker import DeepCheckResults
13. from allmydata.util import hashutil, mathutil, base32, log
14. from allmydata.util.hashutil import netstring
15. from allmydata.util.limiter import ConcurrencyLimiter
16. from allmydata.uri import NewDirectoryURI
17. from pycryptopp.cipher.aes import AES
18.
19. def split_netstring(data, numstrings, allow_leftover=False):
20. """like string.split(), but extracts netstrings. If allow_leftover=False,
21. returns numstrings elements, and throws ValueError if there was leftover
22. data. If allow_leftover=True, returns numstrings+1 elements, in which the
23. last element is the leftover data (possibly an empty string)"""
24. elements = []
25. assert numstrings >= 0
26. while data:
27. colon = data.index(":")
28. length = int(data[:colon])
29. string = data[colon+1:colon+1+length]
30. assert len(string) == length
31. elements.append(string)
32. assert data[colon+1+length] == ","
33. data = data[colon+1+length+1:]
34. if len(elements) == numstrings:
35. break
36. if len(elements) < numstrings:
37. raise ValueError("ran out of netstrings")
38. if allow_leftover:
39. return tuple(elements + [data])
40. if data:
41. raise ValueError("leftover data in netstrings")
42. return tuple(elements)
43.
44. class Deleter:
45. def __init__(self, node, name, must_exist=True):
46. self.node = node
47. self.name = name
48. self.must_exist = True
49. def modify(self, old_contents):
50. children = self.node._unpack_contents(old_contents)
51. if self.name not in children:
52. if self.must_exist:
53. raise KeyError(self.name)
54. self.old_child = None
55. return None
56. self.old_child, metadata = children[self.name]
57. del children[self.name]
58. new_contents = self.node._pack_contents(children)
59. return new_contents
60.
61. class MetadataSetter:
62. def __init__(self, node, name, metadata):
63. self.node = node
64. self.name = name
65. self.metadata = metadata
66.
67. def modify(self, old_contents):
68. children = self.node._unpack_contents(old_contents)
69. children[self.name] = (children[self.name][0], self.metadata)
70. new_contents = self.node._pack_contents(children)
71. return new_contents
72.
73.
74. class Adder:
75. def __init__(self, node, entries=None, overwrite=True):
76. self.node = node
77. if entries is None:
78. entries = []
79. self.entries = entries
80. self.overwrite = overwrite
81.
82. def set_node(self, name, node, metadata):
83. self.entries.append( [name, node, metadata] )
84.
85. def modify(self, old_contents):
86. children = self.node._unpack_contents(old_contents)
87. now = time.time()
88. for e in self.entries:
89. if len(e) == 2:
90. name, child = e
91. new_metadata = None
92. else:
93. assert len(e) == 3
94. name, child, new_metadata = e
95. assert isinstance(name, unicode)
96. if name in children:
97. if not self.overwrite:
98. raise ExistingChildError("child '%s' already exists" % name)
99. metadata = children[name][1].copy()
100. else:
101. metadata = {"ctime": now,
102. "mtime": now}
103. if new_metadata is None:
104. # update timestamps
105. if "ctime" not in metadata:
106. metadata["ctime"] = now
107. metadata["mtime"] = now
108. else:
109. # just replace it
110. metadata = new_metadata.copy()
111. children[name] = (child, metadata)
112. new_contents = self.node._pack_contents(children)
113. return new_contents
114.
115. class NewDirectoryNode:
116. implements(IDirectoryNode, ICheckable)
117. filenode_class = MutableFileNode
118.
119. def __init__(self, client):
120. self._client = client
121. self._most_recent_size = None
122.
123. def __repr__(self):
124. return "<%s %s %s>" % (self.__class__.__name__, self.is_readonly() and "RO" or "RW", hasattr(self, '_uri') and self._uri.abbrev())
125. def init_from_uri(self, myuri):
126. self._uri = IURI(myuri)
127. self._node = self.filenode_class(self._client)
128. self._node.init_from_uri(self._uri.get_filenode_uri())
129. return self
130.
131. def create(self, keypair_generator=None):
132. """
133. Returns a deferred that eventually fires with self once the directory
134. has been created (distributed across a set of storage servers).
135. """
136. # first we create a MutableFileNode with empty_contents, then use its
137. # URI to create our own.
138. self._node = self.filenode_class(self._client)
139. empty_contents = self._pack_contents({})
140. d = self._node.create(empty_contents, keypair_generator)
141. d.addCallback(self._filenode_created)
142. return d
143. def _filenode_created(self, res):
144. self._uri = NewDirectoryURI(IMutableFileURI(self._node.get_uri()))
145. return self
146.
147. def get_size(self):
148. # return the size of our backing mutable file, in bytes, if we've
149. # fetched it.
150. return self._most_recent_size
151.
152. def _set_size(self, data):
153. self._most_recent_size = len(data)
154. return data
155.
156. def _read(self):
157. d = self._node.download_best_version()
158. d.addCallback(self._set_size)
159. d.addCallback(self._unpack_contents)
160. return d
161.
162. def _encrypt_rwcap(self, rwcap):
163. assert isinstance(rwcap, str)
164. IV = os.urandom(16)
165. key = hashutil.mutable_rwcap_key_hash(IV, self._node.get_writekey())
166. cryptor = AES(key)
167. crypttext = cryptor.process(rwcap)
168. mac = hashutil.hmac(key, IV + crypttext)
169. assert len(mac) == 32
170. return IV + crypttext + mac
171.
172. def _decrypt_rwcapdata(self, encwrcap):
173. IV = encwrcap[:16]
174. crypttext = encwrcap[16:-32]
175. mac = encwrcap[-32:]
176. key = hashutil.mutable_rwcap_key_hash(IV, self._node.get_writekey())
177. if mac != hashutil.hmac(key, IV+crypttext):
178. raise hashutil.IntegrityCheckError("HMAC does not match, crypttext is corrupted")
179. cryptor = AES(key)
180. plaintext = cryptor.process(crypttext)
181. return plaintext
182.
183. def _create_node(self, child_uri):
184. return self._client.create_node_from_uri(child_uri)
185.
186. def _unpack_contents(self, data):
187. # the directory is serialized as a list of netstrings, one per child.
188. # Each child is serialized as a list of four netstrings: (name,
189. # rocap, rwcap, metadata), in which the name,rocap,metadata are in
190. # cleartext. The 'name' is UTF-8 encoded. The rwcap is formatted as:
191. # pack("16ss32s", iv, AES(H(writekey+iv), plaintextrwcap), mac)
192. assert isinstance(data, str)
193. # an empty directory is serialized as an empty string
194. if data == "":
195. return {}
196. writeable = not self.is_readonly()
197. children = {}
198. while len(data) > 0:
199. entry, data = split_netstring(data, 1, True)
200. name, rocap, rwcapdata, metadata_s = split_netstring(entry, 4)
201. name = name.decode("utf-8")
202. if writeable:
203. rwcap = self._decrypt_rwcapdata(rwcapdata)
204. child = self._create_node(rwcap)
205. else:
206. child = self._create_node(rocap)
207. metadata = simplejson.loads(metadata_s)
208. assert isinstance(metadata, dict)
209. children[name] = (child, metadata)
210. return children
211.
212. def _pack_contents(self, children):
213. # expects children in the same format as _unpack_contents
214. assert isinstance(children, dict)
215. entries = []
216. for name in sorted(children.keys()):
217. child, metadata = children[name]
218. assert isinstance(name, unicode)
219. assert (IFileNode.providedBy(child)
220. or IMutableFileNode.providedBy(child)
221. or IDirectoryNode.providedBy(child)), (name,child)
222. assert isinstance(metadata, dict)
223. rwcap = child.get_uri() # might be RO if the child is not writeable
224. rocap = child.get_readonly_uri()
225. entry = "".join([netstring(name.encode("utf-8")),
226. netstring(rocap),
227. netstring(self._encrypt_rwcap(rwcap)),
228. netstring(simplejson.dumps(metadata))])
229. entries.append(netstring(entry))
230. return "".join(entries)
231.
232. def is_readonly(self):
233. return self._node.is_readonly()
234. def is_mutable(self):
235. return self._node.is_mutable()
236.
237. def get_uri(self):
238. return self._uri.to_string()
239.
240. def get_readonly_uri(self):
241. return self._uri.get_readonly().to_string()
242.
243. def get_verifier(self):
244. return self._uri.get_verifier().to_string()
245.
246. def get_storage_index(self):
247. return self._uri._filenode_uri.storage_index
248.
249. def check(self, verify=False, repair=False):
250. """Perform a file check. See IChecker.check for details."""
251. return self._node.check(verify, repair)
252.
253. def list(self):
254. """I return a Deferred that fires with a dictionary mapping child
255. name to a tuple of (IFileNode or IDirectoryNode, metadata)."""
256. return self._read()
257.
258. def has_child(self, name):
259. """I return a Deferred that fires with a boolean, True if there
260. exists a child of the given name, False if not."""
261. assert isinstance(name, unicode)
262. d = self._read()
263. d.addCallback(lambda children: children.has_key(name))
264. return d
265.
266. def _get(self, children, name):
267. child = children.get(name)
268. if child is None:
269. raise KeyError(name)
270. return child[0]
271.
272. def get(self, name):
273. """I return a Deferred that fires with the named child node,
274. which is either an IFileNode or an IDirectoryNode."""
275. assert isinstance(name, unicode)
276. d = self._read()
277. d.addCallback(self._get, name)
278. return d
279.
280. def get_metadata_for(self, name):
281. assert isinstance(name, unicode)
282. d = self._read()
283. d.addCallback(lambda children: children[name][1])
284. return d
285.
286. def set_metadata_for(self, name, metadata):
287. assert isinstance(name, unicode)
288. if self.is_readonly():
289. return defer.fail(NotMutableError())
290. assert isinstance(metadata, dict)
291. s = MetadataSetter(self, name, metadata)
292. d = self._node.modify(s.modify)
293. d.addCallback(lambda res: self)
294. return d
295.
296. def get_child_at_path(self, path):
297. """Transform a child path into an IDirectoryNode or IFileNode.
298.
299. I perform a recursive series of 'get' operations to find the named
300. descendant node. I return a Deferred that fires with the node, or
301. errbacks with IndexError if the node could not be found.
302.
303. The path can be either a single string (slash-separated) or a list of
304. path-name elements.
305. """
306.
307. if not path:
308. return defer.succeed(self)
309. if isinstance(path, (list, tuple)):
310. pass
311. else:
312. path = path.split("/")
313. for p in path:
314. assert isinstance(p, unicode)
315. childname = path[0]
316. remaining_path = path[1:]
317. d = self.get(childname)
318. if remaining_path:
319. def _got(node):
320. return node.get_child_at_path(remaining_path)
321. d.addCallback(_got)
322. return d
323.
324. def set_uri(self, name, child_uri, metadata=None, overwrite=True):
325. """I add a child (by URI) at the specific name. I return a Deferred
326. that fires with the child node when the operation finishes. I will
327. replace any existing child of the same name.
328.
329. The child_uri could be for a file, or for a directory (either
330. read-write or read-only, using a URI that came from get_uri() ).
331.
332. If this directory node is read-only, the Deferred will errback with a
333. NotMutableError."""
334. assert isinstance(name, unicode)
335. child_node = self._create_node(child_uri)
336. d = self.set_node(name, child_node, metadata, overwrite)
337. d.addCallback(lambda res: child_node)
338. return d
339.
340. def set_children(self, entries, overwrite=True):
341. # this takes URIs
342. a = Adder(self, overwrite=overwrite)
343. node_entries = []
344. for e in entries:
345. if len(e) == 2:
346. name, child_uri = e
347. metadata = None
348. else:
349. assert len(e) == 3
350. name, child_uri, metadata = e
351. assert isinstance(name, unicode)
352. a.set_node(name, self._create_node(child_uri), metadata)
353. return self._node.modify(a.modify)
354.
355. def set_node(self, name, child, metadata=None, overwrite=True):
356. """I add a child at the specific name. I return a Deferred that fires
357. when the operation finishes. This Deferred will fire with the child
358. node that was just added. I will replace any existing child of the
359. same name.
360.
361. If this directory node is read-only, the Deferred will errback with a
362. NotMutableError."""
363.
364. if self.is_readonly():
365. return defer.fail(NotMutableError())
366. assert isinstance(name, unicode)
367. assert IFilesystemNode.providedBy(child), child
368. a = Adder(self, overwrite=overwrite)
369. a.set_node(name, child, metadata)
370. d = self._node.modify(a.modify)
371. d.addCallback(lambda res: child)
372. return d
373.
374. def set_nodes(self, entries, overwrite=True):
375. if self.is_readonly():
376. return defer.fail(NotMutableError())
377. a = Adder(self, entries, overwrite=overwrite)
378. d = self._node.modify(a.modify)
379. d.addCallback(lambda res: None)
380. return d
381.
382.
383. def add_file(self, name, uploadable, metadata=None, overwrite=True):
384. """I upload a file (using the given IUploadable), then attach the
385. resulting FileNode to the directory at the given name. I return a
386. Deferred that fires (with the IFileNode of the uploaded file) when
387. the operation completes."""
388. assert isinstance(name, unicode)
389. if self.is_readonly():
390. return defer.fail(NotMutableError())
391. d = self._client.upload(uploadable)
392. d.addCallback(lambda results: results.uri)
393. d.addCallback(self._client.create_node_from_uri)
394. d.addCallback(lambda node:
395. self.set_node(name, node, metadata, overwrite))
396. return d
397.
398. def delete(self, name):
399. """I remove the child at the specific name. I return a Deferred that
400. fires (with the node just removed) when the operation finishes."""
401. assert isinstance(name, unicode)
402. if self.is_readonly():
403. return defer.fail(NotMutableError())
404. deleter = Deleter(self, name)
405. d = self._node.modify(deleter.modify)
406. d.addCallback(lambda res: deleter.old_child)
407. return d
408.
409. def create_empty_directory(self, name, overwrite=True):
410. """I create and attach an empty directory at the given name. I return
411. a Deferred that fires (with the new directory node) when the
412. operation finishes."""
413. assert isinstance(name, unicode)
414. if self.is_readonly():
415. return defer.fail(NotMutableError())
416. d = self._client.create_empty_dirnode()
417. def _created(child):
418. entries = [(name, child, None)]
419. a = Adder(self, entries, overwrite=overwrite)
420. d = self._node.modify(a.modify)
421. d.addCallback(lambda res: child)
422. return d
423. d.addCallback(_created)
424. return d
425.
426. def move_child_to(self, current_child_name, new_parent,
427. new_child_name=None, overwrite=True):
428. """I take one of my children and move them to a new parent. The child
429. is referenced by name. On the new parent, the child will live under
430. 'new_child_name', which defaults to 'current_child_name'. I return a
431. Deferred that fires when the operation finishes."""
432. assert isinstance(current_child_name, unicode)
433. if self.is_readonly() or new_parent.is_readonly():
434. return defer.fail(NotMutableError())
435. if new_child_name is None:
436. new_child_name = current_child_name
437. assert isinstance(new_child_name, unicode)
438. d = self.get(current_child_name)
439. def sn(child):
440. return new_parent.set_node(new_child_name, child,
441. overwrite=overwrite)
442. d.addCallback(sn)
443. d.addCallback(lambda child: self.delete(current_child_name))
444. return d
445.
446. def build_manifest(self):
447. """Return a frozenset of verifier-capability strings for all nodes
448. (directories and files) reachable from this one."""
449.
450. # this is just a tree-walker, except that following each edge
451. # requires a Deferred. We use a ConcurrencyLimiter to make sure the
452. # fan-out doesn't cause problems.
453.
454. manifest = set()
455. manifest.add(self.get_verifier())
456. limiter = ConcurrencyLimiter(10) # allow 10 in parallel
457.
458. d = self._build_manifest_from_node(self, manifest, limiter)
459. def _done(res):
460. # LIT nodes have no verifier-capability: their data is stored
461. # inside the URI itself, so there is no need to refresh anything.
462. # They indicate this by returning None from their get_verifier
463. # method. We need to remove any such Nones from our set. We also
464. # want to convert all these caps into strings.
465. return frozenset([IVerifierURI(cap).to_string()
466. for cap in manifest
467. if cap is not None])
468. d.addCallback(_done)
469. return d
470.
471. def _build_manifest_from_node(self, node, manifest, limiter):
472. d = limiter.add(node.list)
473. def _got_list(res):
474. dl = []
475. for name, (child, metadata) in res.iteritems():
476. verifier = child.get_verifier()
477. if verifier not in manifest:
478. manifest.add(verifier)
479. if IDirectoryNode.providedBy(child):
480. dl.append(self._build_manifest_from_node(child,
481. manifest,
482. limiter))
483. if dl:
484. return defer.DeferredList(dl)
485. d.addCallback(_got_list)
486. return d
487.
488. def deep_stats(self):
489. stats = DeepStats()
490. # we track verifier caps, to avoid double-counting children for which
491. # we've got both a write-cap and a read-cap
492. found = set()
493. found.add(self.get_verifier())
494.
495. limiter = ConcurrencyLimiter(10)
496.
497. d = self._add_deepstats_from_node(self, found, stats, limiter)
498. d.addCallback(lambda res: stats.get_results())
499. return d
500.
501. def _add_deepstats_from_node(self, node, found, stats, limiter):
502. d = limiter.add(node.list)
503. def _got_list(children):
504. dl = []
505. dirsize_bytes = node.get_size()
506. dirsize_children = len(children)
507. stats.add("count-directories")
508. stats.add("size-directories", dirsize_bytes)
509. stats.max("largest-directory", dirsize_bytes)
510. stats.max("largest-directory-children", dirsize_children)
511. for name, (child, metadata) in children.iteritems():
512. verifier = child.get_verifier()
513. if verifier in found:
514. continue
515. found.add(verifier)
516. if IDirectoryNode.providedBy(child):
517. dl.append(self._add_deepstats_from_node(child, found,
518. stats, limiter))
519. elif IMutableFileNode.providedBy(child):
520. stats.add("count-files")
521. stats.add("count-mutable-files")
522. # TODO: update the servermap, compute a size, add it to
523. # size-mutable-files, max it into "largest-mutable-file"
524. elif IFileNode.providedBy(child): # CHK and LIT
525. stats.add("count-files")
526. size = child.get_size()
527. stats.histogram("size-files-histogram", size)
528. if child.get_uri().startswith("URI:LIT:"):
529. stats.add("count-literal-files")
530. stats.add("size-literal-files", size)
531. else:
532. stats.add("count-immutable-files")
533. stats.add("size-immutable-files", size)
534. stats.max("largest-immutable-file", size)
535. if dl:
536. return defer.DeferredList(dl)
537. d.addCallback(_got_list)
538. return d
539.
540. def deep_check(self, verify=False, repair=False):
541. # shallow-check each object first, then traverse children
542. root_si = self._node.get_storage_index()
543. self._lp = log.msg(format="deep-check starting (%(si)s),"
544. " verify=%(verify)s, repair=%(repair)s",
545. si=base32.b2a(root_si), verify=verify, repair=repair)
546. results = DeepCheckResults(root_si)
547. found = set()
548. limiter = ConcurrencyLimiter(10)
549.
550. d = self._add_deepcheck_from_node(self, results, found, limiter,
551. verify, repair)
552. def _done(res):
553. log.msg("deep-check done", parent=self._lp)
554. return results
555. d.addCallback(_done)
556. return d
557.
558. def _add_deepcheck_from_node(self, node, results, found, limiter,
559. verify, repair):
560. verifier = node.get_verifier()
561. if verifier in found:
562. # avoid loops
563. return None
564. found.add(verifier)
565.
566. d = limiter.add(node.check, verify, repair)
567. d.addCallback(results.add_check)
568.
569. if IDirectoryNode.providedBy(node):
570. d.addCallback(lambda res: node.list())
571. def _got_children(children):
572. dl = []
573. for name, (child, metadata) in children.iteritems():
574. d2 = self._add_deepcheck_from_node(child, results,
575. found, limiter,
576. verify, repair)
577. if d2:
578. dl.append(d2)
579. if dl:
580. return defer.DeferredList(dl, fireOnOneErrback=True)
581. d.addCallback(_got_children)
582. return d
583.
584.
585. class DeepStats:
586. def __init__(self):
587. self.stats = {}
588. for k in ["count-immutable-files",
589. "count-mutable-files",
590. "count-literal-files",
591. "count-files",
592. "count-directories",
593. "size-immutable-files",
594. #"size-mutable-files",
595. "size-literal-files",
596. "size-directories",
597. "largest-directory",
598. "largest-directory-children",
599. "largest-immutable-file",
600. #"largest-mutable-file",
601. ]:
602. self.stats[k] = 0
603. self.histograms = {}
604. for k in ["size-files-histogram"]:
605. self.histograms[k] = {} # maps (min,max) to count
606. self.buckets = [ (0,0), (1,3)]
607. self.root = math.sqrt(10)
608.
609. def add(self, key, value=1):
610. self.stats[key] += value
611.
612. def max(self, key, value):
613. self.stats[key] = max(self.stats[key], value)
614.
615. def which_bucket(self, size):
616. # return (min,max) such that min <= size <= max
617. # values are from the set (0,0), (1,3), (4,10), (11,31), (32,100),
618. # (101,316), (317, 1000), etc: two per decade
619. assert size >= 0
620. i = 0
621. while True:
622. if i >= len(self.buckets):
623. # extend the list
624. new_lower = self.buckets[i-1][1]+1
625. new_upper = int(mathutil.next_power_of_k(new_lower, self.root))
626. self.buckets.append( (new_lower, new_upper) )
627. maybe = self.buckets[i]
628. if maybe[0] <= size <= maybe[1]:
629. return maybe
630. i += 1
631.
632. def histogram(self, key, size):
633. bucket = self.which_bucket(size)
634. h = self.histograms[key]
635. if bucket not in h:
636. h[bucket] = 0
637. h[bucket] += 1
638.
639. def get_results(self):
640. stats = self.stats.copy()
641. for key in self.histograms:
642. h = self.histograms[key]
643. out = [ (bucket[0], bucket[1], h[bucket]) for bucket in h ]
644. out.sort()
645. stats[key] = out
646. return stats
647.
648.
649. # use client.create_dirnode() to make one of these
650.
651.