Merge pull request #470 from CZerta/rhbz2113900_unix_socket_api

Expose TuneD API to the Unix Domain Socket.

tuned-main.conf

@@ -49,3 +49,29 @@ log_file_max_size = 1MB
 # It can be used to force tuning for specific architecture.
 # If commented, "/proc/cpuinfo" will be read to fill its content.
 # cpuinfo_string = Intel
+
+# Enable TuneD listening on dbus
+# enable_dbus = 1
+
+# Enable TuneD listening on unix domain socket
+# As this functionality is not used commonly, we disable it by default
+# and it is needed to allow it manually
+# enable_unix_socket = 0
+
+# Path to socket for TuneD to listen
+# Existing files on given path will be removed
+# unix_socket_path = /run/tuned/tuned.sock
+
+# Paths to sockets for TuneD to send signals to separated by , or ;
+# unix_socket_signal_paths =
+
+# Default unix socket ownership
+# Can be set as id or name, -1 or non-existing name leaves unchanged
+# unix_socket_ownership = -1 -1
+
+# Permissions for listening sockets
+# unix_socket_permissions = 0o600
+
+# Size of connections backlog for listen function on socket
+# Higher value allows to process requests from more clients
+# connections_backlog = 1024

tuned.py

@@ -41,6 +41,7 @@ if __name__ == "__main__":
 	parser.add_argument("--log", "-l", nargs = "?", const = consts.LOG_FILE, help = "log to file, default file: " + consts.LOG_FILE)
 	parser.add_argument("--pid", "-P", nargs = "?", const = consts.PID_FILE, help = "write PID file, default file: " + consts.PID_FILE)
 	parser.add_argument("--no-dbus", action = "store_true", help = "do not attach to DBus")
+	parser.add_argument("--no-socket", action = "store_true", help = "do not attach to socket")
 	parser.add_argument("--profile", "-p", action = "store", type=str, metavar = "name", help = "tuning profile to be activated")
 	parser.add_argument('--version', "-v", action = "version", version = "%%(prog)s %s.%s.%s" % (ver.TUNED_VERSION_MAJOR, ver.TUNED_VERSION_MINOR, ver.TUNED_VERSION_PATCH))
 	args = parser.parse_args(sys.argv[1:])
@@ -67,13 +68,25 @@ if __name__ == "__main__":
 
 		app = tuned.daemon.Application(args.profile, config)
 
-		# no daemon mode doesn't need DBus
-		if not config.get_bool(consts.CFG_DAEMON, consts.CFG_DEF_DAEMON):
+		# no daemon mode doesn't need DBus or if disabled in config
+		if not config.get(consts.CFG_DAEMON, consts.CFG_DEF_DAEMON) \
+				or not config.get_bool(consts.CFG_ENABLE_DBUS, consts.CFG_DEF_ENABLE_DBUS):
 			args.no_dbus = True
 
+		# no daemon mode doesn't need sockets or if disabled in config
+		if not config.get(consts.CFG_DAEMON, consts.CFG_DEF_DAEMON) \
+				or not config.get_bool(consts.CFG_ENABLE_UNIX_SOCKET, consts.CFG_DEF_ENABLE_UNIX_SOCKET):
+			args.no_socket = True
+
 		if not args.no_dbus:
 			app.attach_to_dbus(consts.DBUS_BUS, consts.DBUS_OBJECT, consts.DBUS_INTERFACE)
 
+		if not args.no_socket:
+			app.attach_to_unix_socket()
+
+		if not args.no_dbus or not args.no_socket:
+			app.register_controller()
+
 		# always write PID file
 		if args.pid is None:
 			args.pid = consts.PID_FILE

tuned/admin/admin.py

@@ -35,7 +35,7 @@ class Admin(object):
 		if self._dbus:
 			self._controller = tuned.admin.DBusController(consts.DBUS_BUS, consts.DBUS_INTERFACE, consts.DBUS_OBJECT, debug)
 			try:
-				self._controller.set_signal_handler(consts.DBUS_SIGNAL_PROFILE_CHANGED, self._signal_profile_changed_cb)
+				self._controller.set_signal_handler(consts.SIGNAL_PROFILE_CHANGED, self._signal_profile_changed_cb)
 			except TunedAdminDBusException as e:
 				self._error(e)
 				self._dbus = False

tuned/consts.py

