Elasticsearch で検索条件に合致した nested の要素だけを抽出

Elasticsearch の nested を用いた検索において、inner_hits を使って検索条件に合致した nested の要素だけを抽出するようにしてみました。

今回のソースは http://github.com/fits/try_samples/tree/master/blog/20211011/

はじめに

Elasticsearch では nested という型を用いる事で、入れ子構造を実現できるようになっています。

ここでは、商品毎に複数のエディションがあって、エディション毎にセールの履歴を保持しているような、2層の nested を持つサンプルデータを用意してみました。

Elasticsearch のマッピング設定は以下。

マッピング設定
$ curl -s http://localhost:9200/items/_mappings | jq
{
  "items": {
    "mappings": {
      "properties": {
        "editions": {
          "type": "nested",
          "properties": {
            "edition": {
              "type": "keyword"
            },
            "price": {
              "type": "scaled_float",
              "scaling_factor": 100
            },
            "sales": {
              "type": "nested",
              "properties": {
                "date_from": {
                  "type": "date"
                },
                "date_to": {
                  "type": "date"
                },
                "sale_price": {
                  "type": "scaled_float",
                  "scaling_factor": 100
                }
              }
            }
          }
        },
        "id": {
          "type": "keyword"
        },
        "name": {
          "type": "text"
        }
      }
    }
  }
}

ドキュメントの内容は以下の通りです。

全ドキュメント内容
$ curl -s http://localhost:9200/items/_search | jq
{
  ・・・
  "hits": {
    ・・・
    "hits": [
      {
        "_index": "items",
        "_type": "_doc",
        "_id": "id-001",
        "_score": 1,
        "_source": {
          "id": "id-001",
          "name": "item-A",
          "editions": [
            {
              "edition": "Standard",
              "price": 1000,
              "sales": [
                {
                  "sale_price": 900,
                  "date_from": "2021-01-01T00:00:00Z",
                  "date_to": "2021-01-03T00:00:00Z"
                },
                {
                  "sale_price": 800,
                  "date_from": "2021-07-01T00:00:00Z",
                  "date_to": "2021-07-05T00:00:00Z"
                }
              ]
            },
            {
              "edition": "Extra",
              "price": 2000,
              "sales": [
                {
                  "sale_price": 1800,
                  "date_from": "2021-01-01T00:00:00Z",
                  "date_to": "2021-01-03T00:00:00Z"
                },
                {
                  "sale_price": 1700,
                  "date_from": "2021-07-01T00:00:00Z",
                  "date_to": "2021-07-05T00:00:00Z"
                },
                {
                  "sale_price": 1500,
                  "date_from": "2021-09-01T00:00:00Z",
                  "date_to": "2021-09-02T00:00:00Z"
                }
              ]
            }
          ]
        }
      },
      {
        "_index": "items",
        "_type": "_doc",
        "_id": "id-002",
        "_score": 1,
        "_source": {
          "id": "id-002",
          "name": "item-B",
          "editions": [
            {
              "edition": "Standard",
              "price": 1500,
              "sales": [
                {
                  "sale_price": 1400,
                  "date_from": "2021-09-01T00:00:00Z",
                  "date_to": "2021-09-05T00:00:00Z"
                }
              ]
            },
            {
              "edition": "Extra",
              "price": 5000,
              "sales": []
            }
          ]
        }
      },
      {
        "_index": "items",
        "_type": "_doc",
        "_id": "id-003",
        "_score": 1,
        "_source": {
          "id": "id-003",
          "name": "item-C",
          "editions": [
            {
              "edition": "Standard",
              "price": 7000,
              "sales": [
                {
                  "sale_price": 6800,
                  "date_from": "2021-01-01T00:00:00Z",
                  "date_to": "2021-01-03T00:00:00Z"
                },
                {
                  "sale_price": 6700,
                  "date_from": "2021-02-01T00:00:00Z",
                  "date_to": "2021-02-05T00:00:00Z"
                },
                {
                  "sale_price": 6600,
                  "date_from": "2021-04-01T00:00:00Z",
                  "date_to": "2021-04-02T00:00:00Z"
                },
                {
                  "sale_price": 6500,
                  "date_from": "2021-07-01T00:00:00Z",
                  "date_to": "2021-07-15T00:00:00Z"
                }
              ]
            }
          ]
        }
      },
      {
        "_index": "items",
        "_type": "_doc",
        "_id": "id-004",
        "_score": 1,
        "_source": {
          "id": "id-004",
          "name": "item-D",
          "editions": [
            {
              "edition": "Standard",
              "price": 4000,
              "sales": []
            },
            {
              "edition": "Extra",
              "price": 6000,
              "sales": []
            },
            {
              "edition": "Premium",
              "price": 9000,
              "sales": []
            }
          ]
        }
      }
    ]
  }
}

検索

Deno 用の TypeScript で検索処理を実装してみます。

Deno 用の公式 Elasticsearch クライアントライブラリは今のところ無さそうなので、fetch を使って REST API を呼び出すような関数を用意してみました。

es_util.ts
// Elasticsearch の REST API 呼び出し
export const send = async (url: string, method: string, body: any) => {
    const res = await fetch(url, {
        method,
        headers: {
          "Content-Type": "application/json"
        },
        body: JSON.stringify(body)
    })

    const resBody = await res.json()

    return {
        ok: res.ok,
        body: resBody
    }
}

ここでは、下記のような条件で検索する事にします。

  • 2021/7/1 以降のセール価格が 1500 以下だったもの

1. inner_hits を用いない検索

まずは、inner_hits を用いず普通に検索します。

nested のフィールドを検索条件として指定するために、Nested query({ nested: path: ・・・, query: ・・・ })を使用します。

editions 内の sales を検索条件として指定するために、Nested query を入れ子にします。

search1.ts
import { send } from './es_util.ts'

const index = 'items'
const baseUrl = 'http://localhost:9200'

const indexUrl = `${baseUrl}/${index}`

const { body } = await send(`${indexUrl}/_search`, 'POST', {
  query: {
    nested: {
      path: 'editions',
      query: {
        nested: {
          path: 'editions.sales',
          query: {
            bool: {
              must: [
                {
                  range: {
                    'editions.sales.date_from': { gte: '2021-07-01T00:00:00Z' }
                  }
                },
                {
                  range: {
                    'editions.sales.sale_price': { lte: 1500 }
                  }
                }
              ]
            }
          }
        }
      }
    }
  }
})

console.log(JSON.stringify(body, null, 2))

実行結果は次の通り、_source は元のドキュメント内容そのままなので、どのエディションの、どのセール履歴が検索条件に合致したかは分かりません。

実行結果
> deno run --allow-net search1.ts

{
  ・・・
  "hits": {
    ・・・
    "hits": [
      {
        "_index": "items",
        "_type": "_doc",
        "_id": "id-001",
        "_score": 2,
        "_source": {
          "id": "id-001",
          "name": "item-A",
          "editions": [
            {
              "edition": "Standard",
              "price": 1000,
              "sales": [
                {
                  "sale_price": 900,
                  "date_from": "2021-01-01T00:00:00Z",
                  "date_to": "2021-01-03T00:00:00Z"
                },
                {
                  "sale_price": 800,
                  "date_from": "2021-07-01T00:00:00Z",
                  "date_to": "2021-07-05T00:00:00Z"
                }
              ]
            },
            {
              "edition": "Extra",
              "price": 2000,
              "sales": [
                {
                  "sale_price": 1800,
                  "date_from": "2021-01-01T00:00:00Z",
                  "date_to": "2021-01-03T00:00:00Z"
                },
                {
                  "sale_price": 1700,
                  "date_from": "2021-07-01T00:00:00Z",
                  "date_to": "2021-07-05T00:00:00Z"
                },
                {
                  "sale_price": 1500,
                  "date_from": "2021-09-01T00:00:00Z",
                  "date_to": "2021-09-02T00:00:00Z"
                }
              ]
            }
          ]
        }
      },
      {
        "_index": "items",
        "_type": "_doc",
        "_id": "id-002",
        "_score": 2,
        "_source": {
          "id": "id-002",
          "name": "item-B",
          "editions": [
            {
              "edition": "Standard",
              "price": 1500,
              "sales": [
                {
                  "sale_price": 1400,
                  "date_from": "2021-09-01T00:00:00Z",
                  "date_to": "2021-09-05T00:00:00Z"
                }
              ]
            },
            {
              "edition": "Extra",
              "price": 5000,
              "sales": []
            }
          ]
        }
      }
    ]
  }
}

