[DE-554] serverId query param (GET/PUT /_admin/log/level)

arango/database.py

@@ -572,13 +572,22 @@ def response_handler(resp: Response) -> Json:
 
         return self._execute(request, response_handler)
 
-    def log_levels(self) -> Result[Json]:
+    def log_levels(self, server_id: Optional[str] = None) -> Result[Json]:
         """Return current logging levels.
 
+        :param server_id: Forward log level to a specific server. This makes it
+            easier to adjust the log levels in clusters because DB-Servers require
+            JWT authentication whereas Coordinators also support authentication
+            using usernames and passwords.
+        :type server_id: str
         :return: Current logging levels.
         :rtype: dict
         """
-        request = Request(method="get", endpoint="/_admin/log/level")
+        params: Params = {}
+        if server_id is not None:
+            params["serverId"] = server_id
+
+        request = Request(method="get", endpoint="/_admin/log/level", params=params)
 
         def response_handler(resp: Response) -> Json:
             if not resp.is_success:
@@ -588,7 +597,9 @@ def response_handler(resp: Response) -> Json:
 
         return self._execute(request, response_handler)
 
-    def set_log_levels(self, **kwargs: str) -> Result[Json]:
+    def set_log_levels(
+        self, server_id: Optional[str] = None, **kwargs: str
+    ) -> Result[Json]:
         """Set the logging levels.
 
         This method takes arbitrary keyword arguments where the keys are the
@@ -604,10 +615,21 @@ def set_log_levels(self, **kwargs: str) -> Result[Json]:
 
         Keys that are not valid logger names are ignored.
 
+        :param server_id: Forward log level to a specific server. This makes it
+            easier to adjust the log levels in clusters because DB-Servers require
+            JWT authentication whereas Coordinators also support authentication
+            using usernames and passwords.
+        :type server_id: str | None
         :return: New logging levels.
         :rtype: dict
         """
-        request = Request(method="put", endpoint="/_admin/log/level", data=kwargs)
+        params: Params = {}
+        if server_id is not None:
+            params["serverId"] = server_id
+
+        request = Request(
+            method="put", endpoint="/_admin/log/level", params=params, data=kwargs
+        )
 
         def response_handler(resp: Response) -> Json:
             if not resp.is_success:

tests/test_database.py

@@ -199,6 +199,10 @@ def test_database_misc_methods(sys_db, db, bad_db):
     # Test get log levels
     assert isinstance(sys_db.log_levels(), dict)
 
+    # Test get log levels (with server_id)
+    server_id = sys_db.replication.server_id()
+    assert isinstance(sys_db.log_levels(server_id), dict)
+
     # Test get log levels with bad database
     with assert_raises(ServerLogLevelError) as err:
         bad_db.log_levels()
@@ -212,6 +216,13 @@ def test_database_misc_methods(sys_db, db, bad_db):
     for key, value in sys_db.log_levels().items():
         assert result[key] == value
 
+    # Test set log levels (with server_id)
+    result = sys_db.set_log_levels(server_id, **new_levels)
+    for key, value in new_levels.items():
+        assert result[key] == value
+    for key, value in sys_db.log_levels(server_id).items():
+        assert result[key] == value
+
     # Test set log levels with bad database
     with assert_raises(ServerLogLevelSetError):
         bad_db.set_log_levels(**new_levels)