@@ -100,6 +100,13 @@ CFG_LOG_FILE_COUNT = "log_file_count"
 CFG_LOG_FILE_MAX_SIZE = "log_file_max_size"
 CFG_UNAME_STRING = "uname_string"
 CFG_CPUINFO_STRING = "cpuinfo_string"
+CFG_ENABLE_DBUS = "enable_dbus"
+CFG_ENABLE_UNIX_SOCKET = "enable_unix_socket"
+CFG_UNIX_SOCKET_PATH = "unix_socket_path"
+CFG_UNIX_SOCKET_SIGNAL_PATHS = "unix_socket_signal_paths"
+CFG_UNIX_SOCKET_OWNERSHIP = "unix_socket_ownership"
+CFG_UNIX_SOCKET_PERMISIONS = "unix_socket_permissions"
+CFG_UNIX_SOCKET_CONNECTIONS_BACKLOG = "connections_backlog"
 CFG_CPU_EPP_FLAG = "hwp_epp"
 
 # no_daemon mode
@@ -130,7 +137,24 @@ CFG_DEF_LOG_FILE_COUNT = 2
 CFG_FUNC_LOG_FILE_COUNT = "getint"
 # default log file max size
 CFG_DEF_LOG_FILE_MAX_SIZE = 1024 * 1024
-
+# default listening on dbus
+CFG_DEF_ENABLE_DBUS = True
+CFG_FUNC_ENABLE_DBUS = "getboolean"
+# default listening on unix socket
+# as it is not used commonly disabled by default
+CFG_DEF_ENABLE_UNIX_SOCKET = False
+CFG_FUNC_ENABLE_UNIX_SOCKET = "getboolean"
+# default unix socket path
+CFG_DEF_UNIX_SOCKET_PATH = "/run/tuned/tuned.sock"
+CFG_DEF_UNIX_SOCKET_SIGNAL_PATHS = ""
+# default unix socket ownership
+# (uid and gid, python2 does not support names out of box, -1 leaves default)
+CFG_DEF_UNIX_SOCKET_OWNERSHIP = "-1 -1"
+# default unix socket permissions
+CFG_DEF_UNIX_SOCKET_PERMISIONS = "0o600"
+# default unix socket conections backlog
+CFG_DEF_UNIX_SOCKET_CONNECTIONS_BACKLOG = "1024"
+CFG_FUNC_UNIX_SOCKET_CONNECTIONS_BACKLOG = "getint"
 
 PATH_CPU_DMA_LATENCY = "/dev/cpu_dma_latency"
 
@@ -138,7 +162,7 @@ PATH_CPU_DMA_LATENCY = "/dev/cpu_dma_latency"
 PROFILE_ATTR_SUMMARY = "summary"
 PROFILE_ATTR_DESCRIPTION = "description"
 
-DBUS_SIGNAL_PROFILE_CHANGED = "profile_changed"
+SIGNAL_PROFILE_CHANGED = "profile_changed"
 
 STR_HINT_REBOOT = "you need to reboot for changes to take effect"
 

tuned/daemon/application.py

@@ -22,6 +22,7 @@ class Application(object):
 		# like e.g. '5.15.13-100.fc34.x86_64'
 		log.info("TuneD: %s, kernel: %s" % (tuned.version.TUNED_VERSION_STR, os.uname()[2]))
 		self._dbus_exporter = None
+		self._unix_socket_exporter = None
 
 		storage_provider = storage.PickleProvider()
 		storage_factory = storage.Factory(storage_provider)
@@ -76,6 +77,19 @@ class Application(object):
 
 		self._dbus_exporter = exports.dbus.DBusExporter(bus_name, interface_name, object_name)
 		exports.register_exporter(self._dbus_exporter)
+
+	def attach_to_unix_socket(self):
+		if self._unix_socket_exporter is not None:
+			raise TunedException("Unix socket interface is already initialized.")
+
+		self._unix_socket_exporter = exports.unix_socket.UnixSocketExporter(self.config.get(consts.CFG_UNIX_SOCKET_PATH),
+																			self.config.get(consts.CFG_UNIX_SOCKET_SIGNAL_PATHS),
+																			self.config.get(consts.CFG_UNIX_SOCKET_OWNERSHIP),
+																			self.config.get_int(consts.CFG_UNIX_SOCKET_PERMISIONS),
+																			self.config.get_int(consts.CFG_UNIX_SOCKET_CONNECTIONS_BACKLOG))
+		exports.register_exporter(self._unix_socket_exporter)
+
+	def register_controller(self):
 		exports.register_object(self._controller)
 
 	def _daemonize_parent(self, parent_in_fd, child_out_fd):

