日本語

背景 仕事ではArgoCDでGitOpsを行っており、Argo Workflowsでバッチ処理を実行するエンジンとして利用しています。 運用初期は利用者が多くなかったため、Argo CDのSSOはGitHub Team (グループ管理)とArgo CDに付随されているDexを使用していました。 また、認可設定はシンプルに保ち、Basic Built-in Rolesを利用して、クラスタ管理者にrole:admin、利用者にrole:readonly権限を付与していました。一方、Argo WorkflowsではSSOを設定せず、デフォルトのAuth Modeであるserverを使っていたため、認証が通れば誰でもバッチのログを閲覧できる状態でした。 しかし、プラットフォームの成長に伴い、機密性の高いログが増える見込みとなり、現在の運用は対応が難しくなってきました。 そこで、先日最小権限の原則（Principle of Least Privilege、必要最小限の権限のみを付与する運用方針）に従い、Argo CDとArgo Workflowsの権限設定を見直しました。 本記事では、公式ドキュメントに記載されていない落とし穴や、実践してみて明らかになったArgo CDとArgo Workflows権限のポイントについて解説します。 Argo CDやWorkflows権限設定の権限設定を進める際の参考になれば幸いです。 Argo CD 先述の通り、今回の構成ではGitHub Teamでグループを管理し、OpenID Connect ProviderとしてDexを利用しています。この構成における認証・認可のフローは、GitHub Teams → Dex → Argo CD Applicationとなります。 Argo CDでは、グループごとにnamespaceのアクセス権限を設定することで、必要最小限の権限のみを付与できます。 ただし、Argo CDのドキュメントRBAC Configurationの記載によると、namespaceごとに設定する場合、事前にApplication in Any Namespacesを有効化する必要があります。 デフォルトでは、Argo CDのApplicationはargocd namespaceに作成されるため、Applicationリソースを各namespaceに分割して配置する必要があります。この設定は比較的簡単な一方で、ApplicationSetのtemplateにおいてnamespaceがサポートされていないため、この方法を採用するのは現実的ではありませんでした。すでにApplicationSetによって複数のnamespaceにアプリケーションをデプロイしているため、ApplicationSetの廃止は容易ではありません。 その解決策として、namespaceごとにAppProjectを作成し、論理的なグループ化を行いました。AppProjectは、Argo CDにおいてアプリケーションをグループ化し、リソースやアクセス制限を設定するためのリソースです。以下は、その例です。 apiVersion: argoproj.io/v1alpha1 kind: AppProject metadata: name: samples namespace: argocd spec: destinations: - namespace: samples server: "https://kubernetes.default.svc" sourceRepos: - "https://github.com/org/repo.git" このAppProjectに対応する権限は、Dex設定用のConfigMapに定義します。...

なぜ 最近のFancyなノートアプリが複雑になりすぎて個人利用には不要な機能が溢れています。 その上、自分が書いたノートなのにローカルに保存されないため、サービス障害やインシデントが起きるとを閲覧できないリスクが高いです。 そのため、MakefileとGitだけで自分用簡易ノートを作りました。 詳細 この15行のMakefileがすべてです。↓ .PHONY: pull note journal pull: git checkout master && git pull origin master today=`date +%Y%m%d` year=`date +%Y` journal_path=./journal/${year}/${today}.md name=newname note: pull @mkdir -p notes && echo "# ${name}" >> notes/${today}_${name}.md && code notes/${today}_${name}.md journal: pull @mkdir -p "./journal/${year}"; if [ ! -f ${journal_path} ]; then echo "# ${today}" >> ${journal_path}; fi; code ${journal_path} ややこしいことはやっておらず、directory journal/とnotes/にmarkdownファイルを作ってvscodeで開いているだけです。 journal/ 日記を保存するディレクトリ 年別に分けるためにjournal/配下に更にサブディレクトリYYYY/を生成するように実装している make journalを実行するとjournal/YYYY/YYYYMMDD.mdが生成される 日付などは自動生成されるので自ら記入する必要なし note/ その他のノードを保存するディレクトリ こちらは年別にわけてない make note name="hogehoge"を実行するとnotes/YYYMMDD_hogehoge....

