动态挂载文件存储NAS

核心能力

函数计算通过会话动态挂载 NAS功能，允许您在创建会话（Session）时，按需为该会话挂载专属的 NAS 目录，并指定独立的用户身份（POSIX UID/GID）。其核心特性包括：

• 动态与按需：存储的挂载与会话生命周期绑定，仅在需要时分配，会话销毁后自动解绑。

• 强隔离性：通过为每个会话指定独立的 User ID/Group ID，从文件系统层面彻底隔离不同租户的数据访问权限。

• 高灵活性：可为不同租户挂载到 NAS 文件系统的不同子目录，轻松实现数据分区。

• 性能优化：结合会话预热，可在实例启动时预先挂载好数据目录，加速模型或数据的加载。

工作原理

该功能将 NAS 挂载操作与 CreateSession API 调用深度集成。以 HeaderField 亲和为例，整体流程如下：

步骤一：配置函数（前置设定）

为使函数支持会话动态挂载NAS，需要在创建函数或更新函数配置时，完成三项关键设置：

1. 开启会话亲和：

   1. 登录函数计算控制台，点击函数管理>函数列表

   2. 在函数列表页，点击需要配置的函数名称，进入函数详情页

   3. 在配置页，找到高级配置，点击

   4. 打开会话亲和开关，选择HeaderField 亲和并配置Header Name，例如：x-affinity-header-v1

      说明

      不能以 x-fc- 前缀开头，以字母开头，非首字符可包含数字、中划线、下划线、字母，长度大于等于5个字符并且不超过40个字符。

   5. 点击部署完成配置更新。

2. 开启实例隔离：开启配置>高级配置>实例隔离，并选择会话隔离；

3. 允许访问 VPC配置

   • 开启配置>高级配置>网络>允许访问 VPC；

   • 配置方式选择自定义配置；

   • 专有网络选择挂载点所在的 VPC。

步骤二：创建带 NAS 配置的会话

1. 安装依赖

   macOS / Linux

   # 使用 pip3 安装
   pip3 install alibabacloud_fc20230330 alibabacloud_credentials alibabacloud_tea_openapi alibabacloud_tea_util
   
   # 如遇权限问题，使用 --user 参数
   pip3 install --user alibabacloud_fc20230330 alibabacloud_credentials alibabacloud_tea_openapi alibabacloud_tea_util
   
   # macOS Homebrew Python 环境需使用 --break-system-packages
   pip3 install --break-system-packages alibabacloud_fc20230330 alibabacloud_credentials alibabacloud_tea_openapi alibabacloud_tea_util

   Windows

   # 使用 pip 安装
   pip install alibabacloud_fc20230330 alibabacloud_credentials alibabacloud_tea_openapi alibabacloud_tea_util
   
   # 或使用 Python 3 指定
   py -3 -m pip install alibabacloud_fc20230330 alibabacloud_credentials alibabacloud_tea_openapi alibabacloud_tea_util

