2021-10-13

【Slack】Slack ～ Access Token ～

Slack

■ はじめに

https://dk521123.hatenablog.com/entry/2021/10/06/141323

で、Airflow で Slack へ通知する際に、
Access Token（アクセストークン）が必要だったのでメモる。
簡単かなーっと思ったら意外と調べたりするのに時間が掛かった、、、

【１】トークンの種類
　１）ボットトークン - Bot tokens -
　２）ユーザトークン - User tokens -
　３）アプリレベルトークン - App-level tokens -
【２】トークン生成および接続テスト
　１）Slackアプリの作成
　２）トークン生成
　３）接続テスト
【３】セキュリティ設定
　１）トークンを特定の IP 範囲に制限する
　２）トークンのローテーションと期限切れ設定

【１】トークンの種類

Slack アプリで使用できるトークンは以下の通り。

１）ボットトークン - Bot tokens -
２）ユーザトークン - User tokens -
３）アプリレベルトークン - App-level tokens -

https://slack.dev/bolt-js/ja-jp/tutorial/getting-started

１）ボットトークン - Bot tokens -

* ボットユーザーに関連づけられ、1 つのワークスペースでは
　最初にそのアプリをインストールした際に一度だけ発行される
* xoxb  で始まるトークン

２）ユーザトークン - User tokens -

* アプリをインストールまたは認証したユーザーに成り代わって
　API メソッドを呼び出すことができる
* xoxp で始まるトークン

３）アプリレベルトークン - App-level tokens -

* 全ての組織を横断できる
* xapp で始まるトークン

【２】トークン生成および接続テスト

１）Slackアプリの作成

[1] 以下のサイトを開き、「Create an App」ボタン押下

https://api.slack.com/apps

[2] 「From scratch」を選択
[3] 以下を入力し、「Create App」ボタン押下
 + App Name : 任意（今回は「HelloWorld」を入力）
 + Pick a workspace to develop your app in:対象のワークスペースを選択

２）トークン生成

[1] [OAuth & Permissions]-[Permission] を選択
[2] Scopes で「chart:write (メッセージの送信)」を追加する
 => 「You’ve changed the permission scopes your app uses. ...」って表示 
[3] 「Install App」を選択し、「Install to Workspace」を選択
[4] 「許可する」ボタン押下
 => トークン(xoxp-XXXXXX)をメモしておく

３）接続テスト

[1] 以下にアクセス

https://api.slack.com/methods/chat.postMessage/test

[2] 以下を入力し、「Test Method」ボタン押下
 + Or, provide your own token:メモったトークン(xoxp-XXXXXX)
 + channel(Required)：送りたいチャンネル
 + text：送りたいメッセージ（今回は「こんにちは世界」）
[3] Slackで対象チャンネルに送ったメッセージが
　表示されていることを確認

https://slack.com/intl/ja-jp/

【３】セキュリティ設定

１）トークンを特定の IP 範囲に制限する

* 以下の公式サイトを参照のこと。

https://api.slack.com/lang/ja-jp/securing-your-slack-app

２）トークンのローテーションと期限切れ設定

https://api.slack.com/authentication/best-practices#rotation

より抜粋
~~~~~~~~~~~~~~~~~~~~~
Rotating and expiring tokens
=> トークンのローテーションと期限切れ

Unfortunately, programmatic rotation and automatic expiration of tokens
 is not yet available for Slack apps.
=> 残念ながら、Slackアプリのための
　トークンのプログラムでのローテーションと自動的な期限期限切れは、
　まだ利用できません

Revoke tokens manually with auth.revoke.
auth.revokeのよる手動的なトークンの無効
~~~~~~~~~~~~~~~~~~~~~

https://api.slack.com/authentication/rotation#turn

より抜粋
~~~~~~~~~~~~~~~~~~~~~
Turn on token rotation 
=> トークンのローテーションONにする

Token rotation is enabled for your app
 via the OAuth & Permissions section in the sidebar of your app config.
=> あなたのアプリについて、あなたのapp configのサイドバーにある
　「OAuth & Permissions」ページ経由で、トークンのローテーションを有効にする

Warning:
=> 警告

 token rotation may not be turned off once it's turned on.
=> トークンのローテーションは、ONにしてすぐに、
　　OFFにならないかもしれない

 You'll want to be sure you're prepared to refresh your token
 within 12 hours before you turn on the toggle.