tuned/daemon/controller.py

@@ -61,8 +61,8 @@ class Controller(tuned.exports.interfaces.ExportableInterface):
 		if daemon:
 			self._terminate.clear()
 			# we have to pass some timeout, otherwise signals will not work
-			while not self._cmd.wait(self._terminate, 10):
-				pass
+			while not self._cmd.wait(self._terminate, 1):
+				exports.period_check()
 
 		log.info("terminating controller")
 		self.stop()
@@ -309,6 +309,23 @@ class Controller(tuned.exports.interfaces.ExportableInterface):
 			return False
 		return self._daemon.get_plugin_hints(str(plugin_name))
 
+	@exports.export("s", "b")
+	def register_socket_signal_path(self, path, caller = None):
+		"""Allows to dynamically add sockets to send signals to
+
+		Parameters:
+		path -- path to socket to register for sending signals
+
+		Return:
+		bool -- True on success
+		"""
+		if caller == "":
+			return False
+		if self._daemon._application and self._daemon._application._unix_socket_exporter:
+			self._daemon._application._unix_socket_exporter.register_signal_path(path)
+			return True
+		return False
+
 	# devices - devices to migrate from other instances, string of form "dev1,dev2,dev3,..."
 	#	or "cpulist:CPULIST", where CPULIST is e.g. "0-3,6,8-9"
 	# instance_name - instance where to migrate devices

tuned/daemon/daemon.py

@@ -173,8 +173,8 @@ class Daemon(object):
 	# send notification when profile is changed (everything is setup) or if error occured
 	# result: True - OK, False - error occured
 	def _notify_profile_changed(self, profile_names, result, errstr):
-		if self._application is not None and self._application._dbus_exporter is not None:
-			self._application._dbus_exporter.send_signal(consts.DBUS_SIGNAL_PROFILE_CHANGED, profile_names, result, errstr)
+		if self._application is not None:
+			exports.send_signal(consts.SIGNAL_PROFILE_CHANGED, profile_names, result, errstr)
 		return errstr
 
 	def _full_rollback_required(self):

tuned/exports/__init__.py

@@ -1,6 +1,7 @@
 from . import interfaces
 from . import controller
 from . import dbus_exporter as dbus
+from . import unix_socket_exporter as unix_socket
 
 def export(*args, **kwargs):
 	"""Decorator, use to mark exportable methods."""
@@ -28,6 +29,10 @@ def register_object(instance):
 	ctl = controller.ExportsController.get_instance()
 	return ctl.register_object(instance)
 
+def send_signal(*args, **kwargs):
+	ctl = controller.ExportsController.get_instance()
+	return ctl.send_signal(*args, **kwargs)
+
 def start():
 	ctl = controller.ExportsController.get_instance()
 	return ctl.start()
@@ -35,3 +40,7 @@ def start():
 def stop():
 	ctl = controller.ExportsController.get_instance()
 	return ctl.stop()
+
+def period_check():
+	ctl = controller.ExportsController.get_instance()
+	return ctl.period_check()

tuned/exports/controller.py

@@ -43,6 +43,16 @@ class ExportsController(tuned.patterns.Singleton):
 			kwargs = method.signal_params[1]
 			exporter.signal(method, *args, **kwargs)
 
+	def send_signal(self, signal, *args, **kwargs):
+		"""Register signal to all exporters."""
+		for exporter in self._exporters:
+			exporter.send_signal(signal, *args, **kwargs)
+
+	def period_check(self):
+		"""Allows to perform checks on exporters without special thread."""
+		for exporter in self._exporters:
+			exporter.period_check()
+
 	def _initialize_exports(self):
 		if self._exports_initialized:
 			return

tuned/exports/interfaces.py

@@ -19,3 +19,6 @@ class ExporterInterface(object):
 
 	def stop(self):
 		raise NotImplementedError()
+
+	def period_check(self):
+		pass

tuned/exports/unix_socket_exporter.py

