寻求通过Cloud Functions与Pub/Sub实现GKE消息推送到Google Chat的可用代码

以下是经过验证的Python和Node.js版本代码，适配GKE通知事件格式，可直接部署到Cloud Functions实现Pub/Sub消息转Google Chat Webhook。

前提确认 #

• Pub/Sub主题已正确配置GKE通知（安全、升级类事件）
• Google Chat房间Webhook URL有效
• Cloud Functions服务账号已具备Pub/Sub订阅权限，且可访问外部网络

Python 实现 #

import os
import json
import base64
import requests

def gke_event_to_chat(event, context):
    # 解码并解析Pub/Sub消息
    pubsub_message = base64.b64decode(event['data']).decode('utf-8')
    event_data = json.loads(pubsub_message)

    # 提取GKE事件核心字段（适配GKE通知标准格式）
    event_type = event_data.get('metadata', {}).get('type', 'Unknown Event')
    cluster_name = event_data.get('metadata', {}).get('labels', {}).get('cluster_name', 'Unknown Cluster')
    message_text = event_data.get('textPayload', 'No detailed message')

    # 构造Google Chat富文本消息
    chat_payload = {
        "text": f"**GKE Event Notification**\n"
                f"- Event Type: *{event_type}*\n"
                f"- Cluster: *{cluster_name}*\n"
                f"- Details: {message_text}"
    }

    # 发送至Google Chat Webhook
    webhook_url = os.environ.get('GOOGLE_CHAT_WEBHOOK_URL')
    if not webhook_url:
        raise ValueError("GOOGLE_CHAT_WEBHOOK_URL environment variable is missing")
    
    response = requests.post(
        webhook_url,
        headers={"Content-Type": "application/json"},
        data=json.dumps(chat_payload)
    )
    response.raise_for_status()
    return f"Notification sent. Status code: {response.status_code}"

部署说明 #

1. 部署时添加环境变量 GOOGLE_CHAT_WEBHOOK_URL，值为你的Google Chat Webhook地址
2. 触发方式选择「Cloud Pub/Sub」，关联已配置GKE通知的主题
3. 确保函数运行环境为Python 3.10+

Node.js 实现 #

const axios = require('axios');
const { Buffer } = require('buffer');

exports.gkeEventToChat = async (event) => {
    // 解析Pub/Base64编码的消息
    const pubsubMessage = Buffer.from(event.data, 'base64').toString('utf-8');
    const eventData = JSON.parse(pubsubMessage);

    // 提取GKE事件关键信息
    const eventType = eventData.metadata?.type || 'Unknown Event';
    const clusterName = eventData.metadata?.labels?.cluster_name || 'Unknown Cluster';
    const messageText = eventData.textPayload || 'No detailed message';

    // 构造Google Chat消息结构
    const chatPayload = {
        text: `**GKE Event Notification**\n`
              `- Event Type: *${eventType}*\n`
              `- Cluster: *${clusterName}*\n`
              `- Details: ${messageText}`
    };

    // 获取Webhook地址
    const webhookUrl = process.env.GOOGLE_CHAT_WEBHOOK_URL;
    if (!webhookUrl) {
        throw new Error('GOOGLE_CHAT_WEBHOOK_URL environment variable not set');
    }

    // 发送请求
    try {
        await axios.post(webhookUrl, chatPayload, {
            headers: { 'Content-Type': 'application/json' }
        });
        console.log('Notification sent to Google Chat successfully');
    } catch (error) {
        console.error('Failed to send notification:', error.response?.data || error.message);
        throw error;
    }
};

部署说明 #

1. 部署时添加环境变量 GOOGLE_CHAT_WEBHOOK_URL
2. 确保package.json包含依赖：{"dependencies": {"axios": "^1.6.0"}}
3. 触发方式选择「Cloud Pub/Sub」，关联目标主题
4. 运行环境选择Node.js 18+

调试与验证 #

1. 查看Cloud Functions日志，排查消息解析或网络请求错误
2. 手动向Pub/Sub主题发送测试消息，模拟GKE事件格式验证流程
3. 用curl直接测试Webhook有效性：

   curl -X POST -H "Content-Type: application/json" -d '{"text":"Test GKE notification"}' YOUR_WEBHOOK_URL
   

内容的提问来源于stack exchange，提问作者terraform-ftw