~~~~~~~~~~~~~~~~~~~~~

* app configに「token_rotation_enabled」って設定があるので
　将来的には使えるだろうが、現状は、「true」に変えても
　以下のエラーになる(redirect URLを設定すればできる？)
~~~~~~
Fix errors to save changes (line 13)

Token rotation requires valid oauth redirect urls
~~~~~~

app config：Manifest例

_metadata:
  major_version: 1
  minor_version: 1
display_information:
  name: HelloWorld
oauth_config:
  scopes:
    user:
      - chat:write
settings:
  org_deploy_enabled: false
  socket_mode_enabled: false
  token_rotation_enabled: false

参考文献

https://qiita.com/ykhirao/items/3b19ee6a1458cfb4ba21
https://zenn.dev/mokomoka/articles/6d281d27aa344e
https://note.com/npaka/n/n4bcb38a1ea74
https://api.slack.com/lang/ja-jp/hello-world-bolt

Slack ～ Incoming Webhooks ～
https://dk521123.hatenablog.com/entry/2021/10/15/091842
Apache Airflow ～通知あれこれ編～
https://dk521123.hatenablog.com/entry/2021/10/06/141323
Github Actions ～ Slack連携～
https://dk521123.hatenablog.com/entry/2024/04/03/003053

2021-10-12

【Airflow】Apache Airflow ～タイムアウト～

Airflow

■ はじめに

Apache Airflow の タイムアウト について扱う。

【１】DAGのタイムアウト関連のプロパティ
　１）dagrun_timeout
　２）dag_file_processor_timeout
　３）dagbag_import_timeout
　使用上の注意
　使用イメージ
【２】タスクのタイムアウト関連のプロパティ
　１）execution_timeout
　使用イメージ
【３】サンプル
　例１：dagrun_timeoutの試験コード
　例２：execution_timeoutの試験コード

【１】DAGのタイムアウト関連のプロパティ

１）dagrun_timeout

https://airflow.apache.org/docs/apache-airflow/stable/_api/airflow/models/index.html#airflow.models.DAG

より抜粋
~~~~~~~~~~~~~~
dagrun_timeout (datetime.timedelta)
 -- specify how long a DagRun should be up before timing out / failing,
　 so that new DagRuns can be created.
　 The timeout is only enforced for scheduled DagRuns.
 -- 新しいDagRunインスタンス を生成するために
　DagRun がどの位で立ち上がり タイムアウト / 失敗にするかを指定する
　タイムアウトは、スケジュールされたDagRunsのためだけに強制される
~~~~~~~~~~~~~~

https://airflow.apache.org/docs/apache-airflow/stable/concepts/tasks.html#timeouts

より抜粋
~~~~~~~~~~~~~~
execution_timeout controls the maximum time allowed for every execution.
If execution_timeout is breached, the task times out and AirflowTaskTimeout is raised.
~~~~~~~~~~~~~~
 => タイムアウト時に例外「AirflowException」が発生する

２）dag_file_processor_timeout

* DAGファイルプロセス DagFileProcessor 時のタイムアウト

https://airflow.apache.org/docs/apache-airflow/stable/configurations-ref.html#dag-file-processor-timeout

New in version 1.10.6.

How long before timing out a DagFileProcessor, which processes a dag file

３）dagbag_import_timeout

* Pythonファイルインポートでのタイムアウト

https://airflow.apache.org/docs/apache-airflow/stable/configurations-ref.html#dagbag-import-timeout

How long before timing out a python file import

使用上の注意

* Airflow v1系は、Timeoutされない？（詳細は以下のサイト参照）

https://stackoverflow.com/questions/57110885/how-to-define-a-timeout-for-apache-airflow-dags

* なお、 v2系(v2.0.2 in AWS MWAA)で試したところうまく動いている

使用イメージ

* DAGインスタンスを生成（New）するときに指定する

イメージ

from datetime import timedelta

dag = DAG(
  'test_timeout',
  schedule_interval=None,
  default_args=args,
  # ★注目
  dagrun_timeout=timedelta(seconds=20),
)

【２】タスクのタイムアウト関連のプロパティ

* タスクのタイムアウトに関して、関連する設定は、以下の通り。

１）execution_timeout

１）execution_timeout

https://airflow.apache.org/docs/apache-airflow/1.10.3/_api/airflow/operators/index.html#package-contents

