initial

server.py

@@ -190,12 +190,24 @@ def announce():
 		server["total_clients"] = server["clients"]
 	server["pop_v"] = server["total_clients"] / server["updates"]
 
+	tmp = getErrorPK(server)
+	old_err = errorTracker(tmp)
+	if old_err:
+		app.logger.info("pk=%r err=%r", tmp, old_err)
+
 	finishRequestAsync(server)
 
 	return "Request has been filed.", 202
 
 # Utilities
 
+# returns a primary key suitable for saving and replaying an error unique to a
+# server that was announced.
+def getErrorPK(server):
+	# We need to include the client IP in here, since some failures
+	# only happen depending on it.
+	return "%s/%s/%d" % (server["ip"], server["address"], server["port"])
+
 # Returns ping time in seconds (up), False (down), or None (error).
 def serverUp(info):
 	sock = None
@@ -211,7 +223,7 @@ def serverUp(info):
 		# [7] u8        type (PACKET_TYPE_ORIGINAL)
 		buf = b"\x4f\x45\x74\x03\x00\x00\x00\x01"
 		sock.send(buf)
-		start = time.time()
+		start = time.monotonic()
 		# receive reliable packet of type CONTROL, subtype SET_PEER_ID,
 		#     with our assigned peer id as data
 		# [0] u32        protocol_id (PROTOCOL_ID)
@@ -223,7 +235,7 @@ def serverUp(info):
 		# [11] u8        controltype (CONTROLTYPE_SET_PEER_ID)
 		# [12] session_t peer_id_new
 		data = sock.recv(1024)
-		end = time.time()
+		end = time.monotonic()
 		if not data:
 			return False
 		peer_id = data[12:14]
@@ -416,14 +428,18 @@ def asyncFinishThread(server):
 			type=socket.SOCK_DGRAM,
 			proto=socket.SOL_UDP)
 	except socket.gaierror:
-		app.logger.warning("Unable to get address info for %s." % (server["address"],))
+		err = "Unable to get address info for %s." % (server["address"],)
+		app.logger.warning(err)
+		errorTracker.put(getErrorPK(server), err)
 		return
 
 	if checkAddress:
 		addresses = set(data[4][0] for data in info)
 		if not server["ip"] in addresses:
-			app.logger.warning("Invalid IP %s for address %s (address valid for %s)."
-					% (server["ip"], server["address"], addresses))
+			err = "Requester IP %s does not match host %s (valid: %s)." % \
+				(server["ip"], server["address"], " ".join(addresses))
+			app.logger.warning(err)
+			errorTracker.put(getErrorPK(server), err)
 			return
 
 	geo = geoip_lookup_continent(info[-1][4][0])
@@ -432,8 +448,10 @@ def asyncFinishThread(server):
 
 	server["ping"] = serverUp(info[0])
 	if not server["ping"]:
-		app.logger.warning("Server %s:%d has no ping."
-				% (server["address"], server["port"]))
+		err = "Server %s port %d did not respond to ping (tried %s)" % \
+			(server["address"], server["port"], info[0][4][0])
+		app.logger.warning(err)
+		errorTracker.push(getErrorPK(server), err)
 		return
 
 	del server["action"]
@@ -561,19 +579,46 @@ def update(self, server):
 			self.save()
 
 
+class ErrorTracker:
+	VALIDITY_TIME = 600
+
+	def __init__(self):
+		self.table = {}
+		self.lock = RLock()
+
+	def put(self, k, info):
+		with self.lock:
+			self.table[k] = (time.monotonic() + ErrorTracker.VALIDITY_TIME, info)
+
+	def get(self, k):
+		with self.lock:
+			e = self.table.get(k)
+		if e and e[0] >= time.monotonic():
+			return e
+
+	def cleanup(self):
+		with self.lock:
+			now = time.monotonic()
+			table = {k: e for e in self.table if e[0] >= now}
+			self.table = table
+
+
 class PurgeThread(Thread):
 	def __init__(self):
-		Thread.__init__(self)
-		self.daemon = True
+		Thread.__init__(self, daemon=True)
 	def run(self):
 		while True:
 			time.sleep(60)
 			serverList.purgeOld()
+			errorTracker.cleanup()
+
 
 # Globals / Startup
 
 serverList = ServerList()
 
+errorTracker = ErrorTracker()
+
 PurgeThread().start()
 
 if __name__ == "__main__":