I am running a service that refresh tokens as a hub managed service. It is working completely fine but i am unable too see the logs in the hubs k8s pod logs (log_level : DEBUG). Along with the logs i am creating prometheus metrics that too are not getting populated.
Can someone help me with this? Any help is much appreciated.
Please can you give us enough information to reproduce your problem? Can you show us your configuration, and tell us which version of Z2JH you’re using? How does your service work? Where does it send its logs to? How are you creating and exposing your Prometheus metrics?
Sorry, let me elaborate.
These the configuration for the service
c.JupyterHub.load_roles.append({
"name": "token-mgr",
"scopes": [
"read:users"
],
"services": [
"token"
]
})
c.JupyterHub.services.append(
{
'name': 'token',
'command': [ "python3", "/usr/local/etc/jupyterhub/config/token_service.py" ],
'display': True,
'environment': token_svc_env
}
)
token_service.py
import json
import os
from jupyterhub.services.auth import HubAuthenticated
from jupyterhub.utils import url_path_join
from tornado.httpserver import HTTPServer
from tornado.ioloop import IOLoop
from tornado.web import Application, RequestHandler, authenticated
from custom_logger import get_logger
from pydantic_models import TokenRefreshRequest
from tenant import TenantConfig
from token_manager import TokenManager
class TokenRefreshHandler(HubAuthenticated, RequestHandler):
hub_scopes = ['custom:token:refresh', 'access:services']
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.tenant_config = TenantConfig()
self.token_manager = TokenManager()
self.logger = get_logger(__name__)
self.logger.debug("TokenRefreshHandler initialized.")
@authenticated
async def post(self):
try:
data = json.loads(self.request.body)
except json.JSONDecodeError:
self._write_error(400, message='Invalid JSON')
return None
tenant = data.get('tenant')
jupyterhub_username = data.get('jupyterhub_username')
auth_header = self.request.headers.get("Authorization")
jpy_token = auth_header.split("Bearer ")[1]
token_request_data = TokenRefreshRequest(
jupyterhub_username=jupyterhub_username,
tenant=tenant,
jpy_token=jpy_token
)
self.token_manager.refresh_token(token_request_data)
self.logger.info(f"User : {jupyterhub_username} - User ID token refreshed successfully.")
return self.finish()
def _write_error(self, status_code, **kwargs):
"""Handle errors with a JSON response."""
self.set_status(status_code)
message = kwargs.get('message', 'An error occurred')
self.set_header('Content-Type', 'application/json')
self.write({'error': message})
self.finish()
def main():
app = Application(
[
(
url_path_join(
os.environ['JUPYTERHUB_SERVICE_PREFIX'], '/refresh'
),
TokenRefreshHandler
)
],
cookie_secret=os.urandom(32),
)
http_server = HTTPServer(app)
http_server.listen(10101, os.getenv('POD_IP'))
IOLoop.current().start()
if __name__ == '__main__':
main()
token_manager.py (contains prometheus) (called by token_service.py)
import requests
from pydantic import ValidationError
from prometheus_client import Histogram, Counter
from aws_secret_manager import AWSSecretManager
from custom_logger import get_logger
from pydantic_models import TokenRefreshRequest, TokenRefreshResponse
from tenant import TenantConfig
from token_cryptor import TokenCryptor
from async_metrics import async_time
TOKEN_REFRESH_TIME = Histogram(
'token_refresh_time',
'Time taken to refresh a user token'
)
TOKEN_REFRESH_REQUESTS = Counter(
'token_refresh_requests',
'Total number of token refresh requests',
['status']
)
class TokenManager:
def __init__(self, *args, **kwargs):
self.tenant_config = TenantConfig()
self.aws_secret_manager = AWSSecretManager()
self.token_cryptor = TokenCryptor()
self.logger = get_logger(__name__)
self.logger.debug("TokenManager initialized.")
TOKEN_REFRESH_TIME.time()
def refresh_token(self, refresh_request: TokenRefreshRequest):
user_name = refresh_request.jupyterhub_username
decrypted_refresh_token = self._get_refresh_token(
refresh_request.jupyterhub_username, refresh_request.tenant)
self.logger.debug(f"User : {user_name} - Successfully retrieved refresh token")
new_token_info = self._refresh_user_token(
decrypted_refresh_token, refresh_request.tenant)
if not new_token_info:
self.logger.error(f"User : {user_name} - Failed to refresh tokens")
TOKEN_REFRESH_REQUESTS.labels(status="failure").inc()
return
self.logger.debug(f"User : {user_name} - Successfully refreshed user id token")
self.store_id_token(
refresh_request.jupyterhub_username, new_token_info.id_token,
refresh_request.jpy_token, refresh_request.tenant)
self.logger.debug(f"User : {user_name} - Successfully stored user id token")
self.store_refresh_token(
refresh_request.jupyterhub_username, new_token_info.refresh_token,
refresh_request.tenant)
self.logger.debug(f"User : {user_name} - Successfully stored refresh token")
TOKEN_REFRESH_REQUESTS.labels(status="success").inc()
def _refresh_user_token(self, refresh_token, tenant):
scopes = self.tenant_config.get_scope(tenant)
scopes_str = ' '.join(scopes) if scopes else ''
client_id = self.tenant_config.get_client_id(tenant)
client_secret = self.tenant_config.get_client_secret(tenant)
scopes = scopes_str
token_url = self.tenant_config.get_token_url(tenant)
"""Request new tokens using the refresh token."""
qparams = {
"client_id": client_id,
"client_secret": client_secret,
"grant_type": "refresh_token",
"refresh_token": refresh_token,
"scope": scopes
}
try:
response = requests.post(token_url, data=qparams)
response.raise_for_status()
try:
refresh_token_response = TokenRefreshResponse(**response.json())
return refresh_token_response
except ValidationError as ve:
self.logger.error(f"Validation error while parsing response: {ve}")
self.logger.error(f"Response data: {response.json()}")
return None
except requests.RequestException as e:
self.logger.error("Error refreshing token: %s", e)
return None
def encrypt_refresh_token(self, refresh_token, tenant):
key = self.tenant_config.get_client_secret(tenant)
return self.token_cryptor.encrypt_token(refresh_token, key)
def store_id_token(self, username, token, encryption_key, tenant):
encrypted = self.token_cryptor.encrypt_token(token, encryption_key)
self.aws_secret_manager.update_or_create_secret(f'elara/ephemeral_tokens/id/{tenant}/{username}', encrypted)
return encrypted
def store_refresh_token(self, username, token, tenant):
client_secret = self.tenant_config.get_client_secret(tenant)
encrypted = self.token_cryptor.encrypt_token(token, client_secret)
self.aws_secret_manager.update_or_create_secret(f'elara/ephemeral_tokens/refresh/{tenant}/{username}',
encrypted)
return encrypted
def delete_id_token(self, username, tenant):
self.aws_secret_manager.delete_secret(f'elara/ephemeral_tokens/id/{tenant}/{username}')
def delete_refresh_token(self, username, tenant):
self.aws_secret_manager.delete_secret(f'elara/ephemeral_tokens/refresh/{tenant}/{username}')
def _get_refresh_token(self, username, tenant):
client_secret = self.tenant_config.get_client_secret(tenant)
encrypted_refresh_token = self.aws_secret_manager.get_secret(
f'elara/ephemeral_tokens/refresh/{tenant}/{username}')
return self.token_cryptor.decrypt_token(encrypted_refresh_token, client_secret)
This service is refreshing the user token and is being called by the user server every few hours. I am not sending logs anywhere, i assumed since it is a hub managed service all the logs will be available in the hub pod logs. Do let me know if any other information is required
I am using hub-managed services and I can see the logs procuded by:
import logging
logger = logging.getLogger("myservice")
# Optional logging config to match JupyterHub's log layout
logging.basicConfig(
level=logging.INFO, # Set the threshold for what messages to log
format="[%(levelname).1s %(asctime)s.%(msecs)03d %(name)s %(filename)s:%(lineno)d] %(message)s", # Define the logging format
datefmt="%Y-%m-%d %H:%M:%S", # Format datetime strings
)
logger.info("Hello from my service")
# [I 2025-04-14 10:30:02.551 myservice handler.py:49] Hello from my service
Since you are importing get_logger from custom_logger, how does the custom logger looks like?
1 Like
custom_logger.py
import logging
from logging import Logger, getLogger, StreamHandler, Formatter
def get_logger(name, jpy_logger: Logger = None):
logger = getLogger(name)
if jpy_logger:
for handler in jpy_logger.handlers:
new_handler = type(handler)()
new_handler.setFormatter(handler.formatter)
new_handler.setLevel(logging.DEBUG)
logger.addHandler(new_handler)
else:
handler = StreamHandler()
handler.setLevel(logging.DEBUG)
formatter = Formatter(
fmt="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
datefmt="%Y-%m-%d %H:%M:%S"
)
handler.setFormatter(formatter)
logger.addHandler(handler)
return logger
Can you try adding some log.error(...) statements to your service and see if they appear in the hub logs?
For prometheus metrics you need to run your own server to expose them, for example prometheus_client.start_http_server
You’ll then need to configure your prometheus server to scrape those metrics, either in the prometheus server configuration, or by adding Kubernetes annotations, depending on how you’ve set things up.
Since this is a non-trivial service you might want to run it as a standalone service in it’s own pod/deployment, so you can configure it independently and rely on Kubernetes to manage the service.