2. inner_hits を用いた検索

それでは、本題の inner_hits を使った検索を行います。

inner_hits は Nested query で指定できます。 inner_hits を指定した場合は _source とは別の inner_hits という項目に該当結果が設定されます。

ここでは、inner_hits で取得する内容は Source filtering を使って、元の _source から取り除くように設定しています。(Source filtering は inner_hits 内でも設定可)

また、_sourceinner_hits の結果を結合して、検索条件に合致した要素だけで構成されるドキュメントを構築する処理(toDoc)も用意してみました。

search2.ts
・・・
const { body } = await send(`${indexUrl}/_search`, 'POST', {
  _source: {
    // _source から editions の内容を取り除く
    excludes: ['editions']
  },
  query: {
    nested: {
      path: 'editions',
      query: {
        nested: {
          path: 'editions.sales',
          query: {
            bool: {
              must: [
                {
                  range: {
                    'editions.sales.date_from': { gte: '2021-07-01T00:00:00Z' }
                  }
                },
                {
                  range: {
                    'editions.sales.sale_price': { lte: 1500 }
                  }
                }
              ]
            }
          },
          // editions.sales に対する inner_hits の設定
          inner_hits: {}
        }
      },
      // editions に対する inner_hits の設定
      inner_hits: {
        _source: {
          // inner_hits の _source から sales の部分を取り除く
          excludes: ['editions.sales']
        }
      }
    }
  }
})

console.log(JSON.stringify(body, null, 2))

console.log('-----')

// _source の内容に inner_hits の内容を再帰的に結合する処理
const toDoc = (res: any) => res.hits.hits.map((r: any) => {
  const doc = { ...r._source }

  for (const [k, v] of Object.entries(r.inner_hits ?? {})) {
    const key = k.split('.').slice(-1)[0]
    doc[key] = toDoc(v)
  }

  return doc
})

const res = toDoc(body)

console.log(JSON.stringify(res, null, 2))

実行結果は次のようになりました。

実行結果
> deno run --allow-net search2.ts 

{
  ・・・
  "hits": {
    ・・・
    "hits": [
      {
        "_index": "items",
        "_type": "_doc",
        "_id": "id-001",
        "_score": 2,
        "_source": {
          "name": "item-A",
          "id": "id-001"
        },
        "inner_hits": {
          "editions": {
            "hits": {
              ・・・
              "hits": [
                {
                  "_index": "items",
                  "_type": "_doc",
                  "_id": "id-001",
                  "_nested": {
                    "field": "editions",
                    "offset": 0
                  },
                  "_score": 2,
                  "_source": {
                    "price": 1000,
                    "edition": "Standard"
                  },
                  "inner_hits": {
                    "editions.sales": {
                      "hits": {
                        ・・・
                        "hits": [
                          {
                            "_index": "items",
                            "_type": "_doc",
                            "_id": "id-001",
                            "_nested": {
                              "field": "editions",
                              "offset": 0,
                              "_nested": {
                                "field": "sales",
                                "offset": 1
                              }
                            },
                            "_score": 2,
                            "_source": {
                              "date_to": "2021-07-05T00:00:00Z",
                              "sale_price": 800,
                              "date_from": "2021-07-01T00:00:00Z"
                            }
                          }
                        ]
                      }
                    }
                  }
                },
                {
                  "_index": "items",
                  "_type": "_doc",
                  "_id": "id-001",
                  "_nested": {
                    "field": "editions",
                    "offset": 1
                  },
                  "_score": 2,
                  "_source": {
                    "price": 2000,
                    "edition": "Extra"
                  },
                  "inner_hits": {
                    "editions.sales": {
                      "hits": {
                        ・・・
                        "hits": [
                          {
                            "_index": "items",
                            "_type": "_doc",
                            "_id": "id-001",
                            "_nested": {
                              "field": "editions",
                              "offset": 1,
                              "_nested": {
                                "field": "sales",
                                "offset": 2
                              }
                            },
                            "_score": 2,
                            "_source": {
                              "date_to": "2021-09-02T00:00:00Z",
                              "sale_price": 1500,
                              "date_from": "2021-09-01T00:00:00Z"
                            }
                          }
                        ]
                      }
                    }
                  }
                }
              ]
            }
          }
        }
      },
      ・・・
    ]
  }
}
-----
[
  {
    "name": "item-A",
    "id": "id-001",
    "editions": [
      {
        "price": 1000,
        "edition": "Standard",
        "sales": [
          {
            "date_to": "2021-07-05T00:00:00Z",
            "sale_price": 800,
            "date_from": "2021-07-01T00:00:00Z"
          }
        ]
      },
      {
        "price": 2000,
        "edition": "Extra",
        "sales": [
          {
            "date_to": "2021-09-02T00:00:00Z",
            "sale_price": 1500,
            "date_from": "2021-09-01T00:00:00Z"
          }
        ]
      }
    ]
  },
  {
    "name": "item-B",
    "id": "id-002",
    "editions": [
      {
        "price": 1500,
        "edition": "Standard",
        "sales": [
          {
            "date_to": "2021-09-05T00:00:00Z",
            "sale_price": 1400,
            "date_from": "2021-09-01T00:00:00Z"
          }
        ]
      }
    ]
  }
]

CDK の Vpc.fromLookup では StringParameter.valueFromLookup を使う

CDK で既存 VPC とのピアリング設定等を行う場合、Vpc.fromLookup を用いて VPC を参照する事になると思います。

VPC ID のハードコーディングを避けるには、SSM(Systems Manager)のパラメータストアから値を取得する方法が考えられます。

CDK で SSM パラメータストアの値を参照するには、以下のような方法が用意されていますが、Vpc.fromLookup の vpcId へ指定できたのは (b) で (a) は駄目でした。

  • (a) StringParameter.valueForStringParameter ※
  • (b) StringParameter.valueFromLookup
 ※ valueForTypedStringParameter や 
    new StringParameter(・・・).stringValue でも同じ

(a) の戻り値を Vpc.fromLookup の vpcId へ設定すると、All arguments to Vpc.fromLookup() must be concrete (no Tokens) というエラーが発生しました。

これは、(a) が返すのは実際の値ではなく、実際の値を参照するためのトークンだという事が原因のようです。

一方、(b) は cdk synth 時にパラメータストアから値を取得して cdk.context.json ファイルにキャッシュするようになっており、パラメータストアから取得した実際の値が返ってきます。

ただし、cdk.context.json にキャッシュした後は、パラメータストアから再取得しないようで、パラメータストア側の値を更新しても反映してくれませんでした。※

 ※ cdk context --clear で cdk.context.json をクリアすると反映されましたが
スタック定義例
import { App, Stack } from '@aws-cdk/core'
import { StringParameter } from '@aws-cdk/aws-ssm'
import * as ec2 from '@aws-cdk/aws-ec2'

const vpcIdParamName = '/sample/vpcid'

const app = new App()

const stack = new Stack(app, 'VPCPeeringSample', {
    env: {
        account: process.env.CDK_DEFAULT_ACCOUNT,
        region: process.env.CDK_DEFAULT_REGION
    }
})

const vpc = new ec2.Vpc(stack, 'Vpc', {
    cidr: '192.168.0.0/16',
    subnetConfiguration: [
        { name: 'sample', subnetType: ec2.SubnetType.PUBLIC }
    ]
})

const peerVpc = ec2.Vpc.fromLookup(stack, 'PeerVpc', {
    vpcId: StringParameter.valueFromLookup(stack, vpcIdParamName)
    // 以下のようにするとエラーになる
    // vpcId: StringParameter.valueForStringParameter(stack, vpcIdParamName)
})

// VPC ピアリング
const vpcPeering = new ec2.CfnVPCPeeringConnection(stack, 'VpcPeering', {
    vpcId: vpc.vpcId,
    peerVpcId: peerVpc.vpcId
})

