GCS文件上传监控方案实现求助：大文件状态追踪与处理触发

核心工作流梳理 #

针对你的需求，结合GCS特性，我们可以通过Firestore维护状态+Cloud Functions处理事件+Cloud Scheduler定时触发的组合实现完整逻辑，精准对应你的需求点：

1. 支持SFTP网关或直接上传2-5GB大文件至GCS；
2. 通过监听临时文件创建事件感知上传开始，将文件名写入Firestore标记为in_progress；
3. 监听正式文件创建事件，将Firestore中对应文件的状态更新为uploaded；
4. 监听文件删除事件，从Firestore中移除对应文件的状态记录；
5. 用Cloud Scheduler每5分钟触发定时任务，查询Firestore中的状态；
6. 筛选出状态为uploaded且实际存在于GCS的文件，调用process_files函数处理。

组件选型与具体实现 #

1. Firestore：维护文件上传状态

创建Firestore集合file_upload_status，每个文档ID使用GCS对象的完整路径（格式：bucket-name/path/to/file），文档包含以下字段：

• status：枚举值in_progress/uploaded/processed，标记文件当前状态
• updated_at：状态更新时间戳

2. Cloud Functions：处理GCS事件

（1）监听临时文件创建，标记上传开始

大部分SFTP网关上传大文件时，会先创建带临时后缀（如.tmp/.part）的文件，上传完成后重命名为正式文件名。我们监听这类临时文件的创建事件，写入in_progress状态：

const { Firestore } = require('@google-cloud/firestore');
const firestore = new Firestore();

exports.onTempFileCreated = async (event) => {
  const file = event.data;
  if (!file) return console.log('无文件数据');

  const tempFilePath = file.name;
  // 替换临时后缀为正式文件名，根据你的SFTP网关配置调整正则
  const targetFilePath = tempFilePath.replace(/\.tmp$/, '');
  if (targetFilePath === tempFilePath) return;

  await firestore.collection('file_upload_status').doc(targetFilePath).set({
    status: 'in_progress',
    updated_at: new Date()
  });
};

（2）监听正式文件创建，标记上传完成

当正式文件上传完成时，更新Firestore中的状态为uploaded：

const { Firestore } = require('@google-cloud/firestore');
const firestore = new Firestore();

exports.onFileUploaded = async (event) => {
  const file = event.data;
  if (!file) return console.log('无文件数据');

  const filePath = file.name;
  await firestore.collection('file_upload_status').doc(filePath).set({
    status: 'uploaded',
    updated_at: new Date()
  }, { merge: true });
};

（3）监听文件删除，清理状态记录

当文件从GCS删除时，同步删除Firestore中的对应状态文档：

const { Firestore } = require('@google-cloud/firestore');
const firestore = new Firestore();

exports.onFileDeleted = async (event) => {
  const file = event.data;
  if (!file) return console.log('无文件数据');

  const filePath = file.name;
  await firestore.collection('file_upload_status').doc(filePath).delete();
};

3. Cloud Scheduler：定时触发文件处理

配置每5分钟触发一次Cloud Function，查询状态为uploaded的文件，验证文件存在后调用process_files：

const { Firestore } = require('@google-cloud/firestore');
const { Storage } = require('@google-cloud/storage');
const firestore = new Firestore();
const storage = new Storage();

exports.processUploadedFiles = async () => {
  const snapshot = await firestore.collection('file_upload_status')
    .where('status', '==', 'uploaded')
    .get();

  if (snapshot.empty) return console.log('无待处理文件');

  const filesToProcess = [];
  for (const doc of snapshot.docs) {
    const filePath = doc.id;
    const [bucketName, ...objectPathParts] = filePath.split('/');
    const objectName = objectPathParts.join('/');

    // 验证文件仍存在于GCS，避免已删除但状态未更新
    try {
      const [exists] = await storage.bucket(bucketName).file(objectName).exists();
      if (exists) filesToProcess.push(filePath);
      else await doc.ref.delete();
    } catch (err) {
      console.log(`文件 ${filePath} 不存在，清理状态`, err);
      await doc.ref.delete();
    }
  }

  if (filesToProcess.length > 0) {
    await process_files(filesToProcess);
    // 处理完成后标记为processed，或直接删除状态文档
    for (const filePath of filesToProcess) {
      await firestore.collection('file_upload_status').doc(filePath).update({
        status: 'processed',
        processed_at: new Date()
      });
    }
  }
};

// 自定义文件处理函数
async function process_files(filePaths) {
  // 替换为你的业务逻辑，比如批量转存、数据分析等
  console.log('处理文件:', filePaths);
}

4. 部署命令示例

# 部署临时文件监听函数
gcloud functions deploy onTempFileCreated \
  --runtime nodejs20 \
  --trigger-event google.storage.object.finalize \
  --trigger-resource YOUR_BUCKET_NAME

# 部署正式文件监听函数
gcloud functions deploy onFileUploaded \
  --runtime nodejs20 \
  --trigger-event google.storage.object.finalize \
  --trigger-resource YOUR_BUCKET_NAME

# 部署文件删除监听函数
gcloud functions deploy onFileDeleted \
  --runtime nodejs20 \
  --trigger-event google.storage.object.delete \
  --trigger-resource YOUR_BUCKET_NAME

# 创建定时任务，每5分钟触发处理函数
gcloud scheduler jobs create http processUploadedFilesJob \
  --schedule "*/5 * * * *" \
  --uri "https://REGION-YOUR_PROJECT_ID.cloudfunctions.net/processUploadedFiles" \
  --http-method POST \
  --oauth-service-account-email YOUR_PROJECT_ID@appspot.gserviceaccount.com

关键问题解答 #

你担心的GCS仅存在/不存在的特性不会影响方案可行性：

• 我们通过SFTP网关的临时文件机制，间接实现了“上传开始”的感知；
• 用Firestore作为中间状态存储，弥补了GCS没有上传中状态的不足；
• 定时任务中额外验证文件存在性，避免了状态与实际文件不一致的问题。

注意事项 #

• 临时文件后缀：不同SFTP网关的临时文件格式可能不同，需根据实际配置调整代码中的正则表达式；
• 幂等性：Cloud Functions可能重复触发，确保Firestore操作和process_files函数支持幂等（比如处理前检查是否已标记为processed）；
• 大文件处理：process_files函数建议采用流式处理，避免内存溢出。

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