feat: add SIGTERM handling and update typer dependency (#4548)
* Update `typer` dependency to version 0.13.0 in `pyproject.toml` * refactor: Simplify exception handling in the CLI * Enhance lifespan function with clean shutdown and logging improvements * Add graceful shutdown handling for SIGTERM and SIGINT signals - Introduce signal handlers to manage SIGTERM and SIGINT for graceful server shutdown. - Update exception handling to ensure processes terminate properly and log shutdown events. - Modify server run logic to support signal-based shutdowns, improving reliability. --------- Co-authored-by: Nadir J <31660040+NadirJ@users.noreply.github.com>
This commit is contained in:
parent
335b649093
commit
f8f9b7cace
4 changed files with 66 additions and 25 deletions
|
|
@ -1,6 +1,7 @@
|
||||||
import asyncio
|
import asyncio
|
||||||
import inspect
|
import inspect
|
||||||
import platform
|
import platform
|
||||||
|
import signal
|
||||||
import socket
|
import socket
|
||||||
import sys
|
import sys
|
||||||
import time
|
import time
|
||||||
|
|
@ -24,9 +25,7 @@ from sqlmodel import select
|
||||||
|
|
||||||
from langflow.logging.logger import configure, logger
|
from langflow.logging.logger import configure, logger
|
||||||
from langflow.main import setup_app
|
from langflow.main import setup_app
|
||||||
from langflow.services.database.models.folder.utils import (
|
from langflow.services.database.models.folder.utils import create_default_folder_if_it_doesnt_exist
|
||||||
create_default_folder_if_it_doesnt_exist,
|
|
||||||
)
|
|
||||||
from langflow.services.database.utils import session_getter
|
from langflow.services.database.utils import session_getter
|
||||||
from langflow.services.deps import async_session_scope, get_db_service, get_settings_service
|
from langflow.services.deps import async_session_scope, get_db_service, get_settings_service
|
||||||
from langflow.services.settings.constants import DEFAULT_SUPERUSER
|
from langflow.services.settings.constants import DEFAULT_SUPERUSER
|
||||||
|
|
@ -77,6 +76,13 @@ def set_var_for_macos_issue() -> None:
|
||||||
logger.debug("Set OBJC_DISABLE_INITIALIZE_FORK_SAFETY to YES to avoid error")
|
logger.debug("Set OBJC_DISABLE_INITIALIZE_FORK_SAFETY to YES to avoid error")
|
||||||
|
|
||||||
|
|
||||||
|
def handle_sigterm(signum, frame): # noqa: ARG001
|
||||||
|
"""Handle SIGTERM signal gracefully."""
|
||||||
|
logger.info("Received SIGTERM signal. Performing graceful shutdown...")
|
||||||
|
# Raise SystemExit to trigger graceful shutdown
|
||||||
|
sys.exit(0)
|
||||||
|
|
||||||
|
|
||||||
@app.command()
|
@app.command()
|
||||||
def run(
|
def run(
|
||||||
*,
|
*,
|
||||||
|
|
@ -150,6 +156,9 @@ def run(
|
||||||
),
|
),
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Run Langflow."""
|
"""Run Langflow."""
|
||||||
|
# Register SIGTERM handler
|
||||||
|
signal.signal(signal.SIGTERM, handle_sigterm)
|
||||||
|
|
||||||
if env_file:
|
if env_file:
|
||||||
load_dotenv(env_file, override=True)
|
load_dotenv(env_file, override=True)
|
||||||
|
|
||||||
|
|
@ -211,13 +220,20 @@ def run(
|
||||||
click.launch(f"http://{host}:{port}")
|
click.launch(f"http://{host}:{port}")
|
||||||
if process:
|
if process:
|
||||||
process.join()
|
process.join()
|
||||||
except KeyboardInterrupt:
|
except (KeyboardInterrupt, SystemExit) as e:
|
||||||
|
logger.info("Shutting down server...")
|
||||||
if process is not None:
|
if process is not None:
|
||||||
process.terminate()
|
process.terminate()
|
||||||
sys.exit(0)
|
process.join(timeout=15) # Wait up to 15 seconds for process to terminate
|
||||||
except Exception as e: # noqa: BLE001
|
if process.is_alive():
|
||||||
|
logger.warning("Process did not terminate gracefully, forcing...")
|
||||||
|
process.kill()
|
||||||
|
raise typer.Exit(0) from e
|
||||||
|
except Exception as e:
|
||||||
logger.exception(e)
|
logger.exception(e)
|
||||||
sys.exit(1)
|
if process is not None:
|
||||||
|
process.terminate()
|
||||||
|
raise typer.Exit(1) from e
|
||||||
|
|
||||||
|
|
||||||
def wait_for_server_ready(host, port) -> None:
|
def wait_for_server_ready(host, port) -> None:
|
||||||
|
|
@ -359,9 +375,6 @@ def print_banner(host: str, port: int) -> None:
|
||||||
def run_langflow(host, port, log_level, options, app) -> None:
|
def run_langflow(host, port, log_level, options, app) -> None:
|
||||||
"""Run Langflow server on localhost."""
|
"""Run Langflow server on localhost."""
|
||||||
if platform.system() == "Windows":
|
if platform.system() == "Windows":
|
||||||
# Run using uvicorn on MacOS and Windows
|
|
||||||
# Windows doesn't support gunicorn
|
|
||||||
# MacOS requires an env variable to be set to use gunicorn
|
|
||||||
import uvicorn
|
import uvicorn
|
||||||
|
|
||||||
uvicorn.run(
|
uvicorn.run(
|
||||||
|
|
@ -374,7 +387,28 @@ def run_langflow(host, port, log_level, options, app) -> None:
|
||||||
else:
|
else:
|
||||||
from langflow.server import LangflowApplication
|
from langflow.server import LangflowApplication
|
||||||
|
|
||||||
LangflowApplication(app, options).run()
|
server = LangflowApplication(app, options)
|
||||||
|
|
||||||
|
def graceful_shutdown(signum, frame): # noqa: ARG001
|
||||||
|
"""Gracefully shutdown the server when receiving SIGTERM."""
|
||||||
|
# Suppress click exceptions during shutdown
|
||||||
|
import click
|
||||||
|
|
||||||
|
click.echo = lambda *args, **kwargs: None # noqa: ARG005
|
||||||
|
|
||||||
|
logger.info("Gracefully shutting down server...")
|
||||||
|
# For Gunicorn workers, we raise SystemExit to trigger graceful shutdown
|
||||||
|
raise SystemExit(0)
|
||||||
|
|
||||||
|
# Register signal handlers
|
||||||
|
signal.signal(signal.SIGTERM, graceful_shutdown)
|
||||||
|
signal.signal(signal.SIGINT, graceful_shutdown)
|
||||||
|
|
||||||
|
try:
|
||||||
|
server.run()
|
||||||
|
except (KeyboardInterrupt, SystemExit):
|
||||||
|
# Suppress the exception output
|
||||||
|
sys.exit(0)
|
||||||
|
|
||||||
|
|
||||||
@app.command()
|
@app.command()
|
||||||
|
|
@ -499,10 +533,7 @@ def api_key(
|
||||||
)
|
)
|
||||||
return None
|
return None
|
||||||
from langflow.services.database.models.api_key import ApiKey, ApiKeyCreate
|
from langflow.services.database.models.api_key import ApiKey, ApiKeyCreate
|
||||||
from langflow.services.database.models.api_key.crud import (
|
from langflow.services.database.models.api_key.crud import create_api_key, delete_api_key
|
||||||
create_api_key,
|
|
||||||
delete_api_key,
|
|
||||||
)
|
|
||||||
|
|
||||||
api_key = (await session.exec(select(ApiKey).where(ApiKey.user_id == superuser.id))).first()
|
api_key = (await session.exec(select(ApiKey).where(ApiKey.user_id == superuser.id))).first()
|
||||||
if api_key:
|
if api_key:
|
||||||
|
|
@ -568,4 +599,8 @@ def main() -> None:
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
main()
|
try:
|
||||||
|
main()
|
||||||
|
except Exception as e:
|
||||||
|
logger.exception(e)
|
||||||
|
raise typer.Exit(1) from e
|
||||||
|
|
|
||||||
|
|
@ -97,6 +97,7 @@ def get_lifespan(*, fix_migration=False, version=None):
|
||||||
@asynccontextmanager
|
@asynccontextmanager
|
||||||
async def lifespan(_app: FastAPI):
|
async def lifespan(_app: FastAPI):
|
||||||
configure(async_file=True)
|
configure(async_file=True)
|
||||||
|
|
||||||
# Startup message
|
# Startup message
|
||||||
if version:
|
if version:
|
||||||
rprint(f"[bold green]Starting Langflow v{version}...[/bold green]")
|
rprint(f"[bold green]Starting Langflow v{version}...[/bold green]")
|
||||||
|
|
@ -108,15 +109,20 @@ def get_lifespan(*, fix_migration=False, version=None):
|
||||||
await asyncio.to_thread(create_or_update_starter_projects, all_types_dict)
|
await asyncio.to_thread(create_or_update_starter_projects, all_types_dict)
|
||||||
telemetry_service.start()
|
telemetry_service.start()
|
||||||
await asyncio.to_thread(load_flows_from_directory)
|
await asyncio.to_thread(load_flows_from_directory)
|
||||||
|
|
||||||
yield
|
yield
|
||||||
|
|
||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
if "langflow migration --fix" not in str(exc):
|
if "langflow migration --fix" not in str(exc):
|
||||||
logger.exception(exc)
|
logger.exception(exc)
|
||||||
raise
|
raise
|
||||||
# Shutdown message
|
finally:
|
||||||
rprint("[bold red]Shutting down Langflow...[/bold red]")
|
# Clean shutdown
|
||||||
await teardown_services()
|
logger.info("Cleaning up resources...")
|
||||||
await logger.complete()
|
await teardown_services()
|
||||||
|
await logger.complete()
|
||||||
|
# Final message
|
||||||
|
rprint("[bold red]Langflow shutdown complete[/bold red]")
|
||||||
|
|
||||||
return lifespan
|
return lifespan
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -119,7 +119,7 @@ dependencies = [
|
||||||
"langchain-experimental>=0.0.61",
|
"langchain-experimental>=0.0.61",
|
||||||
"pydantic>=2.7.0",
|
"pydantic>=2.7.0",
|
||||||
"pydantic-settings>=2.2.0",
|
"pydantic-settings>=2.2.0",
|
||||||
"typer>=0.12.0",
|
"typer>=0.13.0",
|
||||||
"cachetools>=5.3.1",
|
"cachetools>=5.3.1",
|
||||||
"platformdirs>=4.2.0",
|
"platformdirs>=4.2.0",
|
||||||
"python-multipart>=0.0.12",
|
"python-multipart>=0.0.12",
|
||||||
|
|
|
||||||
8
uv.lock
generated
8
uv.lock
generated
|
|
@ -4082,7 +4082,7 @@ requires-dist = [
|
||||||
{ name = "setuptools", specifier = ">=70" },
|
{ name = "setuptools", specifier = ">=70" },
|
||||||
{ name = "spider-client", specifier = ">=0.0.27" },
|
{ name = "spider-client", specifier = ">=0.0.27" },
|
||||||
{ name = "sqlmodel", specifier = "==0.0.18" },
|
{ name = "sqlmodel", specifier = "==0.0.18" },
|
||||||
{ name = "typer", specifier = ">=0.12.0" },
|
{ name = "typer", specifier = ">=0.13.0" },
|
||||||
{ name = "types-google-cloud-ndb", marker = "extra == 'dev'", specifier = ">=2.2.0.0" },
|
{ name = "types-google-cloud-ndb", marker = "extra == 'dev'", specifier = ">=2.2.0.0" },
|
||||||
{ name = "types-markdown", marker = "extra == 'dev'", specifier = ">=3.7.0.20240822" },
|
{ name = "types-markdown", marker = "extra == 'dev'", specifier = ">=3.7.0.20240822" },
|
||||||
{ name = "types-passlib", marker = "extra == 'dev'", specifier = ">=1.7.7.13" },
|
{ name = "types-passlib", marker = "extra == 'dev'", specifier = ">=1.7.7.13" },
|
||||||
|
|
@ -8060,7 +8060,7 @@ wheels = [
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "typer"
|
name = "typer"
|
||||||
version = "0.12.5"
|
version = "0.13.0"
|
||||||
source = { registry = "https://pypi.org/simple" }
|
source = { registry = "https://pypi.org/simple" }
|
||||||
dependencies = [
|
dependencies = [
|
||||||
{ name = "click" },
|
{ name = "click" },
|
||||||
|
|
@ -8068,9 +8068,9 @@ dependencies = [
|
||||||
{ name = "shellingham" },
|
{ name = "shellingham" },
|
||||||
{ name = "typing-extensions" },
|
{ name = "typing-extensions" },
|
||||||
]
|
]
|
||||||
sdist = { url = "https://files.pythonhosted.org/packages/c5/58/a79003b91ac2c6890fc5d90145c662fd5771c6f11447f116b63300436bc9/typer-0.12.5.tar.gz", hash = "sha256:f592f089bedcc8ec1b974125d64851029c3b1af145f04aca64d69410f0c9b722", size = 98953 }
|
sdist = { url = "https://files.pythonhosted.org/packages/e7/87/9eb07fdfa14e22ec7658b5b1147836d22df3848a22c85a4e18ed272303a5/typer-0.13.0.tar.gz", hash = "sha256:f1c7198347939361eec90139ffa0fd8b3df3a2259d5852a0f7400e476d95985c", size = 97572 }
|
||||||
wheels = [
|
wheels = [
|
||||||
{ url = "https://files.pythonhosted.org/packages/a8/2b/886d13e742e514f704c33c4caa7df0f3b89e5a25ef8db02aa9ca3d9535d5/typer-0.12.5-py3-none-any.whl", hash = "sha256:62fe4e471711b147e3365034133904df3e235698399bc4de2b36c8579298d52b", size = 47288 },
|
{ url = "https://files.pythonhosted.org/packages/18/7e/c8bfa8cbcd3ea1d25d2beb359b5c5a3f4339a7e2e5d9e3ef3e29ba3ab3b9/typer-0.13.0-py3-none-any.whl", hash = "sha256:d85fe0b777b2517cc99c8055ed735452f2659cd45e451507c76f48ce5c1d00e2", size = 44194 },
|
||||||
]
|
]
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue