Avoid pf and pfs version mismatch in Python scenario. (#2138)

# Description
- Start different pfs service for each python envionrment
- pfs service stop itself if no request comming in certain time: e.g. 1
hour.
- add pfs restart logic if version mismatch

![image](https://github.com/microsoft/promptflow/assets/26239730/7840e52b-08f3-4c85-a123-530afb13eb52)

![image](https://github.com/microsoft/promptflow/assets/26239730/3d200b81-c19b-4648-8f6a-497ea5a5ff59)

Goal
restart pfs if version mismatch
different port file for different python

Non goal in this pr
handle concurrent create in same env
monitor stop by waitress pid

Please add an informative description that covers that changes made by
the pull request and link all relevant issues.

# All Promptflow Contribution checklist:
- [ ] **The pull request does not introduce [breaking changes].**
- [ ] **CHANGELOG is updated for new features, bug fixes or other
significant changes.**
- [ ] **I have read the [contribution guidelines](../CONTRIBUTING.md).**
- [ ] **Create an issue and link to the pull request to get dedicated
review from promptflow team. Learn more: [suggested
workflow](../CONTRIBUTING.md#suggested-workflow).**

## General Guidelines and Best Practices
- [ ] Title of the pull request is clear and informative.
- [ ] There are a small number of commits, each of which have an
informative message. This means that previously merged commits do not
appear in the history of the PR. For more information on cleaning up the
commits in your PR, [see this
page](https://github.com/Azure/azure-powershell/blob/master/documentation/development-docs/cleaning-up-commits.md).

### Testing Guidelines
- [ ] Pull request includes test coverage for the included changes.

---------

Co-authored-by: Ying Chen <[email protected]>

Author: YingChen1996

src/promptflow/promptflow/_sdk/_constants.py

@@ -71,8 +71,11 @@ def _prepare_home_dir() -> Path:
 PROMPT_FLOW_RUNS_DIR_NAME = ".runs"
 PROMPT_FLOW_EXP_DIR_NAME = ".exps"
 SERVICE_CONFIG_FILE = "pf.yaml"
+PF_SERVICE_PORT_DIT_NAME = "pfs"
 PF_SERVICE_PORT_FILE = "pfs.port"
 PF_SERVICE_LOG_FILE = "pfs.log"
+PF_SERVICE_HOUR_TIMEOUT = 1
+PF_SERVICE_MONITOR_SECOND = 60
 PF_TRACE_CONTEXT = "PF_TRACE_CONTEXT"
 PF_SERVICE_DEBUG = "PF_SERVICE_DEBUG"
 

src/promptflow/promptflow/_sdk/_service/app.py

@@ -2,14 +2,16 @@
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # ---------------------------------------------------------
 import logging
+import sys
 import time
+from datetime import datetime, timedelta
 from logging.handlers import RotatingFileHandler
 
 from flask import Blueprint, Flask, g, jsonify, request
 from flask_cors import CORS
 from werkzeug.exceptions import HTTPException
 
-from promptflow._sdk._constants import HOME_PROMPT_FLOW_DIR, PF_SERVICE_LOG_FILE
+from promptflow._sdk._constants import PF_SERVICE_HOUR_TIMEOUT, PF_SERVICE_LOG_FILE, PF_SERVICE_MONITOR_SECOND
 from promptflow._sdk._service import Api
 from promptflow._sdk._service.apis.collector import trace_collector
 from promptflow._sdk._service.apis.connection import api as connection_api
@@ -18,8 +20,14 @@
 from promptflow._sdk._service.apis.span import api as span_api
 from promptflow._sdk._service.apis.telemetry import api as telemetry_api
 from promptflow._sdk._service.apis.ui import api as ui_api
-from promptflow._sdk._service.utils.utils import FormattedException
-from promptflow._sdk._utils import get_promptflow_sdk_version, overwrite_null_std_logger, read_write_by_user
+from promptflow._sdk._service.utils.utils import (
+    FormattedException,
+    get_current_env_pfs_file,
+    get_port_from_config,
+    kill_exist_service,
+)
+from promptflow._sdk._utils import get_promptflow_sdk_version, overwrite_null_std_logger
+from promptflow._utils.thread_utils import ThreadWithContextVars
 
 overwrite_null_std_logger()
 
@@ -57,13 +65,14 @@ def create_app():
 
         # Enable log
         app.logger.setLevel(logging.INFO)
-        log_file = HOME_PROMPT_FLOW_DIR / PF_SERVICE_LOG_FILE
-        log_file.touch(mode=read_write_by_user(), exist_ok=True)
+        # each env will have its own log file
+        log_file = get_current_env_pfs_file(PF_SERVICE_LOG_FILE)
         # Create a rotating file handler with a max size of 1 MB and keeping up to 1 backup files
         handler = RotatingFileHandler(filename=log_file, maxBytes=1_000_000, backupCount=1)
         formatter = logging.Formatter("[%(asctime)s][%(name)s][%(levelname)s] - %(message)s")
         handler.setFormatter(formatter)
-        app.logger.addHandler(handler)
+        # Set app logger to the only one RotatingFileHandler to avoid duplicate logs
+        app.logger.handlers = [handler]
 
         # Basic error handler
         @api.errorhandler(Exception)
@@ -82,9 +91,19 @@ def handle_exception(e):
 
         @app.before_request
         def log_before_request_info():
+            app.config["last_request_time"] = datetime.now()
             g.start = time.perf_counter()
-            app.logger.debug("Headers: %s", request.headers)
-            app.logger.debug("Body: %s", request.get_data())
+            if "/v1.0/Connections" in request.url:
+                request_body = "Request body not recorded for Connections API"
+            else:
+                request_body = request.get_data()
+
+            app.logger.debug(
+                "Last request time: %s, Headers: %s, Body: %s",
+                app.config["last_request_time"],
+                request.headers,
+                request_body,
+            )
 
         @app.after_request
         def log_after_request_info(response):
@@ -94,4 +113,25 @@ def log_after_request_info(response):
             )
             return response
 
+        # Start a monitor process using detach mode. It will stop pfs service if no request to pfs service in 1h in
+        # python scenario. For C# scenario, pfs will live until the process is killed manually.
+        def monitor_request():
+            while True:
+                time.sleep(PF_SERVICE_MONITOR_SECOND)
+                # For python scenario, since we start waitress in cli, there will be two app. The one used to log in
+                # the parent process will have no "last_request_time" in app.config since the app doesn't run.
+                app.logger.info(f"Promptflow service is running, version: {get_promptflow_sdk_version()}")
+                if "last_request_time" in app.config and datetime.now() - app.config["last_request_time"] > timedelta(
+                    hours=PF_SERVICE_HOUR_TIMEOUT
+                ):
+                    # Todo: check if we have any not complete work? like persist all traces.
+                    port = get_port_from_config()
+                    if port:
+                        app.logger.info(f"Try auto stop pfs service in port {port} since no request to app within 1h")
+                        kill_exist_service(port)
+                    break
+
+        if not sys.executable.endswith("pfcli.exe"):
+            monitor_thread = ThreadWithContextVars(target=monitor_request, daemon=True)
+            monitor_thread.start()
     return app, api

src/promptflow/promptflow/_sdk/_service/entry.py

@@ -22,10 +22,13 @@
     get_started_service_info,
     is_port_in_use,
     kill_exist_service,
+    kill_service_get_from_original_port_file,
 )
 from promptflow._sdk._utils import get_promptflow_sdk_version, print_pf_version
 from promptflow.exceptions import UserErrorException
 
+app, _ = create_app()
+
 
 def get_app(environ, start_response):
     app, _ = create_app()
@@ -80,13 +83,16 @@ def add_show_status_action(subparsers):
 
 def start_service(args):
     # User Agent will be set based on header in request, so not set globally here.
+    global app
     os.environ[PF_NO_INTERACTIVE_LOGIN] = "true"
     port = args.port
-    app, _ = create_app()
     if args.debug:
         os.environ[PF_SERVICE_DEBUG] = "true"
         app.logger.setLevel(logging.DEBUG)
 
+    # add this logic to stop pfs service which is start in the original port file.
+    kill_service_get_from_original_port_file()
+
     def validate_port(port, force_start):
         if is_port_in_use(port):
             if force_start:
@@ -111,14 +117,16 @@ def validate_port(port, force_start):
         )
         waitress.serve(app, host="127.0.0.1", port=port)
     else:
+        # Note: in this scenario, we will have two app, one is the parent process created to log, one is created in
+        # the detached child process.
         # Set host to localhost, only allow request from localhost.
         cmd = ["waitress-serve", f"--listen=127.0.0.1:{port}", "promptflow._sdk._service.entry:get_app"]
         # Start a pfs process using detach mode. It will start a new process and create a new app. So we use environment
         # variable to pass the debug mode, since it will inherit parent process environment variable.
         if platform.system() == "Windows":
-            subprocess.Popen(cmd, creationflags=subprocess.CREATE_NEW_PROCESS_GROUP)
+            subprocess.Popen(cmd, stdout=subprocess.DEVNULL, creationflags=subprocess.CREATE_NEW_PROCESS_GROUP)
         else:
-            subprocess.Popen(cmd, start_new_session=True)
+            subprocess.Popen(cmd, stdout=subprocess.DEVNULL, start_new_session=True)
         is_healthy = check_pfs_service_status(port)
         if is_healthy:
             app.logger.info(
@@ -129,7 +137,7 @@ def validate_port(port, force_start):
 
 
 def stop_service():
-    app, _ = create_app()
+    global app
     port = get_port_from_config()
     if port is not None:
         kill_exist_service(port)

src/promptflow/promptflow/_sdk/_service/utils/utils.py

@@ -2,7 +2,11 @@
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # ---------------------------------------------------------
 import getpass
+import hashlib
+import os
+import re
 import socket
+import sys
 import time
 from dataclasses import InitVar, dataclass, field
 from datetime import datetime
@@ -12,9 +16,14 @@
 import requests
 from flask import abort, make_response, request
 
-from promptflow._sdk._constants import DEFAULT_ENCODING, HOME_PROMPT_FLOW_DIR, PF_SERVICE_PORT_FILE
+from promptflow._sdk._constants import (
+    DEFAULT_ENCODING,
+    HOME_PROMPT_FLOW_DIR,
+    PF_SERVICE_PORT_DIT_NAME,
+    PF_SERVICE_PORT_FILE,
+)
 from promptflow._sdk._errors import ConnectionNotFoundError, RunNotFoundError
-from promptflow._sdk._utils import read_write_by_user
+from promptflow._sdk._utils import get_promptflow_sdk_version, read_write_by_user
 from promptflow._utils.logger_utils import get_cli_sdk_logger
 from promptflow._utils.yaml_utils import dump_yaml, load_yaml
 from promptflow._version import VERSION
@@ -35,13 +44,27 @@ def wrapper(*args, **kwargs):
     return wrapper
 
 
+def get_current_env_pfs_file(file_name):
+    executable_path = sys.executable
+    dir_name = os.path.basename(os.path.dirname(executable_path))
+    # Hash the executable path
+    hash_object = hashlib.sha1(executable_path.encode())
+    hex_dig = hash_object.hexdigest()
+    port_file_name = f"{dir_name}_{hex_dig}_{file_name}"
+    port_file_dir = HOME_PROMPT_FLOW_DIR / PF_SERVICE_PORT_DIT_NAME
+    port_file_dir.mkdir(parents=True, exist_ok=True)
+    port_file_path = port_file_dir / port_file_name
+    port_file_path.touch(mode=read_write_by_user(), exist_ok=True)
+    return port_file_path
+
+
 def get_port_from_config(create_if_not_exists=False):
-    (HOME_PROMPT_FLOW_DIR / PF_SERVICE_PORT_FILE).touch(mode=read_write_by_user(), exist_ok=True)
-    with open(HOME_PROMPT_FLOW_DIR / PF_SERVICE_PORT_FILE, "r", encoding=DEFAULT_ENCODING) as f:
+    port_file_path = get_current_env_pfs_file(PF_SERVICE_PORT_FILE)
+    with open(port_file_path, "r", encoding=DEFAULT_ENCODING) as f:
         service_config = load_yaml(f) or {}
         port = service_config.get("service", {}).get("port", None)
     if not port and create_if_not_exists:
-        with open(HOME_PROMPT_FLOW_DIR / PF_SERVICE_PORT_FILE, "w", encoding=DEFAULT_ENCODING) as f:
+        with open(port_file_path, "w", encoding=DEFAULT_ENCODING) as f:
             # Set random port to ~/.promptflow/pf.yaml
             port = get_random_port()
             service_config["service"] = service_config.get("service", {})
@@ -50,12 +73,25 @@ def get_port_from_config(create_if_not_exists=False):
     return port
 
 
+def kill_service_get_from_original_port_file():
+    if (HOME_PROMPT_FLOW_DIR / PF_SERVICE_PORT_FILE).exists():
+        with open(HOME_PROMPT_FLOW_DIR / PF_SERVICE_PORT_FILE, "r", encoding=DEFAULT_ENCODING) as f:
+            service_config = load_yaml(f) or {}
+            port = service_config.get("service", {}).get("port", None)
+            if port:
+                if is_port_in_use(port):
+                    logger.debug(f"Kill the deprecated port {port} got from service key in thr pfs.port file.")
+                    kill_exist_service(port)
+        # delete original .promptflow/pfs.port
+        (HOME_PROMPT_FLOW_DIR / PF_SERVICE_PORT_FILE).unlink()
+
+
 def dump_port_to_config(port):
-    # Set port to ~/.promptflow/pf.port, if already have a port in file , will overwrite it.
-    (HOME_PROMPT_FLOW_DIR / PF_SERVICE_PORT_FILE).touch(mode=read_write_by_user(), exist_ok=True)
-    with open(HOME_PROMPT_FLOW_DIR / PF_SERVICE_PORT_FILE, "r", encoding=DEFAULT_ENCODING) as f:
+    # Set port to ~/.promptflow/pfs/**_pf.port, if already have a port in file , will overwrite it.
+    port_file_path = get_current_env_pfs_file(PF_SERVICE_PORT_FILE)
+    with open(port_file_path, "r", encoding=DEFAULT_ENCODING) as f:
         service_config = load_yaml(f) or {}
-    with open(HOME_PROMPT_FLOW_DIR / PF_SERVICE_PORT_FILE, "w", encoding=DEFAULT_ENCODING) as f:
+    with open(port_file_path, "w", encoding=DEFAULT_ENCODING) as f:
         service_config["service"] = service_config.get("service", {})
         service_config["service"]["port"] = port
         dump_yaml(service_config, f)
@@ -106,12 +142,24 @@ def make_response_no_content():
 
 
 def is_pfs_service_healthy(pfs_port) -> bool:
-    """Check if pfs service is running."""
+    """Check if pfs service is running and pfs version matches pf version."""
     try:
         response = requests.get("http://localhost:{}/heartbeat".format(pfs_port))
         if response.status_code == 200:
             logger.debug(f"Promptflow service is already running on port {pfs_port}, {response.text}")
-            return True
+            match = re.search(r'"promptflow":"(.*?)"', response.text)
+            if match:
+                version = match.group(1)
+                is_healthy = version == get_promptflow_sdk_version()
+                if not is_healthy:
+                    logger.warning(
+                        f"Promptflow service is running on port {pfs_port}, but the version is not the same as "
+                        f"promptflow sdk version {get_promptflow_sdk_version()}. The service version is {version}."
+                    )
+            else:
+                is_healthy = False
+                logger.warning("/heartbeat response doesn't contain current pfs version.")
+            return is_healthy
     except Exception:  # pylint: disable=broad-except
         pass
     logger.warning(

src/promptflow/tests/sdk_pfs_test/e2etests/test_cli.py

@@ -30,10 +30,11 @@ def _test_start_service(self, port=None, force=False):
             command = f"{command} --force"
         start_pfs = subprocess.Popen(command, shell=True)
         # Wait for service to be started
-        sleep(5)
+        start_pfs.wait()
         assert self._is_service_healthy()
-        start_pfs.terminate()
-        start_pfs.wait(10)
+        stop_command = "pfs stop"
+        stop_pfs = subprocess.Popen(stop_command, shell=True)
+        stop_pfs.wait()
 
     def _is_service_healthy(self, port=None):
         port = port or get_port_from_config()
@@ -48,10 +49,11 @@ def test_start_service(self):
             random_port = get_random_port()
             self._test_start_service(port=random_port, force=True)
 
-            # Force start pfs
+            # start pfs
             start_pfs = subprocess.Popen("pfs start", shell=True)
             # Wait for service to be started
-            sleep(5)
+            start_pfs.wait()
+            assert self._is_service_healthy()
             self._test_start_service(force=True)
             # previous pfs is killed
             assert start_pfs.poll() is not None