2. 编写创建会话的代码

   创建python代码文件（如：creatSession.py），将下列代码复制到文件中并替换核心参数。示例演示了如何为“租户 A”创建一个会话，并将其专属的 NAS 目录 (<YOUR-NAS-SERVER-ADDR>:/tenant-a-data) 挂载到实例的 /mnt/data 路径，同时指定用户身份为 UID=1001 和 GID=1001。

   核心方法及核心参数说明：

   重要

   会话动态挂载NAS和在函数配置>高级配置>存储中挂载NAS可以同时配置，但是需要注意：

   • 下文NASConfig 中定义的 User ID/Group ID必须与函数挂载配置中使用的用户（User ID）/用户组（Group ID）保持一致

   • 同一个挂载路径 (mount_dir) 不能同时用于会话动态挂载和函数挂载

   • config.endpoint：

     • <账号ID>：替换为阿里云账号ID；

     • <Endpoint>：请参考函数计算3.0 服务区域列表，规则为fcv3.[region_id].aliyuns.com；

   • CreateSessionInput：

     • session_ttlin_seconds：会话总生命周期（单位：秒）；

     • session_idle_timeout_in_seconds：会话闲置过期时间（单位：秒）；

   • client.create_session_with_options：将<函数名称>替换为创建Session的函数名称；

     • NASMountConfig：NAS 挂载配置

       • mount_dir：实例内的挂载路径，如：/home/test

       • server_addr：NAS 文件系统地址及租户专属子目录

     • user_id：为此会话指定独立的 POSIX User ID

     • group_id：为此会话指定独立的 POSIX Group ID

   # -*- coding: utf-8 -*-
   from alibabacloud_fc20230330.client import Client as FC20230330Client
   from alibabacloud_credentials.client import Client as CredentialClient
   from alibabacloud_tea_openapi import models as open_api_models
   from alibabacloud_fc20230330 import models as fc20230330_models
   from alibabacloud_tea_util import models as util_models
   
   # 1. 创建账号Client
   credential = CredentialClient()
   config = open_api_models.Config(credential=credential)
   config.endpoint = f'<账号ID>.<Endpoint>'
   client = FC20230330Client(config)
   
   # 2. 准备 NAS 挂载配置
   nas_mount_config = fc20230330_models.NASMountConfig(
       mount_dir='/mnt/data',  # 实例内的挂载路径
       server_addr='<YOUR-NAS-SERVER-ADDR>:/<tenant-a-path>'  # NAS 文件系统地址及租户专属子目录
   )
   
   # 3. 配置 NAS 和用户身份（为租户A分配独立UID/GID）
   nas_config = fc20230330_models.NASConfig(
       mount_points=[nas_mount_config],
       user_id=1001,  # 为此会话指定独立的 POSIX User ID
       group_id=1001  # 为此会话指定独立的 POSIX Group ID
   )
   
   # 4. 构造 CreateSession 请求
   create_session_input = fc20230330_models.CreateSessionInput(
       nas_config=nas_config,
       session_ttlin_seconds=3600,
       session_idle_timeout_in_seconds=600
   )
   create_session_request = fc20230330_models.CreateSessionRequest(
       body=create_session_input
   )
   
   # 5. 发起请求
   runtime = util_models.RuntimeOptions()
   response = client.create_session_with_options('<函数名称>', create_session_request, {}, runtime)
   
   # 6. 从响应中获取 sessionId
   print(response.body.to_map())
   session_id = response.body.session_id
   print(f"Session created successfully. Session ID: {session_id}")
   

3. 运行代码

   export ALIBABA_CLOUD_ACCESS_KEY_ID=LTAI****************
   export ALIBABA_CLOUD_ACCESS_KEY_SECRET=<yourAccessKeySecret>
   python3 creatSession.py

   参数说明：

   • ALIBABA_CLOUD_ACCESS_KEY_ID：阿里云账号或 RAM 用户的AccessKey ID；

   • ALIBABA_CLOUD_ACCESS_KEY_SECRET：阿里云账号或 RAM 用户的AccessKey Secret。

4. 返回结果示例

   {
     'containerId': 'c-********-********-************', 
     'createdTime': '2025-10-30T06:38:10Z', 
     'functionName': '****', 
     'lastModifiedTime': '2025-10-30T06:38:10Z', 
     'nasConfig': 
       {
         'groupId': 1001, 
         'mountPoints': [
           {
             'enableTLS': False, 
             'mountDir': '/home/test', 
             'serverAddr': '*-*.*.nas.aliyuncs.com:/test'
              }
                ], 
                'userId': 1001
                }, 
      'qualifier': 'LATEST', 
      'sessionAffinityType': 'HEADER_FIELD', 
      'sessionId': '******************', 
      'sessionIdleTimeoutInSeconds': 600, 
      'sessionStatus': 'Active', 
      'sessionTTLInSeconds': 3600
      }
   Session created successfully. Session ID: ************