より抜粋
~~~~~~~~~
execution_timeout (datetime.timedelta)
 – max time allowed for the execution of this task instance,
　 if it goes beyond it will raise and fail.
 – このタスクインスタンスの実行するための最大時間。
　もし超えた場合は例外が発生し失敗する
~~~~~~~~~
 => タイムアウト時に例外「AirflowTaskTimeout」が発生する

使用イメージ

* 「デフォルトでの設定」と「タスク個別に設定」することが可能

イメージ

from datetime import timedelta

# デフォルトでの設定例
default_args = {
  "provide_context": True,
  "execution_timeout": timedelta(seconds=10),
}

# オペレータ個別による設定例
task1 = PythonOperator(
    task_id='task1',
    python_callable=say_hello,
    execution_timeout=timedelta(seconds=3),
    dag=dag
)

【３】サンプル

例１：dagrun_timeoutの試験コード

import os
import time
from datetime import timedelta

from airflow import DAG
from airflow.utils.dates import days_ago
from airflow.operators.python import PythonOperator


def say_hello(**context):
  print(f"こんにちは世界。10秒寝ます Zzz")
  # ★注目★
  time.sleep(10)
  print(f"Done")

defalut_args = {
  "start_date": days_ago(2),
  "provide_context": True
}

with DAG(
  dag_id=os.path.basename(__file__).replace(".py", ""),
  description='This is a simple demo.',
  default_args=defalut_args,
  schedule_interval=None,
  # ★注目：期待としては、Job2の途中でタイムアウトする
  dagrun_timeout=timedelta(seconds=15),
  tags=['hello_world'],
) as dag:
  job1 = PythonOperator(
    task_id='say_hello_task1',
    dag=dag,
    python_callable=say_hello,
  )
  job2 = PythonOperator(
    task_id='say_hello_task2',
    dag=dag,
    python_callable=say_hello,
  )
  job3 = PythonOperator(
    task_id='say_hello_task3',
    dag=dag,
    python_callable=say_hello,
  )
  job1 >> job2 >> job3

出力結果

ログより抜粋（タイムアウト時に例外「AirflowException」が発生する）

[2021-10-14 17:16:46,001] {taskinstance.py:1265} ERROR - Received SIGTERM. Terminating subprocesses.
[2021-10-14 17:16:46,011] {taskinstance.py:1482} ERROR - Task failed with exception
Traceback (most recent call last):
...略...
raise AirflowException("Task received SIGTERM signal")

例２：execution_timeoutの試験コード

import os
import time
from datetime import timedelta

from airflow import DAG
from airflow.utils.dates import days_ago
from airflow.operators.python import PythonOperator


def say_hello(**context):
  print(f"こんにちは世界。30秒寝ます Zzz")
  # ★注目★
  time.sleep(30)
  print(f"Done")

defalut_args = {
  "start_date": days_ago(2),
  "provide_context": True,
  # ★注目★
  # Timeout時間が10秒で、上で30秒かかるのでタイムアウトするはず
  "execution_timeout": timedelta(seconds=10),
}

with DAG(
  dag_id=os.path.basename(__file__).replace(".py", ""),
  description='This is a simple demo.',
  default_args=defalut_args,
  schedule_interval=None,
  tags=['hello_world'],
) as dag:
  job1 = PythonOperator(
    task_id='say_hello_task',
    dag=dag,
    python_callable=say_hello,
    # ★注目：下のコメントアウトを外すと、設定が上書きされてタイムアウトにならない★
    # execution_timeout=timedelta(seconds=40)
  )
  job1

出力結果

ログより抜粋（タイムアウト時に例外「AirflowTaskTimeout」が発生する）

[2021-10-15 18:29:40,238] {timeout.py:42} ERROR - Process timed out, PID: 999
[2021-10-15 18:29:40,238] {dagbag.py:259} ERROR - Failed to import: /root/airflow/dags/hello.py
Traceback (most recent call last):
...略...
  File "/root/airflow/lib/python3.6/site-packages/airflow/utils/timeout.py", line 37, in handle_timeout
    raise AirflowTaskTimeout(self.error_message)
airflow.exceptions.AirflowTaskTimeout: Timeout, PID: 999

参考文献

https://qiita.com/munaita_/items/7fe474369a190d8d4ee6

2021-10-11

【Airflow】Apache Airflow ～環境構築 / Docker 編～

