アラートとワークフローのための Slack 連携パターン

[service] -> [alert formatter] -> [slack incoming webhook] -> [channel]

[services] -> [queue topic] -> [slack dispatcher] -> [slack api]
                     |                 |
                     |                 +-> [rate limit handler]
                     +-> [dead letter queue]

[alert] -> [slack message with button] -> [button click]
   -> [action payload] -> [approval handler] -> [internal API] -> [update message]

{
  "text": "checkout error rate elevated",
  "blocks": [
    {
      "type": "header",
      "text": { "type": "plain_text", "text": "checkout error rate elevated" }
    },
    {
      "type": "section",
      "fields": [
        { "type": "mrkdwn", "text": "*severity*\\nwarn" },
        { "type": "mrkdwn", "text": "*context*\\nservice=checkout env=prod region=us-east" }
      ]
    },
    {
      "type": "section",
      "text": { "type": "mrkdwn", "text": "*action_hint*\\nClick Approve to trigger a safe restart." }
    }
  ]
}

package main

import (
  "bytes"
  "crypto/hmac"
  "crypto/sha256"
  "encoding/hex"
  "encoding/json"
  "io"
  "log"
  "net/http"
  "net/url"
  "os"
  "strconv"
  "strings"
  "sync"
  "time"

  "github.com/slack-go/slack"
)

type BlockActionPayload struct {
  Type  string `json:"type"`
  Team  struct{ ID string `json:"id"` } `json:"team"`
  User  struct{ ID string `json:"id"` } `json:"user"`
  Channel struct {
    ID string `json:"id"`
  } `json:"channel"`
  Message struct {
    Ts string `json:"ts"`
  } `json:"message"`
  ResponseURL string `json:"response_url"`
  Actions []struct {
    ActionID string `json:"action_id"`
    Value    string `json:"value"`
    Type     string `json:"type"`
  } `json:"actions"`
}

type InternalAction struct {
  Action     string `json:"action"`
  TeamID     string `json:"team_id"`
  ChannelID  string `json:"channel_id"`
  MessageTS  string `json:"message_ts"`
  UserID     string `json:"user_id"`
  Value      string `json:"value"`
}

var (
  // 本番環境では、TTL を設定して Redis に保存します
  seenMu sync.Mutex
  seen   = map[string]time.Time{}
  ttl    = 10 * time.Minute
)

