如何授权 HTTP POST 请求以使用 REST API 执行数据流模板

Question

我正在尝试通过 NodeJS 后端服务器中的 REST API 将 Cloud Bigtable 执行到 Cloud Storage SequenceFile 模板。

我正在使用 axios 0.17.1 发送请求，我得到 401 状态。 我试图按照谷歌文档进行操作，但是我不知道如何授权 HTTP 请求来运行数据流模板。 我想作为服务帐户进行身份验证，并且成功生成并下载了包含私钥的 json 文件。 任何人都可以通过向https://dataflow.googleapis.com/v1b3/projects/[YOUR_PROJECT_ID]/templates:launch?gcsPath=gs://dataflow-templates/latest/ 发送 HTTP POST 请求的示例来帮助我

Answer 1

您需要从您的服务帐户凭据生成jwt。可以将 jwt 交换为访问令牌，然后该令牌可用于发出执行 Dataflow 作业的请求。完整示例：

import axios from "axios";
import jwt from "jsonwebtoken";
import mem from "mem";
import fs from "fs";

const loadServiceAccount = mem(function(){
    // This is a string containing service account credentials
    const serviceAccountJson = process.env.GOOGLE_APPLICATION_CREDENTIALS;
    if (!serviceAccountJson) {
      throw new Error("Missing GCP Credentials");
    }
})

const loadCredentials = mem(function() {

  loadServiceAccount();

  const credentials = JSON.parse(fs.readFileSync("key.json").toString());

  return {
    projectId: credentials.project_id,
    privateKeyId: credentials.private_key_id,
    privateKey: credentials.private_key,
    clientEmail: credentials.client_email,
  };
});

interface ProjectCredentials {
  projectId: string;
  privateKeyId: string;
  privateKey: string;
  clientEmail: string;
}

function generateJWT(params: ProjectCredentials) {
  const scope = "https://www.googleapis.com/auth/cloud-platform";
  const authUrl = "https://www.googleapis.com/oauth2/v4/token";
  const issued = new Date().getTime() / 1000;
  const expires = issued + 60;

  const payload = {
    iss: params.clientEmail,
    sub: params.clientEmail,
    aud: authUrl,
    iat: issued,
    exp: expires,
    scope: scope,
  };

  const options = {
    keyid: params.privateKeyId,
    algorithm: "RS256",
  };

  return jwt.sign(payload, params.privateKey, options);
}

async function getAccessToken(credentials: ProjectCredentials): Promise<string> {
  const jwt = generateJWT(credentials);
  const authUrl = "https://www.googleapis.com/oauth2/v4/token";
  const params = {
    grant_type: "urn:ietf:params:oauth:grant-type:jwt-bearer",
    assertion: jwt,
  };
  try {
    const response = await axios.post(authUrl, params);
    return response.data.access_token;
  } catch (error) {
    console.error("Failed to get access token", error);
    throw error;
  }
}

function buildTemplateParams(projectId: string, table: string) {
  return {
    jobName: `[job-name]`,
    parameters: {
      bigtableProjectId: projectId,
      bigtableInstanceId: "[table-instance]",
      bigtableTableId: table,
      outputDirectory: `[gs://your-instance]`,
      filenamePrefix: `${table}-`,
    },
    environment: {
      zone: "us-west1-a" // omit or define your own,
      tempLocation: `[gs://your-instance/temp]`,
    },
  };
}

async function backupTable(table: string) {
  console.info(`Executing backup template for table=${table}`);
  const credentials = loadCredentials();
  const { projectId } = credentials;
  const accessToken = await getAccessToken(credentials);
  const baseUrl = "https://dataflow.googleapis.com/v1b3/projects";
  const templatePath = "gs://dataflow-templates/latest/Cloud_Bigtable_to_GCS_Avro";
  const url = `${baseUrl}/${projectId}/templates:launch?gcsPath=${templatePath}`;
  const template = buildTemplateParams(projectId, table);
  try {
    const response = await axios.post(url, template, {
      headers: { Authorization: `Bearer ${accessToken}` },
    });
    console.log("GCP Response", response.data);
  } catch (error) {
    console.error(`Failed to execute template for ${table}`, error.message);
  }
}

async function run() {
  await backupTable("my-table");
}

try {
  run();
} catch (err) {
  process.exit(1);
}