speedup server start/stop (waiting for communicate, etc);

extend server socket with timeouts, extend ping with timeout parameter;

fail2ban/client/csocket.py

@@ -32,10 +32,13 @@ import sys
 
 class CSocket:
 	
-	def __init__(self, sock="/var/run/fail2ban/fail2ban.sock"):
+	def __init__(self, sock="/var/run/fail2ban/fail2ban.sock", timeout=-1):
 		# Create an INET, STREAMing socket
 		#self.csock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 		self.__csock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+		self.__deftout = self.__csock.gettimeout()
+		if timeout != -1:
+			self.settimeout(timeout)
 		#self.csock.connect(("localhost", 2222))
 		self.__csock.connect(sock)
 
@@ -50,6 +53,9 @@ class CSocket:
 		self.__csock.send(obj + CSPROTO.END)
 		return self.receive(self.__csock)
 
+	def settimeout(self, timeout):
+		self.__csock.settimeout(timeout if timeout != -1 else self.__deftout)
+
 	def close(self, sendEnd=True):
 		if not self.__csock:
 			return

fail2ban/client/fail2banclient.py

@@ -36,6 +36,8 @@ from .beautifier import Beautifier
 from .fail2bancmdline import Fail2banCmdLine, ServerExecutionException, ExitException, \
 	logSys, exit, output
 
+from ..server.utils import Utils
+
 PROMPT = "fail2ban> "
 
 
@@ -69,8 +71,9 @@ class Fail2banClient(Fail2banCmdLine, Thread):
 		logSys.warning("Caught signal %d. Exiting" % signum)
 		exit(-1)
 
-	def __ping(self):
+	def __ping(self, timeout=0.1):
-		return self.__processCmd([["ping"]], False)
+		return self.__processCmd([["ping"] + ([timeout] if timeout != -1 else [])],
+			False, timeout=timeout)
 
 	@property
 	def beautifier(self):
@@ -79,7 +82,7 @@ class Fail2banClient(Fail2banCmdLine, Thread):
 		self._beautifier = Beautifier()
 		return self._beautifier
 
-	def __processCmd(self, cmd, showRet=True):
+	def __processCmd(self, cmd, showRet=True, timeout=-1):
 		client = None
 		try:
 			beautifier = self.beautifier
@@ -88,7 +91,11 @@ class Fail2banClient(Fail2banCmdLine, Thread):
 				beautifier.setInputCmd(c)
 				try:
 					if not client:
-						client = CSocket(self._conf["socket"])
+						client = CSocket(self._conf["socket"], timeout=timeout)
+					elif timeout != -1:
+						client.settimeout(timeout)
+					if self._conf["verbose"] > 2:
+						logSys.log(5, "CMD: %r", c)
 					ret = client.send(c)
 					if ret[0] == 0:
 						logSys.log(5, "OK : %r", ret[1])
@@ -101,10 +108,10 @@ class Fail2banClient(Fail2banCmdLine, Thread):
 						streamRet = False
 				except socket.error as e:
 					if showRet or self._conf["verbose"] > 1:
-						if showRet or c != ["ping"]:
+						if showRet or c[0] != "ping":
-							self.__logSocketError()
+							self.__logSocketError(e, c[0] == "ping")
 						else:
-							logSys.log(5, " -- ping failed -- %r", e)
+							logSys.log(5, " -- %s failed -- %r", c, e)
 					return False
 				except Exception as e: # pragma: no cover
 					if showRet or self._conf["verbose"] > 1:
@@ -125,14 +132,18 @@ class Fail2banClient(Fail2banCmdLine, Thread):
 				sys.stdout.flush()
 		return streamRet
 
-	def __logSocketError(self):
+	def __logSocketError(self, prevError="", errorOnly=False):
 		try:
 			if os.access(self._conf["socket"], os.F_OK): # pragma: no cover
 				# This doesn't check if path is a socket,
 				#  but socket.error should be raised
 				if os.access(self._conf["socket"], os.W_OK):
 					# Permissions look good, but socket.error was raised
-					logSys.error("Unable to contact server. Is it running?")
+					if errorOnly:
+						logSys.error(prevError)
+					else:
+						logSys.error("%sUnable to contact server. Is it running?", 
+							("[%s] " % prevError) if prevError else '')
 				else:
 					logSys.error("Permission denied to socket: %s,"
 								 " (you must be root)", self._conf["socket"])
@@ -188,7 +199,7 @@ class Fail2banClient(Fail2banCmdLine, Thread):
 		# Start the server or just initialize started one:
 		try:
 			if background:
