自動実行ファイル加工をOCI FunctionsとObject Storageで実装してみた #oci

はじめに

DBとIaaS以外のOCIを使ってみるシリーズの第一弾です。

OCI Functionsを使って元ファイルの取得と加工を行い、Object Storageに加工したキャッシュファイルを保存し、ユーザがそのキャッシュファイルにアクセスするという構成を考えてみました。

構成概要

ユーザがObject Storageに保存されている加工済みキャッシュファイルにアクセスします
Object Storageのアクセスログを保存します
アクセスしたログがあった際にConnector HubがOCI Functionsを実行
OCI Functionsでは、Object Storageに保存されている加工済みキャッシュファイルが古ければ元ファイルを取得して加工して保存し、古くなければ何もしないというコードが実装されている

想定ユースケース（今回は勉強のために構築するので実用的かどうかは微妙です。）
- ファイルアクセスは朝〜夜に5分に1回程度。深夜帯は無し。（OCI Functionsの実行回数が余裕で無料枠に収まる程度）
- ファイルアクセスしたときの情報は1時間程度古くてもOK。
- 元ファイルへのアクセスはなるべく抑えたい、もしくは、ユーザから元ファイルに直接アクセスさせたくない、ユーザには加工済みのファイルをアクセスさせたい、等。

余談

当初、AWS LambdaやS3と同じ感覚で考えていて、
OCI FunctionsとObject Storageの2つだけで構成を考えて構築を進めていたのですが、
pythonコードが完成してOCI Functionsをスケジューリングしようとした段階で
OCI Functionsでは想定していた方法ではスケジュール実行できないことに気がつきました。
そのため、いろいろ考えた結果、ロギングからOCI Functionsを呼び出す構成となっています。

1. Object Storage

OCIコンソールの左上ハンバーガーマークの
ストレージ > バケット より、
オブジェクト・ストレージのバケットを作成します。

後で使うのでバケットのネームスペースを確認しておきます。

作成できたら、左下のリソースから
「Read Access Events」を有効にします。サービス・コネクタで利用します。
今回はイベント・サービスは使わないので「オブジェクト・イベントの出力」はオフでも大丈夫です。

2. OCI Functions

2.1. VCNの作成

OCIコンソールの左上ハンバーガーマークの
ネットワーキング > 仮想クラウド・ネットワーク より
ファンクションを置くためのVCNとサブネットを作成します。

サブネットは、インターネットからのアクセスは無いのでプライベートサブネットとします。
プライベートサブネットにファンクションを置く場合はサービス・ゲートウェイが必須です。また今回はファンクションからインターネットアクセスするのでNATゲートウェイも作成します。

2.2. レポジトリの作成

OCIコンソールの左上ハンバーガーマークの
開発者サービス > コンテナ・レジストリ から
ファンクションのアプリケーションのレポジトリの作成します。
今回は
testuser/testfunc
というレポジトリ名（仮）にします。

2.3. ポリシーの追加

OCIコンソールの左上ハンバーガーマークの
アイデンティティとセキュリティ > ドメイン から
動的グループを作成します。

ドメインを選択（Defaultの場合、Defaultのリンクをクリック）し、動的グループから「動的グループを作成」をクリック。
今回は、自コンパートメントの全てのファンクションに動的グループを割り当てます。

ALL {resource.type = 'fnfunc', resource.compartment.id = '＜コンパートメントのOCID＞'}

なお、ここでは全てのファンクションに動的グループを割り当てしていますが、
もっと絞りたい場合は、ファンクションを作成した後にそのファンクションのOCIDのみ許可するポリシーを追加する形になるかと思います。

次に、先ほど割り当てた動的グループに対して、オブジェクト・ストレージのバケットのオブジェクト操作を許可するポリシーを追加します。

Allow dynamic-group ＜動的グループ名＞ to manage objects in compartment ＜コンパートメント名＞

なお、ここでは全てのオブジェクトの操作を許可していますが、
もっと絞りたい場合は、指定されたバケットのみに許可するポリシーを追加する形になるかと思います。

2.4. ファンクションの作成

OCIコンソールの左上ハンバーガーマークの
開発者サービス > アプリケーション から
アプリケーションを作成します。
今回は
testapp
という名にします。
先ほど作成したVCNとサブネットを指定します。

アプリケーションが作成できたらスタート・ガイドに沿ってファンクションを作成します。ただし以下の点を変更します。

(4) 一意のリポジトリ名の接頭辞を指定して、・・・

[repo-name-prefix]は今回は
testuser
です。先ほど作成したレポジトリ名のスラッシュまでの部分です。

(8) 'hello-world'ボイラープレート・ファンクションを生成します

先ほど作成したレポジトリ名のスラッシュ以降の部分の
testfuncをファンクション名にしたいので、

fn init --runtime python testfunc

とします。

(11) ファンクションを呼び出します

同様に、

fn invoke testapp testfunc

となります。

また、アプリケーション「testapp」の画面の左下の「ログ」にて
「Function Invocation Logs」をオンにしておくと切り分け等が捗ります。

次に、ファンクションからオブジェクト・ストレージを操作するpythonコードを編集します。

func.py抜粋

import io
import json
import logging
import oci
import datetime
from fdk import response

