はじめに AWS SDK for Goがありますが、現在（2021年7月24日）のところ、AWS SDK for Go v2のAmazon Comporehendのコードサンプルがなかったので、とりあえず自分で作ってみることにしました。 個人的にはPythonを使用することが多いのですが、Go言語も業務で使用することがあるので、Go言語でもチャレンジしてみました。 ※他の言語や他のサービスのコードサンプルは以下から参照できます。 Developer Guide/Code Examples awsdocs/aws-doc-sdk-examples Amazon Comprehendを使用するための事前準備 Go言語のプログラムをLinuxで実行してAmazon Comprehendのエンティティ認識のバッチを呼び出して結果をダウンロードして表示させることを考えます。オンラインのAPI呼び出しではなく、バッチ処理にしたのはAPI呼び出しだと文字数に（少し）きつい制限があるためです。エンティティ認識の処理だと、UTF-8のエンコーディングで5,000バイトが処理の上限です。 Amazon Comprehendを使用するには事前準備として権限の付与とロールの定義が必要です。 コマンド（Goのソースコードをコンパイルしたもの）を実行するユーザ（aws configureで設定したアクセスキーやシークレットキーを持つユーザ）にAmazon Comprehendの呼び出しを許可する コマンドを実行するユーザにAWS のサービスにロールを渡すアクセス権限を付与する。（こちらを参考にしてください。） Amazon Comprehendに渡すロールを定義する。（こちらを参考にしてください。） 処理の流れ 処理対象ファイルをS3にアップロードする。 Amazon Comprehendを呼び出す（エンティティ認識のバッチ処理） Amazon Comprehendのジョブの完了を一定間隔で確認する。 処理結果のtar.gzファイルをダウンロードする。 tar.gzファイルに含まれる結果のJSONからScoreの高い順にエンティティのキーワードと種別、スコアの値を表示する（デフォルトでは20個） ソースコード（Go言語） ソースコードは以下のとおりです。（GitHubにも置いてあります。） 43行目の「roleArn string = "arn:aws:iam::123456789012:role/comprehend-access-role"」のところは、事前準備で定義した「Amazon Comprehendに渡すロール」のARNを指定します。 main.go package main import ( "archive/tar" "bufio" "compress/gzip" "context" "encoding/json" "flag" "fmt" "io" "net/url" "os" "sort" "time" "github.com/aws/aws-sdk-go-v2/config" "github.com/aws/aws-sdk-go-v2/service/comprehend" comprehend_types "github.com/aws/aws-sdk-go-v2/service/comprehend/types" "github.com/aws/aws-sdk-go-v2/service/s3" ) type Entity struct { BeginOffset int `json:"BeginOffset"` EndOffset int `json:"EndOffset"` Score float64 `json:"Score"` Text string `json:"Text"` Type string `json:"Type"` } type JsonOutput struct { File string `json:"File"` Entities []Entity `json:"Entities"` } const ( resultFileName string = "output" bufferSize int = 1024 * 1024 interval int = 10 timeoutCount int = 100 jobName string = "sample-entities-detection-job" // ここのロールは置き換えること roleArn string = "arn:aws:iam::123456789012:role/comprehend-access-role" ) func main() { pBucketName := flag.String("bucket", "testcomprehend-tn", "Bucket to put a content on.") pPrefixName := flag.String("prefix", "comprehend/", "Prefix to store a content file") pContentFileName := flag.String("file", "content.txt", "The content file") pLimitNumber := flag.Int("limit", 20, "The limit to display keywords") flag.Parse() fmt.Println("START!!!!") fmt.Println("Bucket:", *pBucketName) fmt.Println("Prefix:", *pPrefixName) fmt.Println("Limit:", *pLimitNumber) fmt.Println("File:", *pContentFileName) cfg, err := config.LoadDefaultConfig(context.TODO()) if err != nil { panic("configuration error, " + err.Error()) } s3Client := s3.NewFromConfig(cfg) comprehendClient := comprehend.NewFromConfig(cfg) // 処理対象ファイルをS3にアップロードする file, err := os.Open(*pContentFileName) if err != nil { fmt.Println("Unable to open file " + *pContentFileName) return } defer file.Close() objectName := *pPrefixName + "input/" + *pContentFileName fmt.Println("Object Name: " + objectName) input := &s3.PutObjectInput{ Bucket: pBucketName, Key: &objectName, Body: file, } _, err = s3Client.PutObject(context.TODO(), input) if err != nil { fmt.Println("Got error uploading file:") fmt.Println(err) return } // Amazon Comprehend 呼び出し inputS3Uri := "s3://" + *pBucketName + "/" + objectName fmt.Println("InputS3URI: " + inputS3Uri) inputConfig := &comprehend_types.InputDataConfig{ S3Uri: &inputS3Uri, InputFormat: comprehend_types.InputFormatOneDocPerFile, } outputS3Uri := "s3://" + *pBucketName + "/" + *pPrefixName + "output/" fmt.Println("OutputS3URI: " + outputS3Uri) outputConfig := &comprehend_types.OutputDataConfig{ S3Uri: &outputS3Uri, } roleArnForInput := roleArn jobNameForInput := jobName jobInput := &comprehend.StartEntitiesDetectionJobInput{ DataAccessRoleArn: &roleArnForInput, InputDataConfig: inputConfig, LanguageCode: comprehend_types.LanguageCodeJa, OutputDataConfig: outputConfig, JobName: &jobNameForInput, } out, err := comprehendClient.StartEntitiesDetectionJob(context.TODO(), jobInput) if err != nil { fmt.Println("Starting an entities detection job Error:") fmt.Println(err) return } jobId := *out.JobId fmt.Println("Job ID: " + jobId) // Amazon Comprehend 完了確認 describeJobInput := &comprehend.DescribeEntitiesDetectionJobInput{ JobId: &jobId, } var outDesc *comprehend.DescribeEntitiesDetectionJobOutput for i := 0; i < timeoutCount; i++ { fmt.Println("In Progress...") time.Sleep(time.Duration(interval) * time.Second) outDesc, err = comprehendClient.DescribeEntitiesDetectionJob(context.TODO(), describeJobInput) if err != nil { fmt.Println("Getting a status of the entities detection job Error:") fmt.Println(err) return } if outDesc.EntitiesDetectionJobProperties.JobStatus == comprehend_types.JobStatusCompleted { fmt.Println("Job Completed.") break } } if outDesc.EntitiesDetectionJobProperties.JobStatus != comprehend_types.JobStatusCompleted { fmt.Println("Job Timeout.") return } // Amazon Comprehend 結果ダウンロード downloadS3Uri := *outDesc.EntitiesDetectionJobProperties.OutputDataConfig.S3Uri fmt.Println("Output S3 URI: " + downloadS3Uri) parsedUri, err := url.Parse(downloadS3Uri) if err != nil { fmt.Println("URI Parse Error:") fmt.Println(err) return } downloadObjectName := parsedUri.Path[1:] fmt.Println("Download Object Name: " + downloadObjectName) getObjectInput := &s3.GetObjectInput{ Bucket: pBucketName, Key: &downloadObjectName, } outGetObject, err := s3Client.GetObject(context.TODO(), getObjectInput) if err != nil { fmt.Println("Getting S3 object Error:") fmt.Println(err) return } defer outGetObject.Body.Close() // tar.gzファイルから結果のJSONを取り出す gzipReader, err := gzip.NewReader(outGetObject.Body) if err != nil { fmt.Println("Reading a gzip file Error:") fmt.Println(err) return } defer gzipReader.Close() tarfileReader := tar.NewReader(gzipReader) var jsonOutput JsonOutput for { tarfileHeader, err := tarfileReader.Next() if err == io.EOF { break } if err != nil { fmt.Println("Reading a tar file Error:") fmt.Println(err) return } if tarfileHeader.Name == resultFileName { jsonReader := bufio.NewReaderSize(tarfileReader, bufferSize) for { // TODO: isPrefixを見て、行がバッファに対して長すぎる場合の処理を行なう line, _, err := jsonReader.ReadLine() if err == io.EOF { break } if err != nil { fmt.Println("Reading a line Error:") fmt.Println(err) return } json.Unmarshal([]byte(line), &jsonOutput) fmt.Println() fmt.Println("File: " + jsonOutput.File) // Score順にソートする entities := jsonOutput.Entities sort.Slice(entities, func(i, j int) bool { return entities[i].Score > entities[j].Score }) // Amazon Comprehend 結果表示 for i := 0; i < *pLimitNumber; i++ { fmt.Printf("%s (%s): %f\n", entities[i].Text, entities[i].Type, entities[i].Score) } } } } fmt.Println() fmt.Println() fmt.Println("END!!!!") } go buildでビルドできます。また、以下のコマンドでオプション（引数）を確認できます。-fileの指定が解析したいファイルになります。デフォルトはcontent.txtという名前です。UTF-8で文章を格納してください。 ./mycomprehend -h Usage of ./mycomprehend: -bucket string Bucket to put a content on. (default "testcomprehend-tn") -file string The content file (default "content.txt") -limit int The limit to display keywords (default 20) -prefix string Prefix to store a content file (default "comprehend/") 実行結果 content.txtについてエンティティ認識の処理をした結果は以下のようになりました。 （前略） File: content.txt 2020年12月 (DATE): 0.999014 3時間 (QUANTITY): 0.997277 Linux Academy (ORGANIZATION): 0.995822 3年 (QUANTITY): 0.995796 180分 (QUANTITY): 0.995555 2回 (QUANTITY): 0.995218 33,000円 (QUANTITY): 0.993613 https://qiita.com/takanattie/items/7dd188ce14a2a5b9ef14 (OTHER): 0.990118 1度 (QUANTITY): 0.988592 3ヶ月半 (QUANTITY): 0.988081 AWS (ORGANIZATION): 0.987514 AWS (ORGANIZATION): 0.980833 U.S. (ORGANIZATION): 0.976798 AWS (ORGANIZATION): 0.976790 AWS (ORGANIZATION): 0.975587 3年間 (QUANTITY): 0.975035 2周 (QUANTITY): 0.972928 Linux Academy (ORGANIZATION): 0.966450 AWS (ORGANIZATION): 0.966190 Qiita Advent Calendar 2020 (TITLE): 0.962282 認識されたエンティティのタイプ（ORGANIZATIONなど）の情報もあるので、そのタイプでフィルタリングすることもできると思います。 補足 上記のコードは、サンプルとなるように書いたので不充分な点があります。特に以下の点は改善の余地があると考えています。 コマンドの戻り値（Exit Code）を設定する S3にAmazon Comprehendの結果が格納されるので、それをトリガとしてLambdaを起動して結果を取り出す、などの処理を行なうようにする。（イベントドリブンにする） また、今回はエンティティ認識だけでしたが、キーフレーズ抽出なども同じように実行できるはずです（コードの修正は必要になりますが）。APIの定義はソースコードとしてGitHubにもありますし、APIのドキュメントもありますので、そこを読み解けばコードをかけるはずですが、やはり一連の流れを示したサンプルコードが欲しいところです。 以上。

はじめに AWS SDK for Goがありますが、現在（2021年7月24日）のところ、AWS SDK for Go v2のAmazon Comporehendのコードサンプルがなかったので、とりあえず自分で作ってみることにしました。 個人的にはPythonを使用することが多いのですが、Go言語も業務で使用することがあるので、Go言語でもチャレンジしてみました。 ※他の言語や他のサービスのコードサンプルは以下から参照できます。 Developer Guide/Code Examples awsdocs/aws-doc-sdk-examples Amazon Comprehendを使用するための事前準備 Go言語のプログラムをLinuxで実行してAmazon Comprehendのエンティティ認識のバッチを呼び出して結果をダウンロードして表示させることを考えます。オンラインのAPI呼び出しではなく、バッチ処理にしたのはAPI呼び出しだと文字数に（少し）きつい制限があるためです。エンティティ認識の処理だと、UTF-8のエンコーディングで5,000バイトが処理の上限です。 Amazon Comprehendを使用するには事前準備として権限の付与とロールの定義が必要です。 コマンド（Goのソースコードをコンパイルしたもの）を実行するユーザ（aws configureで設定したアクセスキーやシークレットキーを持つユーザ）にAmazon Comprehendの呼び出しを許可する コマンドを実行するユーザにAWS のサービスにロールを渡すアクセス権限を付与する。（こちらを参考にしてください。） Amazon Comprehendに渡すロールを定義する。（こちらを参考にしてください。） 処理の流れ 処理対象ファイルをS3にアップロードする。 Amazon Comprehendを呼び出す（エンティティ認識のバッチ処理） Amazon Comprehendのジョブの完了を一定間隔で確認する。 処理結果のtar.gzファイルをダウンロードする。 tar.gzファイルに含まれる結果のJSONからScoreの高い順にエンティティのキーワードと種別、スコアの値を表示する（デフォルトでは20個） ソースコード（Go言語） ソースコードは以下のとおりです。（GitHubにも置いてあります。） 43行目の「roleArn string = "arn:aws:iam::123456789012:role/comprehend-access-role"」のところは、事前準備で定義した「Amazon Comprehendに渡すロール」のARNを指定します。 main.go package main import ( "archive/tar" "bufio" "compress/gzip" "context" "encoding/json" "flag" "fmt" "io" "net/url" "os" "sort" "time" "github.com/aws/aws-sdk-go-v2/config" "github.com/aws/aws-sdk-go-v2/service/comprehend" comprehend_types "github.com/aws/aws-sdk-go-v2/service/comprehend/types" "github.com/aws/aws-sdk-go-v2/service/s3" ) type Entity struct { BeginOffset int `json:"BeginOffset"` EndOffset int `json:"EndOffset"` Score float64 `json:"Score"` Text string `json:"Text"` Type string `json:"Type"` } type JsonOutput struct { File string `json:"File"` Entities []Entity `json:"Entities"` } const ( resultFileName string = "output" bufferSize int = 1024 * 1024 interval int = 10 timeoutCount int = 100 jobName string = "sample-entities-detection-job" // ここのロールは置き換えること roleArn string = "arn:aws:iam::123456789012:role/comprehend-access-role" ) func main() { pBucketName := flag.String("bucket", "testcomprehend-tn", "Bucket to put a content on.") pPrefixName := flag.String("prefix", "comprehend/", "Prefix to store a content file") pContentFileName := flag.String("file", "content.txt", "The content file") pLimitNumber := flag.Int("limit", 20, "The limit to display keywords") flag.Parse() fmt.Println("START!!!!") fmt.Println("Bucket:", *pBucketName) fmt.Println("Prefix:", *pPrefixName) fmt.Println("Limit:", *pLimitNumber) fmt.Println("File:", *pContentFileName) cfg, err := config.LoadDefaultConfig(context.TODO()) if err != nil { panic("configuration error, " + err.Error()) } s3Client := s3.NewFromConfig(cfg) comprehendClient := comprehend.NewFromConfig(cfg) // 処理対象ファイルをS3にアップロードする file, err := os.Open(*pContentFileName) if err != nil { fmt.Println("Unable to open file " + *pContentFileName) return } defer file.Close() objectName := *pPrefixName + "input/" + *pContentFileName fmt.Println("Object Name: " + objectName) input := &s3.PutObjectInput{ Bucket: pBucketName, Key: &objectName, Body: file, } _, err = s3Client.PutObject(context.TODO(), input) if err != nil { fmt.Println("Got error uploading file:") fmt.Println(err) return } // Amazon Comprehend 呼び出し inputS3Uri := "s3://" + *pBucketName + "/" + objectName fmt.Println("InputS3URI: " + inputS3Uri) inputConfig := &comprehend_types.InputDataConfig{ S3Uri: &inputS3Uri, InputFormat: comprehend_types.InputFormatOneDocPerFile, } outputS3Uri := "s3://" + *pBucketName + "/" + *pPrefixName + "output/" fmt.Println("OutputS3URI: " + outputS3Uri) outputConfig := &comprehend_types.OutputDataConfig{ S3Uri: &outputS3Uri, } roleArnForInput := roleArn jobNameForInput := jobName jobInput := &comprehend.StartEntitiesDetectionJobInput{ DataAccessRoleArn: &roleArnForInput, InputDataConfig: inputConfig, LanguageCode: comprehend_types.LanguageCodeJa, OutputDataConfig: outputConfig, JobName: &jobNameForInput, } out, err := comprehendClient.StartEntitiesDetectionJob(context.TODO(), jobInput) if err != nil { fmt.Println("Starting an entities detection job Error:") fmt.Println(err) return } jobId := *out.JobId fmt.Println("Job ID: " + jobId) // Amazon Comprehend 完了確認 describeJobInput := &comprehend.DescribeEntitiesDetectionJobInput{ JobId: &jobId, } var outDesc *comprehend.DescribeEntitiesDetectionJobOutput for i := 0; i < timeoutCount; i++ { fmt.Println("In Progress...") time.Sleep(time.Duration(interval) * time.Second) outDesc, err = comprehendClient.DescribeEntitiesDetectionJob(context.TODO(), describeJobInput) if err != nil { fmt.Println("Getting a status of the entities detection job Error:") fmt.Println(err) return } if outDesc.EntitiesDetectionJobProperties.JobStatus == comprehend_types.JobStatusCompleted { fmt.Println("Job Completed.") break } } if outDesc.EntitiesDetectionJobProperties.JobStatus != comprehend_types.JobStatusCompleted { fmt.Println("Job Timeout.") return } // Amazon Comprehend 結果ダウンロード downloadS3Uri := *outDesc.EntitiesDetectionJobProperties.OutputDataConfig.S3Uri fmt.Println("Output S3 URI: " + downloadS3Uri) parsedUri, err := url.Parse(downloadS3Uri) if err != nil { fmt.Println("URI Parse Error:") fmt.Println(err) return } downloadObjectName := parsedUri.Path[1:] fmt.Println("Download Object Name: " + downloadObjectName) getObjectInput := &s3.GetObjectInput{ Bucket: pBucketName, Key: &downloadObjectName, } outGetObject, err := s3Client.GetObject(context.TODO(), getObjectInput) if err != nil { fmt.Println("Getting S3 object Error:") fmt.Println(err) return } defer outGetObject.Body.Close() // tar.gzファイルから結果のJSONを取り出す gzipReader, err := gzip.NewReader(outGetObject.Body) if err != nil { fmt.Println("Reading a gzip file Error:") fmt.Println(err) return } defer gzipReader.Close() tarfileReader := tar.NewReader(gzipReader) var jsonOutput JsonOutput for { tarfileHeader, err := tarfileReader.Next() if err == io.EOF { break } if err != nil { fmt.Println("Reading a tar file Error:") fmt.Println(err) return } if tarfileHeader.Name == resultFileName { jsonReader := bufio.NewReaderSize(tarfileReader, bufferSize) for { // TODO: isPrefixを見て、行がバッファに対して長すぎる場合の処理を行なう line, _, err := jsonReader.ReadLine() if err == io.EOF { break } if err != nil { fmt.Println("Reading a line Error:") fmt.Println(err) return } json.Unmarshal([]byte(line), &jsonOutput) fmt.Println() fmt.Println("File: " + jsonOutput.File) // Score順にソートする entities := jsonOutput.Entities sort.Slice(entities, func(i, j int) bool { return entities[i].Score > entities[j].Score }) // Amazon Comprehend 結果表示 for i := 0; i < *pLimitNumber; i++ { fmt.Printf("%s (%s): %f\n", entities[i].Text, entities[i].Type, entities[i].Score) } } } } fmt.Println() fmt.Println() fmt.Println("END!!!!") } go buildでビルドできます。また、以下のコマンドでオプション（引数）を確認できます。-fileの指定が解析したいファイルになります。デフォルトはcontent.txtという名前です。UTF-8で文章を格納してください。 ./mycomprehend -h Usage of ./mycomprehend: -bucket string Bucket to put a content on. (default "testcomprehend-tn") -file string The content file (default "content.txt") -limit int The limit to display keywords (default 20) -prefix string Prefix to store a content file (default "comprehend/") 実行結果 content.txtについてエンティティ認識の処理をした結果は以下のようになりました。 （前略） File: content.txt 2020年12月 (DATE): 0.999014 3時間 (QUANTITY): 0.997277 Linux Academy (ORGANIZATION): 0.995822 3年 (QUANTITY): 0.995796 180分 (QUANTITY): 0.995555 2回 (QUANTITY): 0.995218 33,000円 (QUANTITY): 0.993613 https://qiita.com/takanattie/items/7dd188ce14a2a5b9ef14 (OTHER): 0.990118 1度 (QUANTITY): 0.988592 3ヶ月半 (QUANTITY): 0.988081 AWS (ORGANIZATION): 0.987514 AWS (ORGANIZATION): 0.980833 U.S. (ORGANIZATION): 0.976798 AWS (ORGANIZATION): 0.976790 AWS (ORGANIZATION): 0.975587 3年間 (QUANTITY): 0.975035 2周 (QUANTITY): 0.972928 Linux Academy (ORGANIZATION): 0.966450 AWS (ORGANIZATION): 0.966190 Qiita Advent Calendar 2020 (TITLE): 0.962282 認識されたエンティティのタイプ（ORGANIZATIONなど）の情報もあるので、そのタイプでフィルタリングすることもできると思います。 補足 上記のコードは、サンプルとなるように書いたので不充分な点があります。特に以下の点は改善の余地があると考えています。 コマンドの戻り値（Exit Code）を設定する S3にAmazon Comprehendの結果が格納されるので、それをトリガとしてLambdaを起動して結果を取り出す、などの処理を行なうようにする。（イベントドリブンにする） また、今回はエンティティ認識だけでしたが、キーフレーズ抽出なども同じように実行できるはずです（コードの修正は必要になりますが）。APIの定義はソースコードとしてGitHubにもありますし、APIのドキュメントもありますので、そこを読み解けばコードをかけるはずですが、やはり一連の流れを示したサンプルコードが欲しいところです。 以上。

はじめに 海外ドラマやYouTubeの英会話チャネルを見ているときに「あっ、この表現かっこいい。覚えよう。」と思ったけれど、数日後には「あれ、何を覚えようとしていたっけ。」となったりしませんか？そんなあなた(自分)のために、覚えたい単語、フレーズをしつこく教えてくれる、あなただけの英会話講師(Parrot Tutor)をAlexaに召喚します。 Parrot Tutorはどのようなスキル？ Student: "Alexa, talk to parrot tutor." Alexa: "Welcome, I can help you remember words or phrases. Please ask me to add what you want to remember." Student: "Add a word, righteous." Alexa: "Righteous. Is it ok?" Student: "Yes." Alexa: "I remember the word, righteous, which means 正義の." Student: "I want to learn 3 words." Alexa: "Righteous, 正義の. Excellent, ..." 手順 スキルの作成 Alexa developer consoleにログインし、新しいスキルを作成します。デフォルトの言語は英語(米国)を選択します。バックエンドリソースは、ユーザー定義のプロビジョニングを選択します。 呼び出し名 呼び出し名を変更しておきます。 モデルの作成 Alexa developer console上で対話モデルを作成します。 en-US.json { "interactionModel": { "languageModel": { "invocationName": "parrot tutor", "intents": [ { "name": "AMAZON.CancelIntent", "samples": [] }, { "name": "AMAZON.HelpIntent", "samples": [] }, { "name": "AMAZON.StopIntent", "samples": [] }, { "name": "AMAZON.NavigateHomeIntent", "samples": [] }, { "name": "AMAZON.FallbackIntent", "samples": [] }, { "name": "AddWordIntent", "slots": [ { "name": "wordSlot", "type": "AMAZON.SearchQuery" } ], "samples": [ "add a word {wordSlot}", "append a word {wordSlot}", "store a word {wordSlot}", "save a word {wordSlot}", "remember a word {wordSlot}" ] }, { "name": "LearnIntent", "slots": [ { "name": "countSlot", "type": "AMAZON.NUMBER" }, { "name": "headingTypeSlot", "type": "PluralHeadingType" } ], "samples": [ "I want to learn {headingTypeSlot}", "learn {headingTypeSlot}", "speak {headingTypeSlot}", "tell me {headingTypeSlot}", "tell me {countSlot} {headingTypeSlot}", "speak {countSlot} {headingTypeSlot}", "learn {countSlot} {headingTypeSlot}", "I want to learn {countSlot} {headingTypeSlot}" ] }, { "name": "AddPhraseIntent", "slots": [ { "name": "phraseSlot", "type": "AMAZON.SearchQuery" } ], "samples": [ "remember a phrase {phraseSlot}", "save a phrase {phraseSlot}", "store a phrase {phraseSlot}", "append a phrase {phraseSlot}", "add a phrase {phraseSlot}" ] } ], "types": [ { "name": "PluralHeadingType", "values": [ { "name": { "value": "phrases" } }, { "name": { "value": "words" } } ] } ] }, "dialog": { "intents": [ { "name": "AddWordIntent", "confirmationRequired": true, "prompts": { "confirmation": "Confirm.Intent.321024708461" }, "slots": [ { "name": "wordSlot", "type": "AMAZON.SearchQuery", "confirmationRequired": false, "elicitationRequired": false, "prompts": {} } ] }, { "name": "AddPhraseIntent", "confirmationRequired": true, "prompts": { "confirmation": "Confirm.Intent.334458378973" }, "slots": [ { "name": "phraseSlot", "type": "AMAZON.SearchQuery", "confirmationRequired": false, "elicitationRequired": false, "prompts": {} } ] } ], "delegationStrategy": "ALWAYS" }, "prompts": [ { "id": "Confirm.Intent.321024708461", "variations": [ { "type": "PlainText", "value": "{wordSlot} . Is it ok?" } ] }, { "id": "Confirm.Intent.334458378973", "variations": [ { "type": "PlainText", "value": "{phraseSlot} . Is it ok?" } ] } ] } } Lambda環境の構築 Alexa-hostedスキルを作成すると、AWSアカウントなくAlexa開発者コンソールだけで作成、編集、公開が完結しますが、Translate APIなどAWSの機能を使いたいので、サービスのエンドポイントは独自に構築したAWS Lambdaでホストします。 Lambda Layerパッケージの作成 $ mkdir parrot && cd parrot $ pyenv local 3.8.10 $ python -m venv venv $ . venv/bin/activate $ mkdir -p layer/python && cd layer $ pip install -t python -r python/requirements.txt $ zip -r python.zip python python/requirements.txt ask-sdk-core==1.11.0 ask-sdk-dynamodb-persistence-adapter==1.15.0 boto3==1.9.216 Lambda関数の作成 AWSマネジメントコンソールからLambdaサービスにアクセスして、Lambda関数を作成します。 レイヤーパッケージのアップロード 先程作成したpython.zip をアップロードしてカスタムレイヤーを作成します。 レイヤーの追加 作成したLambda関数を選択して、レイヤーを追加します。 トリガーの設定 Alexa developer consoleのAlexaスキル一覧で、「スキルIDをコピー」を選択します。Lambdaサービスで、Lambda関数にトリガーを追加します。スキルID検証を有効にして、コピーしたスキルIDをペーストします。 lambda_function.pyの実装 lambda_function.py # -*- coding: utf-8 -*- import boto3 import decimal import learning_db import logging import random import ask_sdk_core.utils as ask_utils from ask_sdk_core.skill_builder import SkillBuilder from ask_sdk_core.dispatch_components import AbstractRequestHandler from ask_sdk_core.dispatch_components import AbstractExceptionHandler from ask_sdk_core.handler_input import HandlerInput from ask_sdk_model import Response from datetime import datetime logger = logging.getLogger(__name__) logger.setLevel(logging.INFO) class LaunchRequestHandler(AbstractRequestHandler): """Handler for Skill Launch.""" def can_handle(self, handler_input): return ask_utils.is_request_type("LaunchRequest")(handler_input) def handle(self, handler_input): speak_output = "Welcome, I can help you remember words or phrases. Please ask me to add what you want to remember." return ( handler_input.response_builder .speak(speak_output) .ask(speak_output) .response ) class AddIntentHandler(AbstractRequestHandler): def __init__(self, heading_type): self.heading_type = heading_type """Handler for Add Item Intent.""" def can_handle(self, handler_input): return ask_utils.is_intent_name(f"Add{self.heading_type}Intent")(handler_input) def handle(self, handler_input): global db, translate user_id = ask_utils.get_user_id(handler_input) locale = ask_utils.get_locale(handler_input) heading = ask_utils.get_slot_value(handler_input, f"{self.heading_type.lower()}Slot") response = translate.translate_text( Text=heading, SourceLanguageCode="en", TargetLanguageCode="ja" ) translation = response['TranslatedText'] db.put_item(user_id, heading, self.heading_type.upper(), translation=translation, sequence=0, next_sequence=0) translation_output = f"<voice name=\"Joanna\"><lang xml:lang=\"ja-JP\">{translation}</lang></voice>" speak_output = f"I remember the {self.heading_type}, {heading}, which means {translation_output}." reprompt = "Add another item or learn to memorize?" return ( handler_input.response_builder .speak(speak_output) .ask(reprompt) .response ) class LearnIntentHandler(AbstractRequestHandler): """Handler for Learn Intent.""" def can_handle(self, handler_input): return ask_utils.is_intent_name("LearnIntent")(handler_input) def handle(self, handler_input): global db user_id = ask_utils.get_user_id(handler_input) count_slot = ask_utils.get_slot_value(handler_input, "countSlot") count = int(count_slot) if count_slot is not None else 3 heading_type_slot = ask_utils.get_slot_value(handler_input, "headingTypeSlot") heading_type = "word" if heading_type_slot == "words" else "phrase" heading_type_keyword = heading_type.upper() max_learned_count = db.get_max_learned_count(user_id, heading_type_keyword) filtered_items = [] for learned_count in range(max_learned_count + 1): items = db.query_item(user_id, heading_type_keyword, learned_count) filtered_items.extend(random.sample(items, len(items))) if len(filtered_items) > count: break total = min(count, len(filtered_items)) outputs = [] for item in filtered_items[0:total]: heading = item['heading'] translation = f"<voice name=\"Joanna\"><lang xml:lang=\"ja-JP\">{item['translation']}</lang></voice>" outputs.append(f"{heading}. {translation}") if item['learned_count'] >= max_learned_count: max_learned_count = item['learned_count'] + 1 db.increment_learned_count(user_id, heading) if total > 0: db.update_max_learned_count(user_id, heading_type_keyword, max_learned_count) plural = "s" if total > 1 else "" if total != count: speak_output = f"You've added only {total} {heading_type}{plural}. " + ". ".join(outputs) else: speak_output = f"Let's learn {total} {heading_type}{plural}. " + ". ".join(outputs) else: speak_output = f"You haven't added any {heading_type} yet." reprompt = "Add another item or learn to memorize?" return ( handler_input.response_builder .speak(speak_output) .ask(reprompt) .response ) class HelpIntentHandler(AbstractRequestHandler): """Handler for Help Intent.""" def can_handle(self, handler_input): return ask_utils.is_intent_name("AMAZON.HelpIntent")(handler_input) def handle(self, handler_input): speak_output = "You can say add a word something or add a phrase something or learn words or learn phrases. What would you like to do?" return ( handler_input.response_builder .speak(speak_output) .ask(speak_output) .response ) class CancelOrStopIntentHandler(AbstractRequestHandler): """Single handler for Cancel and Stop Intent.""" def can_handle(self, handler_input): return (ask_utils.is_intent_name("AMAZON.CancelIntent")(handler_input) or ask_utils.is_intent_name("AMAZON.StopIntent")(handler_input)) def handle(self, handler_input): speak_output = "Goodbye!" return ( handler_input.response_builder .speak(speak_output) .response ) class FallbackIntentHandler(AbstractRequestHandler): """Single handler for Fallback Intent.""" def can_handle(self, handler_input): return ask_utils.is_intent_name("AMAZON.FallbackIntent")(handler_input) def handle(self, handler_input): logger.info("In FallbackIntentHandler") speech = "Hmm, I'm not sure. You can say Add, Learn, or Help. What would you like to do?" reprompt = "I didn't catch that. What can I help you with?" return handler_input.response_builder.speak(speech).ask(reprompt).response class SessionEndedRequestHandler(AbstractRequestHandler): """Handler for Session End.""" def can_handle(self, handler_input): return ask_utils.is_request_type("SessionEndedRequest")(handler_input) def handle(self, handler_input): # Clean up logic here. return handler_input.response_builder.response class IntentReflectorHandler(AbstractRequestHandler): def can_handle(self, handler_input): return ask_utils.is_request_type("IntentRequest")(handler_input) def handle(self, handler_input): intent_name = ask_utils.get_intent_name(handler_input) speak_output = "You just triggered " + intent_name + "." return ( handler_input.response_builder .speak(speak_output) .response ) class CatchAllExceptionHandler(AbstractExceptionHandler): def can_handle(self, handler_input, exception): return True def handle(self, handler_input, exception): logger.error(exception, exc_info=True) speak_output = "Sorry, I had trouble doing what you asked. Please try again." return ( handler_input.response_builder .speak(speak_output) .ask(speak_output) .response ) db = learning_db.LearningDB() sb = SkillBuilder() translate = boto3.client(service_name='translate', region_name='us-west-1', use_ssl=True) sb.add_request_handler(LaunchRequestHandler()) sb.add_request_handler(AddIntentHandler("Word")) sb.add_request_handler(AddIntentHandler("Phrase")) sb.add_request_handler(LearnIntentHandler()) sb.add_request_handler(HelpIntentHandler()) sb.add_request_handler(CancelOrStopIntentHandler()) sb.add_request_handler(FallbackIntentHandler()) sb.add_request_handler(SessionEndedRequestHandler()) # make sure IntentReflectorHandler is last so it doesn't override your custom intent handlers sb.add_request_handler(IntentReflectorHandler()) sb.add_exception_handler(CatchAllExceptionHandler()) lambda_handler = sb.lambda_handler() DynamoDB フレーズや単語は、DynamoDBに格納します。まず、Lambda関数からDynamoDBにアクセスできるようにIAMでロールにポリシーを追加します。Lambda関数のロールはLambda関数を作成した際に自動的に生成されています。 グローバルセカンダリキーにuser_idとlearned_count(学習回数)を用いることで、学習回数でクエリできるようにします。ユーザーには学習回数の少ないものから指定個数を復習対象として提示します。 learning_db.py import boto3 import json import decimal from boto3.dynamodb.conditions import Key, Attr from botocore.exceptions import ClientError from datetime import datetime class LearningDB(): def __init__(self): self.create_textbook_table() self.create_learning_record_table() def create_textbook_table(self): TABLE_NAME = "ask.parrot_tutor.learning_db.textbook_table" dynamodb = boto3.resource('dynamodb') try: self.textbook_table = dynamodb.create_table( TableName=TABLE_NAME, KeySchema=[ { 'AttributeName': 'user_id', 'KeyType': 'HASH' }, { 'AttributeName': 'heading', 'KeyType': 'RANGE' } ], AttributeDefinitions=[ { 'AttributeName': 'user_id', 'AttributeType': 'S' }, { 'AttributeName': 'heading', 'AttributeType': 'S' }, { 'AttributeName': 'learned_count', 'AttributeType': 'N' } ], ProvisionedThroughput={ 'ReadCapacityUnits': 5, 'WriteCapacityUnits': 5 }, GlobalSecondaryIndexes=[ { 'IndexName': 'learned_count_index', 'KeySchema': [ { 'AttributeName': 'user_id', 'KeyType': 'HASH' }, { 'AttributeName': 'learned_count', 'KeyType': 'RANGE' } ], 'Projection': { 'ProjectionType': 'INCLUDE', 'NonKeyAttributes': [ 'heading_type', 'learned_at', 'translation' ] }, 'ProvisionedThroughput': { 'ReadCapacityUnits': 5, 'WriteCapacityUnits': 5 } } ] ) except ClientError as e: self.textbook_table = dynamodb.Table(TABLE_NAME) def create_learning_record_table(self): TABLE_NAME = "ask.parrot_tutor.learning_db.learning_record_table" dynamodb = boto3.resource('dynamodb') try: self.learning_record_table = dynamodb.create_table( TableName=TABLE_NAME, KeySchema=[ { 'AttributeName': 'user_id', 'KeyType': 'HASH' }, { 'AttributeName': 'heading_type', 'KeyType': 'RANGE' }, ], AttributeDefinitions=[ { 'AttributeName': 'user_id', 'AttributeType': 'S' }, { 'AttributeName': 'heading_type', 'AttributeType': 'S' } ], ProvisionedThroughput={ 'ReadCapacityUnits': 5, 'WriteCapacityUnits': 5 } ) except ClientError as e: self.learning_record_table = dynamodb.Table(TABLE_NAME) def put_item(self, user_id, heading, heading_type, translation="", sequence=0, next_sequence=0): ts = decimal.Decimal(datetime.now().timestamp()) item = { 'user_id': user_id, 'heading': heading, 'created_at': ts, 'updated_at': ts, 'learned_at': ts, 'learned_count': 0, 'heading_type': heading_type, 'translation': translation, 'sequence': sequence, 'next_sequence': next_sequence } self.textbook_table.put_item( Item=item ) def query_item(self, user_id, heading_type, max_learned_count): response = self.textbook_table.query( IndexName="learned_count_index", KeyConditionExpression=Key('user_id').eq(user_id) & Key('learned_count').eq(max_learned_count), FilterExpression=Attr('heading_type').contains(heading_type) ) return response['Items'] if response else [] def get_max_learned_count(self, user_id, heading_type): try: response = self.learning_record_table.get_item( Key={ 'user_id': user_id, 'heading_type': heading_type } ) if 'Item' in response: return int(response['Item']['max_learned_count']) else: self.learning_record_table.put_item( Item={ 'user_id': user_id, 'heading_type': heading_type, 'max_learned_count': 0 }) return 0 except ClientError as e: return 0 def increment_learned_count(self, user_id, heading): ts = decimal.Decimal(datetime.now().timestamp()) response = self.textbook_table.update_item( Key={ 'user_id': user_id, 'heading': heading }, UpdateExpression='set learned_at=:learned_at, learned_count=learned_count + :one', ExpressionAttributeValues={ ':learned_at': ts, ':one': 1 }, ReturnValues="UPDATED_NEW" ) def update_max_learned_count(self, user_id, heading_type, count): response = self.learning_record_table.update_item( Key={ 'user_id': user_id, 'heading_type': heading_type }, UpdateExpression='set max_learned_count=:max_learned_count', ExpressionAttributeValues={ ':max_learned_count': count }, ReturnValues="UPDATED_NEW" ) Translate 登録した単語やフレーズは、Translate APIを使用して翻訳します。DynamoDBと同様にIAMでロールにTranslate APIのポリシーを登録します。 Translate APIの読み出しは簡単で、次のようにboto3.clientのインスタンスを作り、translate_textを呼び出すだけです。 import boto3 translate = boto3.client(service_name='translate', region_name='us-west-1', use_ssl=True) response = translate.translate_text( Text=heading, SourceLanguageCode="en", TargetLanguageCode="ja" ) translation = response['TranslatedText'] 日本語の翻訳部分は次のようなSSMLで記述して英語と日本語を両方発話できるようにします。 f"<voice name=\"Joanna\"><lang xml:lang=\"ja-JP\">{translation}</lang></voice>" エンドポイントの設定 最後に、AWS Lambda関数のARNをスキルのエンドポイントに設定します。 Alexa Parrot Tutorスキルのエンドポイントを変更する。 テスト Alexa developer consoleのシミュレータで、動かしてみます。 まとめ 単語やフレーズを登録してAlexaに繰り返し話してもらうスキルを作成しました。登録の削除ができなかったり、翻訳が適当だったりと改良点はまだまだあるので、もっと効果的な英会話アプリに育ててから公開予定です。 参考コード Github ParrotTutor

aws-cliではec2 describe-instancesを利用してEC2インスタンスの一覧を取得できます。 ここではインスタンス一覧取得時にインスタンスの起動時刻・停止時刻でフィルタするサンプルとなります。 先に結果を書いてしましますが、インスタンスの開始時刻についてはフィルタするのに問題はないのですが。 停止時刻の方についてはマネージメントコンソールやCLIでAPIを利用してシャットダウンした場合は停止時刻を後から確認できますが、OSからシャットダウンをすると現状、ec2 describe-instancesで停止時刻を取得する方法はなさそう？ 利用するaws-cliのバージョン 今回利用しているaws-cliのバージョンについては下記となります aws-cli/2.2.22 Python/3.8.8 Windows/10 exe/AMD64 prompt/off ec2 describe-instancesのドキュメント describe-instances 起動時刻でフィルタする describe-instancesドキュメントを確認すると、EC2インスタンスの起動時刻はLaunchTimeを参照すればよさそうな事がわかります。 またfiltersオプションでlaunch-timeフィルタ項目としても提供されている事もわかります。 filtersオプションでフィルタする まずはfiltersオプションで提供されているlaunch-timeを利用してフィルタしてみます。 LaunchTime項目はyyyy-MM-ddThh:mm:ss+0000の形式となっている。 現在、起動しているインスタンスで、起動時刻が2021年1月となっているインスタンスを取得する場合は下記の用に記載する ※あくまで2021年1月に起動なので、2021年1月以降ではない点に注意 # 現在起動していて2021年1月に起動したインスタンスを取得 aws ec2 describe-instances --filters "Name=instance-state-name,Values=running" "Name=launch-time,Values=2021-01*" filtersオプションはワイルドカードが利用できますが、これだけだと◯月☓日以降といった細かい条件を指定しての抽出できません。 queryオプションでフィルタする もうすこし細かくフィルタしたい場合は、queryオプションを利用してフィルタします。 下記のようにすると、2021年1月1日以降に起動して、現在も起動しているインスタンスの一覧を抽出できます。 # 2021年1月1日以降に起動して現在も起動しているインスタンスを取得 # queryオプション Instances[?LaunchTime>=`2021-01-01`][] # queryオプションで出力する項目をtag-Name,InstanceId,LaunchTime,Stateを指定 aws ec2 describe-instances --filters "Name=instance-state-name,Values=running" --query 'Reservations[].Instances[?LaunchTime>=`2021-01-01`][].{Name:Tags[?Key==`Name`]|[0].Value,InstanceId:InstanceId,LaunchTime:LaunchTime,State:State.Name}' 停止時刻でフィルタする describe-instances aws-cliのドキュメントを確認した所、インスタンスを停止した時刻を直接持つ項目はなさそうです。 色々と項目を確認してみると、StateTransitionReasonにAPIからインスタンスを停止した場合に下記形式で停止時間が表示されるようなので、今回はこれを利用してフィルタしてみます。 User initiated (yyyy-MM-dd hh:mm:ss GMT) ※注意点 StateTransitionReason項目とは別にStateReasonという項目が存在し。 これはAPIドキュメントのStateReasonの値が入っていそうです。 StateReason APIドキュメントを確認すると APIを利用してシャットダウンをした場合は Client.UserInitiatedShutdown OSからshutdown -hでシャットダウンした場合は　Client.InstanceInitiatedShutdown とシャットダウンの方法によりコードが分けられています。 これはちょっとStateReasonによって、StateTransitionReasonの出力形式も別になってきそう？　な気配がするので確認してみます。 2021年7月現在実際に試してみると下記のようになります。（マネージメントコンソールでもaws-cliでも同様でした） Client.UserInitiatedShutdown(APIでシャットダウン)ではStateTransitionReasonに時刻が表示される。 Client.InstanceInitiatedShutdown（OSからシャットダウン）ではStateTransitionReasonに時刻が表示されない。 ここでは試していませんがスポットインスタンスで入札価格をスポット価格が上回って停止した場合はまた別のメッセージが表示されそうし、AWS基盤障害等で停止した場合は更に別のメッセージが入りそうな気はします。 ですので今回紹介する方法で停止時刻をフィルタできるのは、あくまで、APIでインスタンスを停止したインスタンスのみとなりそうです。 この点、注意が必要となります。 # 2021年1月1日以前に停止して現在も停止しちえるインスタンスを取得（APIを利用して停止したインスタンスのみ） aws ec2 describe-instances --filters "Name=instance-state-name,Values=stopped" --query 'Reservations[].Instances[?StateTransitionReason<=`User initiated (2021-01-01 00:00:00 GMT)`][].{InstanceId:InstanceId,LaunchTime:LaunchTime,StateTransitionReason:StateTransitionReason,Name:Tags[?Key==`Name`].Value|[0],State:State.Name,Message:StateReason.Message}' 総評 起動開始時刻は項目として用意されているので容易に取得することができます。 停止時刻はAWS側で項目として用意されていないようなので、複数項目を組み合わせて抽出できるかとも思いましたが。 APIを利用せずにOSからシャットダウンしたケース等ではec2 describe-instancesでは停止時刻を参照できる項目はなさそうに思います。

エラー発生 AWS CLI上で下記のエラーが発生。 aws kms create-key You must specify a region. You can also configure your region by running "aws configure". (リージョンを指定しなければいけません。"aws configure"を実行することで、リージョンを設定することができます。) 上記エラー内容をググると、下記の記事発見。 AWSクライアントでリージョン(region)を指定する リージョンを指定してみるが、下記のエラーが発生。。 aws kms create-key --region us-east-1 Unable to locate credentials. You can configure credentials by running "aws configure". (認証情報が見つかりません。"aws configure"を実行することで、認証情報を設定することができます。) 解決方法 aws configureの設定をすることで解決できます。 AWS-CLIの初期設定のメモの記事のaws-cliの設定を参考にしました。 $ aws configure --profile user1 AWS Access Key ID [None]: {アクセスキー(各自)} AWS Secret Access Key [None]: {シークレットアクセスキー(各自)} Default region name [None]: ap-northeast-1 Default output format [None]: json AWS Access Key IDとAWS Secret Access KeyはIAMの該当ユーザー>認証情報>アクセスキーから確認可能です。 参照:https://github.com/aws/aws-cli/issues/915

前回の投稿 AWS上にサーバーを作る（その02） Amazon VPCの設定 さて、だいぶ期間が開いてしまいましたが、前回はネットワーク周りの名称と内容の確認までは終わっていたと思います。 ここからは、ようやく初めてAWSのネットワークの設定に入ります。 まずは「ルートユーザー」または「IAMユーザー」でログインします。 Amazon VPCの作成 前回調べた通り、「VPCの作成」自体は無料のようですので、早速作成を開始しちゃいます。 1.VPCウィザードを起動 ステップ 1: VPC 設定の選択 今回は外部へサーバーを公開しない想定ですが、 「プライベートのサブネットのみで、ハードウェア VPN アクセスを持つ VPC」 を選択してセキュアなネットワークを構築する、または 「1 個のパブリックサブネットを持つ VPC」 を選択してセキュリティグループで接続元IPアドレスを限定する方法、どちらでもよいと思います。 前者はそもそも外部に通信をさらされないため安全ですが、拠点間VPN接続は別途金額がかかってしまうようですので、今回はケチって後者を採用しました。 ステップ 2: 1 個のパブリックサブネットを持つ VPC IPアドレスなどを指定していきます。 「アベイラビリティーゾーン」の項目ですが、簡単に説明するとどこのデータセンターに仮想マシンを設置するかということのようです。 現在、AWSのアジアパシフィック東京リージョンは、 ap-northeast-1a ap-northeast-1c ap-northeast-1d の３か所が公開されていて、このうちのどこに仮想マシンを設置するかを指定します。 今後、冗長化を考える際はアベイラビリティーゾーンの1aと1dのような感じで別の場所にプライマリー、セカンダリーサーバーの設置場所を指定すると良さそうです。 それ以外は、デフォルトで進めてしまいましょう。 次回の内容 そろそろ仮想マシン作成できる？？ （まだ引っ張る） To be continued...

VS codeの拡張機能を使って、 Amazon Linux2にログインするやり方があったので、ご紹介致します。 これにより、 ディレクトリなど、かなり見やすいので、おすすめです。 ※なお、今回は既にEC2があることを前提としています。 インストール VS codeで remote ssh などで検索し、 Remote - SSH をインストールします。 configの設定 コンソールを開いて .sshのconfigに設定を入れます （.sshの中にconfigがない場合は新規で作成してください） /Users/{ユーザー名}/.ssh/config Host { 任意の名称 例: test } HostName { 作成したEC2のパブリック IPv4 アドレス 例: 11.11.11.11 } IdentityFile { キーペアのパス 例: /Users/{ユーザー名}/.ssh/test.pem } User ec2-user VS codeで起動する 以下を押下します。 Connect to Hostを押下します。 すると、先ほど configに設定した Host名が表示されるので、 それを押下します。 これで VS codeで Amazon Linux2 にログインできました。 openからディレクトリを開き、ターミナルを開けば、 かなり 見やすくなりました！！！！

はじめに 個人開発したアプリケーションをAWS EC2で本番環境にデプロイをしました。 その後、追加機能としてチャットルーム機能を実装。roomsテーブルを追加しました。 なので再びEC2にデプロイ。 EC2本番環境で新しい機能、rooms/newページでチャットルーム作成をして、rooms/createのページアクセスすると以下のエラーが発生。 room/createのページにアクセスできませんでした。（チャットルームが作成できない状況） ポイントとして ・開発環境(localhost:3000)とherokuでは問題なくrooms/newからrooms/createにアクセスできる。でもEC2ではできない。　ということ。 今回の原因として、roomsテーブル作成後、新しくnameカラムを追加しました。結果、そのカラム追加方法に原因があったようです。そのためEC2のDBにnameカラム反映されなかったみたいでした。内容を詳しく後述します。 まず、使用している開発環境及び追加したテーブル&カラムは以下の通りです。 開発環境 Ruby 2.6.5 Ruby on Rails 6.0.3.7 MySQL 5.6.51 Github 2.30.1 heroku 7.54.0 AWS EC2 自動デプロイツール : Capistrano Webサーバー : nginx APサーバー :Unicorn 追加したテーブル rooms Column Type Options name strings null: false has_many: users through: :room_users has_many: room_users room_users（中間テーブル） Column Type Options room references foreign_key: true user references foreign_key: true has_many: users has_many: rooms 新しくテーブル、そのあとにカラムを追加後、EC２上に再度デプロイ。本番環境でチャットルーム新規作成ページにてルームを作成しようとすると、 と表示されてしまった。 このエラーを解決するのに丸一日以上費やしたので、今後のためにも解決した方法をこの記事に載せることにしました。 やっかいだったのが、AWSの情報ってググってもまだ情報量が少ないので答えになかなか辿り着けませんでした。 そこで僕は、エラー解決でどうしようもないときたまにお世話になっている、MENTAを使うことにしました。 原因 EC2上で tail -f production.log でログを調べると ActiveModel::UnknownAttributeError in RoomsController#create unknown attribute 'name' for Room. ログの中にこいつが見つかりました。前述しましたがこれが今回の原因です。 nameカラムがEC2上のデータベースにないですよ、という意味。 試したこと まず、試したこととして、以下のコマンドで、Nginxの読み直しと再起動をしました。
 sudo systemctl reload nginx sudo systemctl restart nginx その後、以下のコマンドで、プロセスを確認。 ps aux | grep unicorn kill プロセス番号 でプロセスをkillしました。 再度デプロイしてみます。 bundle exec cap production deploy もう一度アクセスしてみましたが、変わりませんでした。 前述しましたが、ググってもAWSの情報はまだ少ないので、解決方法になかなか辿り着けませんでした。 メンターからのアドバイス そこで今回、MENTAでお願いをしたメンターより、以下のアドバイスをいただきました。 「もし、nameカラムを追加したマイグレーションファイルが一番直近のものでかつロールバック可能であればロールバックして再度マイグレーションを適用するのがいいと思います。 一番簡単なのは一度データベースを作り直すことです。ただし、データが消えるのでデータが消えてもいい場合のみです。」 「基本的にデータベースのロールバックというのはやらない方がいいです。 これをやっていいのは開発環境くらいで本番環境では常にマイグレーションファイルを追加する運用が最も安全かと思います。」 僕は、nameカラムはマイグレーションファイル作成後、必要だと気づいたので、 ①roomのマイグレーションファイルをrails db:rollbackでdownさせ ②downさせたマイグレーションファイルにnameカラムを追記 ③再びrails db:migrate →結果これが良くなかったことだったと判明 ＝次回からは、カラム追加時は新しくマイグレーションファイルを別で作成するようにします! 解決方法手順 本場環境のデータベースを作り直すのであれば下記の方法で作り直せます。 DISABLE_DATABASE_ENVIRONMENT_CHECK=1 ./bin/rake db:drop ./bin/rake db:create ./bin/rake db:migrate しかしコマンド実行後、以下のエラーメッセージが出力 エラー内容 FATAL: Listen error: unable to monitor directories for changes. Visit https://github.com/guard/listen/wiki/Increasing-the-amount-of-inotify-watchers for info on how to fix this. ディレクトリの変更を監視することができませんという意味らしい… エラーの原因 railsコンソールが起動できなかったのは、1つの実ユーザIDに対して生成できるinotifyのインスタンスの数の上限が決まっており、その上限に達してしまった為とのことです。 ここで、inotifyとはLinuxファイルやディレクトリのイベントを監視する機能です。 （参考URL: https://qiita.com/yn-misaki/items/c850a07f7858437e4d26） echo fs.inotify.max_user_watches=524288 | sudo tee -a /etc/sysctl.conf && sudo sysctl -p 再度試すも、 ターミナルで出たエラー内容 Can't connect to local MySQL server through socket '/tmp/mysql.sock' (2) Couldn't drop database 'original35700_development' rake aborted! Mysql2::Error::ConnectionError: Can't connect to local MySQL server through socket '/tmp/mysql.sock' (2) -e:1:in `<main>' Tasks: TOP => db:drop:_unsafe というエラーが表示 MySQLに繋げられていないようでした。 メンターから以下のコマンドを実行するよう指示。 RAILS_ENV=production DISABLE_DATABASE_ENVIRONMENT_CHECK=1 ./bin/rake db:drop メンターから「既に削除できているっぽいので下記のコマンドを実行してください。」　とのこと RAILS_ENV=production ./bin/rake db:create RAILS_ENV=production ./bin/rake db:migrate この2つのコマンドを実行し、再度EC2上にデプロイしたところ新たにEC２上でDBが反映され解決！ 開発環境でテーブル作成後、あとからカラムを追加するときはrollbackではなく、addでマイグレーションファイルを新たに作成し追加するようにしましょう。

初めに 以下のチュートリアルをやってみました。このチュートリアルでは S3 を利用した静的ウェブサイトを Fargate や Network Load Balancer、API Gateway 利用して動的ウェブサイトにする過程を学ぶことができました。さらに CI/CD 環境の構築、クリックイベントに対するストリーミング処理の方法も学ぶことができます。このチュートリアルではほとんど CLI、時折 CloudFormation を使用してアプリケーションを構築していきます。 1. 静的 Web サイトの作成 開発環境として Cloud9 を用います。リージョンは us-east-1 を選択しました。以下の手順で開発環境を作成します。 開発環境の名前を入力します。 すべてデフォルトのまま進めます。 「Next step」をクリックします。 「Create environment」をクリックします。 チュートリアルに使用するコードをクローンし、そのルートディレクトリに移動します。 git clone -b python https://github.com/aws-samples/aws-modern-application-workshop.git cd aws-modern-application-workshop ウェブサイトをホスティングするバケットを作成します。mythical-mysfits-python-tutorial-bucket はバケット名です。バケット名はグローバルで一意である必要があります。 aws s3 mb s3://mythical-mysfits-python-tutorial-bucket 次に静的 Web ホスティングを有効にします。静的 Web ホスティングを有効化することで以下のようなウェブサイトエンドポイントを利用して、リクエスト先を index.html に指定することができます。 http://bucket-name.s3-website-Region.amazonaws.com http://bucket-name.s3-website.Region.amazonaws.com 上記 2 つのうちどちらを選択するかどうかはリージョンごとに決まっています。以下のリンクにすべて記載があります。 ユーザーガイドに記載がある通り、これらのエンドポイントには https ではアクセスできません。 Amazon S3 ウェブサイトエンドポイントは HTTPS をサポートしていません。 静的 Web ホスティングを有効にするために以下のコマンドを実行します。 aws s3 website s3://mythical-mysfits-python-tutorial-bucket --index-document index.html なお上記の CLI は以下のコンソール画面の操作を行うのと同じことです。「S3 コンソール画面 → 設定したいバケット名を選択 → プロパティ → 静的ウェブサイトホスティング」 の手順で設定できます。 /aws-modern-application-workshop/module-1/aws-cli/website-bucket-policy.json を編集します。 /aws-modern-application-workshop/module-1/aws-cli/website-bucket-policy.json { "Id": "MyPolicy", "Version": "2012-10-17", "Statement": [ { "Sid": "PublicReadForGetBucketObjects", "Effect": "Allow", "Principal": "*", "Action": "s3:GetObject", "Resource": "arn:aws:s3:::REPLACE_ME_BUCKET_NAME/*" } ] } 上記の REPLACE_ME_BUCKET_NAME を作成したバケット名で置き換えます。このチュートリアルではこのような置き換えが何度も出てきます。チュートリアル通りに進めているのにエラーが発生した場合は、この置き換えがされていないことが原因かもしれません。以下はバケット名を置き換えた後のポリシーです。 /aws-modern-application-workshop/module-1/aws-cli/website-bucket-policy.json { "Version": "2012-10-17", "Id": "MyPolicy", "Statement": [ { "Sid": "PublicReadForGetBucketObjects", "Effect": "Allow", "Principal": "*", "Action": "s3:GetObject", "Resource": "arn:aws:s3:::mythical-mysfits-python-tutorial-bucket/*" } ] } 編集を保存した後、以下のコマンドを実行しバケットポリシーを設定します。 aws s3api put-bucket-policy --bucket mythical-mysfits-python-tutorial-bucket --policy file://~/environment/aws-modern-application-workshop/module-1/aws-cli/website-bucket-policy.json S3 コンソール画面で確認し、設定が反映されていることが確認してみます。 /aws-modern-application-workshop/module-1/web/index.html をバケットにアップロードします。 aws s3 cp ~/environment/aws-modern-application-workshop/module-1/web/index.html s3://mythical-mysfits-python-tutorial-bucket/index.html ウェブサイトエンドポイントの http://mythical-mysfits-python-tutorial-bucket.s3-website-us-east-1.amazonaws.com にアクセスし、ホスティングが正常に行われていることを確認します。 2. 動的ウェブサイトの構築 この章は動的なウェブサイトを構築するためのステップです。Python を使用して、 Docker コンテナに Flask アプリケーションを作成します。 2-1. CloudFormation でインフラを構築する Fargate は VPC 内に配置されるサービスであり、VPC や サブネットなどを作成する必要があるため、このチュートリアルでは CloudFormation を使用します。 リソース 用途 個数 VPC Fargate を配置する 1 NAT Gateway 用 Elastic IP NAT Gateway の変換グローバル IP アドレスとして使用する 2 NAT Gateway プライベートサブネットからインターネットにアクセスする 2 Internet Gateway パブリックサブネットとインターネットを接続する 1 Fargate 用 セキュリティグループ Fargate が NLB からのトラフィックを許可するために使用する 1 パブリックサブネット NLB を配置し、インターネットと通信する 2 プライベートサブネット Fargate をセキュアに配置する 2 DynamoDB VPC エンドポイント Fargate がプライベートサブネットから DynamoDB にアクセスするために使用する 1 ECS タスクロール Fargate が DynamoDB にアクセスすることを許可する 1 ECS サービスロール トラフィックがコンテナに到達できるように ECS がリソースを管理することを許可する。イメージのプルや、ログの書き出しを許可する。 1 CodeBuild 用 ロール ビルドの実行に必要なアクションを許可する 1 CodePipeline 用 ロール パイプラインの実行に必要なアクションを許可する 1 上記の ECS タスクロールと ECS サービスロールはタスク定義内で以下のように設定されます。 以下のコマンドで CloudFormation のスタックを作成します。--capabilities CAPABILITY_NAMED_IAM はスタックリソースに IAM ロールが含まれているので必要なオプションです。 カスタム名を持つIAMリソースがある場合は、CAPABILITY_NAMED_IAMを指定する必要があります。 aws cloudformation create-stack --stack-name MythicalMysfitsCoreStack --capabilities CAPABILITY_NAMED_IAM --template-body file://~/environment/aws-modern-application-workshop/module-2/cfn/core.yml 以下のコマンドでスタックの作成状況を確認します。 aws cloudformation describe-stacks --stack-name MythicalMysfitsCoreStack 結果は以下のように返ってきます。StackStatus というキーを確認し、この値が CREATE_COMPLETE になるまでしばらく待ちます。 { "Stacks": [ { "StackId": "arn:aws:cloudformation:us-east-1:123456789012:stack/MythicalMysfitsCoreStack/3a8d94b0-eb6a-11eb-8b9c-0e443e7db6f3", "DriftInformation": { "StackDriftStatus": "NOT_CHECKED" }, "Description": "This stack deploys the core network infrastructure and IAM resources to be used for a service hosted in Amazon ECS using AWS Fargate.", "Tags": [], "EnableTerminationProtection": false, "CreationTime": "2021-07-23T03:58:24.697Z", "Capabilities": [ "CAPABILITY_NAMED_IAM" ], "StackName": "MythicalMysfitsCoreStack", "NotificationARNs": [], "StackStatus": "CREATE_IN_PROGRESS", "DisableRollback": false, "RollbackConfiguration": {} } ] } コンソール画面からでも作成状況を確認することができます。 スタックの作成が完了後、今後のアプリケーションインフラ構築を進めやすくするために以下のようにスタックの作成結果をファイルに書き出しておきます。ファイルに書き出した結果は、このスタックで作成したリソースの ARN を参照するために今後何度も使用することになります。 aws cloudformation describe-stacks --stack-name MythicalMysfitsCoreStack > ~/environment/cloudformation-core-output.json 2-2. Docker コンテナの準備 このコンテナで実行されるのは Flask アプリケーションです。以下のコードが実行されます。 / にアクセスしたクライアントには 「'/mysfits' にアクセスしてください」というレスポンスを返す /mysfits にアクセスしたクライアントには mysfits たちの名前や年齢などの情報が記載された mysfits-response.json をレスポンスとして返す /aws-modern-application-workshop/module-2/app/service/mythicalMysfitsService.py from flask import Flask, jsonify, json, Response, request from flask_cors import CORS app = Flask(__name__) app.config['JSONIFY_PRETTYPRINT_REGULAR'] = False CORS(app) @app.route("/") def healthCheckResponse(): return jsonify({"message" : "Nothing here, used for health check. Try /mysfits instead."}) @app.route("/mysfits") def getMysfits(): response = Response(open("mysfits-response.json", "rb").read()) response.headers["Content-Type"]= "application/json" return response if __name__ == "__main__": app.run(host="0.0.0.0", port=8080) Fargate を開始するためのインフラ構築後、Fargate でコンテナを起動するために、まず Docker イメージを作成します。 cd ~/environment/aws-modern-application-workshop/module-2/app docker build . -t 123456789012.dkr.ecr.us-east-1.amazonaws.com/mythicalmysfits/service:latest ビルドが完了すると最後に以下のような出力があります。この ID はイメージの ID です。 Successfully built 4611ce22a1cc Successfully tagged 123456789012.dkr.ecr.us-east-1.amazonaws.com/mythicalmysfits/service:latest 上記イメージ ID を渡してコンテナをローカルで起動します。 docker run -p 8080:8080 4611ce22a1cc 以下の「Preview Runnig Application」をクリックします。 URL の最後に「/myfits」を追加し、レスポンスとして JSON ドキュメントが確認できました。 イメージを ECR リポジトリにプッシュします。以下のようにリポジトリを作成します。その後、ログインします。 aws ecr create-repository --repository-name mythicalmysfits/service $(aws ecr get-login --no-include-email) docker push 123456789012.dkr.ecr.us-east-1.amazonaws.com/mythicalmysfits/service:latest コンソールでもプッシュされたイメージを確認してみます。 次に Fargate クラスターを作成します。以下のコマンドでクラスターと CloudWatch Logs のロググループを作成します。 aws ecs create-cluster --cluster-name MythicalMysfits-Cluster aws logs create-log-group --log-group-name mythicalmysfits-logs /cloudformation-core-output.json を参照しながら /aws-modern-application-workshop/module-2/aws-cli/task-definition.json を編集します。編集するのは以下3点です。その後、タスク定義を登録します。 REPLACE_ME_ECS_SERVICE_ROLE_ARN REPLACE_ME_ECS_TASK_ROLE_ARN REPLACE_ME_IMAGE_TAG_USED_IN_ECR_PUSH aws ecs register-task-definition --cli-input-json file://~/environment/aws-modern-application-workshop/module-2/aws-cli/task-definition.json 2-3. NLB を作成する Network Load Balancer を作成します。--subnets オプションには 2 つのパブリックサブネット ID を渡します。この実行結果を nlb-output.json というファイルに書き出しておきます。 aws elbv2 create-load-balancer --name mysfits-nlb --scheme internet-facing --type network --subnets subnet-0ebf9d3a8 subnet-fa7bdfa5 > ~/environment/nlb-output.json ターゲットグループを作成します。この実行結果は target-group-output.json というファイルに書き出しておきます。 aws elbv2 create-target-group --name MythicalMysfits-TargetGroup --port 8080 --protocol TCP --target-type ip --vpc-id vpc-0ebf9d3a8fakm8fa5 --health-check-interval-seconds 10 --health-check-path / --health-check-protocol HTTP --healthy-threshold-count 3 --unhealthy-threshold-count 3 > ~/environment/target-group-output.json リスナーを作成します。 aws elbv2 create-listener --default-actions TargetGroupArn=arn:aws:elasticloadbalancing:us-east-1:123456789012:targetgroup/MythicalMysfits-TargetGroup/0249b754ae522df7,Type=forward --load-balancer-arn arn:aws:elasticloadbalancing:us-east-1:123456789012:loadbalancer/net/mysfits-nlb/e7795589l86fc2d7 --port 80 --protocol TCP 2-4. Fargate でタスクを実行する /aws-modern-application-workshop/module-2/aws-cli/service-definition.json を編集します。以下を置き換えます。 REPLACE_ME_SECURITY_GROUP_ID REPLACE_ME_PRIVATE_SUBNET_ONE REPLACE_ME_PRIVATE_SUBNET_TWO REPLACE_ME_NLB_TARGET_GROUP_ARN 以下のコマンドを実行しサービスを作成します。 aws ecs create-service --cli-input-json file://~/environment/aws-modern-application-workshop/module-2/aws-cli/service-definition.json コンソール画面を確認するとタスクが実行されていることが確認できます。 ブラウザを開き、http://NLB_DNS_NAME/mysfits にアクセスします。以下の NLB_DNS_NAME を NLB の DNS 名で置き換えます。ただし、http でアクセスします。証明書がないため、 https ではアクセスできません。以前 Cloud9 で Docker を起動したときのレスポンスが返ってきていれば、正常に Fargate が稼働しています。 REPLACE_ME を http://NLB_DNS_NAME で置き換えます。 /aws-modern-application-workshop/module-2/web/index.html <script> var mysfitsApiEndpoint = 'REPLACE_ME'; // example: 'http://mythi-publi-abcd12345-01234567890123.elb.us-east-1.amazonaws.com' var app = angular.module('mysfitsApp', []); このエンドポイントは ajax 呼び出しの URL に渡されます。/aws-modern-application-workshop/module-2/web/index.html の該当ソースは以下の通りです。 function getAllMysfits(callback) { var mysfitsApi = mysfitsApiEndpoint + '/mysfits'; $.ajax({ url : mysfitsApi, type : 'GET', success : function(response) { callback(response.mysfits); }, error : function(response) { console.log("could not retrieve mysfits list."); console.log(response.message); } }); 変更を保存した後、バケットにアップロードします。 aws s3 cp ~/environment/aws-modern-application-workshop/module-2/web/index.html s3://mythical-mysfits-python-tutorial-bucket/index.html 2-5. CI/CD 環境を構築する コードの変更を自動でデプロイするために、CodeCommit、CodeBuild、CodePipeline を使用して CI/CD 環境を構築します。 CodeBuild の出力アーティファクトを格納するバケットを作成します。 aws s3 mb s3://mythical-mysfits-python-tutorial-artifacts-bucket /aws-modern-application-workshop/module-2/aws-cli/artifacts-bucket-policy.json を編集します。以下を置き換えます。その後バケットポリシーを設定するコマンドを実行します。 REPLACE_ME_CODEBUILD_ROLE_ARN REPLACE_ME_CODEPIPELINE_ROLE_ARN aws s3api put-bucket-policy --bucket mythical-mysfits-python-tutorial-artifacts-bucket --policy file://~/environment/aws-modern-application-workshop/module-2/aws-cli/artifacts-bucket-policy.json CodeCommit にリポジトリを作成します。 aws codecommit create-repository --repository-name MythicalMysfitsService-Repository ビルドプロジェクトを作成するために /aws-modern-application-workshop/module-2/aws-cli/code-build-project.json を編集します。以下を置き換えます。 REPLACE_ME_ACCOUNT_ID REPLACE_ME_REGION REPLACE_ME_CODEBUILD_ROLE_ARN aws codebuild create-project --cli-input-json file://~/environment/aws-modern-application-workshop/module-2/aws-cli/code-build-project.json パイプラインを作成するために /aws-modern-application-workshop/module-2/aws-cli/code-pipeline.json を編集します。以下を置き換えます。 REPLACE_ME_CODEPIPELINE_ROLE_ARN REPLACE_ME_ARTIFACTS_BUCKET_NAME aws codepipeline create-pipeline --cli-input-json file://~/environment/aws-modern-application-workshop/module-2/aws-cli/code-pipeline.json ECR のリポジトリポリシーを設定します。以下を置き換えます。 REPLACE_ME_CODEBUILD_ROLE_ARN aws ecr set-repository-policy --repository-name mythicalmysfits/service --policy-text file://~/environment/aws-modern-application-workshop/module-2/aws-cli/ecr-policy.json コンソール画面ではリポジトリポリシーを確認できなかったので、CLI で確認してみました。以下のコマンドで確認できます。 aws ecr get-repository-policy --repository-name mythicalmysfits/service CodeCommit のリポジトリをクローンし、ローカルリポジトリを作成します。 git clone https://git-codecommit.us-east-1.amazonaws.com/v1/repos/MythicalMysfitsService-Repository CodeCommit のローカルリポジトリにアプリケーションコードをコピーします。 cp -r ~/environment/aws-modern-application-workshop/module-2/app/* ~/environment/MythicalMysfitsService-Repository/ パイプラインが正常に機能しているかをチェックするために Evangeline の年齢を変更します。 "species": "Chimera", "age": 999, 変更を CodeCommit リポジトリにプッシュします。 git add . git commit -m "I changed the age of one of the mysfits." git push 以下はプッシュ後のリポジトリです。 コンソール画面でパイプラインが実行されていることを確認してみます。 コンソール画面では以下のようにビルドが実行中です。 パイプラインの実行完了後、ウェブサイトを更新し、Evangeline の年齢が変更した値に更新されていることが確認できました。 3. MYSFIT データの保存 この章では mysfit を JSON ファイルにで管理するのではなく、データベースに保存する設定を行います。データベースには DynamoDB を利用します。 3-1. テーブルを作成し、アイテムを追加する テーブルを作成しアイテムを追加します。 aws dynamodb create-table --cli-input-json file://~/environment/aws-modern-application-workshop/module-3/aws-cli/dynamodb-table.json aws dynamodb batch-write-item --request-items file://~/environment/aws-modern-application-workshop/module-3/aws-cli/populate-dynamodb.json コンソール画面で確認してみます。 3-2. DynamoDB を参照するようにコードを編集する Flask アプリケーションで DynamoDB からデータを取得するために /aws-modern-application-workshop/module-3/app/service/mythicalMysfitsService.py を用います。このモジュールは boto3 という SDK を使用して DynamoDB からデータを取得します。例えば、以下のコードはテーブルに登録されている全データを取得します。 /aws-modern-application-workshop/module-3/app/service/mythicalMysfitsService.py import boto3 ... ... def getAllMysfits(): response = client.scan( TableName='MysfitsTable' ) リポジトリに変更を加え、プッシュします。 cp ~/environment/aws-modern-application-workshop/module-3/app/service/* ~/environment/MythicalMysfitsService-Repository/service/ cd ~/environment/MythicalMysfitsService-Repository git add . git commit -m "Add new integration to DynamoDB." git push ビルドに失敗しました。 「詳細」をクリックして CodeBuild のビルドログを確認します。 [Container] 2021/07/23 08:24:48 Running command docker build -t mythicalmysfits/service:latest . Sending build context to Docker daemon 19.97kB Step 1/13 : FROM ubuntu:latest latest: Pulling from library/ubuntu toomanyrequests: You have reached your pull rate limit. You may increase the limit by authenticating and upgrading: https://www.docker.com/increase-rate-limit [Container] 2021/07/23 08:24:49 Command did not exit successfully docker build -t mythicalmysfits/service:latest . exit status 1 公式ドキュメントに以下の記述があります。 プルリクエストを発行し、アカウントタイプの制限を超えた429場合、マニフェストがリクエストされると、DockerHubは次の本文のレスポンスコードを返します。 You have reached your pull rate limit. You may increase the limit by authenticating and upgrading: https://www.docker.com/increase-rate-limits チュートリアルに記載のあった Git 認証を飛ばしていたので、匿名ユーザーに対してレート制限が適用されたようです。Git 認証を追加し再度プッシュします。 git config --global user.name "xxxx" git config --global user.email yyyy@example.com git config --global credential.helper '!aws codecommit credential-helper $@' git config --global credential.UseHttpPath true なお、認証情報は以下のローカルファイルに保存されます。 ~/.gitconfig [credential] helper = !aws codecommit credential-helper $@ UseHttpPath = true [core] editor = nano [user] name = xxxx email = yyyy@example.com プッシュします。 git add . git commit -m "set git config.I retry PiPeline execution" git push 正常にビルドが完了しました。 4. ユーザーの登録 Cognito を使用してユーザー管理を行います。Cognito のユーザープール・アプリクライアントを設定して認証されていないクライアントがサインアップ・サインインできるように実装します。 まずユーザープールを作成します。ユーザープールとはユーザーディレクトリのことです。 aws cognito-idp create-user-pool --pool-name MysfitsUserPool --auto-verified-attributes email 上記で作成したユーザープール ID を使用してアプリクライアントを作成します。 aws cognito-idp create-user-pool-client --user-pool-id us-east-1_BaCZvuQrU --client-name MysfitsUserPoolClient ユーザープール ID と アプリクライアント ID は、JavaScript for SDK によって処理されます。以下は該当ソースになります。 /aws-modern-application-workshop/module-4/web/register.html <script> var cognitoUserPoolId = 'REPLACE_ME'; // example: 'us-east-1_abcd12345' var cognitoUserPoolClientId = 'REPLACE_ME'; // example: 'abcd12345abcd12345abcd12345' ... ... var poolData = { UserPoolId : cognitoUserPoolId, ClientId : cognitoUserPoolClientId }; var userPool = new AmazonCognitoIdentity.CognitoUserPool(poolData); ... ... userPool.signUp(email, pw, attributeList, null, function(err, result){ if (err) { alert(err.message); return; } cognitoUser = result.user; console.log(cognitoUser); localStorage.setItem('email', email); window.location.replace('confirm.html'); }); 上記のコードでの最後に window.location.replace('confirm.html'); とあります。こちらは登録したメールアドレスに送信される認証コードを確認するためのページです。認証コードはサインアップ後、 Cognito から送信されます。認証コードを Cognito に送信してユーザー登録を完了させるための該当ソースは以下の通りです。 /aws-modern-application-workshop/module-4/web/confirm.html <script> var cognitoUserPoolId = 'REPLACE_ME'; // example: 'us-east-1_abcd12345' var cognitoUserPoolClientId = 'REPLACE_ME'; // example: 'abcd12345abcd12345abcd12345' ... ... var confirmCode = document.getElementById('confirmCode').value; var poolData = { UserPoolId : cognitoUserPoolId, ClientId : cognitoUserPoolClientId }; var userName = localStorage.getItem('email'); var userPool = new AmazonCognitoIdentity.CognitoUserPool(poolData); var userData = { Username : userName, Pool : userPool }; var cognitoUser = new AmazonCognitoIdentity.CognitoUser(userData); cognitoUser.confirmRegistration(confirmCode, true, function(err, result) { if (err) { alert(err.message); return; } window.location.replace("index.html"); }); ここからは Cognito によるオーソライザーを使用して、登録されたユーザーのみが特定の機能を使用できるようにします。そのためにはAPI Gateway を NLB に統合する必要があります。 VPC リンクを使用すると、Application Load Balancer または Amazon ECS コンテナベースのアプリケーションなどの、HTTP API ルートを VPC 内のプライベートリソースに接続するプライベート統合を作成できます。 ただし、このチュートリアルでは NLB がパブリックであり、プライベートリソースではありません。この点について、以下の注意書きがあります。 実際の環境では、API Gateway がインターネット接続した API 承認向けの戦略であることがわかっているため、最初から内部用に NLB を作成する必要があります (または、新しい内部ロードバランサーを作成して、既存のものと置き換える必要があります)。ただし、時間を節約するため、既に作成した NLB をパブリックにアクセスできる状態のまま使用します。 以下のコマンドを実行し、VPC リンクを作成します。 aws apigateway create-vpc-link --name MysfitsApiVpcLink --target-arns arn:aws:elasticloadbalancing:us-east-1:123456789012:loadbalancer/net/mysfits-nlb/e7795560f86fc2d7 > ~/environment/api-gateway-link-output.json REST API を作成します。/aws-modern-application-workshop/module-4/aws-cli/api-swagger.json の REPLACE_ME の箇所を置き換えます。その後以下のコマンドを実行します。 aws apigateway import-rest-api --parameters endpointConfigurationTypes=REGIONAL --body file://~/environment/aws-modern-application-workshop/module-4/aws-cli/api-swagger.json --fail-on-warnings コンソール画面で作成された REST API を確認してみます。 上図のスクリーンショットが良くなかったのですが、「認可」の部分が「なし」になっています。これは、/ にはオーソライザーが設定されていないからです。オーソライザーが設定されているのは /mysfits/{mysfitId}/like と /mysfits/{mysfitId}/adopt です。該当ソースは以下の通りです。 /aws-modern-application-workshop/module-4/aws-cli/api-swagger.json "/mysfits/{mysfitId}/like": { "post": { "parameters": [{ "name": "mysfitId", "in": "path", "required": true, "type": "string" }], "responses": { "200": { "description": "Default response for CORS method", "headers": { "Access-Control-Allow-Headers": { "type": "string" }, "Access-Control-Allow-Methods": { "type": "string" }, "Access-Control-Allow-Origin": { "type": "string" } } } }, "security": [{ "MysfitsUserPoolAuthorizer": [ ] }], なお、オーソライザーオブジェクトの設定方法のドキュメントは以下になります。 続いて API をデプロイします。 aws apigateway create-deployment --rest-api-id hnypyxgqve --stage-name prod バックエンドを更新し、プッシュします。 cd ~/environment/MythicalMysfitsService-Repository/ cp -r ~/environment/aws-modern-application-workshop/module-4/app/* . git add . git commit -m "Update service code backend to enable additional website features." git push S3 バケットのオブジェクトも更新します。以下 3 ファイルの REPLACE_ME を置き換えます。 /aws-modern-application-workshop/module-4/web/index.html <script> var mysfitsApiEndpoint = 'REPLACE_ME'; // example: 'https://abcd12345.execute-api.us-east-1.amazonaws.com/prod' var cognitoUserPoolId = 'REPLACE_ME'; // example: 'us-east-1_abcd12345' var cognitoUserPoolClientId = 'REPLACE_ME'; // example: 'abcd12345abcd12345abcd12345' var awsRegion = 'REPLACE_ME'; // example: 'us-east-1' or 'eu-west-1' etc. /aws-modern-application-workshop/module-4/web/register.html <script> var cognitoUserPoolId = 'REPLACE_ME'; // example: 'us-east-1_abcd12345' var cognitoUserPoolClientId = 'REPLACE_ME'; // example: 'abcd12345abcd12345abcd12345' /aws-modern-application-workshop/module-4/web/confirm.html <script> var cognitoUserPoolId = 'REPLACE_ME'; // example: 'us-east-1_abcd12345' var cognitoUserPoolClientId = 'REPLACE_ME'; // example: 'abcd12345abcd12345abcd12345' その後 S3 バケットにアップロードします。 aws s3 cp --recursive ~/environment/aws-modern-application-workshop/module-4/web/ s3://mythical-mysfits-python-tutorial-bucket/ ウェブサイトを更新すると以下のようにログイン機能が実装されていることが確認できます。 サインアップができることを確認し、ユーザープールに登録されていることが確認できました。 5. ユーザークリックの把握 ストリーミング処理をするコードを格納するために、新たに CodeCommit にリポジトリを作成します。 aws codecommit create-repository --repository-name MythicalMysfitsStreamingService-Repository Lambda 関数のコードを S3 にアップロードするために新たにバケットを作成します。 aws s3 mb s3://mythical-mysfits-python-tutorial-lambda-package-bucket Kinesis Firehose から送信されるレコードを Lambda 関数は適切なフォーマットに変換します。該当ソースは以下の通りです。 /aws-modern-application-workshop/module-5/app/streaming/streamProcessor.py def processRecord(event, context): output = [] for record in event['records']: print('Processing record: ' + record['recordId']) click = json.loads(base64.b64decode(record['data'])) mysfitId = click['mysfitId'] mysfit = retrieveMysfit(mysfitId) enrichedClick = { 'userId': click['userId'], 'mysfitId': mysfitId, 'goodevil': mysfit['goodevil'], 'lawchaos': mysfit['lawchaos'], 'species': mysfit['species'] } output_record = { 'recordId': record['recordId'], 'result': 'Ok', 'data': base64.b64encode(json.dumps(enrichedClick).encode('utf-8') + b'\n').decode('utf-8') } output.append(output_record) print('Successfully processed {} records.'.format(len(event['records']))) return {'records': output} SAM CLI を利用して CloudFormation テンプレートを作成します。real-time-streaming.yml が transformed-streaming.yml に変換されます。 sam package --template-file ./real-time-streaming.yml --output-template-file ./transformed-streaming.yml --s3-bucket mythical-mysfits-python-tutorial-lambda-package-bucket SAM によって変換された CloudFormation のテンプレートを使用してスタックを作成します。 aws cloudformation deploy --template-file /home/ec2-user/environment/MythicalMysfitsStreamingService-Repository/transformed-streaming.yml --stack-name MythicalMysfitsStreamingStack --capabilities CAPABILITY_IAM このスタックによって作成されるリソースは以下の通りです。 リソース 用途 API Gateway Kinesis Firehose と統合するREST API S3 バケット Kinesis のログを格納する FirehoseDeliveryRole Lambda、S3 へのアクションを許可する Lambda Kinesis Firehose のレコードを変換する Kinesis Firehose クリックデータを受信し Lambda、S3 に送信する ストリーム用のエンドポイントを追加した新しい index.html を S3 バケットにアップロードします。以下の REPLACE_ME を置き換えます。 /aws-modern-application-workshop/module-5/web/index.html <script> var mysfitsApiEndpoint = 'REPLACE_ME'; // example: 'https://abcd12345.execute-api.us-east-1.amazonaws.com/prod' var streamingApiEndpoint = 'REPLACE_ME'; // example: 'https://abcd12345.execute-api.us-east-1.amazonaws.com/prod' var cognitoUserPoolId = 'REPLACE_ME'; // example: 'us-east-1_abcd12345' var cognitoUserPoolClientId = 'REPLACE_ME'; // example: 'abcd12345abcd12345abcd12345' var awsRegion = 'REPLACE_ME'; // example: 'us-east-1' or 'eu-west-1' etc. 以下のコマンドで編集した index.html をアップロードします。 aws s3 cp ~/environment/aws-modern-application-workshop/module-5/web/index.html s3://mythical-mysfits-python-tutorial-bucket/ S3 バケットに送信された、変換されたクリックイベントのストリーミングデータファイルをダウンロードして確認してみます。 {"userId": "7c25e02a-05fe-4fcc-8569-6e8688933805", "mysfitId": "a901bb08-1985-42f5-bb77-27439ac14300", "goodevil": "Good", "lawchaos": "Neutral", "species": "Haetae"} {"userId": "7c25e02a-05fe-4fcc-8569-6e8688933805", "mysfitId": "a901bb08-1985-42f5-bb77-27439ac14300", "goodevil": "Good", "lawchaos": "Neutral", "species": "Haetae"} {"userId": "7c25e02a-05fe-4fcc-8569-6e8688933805", "mysfitId": "b6d16e02-6aeb-413c-b457-321151bb403d", "goodevil": "Evil", "lawchaos": "Chaotic", "species": "Troll"} {"userId": "7c25e02a-05fe-4fcc-8569-6e8688933805", "mysfitId": "a901bb08-1985-42f5-bb77-27439ac14300", "goodevil": "Good", "lawchaos": "Neutral", "species": "Haetae"} 上記の処理を行った Lambda が出力したログを CloudWatch Logs で確認してみます。 START RequestId: ba9b88c5-a552-4c6e-8df6-9ccb2ba2b10e Version: $LATEST Processing record: 49620403830958916807586905640547610254069909630134255618000000 Processing record: 49620403830958916807586905640568161993003359356896411650000000 Processing record: 49620403830958916807586905640572997696281818079753666562000000 Processing record: 49620403830958916807586905640602011915952570485616672770000000 Successfully processed 4 records. END RequestId: ba9b88c5-a552-4c6e-8df6-9ccb2ba2b10e REPORT RequestId: ba9b88c5-a552-4c6e-8df6-9ccb2ba2b10e Duration: 653.45 ms Billed Duration: 654 ms Memory Size: 128 MB Max Memory Used: 53 MB 正常にレコードが処理されていることが確認できました。 以上でチュートリアルは終了です。 参考記事

アクセス制御サービス IAMサービスの概要 　IAM ユーザーに対してアクセスを安全に制御する仕組み 通常ルートユーザーは使用せず、IAMユーザーでログインする ※ルートユーザーはAWSの全ての操作を行うことができるため、情報漏洩／誤作動の危険があるため IAMサービスを通じたAWSの操作方法 WebブラウザでAWSマネジメントコンソールにログインする 登録したIAMユーザー／パスワードでアクセスする AWS CLIでWindows／Linuxからコマンド操作する リージョン及び、IAMユーザーごとにアクセスキーIDとシークレットアクセスキーを事前に設定する必要がある AWS SDKでプログラムからAPIを利用する アクセスキーIDとシークレットアクセスキーによる認証が必要 アクセスキーIDとシークレットアクセスキーによる認証は、キー流出が懸念されるためIAMロールによる認証が推奨される 参考

目次 1.背景 2.構成 2-1.Lambdaの構築 2-1-1.Lambda関数の作成 2-1-2.IAMロールの権限付与 2-1-3.ソースコード 2-1-3-1.Lambda①：nicehash-balance-notification-LINE のソース 2-1-3-2.Lambda②：nicehash-draw-figure のソース 2-1-3-3.コンフィグ 2-1-4.Module組み込み 2-1-5.基本設定の編集 2-2.NiceHash APIによる収益情報取得 2-3.EventBridgeによるトリガー定義 2-4.S3バケットの作成 2-5.LINE NotifyによるLINE通知 2-6.RDBの作成 3.実行結果 4.終わりに 5.更新履歴 1.背景 　「②NiceHashマイニング収益をAWS Lambda×LINE NotifyでLINE通知する」で日々のマイニング収益を定期的に通知するシステムを構築してから数カ月経って、統計データがたまってきていたので収益と円相場の変遷をグラフ化したいと思った。 2.構成 システム構成は、AWS Lambdaベースのアーキテクチャ。 処理の流れ 　①. EventBridgeの日次実行cronがトリガーとなり、 　　[Lambda①]nicehash-balance-notification-LINEがキックされる 　②. 外部APIから当日のマイニング収益情報を取得 　③. MySQL on EC2に当日の収益情報をレコード追加 　④. MySQL on EC2から収益情報の統計を取得 　⑤. S3へ収益の統計情報をアップロード 　⑥. [Lambda②]nicehash-draw-figureのinvoke 　⑦. 残高/円相場の統計情報をグラフ描画し、S3へアップロード 　⑧. S3から統計情報グラフをダウンロード 　⑨. POSTで通知メッセージ/統計情報グラフをLINE Notifyへ渡して 　　 スマホへLINE上で通知 2-1.Lambdaの構築 2-1-1.Lambda関数の作成 呼び出されるLambda関数本体を作成する ・Lambdaデプロイ上限250MB回避のため、Lambda関数を2つに分けて作成 【Lambda①】 　関数名：「nicehash-balance-notification-LINE」 　ランタイム：「Python 3.6」 　※DBへアクセスするためVPC指定も必要 【Lambda②】 　関数名：「nicehash-draw-figure」 　ランタイム：「Python 3.7」 2-1-2.IAMロールの権限付与 サービス間アクセスに必要となる権限をLambdaへ付与する Lambda①：nicehash-balance-notification-LINEに対して、下記の権限を付与する。 S3に対するread/write権限 EC2に対するアクセス権限 Lambdaに対するinvoke権限 Lambda②：nicehash-draw-figureに対して、下記の権限を付与する。 S3に対するwrite権限 2-1-3.ソースコード 2-1-3-1.Lambda①：nicehash-balance-notification-LINE のソース nicehash-balance-notification-LINE nicehash-balance-notification-LINE/ ├ lambda_function.py ├ db_data_deal.py ├ nicehash.py ├ marketrate.py ├ s3_deal.py ├ create_message.py ├ line_config.py ├ mysql_config.py ├ nicehash_config.py └ s3_config.py Lambda①メインプログラム lambda_function.py import json import requests import os import datetime import boto3 import db_data_deal import s3_deal import create_message import mysql_config as MYSQLconfig import s3_config as S3config import line_config as LINEconfig ### AWS Lambda handler method def lambda_handler(event, context): Messenger = create_message.create_messenger() (today_balance,market_price) = Messenger.get_balance_and_rate() Sqldealer = db_data_deal.sqldealer() db_data_dict = Sqldealer.road_data() Sqldealer.insert_data(today_balance,market_price) db_data_dict = Sqldealer.road_data() Sqldealer.save_dict_to_json(df_dict = db_data_dict, out_path = S3config.dict_file_path) S3dealer = s3_deal.s3_dealer(bucket=S3config.bucket) S3dealer.save_dict_to_s3(local_file_path=S3config.dict_file_path,file_name_base=S3config.dict_file_name_base) ### Call Lambda②(nicehash-draw-figure) response = boto3.client('lambda').invoke( FunctionName='nicehash-draw-figure', InvocationType='RequestResponse', Payload=json.dumps(db_data_dict, cls = db_data_deal.DateTimeEncoder) ) get_file_local_path = S3dealer.get_s3_file_item(file_name_base=S3config.figure_file_name_base) msg = Messenger.create_notification_msg(db_data_dict) notify(msg,get_file_local_path) print(msg) def notify(msg, *args): headers = {"Authorization": "Bearer %s" % LINEconfig.LINE_NOTIFY_ACCESS_TOKEN} url = "https://notify-api.line.me/api/notify" payload = {'message': msg} if len(args) == 0: requests.post(url, data=payload, headers=headers) else: files = {"imageFile": open(args[0], "rb")} requests.post(url, data=payload, headers=headers,files=files) os.remove(args[0]) return 0 DBからの情報取得/DB更新処理を行うクラス db_data_deal.py import os import json from json import JSONEncoder import mysql.connector import boto3 import datetime import mysql_config as SQLconfig class sqldealer: def __init__(self): self.connection = mysql.connector.connect(user=SQLconfig.user, password=SQLconfig.password, host=SQLconfig.host, database=SQLconfig.database) self.columns = ['db-id','date','balance','diff','market'] self.db_data_dict = dict() ### Get statistics information from MySQL def road_data(self): try: with self.connection.cursor() as cur: select_sql = 'SELECT * FROM nicehash_info;' cur.execute(select_sql) row_db_data = cur.fetchall() except: err_msg = 'Error001:DB-data取得に失敗' print(err_msg) return err_msg return self.datashaping_tuple_to_dict(row_db_data) ### Insert today's balance record into MySQL def insert_data(self,today_balance,market_price): try: db_id = self.db_data_dict['db-id'][-1] + 1 date = datetime.datetime.now(datetime.timezone(datetime.timedelta(hours=9))).strftime('%Y-%m-%d') balance = today_balance diff = today_balance - self.db_data_dict['balance'][-1] insert_info = str(db_id)+','+'"'+date+'"'+','+str(balance)+','+str(diff)+','+str(market_price) insert_sql = 'INSERT INTO nicehash_info VALUES('+insert_info+');' with self.connection.cursor() as cur: cur.execute(insert_sql) cur.execute('commit;') print(insert_sql) except: err_msg = 'Error002:DB更新に失敗' print(err_msg) return err_msg ### Cast DB row data to dict type def datashaping_tuple_to_dict(self,tupple_data): try: db_data_list = [[],[],[],[],[]] db_data_dict = dict() for i in range(len(tupple_data)): for j in range(len(tupple_data[i])): db_data_list[j].append(tupple_data[i][j]) self.db_data_dict = dict(zip(self.columns, db_data_list)) except: err_msg = 'Error003:DBデータの型変換に失敗' print(err_msg) return err_msg return self.db_data_dict ### Save dict object in json format def save_dict_to_json(self,df_dict,out_path): if os.path.exists(out_path): with open(out_path,'w') as json_obj: json_obj.write("") print("jsonファイル作成") with open(out_path, 'w') as json_file: json.dump(df_dict, json_file, cls = DateTimeEncoder) class DateTimeEncoder(JSONEncoder): ### Override the default method def default(self, obj): if isinstance(obj, (datetime.date, datetime.datetime)): return obj.isoformat() NiceHashから残高情報を取得するためのクラス nicehash.py from datetime import datetime from time import mktime import uuid import hmac import requests import json from hashlib import sha256 import optparse import sys class private_api: def __init__(self, host, organisation_id, key, secret, verbose=False): self.key = key self.secret = secret self.organisation_id = organisation_id self.host = host self.verbose = verbose def request(self, method, path, query, body): xtime = self.get_epoch_ms_from_now() xnonce = str(uuid.uuid4()) message = bytearray(self.key, 'utf-8') message += bytearray('\x00', 'utf-8') message += bytearray(str(xtime), 'utf-8') message += bytearray('\x00', 'utf-8') message += bytearray(xnonce, 'utf-8') message += bytearray('\x00', 'utf-8') message += bytearray('\x00', 'utf-8') message += bytearray(self.organisation_id, 'utf-8') message += bytearray('\x00', 'utf-8') message += bytearray('\x00', 'utf-8') message += bytearray(method, 'utf-8') message += bytearray('\x00', 'utf-8') message += bytearray(path, 'utf-8') message += bytearray('\x00', 'utf-8') message += bytearray(query, 'utf-8') if body: body_json = json.dumps(body) message += bytearray('\x00', 'utf-8') message += bytearray(body_json, 'utf-8') digest = hmac.new(bytearray(self.secret, 'utf-8'), message, sha256).hexdigest() xauth = self.key + ":" + digest headers = { 'X-Time': str(xtime), 'X-Nonce': xnonce, 'X-Auth': xauth, 'Content-Type': 'application/json', 'X-Organization-Id': self.organisation_id, 'X-Request-Id': str(uuid.uuid4()) } s = requests.Session() s.headers = headers url = self.host + path if query: url += '?' + query if self.verbose: print(method, url) if body: response = s.request(method, url, data=body_json) else: response = s.request(method, url) if response.status_code == 200: return response.json() elif response.content: raise Exception(str(response.status_code) + ": " + response.reason + ": " + str(response.content)) else: raise Exception(str(response.status_code) + ": " + response.reason) def get_epoch_ms_from_now(self): now = datetime.now() now_ec_since_epoch = mktime(now.timetuple()) + now.microsecond / 1000000.0 return int(now_ec_since_epoch * 1000) def algo_settings_from_response(self, algorithm, algo_response): algo_setting = None for item in algo_response['miningAlgorithms']: if item['algorithm'] == algorithm: algo_setting = item if algo_setting is None: raise Exception('Settings for algorithm not found in algo_response parameter') return algo_setting def get_accounts(self): return self.request('GET', '/main/api/v2/accounting/accounts2/', '', None) def get_accounts_for_currency(self, currency): return self.request('GET', '/main/api/v2/accounting/account2/' + currency, '', None) def get_withdrawal_addresses(self, currency, size, page): params = "currency={}&size={}&page={}".format(currency, size, page) return self.request('GET', '/main/api/v2/accounting/withdrawalAddresses/', params, None) def get_withdrawal_types(self): return self.request('GET', '/main/api/v2/accounting/withdrawalAddresses/types/', '', None) def withdraw_request(self, address_id, amount, currency): withdraw_data = { "withdrawalAddressId": address_id, "amount": amount, "currency": currency } return self.request('POST', '/main/api/v2/accounting/withdrawal/', '', withdraw_data) def get_my_active_orders(self, algorithm, market, limit): ts = self.get_epoch_ms_from_now() params = "algorithm={}&market={}&ts={}&limit={}&op=LT".format(algorithm, market, ts, limit) return self.request('GET', '/main/api/v2/hashpower/myOrders', params, None) def create_pool(self, name, algorithm, pool_host, pool_port, username, password): pool_data = { "name": name, "algorithm": algorithm, "stratumHostname": pool_host, "stratumPort": pool_port, "username": username, "password": password } return self.request('POST', '/main/api/v2/pool/', '', pool_data) def delete_pool(self, pool_id): return self.request('DELETE', '/main/api/v2/pool/' + pool_id, '', None) def get_my_pools(self, page, size): return self.request('GET', '/main/api/v2/pools/', '', None) def get_hashpower_orderbook(self, algorithm): return self.request('GET', '/main/api/v2/hashpower/orderBook/', 'algorithm=' + algorithm, None ) def create_hashpower_order(self, market, type, algorithm, price, limit, amount, pool_id, algo_response): algo_setting = self.algo_settings_from_response(algorithm, algo_response) order_data = { "market": market, "algorithm": algorithm, "amount": amount, "price": price, "limit": limit, "poolId": pool_id, "type": type, "marketFactor": algo_setting['marketFactor'], "displayMarketFactor": algo_setting['displayMarketFactor'] } return self.request('POST', '/main/api/v2/hashpower/order/', '', order_data) def cancel_hashpower_order(self, order_id): return self.request('DELETE', '/main/api/v2/hashpower/order/' + order_id, '', None) def refill_hashpower_order(self, order_id, amount): refill_data = { "amount": amount } return self.request('POST', '/main/api/v2/hashpower/order/' + order_id + '/refill/', '', refill_data) def set_price_hashpower_order(self, order_id, price, algorithm, algo_response): algo_setting = self.algo_settings_from_response(algorithm, algo_response) price_data = { "price": price, "marketFactor": algo_setting['marketFactor'], "displayMarketFactor": algo_setting['displayMarketFactor'] } return self.request('POST', '/main/api/v2/hashpower/order/' + order_id + '/updatePriceAndLimit/', '', price_data) def set_limit_hashpower_order(self, order_id, limit, algorithm, algo_response): algo_setting = self.algo_settings_from_response(algorithm, algo_response) limit_data = { "limit": limit, "marketFactor": algo_setting['marketFactor'], "displayMarketFactor": algo_setting['displayMarketFactor'] } return self.request('POST', '/main/api/v2/hashpower/order/' + order_id + '/updatePriceAndLimit/', '', limit_data) def set_price_and_limit_hashpower_order(self, order_id, price, limit, algorithm, algo_response): algo_setting = self.algo_settings_from_response(algorithm, algo_response) price_data = { "price": price, "limit": limit, "marketFactor": algo_setting['marketFactor'], "displayMarketFactor": algo_setting['displayMarketFactor'] } return self.request('POST', '/main/api/v2/hashpower/order/' + order_id + '/updatePriceAndLimit/', '', price_data) def get_my_exchange_orders(self, market): return self.request('GET', '/exchange/api/v2/myOrders', 'market=' + market, None) def get_my_exchange_trades(self, market): return self.request('GET','/exchange/api/v2/myTrades', 'market=' + market, None) def create_exchange_limit_order(self, market, side, quantity, price): query = "market={}&side={}&type=limit&quantity={}&price={}".format(market, side, quantity, price) return self.request('POST', '/exchange/api/v2/order', query, None) def create_exchange_buy_market_order(self, market, quantity): query = "market={}&side=buy&type=market&secQuantity={}".format(market, quantity) return self.request('POST', '/exchange/api/v2/order', query, None) def create_exchange_sell_market_order(self, market, quantity): query = "market={}&side=sell&type=market&quantity={}".format(market, quantity) return self.request('POST', '/exchange/api/v2/order', query, None) def cancel_exchange_order(self, market, order_id): query = "market={}&orderId={}".format(market, order_id) return self.request('DELETE', '/exchange/api/v2/order', query, None) CoinGeckoを利用して仮装通貨相場をリアルタイムで取得するクラス marketrate.py import requests import json class trade_table: def __init__(self, market="BTC"): ### currency-name conversion table self.currency_rename_table = {'BTC':'Bitcoin','ETH':'Ethereum','LTC':'Litecoin', 'XRP':'XRP','RVN':'Ravencoin','MATIC':'Polygon', 'BCH':'Bitcoin Cash','XLM':'Stellar','XMR':'Monero','DASH':'Dash'} self.market = self.currency_rename_table[market] def get_rate(self): body = requests.get('https://api.coingecko.com/api/v3/coins/markets?vs_currency=jpy') coingecko = json.loads(body.text) idx = 0 while coingecko[idx]['name'] != self.market: idx += 1 if idx > 100: return "trade_table_err" else: return int(coingecko[idx]['current_price']) 残高の統計情報及び統計グラフをS3バケットへread/writeするクラス s3_deal.py import boto3 import json import datetime class s3_dealer: def __init__(self, bucket = 'nice-hash-graph-backet'): self.datestamp = str(datetime.datetime.now(datetime.timezone(datetime.timedelta(hours=9))).strftime('%Y-%m-%d')) self.s3 = boto3.resource('s3') self.bucket = self.s3.Bucket(bucket) def save_dict_to_s3(self, local_file_path, file_name_base = 'balance_stat_data'): file_name = file_name_base + '_' + self.datestamp + '.json' self.bucket.upload_file(Filename=local_file_path,Key=file_name) print("Completed json object upload to s3...") def save_figure_to_s3(self, local_file_path, file_name_base = 'balance_stat_graph'): file_name = file_name_base + '_' + self.datestamp + '.png' self.bucket.upload_file(Filename=local_file_path,Key=file_name) print("Completed figure upload to s3...") def get_s3_file_item(self, file_name_base = 'balance_stat_graph'): file_name = file_name_base + '_' + self.datestamp + '.png' local_file_path = '/tmp/'+file_name self.bucket.download_file(Filename=local_file_path,Key=file_name) print("Data download from s3 is completed...") return local_file_path LINE通知メッセージの作成するクラス create_message.py import datetime import nicehash import marketrate import nicehash_config as NICEHASHconfig class create_messenger: def __init__(self): self.host = 'https://api2.nicehash.com' self.organisation_id = NICEHASHconfig.organisation_id self.key = NICEHASHconfig.key self.secret = NICEHASHconfig.secret self.market='BTC' def get_balance_and_rate(self): host = 'https://api2.nicehash.com' ### Get mining information from NiceHash API PrivateApi = nicehash.private_api(self.host, self.organisation_id, self.key, self.secret) accounts_info = PrivateApi.get_accounts_for_currency(self.market) balance_row = float(accounts_info['totalBalance']) ### Get currency_to_JPY_rate from CoinGecko API TradeTable = marketrate.trade_table(self.market) rate = TradeTable.get_rate() balance_jpy = int(balance_row*rate) return (balance_jpy,rate) def create_notification_msg(self, df_dict): diff = df_dict['diff'][-1] rate = df_dict['market'][-1] balance = df_dict['balance'][-1] pre_balance = df_dict['balance'][-2] ### Create nortification message time_text = "時刻: " + str(datetime.datetime.now(datetime.timezone(datetime.timedelta(hours=9))))[:19] market_text = "仮想通貨： " + self.market rate_text = "単位仮想通貨価値: " + str(rate) + "円" balance_text = "現在の残高: " + str(balance) + "円" pre_balance_text = "昨日の残高: " + str(pre_balance) + "円" symbol = "+" if diff > 0 else "" diff_txt = "【日次収益: " + str(symbol) + str(diff) + "円】" mon_revenue = "推定月次収益: " + str(diff*30) + "円" ann_revenue = "推定年次収益: " + str(diff*365) + "円" msg = '\n'.join(["",time_text,market_text,rate_text,balance_text,pre_balance_text,diff_txt,mon_revenue,ann_revenue]) return msg 2-1-3-2.Lambda②：nicehash-draw-figure のソース nicehash-draw-figure nicehash-draw-figure/ ├ lambda_function.py ├ plot_stat.py ├ s3_deal.py └ s3_config.py nicehash-balance-notification-LINEによって呼び出されるメインプログラム lambda_function.py import json import plot_stat import s3_deal import s3_config as S3config ### AWS Lambda handler method def lambda_handler(event, context): df_dict = event ### Graph drawing of statistical data Statdrawer = plot_stat.statdrawer() Statdrawer.drawfig(df_dict) S3dealer = s3_deal.s3_dealer(bucket=S3config.bucket) S3dealer.save_figure_to_s3(local_file_path=S3config.figure_file_path,file_name_base=S3config.figure_file_name_base) return event 統計情報をグラフ描画するクラス plot_stat.py import matplotlib.pyplot as plt import datetime import s3_config as S3config class statdrawer: def __init__(self): self.fig = plt.figure() def drawfig(self,df_dict): ax1 = self.fig.add_subplot(111) date1 = [datetime.datetime.strptime(str(s),'%Y-%m-%d') for s in df_dict['date']] ### Balance drawing ln1=plt.bar(date1,df_dict['balance'], width=0.5,linewidth=0.5,label='Balance') h1, l1 = ax1.get_legend_handles_labels() ax1.set_xlabel('Date') ax1.set_ylabel('Balance [yen]') ax1.grid(True) plt.xticks(rotation=45,fontsize=6) ### Adjustment of drawing range ax1_min = min(df_dict['balance']) ax1_max = max(df_dict['balance']) ax1.set_ylim(ax1_min, ax1_max*1.1) ax2_min = min(df_dict['market']) ax2_max = max(df_dict['market']) ax2 = ax1.twinx() ### Market price drawing ln2=plt.plot(date1,df_dict['market'], color="red", linestyle="solid", markersize=8, label='BTC-rate') ax2.set_ylabel('BTC-rate [million]') ax2.set_ylim(ax2_min*0.9, ax2_max*1.01) h2, l2 = ax2.get_legend_handles_labels() ax1.legend(h1+h2, l1+l2, loc='lower left') ### Output of statistical graph self.fig.subplots_adjust(bottom=0.1) self.fig.savefig(S3config.figure_file_path,bbox_inches='tight') 2-1-3-3.コンフィグ 各サービス/外部APIと連携するためにコンフィグに必要な設定値を指定する 下記コンフィグの設定値詳細については、②NiceHashマイニング収益をAWS Lambda×LINE NotifyでLINE通知するを参照。 line_config.py LINE_NOTIFY_ACCESS_TOKEN = '[LINEアクセストークン]' ### ※設定値は2-5節参照 mysql_config.py user='[MySQLアクセスユーザ]' password='[MySQLアクセスユーザpw]' host='[EC2インスタンスの静的IP]' database='[MySQLに構築したDatabase名]' nicehash_config.py organisation_id = '[NiceHash組織ID]' key = '[NiceHash APIアクセスキー]' secret = '[NiceHash APIシークレットアクセスキー]' ### ※設定値は2-2節参照 s3_config.py bucket = '[S3バケット名]' dict_file_name_base = 'balance_stat_data' dict_file_path = '/tmp/balance_stat_data.json' figure_file_name_base = 'balance_stat_graph' figure_file_path = '/tmp/balance_stat_graph.png' ### ※設定値は2-4節参照 2-1-4.Module組み込み 実行に必要なパッケージを取り込む ・Lambda①：nicehash-balance-notification-LINEには「mysql-connector-python」が必要なので、AWS Cloud9上でディレクトリを切って、下記コマンドを実行して環境を整備する。LambdaへのデプロイもCloud9上で行う。 nicehash-balance-notification-LINE ec2-user:~/environment (master) $ mkdir nicehash-balance-notification-LINE ec2-user:~/environment (master) $ cd nicehash-balance-notification-LINE ec2-user:~/environment/nicehash-balance-notification-LINE (master) $ pip install -t ./ mysql-connector-python ・Lambda②：nicehash-draw-figureについても「matplotlib」が必要なので、同様にCloud9上にディレクトリを切って環境を整備しLambdaへデプロイする。 nicehash-draw-figure ec2-user:~/environment (master) $ mkdir nicehash-draw-figure ec2-user:~/environment (master) $ cd nnicehash-draw-figure ec2-user:~/environment/nicehash-draw-figure (master) $ pip install -t ./ matplotlib 2-1-5.基本設定の編集 メモリ/タイムアウトエラーを回避するために基本設定値を変更する ・Lambdaはデフォルトだと、メモリ：128MB、タイムアウト：3秒になっているため、実行状況の様子をみてメモリ「128MB ⇒ 200MB」、タイムアウト「3秒 ⇒ 15秒」程度へ変更しておく。 2-2.NiceHash APIによる収益情報取得 外部APIからマイニング収益情報を取得できるようKEYを取得する ・API Keys取得手順はこちらを参照。 2-3.EventBridgeによるトリガー定義 日次ジョブとしてLambdaをキックするためのトリガーを定義する ・下記トリガーを作成して、Lambda①：nicehash-balance-notification-LINEにアタッチする ルール：「新規ルールの作成」 ルール名：DailyTrigger ルールタイプ：スケジュール式 スケジュール式：cron(0 15 * * ? *) # 毎日0:00に実行するcron 2-4.S3バケットの作成 ファイルの受け渡しを行うS3バケットを用意する ・s3_config.pyに指定したバケットをあらかじめS3上に作成しておく 2-5.LINE NotifyによるLINE通知 AWSへLINE Nortifyを連携するために必要なトークンを発行する ・LINE連携、Access tokenの取得方法については、こちら を参照。 2-6.RDBの作成 収益の統計情報を管理するDBを用意する ・RDBでNiceHash収益の統計情報を管理するために、EC2上にMySQLを導入する 　※Amazon RDSを使うべきだが、料金的な都合からMySQL on EC2で代用 ・Lambda①：nicehash-balance-notification-LINEを配置したVPC上の同サブネットにEC2インスタンスを作成 ・作成したEC2インスタンスにMySQLをインストール 　※MySQLのインストールはこの辺を参照 ・シンプルなテーブル一つで事足りるので、とりあえず下記のようにnicehash_infoテーブルを定義 nicehash_infoテーブル構造 mysql> SHOW COLUMNS FROM nicehash_info; +---------+------+------+-----+---------+----------------+ | Field | Type | Null | Key | Default | Extra | +---------+------+------+-----+---------+----------------+ | id | int | NO | PRI | NULL | auto_increment | | date | date | YES | | NULL | | | balance | int | NO | | NULL | | | diff | int | YES | | NULL | | | market | int | NO | | NULL | | +---------+------+------+-----+---------+----------------+ 5 rows in set (0.00 sec) id：レコードID date：日付 balance：残高 diff：前日残高との差分 market：BTCの円相場 3. 実行結果 ・毎日0:00になると、日次収益と残高/円相場の変遷グラフがLINE通知されるようになりました。 ・5月の暴落が顕著すぎて、分かってたけど悲しくなった。。。 4. 終わりに ・pandas.DataFrameなどは使わずlistやdict等の組み込み関数のみで実装したが、パッケージの都合上、250MB以内には収められなかったのでLambdaのデプロイ上限250MB(圧縮50MB)は結構ボトルネックになると痛感した。。。 ・そもそもLambdaは、機能毎に分割した最小単位で定義してシステムはLambdaを組み合わせて構築するものという前提があるのかと思った。 5. 更新履歴 ver. 1.0 初版投稿 2021/07/24

※記事について著作権等で問題がありましたら、お手数ですがコメントいただけると幸いです。早急に修正か、必要に応じて記事を削除いたします。 AWS初心者による、SAA取得に向けた学習の記録④ の続きです。 関連記事はこちらからどうぞ。 AWS初心者による、SAA取得に向けた学習の記録① AWS初心者による、SAA取得に向けた学習の記録② AWS初心者による、SAA取得に向けた学習の記録③ AWS初心者による、SAA取得に向けた学習の記録④ AWS初心者による、SAA取得に向けた学習の記録⑤ AWS初心者による、SAA取得に向けた学習の記録⑥ ※参考書籍 AWS認定資格試験テキスト AWS認定ソリューションアーキテクト・アソシエイト 改訂版第2段 今回は EBS EFS S3 Storage Gateway FSx についての内容となります。 まずは前提知識として以下の説明をします。 ブロックストレージ ファイルストレージ オブジェクトストレージ ブロックストレージ データを物理的なディスクにブロック単位で管理するストレージ。更新頻度が高い場合や高速アクセスが必要な場合に使用される。 AWSではEBSがブロックストレージに該当。 ファイルストレージ ブロックストレージ上にファイルシステムを構成し、ファイル単位でデータを管理するストレージ。 複数クライアントによるデータ共有や、まとまった過去データの保存が主な用途となる。 AWSではEFSが該当。 オブジェクトストレージ ファイルにメタデータを付与してオブジェクトとして管理するストレージ。 ストレージ内でファイルを直接操作することはできないが、HTTP経由で操作が可能。 更新頻度の低いデータや大容量のデータの保存が主な用途となる。 AWSではS3とS3 Glacierが該当。 EBS（Elastic Block Store） EBS は、AWSが提供するブロックストレージサービス。 機能概要 EC2のOS領域として使用 追加で複数のEBSをEC2にアタッチが可能 RDSのデータ保存が可能 マルチアタッチ機能により複数EC2から同時にアタッチすることも可能だが、制約が多く用途が限定的 作成時にAZを指定する 別のAZのEC2にアタッチする場合はEBSのスナップショットを作成し、そのスナップショットを使用して別AZでEBSボリュームを生成する バックアップも同様の手順でスナップショットから新規EBSボリュームを作成し、EC2にアタッチすることで実現 画像引用元：AWS構成パターン（EBSデータ領域バックアップ） EBSのボリュームタイプ IOPS：性能の指標。1秒あたりに処理できるI/Oアクセス数 汎用SSD（gp2）：最も一般的なSSDベースのボリュームタイプ。 プロビジョンドIOPS SSD（io1）：最も高性能なSSDベースのボリュームタイプ。DBサーバーを構成する場合など、高いIOPS性能が必要な場合に使用。 スループット最適化HDD（st1）：HDDベースのスループット重視のボリュームタイプ。ログデータへの処理やバッチ処理の入力ファイルなど、大容量ファイルを高速に読み取る場合に使用。 Cold HDD（sc1）：性能よりも低コストを重視したボリュームタイプ。利用頻度が低く、低アクセスなデータをシーケンシャルにアクセスする場合やアーカイブとしての用途で使用。 プロビジョンドIOPS以外には、IOPS容量に応じてEBS利用時間の99%を満たすよう設計されている ベースライン性能 と、1TB未満のボリュームには一時的なIOPSの上昇に対応するための バースト機能 がある。 EBSのボリュームは必要に応じてディスクサイズ（最大16TB）の変更が可能だが、拡張後はEC2条でOSに応じたファイルシステムの拡張作業を行い、OS側で認識させる必要がある。 しかしサイズを縮小することは不可能なため、一時的なデータの増加の場合には新規EBSとしてEC2にアタッチする。 また、ボリュームタイプの変更は自由にできる。 暗号化 EBSは、暗号化オプションを有効にしストレージ自体を暗号化することで、ボリュームが暗号化され、そこから取得したスナップショットも併せて暗号化される。 EBSマルチアタッチ 複数のEC2インスタンスから同一のEBSをアタッチ可能。 別AZからはアタッチが不可能。 また、プロビジョンドIOPS SSDボリュームのみ利用可能。 EFS（Elastic File System） EFS は、容量無制限で複数EC2からアクセス可能なファイルストレージサービス。 機能概要 クライアントからEFSへの接続は、NFSクライアントを用意することで実現できる amazon-efs-utils ツールには、EFSへのマウント推奨オプションやトラブルシューティングに役立つログの記録機能がある パフォーマンスモードとスループットモードがある EFSの構成要素 ファイルシステム マウントターゲット セキュリティグループ EFSは、ファイルが作成されると自動で 3か所以上のAZに保存される分散ファイルシステムを構成 する。 このファイルシステムにアクセスするために、AZごとでサブネットを指定し マウントターゲット を作成する。 マウントターゲットにはセキュリティグループを指定できる。 画像引用元：マウント・ターゲットの作成と管理 パフォーマンスモード 汎用パフォーマンスモード：最も汎用的なモード 最大I/Oパフォーマンスモード：EFSへの数百～数千の同時アクセス（ビッグデータ解析アプリの並列処理に使用するデータの参照など） こちらを選択した場合、スループットを最大化する代わりにファイル操作のレイテンシーが微妙に高くなる モードの選択を見分ける指標にはCloudWatchの PercentIOLimit を使用する。 ※パフォーマンスモードは後から変更が不可能。 スループットモード バーストスループットモード：EFSのデータ容量に応じてベースラインとなるスループットが設定されている プロビジョニングスループットモード：バーストで指定されているベースラインスループットを上回るスループットが必要な場合に、任意のスループット値を指定可能 モードの選択を見分ける指標にはCloudWatchの BurstCreditBalance を使用する。 ※プロビジョニングスループットで指定するスループット値は増減が可能。また、プロビジョニングスループットでのスループット値の削減やモードの変更は、前回の作業から24時間開ける必要がある S3（Simple Storage Service） S3 は優れた耐久性を持つ、容量無制限のオブジェクトストレージサービス。 「大容量」「長期保存」「重要」に当てはまるデータの保管に最適。 保存されたデータは複数のAZとAZ内の複数の物理ストレージに複製される。 機能概要 オブジェクトの参照にはHTTPベースのWeb APIを使用 利用者のデータ保存 EBSスナップショットの保存 データのバックアップ ビッグデータ解析用のデータレイク EC2やコンテナからのログ保存先 静的コンテンツのホスティング 簡易的なKey-Value型のDB 構成要素 バケット：オブジェクトを保存する領域。バケット名はAWS内で一意である必要がある オブジェクト：S3に格納されるデータ。キー（オブジェクト名）が付与され、「バケット名＋キー名＋バージョンID」で必ず一意になるURLが作成される。サイズは最大5TB メタデータ：オブジェクトを管理する情報を任意に定義できる ストレージクラス S3標準：デフォルトのストレージクラス S3標準-低頻度アクセス：上記に比べてコストが安価 S3 1ゾーン：単一のAZのみでデータを複製するストレージクラス S3 Intelligent-Tiering：参照頻度を判断できないデータを扱う場合に使用。上記のS3標準とS3標準-低頻度アクセスの2層構成 S3 Glacier：ほとんど参照されないアーカイブ用のデータ保管に使用。オブジェクトの新規作成時に指定はできないが、ライフサイクル管理機能で指定することで利用可能 S3 Glacier Deep Archive：アーカイブ用途のストレージ。Glacierより更にアクセス頻度が低いデータを保管し、安価に利用が可能 オブジェクトの利用頻度に応じてライフサイクル管理を設定することができる 移行アクション：データの利用頻度に応じてストレージクラスを変更するアクション 有効期限アクション：期限を越えたオブジェクトをS3から削除するアクション また、その他にバージョニング機能でオブジェクトを差分管理したり、静的コンテンツに限りWebサイトとしてホスティングしたりすることが可能。 Storage Gateway Storage Gateway は、オンプレミスにあるデータをクラウドに対して連携するためのインターフェースを提供するサービス。 データの保存先にはS3やS3 Glacierが、キャッシュストレージとしてはEBSが利用される。 画像引用元：(レポート) STG311: AWS Storage Gatewayによるセキュアでコスト効率のよりバックアップ #reinvent Storage Gatewayのタイプ ファイルゲートウェイ：S3をクライアントサーバーからNFSマウントして、ファイルシステムのように使用する ボリュームゲートウェイ：各ファイルをオブジェクトとしてではなく、S3のデータ保存領域全体を一つのボリュームとして管理する。ボリュームからスナップショットを取得することができる。 キャッシュ型ボリューム：頻繁に使用するデータはStorage Gateway内のキャッシュディスク（オンプレ）に保存し、全てのデータを保存するストレージ（プライマリストレージ）としてS3を利用する 保管型ボリューム：全てのデータを保存するストレージ（プライマリストレージ）としてローカルストレージを使用し、データをスナップショット形式で定期的にS3へ転送する。このスナップショットはEBSとしてリストア可能。 テープゲートウェイ：テープデバイスの代替としてS3やS3 Glacierにデータをバックアップする FSx FSx はフルマネージドなファイルストレージ。 Amazon FSx for Windows：Windows向けでビジネスアプリで利用される Amazon FSx for Lustre：ハイパフォーマンス向け ファイルシステム作成時にS3のバケットと関連付けされ、自前のファイルように管理することが可能。 機械学習やビッグデータ処理に使用 Linux用のサービスで、専用のクライアントソフトが必要 インストール後は通常のNASのようにマウントして利用可能 終わりに 最後まで読んでいただきありがとうございました。 関連記事はこちらからどうぞ。 AWS初心者による、SAA取得に向けた学習の記録① AWS初心者による、SAA取得に向けた学習の記録② AWS初心者による、SAA取得に向けた学習の記録③ AWS初心者による、SAA取得に向けた学習の記録④ AWS初心者による、SAA取得に向けた学習の記録⑤ AWS初心者による、SAA取得に向けた学習の記録⑥ ※参考書籍 AWS認定資格試験テキスト AWS認定ソリューションアーキテクト・アソシエイト 改訂版第2段

はじめに Amazon Elasticsearch Service 5.3以降では、基本的に1時間ごとに自動でスナップショットを作成し、最大で2週間分(336個)のスナップショットが作成されます。 今回は、インデックスをスナップショットから復元する手順を記述します。 手順 復元するスナップショットを特定する 最初に、復元するスナップショットを特定します。 以下のコマンドを実行することで、スナップショットの情報を取得出来ます。 curl -XGET 'https://{ESエンドポイント}/_snapshot/cs-automated/_all?pretty' terminal $ curl -XGET 'https://{ESエンドポイント}/_snapshot/cs-automated/_all?pretty' { "snapshot" : "2021-06-15t02-19-46.**********-*******-********", "uuid" : "***************", "version_id" : ******, "version" : "6.3.1", "indices" : [ "index-1", "**********", "**********", "**********", "**********", "**********", "**********", ...(省略) }, ...(省略) ここでは、snapshotに記述されている2021-06-15t02-19-46.**********-*******-********をメモしましょう。 cs-autometedとは ここにはスナップショットリポジトリ名を記述するが、自動スナップショットの場合はcs-automatedリポジトリに保存される。 スナップショットリポジトリの一覧を確認したいときは以下のコマンドを実行する。 curl -XGET '{ESエンドポイント}/_snapshot?pretty' 1つのインデックスを復元する 1つのインデックスを復元するには以下のコマンドを実行します。 curl -XPOST 'https://{ESエンドポイント}/_snapshot/cs-automated/{先程メモしたスナップショット名}/_restore?pretty' -d '{"indices": "{インデックス名}"}' -H 'Content-Type: application/json' 以下に、index-1のインデックスを復元する例を示します。 terminal $ curl -XPOST 'https://{ESエンドポイント}/_snapshot/cs-automated/2021-06-15t02-19-46.**********-*******-********/_restore?pretty' -d '{"indices": "index-1"}' -H 'Content-Type: application/json' { "accepted" : true } 全てのインデックスを復元する 全てのインデックスを復元するには以下のコマンドを実行します。 curl -XPOST 'https://{エンドポイント}/_snapshot/cs-automated/{スナップショット名}/_restore?pretty' terminal $ curl -XPOST 'https://{esエンドポイント}/_snapshot/cs-automated/2021-06-15t02-19-46.**********-*******-********/_restore?pretty' { "accepted" : true } 検索可能なドキュメントが増えていることを確認しましょう。 ドキュメントが増えていることが確認出来ました。以上です。 参考 Amazon Elasticsearch Service でのインデックスのスナップショットを作成する - Amazon Elasticsearch Service

Snowflake における AWS PrivateLinkの役割 AWS上のSnowflakeサービス(VPC)に、AWSの閉域からアクセスするための機能です。 自分専用のシングルテナントのSnowflake環境が作成されるものではなく、マルチテナントの環境に閉域でアクセスするための専用の入り口が用意されます。 PrivateLinkを使う場合であっても、インターネットからSnowflakeにアクセスすることができます。 つまりセキュリティ的な観点でいえばPrivateLinkを使ったからといって完全に閉域でSnowflakeを利用できるわけではありません。インターネットからのアクセスを制限したい場合はIPアドレス制限ができるSnowflakeの[ネットワークポリシー]機能を使う必要があります。 AWS PrivateLinkの使用条件は以下の通りです。 SnowflakeをBusinessCritical以上のエディションで利用していること AWS VPCのリージョンとSnowflakeのリージョンが一致していること 大まかな手順 Snowflakeへの利用申請(アクセス元AWSアカウントID、SnowflakeアカウントIDを伝える) AWSでVPCエンドポイントの作成(Snowflakeから指示のあったエンドポイント名を検索して作成) 権威DNSにCNAMEレコードの設定 S3のVPCエンドポイント(ゲートウェイ型)の作成 ３番目の「権威DNSにCNAMEレコードの設定」という部分が気になるところです。 というのもVPCエンドポイントを作成すれば、専用のDNS名が作成されるはずです。 VPCエンドポイントのDNS名にアクセスすればいいはずなのに、なぜCNAMEレコードを追加する必要があるのでしょうか。 公式サイトを見ると、以下のように記載されています。 ①AWSPrivateLinkエンドポイントを介してSnowflakeにアクセスするには、DNSにCNAME記録を作成して、 SYSTEM$GET_PRIVATELINK_CONFIG 関数からの privatelink-account-url 値と privatelink-ocsp-url 値を許可する必要があります。 ②一部のSnowflake機能では、AWSPrivateLinkで機能を使用するために、DNSに追加のCNAME記録が必要です。 Snowflake Data MarketplaceまたはSnowsight：app.<地域ID>.privatelink.snowflakecomputing.com 組織：<組織名>-<アカウント名>.privatelink.snowflakecomputing.com 公式の説明が少し分かりづらいため、番号は私が振りました。 ①が必須であり、②は任意という理解であっているようです。 ①の必須レコードを使わず、VPCエンドポイント名でSnowflakeにアクセスしたらどうなるのでしょうか。そのへんも検証してみます。 ②のSnowsightはクエリ結果の単純な可視化ができて便利なので登録してみます。組織機能はまだ使っていないので登録は見合わせます。 手順の詳細は公式の以下ページもご確認ください。 具体的な手順 では検証していきます。 まず、SnowflakeのアカウントURLについて少し説明します。 Snowflakeのアクセス用URLは上記リンクにあるとおり、(「組織」機能を使っていない場合は)以下のような形式になっており、このアカウントロケータがユーザ固有の名前になっています。 https://<account_locator>.<region_name>.snowflakecomputing.com ①Snowflakeへの利用申請 Snowflake Communityアカウントにログインし、サポートケースを起票します。 サポートケースの起票の仕方は以下のページを参考にしてください。 サポートケースのDescriptionでは以下の情報を記載します。 SnowflakeアカウントのURL AWSのアカウントID(12桁) AWSのアカウントのリージョン 申請後、最大2営業日でサポートケースからのメールが来ます。(私の場合は翌営業日にメールが来ました) メッセージには以下のようにselect文で情報を取得してね、と記載されています。 You can get the VPCE ID by running the system function SYSTEM$GET_PRIVATELINK_CONFIG. Snowflakeのワークシートにてクエリを実行します。 以下のようなJSONデータが出力されます。 上記のデータを分解すると、以下のようなデータになります。 # Key Value 備考 1 regionless-privatelink_ocsp-url ocsp.ocsp-xxxxxxx-xxxxxxx.privatelink.snowflakecomputing.com 2 privatelink-account-name xx12345.ap-northeast-1.privatelink 3 privatelink-vpce-id com.amazonaws.vpce.ap-northeast-1.vpce-svc-xxxxxxxxxxx VPCエンドポイントで利用 4 privatelink-account-url xx12345.ap-northeast-1.privatelink.snowflakecomputing.com DNSにCNAME登録必須① 5 regionless-privatelink-account-url xxxxxxx-xxxxxxx.privatelink.snowflakecomputing.com 6 privatelink_ocsp-url ocsp.xx12345.ap-northeast-1.privatelink.snowflakecomputing.com DNSにCNAME登録必須② 7 privatelink-connection-urls [] privatelink-vpce-idの値をこのあとのVPCエンドポイントの検索で利用するためメモしておきます。 ②AWSでVPCエンドポイントの作成 Snowflakeに申請したAWSのアカウントにログインします。マネジメントコンソールから[VPC]->[エンドポイント]を選択し、[エンドポイントの作成]をクリックすると、以下の画面が表示されます。 [サービスを名前で検索]を選択し、[サービス名]に先程メモしておいたprivatelink-vpce-idの値を入力すると、正常に申請が完了していればサービス名が発見できます。 紐付けたいVPCとサブネットを選択します。 VPCエンドポイントのセキュリティグループについては適切に変更してください。 https://docs.snowflake.com/ja/user-guide/admin-security-privatelink.html#step-1-create-and-configure-a-vpc-endpoint-vpce 443: 一般的なSnowflakeのトラフィックに必要です。 80 ：このポートですべてのSnowflakeのクライアント通信をリッスンする、Snowflake OCSP キャッシュサーバーで必要です。Snowflakeの顧客データが port 80 に流出することはありません。 VPCエンドポイントのステータスが「保留中」から「使用可能」になったことを確認します。 使用可能となったあと、対象のVPCエンドポイントを選択した際に右下に表示されるDNS名をメモします。 VPC単位のDNS名、サブネット単位のDNS名等が表示されます。ここでは一番上に表示されるVPC単位のDNS名をメモします。 このDNS名をのちほどDNSのCNAMEレコードの値として登録します。 ③権威DNSにCNAMEレコードの設定 今回のテストではSnowflakeにPrivateLinkでアクセスするクライアントはAWS上にあるEC2を想定しています。 この環境ではWindowsServerをシンプルに立てただけですので、EC2のDNSリゾルバはAmazonProvidedDNS(最近の呼称でいうとRoute 53 Resolver )を指定しています。 AmazonProvidedDNSはRoute53を参照してくれます。 そのためこれからCNAMEレコードを追加する権威DNSは、Route53を利用します。 ユーザの利用環境によってはクライアントが参照するDNSリゾルバはRoute53ではない場合もありますので、その場合は手順を読み替えてください。 Route53でゾーンを新規作成します。先程のJSONデータのアカウントURLの以下のドメイン部分のプライベートホストゾーンを作成します。 ap-northeast-1.privatelink.snowflakecomputing.com これはパブリックゾーンではないため、snowflakecomputing.comからのDNS委任は必要ありません。あくまでプライベートネットワークのみで通用するゾーンになります。 ホストゾーンを紐付けるVPCは、SnowflakeのPrivateLinkを展開したVPCを選択します。 プライベートホストゾーンが作成されました。 繰り返しますがプライベートゾーンであるためゾーン委任は必要なく、NSレコードも変更する必要はありません。 ではレコードを登録していきます。 # レコード名 レコードタイプ 値 備考 1 (アカウントロケータ) CNAME (VPCエンドポイントのDNS名) 必須レコード① 2 ocsp.(アカウントロケータ) CNAME (VPCエンドポイントのDNS名) 必須レコード② 3 app CNAME (VPCエンドポイントのDNS名) Snowflake Data MarketplaceまたはSnowsightで利用 4 (org_name)-(account_name) CNAME (VPCエンドポイントのDNS名) 「組織」機能。今回有効化していないため設定しない 1つ目のレコードを登録します。 レコード名 レコードタイプ 値 備考 (アカウントロケータ) CNAME (VPCエンドポイントのDNS名) 必須レコード① アカウントロケータとは先に説明したとおり、アカウントURLのうち、以下の太字の部分を指します。 xx12345.ap-northeast-1.privatelink.snowflakecomputing.com このアカウントロケータは例ですので、値は環境毎で読み替えてください。 2つ目のレコードを登録します。 レコード名 レコードタイプ 値 備考 ocsp.(アカウントロケータ) CNAME (VPCエンドポイントのDNS名) 必須レコード② こちらはJSONで取得した[privatelink_ocsp-url]の値の以下の太字の部分をレコードで指定します。 ocsp.xx12345.ap-northeast-1.privatelink.snowflakecomputing.com このアカウントロケータは例ですので、値は環境毎で読み替えてください。 3つ目のレコードを登録します。 レコード名 レコードタイプ 値 備考 app CNAME (VPCエンドポイントのDNS名) Snowflake Data MarketplaceまたはSnowsightで利用 最終的にこのようなレコードになりました。 ④S3のVPCエンドポイント(ゲートウェイ型)の作成 こちらは省略します。 公式には以下のように記載されています。 Snowflakeクライアント（SnowSQL、 JDBC ドライバー、 ODBC ドライバーなど）は、さまざまなランタイム操作を実行するためにS3へのアクセスを必要とします。 PrivateLink VPC ネットワークはインターネット経由のS3アクセスを許可しないため、Snowflakeクライアントに必要なS3ホスト名に1つ以上のゲートウェイエンドポイントを構成する必要があります。 このステップは、SnowflakeクライアントからのS3トラフィックが AWS バックボーンにとどまるために必要です。 SnowflakeクライアントがS3にアクセスすることがあるようです。 その経路も閉域網化したい場合は、Snowflakeクライアントが存在するVPC上にS3のゲートウェイ型VPCエンドポイントを置くようです。 ただしゲートウェイ型のエンドポイントはVPC外からアクセスすることができないため、あくまでSnowflakeクライアントがAWSのVPC内にあることを想定した手順のようです。 今回検証したEC2のサブネットにはすでにVPCエンドポイントのS3ゲートウェイ型が設置済みでした。 もしVPC外からアクセスした場合、S3の経路だけインターネットで接続できればいいということかもしれませんが、そこまでは検証していません。 動作確認 CLIによるテスト まず、今回SnowflakeクライアントとしているEC2で、VPCエンドポイントのDNS名がプライベートIPで名前解決できるか確かめます。 $ nslookup vpce-xxxxxxxxxx-xxxxxxxx.vpce-svc-xxxxxxxxxxxxxxxxx.ap-northeast-1.vpce.amazonaws.com CNAMEで登録したレコードも名前解決でプライベートIPが返ってくるか試します。 $ nslookup xxxxxxxx.ap-northeast-1.privatelink.snowflakecomputing.com 問題なければ、まずはVPCエンドポイントのDNS名で、ワークシートのURLにたいしてcurlコマンドでHTTPアクセスを試してみます。 $ curl https://vpce-xxxxxxxxxx-xxxxxxxx.vpce-svc-xxxxxxxxxxxxxxxxx.ap-northeast-1.vpce.amazonaws.com/console#/internal/worksheet curl: (51) SSL: no alternative certificate subject name matches target host name 'vpce-xxxxxxxxxx-xxxxxxxx.vpce-svc-xxxxxxxxxxxxxxxxx.ap-northeast-1.vpce.amazonaws.com' SSL/TLS証明書エラーで怒られてしまいましたね。 ホスト名が一致せず、このようなエラーが出る場合、証明書を自分で持っている場合は証明書側のSAN(Subject Alternative Name)に代替ホスト名を登録するような対応をしますが、証明書管理はSnowflake側なのでどうにもしようがないです。 苦し紛れに、警告を無視するオプション -k をつけて試してみます。 $ curl -k https://vpce-xxxxxxxxxx-xxxxxxxx.vpce-svc-xxxxxxxxxxxxxxxxx.ap-northeast-1.vpce.amazonaws.com/console#/internal/worksheet <html> <head><title>404 Not Found</title></head> <body> <center><h1>404 Not Found</h1></center> </body> </html> 404 Not Foundになってしまいました。 ではCNAME登録した名前でアクセスしてみます。 $ curl https://xxxxxxxx.ap-northeast-1.privatelink.snowflakecomputing.com/console#/internal/worksheet <!DOCTYPE html> <html> <head> <meta http-equiv="X-UA-Compatible" content="IE=edge" /> <meta charset="UTF-8" /> <meta (snip) コンテンツが正常に取得できました。 ブラウザによるテスト 今度は同じEC2でブラウザからのアクセスを試してみます。 VPCエンドポイントのDNS名によるテスト まずはVPCエンドポイントのDNS名でアクセスしてみます。 証明書不一致警告が出たため、不一致を無視してアクセスしてみると、、 404 Not Foundでアクセスに失敗しました。 この結果から推測すると、SnowflakeのWebサーバは(apacheでいう)VirtualHostのような機能をつかってアクセス時のドメイン名をチェックし、それでユーザ毎のテナントに振り分けていると考えられます。 そう考えると当初の疑問、「なぜCNAMEレコードを追加する必要があるのか」は、「SnowflakeのPrivateLinkを使う上で必須だから」といえそうです。 CNAMEを設定したDNS名によるテスト Snowflakeコンソール 以下のようなURLにアクセスすると、問題なくログイン画面が表示され、ログインもできました。 https://xx12345.ap-northeast-1.privatelink.snowflakecomputing.com/console#/internal/worksheet Snowsightコンソール Snowsightも試してみます。 Snowflakeコンソールから「アプリをプレビュー」をクリックします。 CNAMEに登録したドメイン名にリダイレクトされます。 https://app.ap-northeast-1.privatelink.snowflakecomputing.com/ap-northeast-1.aws/(アカウントロケータ) サインインすると、 Snowsightに正常にアクセスできました。 Snowsightのチャートも正常に表示されます。 システム連携テスト(Tableau) AWS内のWindowsServerからテストするため、SnowflakeのODBCドライバをダウンロードしインストールします。 インストール後、Windowsの[ODBC データソースアドミニストレーター]を起動し、セットアップします。 CNAMEを設定したDNS名によるテスト ODBCドライバのセットアップで指定するServerはCNAME登録済みのドメイン（以下例）です。 xx12345.ap-northeast-1.privatelink.snowflakecomputing.com ODBCドライバをセットアップしたあとは、TableauのコネクタでSnowflakeを選択すればTableau専用の接続画面が出てきます。 Tableauの[サーバー]欄は先程ODBCドライバに入力したものと同じものを指定します。 xx12345.ap-northeast-1.privatelink.snowflakecomputing.com 無事プライベートリンクのドメイン名で接続ができました。 接続時の[役割]でACCOUNT ADMINを指定したため、SNOWFLAKEデータベースも見えています。 まとめ 以下の検証を行い、正常にアクセスできることを確認しました。 CNAMEを設定したDNS名でSnowflakeコンソールにブラウザから閉域接続できること CNAMEを設定したDNS名でSnowsightコンソールにブラウザから閉域接続できること CNAMEを設定したDNS名でTableauからSnowflakeにシステム的な閉域接続できること CNAMEの登録せずにVPCエンドポイントのDNS名を使うと正常にアクセスができないこと(404 NotFoundになる) 今回の例では AWS(EC2) -> AWS(Snowflake)の通信でしたので、以下の公式FAQにあるとおり、仮にグローバルIPでアクセスしたとしてもAWS構内経路で完結するはずです(試してはいません)。しかしアクセス元がDirectConnectを通したオンプレにあるような場合では、やはりこのPrivteLinkの機能で閉域を通すのは有効な方法ではあります。 Q:2 つのインスタンスがパブリック IP アドレスを使用して通信する場合、またはインスタンスが AWS のサービスのパブリックエンドポイントと通信する場合、トラフィックはインターネットを経由しますか? いいえ。パブリックアドレススペースを使用する場合、AWS でホストされているインスタンスとサービス間のすべての通信は AWS のプライベートネットワークを使用します。AWS ネットワークから発信され、AWS ネットワーク上の送信先を持つパケットは、AWS 中国リージョンとの間のトラフィックを除いて、AWS グローバルネットワークにとどまります。 以上です。