src/allmydata/test/test_hung_server.py

   1 # -*- coding: utf-8 -*-
   2
   3 import os, shutil
   4 from twisted.trial import unittest
   5 from twisted.internet import defer
   6 from allmydata import uri
   7 from allmydata.util.consumer import download_to_data
   8 from allmydata.immutable import upload
   9 from allmydata.mutable.common import UnrecoverableFileError
  10 from allmydata.storage.common import storage_index_to_dir
  11 from allmydata.test.no_network import GridTestMixin
  12 from allmydata.test.common import ShouldFailMixin, _corrupt_share_data
  13 from allmydata.interfaces import NotEnoughSharesError
  14
  15 immutable_plaintext = "data" * 10000
  16 mutable_plaintext = "muta" * 10000
  17
  18 class HungServerDownloadTest(GridTestMixin, ShouldFailMixin, unittest.TestCase):
  19     # Many of these tests take around 60 seconds on François's ARM buildslave:
  20     # http://tahoe-lafs.org/buildbot/builders/FranXois%20lenny-armv5tel
  21     # allmydata.test.test_hung_server.HungServerDownloadTest.test_2_good_8_broken_duplicate_share_fail once ERRORed after 197 seconds on Midnight Magic's NetBSD buildslave:
  22     # http://tahoe-lafs.org/buildbot/builders/MM%20netbsd4%20i386%20warp
  23     # MM's buildslave varies a lot in how long it takes to run tests.
  24
  25     timeout = 240
  26     skip="not ready"
  27
  28     def _break(self, servers):
  29         for (id, ss) in servers:
  30             self.g.break_server(id)
  31
  32     def _hang(self, servers, **kwargs):
  33         for (id, ss) in servers:
  34             self.g.hang_server(id, **kwargs)
  35
  36     def _unhang(self, servers, **kwargs):
  37         for (id, ss) in servers:
  38             self.g.unhang_server(id, **kwargs)
  39
  40     def _delete_all_shares_from(self, servers):
  41         serverids = [id for (id, ss) in servers]
  42         for (i_shnum, i_serverid, i_sharefile) in self.shares:
  43             if i_serverid in serverids:
  44                 os.unlink(i_sharefile)
  45
  46     def _corrupt_all_shares_in(self, servers, corruptor_func):
  47         serverids = [id for (id, ss) in servers]
  48         for (i_shnum, i_serverid, i_sharefile) in self.shares:
  49             if i_serverid in serverids:
  50                 self._corrupt_share((i_shnum, i_sharefile), corruptor_func)
  51
  52     def _copy_all_shares_from(self, from_servers, to_server):
  53         serverids = [id for (id, ss) in from_servers]
  54         for (i_shnum, i_serverid, i_sharefile) in self.shares:
  55             if i_serverid in serverids:
  56                 self._copy_share((i_shnum, i_sharefile), to_server)
  57
  58     def _copy_share(self, share, to_server):
  59         (sharenum, sharefile) = share
  60         (id, ss) = to_server
  61         shares_dir = os.path.join(ss.original.storedir, "shares")
  62         si = uri.from_string(self.uri).get_storage_index()
  63         si_dir = os.path.join(shares_dir, storage_index_to_dir(si))
  64         if not os.path.exists(si_dir):
  65             os.makedirs(si_dir)
  66         new_sharefile = os.path.join(si_dir, str(sharenum))
  67         shutil.copy(sharefile, new_sharefile)
  68         self.shares = self.find_uri_shares(self.uri)
  69         # Make sure that the storage server has the share.
  70         self.failUnless((sharenum, ss.original.my_nodeid, new_sharefile)
  71                         in self.shares)
  72
  73     def _corrupt_share(self, share, corruptor_func):
  74         (sharenum, sharefile) = share
  75         data = open(sharefile, "rb").read()
  76         newdata = corruptor_func(data)
  77         os.unlink(sharefile)
  78         wf = open(sharefile, "wb")
  79         wf.write(newdata)
  80         wf.close()
  81
  82     def _set_up(self, mutable, testdir, num_clients=1, num_servers=10):
  83         self.mutable = mutable
  84         if mutable:
  85             self.basedir = "hung_server/mutable_" + testdir
  86         else:
  87             self.basedir = "hung_server/immutable_" + testdir
  88
  89         self.set_up_grid(num_clients=num_clients, num_servers=num_servers)
  90
  91         self.c0 = self.g.clients[0]
  92         nm = self.c0.nodemaker
  93         self.servers = [(id, ss) for (id, ss) in nm.storage_broker.get_all_servers()]
  94
  95         if mutable:
  96             d = nm.create_mutable_file(mutable_plaintext)
  97             def _uploaded_mutable(node):
  98                 self.uri = node.get_uri()
  99                 self.shares = self.find_uri_shares(self.uri)
 100             d.addCallback(_uploaded_mutable)
 101         else:
 102             data = upload.Data(immutable_plaintext, convergence="")
 103             d = self.c0.upload(data)
 104             def _uploaded_immutable(upload_res):
 105                 self.uri = upload_res.uri
 106                 self.shares = self.find_uri_shares(self.uri)
 107             d.addCallback(_uploaded_immutable)
 108         return d
 109
 110     def _start_download(self):
 111         n = self.c0.create_node_from_uri(self.uri)
 112         if self.mutable:
 113             d = n.download_best_version()
 114             stage_4_d = None # currently we aren't doing any tests which require this for mutable files
 115         else:
 116             d = download_to_data(n)
 117             #stage_4_d = n._downloader._all_downloads.keys()[0]._stage_4_d # too ugly! FIXME
 118             stage_4_d = None
 119         return (d, stage_4_d,)
 120
 121     def _wait_for_data(self, n):
 122         if self.mutable:
 123             d = n.download_best_version()
 124         else:
 125             d = download_to_data(n)
 126         return d
 127
 128     def _check(self, resultingdata):
 129         if self.mutable:
 130             self.failUnlessEqual(resultingdata, mutable_plaintext)
 131         else:
 132             self.failUnlessEqual(resultingdata, immutable_plaintext)
 133
 134     def _download_and_check(self):
 135         d, stage4d = self._start_download()
 136         d.addCallback(self._check)
 137         return d
 138
 139     def _should_fail_download(self):
 140         if self.mutable:
 141             return self.shouldFail(UnrecoverableFileError, self.basedir,
 142                                    "no recoverable versions",
 143                                    self._download_and_check)
 144         else:
 145             return self.shouldFail(NotEnoughSharesError, self.basedir,
 146                                    "ran out of shares",
 147                                    self._download_and_check)
 148
 149
 150     def test_10_good_sanity_check(self):
 151         d = defer.succeed(None)
 152         for mutable in [False, True]:
 153             d.addCallback(lambda ign: self._set_up(mutable, "test_10_good_sanity_check"))
 154             d.addCallback(lambda ign: self._download_and_check())
 155         return d
 156
 157     def test_10_good_copied_share(self):
 158         d = defer.succeed(None)
 159         for mutable in [False, True]:
 160             d.addCallback(lambda ign: self._set_up(mutable, "test_10_good_copied_share"))
 161             d.addCallback(lambda ign: self._copy_all_shares_from(self.servers[2:3], self.servers[0]))
 162             d.addCallback(lambda ign: self._download_and_check())
 163             return d
 164
 165     def test_3_good_7_noshares(self):
 166         d = defer.succeed(None)
 167         for mutable in [False, True]:
 168             d.addCallback(lambda ign: self._set_up(mutable, "test_3_good_7_noshares"))
 169             d.addCallback(lambda ign: self._delete_all_shares_from(self.servers[3:]))
 170             d.addCallback(lambda ign: self._download_and_check())
 171         return d
 172
 173     def test_2_good_8_broken_fail(self):
 174         d = defer.succeed(None)
 175         for mutable in [False, True]:
 176             d.addCallback(lambda ign: self._set_up(mutable, "test_2_good_8_broken_fail"))
 177             d.addCallback(lambda ign: self._break(self.servers[2:]))
 178             d.addCallback(lambda ign: self._should_fail_download())
 179         return d
 180
 181     def test_2_good_8_noshares_fail(self):
 182         d = defer.succeed(None)
 183         for mutable in [False, True]:
 184             d.addCallback(lambda ign: self._set_up(mutable, "test_2_good_8_noshares_fail"))
 185             d.addCallback(lambda ign: self._delete_all_shares_from(self.servers[2:]))
 186             d.addCallback(lambda ign: self._should_fail_download())
 187         return d
 188
 189     def test_2_good_8_broken_copied_share(self):
 190         d = defer.succeed(None)
 191         for mutable in [False, True]:
 192             d.addCallback(lambda ign: self._set_up(mutable, "test_2_good_8_broken_copied_share"))
 193             d.addCallback(lambda ign: self._copy_all_shares_from(self.servers[2:3], self.servers[0]))
 194             d.addCallback(lambda ign: self._break(self.servers[2:]))
 195             d.addCallback(lambda ign: self._download_and_check())
 196         return d
 197
 198     def test_2_good_8_broken_duplicate_share_fail(self):
 199         d = defer.succeed(None)
 200         for mutable in [False, True]:
 201             d.addCallback(lambda ign: self._set_up(mutable, "test_2_good_8_broken_duplicate_share_fail"))
 202             d.addCallback(lambda ign: self._copy_all_shares_from(self.servers[1:2], self.servers[0]))
 203             d.addCallback(lambda ign: self._break(self.servers[2:]))
 204             d.addCallback(lambda ign: self._should_fail_download())
 205         return d
 206
 207     # The tests below do not currently pass for mutable files.
 208
 209     def test_3_good_7_hung(self):
 210         d = defer.succeed(None)
 211         for mutable in [False]:
 212             d.addCallback(lambda ign: self._set_up(mutable, "test_3_good_7_hung"))
 213             d.addCallback(lambda ign: self._hang(self.servers[3:]))
 214             d.addCallback(lambda ign: self._download_and_check())
 215         return d
 216
 217     def test_2_good_8_hung_then_1_recovers(self):
 218         d = defer.succeed(None)
 219         for mutable in [False]:
 220             d.addCallback(lambda ign: self._set_up(mutable, "test_2_good_8_hung_then_1_recovers"))
 221             d.addCallback(lambda ign: self._hang(self.servers[2:3]))
 222             d.addCallback(lambda ign: self._hang(self.servers[3:]))
 223             d.addCallback(lambda ign: self._unhang(self.servers[2:3]))
 224             d.addCallback(lambda ign: self._download_and_check())
 225         return d
 226
 227     def test_2_good_8_hung_then_1_recovers_with_2_shares(self):
 228         d = defer.succeed(None)
 229         for mutable in [False]:
 230             d.addCallback(lambda ign: self._set_up(mutable, "test_2_good_8_hung_then_1_recovers_with_2_shares"))
 231             d.addCallback(lambda ign: self._copy_all_shares_from(self.servers[0:1], self.servers[2]))
 232             d.addCallback(lambda ign: self._hang(self.servers[2:3]))
 233             d.addCallback(lambda ign: self._hang(self.servers[3:]))
 234             d.addCallback(lambda ign: self._unhang(self.servers[2:3]))
 235             d.addCallback(lambda ign: self._download_and_check())
 236         return d
 237
 238     def test_failover_during_stage_4(self):
 239         raise unittest.SkipTest("needs rewrite")
 240         # See #287
 241         d = defer.succeed(None)
 242         for mutable in [False]:
 243             d.addCallback(lambda ign: self._set_up(mutable, "test_failover_during_stage_4"))
 244             d.addCallback(lambda ign: self._corrupt_all_shares_in(self.servers[2:3], _corrupt_share_data))
 245             d.addCallback(lambda ign: self._set_up(mutable, "test_failover_during_stage_4"))
 246             d.addCallback(lambda ign: self._hang(self.servers[3:]))
 247             d.addCallback(lambda ign: self._start_download())
 248             def _after_starting_download((doned, started4d)):
 249                 started4d.addCallback(lambda ign: self._unhang(self.servers[3:4]))
 250                 doned.addCallback(self._check)
 251                 return doned
 252             d.addCallback(_after_starting_download)
 253
 254         return d