def handler(ctx, data: io.BytesIO = None):

  #今回はsignerを使ってObjectStorageClientでバケットにアクセスします
  signer = oci.auth.signers.get_resource_principals_signer()
  client = oci.object_storage.ObjectStorageClient(config={}, signer=signer)

  #オブジェクト・ストレージのバケットの情報を記載する
  bucket_namespace = "＜バケットのネームスペース＞"
  bucket_name = "＜バケット名＞"

  #オブジェクトストレージに保存するオブジェクトのファイル名を記載する
  object_name = "＜ファイル名＞"

  #対象オブジェクトの更新日時を取得する
  rtn = client.head_object(bucket_namespace, bucket_name, object_name)
  lasttime = datetime.datetime.strptime(rtn.headers['last-modified'], '%a, %d %b %Y %H:%M:%S %Z')
  nowtime = datetime.datetime.now()

  #保存されているファイルの更新日時が45分前より古いかどうか
  if nowtime - lasttime > datetime.timedelta(minutes = 45):

    #元ファイルを取得し加工する処理がここに入ります。
    #今回は省略します。
    #なおオブジェクト・ストレージに保存したいファイルの内容を変数outputに文字列で格納しています。streamを受け付けているのでバイナリでも可です。
    output = '＜保存したい内容がここに入る＞'
  
    #オブジェクト・ストレージにファイルを保存する
    client.put_object(bucket_namespace, bucket_name, object_name, output)

    #オブジェクト・ストレージにファイルを保存したことをログに出力
    logging.getLogger().info("put_object done")

  else:
    #オブジェクト・ストレージにファイルを保存しなかったことをログに出力
    logging.getLogger().info("put_object skipped")

  #returnする値は簡素でOK
  return response.Response(
    ctx, response_data=json.dumps({}),
    headers={"Content-Type": "application/json"}
  )

詳細は、最新のAPI Referenceの Object Storage > ObjectStorageClient を参照ください。

今回のpythonコードでimport ociしているので、requirements.txtファイルに以下追記します。

requirements.txt抜粋

oci

コードが完成したら以下のコマンドでデプロイします。

fn -v deploy --app testapp

3. Connector Hub

OCIコンソールの左上ハンバーガーマークの
監視および管理 > コネクタ から
サービス・コネクタを作成します。

ソース: ロギング
ターゲット: ファンクション
を選択します。

ソースの構成にて
「1. Object Storage」で作成した「Read Access Events」ログのコンパートメント、ロググループ、ログ名を指定します。

このままだとオブジェクトアクセス以外のログでもファンクションが動作してしまうので、
フィルタ（ログ・フィルタ・タスク）で
type = com.oraclecloud.objectstorage.getobject
を指定し、オブジェクトがgetされた時のみファンクションが動作するようにします。

次に、タスクの構成は飛ばして、ターゲットの構成にて、
「2. OCI Functions」で作成したアプリケーション名、ファンクション名を指定します。

4. 全体の動作確認

今回作成したバケットに対象ファイルキャッシュとなるオブジェクトが保存されている状態で全体の動作確認します。

4.1 事前認証済リクエストの作成

オブジェクト・ストレージ > バケットの詳細 > 事前認証済リクエスト にて

事前認証済リクエスト・ターゲット: バケット
アクセス・タイプ: オブジェクトの読取りを許可
オブジェクト・リストの有効化: オフ
有効期限: 任意（デフォルトは1週間後）

を設定します。
バケットからオブジェクトをダウンロードする際に利用できる事前認証済リクエストURLのオブジェクトのファイル名を除いた文字列が取得できます。
例：　https://＜バケットのネームスペース＞.objectstorage.＜リージョン＞.oci.customer-oci.com/＜文字列＞/n/＜バケットのネームスペース＞/b/＜バケット名＞/o/
この文字列はここで表示された以降は二度と表示されないので安全に保管します。

4.2 ダウンロードの実行

「4.1 事前認証済リクエストの作成」で出力されたURLの末尾にオブジェクトのファイル名を付加してブラウザアクセスしてみます。

例：　https://＜バケットのネームスペース＞.objectstorage.＜リージョン＞.oci.customer-oci.com/＜文字列＞/n/＜バケットのネームスペース＞/b/＜バケット名＞/o/＜オブジェクトのファイル名＞

アプリケーションの「ログ」の設定にて「Function Invocation Logs」をオンにしているので、
アクセスしてから数分後に、そのログを参照すると
INFO - put_object done （キャッシュが古かった場合）
もしくは
INFO - put_object skipped （キャッシュが古くなかった場合）
のログが出力されているはずです。
また、前者の場合は、オブジェクトの更新日時が新しくなっていることも確認できます。

あとがき

現状だと、他の基本有料OCIサービスを利用しないとできないようなのでOCI Functionsのスケジュール実行は諦めましたが、
将来、もし簡単にスケジュール実行できるようになったら試してみたいと思います。

P.S.

Qiitaの編集をMacとWindowsからやったのでスクショのフォントがバラバラになってます。
あと、前回の記事はQiitaガイドラインに沿わない気がする（Qiita非推奨の「日記」にある意味該当するかも）ので削除しました。

自動実行ファイル加工をOCI FunctionsとObject Storageで実装してみた