Azure Data Factory调用Function App报BadRequest错误排查求助

报错信息 #

错误代码：3608 - 调用Azure函数'CanStart?id=66&forceRun=false'时，在'https://functionapptest.azurewebsites.net'上执行GET请求返回BadRequest状态，错误消息为'{"status": "ERROR", "message": "'https://cosmosdb_dev.documents.azure.com:443/'", "version": "", "id": "66"}'。

场景说明 #

测试Azure Data Factory（ADF）调用Azure Function App从CosmosDB获取结果的流程时，Function App活动失败。ADF已配置参数：(1) forceRun (2) TriggerDateTime。已尝试重启Azure Function App，且CosmosDB设置为公网访问、未配置防火墙，但问题仍未解决。

相关代码 #

init.py #

import os
import sys
import json
import logging


from datetime import datetime, timedelta

from dateutil.parser import parse
import azure.functions as func
import pytz

from shared.keyvault import get_config
from shared.utils import get_datetime_with_offset, get_run_time

from shared.bottler import get_bottler_config
from shared.stage_status import get_bottler_stage_status
from shared.version import VERSION

sys.path.insert(0, os.path.abspath(".."))



def main(req: func.HttpRequest) -> func.HttpResponse:
    logging.info("Python HTTP trigger function processed a request.")

    bottler = str(req.params.get("bottler", None))
    force_run = req.params.get("forceRun", None)

    if req.method == "POST" and (bottler is None or force_run is None):
        bottler = str(req.get_json().get("bottler", None))
        force_run = req.get_json().get("forceRun", None)

    if force_run == "true":
        return func.HttpResponse(json.dumps({"status": "OK", "version": VERSION}),
                                 mimetype="application/json",
                                 headers={"Content-type": "application/json"})

    if not bottler:
        return func.HttpResponse(
            json.dumps(
                {"status": "ERROR", "message": "Please set bottler to check", "version": VERSION,
                 "bottler": bottler}),
            status_code=400,
            mimetype="application/json",
            headers={"Content-type": "application/json"}
        )

    try:
        config = get_config()
        bconfig = get_bottler_config(bottler)
        logging.info("Got bottler [%s] configs.", bottler)

        logging.info("Fetching bottler[%s] statuses.", bottler)
        statuses = get_bottler_stage_status(bottler)
        logging.info("Finish fetching bottler[%s] statuses.", bottler)
        run_time = get_run_time(bconfig, config)

        return func.HttpResponse(
            json.dumps({"status": "OK" if is_time_to_run(statuses, bconfig, run_time) else "NOK", "version": VERSION})
        )
    except Exception as my_e:
        return func.HttpResponse(
            json.dumps(
                {"status": "ERROR", "message": str(my_e), "version": VERSION, "bottler": bottler}),
            status_code=400,
            mimetype="application/json"
        )


def is_time_to_run(statuses: list, bottler_config: dict, run_time: datetime) -> bool:
    """
    Seek bottler config in CosmoDB and match scheduled run time to current time. If current time is in valid range it
    return True. Otherwise return False
    :param statuses:
    :param bottler_config:
    :param run_time
    :return: boolean
    """
    bottler = bottler_config["entityid"]

    if not statuses:
        logging.info("No statuses for bottler[%s]", bottler)
        return True
    else:
        logging.info("Fetched %s statuses for bottler[%s]", len(statuses), bottler)

        statuses = filter_today_statuses(statuses, bottler_config, run_time)

        if not statuses:
            logging.info("No today statuses for bottler[%s]", bottler)
            return check_time_to_run(bottler_config)

        ingest_done_statuses = filter_statuses(statuses, "INGEST", "DONE")
        ingest_in_progress_statuses = filter_statuses(statuses, "INGEST", "IN_PROGRESS")
        is_dispatcher_done = check_last_stage_status(statuses, "Dispatcher", "DONE")

        if not (ingest_done_statuses or ingest_in_progress_statuses) and is_dispatcher_done:
            logging.info("Last bottler[%s] status is: %s", bottler, statuses[-1])
            if statuses[-1]["status"] != "IN_PROGRESS":
                return check_time_to_run(bottler_config)
    return False


def check_last_stage_status(statuses: list, stage: str, req_status: str):
    if statuses is None:
        statuses = []
    statuses = [
        s for s in statuses
        if s["stage"] == stage
    ]
    if not statuses:
        return True
    return statuses[-1]["status"] == req_status


def filter_statuses(statuses: list, stage: str, status: str) -> list:
    statuses = [
        s for s in statuses
        if s["status"] == status and s["stage"] == stage
    ]
    logging.info("Found statuses[%s, %s] : %s", stage, status, statuses)
    return statuses


def filter_today_statuses(statuses: list, bottler_config: dict, run_time: datetime) -> list:
    logging.info("Current time: %s", run_time)
    statuses = [
        s for s in statuses
        if get_datetime_with_offset(bottler_config["timezone"], parse(s["timestamp"])) >= run_time
    ]
    logging.info("Found today statuses: %s", statuses)
    return statuses


def check_time_to_run(bottler_config: dict):
    timezone = pytz.timezone(bottler_config["timezone"])
    offset = timezone.utcoffset(datetime.utcnow())
    now = datetime.utcnow() + offset
    today = datetime(year=now.year, month=now.month, day=now.day)

    if bottler_config.get("processingStartTimeDayTotalMinutes", None) is None:
        bottler_config["processingStartTimeDayTotalMinutes"] = 0

    if bottler_config.get("processingEndTimeDayTotalMinutes", None) is None:
        bottler_config["processingEndTimeDayTotalMinutes"] = 0

    start_time = today + timedelta(minutes=bottler_config["processingStartTimeDayTotalMinutes"])
    end_time = today + timedelta(minutes=bottler_config["processingEndTimeDayTotalMinutes"])

    if start_time.timestamp() <= now.timestamp() <= end_time.timestamp():
        return True
    return False

function.json #

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "authLevel": "anonymous",
      "type": "httpTrigger",
      "direction": "in",
      "name": "req",
      "methods": [
        "get",
        "post"
      ]
    },
    {
      "type": "http",
      "direction": "out",
      "name": "$return"
    }
  ]
}

其他可能的错误原因 #

• CosmosDB连接字符串配置错误：函数中get_config()或get_bottler_stage_status()从Key Vault加载的CosmosDB连接字符串可能不完整、格式错误，或缺少主键、数据库/容器名称等必要参数，导致连接时抛出包含CosmosDB端点的异常。
• Key Vault访问权限不足：Function App的托管标识没有权限读取Key Vault中存储的CosmosDB配置项，导致获取连接信息失败，触发相关异常。
• CosmosDB资源不存在：代码尝试访问的CosmosDB数据库或容器不存在，或名称拼写错误，引发连接失败的异常。
• 函数依赖包问题：shared模块中的CosmosDB相关依赖（如azure-cosmos包）版本不兼容或未正确安装，导致调用CosmosDB API时出错。
• 请求参数异常：ADF调用的GET请求中bottler参数值无效，导致加载对应配置时出错，进而触发CosmosDB相关异常。
• 运行时环境不兼容：Function App的Python运行时版本与代码依赖不匹配，或环境变量配置缺失，影响CosmosDB客户端的正常初始化。

内容的提问来源于stack exchange，提问作者iamrhass