src/allmydata/storage_client.py

   1
   2 """
   3 I contain the client-side code which speaks to storage servers, in particular
   4 the foolscap-based server implemented in src/allmydata/storage/*.py .
   5 """
   6
   7 # roadmap:
   8 #
   9 # 1: implement StorageFarmBroker (i.e. "storage broker"), change Client to
  10 # create it, change uploader/servermap to get rrefs from it. ServerFarm calls
  11 # IntroducerClient.subscribe_to . ServerFarm hides descriptors, passes rrefs
  12 # to clients. webapi status pages call broker.get_info_about_serverid.
  13 #
  14 # 2: move get_info methods to the descriptor, webapi status pages call
  15 # broker.get_descriptor_for_serverid().get_info
  16 #
  17 # 3?later?: store descriptors in UploadResults/etc instead of serverids,
  18 # webapi status pages call descriptor.get_info and don't use storage_broker
  19 # or Client
  20 #
  21 # 4: enable static config: tahoe.cfg can add descriptors. Make the introducer
  22 # optional. This closes #467
  23 #
  24 # 5: implement NativeStorageClient, pass it to Tahoe2PeerSelector and other
  25 # clients. Clients stop doing callRemote(), use NativeStorageClient methods
  26 # instead (which might do something else, i.e. http or whatever). The
  27 # introducer and tahoe.cfg only create NativeStorageClients for now.
  28 #
  29 # 6: implement other sorts of IStorageClient classes: S3, etc
  30
  31
  32 import time
  33 from zope.interface import implements, Interface
  34 from foolscap.api import eventually
  35 from allmydata.interfaces import IStorageBroker
  36 from allmydata.util import idlib, log
  37 from allmydata.util.assertutil import precondition
  38 from allmydata.util.rrefutil import add_version_to_remote_reference
  39 from allmydata.util.hashutil import sha1
  40
  41 # who is responsible for de-duplication?
  42 #  both?
  43 #  IC remembers the unpacked announcements it receives, to provide for late
  44 #  subscribers and to remove duplicates
  45
  46 # if a client subscribes after startup, will they receive old announcements?
  47 #  yes
  48
  49 # who will be responsible for signature checking?
  50 #  make it be IntroducerClient, so they can push the filter outwards and
  51 #  reduce inbound network traffic
  52
  53 # what should the interface between StorageFarmBroker and IntroducerClient
  54 # look like?
  55 #  don't pass signatures: only pass validated blessed-objects
  56
  57 class StorageFarmBroker:
  58     implements(IStorageBroker)
  59     """I live on the client, and know about storage servers. For each server
  60     that is participating in a grid, I either maintain a connection to it or
  61     remember enough information to establish a connection to it on demand.
  62     I'm also responsible for subscribing to the IntroducerClient to find out
  63     about new servers as they are announced by the Introducer.
  64     """
  65     def __init__(self, tub, permute_peers):
  66         self.tub = tub
  67         assert permute_peers # False not implemented yet
  68         self.permute_peers = permute_peers
  69         # self.servers maps serverid -> IServer, and keeps track of all the
  70         # storage servers that we've heard about. Each descriptor manages its
  71         # own Reconnector, and will give us a RemoteReference when we ask
  72         # them for it.
  73         self.servers = {}
  74         self.introducer_client = None
  75
  76     # these two are used in unit tests
  77     def test_add_rref(self, serverid, rref):
  78         s = NativeStorageServer(serverid, {})
  79         s.rref = rref
  80         self.servers[serverid] = s
  81
  82     def test_add_server(self, serverid, s):
  83         self.servers[serverid] = s
  84
  85     def use_introducer(self, introducer_client):
  86         self.introducer_client = ic = introducer_client
  87         ic.subscribe_to("storage", self._got_announcement)
  88
  89     def _got_announcement(self, serverid, ann_d):
  90         precondition(isinstance(serverid, str), serverid)
  91         precondition(len(serverid) == 20, serverid)
  92         assert ann_d["service-name"] == "storage"
  93         old = self.servers.get(serverid)
  94         if old:
  95             if old.get_announcement() == ann_d:
  96                 return # duplicate
  97             # replacement
  98             del self.servers[serverid]
  99             old.stop_connecting()
 100             # now we forget about them and start using the new one
 101         dsc = NativeStorageServer(serverid, ann_d)
 102         self.servers[serverid] = dsc
 103         dsc.start_connecting(self.tub, self._trigger_connections)
 104         # the descriptor will manage their own Reconnector, and each time we
 105         # need servers, we'll ask them if they're connected or not.
 106
 107     def _trigger_connections(self):
 108         # when one connection is established, reset the timers on all others,
 109         # to trigger a reconnection attempt in one second. This is intended
 110         # to accelerate server connections when we've been offline for a
 111         # while. The goal is to avoid hanging out for a long time with
 112         # connections to only a subset of the servers, which would increase
 113         # the chances that we'll put shares in weird places (and not update
 114         # existing shares of mutable files). See #374 for more details.
 115         for dsc in self.servers.values():
 116             dsc.try_to_connect()
 117
 118     def get_servers_for_psi(self, peer_selection_index):
 119         # return a list of server objects (IServers)
 120         assert self.permute_peers == True
 121         def _permuted(server):
 122             seed = server.get_permutation_seed()
 123             return sha1(peer_selection_index + seed).digest()
 124         return sorted(self.get_connected_servers(), key=_permuted)
 125
 126     def get_all_serverids(self):
 127         return frozenset(self.servers.keys())
 128
 129     def get_connected_servers(self):
 130         return frozenset([s for s in self.servers.values() if s.get_rref()])
 131
 132     def get_known_servers(self):
 133         return frozenset(self.servers.values())
 134
 135     def get_nickname_for_serverid(self, serverid):
 136         if serverid in self.servers:
 137             return self.servers[serverid].get_nickname()
 138         return None
 139
 140 class IServer(Interface):
 141     """I live in the client, and represent a single server."""
 142     def start_connecting(tub, trigger_cb):
 143         pass
 144     def get_nickname():
 145         pass
 146     def get_rref():
 147         pass
 148
 149 class NativeStorageServer:
 150     """I hold information about a storage server that we want to connect to.
 151     If we are connected, I hold the RemoteReference, their host address, and
 152     the their version information. I remember information about when we were
 153     last connected too, even if we aren't currently connected.
 154
 155     @ivar announcement_time: when we first heard about this service
 156     @ivar last_connect_time: when we last established a connection
 157     @ivar last_loss_time: when we last lost a connection
 158
 159     @ivar version: the server's versiondict, from the most recent announcement
 160     @ivar nickname: the server's self-reported nickname (unicode), same
 161
 162     @ivar rref: the RemoteReference, if connected, otherwise None
 163     @ivar remote_host: the IAddress, if connected, otherwise None
 164     """
 165     implements(IServer)
 166
 167     VERSION_DEFAULTS = {
 168         "http://allmydata.org/tahoe/protocols/storage/v1" :
 169         { "maximum-immutable-share-size": 2**32,
 170           "tolerates-immutable-read-overrun": False,
 171           "delete-mutable-shares-with-zero-length-writev": False,
 172           },
 173         "application-version": "unknown: no get_version()",
 174         }
 175
 176     def __init__(self, serverid, ann_d, min_shares=1):
 177         self.serverid = serverid
 178         self._tubid = serverid
 179         self.announcement = ann_d
 180         self.min_shares = min_shares
 181
 182         self.serverid_s = idlib.shortnodeid_b2a(self.serverid)
 183         self.announcement_time = time.time()
 184         self.last_connect_time = None
 185         self.last_loss_time = None
 186         self.remote_host = None
 187         self.rref = None
 188         self._reconnector = None
 189         self._trigger_cb = None
 190
 191     def __repr__(self):
 192         return "<NativeStorageServer for %s>" % self.get_name()
 193     def get_serverid(self):
 194         return self._tubid
 195     def get_permutation_seed(self):
 196         return self._tubid
 197     def get_version(self):
 198         if self.rref:
 199             return self.rref.version
 200         return None
 201     def get_name(self): # keep methodname short
 202         return self.serverid_s
 203     def get_longname(self):
 204         return idlib.nodeid_b2a(self._tubid)
 205     def get_lease_seed(self):
 206         return self._tubid
 207     def get_foolscap_write_enabler_seed(self):
 208         return self._tubid
 209
 210     def get_nickname(self):
 211         return self.announcement["nickname"].decode("utf-8")
 212     def get_announcement(self):
 213         return self.announcement
 214     def get_remote_host(self):
 215         return self.remote_host
 216     def get_last_connect_time(self):
 217         return self.last_connect_time
 218     def get_last_loss_time(self):
 219         return self.last_loss_time
 220     def get_announcement_time(self):
 221         return self.announcement_time
 222
 223     def start_connecting(self, tub, trigger_cb):
 224         furl = self.announcement["FURL"]
 225         self._trigger_cb = trigger_cb
 226         self._reconnector = tub.connectTo(furl, self._got_connection)
 227
 228     def _got_connection(self, rref):
 229         lp = log.msg(format="got connection to %(name)s, getting versions",
 230                      name=self.get_name(),
 231                      facility="tahoe.storage_broker", umid="coUECQ")
 232         if self._trigger_cb:
 233             eventually(self._trigger_cb)
 234         default = self.VERSION_DEFAULTS
 235         d = add_version_to_remote_reference(rref, default)
 236         d.addCallback(self._got_versioned_service, lp)
 237         d.addErrback(log.err, format="storageclient._got_connection",
 238                      name=self.get_name(), umid="Sdq3pg")
 239
 240     def _got_versioned_service(self, rref, lp):
 241         log.msg(format="%(name)s provided version info %(version)s",
 242                 name=self.get_name(), version=rref.version,
 243                 facility="tahoe.storage_broker", umid="SWmJYg",
 244                 level=log.NOISY, parent=lp)
 245
 246         self.last_connect_time = time.time()
 247         self.remote_host = rref.getPeer()
 248         self.rref = rref
 249         rref.notifyOnDisconnect(self._lost)
 250
 251     def get_rref(self):
 252         return self.rref
 253
 254     def _lost(self):
 255         log.msg(format="lost connection to %(name)s", name=self.get_name(),
 256                 facility="tahoe.storage_broker", umid="zbRllw")
 257         self.last_loss_time = time.time()
 258         self.rref = None
 259         self.remote_host = None
 260
 261     def stop_connecting(self):
 262         # used when this descriptor has been superceded by another
 263         self._reconnector.stopConnecting()
 264
 265     def try_to_connect(self):
 266         # used when the broker wants us to hurry up
 267         self._reconnector.reset()
 268
 269 class UnknownServerTypeError(Exception):
 270     pass