更新履歴 2022/06/26: 初稿 2024/04/20: 起動時I/O error, dev sda, sector xxxxxxxxxxのエラーが発生した。smartctlで確認したところ、問題がなかったので、手順の最新化を行い再度インストールした アップルをはじめとして、最近スペックがそこそこ良いPCがわけわからないくらい高くなり、誰でも使える個人用のコンピュータというPersnal Computerの趣旨と乖離しています。独断と偏見ですが、せっかく数十万円でPCを購入したのにもかかわらず、 パーツ交換不可なので、数年後は電子ゴミになり、自然環境に優しくない 不要なソフトウェア・サービスが多い（たとえば、2022年なのに、バージョン管理非対応のクラウドサービスiCloudが大きなシェアを占めているっていう皮肉） なんとかStoreの通知がうるさい。利用するのにアカウント（個人情報）が必要 プライバシー侵害問題 … https://www.theguardian.com/world/2021/oct/15/apple-plan-scan-child-abuse-images-tears-heart-of-privacy 別にゲーマーではないですが、数十万払ったのに自由が奪われる感じがどうしても気に入らないので、自作PCからLinuxの開発環境を構築することにしました。 ハードウェア 秋葉原まで行く元気がなかったので、すべてのパーツはAmazonやメルカリから購入することにしました。また、PCパーツの価格変動が激しいのでkeepaのメール通知機能を活用すると簡単に低価格でお気に入りのパーツ狙えます。 https://keepa.com/ 最終的に以下のパーツを買いました。 Model Price Motherboard ASUS Intel B560M-A ¥10,563 CPU Intel Core i5 10400 ¥18,943 GPU ASUS GTX 1650-4G DUAL ¥21,500 Power 玄人志向 KRPW-BK 80PLUS Bronze 750W ¥5,535 Memory KLEVV DDR4 2666 PC4-21300 8GB*2 ¥7,002 Storage Crucial SSD 500GB MX500 ¥5,980 CPU Cooler CPUクーラー 虎徹 Mark II ¥3,400 Case SAMA JAX-03W (second hand) ¥4,480 LAN cable ¥945 Total ¥78,348 あくまで筆者の例ですが、他のパーツに替えても全然問題がありません。パーツ間の互換性と電源容量だけ気をつけてください...

マネージド型k8sのDaemonSet EKSなどのKubernetesマネージドサービスは、DaemonSetを介してNodeにkube-proxy, ebs csi, cniなどのPodを適切に配置してくれます。 k8sドキュメントに記載している自動的に追加されるTolerations以外で、ワイルドカードのTolerationsもデフォルトで入っています。 tolerations: - operator: Exists EKS管轄外のDaemonSet しかし、DatadogなどEKS外でデプロイされたDaemonSet Podsを入れる際に、TaintsがついているNodeにPodがスケジューリングされないため、注意が必要です。 解決方法としては、ドキュメントに書かれているように https://kubernetes.io/docs/concepts/workloads/controllers/daemonset/#taints-and-tolerations You can add your own tolerations to the Pods of a DaemonSet as well, by defining these in the Pod template of the DaemonSet. DaemonSetのPodテンプレートにtolerationsを定義することです。 tolerations: - operator: Exists また、このようなワイルドカードのTolerationsを追加するのは便利かもしれないですが、（EKSの場合）Fargateにスケジューリングされてしまうので、Affinityもちゃんと書きましょう affinity: nodeAffinity: requiredDuringSchedulingIgnoredDuringExecution: nodeSelectorTerms: - matchExpressions: - key: eks.amazonaws.com/compute-type operator: NotIn values: - fargat 余談ですが、同じくDaemonSet介してデプロイされるfluent bitでは、不必要なTolerationsによるバグが過去起きていました。 https://github.com/aws-samples/amazon-cloudwatch-container-insights/issues/61