func main() {
  botToken := os.Getenv("SLACK_BOT_TOKEN")
  signingSecret := os.Getenv("SLACK_SIGNING_SECRET")
  channelID := os.Getenv("SLACK_CHANNEL_ID")
  internalURL := os.Getenv("INTERNAL_API_URL")
  listenAddr := os.Getenv("LISTEN_ADDR") // e.g. :8080

  if botToken == "" || signingSecret == "" || channelID == "" || internalURL == "" || listenAddr == "" {
    log.Fatal("missing env vars SLACK_BOT_TOKEN SLACK_SIGNING_SECRET SLACK_CHANNEL_ID INTERNAL_API_URL LISTEN_ADDR")
  }

  api := slack.New(botToken)

  // 承認ボタン付きのアラートメッセージを送信します。
  // ボタンは、action_id と value を持つインタラクティブな Block Kit 要素です。
  // Slack Block Kit ボタン要素のドキュメントを参照してください。
  blocks := slack.Blocks{
    BlockSet: []slack.Block{
      slack.NewHeaderBlock(slack.NewTextBlockObject("plain_text", "checkout error rate elevated", false, false)),
      slack.NewSectionBlock(
        slack.NewTextBlockObject("mrkdwn", "*severity*\\nwarn\\n*context*\\nservice=checkout env=prod", false, false),
        nil,
        nil,
      ),
      slack.NewActionBlock(
        "actions_1",
        slack.NewButtonBlockElement("approve_restart", "restart", slack.NewTextBlockObject("plain_text", "Approve restart", false, false)),
      ),
    },
  }

  _, ts, err := api.PostMessage(channelID, slack.MsgOptionBlocks(blocks.BlockSet...))
  if err != nil {
    log.Fatalf("PostMessage failed: %v", err)
  }
  log.Printf("posted alert message_ts=%s", ts)

  // インタラクティブ性エンドポイント
  http.HandleFunc("/slack/actions", func(w http.ResponseWriter, r *http.Request) {
    rawBody, err := io.ReadAll(r.Body)
    if err != nil {
      http.Error(w, "read body failed", http.StatusBadRequest)
      return
    }
    r.Body.Close()

    // 生ボディとタイムスタンプに対する Slack リクエスト署名を検証します。
    // Slack のリクエスト検証ドキュメントを参照してください。
    if !verifySlackRequest(r.Header, rawBody, signingSecret) {
      http.Error(w, "invalid signature", http.StatusUnauthorized)
      return
    }

    // Slack は、payload=JSON を持つ application/x-www-form-urlencoded を送信します
    vals, err := url.ParseQuery(string(rawBody))
    if err != nil {
      http.Error(w, "bad form body", http.StatusBadRequest)
      return
    }
    payloadStr := vals.Get("payload")
    if payloadStr == "" {
      http.Error(w, "missing payload", http.StatusBadRequest)
      return
    }

    var p BlockActionPayload
    if err := json.Unmarshal([]byte(payloadStr), &p); err != nil {
      http.Error(w, "bad payload json", http.StatusBadRequest)
      return
    }

    // 3 秒以内に承認します。実際の作業は非同期で行い、更新には response_url を使用します。
    // Slack のインタラクティブ性ドキュメントの承認タイミングを参照してください。
    w.WriteHeader(http.StatusOK)
    _, _ = w.Write([]byte(""))

    go func() {
      if p.Type != "block_actions" || len(p.Actions) == 0 {
        return
      }
      a := p.Actions[0]
      if a.ActionID != "approve_restart" {
        return
      }

      // 承認の重複排除
      key := strings.Join([]string{p.Team.ID, p.Channel.ID, p.Message.Ts, p.User.ID, a.ActionID, a.Value}, "|")
      if !tryOnce(key) {
        return
      }

      req := InternalAction{
        Action:    "approve_restart",
        TeamID:    p.Team.ID,
        ChannelID: p.Channel.ID,
        MessageTS: p.Message.Ts,
        UserID:    p.User.ID,
        Value:     a.Value,
      }

      if err := postJSON(internalURL, req); err != nil {
        log.Printf("internal action failed: %v", err)
        _ = replyViaResponseURL(p.ResponseURL, "action failed, check logs")
        return
      }

      _ = replyViaResponseURL(p.ResponseURL, "approval received, internal action triggered")
    }()
  })

  log.Printf("listening on %s", listenAddr)
  log.Fatal(http.ListenAndServe(listenAddr, nil))
}

func tryOnce(key string) bool {
  now := time.Now()

  seenMu.Lock()
  defer seenMu.Unlock()

  for k, t := range seen {
    if now.Sub(t) > ttl {
      delete(seen, k)
    }
  }
  if _, ok := seen[key]; ok {
    return false
  }
  seen[key] = now
  return true
}

func verifySlackRequest(h http.Header, body []byte, signingSecret string) bool {
  ts := h.Get("X-Slack-Request-Timestamp")
  sig := h.Get("X-Slack-Signature")
  if ts == "" || sig == "" {
    return false
  }

  tsInt, err := strconv.ParseInt(ts, 10, 64)
  if err != nil {
    return false
  }

  // リプレイリスクを減らすために、5 分以上前のリクエストを拒否します。
  if time.Since(time.Unix(tsInt, 0)) > 5*time.Minute {
    return false
  }

  base := "v0:" + ts + ":" + string(body)
  mac := hmac.New(sha256.New, []byte(signingSecret))
  mac.Write([]byte(base))
  sum := hex.EncodeToString(mac.Sum(nil))
  expected := "v0=" + sum

  return hmac.Equal([]byte(expected), []byte(sig))
}

func postJSON(url string, body any) error {
  b, err := json.Marshal(body)
  if err != nil {
    return err
  }
  req, err := http.NewRequest(http.MethodPost, url, bytes.NewReader(b))
  if err != nil {
    return err
  }
  req.Header.Set("Content-Type", "application/json")
  c := &http.Client{Timeout: 5 * time.Second}
  res, err := c.Do(req)
  if err != nil {
    return err
  }
  defer res.Body.Close()
  if res.StatusCode < 200 || res.StatusCode >= 300 {
    return io.ErrUnexpectedEOF
  }
  return nil
}

func replyViaResponseURL(responseURL string, text string) error {
  if responseURL == "" {
    return nil
  }
  // response_url は JSON ペイロードを受け取り、デフォルトでは一時的（ephemeral）に投稿できます。
  b, _ := json.Marshal(map[string]string{
    "text": text,
  })
  req, _ := http.NewRequest(http.MethodPost, responseURL, bytes.NewReader(b))
  req.Header.Set("Content-Type", "application/json")
  c := &http.Client{Timeout: 5 * time.Second}
  res, err := c.Do(req)
  if err != nil {
    return err
  }
  defer res.Body.Close()
  return nil
}

import os
import json
import time
import threading
import requests
from flask import Flask, request, make_response
from slack_sdk import WebClient
from slack_sdk.signature import SignatureVerifier

SLACK_BOT_TOKEN = os.environ["SLACK_BOT_TOKEN"]
SLACK_SIGNING_SECRET = os.environ["SLACK_SIGNING_SECRET"]
SLACK_CHANNEL_ID = os.environ["SLACK_CHANNEL_ID"]
INTERNAL_API_URL = os.environ["INTERNAL_API_URL"]

client = WebClient(token=SLACK_BOT_TOKEN)
verifier = SignatureVerifier(signing_secret=SLACK_SIGNING_SECRET)

app = Flask(__name__)

# 本番環境では、これらを Redis に保存します
_seen = {}
_TTL_SECONDS = 600

def try_once(key: str) -> bool:
    now = int(time.time())
    expired = [k for k, t in _seen.items() if now - t > _TTL_SECONDS]
    for k in expired:
        _seen.pop(k, None)
    if key in _seen:
        return False
    _seen[key] = now
    return True

def post_internal_action(payload: dict) -> None:
    requests.post(INTERNAL_API_URL, json=payload, timeout=5)

def reply_via_response_url(response_url: str, text: str) -> None:
    if not response_url:
        return
    requests.post(response_url, json={"text": text}, timeout=5)

def send_alert_with_button() -> None:
    blocks = [
        {"type": "header", "text": {"type": "plain_text", "text": "checkout error rate elevated"}},
        {"type": "section", "text": {"type": "mrkdwn", "text": "*severity*\\nwarn\\n*context*\\nservice=checkout env=prod"}},
        {
            "type": "actions",
            "block_id": "actions_1",
            "elements": [
                {
                    "type": "button",
                    "text": {"type": "plain_text", "text": "Approve restart"},
                    "action_id": "approve_restart",
                    "value": "restart"
                }
            ]
        }
    ]
    # chat.postMessage には chat:write スコープが必要です。
    client.chat_postMessage(channel=SLACK_CHANNEL_ID, text="checkout error rate elevated", blocks=blocks)

@app.post("/slack/actions")
def slack_actions():
    raw_body = request.get_data()  # Slack はパース前の生ボディを検証することを推奨
    if not verifier.is_valid_request(raw_body, request.headers):
        return make_response("invalid signature", 401)

    # Slack は、JSON を含む payload フィールドを持つ application/x-www-form-urlencoded を送信します。
    payload_str = request.form.get("payload", "")
    if not payload_str:
        return make_response("missing payload", 400)
    payload = json.loads(payload_str)

    # 3 秒以内に承認します。そうしないと Slack ユーザーはエラーを表示します。
    # Slack のインタラクティブ性ドキュメントの承認を参照してください。
    resp = make_response("", 200)

    def work():
        if payload.get("type") != "block_actions":
            return
        actions = payload.get("actions", [])
        if not actions:
            return

        a = actions[0]
        if a.get("action_id") != "approve_restart":
            return

        team_id = payload.get("team", {}).get("id", "")
        channel_id = payload.get("channel", {}).get("id", "")
        user_id = payload.get("user", {}).get("id", "")
        message_ts = payload.get("message", {}).get("ts", "")
        value = a.get("value", "")

        key = "|".join([team_id, channel_id, message_ts, user_id, "approve_restart", value])
        if not try_once(key):
            return

        internal_payload = {
            "action": "approve_restart",
            "team_id": team_id,
            "channel_id": channel_id,
            "message_ts": message_ts,
            "user_id": user_id,
            "value": value,
        }

        try:
            post_internal_action(internal_payload)
            reply_via_response_url(payload.get("response_url", ""), "approval received, internal action triggered")
        except Exception:
            reply_via_response_url(payload.get("response_url", ""), "action failed, check logs")

    threading.Thread(target=work, daemon=True).start()
    return resp

if __name__ == "__main__":
    send_alert_with_button()
    app.run(host="0.0.0.0", port=int(os.getenv("PORT", "8080")))