Airflow Docker

■ はじめに

https://dk521123.hatenablog.com/entry/2021/07/18/004531

で、ローカル上に Airflowの環境を構築したが、
いつの間にか環境がぶっつぶれた。
どうせなら、Dockerでやってみようと思ったので、メモ。

【１】前提条件

* Docker がインストールされていること
* 以下の関連記事を参照のこと。

Docker Desktop / WSL2 ～ Windows / 環境構築編～
https://dk521123.hatenablog.com/entry/2020/12/08/165505
Docker ～ Linux / 環境構築編～
https://dk521123.hatenablog.com/entry/2018/04/10/234030

【２】構築環境例

* OS ： Windows10 / Ubuntu 20.04.2 LTS
* Python ： Python 3.9.7
* Airflow : v2.1.4

【３】構築手順

* 公式サイトに記載されている

https://airflow.apache.org/docs/apache-airflow/stable/howto/docker-compose/index.html

１）docker-compose.yml のダウンロード

mkdir airflow-docker
cd airflow-docker/

curl -LfO https://airflow.apache.org/docs/apache-airflow/stable/docker-compose.yaml

２）インストール前の下準備

# 必要なディレクトリやファイルを作成
mkdir ./dags ./logs ./plugins
echo -e "AIRFLOW_UID=$(id -u)\nAIRFLOW_GID=0" > .env

３）airflow-initを実行

* 以下を実行

コマンド

# docker-compose up airflow-init
docker compose up airflow-init

注意点１

もし、コマンド後に以下のようなエラーになった場合、
docker-compose のバージョンが古い可能性があり。
~~~~
ERROR: The Compose file './docker-compose.yaml' is invalid because:
Unsupported config option for services.airflow-cli: 'profiles'
Invalid top-level property "x-airflow-common". Valid top-level sections
 for this Compose file are: version, services, networks, volumes,
 and extensions starting with "x-".
~~~~

対応方法などの詳細は、以下の関連記事を参照のこと。

https://dk521123.hatenablog.com/entry/2021/10/03/000000

注意点２

* ちなみに、v2.0.2にしたら怒られた。。。
~~~~~~
ERROR!!!: Too old Airflow version 2.0.2!
The minimum Airflow version supported: 2.1.0. Only use this or higher!
~~~~~~

４）airflowを起動

* 以下を実行
 => 結構、時間が掛かる。

コマンド

# docker-compose up
docker compose up

５）確認

a) Web UI

* ブラウザで Web UI にアクセスするために、以下にアクセス。

http://localhost:8080

* Login ページになったら、以下を入力。
 + Login : airflow
 + Password : airflow

https://codeclef.medium.com/apache-airflow-using-docker-along-with-the-authentication-caa3522111c4

b) Flower

 => タスクやワーカー数の調整などを行うWeb UI

* ブラウザで Flower にアクセスするために、以下にアクセス。

http://localhost:5555/

６）後片付け

docker compose stop

docker compose down

参考文献

公式ドキュメント
https://airflow.apache.org/docs/apache-airflow/stable/start/docker.html
動画
https://www.youtube.com/watch?v=aTaytcxy2Ck
その他
https://dev.classmethod.jp/articles/trying-apache-airflow-quick-start-docker/

■ はじめに

目次

【１】トークンの種類

１）ボットトークン - Bot tokens -

２）ユーザトークン - User tokens -

３）アプリレベルトークン - App-level tokens -

【２】トークン生成および接続テスト

１）Slackアプリの作成

２）トークン生成

３）接続テスト

【３】セキュリティ設定

１）トークンを特定の IP 範囲に制限する

２）トークンのローテーションと期限切れ設定

参考文献

関連記事

■ はじめに

目次

【１】DAGのタイムアウト関連のプロパティ

１）dagrun_timeout

２）dag_file_processor_timeout

３）dagbag_import_timeout

使用上の注意

使用イメージ

【２】タスクのタイムアウト関連のプロパティ

１）execution_timeout

使用イメージ

【３】サンプル

例１：dagrun_timeoutの試験コード

例２：execution_timeoutの試験コード

参考文献

関連記事

■ はじめに

目次

【１】前提条件

【２】構築環境例

【３】構築手順

１）docker-compose.yml のダウンロード

２）インストール前の下準備

３）airflow-initを実行

４）airflowを起動

５）確認

６）後片付け

参考文献

関連記事