本記事に書かれていることを実践する際は自己責任でお願いします。不都合などが生じた場合、責任を負いかねます。 背景 日本のインターネット環境は比較的自由ですが、海外のサービスを利用際には大人の諸事情によってサイトがリダイレクトされて利用できない場合があります。 制限されるのはあまり気持ちよく感じないのとWeb閲覧時自分のプライバシーを保護するため、 先日自作プロキシサーバを自作しました。 本記事では主にHowに焦点を当てて紹介します。関連技術のWhatとWhyについては深く言及しないので、公式ドキュメントまたは他の解説記事をご参照ください。 雑なアーキテクチャ図 利用技術・サービス AWS Lightsail 仮想プライベートサーバ 3ヶ月無料 $3.5/月 ムームードメイン 3000~4000円/年（申請するドメインによる） gost OSS Webサービスを装うために使う Cloudflare gostを使うのに必要 無料プランで十分 SwitchyOmega OSS クライアント側の設定 あくまでも一例で最適解ではないです。 事前準備 VPS（仮想プライベートサーバ）を契約 今回は、AWSが提供しているLightsailという軽量仮想プライベートサーバのサービスを利用しますが、EC2やHerokuなど他のサービスを利用する場合もまったく問題ありません。リージョンは日本以外（例えばUS）に設定しておきます。AWSのCLIを利用する場合、AWS IAMなどを設定する必要がありますが、今回は基本的にいじらなくても良いです。 Lightsailでサーバを立ち上げたら、AWS CloudShellからサーバにアクセスできます。ローカルに慣れている方はキーをダウンロードして、sshでアクセスしても構いません。 念のため、 curl ipinfo.io を実行して、設定しているリージョンの住所と一致しているか確認します。 また、HTTPSを利用するため、Networking -> IPv4 FirewallからHTTPSを追加しておきます。 ドメインを取得 昨今の円安の影響で海外のドメインレジスターサービス（例えばGoDaddy）がかなり高くなっているため、国内サービスムームードメインを利用してドメインを取得しました。 Cloudflareアカウントを作成 WebサービスでもないなのになぜわざわざCloudflare使う理由 簡単に言えば、VPSのIPがブロックされる可能性はゼロではないのとプロバイダ側にIPアドレスを変更してくれないケースが多いためです。CloudflareなどのCDNサービスを挟むことでわずかな遅延が発生するかもしれませんが、可用性を向上させることができます。また、Cloudflareは無料枠を提供しており、WebSocketプロトコル (gostを使うのに必要なもの)に対応しているため、今回ユースケースに適しています。 ログインした後ガイダンスにしたがってネームサーバを設定します。 続いて、ムームードメインからデアフォルトのネームサーバを変更します。 詳細はネームサーバのセットアップ方法（GMOペパボ以外のサービス）にご参照ください。 最後はサブドメインのレコードを追加して、立ち上げたサーバのIPアドレスと紐付けます。 IPv4 FirewallにHTTPSを追加しておかないと動かないので気をつけてください。 https://developers.cloudflare.com/dns/manage-dns-records/how-to/create-subdomain/ 事前準備が完了したので、いよいよサーバ側を設定を始めます。 サーバ側の設定 Dockerエンジンをインストール Docker公式ドキュメントを参考にしてDockerエンジンをインストールします。 # Add Docker's official GPG key: sudo apt-get update sudo apt-get install ca-certificates curl gnupg sudo install -m 0755 -d /etc/apt/keyrings curl -fsSL https://download....

先日の記事にてパーティション化されたCSVオブジェクトをCloudSQLにimportする方法を紹介しました。 SDKを利用して32より多いオブジェクトをまとめる場合、GCPのコミュニティのチュートリアルのコードをそのまま使っていました。（2023.8.10にアーカイブされ済み） https://github.com/GoogleCloudPlatform/community/blob/master/archived/cloud-storage-infinite-compose/index.md 文章最後に書いてあるように This code is offered for demonstration purposes only, and should not be considered production-ready. Applying it to your production workloads is up to you! コードはデモ用途のため、そのまま本番環境で使うのはで推奨しないです。先日この文を見逃して自分の環境にデプロイし、平常運転1ヶ月後、パーティション化されたCSVオブジェクトの数が増えたらバグが出ました。 事象 起きていた事象は、composeによって作られた中間オブジェクトが消されず残り続け、最終的に数TBのとてつもなく大きいオブジェクトが作成されてしまいました。一つのオブジェクトは5TiBまでというGCSの上限を超えてしまうため、処理が失敗しました。 どこがバグ 問題は関数compose_and_cleanupから呼び出している関数delete_objects_concurrentにあります。オブジェクトをまとめた後、毎回中間オブジェクトを削除していますが、そのdelete処理自体が非同期処理ですべての処理が終了するのを待たずに次のblob.composeの実行が始まります。 The delete_objects_concurrent function is very simple, using fire-and-forget delete tasks in an executor. A more robust implementation might check the futures from the submitted tasks. チュートリアルの中にちゃんと書いてあります。（もっとロバスト性のある実装は、submit済みのタスクのfuturesをチェックするとのこと） まとめようとするオブジェクトの数が少ない場合はほとんど問題がないですが、1000オブジェクトあたりを超えるとdeleteがcompose処理より遅くなるため、next_chunkが永遠に存在しているままwhileループから脱出できない状態になります。 delete処理自体は特に制限ないようですが、数百のオブジェクトを削除する場合時間かかるとドキュメントに記載されています。 https://cloud.google.com/storage/docs/deleting-objects#delete-objects-in-bulk 解決方法 最初に試したことは、記事に書いてあるようにsubmit済みのタスクのfuturesをwait処理でチェックします。つまりすべてのdelete処理が終わるまでに、compose処理を行いません。 （python並行処理の詳細は公式ドキュメントまたは他の記事をご覧ください） from concurrent.futures import ALL_COMPLETED, ThreadPoolExecutor, wait def delete_objects_concurrent(blobs, executor, client) -> None: """Delete Cloud Storage objects concurrently....

