如何获取已完成Google Cloud Tasks队列统计或配置空队列Webhook

一、Google Cloud CLI 统计方案 #

通过gcloud命令结合jq工具，直接筛选指定日期范围内的任务并统计指标，适合快速查询或脚本自动化。

命令示例 #

# 配置参数
QUEUE_NAME="你的队列名称"
LOCATION="你的队列所在区域（如us-central1）"
TARGET_DATE="2024-05-20" # 要查询的日期
START_TIME="${TARGET_DATE}T00:00:00Z"
END_TIME="${TARGET_DATE}T23:59:59Z"

# 获取任务数据并输出统计结果
gcloud tasks tasks list --queue=$QUEUE_NAME --location=$LOCATION \
--filter="createTime >= '$START_TIME' AND createTime <= '$END_TIME' AND (state=SUCCEEDED OR state=FAILED)" \
--format=json | jq '
{
  "已分发任务数": length,
  "成功任务数(200状态)": map(select(.responseStatus.code == 200)) | length,
  "失败任务数(重试耗尽)": map(select(.state == "FAILED")) | length,
  "队列开始时间戳": (min_by(.createTime).createTime),
  "队列完成时间戳": (max_by(.updateTime).updateTime)
}'

注意事项 #

• 需提前安装jq工具用于JSON数据处理
• 确保gcloud CLI已完成GCP账号授权
• 可根据实际重试逻辑调整失败任务的过滤条件

二、Node.js SDK 统计方案 #

使用官方Cloud Tasks SDK编写代码，适合集成到现有Node.js服务中实现自动化统计。

步骤1：安装依赖 #

npm install @google-cloud/tasks

步骤2：统计代码示例 #

const { CloudTasksClient } = require('@google-cloud/tasks');

async function getQueueStatsByDate(queueName, location, targetDate) {
  const client = new CloudTasksClient();
  const parent = client.queuePath(process.env.GCP_PROJECT, location, queueName);
  
  // 构造日期范围
  const startTime = new Date(targetDate);
  startTime.setHours(0, 0, 0, 0);
  const endTime = new Date(targetDate);
  endTime.setHours(23, 59, 59, 999);

  const tasks = [];
  let nextPageToken = null;

  // 分页获取所有符合条件的任务（避免数据量过大导致遗漏）
  do {
    const request = {
      parent,
      filter: `createTime >= "${startTime.toISOString()}" AND createTime <= "${endTime.toISOString()}" AND (state=SUCCEEDED OR state=FAILED)`,
      pageToken: nextPageToken,
    };

    const response = await client.listTasks(request);
    tasks.push(...response[0]);
    nextPageToken = response[1]?.nextPageToken;
  } while (nextPageToken);

  // 计算统计指标
  const totalDispatched = tasks.length;
  const successCount = tasks.filter(task => task.responseStatus?.code === 200).length;
  const failedCount = tasks.filter(task => task.state === 'FAILED').length;
  const queueStartTime = tasks.length ? new Date(Math.min(...tasks.map(t => new Date(t.createTime).getTime()))).toISOString() : null;
  const queueEndTime = tasks.length ? new Date(Math.max(...tasks.map(t => new Date(t.updateTime).getTime()))).toISOString() : null;

  return {
    已分发任务数: totalDispatched,
    成功任务数(200状态): successCount,
    失败任务数(重试耗尽): failedCount,
    队列开始时间戳: queueStartTime,
    队列完成时间戳: queueEndTime
  };
}

// 调用示例
getQueueStatsByDate('你的队列名称', 'us-central1', '2024-05-20')
  .then(stats => console.log(stats))
  .catch(err => console.error('统计失败:', err));

注意事项 #

• 需设置GCP_PROJECT环境变量指定你的GCP项目ID
• 若任务量极大，可添加异步分页处理逻辑避免内存溢出

三、Webhook触发日志统计方案 #

如果官方QueueStats无法满足需求，可通过Webhook将任务状态上报到数据库，再从数据库查询统计数据，适合长期存储和复杂统计场景。

步骤1：搭建Webhook服务 #

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

const app = express();
app.use(express.json());
const firestore = new Firestore();

// 接收任务状态上报的接口
app.post('/task-status-logger', async (req, res) => {
  const { taskId, statusCode, state, createTime, updateTime, batchDate } = req.body;

  // 将日志写入Firestore（也可使用BigQuery、MySQL等）
  await firestore.collection('task_execution_logs').doc(taskId).set({
    taskId,
    statusCode,
    state,
    createTime: new Date(createTime),
    updateTime: new Date(updateTime),
    batchDate // 标识任务所属的执行日期（如周一、三、五的日期）
  });

  res.status(200).send('日志已记录');
});

const PORT = process.env.PORT || 8080;
app.listen(PORT, () => {
  console.log(`Webhook服务运行在端口 ${PORT}`);
});

步骤2：创建任务时携带批次信息 #

在创建Cloud Tasks任务时，添加batchDate参数，方便后续按日期统计：

// 创建任务的代码片段
const task = {
  httpRequest: {
    httpMethod: 'POST',
    url: '你的任务处理接口URL',
    // 携带上报用的元数据
    body: Buffer.from(JSON.stringify({
      batchDate: new Date().toISOString().split('T')[0]
    })).toString('base64'),
    headers: {
      'Content-Type': 'application/json'
    }
  }
};

步骤3：从数据库查询统计数据 #

async function getStatsFromDb(batchDate) {
  const snapshot = await firestore.collection('task_execution_logs')
    .where('batchDate', '==', batchDate)
    .get();

  const logs = snapshot.docs.map(doc => doc.data());
  
  return {
    已分发任务数: logs.length,
    成功任务数(200状态): logs.filter(log => log.statusCode === 200).length,
    失败任务数(重试耗尽): logs.filter(log => log.state === 'FAILED').length,
    队列开始时间戳: logs.length ? new Date(Math.min(...logs.map(l => l.createTime.getTime()))).toISOString() : null,
    队列完成时间戳: logs.length ? new Date(Math.max(...logs.map(l => l.updateTime.getTime()))).toISOString() : null
  };
}

注意事项 #

• 需确保Webhook服务的可靠性，可添加重试机制避免日志丢失
• 数据库选择可根据业务需求调整，BigQuery适合大数据量的统计分析

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