步骤三：在函数中使用挂载的 NAS

现在，调用函数时携带上一步获取的 SessionID，函数实例内部即可访问到已挂载的 /mnt/data 目录。下面以Web函数为例：

1. 修改函数内代码并重新部署

   import os
   from flask import Flask, request, jsonify
   
   # ... (app setup) ...
   
   # 假设 NAS 挂载点在 /mnt/data
   NAS_MOUNT_PATH = '/mnt/data'
   app = Flask(__name__)
   
   
   @app.route('/<path:path>', methods=['GET', 'POST'])
   def handle_nas_request(path):
       rid = request.headers.get('x-fc-request-id')
       print(f"FC Invoke Start RequestId: {rid}")
   
       # 构造文件在 NAS 上的完整路径
       # 注意：需要防止路径穿越攻击
       safe_path = os.path.normpath(os.path.join(NAS_MOUNT_PATH, path))
       if not safe_path.startswith(NAS_MOUNT_PATH):
           return "Path traversal attempt detected!", 400
   
       response_data = {}
   
       if request.method == 'POST':
           # 写入文件
           body = request.data.decode('utf-8')
           try:
               with open(safe_path, 'w') as f:
                   f.write(body)
               response_data['message'] = f"Successfully wrote to {safe_path}"
               print(f"Wrote to {safe_path}")
           except Exception as e:
               return str(e), 500
   
       elif request.method == 'GET':
           # 读取文件
           try:
               if os.path.exists(safe_path) and os.path.isfile(safe_path):
                   with open(safe_path, 'r') as f:
                       content = f.read()
                   response_data['content'] = content
                   print(f"Read from {safe_path}")
               else:
                   return f"File not found: {safe_path}", 404
           except Exception as e:
               return str(e), 500
   
       print(f"FC Invoke End RequestId: {rid}")
       return jsonify(response_data)  # 使用 jsonify 返回 JSON 格式的响应
   
   # ... (if __name__ == '__main__': block) ...
   

2. 调用函数并验证

   调用API InvokeFunction- 调用函数，使用会话调用函数。

   核心代码示例及说明：

   InvokeFunctionHeaders：构造请求头，在请求Header中携带上一步返回的sessionId，Header Key的值与开启会话亲和设置的值保持一致（如：x-affinity-header-v1），实现会话绑定路由。

   # -*- coding: utf-8 -*-
   from alibabacloud_fc20230330.client import Client as FC20230330Client
   from alibabacloud_credentials.client import Client as CredentialClient
   from alibabacloud_tea_openapi import models as open_api_models
   from alibabacloud_fc20230330 import models as fc20230330_models
   from alibabacloud_tea_util import models as util_models
   
   session_id = '************'
   function_name = 'my-session-nas'
   # 1. 创建账号Client
   credential = CredentialClient()
   config = open_api_models.Config(credential=credential)
   config.endpoint = f'<账号ID>.<Endpoint>'
   client = FC20230330Client(config)
   
   # 2. 构造请求头。Header Key（x-affinity-header-v1）必须与函数配置的会话亲和 Key 一致。
   headers = fc20230330_models.InvokeFunctionHeaders(
       common_headers={
           "x-affinity-header-v1": session_id
       }
   )
   
   # 3. 构造调用请求（可根据需要传入 body）
   invoke_request = fc20230330_models.InvokeFunctionRequest(
       body='your_request_payload'.encode('utf-8')  # 示例 payload
   )
   
   runtime = util_models.RuntimeOptions()
   
   try:
       # 4. 发起调用
       invoke_response = client.invoke_function_with_options(
           function_name,
           invoke_request,
           headers,
           runtime
       )
       # 5. 处理响应
       print(f"Status Code: {invoke_response.status_code}")
       print(f"Response Body: {invoke_response.body.decode('utf-8')}")
   except Exception as error:
       print(error.message)