🪁 技術調査 - Kestra（宣言型ワークフローオーケストレーター）

docker run --pull=always --rm -it \
  -p 8080:8080 \
  --user=root \
  --name kestra \
  -v kestra_data:/app/storage \
  -v kestra_db:/app/data \
  -v /var/run/docker.sock:/var/run/docker.sock \
  -v /tmp:/tmp \
  kestra/kestra:latest server local

curl -o docker-compose.yml \
  https://raw.githubusercontent.com/kestra-io/kestra/develop/docker-compose.yml

docker compose up -d

services:
  kestra-executor:
    image: kestra/kestra:latest
    command: server executor
    environment:
      KESTRA_CONFIGURATION: *common_configuration
    depends_on: [postgres]
  kestra-worker:
    image: kestra/kestra:latest
    command: server worker --thread=64
    environment:
      KESTRA_CONFIGURATION: *common_configuration
    depends_on: [postgres]
  kestra-scheduler:
    image: kestra/kestra:latest
    command: server scheduler
    environment:
      KESTRA_CONFIGURATION: *common_configuration
    depends_on: [postgres]
  kestra-webserver:
    image: kestra/kestra:latest
    command: server webserver
    ports: ["8080:8080", "8081:8081"]
    environment:
      KESTRA_CONFIGURATION: *common_configuration
    depends_on: [postgres]

helm repo add kestra https://helm.kestra.io/
helm repo update

# 評価用（PostgreSQL + Versity 同梱）
helm install my-kestra kestra/kestra-starter

# 本番用（依存なし・設定要）
helm install my-kestra kestra/kestra

deployments:
  webserver:
    enabled: true
  executor:
    enabled: true
  indexer:
    enabled: true
  scheduler:
    enabled: true
  worker:
    enabled: true
  standalone:
    enabled: false

# ローカルモード（H2 DB 使用）
./kestra-<VERSION> server local

# スタンドアロンモード（外部 DB 接続が必要）
./kestra-<VERSION> server standalone

# 設定ファイルを指定する場合
./kestra-<VERSION> server local --config confs/application.yaml

kestra:
  queue:
    type: postgres
  repository:
    type: postgres

datasources:
  postgres:
    url: jdbc:postgresql://localhost:5432/kestra
    driver-class-name: org.postgresql.Driver
    username: kestra
    password: k3str4

kestra:
  queue:
    type: mysql
  repository:
    type: mysql

datasources:
  mysql:
    url: jdbc:mysql://localhost:3306/kestra
    driver-class-name: com.mysql.cj.jdbc.Driver
    username: kestra
    password: k3str4
    dialect: MYSQL

datasources:
  h2:
    url: "jdbc:h2:mem:public;DB_CLOSE_DELAY=-1;DB_CLOSE_ON_EXIT=FALSE"
    driver-class-name: org.h2.Driver
    username: sa
    password: ""

datasources:
  postgres:
    url: jdbc:postgresql://localhost:5432/kestra
    driver-class-name: org.postgresql.Driver
    username: kestra
    password: k3str4
    hikari:
      maximum-pool-size: 10
      connection-timeout: 30000

concurrency:
  limit: 3
  behavior: QUEUE   # QUEUE | CANCEL | FAIL

id: my-flow
namespace: company.team

description: |
  サンプルフローです。

labels:
  env: dev
  team: data

inputs:
  - id: run_task
    type: BOOLEAN
    defaults: true
  - id: target_date
    type: DATE
    required: false

variables:
  greeting: "Hello"

tasks:
  - id: step1
    type: io.kestra.plugin.core.debug.Return
    format: "{{ vars.greeting }} {{ inputs.target_date }}"

outputs:
  - id: flow_output
    type: STRING
    value: "{{ outputs.step1.value }}"

pluginDefaults:
  - type: io.kestra.plugin.core.log.Log
    values:
      level: INFO

triggers:
  - id: every_hour
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "@hourly"

tasks:
  - id: greet
    type: io.kestra.plugin.core.debug.Return
    format: "Hello, {{ inputs.name }}!"

tasks:
  - id: log_message
    type: io.kestra.plugin.core.log.Log
    message: "Processing {{ taskrun.value }}"
    level: INFO

tasks:
  - id: parallel_tasks
    type: io.kestra.plugin.core.flow.Parallel
    tasks:
      - id: task_a
        type: io.kestra.plugin.core.debug.Return
        format: "Task A: {{ taskrun.startDate }}"
      - id: task_b
        type: io.kestra.plugin.core.debug.Return
        format: "Task B: {{ taskrun.id }}"

tasks:
  - id: process_each
    type: io.kestra.plugin.core.flow.ForEach
    values: ["value1", "value2", "value3"]
    tasks:
      - id: process_item
        type: io.kestra.plugin.core.log.Log
        message: "Processing: {{ taskrun.value }}"