問題 パーティション化されたCSVファイルをCloudSQLにimportする場面は時々あると思います。 残念ながらCloudSQLはBigQueryのようにwildcardsによるimportを対応していません。需要はあるようですが↓ https://issuetracker.google.com/issues/132058570?pli=1 ファイルごとにimportするとオーバーヘッドが毎回発生するため、速度的に実用性があまりないと思います。一方、importはオペレーションの1種なので、並列処理はできません。 https://cloud.google.com/sql/docs/troubleshooting#import-export HTTP Error 409: Operation failed because another operation was already in progress. There is already a pending operation for your instance. Only one operation is allowed at a time. Try your request after the current operation is complete. なので、ファイルを結合してimportするのはより現実的な解決策だと思います。 gsutil compose gsutil composeを利用すると、GCSにある複数のファイルを結合できます。 cliのみならず、SDK(google.cloud.storage.Blob.compose)も同じ機能が提供されています。 https://cloud.google.com/storage/docs/composing-objects#create-composite-client-libraries https://cloud.google.com/storage/docs/gsutil/commands/compose ただし、結合できるファイルは最大32個という制約があります。 There is a limit (currently 32) to the number of components that can be composed in a single operation....

先日GCPのDataformがGAリリースされました。 せっかくなので、まずAirflowにある既存ワークフローの一部をDataformで書き換えようと思いました。 AirflowからDataformをトリッガーする ドキュメントを調べると、AirflowからDataformをトリッガーするoperatorはすでに存在しています。 https://cloud.google.com/dataform/docs/schedule-executions-composer#create_an_airflow_dag_that_schedules_workflow_invocations 簡単にまとめると DataformCreateCompilationResultOperator: sqlxをsqlにコンパイルする DataformCreateWorkflowInvocationOperator: sqlを実行する しかし、どのようにAirflowからDataformへ変数を渡すかについてはドキュメントに記載されていません。 Dataformに変数を渡す まず、Dataformの設定ファイルdataform.jsonに変数varsを追加しておきましょう。 { "defaultSchema": "dataform", "assertionSchema": "dataform_assertions", "warehouse": "bigquery", "defaultDatabase": "project-stg", "defaultLocation": "asia-northeast1", "vars": { "bq_suffix": "_stg", "execution_date": "2023-05-24" } } DataformCreateCompilationResultOperatorのソースを調べてみたところ、compilation_resultという引数があることを発見しました。 https://github.com/apache/airflow/blob/739e6b5d775412f987a3ff5fb71c51fbb7051a89/airflow/providers/google/cloud/operators/dataform.py#LL73C29-L73C46 compilation_resultの中身を確認するため、APIの詳細を調べました。 https://cloud.google.com/dataform/reference/rest/v1beta1/CodeCompilationConfig CodeCompilationConfig内にvarsという変数を指定できるようです。 { "defaultDatabase": string, "defaultSchema": string, "defaultLocation": string, "assertionSchema": string, "vars": { string: string, ... }, "databaseSuffix": string, "schemaSuffix": string, "tablePrefix": string } BigQueryのsuffixをcode_compilation_configのvarsへ渡してみたら問題なく実行できました。ちなみに、Dataform側からはdataform.projectConfig.vars.bq_suffixで変数を呼び出せます。 DataformCreateCompilationResultOperator( task_id="create_compilation_result", project_id=PROJECT_ID, region=REGION, repository_id=REPOSITORY_ID, compilation_result={ "git_commitish": GIT_COMMITISH, "code_compilation_config": { "vars": { "bq_suffix": "_stg", } }, }, ) Dataformにcontext変数を渡す 増分処理する際によくdata_interval_endなどのcontext変数を利用して当日の差分だけ取り入れます。 しかし、DataformCreateCompilationResultOperatorではtemplate_fieldsが実装されていないため、直接{{ data_interval_end }}のようなjinjaテンプレートを渡すことはできません。...

