要約処理が重いので、キューをかますなどして、複数アクセスに耐えられる設計とする

[sakamomo554101]

単純にFlaskで組むと、余裕で耐えられなくなるので、前段にキューをかますなどの検討が必要。

[sakamomo554101]

https://docs.microsoft.com/ja-jp/azure/architecture/patterns/async-request-reply

上記のようなHTTPの非同期処理パターンが良い気がするな。 ようは、

• 構成としては、APIゲートウェイ + タスクを積むQueue + 推論処理をするコンテナのようなイメージ
• まず、推論リクエストを投げて、即座にレスポンスを返す（トリガーの発火）
• バックエンドでは推論処理を実施し、クライアント側ではポーリングして、状態を監視（ここが負荷にならないかどうか）
  • 推論内容の識別のために、ID発行が必要
• 推論が完了したら、結果を返す

[sakamomo554101]

FastAPI + Queue（Kafkaとか？）で作ってみようかな。

[sakamomo554101]

APIとして必要なのは、下記かな。

• 要約リクエストのAPI（POST : 本文をリクエストボディで渡す）
• 要約結果を取得するためのAPI（GET : リクエストIDをパラメーターで渡す）
• 要約結果を補正するためのAPI（POST : リクエストIDをパラメーターで渡し、リクエストボディに補正した要約内容を入れる）

https://fastapi.tiangolo.com/ja/tutorial/body/

[sakamomo554101]

リクエスト時の必要なパラメーターはjsonフォーマットで渡す感じにするか。

[sakamomo554101]

設計として、良さそうなのが以下のどちらか。（多分、後者だけど、一旦は簡易的に前者でやってもいいかな）

• 推論リクエスト時には、一時的なuuidを返して、uuidと本文情報を格納したテーブルのidを紐づける
• 推論リクエスト時に、本文情報のDB追加処理を行い、テーブルのidを返す

[sakamomo554101]

いや、API gateway側でDBにアクセスしない方がいいか。 そしたら、一時的なuuidを作成して、Queue側に渡す感じにしよう。

[sakamomo554101]

SQS互換なElasticMQを使うか、Kafkaを使うか。 SQSだと、メッセージサイズの上限が256kb（拡張ライブラリであげられる）か。 https://github.com/softwaremill/elasticmq

[sakamomo554101]

kafka使うかな。 https://hub.docker.com/r/bitnami/kafka/

自前でコンテナイメージ作ってもいいけど、上記のようにすでにあるイメージを使う。

[sakamomo554101]

https://aiokafka.readthedocs.io/en/stable/examples/serialize_and_compress.html?highlight=json#serialization-and-compression

jsonをメッセージとして渡す場合は、serializer/deserializerを作れば良さそう。

[sakamomo554101]

https://aiokafka.readthedocs.io/en/stable/examples/batch_produce.html send_batchを使っても良い（一旦、愚直なforループにする）

[sakamomo554101]

https://docs.python.org/ja/3.6/library/asyncio-task.html async def と asyncio.coroutineについて（両者は異なる）

[sakamomo554101]

======================================================================
ERROR: test_send_and_receive_body (test.test_queue_helper.TestQueueHelper)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/workspace/queue/test/test_queue_helper.py", line 24, in test_send_and_receive_body
    queue_helper.send_body_into_queue(body_text=body_text, config=self.__config)
  File "/workspace/queue/queue_helper.py", line 17, in send_body_into_queue
    producer.produce(messages=[message])
  File "/workspace/queue/queue_client.py", line 58, in produce
    loop.run_until_complete(self.__produce(loop=loop, messages=messages))
  File "/usr/local/lib/python3.8/asyncio/base_events.py", line 616, in run_until_complete
    return future.result()
  File "/workspace/queue/queue_client.py", line 71, in __produce
    await producer.send(self._config.optional_param["topic_name"], message)
  File "/usr/local/lib/python3.8/site-packages/aiokafka/producer/producer.py", line 420, in send
    await self.client._wait_on_metadata(topic)
  File "/usr/local/lib/python3.8/site-packages/aiokafka/client.py", line 628, in _wait_on_metadata
    raise UnknownTopicOrPartitionError()
kafka.errors.UnknownTopicOrPartitionError: [Error 3] UnknownTopicOrPartitionError

あれ、事前にtopicは作らないとだっけ？

[sakamomo554101]

https://stackoverflow.com/questions/26021541/how-to-programmatically-create-a-topic-in-apache-kafka-using-python/55524560

aiokafkaだと、topic作れないっけ、、ドキュメント探したが、わからん。

[sakamomo554101]

======================================================================
FAIL: test_send_and_receive_body (test.test_queue_helper.TestQueueHelper)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/workspace/queue/test/test_queue_helper.py", line 18, in setUp
    queue_helper.create_kafka_topics(topics=[self.__config.optional_param["topic_name"]], cliend_id="test", config=self.__config)
  File "/workspace/queue/queue_helper.py", line 32, in create_kafka_topics
    client = KafkaAdminClient(
  File "/usr/local/lib/python3.8/site-packages/kafka/admin/client.py", line 218, in __init__
    self._refresh_controller_id()
  File "/usr/local/lib/python3.8/site-packages/kafka/admin/client.py", line 278, in _refresh_controller_id
    controller_version = self._client.check_version(controller_id, timeout=(self.config['api_version_auto_timeout_ms'] / 1000))
  File "/usr/local/lib/python3.8/site-packages/kafka/client_async.py", line 901, in check_version
    self._maybe_connect(try_node)
  File "/usr/local/lib/python3.8/site-packages/kafka/client_async.py", line 372, in _maybe_connect
    assert broker, 'Broker id %s not in current metadata' % (node_id,)
AssertionError: Broker id -1 not in current metadata

うーん、topicが上手く作れん。 kafkaのシェルスクリプト経由で作れるかを確認する。

[sakamomo554101]

kafkaのコンテナにはいって、下記を実行すると、ちゃんとトピックは作られてそう。

I have no name!@c8b57d77ce69:/$ kafka-topics.sh --zookeeper youyaku_ai_queue_zookeeper --create --replication-factor 1 --partitions 1 --topic test1
Created topic test1.
I have no name!@c8b57d77ce69:/$ kafka-topics.sh --list --zookeeper youyaku_ai_queue_zookeeper
test1

[sakamomo554101]

https://kafka-python.readthedocs.io/en/master/apidoc/KafkaAdminClient.html#kafka.KafkaAdminClient KafkaAdminClientの取得処理で失敗してる

[sakamomo554101]

KAFKA_CFG_ADVERTISED_LISTENERSの設定が誤っていて、コンテナ経由でのqueueのアクセスが失敗していた。

[sakamomo554101]

api gateway部分は正常系は動作するようになった。 summarizer部分をQueueからデータ取得して、動作させるようにする。

[sakamomo554101]

https://matsuand.github.io/docs.docker.jp.onthefly/compose/environment-variables/

docker-composeだと、変数定義をenvファイルに設定できる。 これを利用して、DBのホスト名とかを設定しようかな。

[sakamomo554101]

8 で対応