tasks:
  - id: condition_check
    type: io.kestra.plugin.core.flow.If
    condition: "{{ inputs.run_task }}"
    then:
      - id: when_true
        type: io.kestra.plugin.core.log.Log
        message: "条件が true です"
    else:
      - id: when_false
        type: io.kestra.plugin.core.log.Log
        message: "条件が false です"

tasks:
  - id: call_subflow
    type: io.kestra.plugin.core.flow.Subflow
    namespace: company.team
    flowId: data_processing
    inputs:
      file: "{{ outputs.upload.uri }}"
      date: "{{ inputs.target_date }}"

tasks:
  - id: python_task
    type: io.kestra.plugin.scripts.python.Script
    script: |
      print("Hello from Python!")
      result = 1 + 2
      print(f"Result: {result}")

  - id: bash_task
    type: io.kestra.plugin.scripts.shell.Script
    script: |
      echo "Current date: $(date)"
      echo "Files: $(ls -la)"

tasks:
  - id: run_in_docker
    type: io.kestra.plugin.scripts.python.Script
    containerImage: python:3.11-slim
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    script: |
      print("Hello from Docker")

pluginDefaults:
  - type: io.kestra.plugin.scripts.python.Script
    values:
      containerImage: python:3.11-slim
      taskRunner:
        type: io.kestra.plugin.scripts.runner.docker.Docker

triggers:
  - id: daily_schedule
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 9 * * 1-5"
    timezone: Asia/Tokyo
    allowConcurrent: false
    recoverMissedSchedules: LAST

triggers:
  - id: webhook
    type: io.kestra.plugin.core.trigger.Webhook
    key: "{{ secret('WEBHOOK_KEY') }}"

triggers:
  - id: upstream_done
    type: io.kestra.plugin.core.trigger.Flow
    preconditions:
      id: my_precondition
      flows:
        - namespace: company.team
          flowId: upstream_flow
          states: [SUCCESS]
      timeWindow:
        type: DAILY_TIME_DEADLINE
        deadline: "09:00:00+09:00"

triggers:
  - id: kafka_trigger
    type: io.kestra.plugin.kafka.RealtimeTrigger
    topic: my-topic
    bootstrapServers: "localhost:9092"
    groupId: kestra-group

# namespace 一覧
./kestra namespace list

# flow 一覧（namespace 指定）
./kestra flow list --namespace company.team

# flow の作成・更新（YAML ファイルから）
./kestra flow create --file my-flow.yaml
./kestra flow update --file my-flow.yaml

# flow の実行
./kestra flow execute --namespace company.team --id my-flow

# プラグインのインストール
./kestra plugins install io.kestra.plugin:plugin-script-python:LATEST

# Flow CRUD
GET    /api/v1/main/flows/{namespace}
GET    /api/v1/main/flows/{namespace}/{id}
POST   /api/v1/main/flows
PUT    /api/v1/main/flows/{namespace}/{id}
DELETE /api/v1/main/flows/{namespace}/{id}

# Execution 操作
POST   /api/v1/main/executions/{namespace}/{flowId}
GET    /api/v1/main/executions?namespace={namespace}&flowId={flowId}
GET    /api/v1/main/executions/{executionId}
DELETE /api/v1/main/executions/{executionId}

curl -X POST \
  "http://localhost:8080/api/v1/main/executions/company.team/my-flow" \
  -F env=prod

terraform {
  required_providers {
    kestra = {
      source  = "kestra-io/kestra"
      version = "~> 0.19"
    }
  }
}

provider "kestra" {
  url = "http://localhost:8080"
}

resource "kestra_flow" "example" {
  namespace = "company.team"
  flow_id   = "my-flow"
  content   = file("flows/my-flow.yaml")
}

id: git-sync
namespace: company.ops
tasks:
  - id: sync
    type: io.kestra.plugin.git.SyncFlows
    url: https://github.com/myorg/kestra-flows
    branch: main
    targetNamespace: company.team
triggers:
  - id: every_5min
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "*/5 * * * *"

# Standalone（全コンポーネント一体起動）
./kestra server standalone --worker-thread=128

# 分散モード（コンポーネント個別起動）
./kestra server webserver   -c ./application.yaml --port=8080
./kestra server executor    -c ./application.yaml
./kestra server scheduler   -c ./application.yaml
./kestra server worker      -c ./application.yaml --thread=64
./kestra server indexer     -c ./application.yaml

curl -s http://localhost:8081/health | jq .

deployments:
  worker:
    replicaCount: 4
    workerThreads: 64
    resources:
      requests:
        cpu: "4"
        memory: "8Gi"
      limits:
        cpu: "8"
        memory: "16Gi"
  executor:
    replicaCount: 3
    resources:
      requests:
        cpu: "4"
        memory: "4Gi"
  scheduler:
    replicaCount: 2
    resources:
      requests:
        cpu: "2"
        memory: "2Gi"