Google Providersのバグを見つけた 先日DAGを開発中にGoogle Providers (apache-airflow-providers-google==8.9.0)のCloudDataTransferServiceJobStatusSensorを使用したところ、 project_idはオプション引数であるにも関わらず、省略するとエラーが発生するというバグに遭遇しました。 [2023-03-09, 02:31:24 UTC] {taskinstance.py:1774} ERROR - Task failed with exception Traceback (most recent call last): File "/home/airflow/.local/lib/python3.8/site-packages/airflow/sensors/base.py", line 236, in execute while not self.poke(context): File "/home/airflow/.local/lib/python3.8/site-packages/airflow/providers/google/cloud/sensors/cloud_storage_transfer_service.py", line 91, in poke operations = hook.list_transfer_operations( File "/home/airflow/.local/lib/python3.8/site-packages/airflow/providers/google/cloud/hooks/cloud_storage_transfer_service.py", line 380, in list_transfer_operations request_filter = self._inject_project_id(request_filter, FILTER, FILTER_PROJECT_ID) File "/home/airflow/.local/lib/python3.8/site-packages/airflow/providers/google/cloud/hooks/cloud_storage_transfer_service.py", line 459, in _inject_project_id raise AirflowException( airflow.exceptions.AirflowException: The project id must be passed either as `project_id` key in `filter` parameter or as project_id extra in Google Cloud connection definition....

データ基盤は下流の分析・可視化・モデリングの「基盤」となるので、品質の担保は言うまでもなく重要ですね。品質を確保するには、ワークフローの監視・検証、ワークフローのテスト、そして加工用クエリのテストがいずれも欠かせません。この記事では、ワークフロー（Airflow）の単体テスト方法について紹介します。また、ワークフローの監視・検証に関しては、過去の記事も合わせてご覧いただけると幸いです。 ワークフローの監視 ワークフローの検証 DAGの単体テスト まずは、DAGの単体テストについて説明します。厳密に言えば、DAGの実行ではなく、DAGが正確に構築されたかどうかのテストを行います。 https://airflow.apache.org/docs/apache-airflow/stable/best-practices.html#unit-tests Airflowの公式ベストプラクティスでは簡潔に紹介されていますが、具体例を挙げてさらに詳しく説明しましょう。 importのテスト importが正常にできることを確認する（importが失敗するとWeb UIからも確認できるが、単体テストする時点で確認するともっと便利） import時間を制限する。（冗長なDAGがあると解析するのに時間がかかるので、import時間を制限することで、事前に冗長なDAGを発見できる） 最低でも1つのタスクが含まれていることを確認する。 import unittest from datetime import timedelta from airflow.models import DagBag class TestImportDags(unittest.TestCase): IMPORT_TIMEOUT = 120 @classmethod def setUpClass(cls) -> None: cls.dagbag = DagBag() cls.stats = cls.dagbag.dagbag_stats def test_import_dags(self): self.assertFalse( len(self.dagbag.import_errors), f"DAG import failures. Errors: {self.dagbag.import_errors}", ) def test_import_dags_time(self): duration = sum((o.duration for o in self.stats), timedelta()).total_seconds() self.assertLess(duration, self.IMPORT_TIMEOUT) def test_dags_have_at_least_one_task(self): for key, dag in self.dagbag.dags.items(): self.assertTrue(dag, f"DAG {key} not exsit") self....