-				# Start server daemon as fork of client process:
+				# Start server daemon as fork of client process (or new process):
 				Fail2banServer.startServerAsync(self._conf)
 				# Send config stream to server:
 				if not self.__processStartStreamAfterWait(stream, False):
@@ -233,6 +244,11 @@ class Fail2banClient(Fail2banCmdLine, Thread):
 			logSys.log(5, '  client phase %s', phase)
 		if not stream:
 			return False
+		# wait a litle bit for phase "start-ready" before enter active waiting:
+		if phase is not None:
+			Utils.wait_for(lambda: phase.get('start-ready', None) is not None, 0.5, 0.001)
+			phase['configure'] = (True if stream else False)
+			logSys.log(5, '  client phase %s', phase)
 		# configure server with config stream:
 		ret = self.__processStartStreamAfterWait(stream, False)
 		if phase is not None:
@@ -293,7 +309,7 @@ class Fail2banClient(Fail2banCmdLine, Thread):
 						return False
 					# stop options - jail name or --all
 					break
-			if self.__ping():
+			if self.__ping(timeout=-1):
 				if len(cmd) == 1:
 					jail = '--all'
 					ret, stream = self.readConfig()
@@ -311,6 +327,9 @@ class Fail2banClient(Fail2banCmdLine, Thread):
 				logSys.error("Could not find server")
 				return False
 
+		elif len(cmd) > 1 and cmd[0] == "ping":
+			return self.__processCmd([cmd], timeout=float(cmd[1]))
+
 		else:
 			return self.__processCmd([cmd])
 
@@ -342,21 +361,23 @@ class Fail2banClient(Fail2banCmdLine, Thread):
 		# Wait for the server to start (the server has 30 seconds to answer ping)
 		starttime = time.time()
 		logSys.log(5, "__waitOnServer: %r", (alive, maxtime))
-		test = lambda: os.path.exists(self._conf["socket"]) and self.__ping()
+		sltime = 0.0125 / 2
+		test = lambda: os.path.exists(self._conf["socket"]) and self.__ping(timeout=sltime)
 		with VisualWait(self._conf["verbose"]) as vis:
-			sltime = 0.0125 / 2
 			while self._alive:
 				runf = test()
 				if runf == alive:
 					return True
-				now = time.time()
+				waittime = time.time() - starttime
+				logSys.log(5, "  wait-time: %s", waittime)
 				# Wonderful visual :)
-				if now > starttime + 1:
+				if waittime > 1:
 					vis.heartbeat()
 				# f end time reached:
-				if now - starttime >= maxtime:
+				if waittime >= maxtime:
 					raise ServerExecutionException("Failed to start server")
-				sltime = min(sltime * 2, 0.5)
+				# first 200ms faster:
+				sltime = min(sltime * 2, 0.5 if waittime > 0.2 else 0.1)
 				time.sleep(sltime)
 		return False
 

fail2ban/client/fail2banserver.py

@@ -178,13 +178,17 @@ class Fail2banServer(Fail2banCmdLine):
 				logSys.debug('Configure via async client thread')
 				cli.configureServer(async=True, phase=phase)
 				# wait, do not continue if configuration is not 100% valid:
-				Utils.wait_for(lambda: phase.get('ready', None) is not None, self._conf["timeout"])
+				Utils.wait_for(lambda: phase.get('ready', None) is not None, self._conf["timeout"], 0.001)
+				logSys.log(5, '  server phase %s', phase)
 				if not phase.get('start', False):
 					raise ServerExecutionException('Async configuration of server failed')
 
 			# Start server, daemonize it, etc.
 			pid = os.getpid()
 			server = Fail2banServer.startServerDirect(self._conf, background)
+			if not async:
+				phase['start-ready'] = True
+				logSys.log(5, '  server phase %s', phase)
 			# If forked - just exit other processes
 			if pid != os.getpid(): # pragma: no cover
 				os._exit(0)
@@ -193,7 +197,7 @@ class Fail2banServer(Fail2banCmdLine):
 
 			# wait for client answer "done":
 			if not async and cli:
-				Utils.wait_for(lambda: phase.get('done', None) is not None, self._conf["timeout"])
+				Utils.wait_for(lambda: phase.get('done', None) is not None, self._conf["timeout"], 0.001)
 				if not phase.get('done', False):
 					if server: # pragma: no cover
 						server.quit()

fail2ban/tests/fail2banclienttestcase.py