helm upgrade kestra kestra/kestra \
  --set image.tag=v0.22.0 \
  -f values.yaml \
  --namespace kestra

./kestra sys database migrate

# バックアップ作成（フル）
./kestra backups create FULL

# 特定リソースのみバックアップ
./kestra backups create FULL --resources KV_STORE,FLOW,SECRET

# 実行データ・ログ・メトリクスを含む
./kestra backups create FULL --include-data

# リストア
./kestra backups restore kestra:///backups/full/backup-20240917163312.kestra

pg_dump -h localhost -p 5432 -U kestra -d kestra -F tar -f kestra.tar
pg_restore -h localhost -p 5432 -U kestra -d kestra kestra.tar

GET :8081/prometheus

groups:
  - name: kestra
    rules:
      - alert: WorkerQueueBacklog
        expr: kestra_worker_job_pending > 100
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "Worker queue backlog exceeding 100 jobs"

errors:
  - id: alert_on_failure
    type: io.kestra.plugin.notifications.slack.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK') }}"
    payload: |
      {"text": "Flow {{ flow.id }} failed: {{ execution.id }}"}

# /loggers エンドポイントで動的に変更
curl -X POST http://localhost:8081/loggers/io.kestra \
  -H 'Content-Type: application/json' \
  -d '{"level": "TRACE"}'

# スレッドダンプ取得
curl http://localhost:8081/threaddump > threaddump.txt

./kestra flow validate flows/ \
  --server http://kestra:8080 \
  --api-token "${KESTRA_API_TOKEN}"

./kestra flow namespace update company.team flows/ \
  --no-delete \
  --server http://kestra:8080 \
  --api-token "${KESTRA_API_TOKEN}"

jobs:
  deploy:
    steps:
      - uses: kestra-io/deploy-action@main
        with:
          resource: flow
          namespace: company.team
          directory: flows/
          server: ${{ vars.KESTRA_URL }}
          apiToken: ${{ secrets.KESTRA_API_TOKEN }}

labels:
  system.readOnly: "true"

tasks:
  - id: query
    type: io.kestra.plugin.jdbc.postgresql.Query
    url: "{{ secret('DB_URL') }}"
    username: "{{ secret('DB_USER') }}"
    password: "{{ secret('DB_PASS') }}"

# 値を Base64 エンコードして環境変数に設定
export SECRET_DB_PASS=$(echo -n "mysecret" | base64)

tasks:
  - id: heavy_job
    type: io.kestra.plugin.scripts.python.Script
    workerGroup:
      key: gpu-workers
    script: |
      import torch

tasks:
  - id: isolated_task
    type: io.kestra.plugin.scripts.python.Script
    taskRunner:
      type: io.kestra.plugin.ee.kubernetes.runner.Kubernetes
      namespace: kestra-jobs
      resources:
        request:
          cpu: "2"
          memory: "4Gi"
        limit:
          cpu: "4"
          memory: "8Gi"
    script: |
      print("Isolated in K8s Pod")

kestra:
  jdbc:
    queues:
      minPollInterval: PT0.1S
      maxPollInterval: PT0.5S
      pollSwitchInterval: PT60S
      pollSize: 100
  executor:
    thread-count: 8

kestra:
  kafka:
    defaults:
      topic:
        partitions: 32
        replication-factor: 3
      stream:
        properties:
          num.stream.threads: "4"
          poll.ms: "25"
          commit.interval.ms: "25"

# 良い例
id: fetch_customer_data
id: processOrderEvents

# 避けるべき例
id: fetch-customer-data

# 1. Worker の状態確認
curl http://localhost:8081/worker | jq .

# 2. ログで Queue 受信エラーを確認
curl -X POST http://localhost:8081/loggers/io.kestra.core \
  -H 'Content-Type: application/json' -d '{"level": "DEBUG"}'

# 3. JDBC Queue の minPollInterval を短縮
# application.yaml:
# kestra:
#   jdbc:
#     queues:
#       minPollInterval: PT0.05S

# スレッドダンプ取得
curl http://localhost:8081/threaddump > threaddump.txt

# Worker JVM ヒープ設定
JVM_OPTS="-Xmx4g -Xms2g" ./kestra server worker

# DinD コンテナのログ確認
docker logs kestra-dind 2>&1 | tail -50

# insecure モードへの切り替え（Helm）
helm upgrade kestra kestra/kestra \
  --set dind.mode=insecure \
  --set dind.securityContext.privileged=true

# プラグイン一覧確認
./kestra plugins list

# 特定プラグインのインストール
./kestra plugins install io.kestra.plugin:plugin-jdbc-postgresql:LATEST