WorkflowsのCloud Storage用コネクタを用いてCloud Storage上のオブジェクトの内容を取得する

概要

WorkflowsからCloud Storage上のオブジェクト(特にjsonファイル)の内容を読み取りたいと思いたったのがこの記事のきっかけです。
今までこれをするのにCloud Run関数をWorkflowsから起動してRun関数が読み取った結果をレスポンスで受け取ってWorkflows側で処理をしていました。
が、コネクタを使えば読み取りも当然できるよなと思い試してみました。

イメージとしては以下です。

1. WorkflowsからCloud Storage上のjsonオブジェクトを読み取り
2. 読み取った結果をforループでまわす
3. ちゃんと読み取れているかログで確認
   

それではやってみます。

やってみる

準備

Workflowsから読み取るためのオブジェクト(jsonファイル)を準備します。
今回は設定ファイルぽい感じのテストデータを作成してみました。

{
  "files": [
    {
      "filename": "sales_report.csv",
      "processing_type": "typeA"
    },
    {
      "filename": "monthly_report.txt",
      "processing_type": "typeB"
    }
  ]
}

filesをキーに値として配列を設定しています。配列の中身としてはfilename(ファイル名)とprocessing_type(処理方法)を設定しています。
このjsonファイルを読み込んでWorkflows側でforループでファイル内容をログに書き込みます。
ファイルを適当なCloud Storageバケットにアップロードしてください。

Workflows側の実装

続いてWorkflows側の実装を行います。今回用いるAPIは以下です。

googleapis.storage.v1.objects.get

https://cloud.google.com/workflows/docs/reference/googleapis/storage/v1/objects/get

このAPIですがオブジェクトのプレフィックスにフォルダ(/)が入る場合はURLエンコードが必要となります。
なぜURLエンコードが必要なのかは以前のブログでまとめたのでよかったら読んでみてください。
今回の実装ではURLエンコードはurl_encode関数で行います。

呼び出しのための引数としては最低限以下を指定すれば呼び出すことができ、またファイルの内容を取得することができます。

注意点としてはaltです。altを設定しなかった場合はデフォルト動作としてオブジェクトのメタデータが取得されます(=alt:json)。今回のようにファイルの内容を取得したい場合はalt: mediaを指定してあげる必要があります。

上記踏まえWorkflowsを実装すると以下となります。

- init:
    assign:
        - targetBucket: "バケット名"
        - targetObject: ${text.url_encode("オブジェクト名")}
- getObject:
    call: googleapis.storage.v1.objects.get
    args:
        bucket: ${targetBucket}
        object: ${targetObject}
        alt: media
    result: jsonData
- printData:
    for:
        value: j
        in: ${jsonData.files}
        steps:
        - logStep:
            call: sys.log
            args:
                text: ${"Filename " + j.filename + " processing_type " + j.processing_type}
                severity: "INFO"
- return:
    return: ${jsonData}

簡単に解説します。

1. init ステップ:

   • 目的: 変数の初期化を行います。
   • assign: 2つの変数を初期化します。
     • targetBucket: Cloud Storageのバケット名を指定します。
     • targetObject: オブジェクト名をURLエンコードして指定します。text.url_encode関数を使用して、オブジェクト名をURLエンコードします。プレフィックスがある場合は必須です。

2. getObject ステップ:

   • 目的: 指定したバケットとオブジェクトからデータを取得します。
   • call: googleapis.storage.v1.objects.getAPIを呼び出します。
   • args: API呼び出しに必要な引数を指定します。
     • bucket: targetBucket変数を使用してバケット名を指定します。
     • object: targetObject変数を使用してオブジェクト名を指定します。
     • alt: mediaを指定することで、オブジェクトのコンテンツを取得します。
   • result: 取得したデータをjsonData変数に格納します。

3. printData ステップ:

   • 目的: 取得したJSONデータのfiles配列をループし、各配列の情報をログに出力します。
   • for: jsonData.files配列をループします。
     • value: ループ内で使用する変数jを指定します。
     • in: ループ対象の配列を指定します。
   • steps: ループ内で実行するステップを定義します。
     • logStep:
       • call: sys.logを呼び出してログ出力を行います。
       • args: ログ出力に必要な引数を指定します。
         • text: ログに出力するテキストを指定します。ファイル名と処理タイプを確認のためログに出力します。
         • severity: ログの重要度を指定します。この例では"INFO"を指定しています。

4. return ステップ:

   • 目的: 取得したJSONデータを返します。
   • return: jsonData変数を返します。

実装できたらWorkflowsを実行してみます。

動作させてみる

ワークフローを実行すると、ワークフローではreturnで取得したjsonデータを返却しているので出力に以下の通り作成したjsonファイルの内容が出力できていることが確認できます。

続いてログを確認します。以下の通りjsonファイルの内容を行ごとにfilenameとprocessing_typeを出力できていることが確認できます。

今回の実装ではただログに出力しているだけですが、結局のところCloud Storageから取得したjsonオブジェクトを変数に詰めることができるのでj.filenameというようにYAML上で扱うことができるのでswitch文などで処理の出し分けを行うことも可能になると思います。

ちなみにaltを設定しないで動作させた場合も確認してみます。
検証の実装としては以下となります。(forループのステップを消しています)

- init:
    assign:
        - targetBucket: "バケット名"
        - targetObject: ${text.url_encode("オブジェクト名")}
- getObject:
    call: googleapis.storage.v1.objects.get
    args:
        bucket: ${targetBucket}
        object: ${targetObject}
#        alt: media
    result: jsonData
- return:
    return: ${jsonData}

このワークフローを実行した出力を見てみます。

赤枠ですが、先ほどとことなりバケット名やハッシュ値などメタデータが出力されていることが確認できます。
alt:jsonとした場合も同じ結果です。
上記より、ファイル内容を取得したい場合はalt:media、メタデータを取得したい場合はaltなしかalt:jsonと設定すればよいことがわかります。

所感

WorkflowsからCloud Storage上のオブジェクトの内容を読み取れることで、設定ファイルをCloud Storage上に保存しておけばWorkflowsから読み取って処理の出し分けやCloud Run関数をWorkflowsから呼び出す時に引数に渡すことができると思います。
使い道は結構ありそうなので覚えておきたいなと思います。

それではまた。ナマステー

参考

https://cloud.google.com/workflows/docs/connectors?hl=ja