vpc.publicSubnets.map((s, i) => {
    new ec2.CfnRoute(stack, `src-peering-${i}`, {
        routeTableId: s.routeTable.routeTableId,
        destinationCidrBlock: peerVpc.vpcCidrBlock,
        vpcPeeringConnectionId: vpcPeering.ref
    })
})

・・・

今回のソースは http://github.com/fits/try_samples/tree/master/blog/20210926/

Jest と Vue Test Utils による Vue コンポーネントのテスト

Vue CLI で作成した TypeScript 用の Vue プロジェクトに対して、JestVue Test Utils(@vue/test-utils) を追加導入し、Vue コンポーネントのテスト(TypeScript で実装)を実施するようにしてみました。

今回は、Vue CLIvue create 時に、Manually select features を選択して、Choose Vue versionTypeScript にだけチェックを付け、Vue.js 2.x と 3.x のプロジェクトをそれぞれ作成して実施しています。

今回のソースは http://github.com/fits/try_samples/tree/master/blog/20210621/

(a) Vue.js 2.x の場合

まずは、Vue.js 2.x のプロジェクトで実施してみます。

  • Vue.js 2.6.14

a-1. テストモジュールの導入

Jest と Vue Test Utils に必要なモジュールをインストールして設定します。

jest のインストールと設定

jestts-jest(テストコードを TypeScript で実装するため)、vue-jest(テストコードで Vue コンポーネントを扱うため)をインストールします。

> npm i -D jest ts-jest vue-jest

ts-jest と vue-jest を適用するように Jest を設定し、jsdom を適用してテストを実施するよう testEnvironmentjsdom を設定しておきます。

jest.config.js 設定例
module.exports = {
    testEnvironment: 'jsdom',
    preset: 'ts-jest',
    transform: {
        '.*\\.(vue)$': 'vue-jest'
    }
}

Vue Test Utils のインストール

@vue/test-utils をインストールします。

> npm i -D @vue/test-utils

@types/jest のインストールと設定

ついでに、VS Code で jest の関数を使ってもエラー表示にならないようにするための措置を行っておきます。

まずは、jest の型定義をインストールします。

> npm i -D @types/jest

tsconfig.jsontypesjest を追加します。

tsconfig.json 設定例
{
・・・
    "types": [
      ・・・
      "jest"
    ],
・・・
}

a-2. コンポーネントの実装

テスト対象のコンポーネントを実装します。

src/components/Counter.vue
<template>
  <div>
    <p>
      <button @click="countUp">count up</button>
    </p>
    <p>
      counter: {{ count }}
    </p>
  </div>
</template>

<script lang="ts">
import Vue from 'vue'

export default Vue.extend({
  data() {
    return {
      count: 0
    }
  },
  methods: {
    countUp() {
      this.count++
    }
  }
})
</script>

a-3. テストの実装と実施

テストコードを実装し、テストしてみます。

TypeScript で実装する場合、型の都合上 counter.vm.count とする事はできなかったので、とりあえずは counter.vm.$data.count としています。

(counter.vm as any).count とする事も可能ですが、正直どちらも微妙な気がします。

tests/Counter.test.vue
import { mount } from '@vue/test-utils'
import Counter from '../src/components/Counter.vue'

test('count up', async () => {
    const counter = mount(Counter)

    expect(counter.vm.$data.count).toBe(0)
    // 以下でも可
    // expect((counter.vm as any).count).toBe(0)

    await counter.get('button').trigger('click')

    expect(counter.vm.$data.count).toBe(1)
})

jest コマンドでテストを実施します。

テスト実行
> npx jest

 PASS  tests/Counter.test.ts (5.907 s)
  ・・・

Test Suites: 1 passed, 1 total
Tests:       1 passed, 1 total
Snapshots:   0 total
Time:        7.434 s
Ran all test suites.

(b) Vue.js 3.x の場合

次は、Vue.js 3.x のプロジェクトで実施してみます。

  • Vue.js 3.1.1

b-1. テストモジュールの導入

基本的には 2.x と同じですが、インストールするモジュールのバージョンが多少異なります。

jest のインストールと設定

現時点で、Vue.s 3.x に対応した vue-jest をインストールするにはバージョンに next を指定する必要がありました。

また、(現時点で)インストールされる vue-jest 5.0.0-alpha.10 は、jest のバージョン 27 には対応しておらず、jest と ts-jest はバージョン 26 をインストールする必要がありました。

> npm i -D jest@26 ts-jest@26 vue-jest@next

Jest の設定は 2.x と同じですが、こちらは 2.x とは違って testEnvironment を設定しなくても特に支障は無さそうでした。

jest.config.js 設定例
module.exports = {
    testEnvironment: 'jsdom',
    preset: 'ts-jest',
    transform: {
        '.*\\.(vue)$': 'vue-jest'
    }
}

Vue Test Utils のインストール

@vue/test-utils のバージョンも next とする必要がありました。

> npm i -D @vue/test-utils@next

また、@types/jest は 2.x と同様に、必要に応じてインストールして設定しておきます。

b-2. コンポーネントの実装

テスト対象コンポーネントを実装します。

src/components/Counter.vue
<template>
  <div>
    <p>
      <button @click="countUp">count up</button>
    </p>
    <p>
      counter: {{ count }}
    </p>
  </div>
</template>

<script lang="ts">
import { defineComponent } from 'vue'

export default defineComponent({
  data() {
    return {
      count: 0
    }
  },
  methods: {
    countUp() {
      this.count++
    }
  }
})
</script>

上記を以下のように Composition API で実装しても、同じテストコードでテストできました。

src/components/Counter.vue(Composition API 版)
<template>
  <div>
    <p>
      <button @click="countUp">count up</button>
    </p>
    <p>
      counter: {{ count }}
    </p>
  </div>
</template>

<script lang="ts">
import { ref } from 'vue'

export default {
  setup() {
    const count = ref(0)
    const countUp = () => count.value++

    return {
      count,
      countUp
    }
  }
}
</script>

b-3. テストの実装と実施

こちらは 2.x とは違って、counter.vm.count とする事ができました。

tests/Counter.test.vue
import { mount } from '@vue/test-utils'
import Counter from '../src/components/Counter.vue'

test('count up', async () => {
    const counter = mount(Counter)

    expect(counter.vm.count).toBe(0)

    await counter.get('button').trigger('click')

    expect(counter.vm.count).toBe(1)
})

jest コマンドでテストを実施します。

テスト実行
> npx jest

 PASS  tests/Counter.test.ts
 ・・・

Test Suites: 1 passed, 1 total
Tests:       1 passed, 1 total
Snapshots:   0 total
Time:        3.78 s, estimated 4 s
Ran all test suites.

CDK と LocalStack でローカルに Lambda と DynamoDB の実行環境を構築

AWS CDK (Cloud Development Kit) を使って、ローカル環境の LocalStack に Lambda 関数と DynamoDB のテーブルを構築してみました。

下記のようなツールを使用し、CDK によるスタックと Lambda 関数ハンドラーは TypeScript で実装しました。

今回のソースは http://github.com/fits/try_samples/tree/master/blog/20210524/

1. はじめに

今回は cdk init を使わずに CDK のコードを自前で構築する事にします。

まず、TypeScript で実装するために下記モジュールを

  • typescript
  • ts-node
  • @types/node

CDK を使って Lambda と DynamoDB を定義するために下記モジュールを

TypeScript のビルド用に下記モジュールを

  • esbuild

Lambda から DynamoDB へ接続するために下記モジュールを

そして、aws-cdk を使って LocalStack に対してデプロイ等を実施するために下記モジュールを

  • aws-cdk-local

これらをインストールして package.json は以下のようになりました。

package.json
{
  "name": "cdk_localstack_sample",
  "version": "1.0.0",
  "description": "",
  "devDependencies": {
    "@types/node": "^14.17.0",
    "aws-cdk": "^1.105.0",
    "aws-cdk-local": "^1.65.4",
    "esbuild": "^0.12.1",
    "ts-node": "^9.1.1",
    "typescript": "^4.2.4"
  },
  "dependencies": {
    "@aws-cdk/aws-dynamodb": "^1.105.0",
    "@aws-cdk/aws-lambda": "^1.105.0",
    "@aws-cdk/aws-lambda-nodejs": "^1.105.0",
    "@aws-sdk/client-dynamodb": "^3.16.0"
  }
}

