在 BigQuery 中写入的云函数（异步函数 ... 等待 bigquery ...）因未处理 rejection/PartialFailureError 而失败？
cloud function to write in BigQuery (async function ... await bigquery ...) failing with Unhandled rejection/PartialFailureError?
在 GCP，我创建了一个 CloudFunction，它由来自 Pub/Sub 的计费事件触发，并在 Slack 上发布了一些消息。我正在使用 node.js 10。我有以下依赖项：
{
"name": "google-container-slack",
"version": "0.0.1",
"description": "Slack integration for Google Cloud Build, using Google Cloud Functions",
"main": "index.js",
"dependencies": {
"@slack/webhook": "5.0.3",
"@google-cloud/bigquery": "4.7.0",
"@google-cloud/pubsub": "1.6.0"
}
}
当我在 BigQuery 中添加一个函数来写入新的预算信息时，我遇到了一些问题，这是基于一些官方示例：
https://github.com/googleapis/nodejs-bigquery/blob/master/samples/insertRowsAsStream.js
// writeInBigQuery update BigQuery table
async function writeInBigQuery(pubsubdata, createdAt, project, threshold) {
const bigquery = new BigQuery({projectId: billing_project});
const rows = [{createdAt: createdAt},
{budgetAmount:pubsubdata.budgetAmount},
{projectName: project},
{thresholdValue: threshold}];
console.log(rows);
console.log('start insert row in bigquery');
await bigquery
.dataset(dataset)
.table(table)
.insert(rows);
console.log('end insert row in bigquery');
console.log(`Inserted ${rows.length} rows`);
}
我猜这个问题与异步和等待有关。这是我的第一个 node.js 代码，错误消息对我来说很神秘：
Function execution took 266 ms, finished with status: 'ok'
Unhandled rejection
PartialFailureError
at request (/layers/google.nodejs.npm/npm/node_modules/@google-cloud/bigquery/build/src/table.js:1550:23)
at util.makeRequest.params (/layers/google.nodejs.npm/npm/node_modules/@google-cloud/common/build/src/util.js:367:25)
8 at Util.handleResp (/layers/google.nodejs.npm/npm/node_modules/@google-cloud/common/build/src/util.js:144:9)
8 at retryRequest (/layers/google.nodejs.npm/npm/node_modules/@google-cloud/common/build/src/util.js:432:22)
at onResponse (/layers/google.nodejs.npm/npm/node_modules/retry-request/index.js:206:7)
8 at /layers/google.nodejs.npm/npm/node_modules/teeny-request/build/src/index.js:233:13
at process._tickCallback (internal/process/next_tick.js:68:7)
8 Error: Process exited with code 16
at process.on.code (/layers/google.nodejs.functions-framework/functions-framework/node_modules/@google-cloud/functions-framework/build/src/invoker.js:393:29)
at process.emit (events.js:198:13)
at process.EventEmitter.emit (domain.js:448:20)
at process.exit (internal/process/per_thread.js:168:15)
at logAndSendError (/layers/google.nodejs.functions-framework/functions-framework/node_modules/@google-cloud/functions-framework/build/src/invoker.js:184:9)
at process.on.err (/layers/google.nodejs.functions-framework/functions-framework/node_modules/@google-cloud/functions-framework/build/src/invoker.js:390:13)
at process.emit (events.js:198:13)
at process.EventEmitter.emit (domain.js:448:20)
at emitPromiseRejectionWarnings (internal/process/promises.js:140:18)
at process._tickCallback (internal/process/next_tick.js:69:34)
可能与代码的整体结构有关：
const { IncomingWebhook } = require('@slack/webhook');
const {BigQuery} = require('@google-cloud/bigquery');
const url = process.env.SLACK_WEBHOOK_URL;
const project =process.env.PROJECT_LIST.split(',');
const dataset = process.env.DATASET;
const table = process.env.TABLE;
const billing_project = process.env.PROJECT;
const webhook = new IncomingWebhook(url);
// subscribeSlack is the main function called by Cloud Functions.
module.exports.subscribeSlack= (pubSubEvent, context) => {
const pubsubdata = eventToBuild(pubSubEvent.data);
//select for which project to send budget alert
if (project.indexOf(pubsubdata.budgetDisplayName) === -1) {
console.log(`skip project: ${pubsubdata.budgetDisplayName.substr(0,pubsubdata.budgetDisplayName.indexOf(' '))}`);
return;
}
console.log(`project: ${pubsubdata.budgetDisplayName.substr(0,pubsubdata.budgetDisplayName.indexOf(' '))}`);
// Send message to Slack.
const message = createSlackMessage(pubsubdata);
webhook.send(message);
};
// eventToBuild transforms pubsub event message to a build object.
const eventToBuild = (data) => {
return JSON.parse(Buffer.from(data, 'base64').toString());
}
// writeInBigQuery update BigQuery table
async function writeInBigQuery(pubsubdata, createdAt, project, threshold) {
const bigquery = new BigQuery({projectId: billing_project});
const rows = [{createdAt: createdAt},
{budgetAmount:pubsubdata.budgetAmount},
{projectName: project},
{thresholdValue: threshold}];
console.log(rows);
console.log('start insert row in bigquery');
await bigquery
.dataset(dataset)
.table(table)
.insert(rows);
console.log('end insert row in bigquery');
console.log(`Inserted ${rows.length} rows`);
}
// createSlackMessage creates a message from a build object.
const createSlackMessage = (pubsubdata) => {
const formatter = new Intl.NumberFormat('de-DE', {style: 'currency', currency: 'EUR', minimumFractionDigits: 2})
const costAmount = formatter.format(pubsubdata.costAmount);
const budgetAmount = formatter.format(pubsubdata.budgetAmount);
const budgetName = pubsubdata.budgetDisplayName;
const createdAt = new Date().toISOString();
const project = budgetName.substr(0,budgetName.indexOf(' '))
let threshold = (pubsubdata.alertThresholdExceeded*100).toFixed(0);
if (!isFinite(threshold)){
threshold = 0;
}
// write current budget info in BigQuery
console.log('big query call start');
writeInBigQuery(pubsubdata, createdAt, project, threshold);
console.log('big query call end');
// create Slack message
const emoticon = threshold >= 90 ? ':fire:' : ':white_check_mark:';
notification = `${emoticon} Project: ${project}\nOverall cost: ${costAmount} \nTotal Budget: ${budgetAmount}\nThresold: ${threshold}%`
const message = {
text: notification
};
return message;
}
这个问题与 asyn ... wait 无关，而是我准备数据以写入 BigQuery 的方式中的一个错误：
const rows = [{createdAt: createdAt},
{budgetAmount:pubsubdata.budgetAmount},
{projectName: project},
{thresholdValue: threshold}];
应该是：
const rows = [{createdAt: createdAt,
budgetAmount:pubsubdata.budgetAmount,
projectName: project,
thresholdValue: threshold}];
通过此修复，一切正常。
可以在这里找到最终的和正在运行的查找：index.js
在 GCP，我创建了一个 CloudFunction，它由来自 Pub/Sub 的计费事件触发，并在 Slack 上发布了一些消息。我正在使用 node.js 10。我有以下依赖项：
{
"name": "google-container-slack",
"version": "0.0.1",
"description": "Slack integration for Google Cloud Build, using Google Cloud Functions",
"main": "index.js",
"dependencies": {
"@slack/webhook": "5.0.3",
"@google-cloud/bigquery": "4.7.0",
"@google-cloud/pubsub": "1.6.0"
}
}
当我在 BigQuery 中添加一个函数来写入新的预算信息时，我遇到了一些问题，这是基于一些官方示例：
https://github.com/googleapis/nodejs-bigquery/blob/master/samples/insertRowsAsStream.js
// writeInBigQuery update BigQuery table
async function writeInBigQuery(pubsubdata, createdAt, project, threshold) {
const bigquery = new BigQuery({projectId: billing_project});
const rows = [{createdAt: createdAt},
{budgetAmount:pubsubdata.budgetAmount},
{projectName: project},
{thresholdValue: threshold}];
console.log(rows);
console.log('start insert row in bigquery');
await bigquery
.dataset(dataset)
.table(table)
.insert(rows);
console.log('end insert row in bigquery');
console.log(`Inserted ${rows.length} rows`);
}
我猜这个问题与异步和等待有关。这是我的第一个 node.js 代码，错误消息对我来说很神秘：
Function execution took 266 ms, finished with status: 'ok'
Unhandled rejection
PartialFailureError
at request (/layers/google.nodejs.npm/npm/node_modules/@google-cloud/bigquery/build/src/table.js:1550:23)
at util.makeRequest.params (/layers/google.nodejs.npm/npm/node_modules/@google-cloud/common/build/src/util.js:367:25)
8 at Util.handleResp (/layers/google.nodejs.npm/npm/node_modules/@google-cloud/common/build/src/util.js:144:9)
8 at retryRequest (/layers/google.nodejs.npm/npm/node_modules/@google-cloud/common/build/src/util.js:432:22)
at onResponse (/layers/google.nodejs.npm/npm/node_modules/retry-request/index.js:206:7)
8 at /layers/google.nodejs.npm/npm/node_modules/teeny-request/build/src/index.js:233:13
at process._tickCallback (internal/process/next_tick.js:68:7)
8 Error: Process exited with code 16
at process.on.code (/layers/google.nodejs.functions-framework/functions-framework/node_modules/@google-cloud/functions-framework/build/src/invoker.js:393:29)
at process.emit (events.js:198:13)
at process.EventEmitter.emit (domain.js:448:20)
at process.exit (internal/process/per_thread.js:168:15)
at logAndSendError (/layers/google.nodejs.functions-framework/functions-framework/node_modules/@google-cloud/functions-framework/build/src/invoker.js:184:9)
at process.on.err (/layers/google.nodejs.functions-framework/functions-framework/node_modules/@google-cloud/functions-framework/build/src/invoker.js:390:13)
at process.emit (events.js:198:13)
at process.EventEmitter.emit (domain.js:448:20)
at emitPromiseRejectionWarnings (internal/process/promises.js:140:18)
at process._tickCallback (internal/process/next_tick.js:69:34)
可能与代码的整体结构有关：
const { IncomingWebhook } = require('@slack/webhook');
const {BigQuery} = require('@google-cloud/bigquery');
const url = process.env.SLACK_WEBHOOK_URL;
const project =process.env.PROJECT_LIST.split(',');
const dataset = process.env.DATASET;
const table = process.env.TABLE;
const billing_project = process.env.PROJECT;
const webhook = new IncomingWebhook(url);
// subscribeSlack is the main function called by Cloud Functions.
module.exports.subscribeSlack= (pubSubEvent, context) => {
const pubsubdata = eventToBuild(pubSubEvent.data);
//select for which project to send budget alert
if (project.indexOf(pubsubdata.budgetDisplayName) === -1) {
console.log(`skip project: ${pubsubdata.budgetDisplayName.substr(0,pubsubdata.budgetDisplayName.indexOf(' '))}`);
return;
}
console.log(`project: ${pubsubdata.budgetDisplayName.substr(0,pubsubdata.budgetDisplayName.indexOf(' '))}`);
// Send message to Slack.
const message = createSlackMessage(pubsubdata);
webhook.send(message);
};
// eventToBuild transforms pubsub event message to a build object.
const eventToBuild = (data) => {
return JSON.parse(Buffer.from(data, 'base64').toString());
}
// writeInBigQuery update BigQuery table
async function writeInBigQuery(pubsubdata, createdAt, project, threshold) {
const bigquery = new BigQuery({projectId: billing_project});
const rows = [{createdAt: createdAt},
{budgetAmount:pubsubdata.budgetAmount},
{projectName: project},
{thresholdValue: threshold}];
console.log(rows);
console.log('start insert row in bigquery');
await bigquery
.dataset(dataset)
.table(table)
.insert(rows);
console.log('end insert row in bigquery');
console.log(`Inserted ${rows.length} rows`);
}
// createSlackMessage creates a message from a build object.
const createSlackMessage = (pubsubdata) => {
const formatter = new Intl.NumberFormat('de-DE', {style: 'currency', currency: 'EUR', minimumFractionDigits: 2})
const costAmount = formatter.format(pubsubdata.costAmount);
const budgetAmount = formatter.format(pubsubdata.budgetAmount);
const budgetName = pubsubdata.budgetDisplayName;
const createdAt = new Date().toISOString();
const project = budgetName.substr(0,budgetName.indexOf(' '))
let threshold = (pubsubdata.alertThresholdExceeded*100).toFixed(0);
if (!isFinite(threshold)){
threshold = 0;
}
// write current budget info in BigQuery
console.log('big query call start');
writeInBigQuery(pubsubdata, createdAt, project, threshold);
console.log('big query call end');
// create Slack message
const emoticon = threshold >= 90 ? ':fire:' : ':white_check_mark:';
notification = `${emoticon} Project: ${project}\nOverall cost: ${costAmount} \nTotal Budget: ${budgetAmount}\nThresold: ${threshold}%`
const message = {
text: notification
};
return message;
}
这个问题与 asyn ... wait 无关，而是我准备数据以写入 BigQuery 的方式中的一个错误：
const rows = [{createdAt: createdAt},
{budgetAmount:pubsubdata.budgetAmount},
{projectName: project},
{thresholdValue: threshold}];
应该是：
const rows = [{createdAt: createdAt,
budgetAmount:pubsubdata.budgetAmount,
projectName: project,
thresholdValue: threshold}];
通过此修复，一切正常。
可以在这里找到最终的和正在运行的查找：index.js