@@ -184,6 +184,19 @@ def _start_params(tmp, use_stock=False, logtarget="/dev/null", db=":memory:"):
 		"--timeout", str(fail2bancmdline.MAX_WAITTIME),
 	)
 
+def _get_pid_from_file(pidfile):
+	f = pid = None
+	try:
+		f = open(pidfile)
+		pid = f.read()
+		pid = re.match(r'\S+', pid).group()
+		return int(pid)
+	except Exception as e: # pragma: no cover
+		logSys.debug(e)
+	finally:
+		if f is not None:
+			f.close()
+	return pid
 
 def _kill_srv(pidfile):
 	logSys.debug("cleanup: %r", (pidfile, isdir(pidfile)))
@@ -193,23 +206,22 @@ def _kill_srv(pidfile):
 		if not isfile(pidfile): # pragma: no cover
 			pidfile = pjoin(piddir, "fail2ban.pid")
 
+	# output log in heavydebug (to see possible start errors):
+	if unittest.F2B.log_level < logging.DEBUG: # pragma: no cover
+		logfile = pjoin(piddir, "f2b.log")
+		if isfile(logfile):
+			_out_file(logfile)
+		else:
+			logSys.log(5, 'no logfile %r', logfile)
+
 	if not isfile(pidfile):
 		logSys.debug("cleanup: no pidfile for %r", piddir)
 		return True
 
-	f = pid = None
+	logSys.debug("cleanup pidfile: %r", pidfile)
-	try:
+	pid = _get_pid_from_file(pidfile)
-		logSys.debug("cleanup pidfile: %r", pidfile)
+	if pid is None: # pragma: no cover
-		f = open(pidfile)
-		pid = f.read()
-		pid = re.match(r'\S+', pid).group()
-		pid = int(pid)
-	except Exception as e: # pragma: no cover
-		logSys.debug(e)
 		return False
-	finally:
-		if f is not None:
-			f.close()
 
 	try:
 		logSys.debug("cleanup pid: %r", pid)
@@ -443,14 +455,18 @@ class Fail2banClientTest(Fail2banClientServerBase):
 	def testClientStartBackgroundCall(self, tmp):
 		global INTERACT
 		startparams = _start_params(tmp, logtarget=pjoin(tmp, "f2b.log"))
-		# start (in new process, using the same python version):
+		# if fast, start server process from client started direct here:
-		cmd = (sys.executable, pjoin(BIN, CLIENT))
+		if unittest.F2B.fast: # pragma: no cover
-		logSys.debug('Start %s ...', cmd)
+			self.execSuccess(startparams + ("start",))
-		cmd = cmd + startparams + ("--async", "start",)
+		else:
-		ret = Utils.executeCmd(cmd, timeout=MAX_WAITTIME, shell=False, output=True)
+			# start (in new process, using the same python version):
-		self.assertTrue(len(ret) and ret[0])
+			cmd = (sys.executable, pjoin(BIN, CLIENT))
-		# wait for server (socket and ready):
+			logSys.debug('Start %s ...', cmd)
-		self._wait_for_srv(tmp, True, startparams=cmd)
+			cmd = cmd + startparams + ("--async", "start",)
+			ret = Utils.executeCmd(cmd, timeout=MAX_WAITTIME, shell=False, output=True)
+			self.assertTrue(len(ret) and ret[0])
+			# wait for server (socket and ready):
+			self._wait_for_srv(tmp, True, startparams=cmd)
 		self.assertLogged("Server ready")
 		self.pruneLog()
 		try:
@@ -459,6 +475,24 @@ class Fail2banClientTest(Fail2banClientServerBase):
 			self.assertLogged("TEST-ECHO")
 			self.assertLogged("Exit with code 0")
 			self.pruneLog()
+			# test ping timeout:
+			self.execSuccess(startparams, "ping", "0.1")
+			self.assertLogged("Server replied: pong")
+			self.pruneLog()
+			# python 3 seems to bypass such short timeouts also, 
+			# so suspend/resume server process and test between it...
+			pid = _get_pid_from_file(pjoin(tmp, "f2b.pid"))
+			try:
+				# suspend:
+				os.kill(pid, signal.SIGSTOP); # or SIGTSTP?
+				time.sleep(Utils.DEFAULT_SHORT_INTERVAL)
+				# test ping with short timeout:
+				self.execFailed(startparams, "ping", "1e-10")
+			finally:
+				# resume:
+				os.kill(pid, signal.SIGCONT)
+			self.assertLogged("timed out")
+			self.pruneLog()
 			# interactive client chat with started server:
 			INTERACT += [
 				"echo INTERACT-ECHO",