CDK 用の設定ファイル cdk.json はシンプルに以下のようにしました。

cdk.json
{
    "app": "npx ts-node --prefer-ts-exts app.ts"
}

2. 実装

DynamoDB のテーブルと Lambda 関数を構築するだけの簡単なスタックを定義しました。

app.ts (スタック定義)
import { App, Construct, Stack, StackProps, CfnOutput } from '@aws-cdk/core'
import * as dynamodb from '@aws-cdk/aws-dynamodb'
import * as lambda from '@aws-cdk/aws-lambda'
import { NodejsFunction } from '@aws-cdk/aws-lambda-nodejs'

class SampleStack extends Stack {
    constructor(scope: Construct, id: string, props?: StackProps) {
        super(scope, id, props)

        const table = new dynamodb.Table(this, 'Items', {
            partitionKey: { name: 'id', type: dynamodb.AttributeType.STRING}
        })

        const func = new NodejsFunction(this, 'SampleFunc', {
            runtime: lambda.Runtime.NODEJS_14_X,
            entry: './src/handler.ts',
            environment: {
                'TABLE_NAME': table.tableName
            }
        })

        // テーブルへの書き込み権限を Lambda に付与
        table.grantWriteData(func)

        // Lambda 関数名の出力
        new CfnOutput(this, 'functionName', {
            value: func.functionName
        })
    }
}

const app = new App()

new SampleStack(app, 'SampleStack')

続いて Lambda 関数ハンドラーです。

LocalStack で実行している場合に、接続先の DynamoDB を LocalStack のものに切り替える必要があります。

また、LocalStack では Node.js ランタイムの Lambda 関数ハンドラーを docker で実行するようになっており、その際に LocalStack のホスト名は LOCALSTACK_HOSTNAME 環境変数で、ポート番号は EDGE_PORT 環境変数でそれぞれ渡されるようになっていました。

そこで、今回はこれらの環境変数の値を利用して DynamoDB の接続先を変えるようにしてみました。

src/handler.ts(Lambda 関数ハンドラー)
import { DynamoDBClient, PutItemCommand } from '@aws-sdk/client-dynamodb'

const tableName = process.env.TABLE_NAME

const config = {}

if (process.env.LOCALSTACK_HOSTNAME) {
    // LocalStack の DynamoDB へ接続するための設定
    config['endpoint'] = `http://${process.env.LOCALSTACK_HOSTNAME}:${process.env.EDGE_PORT}`
}

const client = new DynamoDBClient(config)

export interface Input {
    id: string
}

export const handler = async (event: Input) => {
    const res = await client.send(new PutItemCommand({
        TableName: tableName,
        Item: {
            id: { S: event.id }
        }
    }))

    console.log(`dynamodb put-item: ${JSON.stringify(res)}`)

    return {
        statusCode: 201,
        body: {
            id: event.id
        }
    }
}

3. デプロイと動作確認

それでは、LocalStack を実行してデプロイを実施してみます。

LocalStack 実行

docker コマンドを使って LocalStack を実行します。

Node.js ランタイムの Lambda 関数を実行するために LAMBDA_EXECUTOR 環境変数の値を docker にして、/var/run/docker.sockマッピングしています。

docker で LocalStack 実行
$ docker run --rm -it -p 4566:4566 -e LAMBDA_EXECUTOR=docker -v /var/run/docker.sock:/var/run/docker.sock localstack/localstack

デプロイ

aws-cdk-local モジュールには、cdklocal コマンドという LocalStack に対してデプロイ等を実施する cdk コマンドのラッパーが用意されているので、このコマンドを使います。

cdk コマンドと同様に初回時は bootstrap を実行します。

ブートストラップ処理
$ npx cdklocal bootstrap

 ⏳  Bootstrapping environment aws://000000000000/ap-northeast-1...
CDKToolkit: creating CloudFormation changeset...
7:38:43 AM | UPDATE_IN_PROGRESS   | AWS::CloudFormation::Stack | UsePublicAccessBlockConfiguration

 ✅  Environment aws://000000000000/ap-northeast-1 bootstrapped.

続いてデプロイを実施します。

デプロイ処理
$ npx cdklocal deploy --require-approval never

Bundling asset SampleStack/SampleFunc/Code/Stage...
  ・・・
SampleStack: deploying...
・・・
SampleStack: creating CloudFormation changeset...
・・・


 ✅  SampleStack

Outputs:
SampleStack.functionName = SampleStack-lambda-c75c6ee1

Stack ARN:
arn:aws:cloudformation:us-east-1:000000000000:stack/SampleStack/c83384b9

これで DynamoDB のテーブルと Lambda 関数が作られ、Lambda の関数名は SampleStack-lambda-c75c6ee1 となりました。

動作確認

ここからは AWS CLI v2 を使って動作確認してみます。

LocalStack へ接続するには --endpoint-urlhttp://localhost:4566 を設定して aws コマンドを使います。

それでは、lambda invoke を使って SampleStack-lambda-c75c6ee1 を実行してみます。

Lambda の実行
$ aws --endpoint-url=http://localhost:4566 lambda invoke --function-name SampleStack-lambda-c75c6ee1  --payload '{"id":"id1"}' --cli-binary-format raw-in-base64-out output.json

{
    "StatusCode": 200,
    "LogResult": "",
    "ExecutedVersion": "$LATEST"
}

output.json の内容は以下のようになり、正常に実行できているようです。

output.json
{"body":{"id":"id1"},"statusCode":201}

次に、この Lambda 関数が出力した CloudWatch のログを確認してみます。 こちらも特に問題は無さそうです。

CloudWatch のログ確認
$ aws --endpoint-url=http://localhost:4566 logs tail "/aws/lambda/SampleStack-lambda-c75c6ee1"

2021-05-23T22:42:30.401000+00:00 2021/05/23/[LATEST]24101d6b START RequestId: ced21627-9622-15e9-45cd-e1c1c93c39a9 Version: $LATEST
2021-05-23T22:42:30.419000+00:00 2021/05/23/[LATEST]24101d6b 
2021-05-23T22:42:30.437000+00:00 2021/05/23/[LATEST]24101d6b 2021-05-23T22:42:28.141Z   ced21627-9622-15e9-45cd-e1c1c93c39a9    INFO    dynamodb put-item: {"$metadata":{"httpStatusCode":200,"requestId":"0f5854a3-1110-42e7-bd95-3ff8bd8bb003","attempts":1,"totalRetryDelay":0},"ConsumedCapacity":{"CapacityUnits":1,"TableName":"SampleStack-Items5C12978B-b5cec818"}}
2021-05-23T22:42:30.473000+00:00 2021/05/23/[LATEST]24101d6b END RequestId: ced21627-9622-15e9-45cd-e1c1c93c39a9
・・・

最後に、SampleStack-Items5C12978B-b5cec818 テーブルの内容も出力してみました。

DynamoDB テーブルの検索結果
$ aws --endpoint-url=http://localhost:4566 dynamodb scan --table-name SampleStack-Items5C12978B-b5cec818

{
    "Items": [
        {
            "id": {
                "S": "id1"
            }
        }
    ],
    "Count": 1,
    "ScannedCount": 1,
    "ConsumedCapacity": null
}

CDK で作った CloudFormation テンプレートをプログラム内からデプロイする

AWS CDK (Cloud Development Kit) では通常 cdk deploy コマンドを使ってデプロイ処理を実施します。

これを cdk コマンドを使わずにプログラム内から実施できないか、以下の 2通りで試してみました。

  • (a) AWS CDK の API を利用
  • (b) AWS SDK の CloudFormation API を利用

なお、実際のところ (a) の場合でも、内部的には CloudFormation の API が呼び出されています。

今回のソースは http://github.com/fits/try_samples/tree/master/blog/20210418/

はじめに

今回は DynamoDB のテーブルを作るだけの単純なスタック(以下)を使用します。

a_sample/stack.ts
import { Construct, Stack, StackProps, CfnOutput } from '@aws-cdk/core'
import * as dynamodb from '@aws-cdk/aws-dynamodb'

export class SampleStack extends Stack {
    constructor(scope: Construct, id: string, props?: StackProps) {
        super(scope, id, props)

        const table = new dynamodb.Table(this, 'Items', {
            partitionKey: { name: 'id', type: dynamodb.AttributeType.STRING }
        })

        new CfnOutput(this, 'TableName', {
            value: table.tableName
        })
    }
}

このスタックの CloudFormation テンプレートは、以下のような処理で文字列として取得できます。

a_sample/sample.ts
import { App } from '@aws-cdk/core'
import { SampleStack } from './stack'

const app = new App()
new SampleStack(app, 'SampleStack')
// CloudFormation テンプレートの文字列化
const tpl = JSON.stringify(app.synth().stacks[0].template)

console.log(tpl)
実行結果例
> cd a_sample

> npx ts-node sample.ts
{"Resources":{"Items5C12978B":{"Type":"AWS::DynamoDB::Table","Properties":{"KeySchema":[{"AttributeName":"id","KeyType":"HASH"}],"AttributeDefinitions":[{"AttributeName":"id","AttributeType":"S"}],"ProvisionedThroughput":{"ReadCapacityUnits":5,"WriteCapacityUnits":5}},"UpdateReplacePolicy":"Retain","DeletionPolicy":"Retain"}},"Outputs":{"TableName":{"Value":{"Ref":"Items5C12978B"}}}}

(a) AWS CDK の API を利用

cdk deploy コマンド処理内で使われている CloudFormationDeployments を直接使うのが簡単そうだったので、今回はこれを使います。

デフォルトでは cdk deploy 時と同じように進捗状況が出力されますが、これは quiettrue にする事で抑制できました。

a_sample/deploy_a.ts
import { App } from '@aws-cdk/core'
import { SdkProvider } from 'aws-cdk/lib/api/aws-auth'
import { CloudFormationDeployments } from 'aws-cdk/lib/api/cloudformation-deployments'

import { SampleStack } from './stack'

const app = new App()
new SampleStack(app, 'Sample1Stack')

const run = async () => {
    const deployer = new CloudFormationDeployments({
        sdkProvider: await SdkProvider.withAwsCliCompatibleDefaults()
    })
    // デプロイ処理
    const r = await deployer.deployStack({
        stack: app.synth().stacks[0],
        quiet: true
    })

    console.log(r)
}

run().catch(err => console.error(err))
パッケージインストール例
> cd a_sample
> npm install -D ts-node typescript @types/node
・・・
> npm install @aws-cdk/aws-dynamodb aws-cdk
・・・

実行結果は以下の通りです。 CloudFormation のスタックが登録され、DynamoDB のテーブルが問題なく作成されました。

実行結果
> npx ts-node deploy_a.ts

Sample1Stack: creating CloudFormation changeset...
{
  noOp: false,
  outputs: { TableName: 'Sample1Stack-Items・・・' },
  ・・・
  stackArtifact: CloudFormationStackArtifact {
    assembly: CloudAssembly {
      ・・・
    },
    id: 'Sample1Stack',
    manifest: {
      type: 'aws:cloudformation:stack',
      environment: 'aws://unknown-account/unknown-region',
      properties: [Object],
      metadata: [Object]
    },
    messages: [],
    _dependencyIDs: [],
    environment: {
      account: 'unknown-account',
      region: 'unknown-region',
      name: 'aws://unknown-account/unknown-region'
    },
    templateFile: 'Sample1Stack.template.json',
    parameters: {},
    tags: {},
    assumeRoleArn: undefined,
    cloudFormationExecutionRoleArn: undefined,
    stackTemplateAssetObjectUrl: undefined,
    requiresBootstrapStackVersion: undefined,
    bootstrapStackVersionSsmParameter: undefined,
    terminationProtection: undefined,
    stackName: 'Sample1Stack',
    assets: [],
    displayName: 'Sample1Stack',
    name: 'Sample1Stack',
    originalName: 'Sample1Stack',
    _deps: [],
    _template: { Resources: [Object], Outputs: [Object] }
  }
}

(b) AWS SDK の CloudFormation API を利用

今回は、AWS SDK for JavaScript v3 を使います。

実質的に AWS CLIcloudformation deploy コマンドと同じ事をするだけですが、AWS SDK にはこれに相当する API が用意されていないようなので、下記ソースコードを参考に自作してみました。

なお、CDK の下記ソースコードでも同じような処理になっていました。

簡単にまとめると、以下のような処理を実装する事になります。

  • (1) DescribeStacks で Stack の有無を確認して、ChangeSetType(CREATEUPDATE)を決定
  • (2) CreateChangeSet で変更セットを作成
  • (3) 処理が完了するまで DescribeChangeSet をポーリング
  • (4) ExecuteChangeSet で変更セットを実行
  • (5) 処理が完了するまで DescribeStacks をポーリング

(1) の DescribeStacks では該当するスタックが存在しない場合にもエラーとなってしまうため、他のエラーと区別するためにエラーメッセージが指定の文字列に合致するかどうかで判定しています。(AWS CLI や CDK と同様)

(3) では変更セットのステータスが CREATE_PENDINGCREATE_IN_PROGRESS 以外になるまでポーリングし、(5) ではスタックのステータスが xxx_IN_PROGRESS 以外になるまでポーリングすれば良さそうです。

また、(3) におけるポーリングの間隔は AWS CLI と CDK 共に 5秒毎となっていましたが、(5) の方は AWS CLI が 30秒で CDK は 5秒となっているようでした。

b_sample/deployer.ts(CloudFormation の API を使ったデプロイ処理の実装)
import { CloudFormationStackArtifact } from '@aws-cdk/cx-api'

import { 
    CloudFormationClient, 
    CreateChangeSetCommand, ExecuteChangeSetCommand, 
    DescribeChangeSetCommand, DescribeStacksCommand,
    Change, Stack
} from '@aws-sdk/client-cloudformation'

const MAX_RETRY = 100
const WAIT_TIME_CREATE = 5000
const WAIT_TIME_EXECUTE = 10000

const sleep = (ms: number) =>
    new Promise(resolve => setTimeout(resolve, ms))

const cf = new CloudFormationClient({})

type ChangeSetType = 'CREATE' | 'UPDATE'

export class CfDeployer {
    static async deploy(stackArtifact: CloudFormationStackArtifact): Promise<Stack | undefined> {
        const stackName = stackArtifact.stackName

        // (1)
        const changeSetType = await CfDeployer.selectChangeSetType(stackName)
        // (2)
        const r1 = await cf.send(new CreateChangeSetCommand({
            ChangeSetName: `ChangeSet-${Date.now()}`,
            StackName: stackName,
            TemplateBody: JSON.stringify(stackArtifact.template),
            ChangeSetType: changeSetType
        }))

        console.log(r1)

        const changeSetId = r1.Id

        if (!changeSetId) {
            throw new Error(`failed create ChangeSet: StackName=${stackName}`)
        }

        // (3)
        const cs = await CfDeployer.waitForCreate(changeSetId, stackName)

        if (cs.length < 1) {
            console.log('*** NO CHANGE')
            return CfDeployer.getStack(stackName)
        }
        // (4)
        const r2 = await cf.send(new ExecuteChangeSetCommand({
            ChangeSetName: changeSetId,
            StackName: stackName
        }))

        console.log(r2)
        // (5)
        return CfDeployer.waitForExecute(stackName)
    }

    static async getStackStatus(stackName: string): Promise<string> {
        const stack = await CfDeployer.getStack(stackName)
        return stack?.StackStatus ?? 'NONE'
    }

    static async getStack(stackName: string): Promise<Stack | undefined> {
        try {
            const r = await cf.send(new DescribeStacksCommand({
                StackName: stackName
            }))

            return r.Stacks?.[0]
        } catch(e) {
            // スタックが存在しなかった場合のエラー以外はそのまま throw
            if (e.Code !== 'ValidationError' || 
                e.message !== `Stack with id ${stackName} does not exist`) {
                throw e
            }
        }

        return undefined
    }
    // (3)
    private static async waitForCreate(changeSetId: string, stackName: string): Promise<Change[]> {

        for (let i = 0; i < MAX_RETRY; i++) {
            await sleep(WAIT_TIME_CREATE)

            const r = await cf.send(new DescribeChangeSetCommand({
                ChangeSetName: changeSetId,
                StackName: stackName
            }))

            if (r.Status === 'CREATE_PENDING' || 
                r.Status === 'CREATE_IN_PROGRESS') {

                continue
            }

            if (r.Status == 'CREATE_COMPLETE') {
                return r.Changes ?? []
            }

            if (r.Status == 'FAILED') {
                console.log(`*** failed: ${JSON.stringify(r)}`)
                return []
            }

            throw new Error(`failed create ChangeSet: ChangeSetId=${changeSetId}, StackName=${stackName}`)
        }

        throw new Error(`create ChangeSet timeout: ChangeSetId=${changeSetId}, StackName=${stackName}`)
    }
    // (5)
    private static async waitForExecute(stackName: string): Promise<Stack> {
        for (let i = 0; i < MAX_RETRY; i++) {
            await sleep(WAIT_TIME_EXECUTE)

            const stack = await CfDeployer.getStack(stackName)

            if (!stack) {
                throw new Error('not found')
            }

            if (!stack.StackStatus?.endsWith('_IN_PROGRESS')) {
                return stack
            }
        }

        throw new Error(`execute ChangeSet timeout: StackName=${stackName}`)
    }

    private static async selectChangeSetType(stackName: string): Promise<ChangeSetType> {

        const status = await CfDeployer.getStackStatus(stackName)

        return (status == 'NONE' || status == 'REVIEW_IN_PROGRESS') ? 
            'CREATE' : 'UPDATE'
    }
}

上記を使ってスタックをデプロイする処理は以下のようになります。

b_sample/deploy_b.ts
import { App } from '@aws-cdk/core'

import { CfDeployer } from './deployer'
import { SampleStack } from './stack'

const app = new App()
new SampleStack(app, 'Sample2Stack')

CfDeployer.deploy(app.synth().stacks[0])
    .then(r => console.log(r))
    .catch(err => console.error(err))
パッケージインストール例
> cd b_sample
> npm install -D ts-node typescript @types/node
・・・
> npm install @aws-cdk/aws-dynamodb @aws-sdk/client-cloudformation
・・・

実行結果は以下の通りです。 こちらも CloudFormation のスタックが登録され、DynamoDB のテーブルが問題なく作成されました。

実行結果
> npx ts-node deploy_b.ts

{
  '$metadata': {
    httpStatusCode: 200,
    requestId: '・・・',
    extendedRequestId: undefined,
    cfId: undefined,
    attempts: 1,
    totalRetryDelay: 0
  },
  Id: 'arn:aws:cloudformation:・・・:changeSet/ChangeSet-・・・',
  StackId: 'arn:aws:cloudformation:・・・:stack/Sample2Stack/・・・'
}
{
  '$metadata': {
    httpStatusCode: 200,
    requestId: '・・・',
    extendedRequestId: undefined,
    cfId: undefined,
    attempts: 1,
    totalRetryDelay: 0
  }
}
{
  StackId: 'arn:aws:cloudformation:・・・',
  StackName: 'Sample2Stack',
  ChangeSetId: 'arn:aws:cloudformation:・・・',
  Description: undefined,
  Parameters: undefined,
  ・・・
  StackStatus: 'CREATE_COMPLETE',
  StackStatusReason: undefined,
  DisableRollback: false,
  NotificationARNs: [],
  TimeoutInMinutes: undefined,
  Capabilities: undefined,
  Outputs: [
    {
      OutputKey: 'TableName',
      OutputValue: 'Sample2Stack-Items・・・',
      Description: undefined,
      ExportName: undefined
    }
  ],
  RoleARN: undefined,
  Tags: [],
  EnableTerminationProtection: false,
  ParentId: undefined,
  RootId: undefined,
  DriftInformation: { StackDriftStatus: 'NOT_CHECKED', LastCheckTimestamp: undefined }
}

Amplify AppSync Simulator を直接使ってマッピングテンプレートを検証

Amplify AppSync Simulator は、AWS Amplify CLI に含まれているモジュールで、AppSync をローカル環境で動作確認するためのものです。(AppSync の GraphQL を処理する Web サーバーが起動するようになっている)

ソースコードを見てみたところ、AppSync 用の処理を適用した GraphQL.jsGraphQLSchema を作り、これを graphql 関数で実行するようになっていました。

そこで試しに、Amplify AppSync Simulator の API を直接使う事で、Amplify CLI を使わず Web サーバーも起動せずに、リクエスマッピングテンプレートの処理結果を出力する Node.js スクリプトを作ってみました。

今回のソースは http://github.com/fits/try_samples/tree/master/blog/20210315/

準備

amplify-appsync-simulator モジュールをインストールしておきます。

インストール例
> npm install amplify-appsync-simulator

Lambda 用マッピングテンプレートの処理

今回は、Lambda 用のマッピングテンプレートを処理してみます。

AppSync では下記を設定する事で GraphQL の処理を定義するようになっています。

  • (1) GraphQL スキーマから GraphQL API を作成
  • (2) データソース(実際の処理を行う部分)を追加
  • (3) リゾルバー ※ を作成
 ※ リゾルバーは、GraphQL API とデータソースとの紐づけ、
    入出力データの加工をマッピングテンプレートとして設定するようになっています

マッピングテンプレートは、VTL(Apache Velocity Template Language)の形式でリクエストやレスポンスの内容を加工して JSON 文字列を作ります。

Amplify AppSync Simulator では AmplifyAppSyncSimulator クラスの init メソッドに対して、上記 (1) ~ (3) と defaultAuthenticationType を設定した AmplifyAppSyncSimulatorConfig を渡す事で、AppSync 用の処理を適用した GraphQLSchema を内部的に作るようになっており、それを schema で取得する事が可能です。

defaultAuthenticationType の設定は、今回のように GraphQLSchema を直接利用するケースでは使われませんが、設定自体は必須のようです。

また、Amplify AppSync Simulator が対応しているデータソースは今のところ下記 3タイプのようで、タイプ毎に用意されている DataLoader が処理を担うようになっています。

各 DataLoader の load メソッドの第一引数にリクエスマッピングテンプレートの処理結果がそのまま渡ってくるようになっています。

NONE データソースの利用

NONE タイプ用 NoneDataLoader の load メソッドは、単純に payload の値を処理結果として返すような実装になっています。

そこで、この load メソッドを上書きする事でリクエスマッピングテンプレートの結果を出力するようにしてみました。

なお、AmplifyAppSyncSimulator が作成した GraphQLSchema を graphql 関数で処理する際のコンテキストに requestAuthorizationModeappsyncErrors の項目が最低限必要でした。

defaultAuthenticationType は必須のようなので API_KEY を設定していますが、実際には使われないので API_KEY の値を設定したりする必要はありませんでした。

sample1.js
const { AmplifyAppSyncSimulator } = require('amplify-appsync-simulator')
const { graphql } = require('graphql')
// GraphQL スキーマ定義
const schema = `
    type Item {
        id: ID!
        value: Int!
    }

    input FindInput {
        id: ID!
    }

    type Query {
        find(input: FindInput!): Item
    }
`
// データソースの設定
const dataSources = [
    { type: 'NONE', name: 'ItemFunc' }
]
// リゾルバーの設定
const resolvers = [
    {
        kind: 'UNIT', 
        // GraphQL の Query で定義した find フィールドと ItemFunc データソースとのマッピング
        typeName: 'Query', fieldName: 'find',
        dataSourceName: 'ItemFunc'
        // リクエストマッピングテンプレートの設定(GraphQL クエリの input をそのまま payload に設定)
        requestMappingTemplate: `
            {
                "version": "2018-05-29",
                "operation": "Invoke",
                "payload": $utils.toJson($ctx.args.input)
            }
        `,
        // レスポンスマッピングテンプレートの設定
        responseMappingTemplate: '$utils.toJson($context.result)'
    }
]
// 設定
const config = {
    schema: { content: schema },
    resolvers,
    dataSources,
    appSync: {
        // 下記の設定は必須
        defaultAuthenticationType: {
            authenticationType: 'API_KEY'
        }
    }
}

const simulator = new AmplifyAppSyncSimulator()

simulator.init(config)

// ItemFunc(NoneDataLoader)の load メソッド上書き
simulator.getDataLoader('ItemFunc').load = (req) => {
    console.log(`*** load: ${JSON.stringify(req)}`)

    return { id: req.payload.id, value: 123 }
}
// AmplifyAppSyncSimulator用 GraphQL の実行コンテキスト
const ctx = {
    requestAuthorizationMode: 'API_KEY',
    appsyncErrors: []
}
// GraphQL クエリ
const q = `
    query FindItem($id: ID!) {
        find(input: {id: $id}) {
            id
            value
        }
    }
`

const run = async () => {
    // GraphQL クエリ実行
    const r = await graphql(simulator.schema, q, null, ctx, {id: 'id1'})
    console.log(JSON.stringify(r))
}

run().catch(err => console.error(err))

実行結果は以下の通り。 リクエスマッピングテンプレートの処理結果が出力されています。

実行結果
> node sample1.js
*** load: {"version":"2018-05-29","operation":"Invoke","payload":{"id":"id1"}}
{"data":{"find":{"id":"id1","value":123}}}

AWS_LAMBDA データソースの利用

AWS_LAMBDA タイプの場合は、データソース設定の invoke へ設定した関数が呼び出されるようになっているので ※、これを使って Lambda の関数ハンドラーを直接実行する事もできます。

 ※ リクエストマッピングテンプレート処理結果の
    payload の値が引数として与えられるようになっています
lambda_func.js(関数ハンドラー)
exports.handler = async (event) => {
    console.log(`*** handler: ${JSON.stringify(event)}`)

    return { id: event.id, value: 234 }
}
sample2.js
const { AmplifyAppSyncSimulator } = require('amplify-appsync-simulator')
const { graphql } = require('graphql')

const schema = `
    ・・・
`

const dataSources = [
    // invoke へ lambda_func.js の handler を設定
    { type: 'AWS_LAMBDA', name: 'ItemFunc', invoke: require('./lambda_func.js').handler }
]

const resolvers = [
    ・・・
]

const config = {
    schema: { content: schema },
    resolvers,
    dataSources,
    appSync: {
        defaultAuthenticationType: {
            authenticationType: 'API_KEY'
        }
    }
}

const simulator = new AmplifyAppSyncSimulator()

simulator.init(config)

const ctx = {
    requestAuthorizationMode: 'API_KEY',
    appsyncErrors: []
}

const q = `
    query FindItem($id: ID!) {
        find(input: {id: $id}) {
            id
            value
        }
    }
`

const run = async () => {
    const r = await graphql(simulator.schema, q, null, ctx, {id: 'id2'})
    console.log(JSON.stringify(r))
}

run().catch(err => console.error(err))
実行結果
> node sample2.js
*** handler: {"id":"id2"}
{"data":{"find":{"id":"id2","value":234}}}

AWS Lambda のランタイム API サーバーを自作して関数ハンドラーをローカル実行

AWS Lambda では、(Lambda 関数の)ランタイムがランタイム API(ランタイムインターフェース)から呼び出しイベントを受け取って、関数ハンドラーを実行し、その結果をランタイム API へ返すような流れで処理が実施されているようです。

そこで、自作のランタイム API サーバーを使って、Go と Node.js でそれぞれ実装した AWS Lambda の関数ハンドラーを(Docker 等を使わずに)ローカル実行してみました。

ソースは http://github.com/fits/try_samples/tree/master/blog/20210211/

ランタイム API サーバーの実装

以下のようなオープンソースのランタイム API サーバーが提供されており、自作する必要は無かったりするのですが。

今回は、fastify を使ってランタイム API サーバーを実装してみました。

AWS Lambda ランタイム API のページから OpenAPI 仕様(2018-06-01)を入手できるので、このスキーマ定義を参考に必要最小限の処理だけを実装します。

まず、ランタイム API として下記を実装する事になります。(ただし、今回の用途では (d) を使いません)

  • (a) 次の呼び出し : GET /runtime/invocation/next
  • (b) 呼び出しレスポンス : POST /runtime/invocation/{AwsRequestId}/response
  • (c) 呼び出しエラー : POST /runtime/invocation/{AwsRequestId}/error
  • (d) 初期化エラー : POST /runtime/init/error

(b) ~ (d) は成功時に 202 ステータスコードを返す事になります。

(b) や (c) における AwsRequestId の値は、(a) のレスポンスヘッダーで通知する事になり、(a) のレスポンスヘッダーで最低限必要な項目は下記でした。(aws-lambda-go の場合)

  • Lambda-Runtime-Aws-Request-Id (AwsRequestId の値)
  • Lambda-Runtime-Deadline-Ms (実行期限)

一方で、ランタイム側は以下のような手順でランタイム API を呼び出すようになっているようです。

  • (1) (a) を GET してレスポンスが返ってくるのを待つ
  • (2) (a) からレスポンス(呼び出しイベント)が返ってくると関数ハンドラーを呼び出す
  • (3) 関数ハンドラーの呼び出し結果を (b) へ POST、エラーが発生した場合は (c) へ POST
  • (4) (1) へ戻る

つまり、(a) で固定の内容をすぐに返すような実装にしてしまうと、(1) ~ (4) が繰り返されてしまい不都合が生じます。

そこで、下記では /invoke を呼び出した際に (a) のレスポンスを返すようにしています。

また、Go と Node.js の関数ハンドラーを同時に扱うために、(a) が呼び出された際の reply を replies へ溜めておいて、forEach で処理するようにしています。

runtime-api-server/server.js (ランタイム API サーバー)
const logger = false
const fastify = require('fastify')({ logger })

const RUNTIME_PATH = '/2018-06-01/runtime'
const TIMEOUT = 5 * 1000

const port = process.env['SERVER_PORT'] ? 
    parseInt(process.env['SERVER_PORT']) : 8080

const replies = []

fastify.post('/invoke', (req, reply) => {
    const data = req.body

    console.log(`*** invoke: ${JSON.stringify(data)}`)

    replies.splice(0).forEach(r => {
        const deadline = Date.now() + TIMEOUT
        // (a) のレスポンスを返す
        r.code(200)
            .header('Lambda-Runtime-Aws-Request-Id', data['request-id'])
            .header('Lambda-Runtime-Deadline-Ms', deadline.toString())
            .header('Lambda-Runtime-Trace-Id', data['trace-id'])
            .header('Lambda-Runtime-Invoked-Function-Arn', data['function-arn'])
            .send(data.body)
    })

    reply.code(200).send({})
})

// Runtime API の実装

// (a) 次の呼び出し
fastify.get(`${RUNTIME_PATH}/invocation/next`, (req, reply) => {
    console.log('*** next')
    console.log(req.body)

    replies.push(reply)
})
// (b) 呼び出しレスポンス
fastify.post(`${RUNTIME_PATH}/invocation/:id/response`, (req, reply) => {
    console.log(`*** response: id = ${req.params['id']}`)
    console.log(req.body)

    reply.code(202).send({ status: '' })
})
// (c) 呼び出しエラー
fastify.post(`${RUNTIME_PATH}/invocation/:id/error`, (req, reply) => {
    console.log(`*** error: id = ${req.params['id']}`)
    console.log(req.body)

    reply.code(202).send({ status: '' })
})
// (d) 初期化エラー
fastify.post(`${RUNTIME_PATH}/init/error`, (req, reply) => {
    console.log('*** init error')
    console.log(req.body)

    reply.code(202).send({ status: '' })
})

fastify.listen(port)
    .then(r => console.log(`started: ${r}`))
    .catch(err => console.error(err))

実行

動作確認のために実行しておきます。

> cd runtime-api-server
> node server.js
started: http://127.0.0.1:8080

Go による関数ハンドラー

Go の場合、aws-lambda-go にランタイム API とやり取りを行うランタイムの処理が実装されています。

そのため、aws-lambda-go を使って実装した関数ハンドラーをローカル環境でビルドして実行するだけです。

接続するランタイム API サーバーのアドレスは AWS_LAMBDA_RUNTIME_API 環境変数で設定します。

ここでは、以下の関数ハンドラーを使用します。

sample_go/app.go (関数ハンドラー)
package main

import (
    "context"
    "fmt"

    "github.com/aws/aws-lambda-go/lambda"
    "github.com/aws/aws-lambda-go/lambdacontext"
)

func handler(ctx context.Context, event interface{}) (string, error) {
    c, _ := lambdacontext.FromContext(ctx)

    fmt.Println("*** call handler")
    fmt.Printf("event = %#v\n", event)
    fmt.Printf("context = %#v\n", c)

    return "sample-go", nil
}

func main() {
    lambda.Start(handler)
}

動作確認1

AWS_LAMBDA_RUNTIME_API 環境変数にランタイム API サーバーのアドレスを設定し、app.go をビルドして実行します。

ビルドと実行
> set AWS_LAMBDA_RUNTIME_API=localhost:8080
> cd sample_go
> go build app.go
> app

ランタイム API サーバー側の出力内容は下記のようになり、app.go から (a) の API が呼び出されている事を確認できます。

server.js(ランタイム API サーバー)の出力内容1
> node server.js
started: http://127.0.0.1:8080
*** next
null

/invoke を呼び出して、関数ハンドラーを実行してみます。

invoke の呼び出し1
$ curl -s -XPOST -H "Content-Type: application/json" http://localhost:8080/invoke -d '{"request-id":"a1", "function-arn":"sample", "body":{"name":"abc","value":1}}'
{}

app.go の出力内容は以下のようになり、関数ハンドラーの実行を確認できました。

app.go の出力内容
> app
*** call handler
event = map[string]interface {}{"name":"abc", "value":1}
context = &lambdacontext.LambdaContext{AwsRequestID:"a1", InvokedFunctionArn:"sample", Identity:lambdacontext.CognitoIdentity{CognitoIdentityID:"", CognitoIdentityPoolID:""}, ClientContext:lambdacontext.ClientContext{Client:lambdacontext.ClientApplication{InstallationID:"", AppTitle:"", AppVersionCode:"", AppPackageName:""}, Env:map[string]string(nil), Custom:map[string]string(nil)}}

ランタイム API サーバー側は以下のように、app.go からのレスポンスを受け取っている事が確認できます。

server.js(ランタイム API サーバー)の出力内容2
> node server.js
started: http://127.0.0.1:8080
・・・
*** invoke: {"request-id":"a1","function-arn":"sample","body":{"name":"abc","value":1}}
*** response: id = a1
sample-go
*** next
null

続いて、エラー時の挙動を確認するために下記を実施します。

invoke の呼び出し2
$ curl -s -XPOST -H "Content-Type: application/json" http://localhost:8080/invoke -d '{"request-id":"b2"}'
{}

app.go には何も出力されず、ランタイム API サーバー側は以下のようになり、エラーの内容を受け取っています。

server.js(ランタイム API サーバー)の出力内容3
> node server.js
started: http://127.0.0.1:8080
・・・
*** invoke: {"request-id":"b2"}
*** error: id = b2
{
  errorMessage: 'unexpected end of JSON input',
  errorType: 'SyntaxError'
}
*** next
null

Node.js による関数ハンドラー

Node.js の場合、通常は関数ハンドラーそのものを実装するだけなので、Go とは違ってランタイムが別途必要となります。

Node.js 用のランタイムとしては下記が提供されており、ネイティブコードなどを使った本格的な作りとなっています。(本番利用を想定していると思われる)

今回はこれを使わず、簡易的なランタイムを node-fetch を使って自作してみました。

node_runtime/runtime.js (Node.js 用の簡易ランタイム)
const fetch = require('node-fetch')

const runtimeUrl = `http://${process.env["AWS_LAMBDA_RUNTIME_API"]}/2018-06-01/runtime/invocation`

const appRoot = process.cwd()
const [moduleName, handlerName] = process.argv[2].split('.')

const app = require(`${appRoot}/${moduleName}`)

const envData = {
    functionVersion: process.env["AWS_LAMBDA_FUNCTION_VERSION"],
    functionName: process.env["AWS_LAMBDA_FUNCTION_NAME"],
    memoryLimitInMB: process.env["AWS_LAMBDA_FUNCTION_MEMORY_SIZE"],
    logGroupName: process.env["AWS_LAMBDA_LOG_GROUP_NAME"],
    logStreamName: process.env["AWS_LAMBDA_LOG_STREAM_NAME"],
}

const run = async () => {
    // (a) の API を呼び出す
    const nextRes = await fetch(`${runtimeUrl}/next`)

    const deadline = parseInt(nextRes.headers.get('lambda-runtime-deadline-ms'))

    const context = Object.assign(
        {
            getRemainingTimeInMillis: () => deadline - Date.now(),
            awsRequestId: nextRes.headers.get('lambda-runtime-aws-request-id'),
            invokedFunctionArn: 
                nextRes.headers.get('lambda-runtime-invoked-function-arn'),
            identity: {},
            clientContext: {},
        }, 
        envData
    )

    try {
        const event = await nextRes.json()
        // 関数ハンドラーの呼び出し
        const res = await app[handlerName](event, context)
        console.log(`* handler result: ${res}`)
        // (b) の API を呼び出す
        await fetch(
            `${runtimeUrl}/${context.awsRequestId}/response`, 
            {method: 'POST', body: res}
        )

    } catch (e) {
        // (c) の API を呼び出す
        await fetch(
            `${runtimeUrl}/${context.awsRequestId}/error`, 
            {method: 'POST', body: JSON.stringify({type: e.type, message: e.message})}
        )
    }
}

const loop = async () => {
    while (true) {
        await run()
    }
}

loop().catch(err => console.error(err))

ここでは、下記の関数ハンドラーを使う事にします。

sample_node/app.js (関数ハンドラー)
exports.handler = async (event, context) => {
    console.log('*** call handler')
    console.log(`event = ${JSON.stringify(event)}`)
    console.log(`context = ${JSON.stringify(context)}`)
    console.log(`remaining time = ${context.getRemainingTimeInMillis()}`)

    return 'sample-node'
}

動作確認2

runtime.js を使って app.js の handler を呼び出すように実行します。

実行
> set AWS_LAMBDA_RUNTIME_API=localhost:8080
> cd sample_node
> node ../node_runtime/runtime.js app.handler

ランタイム API サーバーの next が呼び出されました。

server.js(ランタイム API サーバー)の出力内容4
> node server.js
started: http://127.0.0.1:8080
・・・
*** next
null

/invoke を呼び出します。

invoke の呼び出し3
$ curl -s -XPOST -H "Content-Type: application/json" http://localhost:8080/invoke -d '{"request-id":"c3", "function-arn":"sample2", "body":{"name":"def","value":2}}'
{}

runtime.js + app.js の出力内容は以下のようになり、関数ハンドラーが呼び出されました。

runtime.js + app.js の出力内容
> node ../node_runtime/runtime.js app.handler
*** call handler
event = {"name":"def","value":2}
context = {"awsRequestId":"c3","invokedFunctionArn":"sample2","identity":{},"clientContext":{}}
remaining time = 4945
* handler result: sample-node

ランタイム API サーバーの出力も以下のようになり、問題は無さそうです。

server.js(ランタイム API サーバー)の出力内容5
> node server.js
started: http://127.0.0.1:8080
・・・
*** invoke: {"request-id":"c3","function-arn":"sample2","body":{"name":"def","value":2}}
・・・
*** response: id = c3
sample-node
*** next
null

エラーが発生するように /invoke を呼び出します。

invoke の呼び出し4
$ curl -s -XPOST -H "Content-Type: application/json" http://localhost:8080/invoke -d '{"request-id":"d4"}'
{}

こちらも問題は無さそうです。

server.js(ランタイム API サーバー)の出力内容6
> node server.js
started: http://127.0.0.1:8080
・・・
*** invoke: {"request-id":"d4"}
・・・
*** error: id = d4
{"type":"invalid-json","message":"invalid json response body at http://localhost:8080/2018-06-01/runtime/invocation/next reason: Unexpected end of JSON input"}
*** next
null