@@ -0,0 +1,246 @@
+import os
+import re
+import pwd, grp
+
+from . import interfaces
+import tuned.logs
+import tuned.consts as consts
+from inspect import ismethod
+import socket
+import json
+import select
+
+log = tuned.logs.get()
+
+class UnixSocketExporter(interfaces.ExporterInterface):
+	"""
+	Export method calls through Unix Domain Socket Interface.
+
+	We take a method to be exported and create a simple wrapper function
+	to call it. This is required as we need the original function to be
+	bound to the original object instance. While the wrapper will be bound
+	to an object we dynamically construct.
+	"""
+
+	def __init__(self, socket_path=consts.CFG_DEF_UNIX_SOCKET_PATH,
+				 signal_paths=consts.CFG_DEF_UNIX_SOCKET_SIGNAL_PATHS,
+				 ownership=consts.CFG_DEF_UNIX_SOCKET_OWNERSHIP,
+				 permissions=consts.CFG_DEF_UNIX_SOCKET_PERMISIONS,
+				 connections_backlog=consts.CFG_DEF_UNIX_SOCKET_CONNECTIONS_BACKLOG):
+
+		self._socket_path = socket_path
+		self._socket_object = None
+		self._socket_signal_paths = re.split(r",;", signal_paths) if signal_paths else []
+		self._socket_signal_objects = []
+		self._ownership = [-1, -1]
+		if ownership:
+			ownership = ownership.split()
+			for i, o in enumerate(ownership[:2]):
+				try:
+					self._ownership[i] = int(o)
+				except ValueError:
+					try:
+						# user
+						if i == 0:
+							self._ownership[i] = pwd.getpwnam(o).pw_uid
+						# group
+						else:
+							self._ownership[i] = grp.getgrnam(o).gr_gid
+					except KeyError:
+						log.error("%s '%s' does not exists, leaving default" % ("User" if i == 0 else "Group", o))
+		self._permissions = permissions
+		self._connections_backlog = connections_backlog
+
+		self._unix_socket_methods = {}
+		self._signals = set()
+		self._conn = None
+		self._channel = None
+
+	def running(self):
+		return self._socket_object is not None
+
+	def export(self, method, in_signature, out_signature):
+		if not ismethod(method):
+			raise Exception("Only bound methods can be exported.")
+
+		method_name = method.__name__
+		if method_name in self._unix_socket_methods:
+			raise Exception("Method with this name (%s) is already exported." % method_name)
+
+		class wrapper(object):
+			def __init__(self, in_signature, out_signature):
+				self._in_signature = in_signature
+				self._out_signature = out_signature
+				
+			def __call__(self, *args, **kwargs):
+				return method(*args, **kwargs)
+
+		self._unix_socket_methods[method_name] = wrapper(in_signature, out_signature)
+
+	def signal(self, method, out_signature):
+		if not ismethod(method):
+			raise Exception("Only bound methods can be exported.")
+		
+		method_name = method.__name__
+		if method_name in self._unix_socket_methods:
+			raise Exception("Method with this name (%s) is already exported." % method_name)
+		
+		class wrapper(object):
+			def __init__(self, out_signature):
+				self._out_signature = out_signature
+			
+			def __call__(self, *args, **kwargs):
+				return method(*args, **kwargs)
+		
+		self._unix_socket_methods[method_name] = wrapper(out_signature)
+		self._signals.add(method_name)
+
+	def send_signal(self, signal, *args, **kwargs):
+		if not signal in self._signals:
+			raise Exception("Signal '%s' doesn't exist." % signal)
+		for p in self._socket_signal_paths:
+			log.debug("Sending signal on socket %s" % p)
+			try:
+				s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+				s.setblocking(False)
+				s.connect(p)
+				self._send_data(s, {"jsonrpc": "2.0", "method": signal, "params": args})
+				s.close()
+			except OSError as e:
+				log.warning("Error while sending signal '%s' to socket '%s': %s" % (signal, p, e))
+
+	def register_signal_path(self, path):
+		self._socket_signal_paths.append(path)
+
+	def _construct_socket_object(self):
+		if self._socket_path:
+			if os.path.exists(self._socket_path):
+				os.unlink(self._socket_path)
+			self._socket_object = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+			self._socket_object.bind(self._socket_path)
+			self._socket_object.listen(self._connections_backlog)
+			os.chown(self._socket_path, self._ownership[0], self._ownership[1])
+			if self._permissions:
+				os.chmod(self._socket_path, self._permissions)
+
+	def start(self):
+		if self.running():
+			return
+
+		self.stop()
+		self._construct_socket_object()
+
+	def stop(self):
+		if self._socket_object:
+			self._socket_object.close()
+
+	def _send_data(self, s, data):
+		log.debug("Sending socket data: %s)" % data)
+		try:
+			s.send(json.dumps(data).encode("utf-8"))
+		except Exception as e:
+			log.warning("Failed to send data '%s': %s" % (data, e))
+
+	def _create_response(self, data, id, error=False):
+		res = {
+			"jsonrpc": "2.0",
+			"id": id
+		}
+		if error:
+			res["error"] = data
+		else:
+			res["result"] = data
+		return res
+
+	def _create_error_responce(self, code, message, id=None, data=None):
+		return self._create_response({
+			"code": code,
+			"message": message,
+			"data": data,
+		}, error=True, id=id)
+
+	def _create_result_response(self, result, id):
+		return self._create_response(result, id)
+
+	def _check_id(self, data):
+		if data.get("id"):
+			return data
+		return None
+
+	def _process_request(self, req):
+		if type(req) != dict or req.get("jsonrpc") != "2.0" or not req.get("method"):
+			return self._create_error_responce(-32600, "Invalid Request")
+		id = req.get("id")
+		ret = None
+		if req["method"] not in self._unix_socket_methods:
+			return self._check_id(self._create_error_responce(-32601, "Method not found", id))
+		try:
+			if not req.get("params"):
+				ret = self._unix_socket_methods[req["method"]]()
+			elif type(req["params"]) in (list, tuple):
+				ret = self._unix_socket_methods[req["method"]](*req["params"])
+			elif type(req["params"]) == dict:
+				ret = self._unix_socket_methods[req["method"]](**req["params"])
+			else:
+				return self._check_id(self._create_error_responce(-32600, "Invalid Request", id))
+		except TypeError as e:
+			return self._check_id(self._create_error_responce(-32602, "Invalid params", id, str(e)))
+		except Exception as e:
+			return self._check_id(self._create_error_responce(1, "Error", id, str(e)))
+		return self._check_id(self._create_result_response(ret, id))
+
+	def period_check(self):
+		"""
+		Periodically checks socket object for new calls. This allows to function without special thread.
+		Interface is according JSON-RPC 2.0 Specification (see https://www.jsonrpc.org/specification)
+		
+		Example calls:
+		
+		printf '[{"jsonrpc": "2.0", "method": "active_profile", "id": 1}, {"jsonrpc": "2.0", "method": "profiles", "id": 2}]' | nc -U /run/tuned/tuned.sock
+		printf '{"jsonrpc": "2.0", "method": "switch_profile", "params": {"profile_name": "balanced"}, "id": 1}' | nc -U /run/tuned/tuned.sock
+		"""
+		if not self.running():
+			return
+		while True:
+			r, _, _ = select.select([self._socket_object], (), (), 0)
+			if r:
+				conn, _ = self._socket_object.accept()
+				try:
+					data = ""
+					while True:
+						rec_data = conn.recv(4096).decode()
+						if not rec_data:
+							break
+						data += rec_data
+				except Exception as e:
+					log.error("Failed to load data of message: %s" % e)
+					continue
+				if data:
+					try:
+						data = json.loads(data)
+					except Exception as e:
+						log.error("Failed to load json data '%s': %s" % (data, e))
+						self._send_data(conn, self._create_error_responce(-32700, "Parse error", str(e)))
+						continue
+					if type(data) not in (tuple, list, dict):
+						log.error("Wrong format of call")
+						self._send_data(conn, self._create_error_responce(-32700, "Parse error", str(e)))
+						continue
+					if type(data) in (tuple, list):
+						if len(data) == 0:
+							self._send_data(conn, self._create_error_responce(-32600, "Invalid Request", str(e)))
+							continue
+						res = []
+						for req in data:
+							r = self._process_request(req)
+							if r:
+								res.append(r)
+						if res:
+							self._send_data(conn, res)
+					else:
+						res = self._process_request(data)
+						if r:
+							self._send_data(conn, res)
+			else:
+				return
+		

tuned/utils/global_config.py

@@ -68,6 +68,15 @@ class GlobalConfig():
 			return True
 		return False
 
+	def get_int(self, key, default = 0):
+		i = self._cfg.get(key, default)
+		if i:
+			if isinstance(i, int):
+				return i
+			else:
+				return int(i, 0)
+		return default
+
 	def set(self, key, value):
 		self._cfg[key] = value