Make Celery check less verbose (#974)
This commit is contained in:
commit
2fe518a5dc
6 changed files with 73 additions and 25 deletions
2
Makefile
2
Makefile
|
|
@ -20,7 +20,7 @@ coverage:
|
||||||
|
|
||||||
tests:
|
tests:
|
||||||
@make install_backend
|
@make install_backend
|
||||||
poetry run pytest tests -n auto
|
poetry run pytest tests
|
||||||
|
|
||||||
format:
|
format:
|
||||||
poetry run black .
|
poetry run black .
|
||||||
|
|
|
||||||
|
|
@ -23,7 +23,7 @@ from rich.table import Table
|
||||||
|
|
||||||
console = Console()
|
console = Console()
|
||||||
|
|
||||||
app = typer.Typer()
|
app = typer.Typer(no_args_is_help=True)
|
||||||
|
|
||||||
|
|
||||||
def get_number_of_workers(workers=None):
|
def get_number_of_workers(workers=None):
|
||||||
|
|
@ -141,7 +141,7 @@ def run(
|
||||||
),
|
),
|
||||||
):
|
):
|
||||||
"""
|
"""
|
||||||
Run the Langflow server.
|
Run the Langflow.
|
||||||
"""
|
"""
|
||||||
# override env variables with .env file
|
# override env variables with .env file
|
||||||
if env_file:
|
if env_file:
|
||||||
|
|
@ -299,7 +299,14 @@ def superuser(
|
||||||
password: str = typer.Option(
|
password: str = typer.Option(
|
||||||
..., prompt=True, hide_input=True, help="Password for the superuser."
|
..., prompt=True, hide_input=True, help="Password for the superuser."
|
||||||
),
|
),
|
||||||
|
log_level: str = typer.Option(
|
||||||
|
"critical", help="Logging level.", envvar="LANGFLOW_LOG_LEVEL"
|
||||||
|
),
|
||||||
):
|
):
|
||||||
|
"""
|
||||||
|
Create a superuser.
|
||||||
|
"""
|
||||||
|
configure(log_level=log_level)
|
||||||
initialize_services()
|
initialize_services()
|
||||||
db_service = get_db_service()
|
db_service = get_db_service()
|
||||||
with session_getter(db_service) as session:
|
with session_getter(db_service) as session:
|
||||||
|
|
@ -321,7 +328,10 @@ def superuser(
|
||||||
|
|
||||||
|
|
||||||
@app.command()
|
@app.command()
|
||||||
def migration(test: bool = typer.Option(False, help="Run migrations in test mode.")):
|
def migration(test: bool = typer.Option(True, help="Run migrations in test mode.")):
|
||||||
|
"""
|
||||||
|
Run or test migrations.
|
||||||
|
"""
|
||||||
initialize_services()
|
initialize_services()
|
||||||
db_service = get_db_service()
|
db_service = get_db_service()
|
||||||
if not test:
|
if not test:
|
||||||
|
|
|
||||||
|
|
@ -102,7 +102,10 @@ class CustomComponent(Component, extra=Extra.allow):
|
||||||
status_code=400,
|
status_code=400,
|
||||||
detail={
|
detail={
|
||||||
"error": "Type hint Error",
|
"error": "Type hint Error",
|
||||||
"traceback": "Prompt type is not supported in the build method. Try using PromptTemplate instead.",
|
"traceback": (
|
||||||
|
"Prompt type is not supported in the build method."
|
||||||
|
" Try using PromptTemplate instead."
|
||||||
|
),
|
||||||
},
|
},
|
||||||
)
|
)
|
||||||
return args
|
return args
|
||||||
|
|
|
||||||
|
|
@ -94,9 +94,7 @@ class DatabaseService(Service):
|
||||||
return True
|
return True
|
||||||
|
|
||||||
def run_migrations(self):
|
def run_migrations(self):
|
||||||
logger.info(
|
logger.info(f"Running DB migrations in {self.script_location}")
|
||||||
f"Running DB migrations in {self.script_location} on {self.database_url}"
|
|
||||||
)
|
|
||||||
alembic_cfg = Config()
|
alembic_cfg = Config()
|
||||||
alembic_cfg.set_main_option("script_location", str(self.script_location))
|
alembic_cfg.set_main_option("script_location", str(self.script_location))
|
||||||
alembic_cfg.set_main_option("sqlalchemy.url", self.database_url)
|
alembic_cfg.set_main_option("sqlalchemy.url", self.database_url)
|
||||||
|
|
|
||||||
|
|
@ -1,4 +1,5 @@
|
||||||
from typing import Any, Callable, Coroutine, Union
|
from typing import Any, Callable, Coroutine, Union
|
||||||
|
from langflow.utils.logger import configure
|
||||||
from loguru import logger
|
from loguru import logger
|
||||||
from langflow.services.base import Service
|
from langflow.services.base import Service
|
||||||
from langflow.services.task.backends.anyio import AnyIOBackend
|
from langflow.services.task.backends.anyio import AnyIOBackend
|
||||||
|
|
@ -7,18 +8,19 @@ from langflow.services.task.utils import get_celery_worker_status
|
||||||
|
|
||||||
|
|
||||||
def check_celery_availability():
|
def check_celery_availability():
|
||||||
from langflow.worker import celery_app
|
|
||||||
|
|
||||||
try:
|
try:
|
||||||
|
from langflow.worker import celery_app
|
||||||
|
|
||||||
status = get_celery_worker_status(celery_app)
|
status = get_celery_worker_status(celery_app)
|
||||||
logger.debug(f"Celery status: {status}")
|
logger.debug(f"Celery status: {status}")
|
||||||
except Exception as e:
|
except Exception as exc:
|
||||||
logger.error(f"An error occurred: {e}")
|
logger.debug(f"Celery not available: {exc}")
|
||||||
status = {"availability": None}
|
status = {"availability": None}
|
||||||
return status
|
return status
|
||||||
|
|
||||||
|
|
||||||
try:
|
try:
|
||||||
|
configure()
|
||||||
status = check_celery_availability()
|
status = check_celery_availability()
|
||||||
|
|
||||||
USE_CELERY = status.get("availability") is not None
|
USE_CELERY = status.get("availability") is not None
|
||||||
|
|
|
||||||
|
|
@ -1,13 +1,44 @@
|
||||||
|
from math import log
|
||||||
from typing import Optional
|
from typing import Optional
|
||||||
from loguru import logger
|
from loguru import logger
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from rich.logging import RichHandler
|
from rich.logging import RichHandler
|
||||||
|
import os
|
||||||
|
import orjson
|
||||||
|
import appdirs
|
||||||
|
|
||||||
|
|
||||||
def configure(log_level: str = "DEBUG", log_file: Optional[Path] = None):
|
VALID_LOG_LEVELS = ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
|
||||||
log_format = "<green>{time:HH:mm:ss}</green> - <level>{level: <8}</level> - <level>{message}</level>"
|
|
||||||
|
|
||||||
|
def serialize(record):
|
||||||
|
subset = {
|
||||||
|
"timestamp": record["time"].timestamp(),
|
||||||
|
"message": record["message"],
|
||||||
|
"level": record["level"].name,
|
||||||
|
"module": record["module"],
|
||||||
|
}
|
||||||
|
return orjson.dumps(subset)
|
||||||
|
|
||||||
|
|
||||||
|
def patching(record):
|
||||||
|
record["extra"]["serialized"] = serialize(record)
|
||||||
|
|
||||||
|
|
||||||
|
def configure(log_level: Optional[str] = None, log_file: Optional[Path] = None):
|
||||||
|
if os.getenv("LANGFLOW_LOG_LEVEL") in VALID_LOG_LEVELS and log_level is None:
|
||||||
|
log_level = os.getenv("LANGFLOW_LOG_LEVEL")
|
||||||
|
if log_level is None:
|
||||||
|
log_level = "INFO"
|
||||||
|
# Human-readable
|
||||||
|
log_format = (
|
||||||
|
"<green>{time:YYYY-MM-DD HH:mm:ss}</green> - <level>"
|
||||||
|
"{level: <8}</level> - {module} - <level>{message}</level>"
|
||||||
|
)
|
||||||
|
|
||||||
|
# log_format = log_format_dev if log_level.upper() == "DEBUG" else log_format_prod
|
||||||
logger.remove() # Remove default handlers
|
logger.remove() # Remove default handlers
|
||||||
|
logger.patch(patching)
|
||||||
# Configure loguru to use RichHandler
|
# Configure loguru to use RichHandler
|
||||||
logger.configure(
|
logger.configure(
|
||||||
handlers=[
|
handlers=[
|
||||||
|
|
@ -19,17 +50,21 @@ def configure(log_level: str = "DEBUG", log_file: Optional[Path] = None):
|
||||||
]
|
]
|
||||||
)
|
)
|
||||||
|
|
||||||
if log_file:
|
if not log_file:
|
||||||
log_file = Path(log_file)
|
cache_dir = Path(appdirs.user_cache_dir("langflow"))
|
||||||
log_file.parent.mkdir(parents=True, exist_ok=True)
|
log_file = cache_dir / "langflow.log"
|
||||||
|
|
||||||
logger.add(
|
log_file = Path(log_file)
|
||||||
sink=str(log_file),
|
log_file.parent.mkdir(parents=True, exist_ok=True)
|
||||||
level=log_level.upper(),
|
|
||||||
format=log_format,
|
|
||||||
rotation="10 MB", # Log rotation based on file size
|
|
||||||
)
|
|
||||||
|
|
||||||
logger.info(f"Logger set up with log level: {log_level}")
|
logger.add(
|
||||||
|
sink=str(log_file),
|
||||||
|
level=log_level.upper(),
|
||||||
|
format=log_format,
|
||||||
|
rotation="10 MB", # Log rotation based on file size
|
||||||
|
serialize=True,
|
||||||
|
)
|
||||||
|
|
||||||
|
logger.debug(f"Logger set up with log level: {log_level}")
|
||||||
if log_file:
|
if log_file:
|
||||||
logger.info(f"Log file: {log_file}")
|
logger.info(f"Log file: {log_file}")
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue