Elastic Workflows実践：Logstash × AI要約 × Slack通知をYAMLだけでつなぐ自動化パイプラインの作り方

input {
  rss {
    url => "https://discuss.elastic.co/c/announcements/5.rss"
    interval => 10800
  }
}

filter {
  mutate {
    add_field => {
      "source_type" => "elastic_rss"
    }
  }

  mutate {
    gsub => [
      "message", "<[^>]*>", ""
    ]
  }

  if [title] !~ /(Security Update|ESA-|CVE|What'?s new in Elastic|Elastic Stack|release|announcement|Announcing)/ {
    drop { }
  }
}

output {
  elasticsearch {
    hosts => ["https://YOUR-ENDPOINT"]
    api_key => "YOUR_API_KEY"             #Elastic Cloud URL
    index => "elastic-announcements-raw"  #APIキー
    document_id => "%{link}"              # ← 記事URLをIDにして重複を防ぐ
    manage_template => false
  }

  stdout { codec => rubydebug }
}

- pipeline.id: main

path.config: "/opt/homebrew/etc/logstash/conf.d/rss.conf"

cat /opt/homebrew/etc/logstash/pipelines.yml

# 起動 / 再起動
brew services restart logstash

# 状態確認
brew services list

# 動作ログ確認
tail -n 50 /opt/homebrew/var/log/logstash.log

tail -n 50 /opt/homebrew/var/log/logstash.log

// 件数確認
GET elastic-announcements-raw/_count

// 最新1件確認
GET elastic-announcements-raw/_search
{
  "size": 1,
  "sort": [{ "published": { "order": "desc" } }]
}

triggers:

  - type: manual

# ═══════════════════════════════════════════════════════════════
# METADATA - Identifies and describes the workflow
# この Workflow は Elastic 公式アナウンスを処理し、
# 日本語要約・分類・保存・Slack 通知まで自動化します
# ═══════════════════════════════════════════════════════════════
name: elastic_announcements_ai_production_V5_slack
description: Process only new Elastic announcements, analyze once with AI, generate actions, store, and notify to Slack
enabled: true

# ═══════════════════════════════════════════════════════════════
# TRIGGER - Starts the workflow on a fixed schedule
# 毎日 9:00 JST に Workflow を起動します
# 朝の業務開始時に新着アナウンスを確認しやすくする設定です
# ═══════════════════════════════════════════════════════════════
triggers:
  - type: scheduled
    with:
      rrule:
        freq: DAILY
        interval: 1
        tzid: Asia/Tokyo
        dtstart: 2026-03-16T09:00:00+09:00
        byhour: [9]
        byminute: [0]

# ═══════════════════════════════════════════════════════════════
# STEPS - Main workflow logic
# 検索 → 繰り返し処理 → 未処理判定 → AI要約 → 保存 → 通知
# という流れで処理します
# ═══════════════════════════════════════════════════════════════
steps:
  # ────────────────────────────────────────────────────────────
  # STEP 1 - Get raw announcements from Elasticsearch
  # raw インデックスから過去24時間分の記事を取得します
  # Output は hits.hits 配列で、次の foreach で1件ずつ処理します
  # ────────────────────────────────────────────────────────────
  - name: get_raw
    type: elasticsearch.search
    with:
      index: elastic-announcements-raw
      size: 10
      sort: "published"
      query:
        range:
          published:
            gte: "now-24h"
            lte: "now"
    on-failure:
      retry:
        max-attempts: 2
        delay: "5s"

  # --------------------------------------------------------
  # STEP 2.1 - Check whether this article was already processed
  # processed インデックスに同じ ID の記事があるか確認します
  # raw と processed で同じ ID を使い、重複処理を防ぎます
  # --------------------------------------------------------
  - name: process_items
    type: foreach
    foreach: "{{ steps.get_raw.output.hits.hits }}"
    steps:
      - name: lookup_processed
        type: elasticsearch.search
        with:
          index: elastic-announcements
          size: 1
          query:
            ids:
              values:
                - "{{ foreach.item._id }}"
        on-failure:
          retry:
            max-attempts: 2
            delay: "3s"

      # --------------------------------------------------------
      # STEP 2.2 - Continue only if the article is new
      # lookup_processed の結果が 0 件なら未処理とみなし、
      # AI 要約や保存処理へ進みます
      # --------------------------------------------------------
      - name: if_new_item
        type: if
        condition: "${{ steps.lookup_processed.output.hits.total.value == 0 }}"
        steps:

          # ----------------------------------------------------
          # STEP 2.2.1 - Generate Japanese summary and category
          # 記事タイトルと本文を使って、日本語要約とカテゴリ分類を行います
          # summary_ja と category を schema で固定し、
          # 後続ステップで扱いやすい構造化 output を返します
          # ----------------------------------------------------
          - name: analyze_article
            type: ai.prompt
            with:
              prompt: |
                あなたはElastic公式アナウンスの整理担当です。

                以下の記事を分析し、2つの項目を出力してください。

                1. 日本語サマリー（2〜3段落）
                - 元の内容に忠実に
                - 余計な推測はしない
                - タイトルは含めない
                - 元記事にない内容は追加しない

                2. category（1つ選択）
                - security_update
                - release
                - product_update
                - maintenance
                - other

                判定ルール:
                - CVE / 脆弱性 / ESA / セキュリティ修正 → security_update
                - バージョン公開 / 一般的なリリース案内 → release
                - 機能追加 / 製品更新 / 改善案内 → product_update
                - 運用 / 停止 / 保守作業 → maintenance
                - どれにも明確に当てはまらない → other

                タイトル:
                {{ foreach.item._source.title }}

                本文:
                {{ foreach.item._source.message }}

              schema:
                type: object
                properties:
                  summary_ja:
                    type: string
                  category:
                    type: string
                    enum:
                      - security_update
                      - release
                      - product_update
                      - maintenance
                      - other
                required:
                  - summary_ja
                  - category

              temperature: 0.2
            on-failure:
              retry:
                max-attempts: 2
                delay: "8s"

          # ----------------------------------------------------
          # STEP 2.2.2 - Generate short action items
          # 記事の内容をもとに、最初に確認すべき対応事項だけを
          # 短く箇条書きで生成します
          # 要約だけでなく、次の行動につながる情報を作るステップです
          # ----------------------------------------------------
          - name: generate_actions
            type: ai.prompt
            with:
              prompt: |
                次のElastic公式アナウンス記事をもとに、ユーザーが最初に確認すべき対応事項だけを書いてください。

                出力ルール:
                - 1〜3項目
                - 各項目は箇条書きで短く書く
                - 元記事にない内容は追加しない
                - 冗長な説明は書かない
                - 見出しやタグは書かない

                例:
                - 影響を受けるバージョンを確認する
                - 必要ならアップデートを検討する

                タイトル:
                {{ foreach.item._source.title }}

                本文:
                {{ foreach.item._source.message }}
            on-failure:
              retry:
                max-attempts: 2
                delay: "8s"

          # ----------------------------------------------------
          # STEP 2.2.3 - Save processed result to Elasticsearch
          # AI が作成した要約・カテゴリ・対応項目を
          # processed インデックスに保存します
          # doc_as_upsert: true により、同じ ID があれば更新、
          # なければ新規作成になります
          # ----------------------------------------------------
          - name: upsert_processed
            type: elasticsearch.update
            with:
              index: elastic-announcements
              id: "{{ foreach.item._id }}"
              doc:
                source_link: "{{ foreach.item._source.link }}"
                title: "{{ foreach.item._source.title }}"
                published: "{{ foreach.item._source.published }}"
                summary_ja: "{{ steps.analyze_article.output.content.summary_ja }}"
                actions_ja: "{{ steps.generate_actions.output.content }}"
                category: "{{ steps.analyze_article.output.content.category }}"
                processed_at: "{{ execution.startedAt }}"
              doc_as_upsert: true
            on-failure:
              retry:
                max-attempts: 2
                delay: "5s"

          # ----------------------------------------------------
          # STEP 2.2.4 - Send notification to Slack
          # 整理した記事情報を Slack に通知します
          # 保存後に関係者へ共有するための最終ステップです
          # continue: true により、Slack 送信だけ失敗しても
          # 保存処理全体は止めません
          # ----------------------------------------------------
          - name: send_slack
            type: slack
            connector-id: e3****01-****-****-****-a****c****ae
            with:
              message: |
               
                🚨 *【お知らせ】{{ foreach.item._source.title }}*

                *Published:* {{ foreach.item._source.published }}
                *Category:* {{ steps.analyze_article.output.content.category }}
                *Link:* {{ foreach.item._source.link }}

                *Summary*
                {{ steps.analyze_article.output.content.summary_ja }}

                *Actions*
                {{ steps.generate_actions.output.content }}

                *※この通知はAIにより自動生成されています。必ず原文をご確認ください。*
            on-failure:
              retry:
                max-attempts: 1
                delay: "10s"
              continue: true

condition: "${{ steps.lookup_processed.output.hits.total.value == 0 }}"