[AWS Cloud9]EC2自動停止を検知して別の処理を始める

はじめに

AWS Cloud9はIDEのブラウザを閉じたとき、自動でEC2インスタンスが停止する機能があります。
何分後に、自動停止するかは、設定から変更が可能です。
1.PNG

AWS Cloud9を1週間使うと、EC2インスタンスが自動停止するのが「便利 → 当たり前」という感覚になりました。そして、思いました。自動停止を検知してもっと楽できないかと。

やりたいこと

AWS Cloud9の機能で

  • 検知した情報をトリガーに別の処理を動作させてみる。
  • 今回は、サンプルに以下を動作確認
    • 自動停止のお知らせメール (実装したら、イチイチお知らせが来てウザくなった)
    • RDSの自動停止 (IDEとセットで使ってたので、止め忘れがなくなり非常に便利になった)

自動停止したことを検知

CloudWatchのルールを作成。

  • イベントパターンを選択
  • サービス名:EC2
  • イベントタイプ:EC2 Instance Status-change Notification
  • 特定の状態:インスタンスの停止を設定
  • Specific instance Id(s):AWS Cloud9のインスタンスIDを設定
    2.PNG
{
  "source": [
    "aws.ec2"
  ],
  "detail-type": [
    "EC2 Instance State-change Notification"
  ],
  "detail": {
    "state": [
      "running",
      "stopped"
    ],
    "instance-id": [
      "AWS Cloud9のインスタンスID"
    ]
  }
}

検知した情報をトリガーに別の処理を動作

ターゲットに今回は2つ(SNS(メール送信)とLambda(rds停止)を設定しました。)
SNSは使い道を思いつかなかったのでEメールを設定したのみ。jsonがそのままメールに飛んでくる。

3.PNG

LambdaはついでにIDE起動したときの動作も記述。

  • IDE起動→EC2開始→RDS開始
  • IDE終了→30分後EC2停止→RDS停止
IAMロール
{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Action": [
                "logs:CreateLogGroup",
                "logs:CreateLogStream",
                "logs:PutLogEvents",
                "rds:StartDBInstance",
                "rds:StopDBInstance"
            ],
            "Effect": "Allow",
            "Resource": "*"
        }
    ]
}
lambda_function.py
import boto3

def lambda_handler(event, context):
    dbinstance = 'RDSインスタンス名'
    rds = boto3.client('rds')
    instanceState = event['detail']['state']
    if instanceState == 'stopped':
        result = rds.stop_db_instance(DBInstanceIdentifier = dbinstance)
    else:
        result = rds.start_db_instance(DBInstanceIdentifier = dbinstance)
    print(result)    
  • 開始状態のRDS

4.PNG

  • EC2が停止するイベントが発生

5.PNG

  • RDSが自動で停止
    6.PNG

まとめ

  • CloudWatchを使うことでIDEの終了後、EC2自動停止を検知できました。
  • 後続処理を自動化できることが分かったので、一日の作業終了をブラウザを閉じるだけに出来そうです。

関連

続きを読む

CloudWatchでエラーログの内容を通知させたい

前置き

CloudWatch Logsの収集対象としているログにErrorという文字列が出力されたらSNSで通知したい。
CloudWatch Logsのロググループにメトリクスフィルタを設定し、このメトリクスが所定のしきい値を超えたらSNSへ連携するアクションをCloudWatch Alarmとして設定すれば実現できる。しかしこの方法だとErrorという文字列が出力されたことは認識できるが、ログの内容までは通知されない。

CloudWatch Alarm

まず、CloudWatch AlarmからSNSに渡される情報を見てみよう。

message.json
{
    "AlarmName": "sample-error",
    "AlarmDescription": "sampleでエラーが発生しました。",
    "AWSAccountId": "xxxxxxxxxxxx",
    "NewStateValue": "ALARM",
    "NewStateReason": "Threshold Crossed: 1 datapoint [2.0 (29/11/17 01:09:00)] was greater than or equal to the threshold (1.0).",
    "StateChangeTime": "2017-11-29T01:10:32.907+0000",
    "Region": "Asia Pacific (Tokyo)",
    "OldStateValue": "OK",
    "Trigger": {
        "MetricName": "sample-metric",
        "Namespace": "LogMetrics",
        "StatisticType": "Statistic",
        "Statistic": "SUM",
        "Unit": null,
        "Dimensions": [],
        "Period": 60,
        "EvaluationPeriods": 1,
        "ComparisonOperator": "GreaterThanOrEqualToThreshold",
        "Threshold": 1,
        "TreatMissingData": "- TreatMissingData:                    NonBreaching",
        "EvaluateLowSampleCountPercentile": ""
    }
}

たしかに、ログの内容は全く含まれていない。その代わりTrigger内にこのアラームに紐づくメトリクスの情報が含まれている。これを使って欲しい情報を辿っていく。

メトリクスフィルタ

MetricNameNamespaceを指定すればメトリクスフィルタの情報を取得することができる。

sample.py
        logs = boto3.client('logs')

        metricfilters = logs.describe_metric_filters(
            metricName = message['Trigger']['MetricName'] ,
            metricNamespace = message['Trigger']['Namespace']
        )

取得したメトリクスフィルタはこんなかんじ。

metricsfilters.json
{
  "metricFilters": [
    {
      "filterName": "sample-filter",
      "filterPattern": "Error",
      "metricTransformations": [
        {
          "metricName": "sample-metric",
          "metricNamespace": "LogMetrics",
          "metricValue": "1"
        }
      ],
      "creationTime": 1493029160596,
      "logGroupName": "sample-loggroup"
    }
  ],
  "ResponseMetadata": {
    "RequestId": "xxxxxxxxxxxxxxxxxxxxxxxxxxx",
    "HTTPStatusCode": 200,
    "HTTPHeaders": {
      "x-amzn-requestid": "xxxxxxxxxxxxxxxxxxxxxxxxxxx",
      "content-type": "application/x-amz-json-1.1",
      "content-length": "210",
      "date": "Wed, 29 Nov 2017 01:10:33 GMT"
    },
    "RetryAttempts": 0
  }
}

filterPatternlogGroupNameが取得できる。これがあればCloudWatch Logsからログイベントを抽出できそうだ。

開始時刻と終了時刻

ログデータを抽出するためにはfilterPatternlogGroupNameのほかに、フィルタリング対象の開始時刻と終了時刻を指定したい。この時刻はUNIX Timeである必要がある。こちらのサイトを参考にさせていただく。

sample.py
        #ログストリームの抽出対象時刻をUNIXタイムに変換(取得期間は TIME_FROM_MIN 分前以降)
        #終了時刻はアラーム発生時刻の1分後
        timeto = datetime.datetime.strptime(message['StateChangeTime'][:19] ,'%Y-%m-%dT%H:%M:%S') + datetime.timedelta(minutes=1)
        u_to = calendar.timegm(timeto.utctimetuple()) * 1000
        #開始時刻は終了時刻のTIME_FROM_MIN分前
        timefrom = timeto - datetime.timedelta(minutes=TIME_FROM_MIN)
        u_from = calendar.timegm(timefrom.utctimetuple()) * 1000

ログイベントの取得

これで材料が揃った。CloudWatch Logsからログイベントを取得する。

sample.py
        response = logs.filter_log_events(
            logGroupName = loggroupname ,
            filterPattern = filterpattern,
            startTime = u_from,
            endTime = u_to,
            limit = OUTPUT_LIMIT
        )

responseはこんなかんじ。

response.json
{
  "events": [
    {
      "logStreamName": "sample-stream",
      "timestamp": 1511942974313,
      "message": "Errorが発生しました。sample messageです。",
      "ingestionTime": 1510943004111,
      "eventId": "11111111111111111111111111111111111111111111111111111112"
    },
    {
      "logStreamName": "sample-stream",
      "timestamp": 1511942974443,
      "message": "またまたErrorが発生しました。sample messageです。",
      "ingestionTime": 1510943004111,
      "eventId": "11111111111111111111111111111111111111111111111111111112"
    }
  ],
  "searchedLogStreams": [
    {
      "logStreamName": "sample-stream",
      "searchedCompletely": true
    }
  ],
  "ResponseMetadata": {
    "RequestId": "xxxxxxxxxxxxxxxxxxxxxx",
    "HTTPStatusCode": 200,
    "HTTPHeaders": {
      "x-amzn-requestid": "xxxxxxxxxxxxxxxxxxxxxx",
      "content-type": "application/x-amz-json-1.1",
      "content-length": "1000",
      "date": "Wed, 29 Nov 2017 01:10:33 GMT"
    },
    "RetryAttempts": 0
  }
}

フィルタ条件にマッチする複数件のログイベントを取得できる。

メッセージの整形

仕上げに、通知する文面の整形を行う。またまた先程のサイトを参考にさせていただく。

sample.py
        #メッセージの整形
        log_message = u""
        for e in response['events']:
            #UNIX時刻をUTCへ変換後、日本時間に変更している
            date = datetime.datetime.fromtimestamp(int(str(e['timestamp'])[:10])) + datetime.timedelta(hours=9)
            log_message = log_message + 'n' + str(date) + ' : ' + e['message']

        #SNSのタイトル、本文
        title = message['NewStateValue'] + " : " + message['AlarmName']
        sns_message = message['AlarmDescription'] + 'n' + log_message

最終的な構成

CloudWatch Alarm -> SNS -> Lambda -> SNS
一つ目のSNSはLambdaファンクションを呼び出すためのもの。二つ目のSNSはいい感じに整形した文面を関係者へ通知するためのもの。

スクリプト

サンプルとして拙いスクリプトを挙げておく。例外処理がテキトウなのはご愛嬌。

lambda_function.py
# -*- coding: utf-8 -*-
import boto3
import json
import datetime
import calendar

#通知先SNSトピックのARN
TOPIC_ARN = "arn:aws:sns:ap-northeast-1:xxxxxxxxxxxx:topic-name"
#抽出するログデータの最大件数
OUTPUT_LIMIT=5
#何分前までを抽出対象期間とするか
TIME_FROM_MIN=10

sns = boto3.client('sns')

def lambda_handler(event, context):
    message = json.loads(event['Records'][0]['Sns']['Message'])

    #SNSフォーマットの作成
    try:
        #CloudWatchのリージョン情報は message['Region'] に含まれる。
        #もしLambdaのリージョンと異なる場合は考慮が必要だが、ここでは同一リージョンを想定する。
        #(逆に言うと複数リージョンのCloudWatchAlarmを単一リージョンのLambdaが捌くことも可能)
        logs = boto3.client('logs')
        #logs = boto3.client('logs',region_name='xxxxxxxx') #←リージョンを指定する場合

        #MetricNameとNamespaceをキーにメトリクスフィルタの情報を取得する。
        metricfilters = logs.describe_metric_filters(
            metricName = message['Trigger']['MetricName'] ,
            metricNamespace = message['Trigger']['Namespace']
        )

        #ログストリームの抽出対象時刻をUNIXタイムに変換(取得期間は TIME_FROM_MIN 分前以降)
        #終了時刻はアラーム発生時刻の1分後
        timeto = datetime.datetime.strptime(message['StateChangeTime'][:19] ,'%Y-%m-%dT%H:%M:%S') + datetime.timedelta(minutes=1)
        u_to = calendar.timegm(timeto.utctimetuple()) * 1000
        #開始時刻は終了時刻のTIME_FROM_MIN分前
        timefrom = timeto - datetime.timedelta(minutes=TIME_FROM_MIN)
        u_from = calendar.timegm(timefrom.utctimetuple()) * 1000

        #ログストリームからログデータを取得
        response = logs.filter_log_events(
            logGroupName = metricfilters['metricFilters'][0]['logGroupName'] ,
            filterPattern = metricfilters['metricFilters'][0]['filterPattern'],
            startTime = u_from,
            endTime = u_to,
            limit = OUTPUT_LIMIT
        )

        #メッセージの整形
        log_message = u""
        for e in response['events']:
            #UNIX時刻をUTCへ変換後、日本時間に変更している
            date = datetime.datetime.fromtimestamp(int(str(e['timestamp'])[:10])) + datetime.timedelta(hours=9)
            log_message = log_message + 'n' + str(date) + ' : ' + e['message']

        #SNSのタイトル、本文整形
        title = message['NewStateValue'] + " : " + message['AlarmName']
        sns_message = message['AlarmDescription'] + 'n' + log_message

    except Exception as e:
        print(e)
        sns_message = message
        title = "error"

    #SNS Publish
    try:
        response = sns.publish(
            TopicArn = TOPIC_ARN,
            Message = sns_message,
            Subject = title
        )

    except Exception as e:
        print(e)
        raise e

参考文献

続きを読む

AWS SageMakerをコンソールから試したらハマったのでメモ

はじめに

先週のRe:Invent2017でリリースされたAWS SageMakerを早速試してみました。
AWSのドキュメントでは
ノートブックインスタンスを作成し、コマンドでその後の手順を実行していますが、
今回はコンソールから試してみました。けっこうハマったのでメモです。

使用データ

今回はモデルデータとしてirisのデータセットを使用。
こちらから生データを入手しました。

やってみる

SageMakerでは、

  1. ノートブックインスタンスの作成
  2. ジョブの作成
  3. モデルの作成
  4. エンドポイントの作成

の順で実施していけばいいらしい。

sagemaker1.png

ノートブックインスタンスの作成

設定項目をいい感じで入力すると、jupyter notebookのインスタンスが作成される。
この中でデータの整形だったりの前準備をするといいらしい。

今回はpythonを書いて行うような前処理もないので、ここは割愛。

ジョブの作成

早速、ハマる。

今回、K-Meansを試してみる。
この他にも全10種類ほどのアルゴリズムが用意されている。
また、自前のDockerイメージを使えば、独自のアルゴリズムを搭載することもできる。

設定をポチポチ入力していく。

sagemaker8.png

アルゴリズムの調整パラメータもポチポチ入力していく。
ここは選択したアルゴリズムによって変わる。

sagemaker9.png

で、次がハマりポイント。
コンテンツタイプをtest/csvみたいに誤字ったり
S3の場所設定をミスったりすると、
わけの分からないエラーがでる。
(新しいサービスだからエラーメッセージが追いついてない…?)

ちゃんと入力するとこんな感じです。

sagemaker10.png

モデルの作成

ここでもちょいハマり。
ジョブの作成時に結果としてS3に格納されるのは、
学習結果のパラメータ(アーティファクト)のみなので、
トレーニングに用いた推論の本体部分イメージが別途必要となる。

なので、
先ほど作成したジョブの詳細から、トレーニングイメージのアドレスをコピーして、
sagemaker13.png

モデルの設定画面に入力する。
(私は空のECRレポジトリのアドレスを入力し続けてハマりました。。)
sagemaker14.png

エンドポイントの作成

最後にエンドポイント設定の作成、エンドポイントの作成を行う。
エンドポイントの作成が完了すると、以下のようにAPIが作成され、
テストデータをPOSTすることで、結果を得ることができる。
sagemaker21.png

エンドポイントがInServiceの間は課金が続くが、
現状、エンドポイント停止の操作ができないので、
使用しなくなったら一旦エンドポイントを消去し、
使うときに再び、エンドポイント設定からエンドポイントを作成するのが良さそう…?

おわりに

今回はAWS SageMakerを試してみました。
機械学習をちょっと導入してみたい方にはとてもおすすめなサービスだと思います。

また、Lambdaにデフォルトで導入されているboto3は
まだSageMakerに対応していないので、
Lambdaを使ってPOSTするときは、最新のboto3を自分で導入する必要があります。

次は自前Dockerイメージからの作成を試してみたいと思います。では。

続きを読む

Amazon SESで受信したメールを転送したい

前置き

SESで受信する特定のアドレス宛のメールを外部の別アドレスへ転送したい。SESの設定でこのへんをポチポチっとやれば…と思ったが、できない。できそうでできない。
そう、SESの標準機能ではメールの転送はできない。

対応方針

思い当たる案は2つ。
1. 受信メッセージをS3へ出力し、それをLambdaで拾って別アドレスへ送信する案
2. Amazon WorkMailを使う案
Lambdaを使う案が主流かもしれないが、自分の場合は品質とスピードを重視して結果的には案2を採用した。

案1 S3への出力とLambdaを使う

SESの受信ルールで、受信メッセージをS3へ出力し、それをLambdaでどうにかして別アドレスへメール送信するのが定石のようだ。

こういった情報を参考にプログラムを作成することになるのだが、文字コードや添付ファイルの有無、メッセージを編集してから転送したい、などいろいろ考慮していくとMIMEの深遠なる世界に迷い込んでなかなか大変なことになる。

加えて、SESのS3アクションで「Encrypt Message」オプションを有効化するとS3に格納するメッセージデータが暗号化されるが、それをLambda(Python)側で複合する方法がわからずハマった。というか今も解決できていない。(KMSも絡んで話が長くなるのでこれについてはあらためて別の記事にしたい。)
公式ドキュメント

スクリプト

lambda_function.py
# -*- coding: utf-8 -*-
import boto3
import json
import re
import os

#転送先アドレス(カンマ区切りで複数指定)
FORWARD_TO = os.environ['forward_to'].split(",")

#メール保存先バケット名
S3_BUCKET = "s3-bucket-name"

#転送メールの送信元ドメイン名
DOMAIN_NAME = "hogehoge.jp"

s3  = boto3.client('s3')
ses = boto3.client('ses', region_name="us-east-1")

def lambda_handler(event, context):
    #本来の送信者
    MAIL_SOURCE = event['Records'][0]['ses']['mail']['source']

    #転送メールの送信者
    MAIL_FROM = MAIL_SOURCE.replace('@','=') + "@" + DOMAIN_NAME

    #メール保存先のフォルダ名
    S3_OBJECT_PREFIX = event['Records'][0]['ses']['receipt']['recipients'][0].split("@")[0] + "/"

    #S3上のメールファイル
    s3_key = S3_OBJECT_PREFIX + event['Records'][0]['ses']['mail']['messageId']

    #メールファイル取得
    try:
        response = s3.get_object(
            Bucket = S3_BUCKET,
            Key    = s3_key
        )
    except Exception as e:
        raise e

    #メールヘッダの書き換え
    try:
        replaced_message = response['Body'].read().decode('utf-8')
        replaced_message = re.sub("\nTo: .+?\n", "\nTo: %s\n" % ", ".join(FORWARD_TO), replaced_message,1)
        replaced_message = re.sub("\nFrom: .+?\n", "\nFrom: %s\n" % MAIL_FROM, replaced_message,1)
        replaced_message = re.sub("^Return-Path: .+?\n", "Return-Path: %s\n" % MAIL_FROM, replaced_message,1)
    except Exception as e:
        raise e

    #メール送信
    try:
        response = ses.send_raw_email(
            Source = MAIL_FROM,
            Destinations= FORWARD_TO ,
            RawMessage={
                'Data': replaced_message
            }
        )
    except Exception as e:
        raise e

send_raw_emailを使い、受信したメールのBODYをそのまま送信するのであれば比較的シンプルに実装できそうだ。ただし、一般的なメール転送でよくあるような、メール本文の冒頭に元メールのヘッダ情報を追記するようなことは行っていないため、元メールの送信者がわからないという問題がある。メール本文を編集しようとするとsend_emailを使えば良さそうだが、文字コードやHTMLメール、添付ファイルの考慮など、前述のとおりディープな世界へ足を踏み入れることになり、大変。なのでここでは諦めてシンプルに。

FORWARD_TO = os.environ['forward_to'].split(",")
転送先のメールアドレスはLambdaの環境変数を使って設定する。カンマ区切りの文字列で複数を指定することも可能にしている。

MAIL_FROM = MAIL_SOURCE.replace('@','=') + "@" + domain_name
ここでひと工夫。元メールの差出人アドレスを転送メールの差出人アドレス内に埋め込む。

S3_OBJECT_PREFIX = event['Records'][0]['ses']['receipt']['recipients'][0].split("@")[0] + "/"
SESがS3へ出力する際のPrefixに合わせていれば何でも良いが、ここでは受信アカウントをPrefixとしている。

案2 Amazon WorkMailを使う

月額4USDのコストが許容できるのであればSESの受け口としてWorkMailが使える。WorkMailには転送機能がある。

WorkMailのセットアップ方法は以下がわかりやすい。
AWS WorkMailを使ってみたら想像以上に便利だった

WorkMailでの転送設定は以下(英語の公式ドキュメント)を参考に。
How do I set up an email forwarding rule in Amazon WorkMail?

Lambdaで実現できるはずの機能を月額4USDで逃げるのはエンジニアとして負けた気がしてしまうが・・・スピードと確実性を優先するならアリかと。

続きを読む

api-gateway+lambdaからlambda・AWS batchを非同期でキックしてみた記事

最近はやりのchatopsをrubyで書いていたのですが、botが死ぬたびに原因調査を行うのがめんどくさくて、slackでメッセージを受け取るところと内部の処理を分離しました。
分離した処理はjenkinsのapiを使って処理していたのですが、せっかくなのでAWSに全てあげてしまうことにしました。
そこでlambdaからLambda・AWS Batchを非同期でキックした時のことを書こうと思います。

はじめに

本稿はslackにbotを多数動かす想定で設計しているので、個人でちゃちゃっと作りたい人はbotで完結したほうがいいと思います。
またlambdaの非同期でキックするところを中心に書くのでbot部分や周辺は割愛します。
(litaをECS上で動作 passなどはkmsで管理)

想定ユーザー

  • データ基盤の機能などの一部を社内ユーザーに解放したい
  • いくつかのbotを動かしたい
  • terraformで環境構築している
  • サーバーレスが好き

全体的な構成

以下構成図になります
– メッセージは全てlita一台で受け取ってApi-GatewayにPOST
– LambdaからLambdaかAWS batchを非同期で起動

スクリーンショット 2017-12-05 12.42.21.png

post.json
{
    "method":{
         "channel_name" : "****",
         "req_usr" : "***",
         "magic_words" : "特定の情報を渡す場所(ddl取る場合はスキーマとテーブル名など)"
    }
}

api-gatewayの部分

以下の記事にlambdaをキックするところまでを書いたので割愛します。
API Gatewayを叩いてLambdaからRedshiftにSQLを投げる(ついでにslackにsnippet通知)

lambda(kicker)の部分

  • post内容から呼び出したい関数名を取り出す
  • 関数名のlambdaがあればそれをキック
  • 関数名に対応するAWS batchがあればそれをキック
kicker.py
import simplejson as json
import boto3
import logging


logger = logging.getLogger()
logger.setLevel(logging.INFO)

def handler(event, context):
    event_dict = json.loads(event["body"])
    function_name = list(event_dict.keys())[0]
    if function_name == "kicker":
        status_code = 404
        body_text = "Kicker loop error : " + function_name
        logger.error(body_text)
    elif kick_lambda(function_name, event):
        status_code = 201
        body_text = "kick lambda : " + function_name
        logger.info(body_text)
    elif kick_batch(function_name, event):
        status_code = 201
        body_text = "kick batch : " + function_name
        logger.info(body_text)
    else:
        status_code = 404
        body_text = "No Method Error : " + function_name
        logger.error(body_text)

    responseObj = {
        "statusCode": status_code,
        "body": body_text
    }
    return responseObj


def kick_lambda(function_name, event):
    try:
        clientLambda = boto3.client("lambda")
        res = clientLambda.invoke(
            FunctionName=function_name,
            InvocationType="Event",
            Payload=json.dumps(event)
        )
        return True
    except:
        return False


def kick_batch(function_name, event):
    try:
        clientBatch = boto3.client("batch")
        res = clientBatch.submit_job(
            jobName=function_name,
            jobQueue=function_name + "_queue",
            jobDefinition=function_name + "_job_definition",
            parameters={
                'event' : json.dumps(event)
            }
        )
        return True
    except:
        return False
    return False


if __name__ == '__main__':
    handler('', '')

非同期で呼び出されたlambda(右側)の部分

eventをそのまま渡しているので以下の記事のように色々作ればOK
API Gatewayを叩いてLambdaからRedshiftにSQLを投げる(ついでにslackにsnippet通知)

非同期で呼び出されたAWS batch(右側)の部分

kickerで定義された名前に対応するものを作成しておけばOK
例のものであれば
– aws_batch_job_queueのname属性が{function_name}_queue
– aws_batch_job_definitionのname属性が{function_name}_job_definition

kicker.pyの一部
def kick_batch(function_name, event):
    try:
        clientBatch = boto3.client("batch")
        res = clientBatch.submit_job(
            jobName=function_name,
            jobQueue=function_name + "_queue",
            jobDefinition=function_name + "_job_definition",
            parameters={
                'event' : json.dumps(event)
            }
        )
        return True
    except:
        return False
    return False

終わりに

全体的にスッキリして機能追加などがかなり簡単になりました。
あとはterraform・lambdaのzipファイル・AWS batchのコンテナイメージをどうやって運用するかが考え所な感じがしました

続きを読む

AWS Step FunctionsでRedshift準備完了まで待機する

目的

AWSでサーバーインスタンス起動系などのリクエストをすると数分待つことになります。
対象や実現方法は色々考えられますが、今回はRedshiftを起動し、使える状態になるまで待機する部分をAWS Step Functionsで実装してみます。

概要

  • チュートリアルのターゲットをRedshiftに変えて実際にやってみたという内容です。
  • LambdaはPython3.6で記述します。

手順

  • 実行用のロールや環境設定は省略します

Redshift起動Lambdaを作成

createRedshiftCluster という名前でLambda Functionを作成します。
パラメータはハードコーディングになっています。(記事用にシンプルにするためです 汗)

createRedshiftCluster
import boto3

def lambda_handler(event, context):
    redshift = boto3.client('redshift')
    redshift.create_cluster(
        ClusterIdentifier='test-cluster',
        DBName='dev',
        Port=5439,
        MasterUsername='testuser',
        MasterUserPassword='myPassword1234',
        VpcSecurityGroupIds=['sg-abcdef12'], 
        ClusterSubnetGroupName='csg1',
        NodeType='dc2.large',
        ClusterType='single-node',
        PubliclyAccessible=True,
        EnhancedVpcRouting=False)
    event['result'] = 'SUCCESS'
    return event

Redshift起動完了確認Lambdaを作成

waitRedshift という名前でLambda Functionを作成します。
WaiterAPIを利用しますがここではリトライはしません。(MaxAttempts=1は1回実行して実行中なら再試行しません)
StepFunctions側でリトライを制御します。
エラー処理は微妙かもしれません。現物合わせです。
起動中ならCREATING、準備完了なら RUNNING を返すようにしてあります。

waitRedshift
import boto3
import botocore

def lambda_handler(event, context):
    redshift = boto3.client('redshift')
    waiter = redshift.get_waiter('cluster_available')
    try:
        waiter.wait(
            ClusterIdentifier='test-cluster',
            WaiterConfig={
                'Delay': 3,
                'MaxAttempts': 1})
        event['result'] = 'RUNNING'
    except botocore.exceptions.WaiterError as e:
        if not hasattr(e, 'last_response'):
            raise e
        if 'Error' in e.last_response:
            raise e
        event['result'] = 'CREATING'
    return event

全体の流れをStepFunctionsで作成

PrepareRedshift という名前でStepFunctionsを作成します。
上記で作成した lambda:createRedshiftCluster を実行し、20秒待ち。
lambda:waitRedshiftで完了チェックし、まだなら再度20秒待ち。これを繰り返します。
APIコール部分はリトライも記述してあります。

PrepareRedshift
{
  "Comment": "Create and Wait for Redshift Cluster",
  "StartAt": "CreateCluster",
  "States": {
    "CreateCluster": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-east-1:123456789012:function:createRedshiftCluster",
      "Next": "Wait",
      "Retry": [
        {
          "ErrorEquals": ["States.ALL"],
          "IntervalSeconds": 5,
          "MaxAttempts": 3,
          "BackoffRate": 2.0
        }
      ]
    },
    "Wait": {
      "Type": "Wait",
      "Seconds" : 20,
      "Next": "CheckStatus"
    },
    "CheckStatus": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-east-1:123456789012:function:waitRedshift",
      "Next": "IsRunning",
      "Retry": [
        {
          "ErrorEquals": ["States.ALL"],
          "IntervalSeconds": 5,
          "MaxAttempts": 3,
          "BackoffRate": 2.0
        }
      ]
    },
    "IsRunning": {
      "Type": "Choice",
      "Choices": [
        {
          "Variable": "$.result",
          "StringEquals": "RUNNING",
          "Next": "Success"
        },
        {
          "Variable": "$.result",
          "StringEquals": "CREATING",
          "Next": "Wait"
        }
      ],
      "Default": "Abort"
    },
    "Abort": {
      "Type": "Fail"
    },
    "Success": {
      "Type": "Succeed"
    }
  }
}

記述について

素晴らしい機能ですが、Amazon States Language なる言語(?)で記述する必要があります。
あとフロー中の変数取り回しができるのですが、 $.var のような記述でこれはJsonPathというようです。
ちょっと大変ですが、ひと通り目を通しておいたほうが良さそうです。
ビジュアライズしてくれるのでそこはホッとします。

表示例:(Endがつながっていませんが大丈夫です :sweat_drops:

image.png

実行結果

マネージメントコンソールから実行すると実行中、実行終了のステップに着色されていきます。
今回は4分ほどで成功しました。

image.png

まとめ

Redshiftクラスタが操作可能になるまで待機する、というパーツはできました。(※エラー処理とかテストとか全然できてませんが :smile:
現実的には、処理の起動トリガーをどうするか、後続処理を実装する等、システムとして組み立てることになると思います。
プロジェクトとしては、いわゆるビジネスロジック(古い表現かも)部分に時間を掛けたいはずですよね。
今回のような準備処理は時間をかけず、サラリと実現するためのツールが Cloud Automatorですねー

続きを読む

Alexa for Businessのスキルを作成してみた

Alexa for Businessのスキルを作成してみました。
リージョンを指定するとそのリージョンで発生中のCloudwatchAlarmを答えるスキルです。

必要なもの

  • AWSアカウント
  • Amazon Developerアカウント
  • Alexa Skills Kit Command Line Interface (ASK CLI)

作業概要

  1. スキル用のLambdaの作成
  2. スキルの作成とテスト
  3. プライベートスキルの設定
  4. 公開情報とプライバシーとコンプライアンスの設定
  5. 申請とスキルの有効化

1. スキル用のLambdaの作成

以下AWSマネジメントコンソールにログインして作業します。

1. Lambda用のIAMロールを作成する

以下ポリシーを持ったIAMロールを作成します。
Lambda用のベースのポリシーにCloudwatchアラームへのアクセス許可を追加しています。

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "logs:CreateLogGroup",
                "logs:CreateLogStream",
                "logs:PutLogEvents"
            ],
            "Resource": "arn:aws:logs:*:*:*"
        },
        {
            "Effect": "Allow",
            "Action": "cloudwatch:DescribeAlarms",
            "Resource": "*"
        }
    ]
}

 
2. blueprintsで[alexa-skills-kit-color-expert-python]を選択・編集しLambdaを作成する

blueprintはpython2.7用ですが、python3.6用にしています。

# -*- coding: utf-8 -*-
import boto3


# --------------- Helpers that build all of the responses ----------------------

def build_speechlet_response(title, output, reprompt_text, should_end_session):
    return {
        'outputSpeech': {
            'type': 'PlainText',
            'text': output
        },
        'card': {
            'type': 'Simple',
            'title': "SessionSpeechlet - " + title,
            'content': "SessionSpeechlet - " + output
        },
        'reprompt': {
            'outputSpeech': {
                'type': 'PlainText',
                'text': reprompt_text
            }
        },
        'shouldEndSession': should_end_session
    }


def build_response(session_attributes, speechlet_response):
    return {
        'version': '1.0',
        'sessionAttributes': session_attributes,
        'response': speechlet_response
    }


# --------------- Functions that control the skill's behavior ------------------

def get_welcome_response():
    session_attributes = {}
    card_title = "Welcome"
    speech_output = "こんにちは。AWSのアラート状況をお知らせするスキルです。 " \
                    "例えば次のように質問して下さい。 " \
                    "東京リージョンの状況は"
    reprompt_text = "例えば次のように質問して下さい。 " \
                    "東京リージョンの状況は"
    should_end_session = False
    return build_response(session_attributes, build_speechlet_response(
        card_title, speech_output, reprompt_text, should_end_session))


def handle_session_end_request():
    card_title = "Session Ended"
    speech_output = "Thank you for trying the Alexa Skills Kit sample. " \
                    "Have a nice day! "
    # Setting this to true ends the session and exits the skill.
    should_end_session = True
    return build_response({}, build_speechlet_response(
        card_title, speech_output, None, should_end_session))


def get_alarms(region):
    cloudwatch = boto3.client('cloudwatch', region_name=region)
    try:
        response = cloudwatch.describe_alarms(StateValue='ALARM')
        alarms = [alarm['AlarmName'] for alarm in response['MetricAlarms']]
        print(alarms)
        return alarms
    except Exception as e:
        print(e)
        raise e


def response_alarms(intent, session):
    card_title = intent['name']
    session_attributes = {}
    should_end_session = False
    region_dic = {
        '東京':'ap-northeast-1',
        'バージニア':'us-east-1',
        'シドニー':'ap-southeast-2'
    }

    if 'Region' in intent['slots'] \
    and intent['slots']['Region']['value'] in region_dic:
        region_jp = intent['slots']['Region']['value']
        region = region_dic[region_jp]
        alarms = get_alarms(region)
        if len(alarms) == 0:
            speech_output = "現在" + \
                            region_jp + \
                            "リージョンにアラートはありません。 "
        else:
            speech_output = "現在" + \
                            region_jp + \
                            "リージョンに" + \
                            str(len(alarms)) + \
                            "件アラートが発生しています。" + \
                            "アラート名は" + \
                            "と".join(alarms) + \
                            "です。"
        reprompt_text = None
        should_end_session = True
    else:
        speech_output = "リージョン名がわかりませんでした。" \
                        "もう一度お願いします。"
        reprompt_text = "リージョン名がわかりませんでした。 " \
                        "もう一度お願いします。"
        should_end_session = False
    return build_response(session_attributes, build_speechlet_response(
        card_title, speech_output, reprompt_text, should_end_session))


# --------------- Events ------------------

def on_session_started(session_started_request, session):
    print("on_session_started requestId=" + session_started_request['requestId']
          + ", sessionId=" + session['sessionId'])


def on_launch(launch_request, session):
    print("on_launch requestId=" + launch_request['requestId'] +
          ", sessionId=" + session['sessionId'])
    return get_welcome_response()


def on_intent(intent_request, session):
    print("on_intent requestId=" + intent_request['requestId'] +
          ", sessionId=" + session['sessionId'])

    intent = intent_request['intent']
    intent_name = intent_request['intent']['name']

    if intent_name == "MyAwsAlarmsIntent":
        return response_alarms(intent, session)
    elif intent_name == "AMAZON.HelpIntent":
        return get_welcome_response()
    elif intent_name == "AMAZON.CancelIntent" or intent_name == "AMAZON.StopIntent":
        return handle_session_end_request()
    else:
        raise ValueError("Invalid intent")


def on_session_ended(session_ended_request, session):
    print("on_session_ended requestId=" + session_ended_request['requestId'] +
          ", sessionId=" + session['sessionId'])



# --------------- Main handler ------------------

def lambda_handler(event, context):
    """ Route the incoming request based on type (LaunchRequest, IntentRequest,
    etc.) The JSON body of the request is provided in the event parameter.
    """
    print("event.session.application.applicationId=" +
          event['session']['application']['applicationId'])

    """
    Uncomment this if statement and populate with your skill's application ID to
    prevent someone else from configuring a skill that sends requests to this
    function.
    """
    # if (event['session']['application']['applicationId'] !=
    #         "amzn1.echo-sdk-ams.app.[unique-value-here]"):
    #     raise ValueError("Invalid Application ID")

    if event['session']['new']:
        on_session_started({'requestId': event['request']['requestId']},
                           event['session'])

    if event['request']['type'] == "LaunchRequest":
        return on_launch(event['request'], event['session'])
    elif event['request']['type'] == "IntentRequest":
        return on_intent(event['request'], event['session'])
    elif event['request']['type'] == "SessionEndedRequest":
        return on_session_ended(event['request'], event['session'])

2. スキルの作成とテスト

以下Amazon開発者コンソールにログインして作業します。

1. スキル情報を設定する

  • スキルの種類:Custom
  • 言語:Japanese
  • スキル名:AWSアラート状況
  • 呼び出し名:awsアラート状況

他はデフォルト
 
2. 対話モデルを設定する

  • インテントスキーマ
{
  "intents": [
    {
      "slots": [
        {
          "name": "Region",
          "type": "LIST_OF_REGIONS"
        }
      ],
      "intent": "MyAwsAlarmsIntent"
    },
    {
      "intent": "AMAZON.HelpIntent"
    },
    {
      "intent": "AMAZON.StopIntent"
    },
    {
      "intent": "AMAZON.CancelIntent"
    }
  ]
}
  • カスタムスロットタイプ

    • タイプ:LIST_OF_REGIONS
    • 値:
      • 東京
      • バージニア
      • シドニー
  • サンプル発話

    • MyAwsAlarmsIntent {Region} は
    • MyAwsAlarmsIntent {Region} の状況は
    • MyAwsAlarmsIntent {Region} のアラート状況は

 
3. 「設定」を設定する

  • サービスエンドポイントのタイプ:AWS Lambda の ARN

    • 作成したLambdaのARNを指定する

他はデフォルト

 
4. テストする

4-1. あらかじめ、バージニアリージョンに3件Cloudwatchアラームが発生している状態にしておきます。

4-2. 発話を入力してください欄に「バージニアは」と入力してテストすると

4-3. サービスレスポンスに下記が返ってきます。

{
  "version": "1.0",
  "response": {
    "outputSpeech": {
      "text": "現在バージニアリージョンに3件アラートが発生しています。アラート名はalarm_1とalarm_2とalarm_3です。",
      "type": "PlainText"
    },
    "card": {
      "content": "SessionSpeechlet - 現在バージニアリージョンに3件アラートが発生しています。アラート名はalarm_1とalarm_2とalarm_3です。",
      "title": "SessionSpeechlet - MyAwsAlarmsIntent"
    },
    "reprompt": {
      "outputSpeech": {
        "type": "PlainText"
      }
    },
    "speechletResponse": {
      "outputSpeech": {
        "text": "現在バージニアリージョンに3件アラートが発生しています。アラート名はalarm_1とalarm_2とalarm_3です。"
      },
      "card": {
        "content": "SessionSpeechlet - 現在バージニアリージョンに3件アラートが発生しています。アラート名はalarm_1とalarm_2とalarm_3です。",
        "title": "SessionSpeechlet - MyAwsAlarmsIntent"
      },
      "reprompt": {
        "outputSpeech": {}
      },
      "shouldEndSession": true
    }
  },
  "sessionAttributes": {}
}

3. プライベートスキルの設定

1. alexa Skills Kit Command Line Interface (ASK CLI)をインストールする

下記URLを参考にインストールします。
https://developer.amazon.com/docs/smapi/quick-start-alexa-skills-kit-command-line-interface.html
 
2. プライベートスキルの設定をする

下記URLを参考にプライベートスキルの設定をします。
http://docs.aws.amazon.com/a4b/latest/ag/private-skills.html

スキルのIDは、スキル情報に表示されます。

$ skill_id=<スキルのID>
$ ask api get-skill -s ${skill_id} > skill.json
$ vi skill.json

“distributionMode”: “PRIVATE”を1行追加する

skill.json
{
  "skillManifest": {
    "publishingInformation": {
      "locales": {
        "ja-JP": {
          "name": "AWSアラート状況"
        }
      },
      "isAvailableWorldwide": true,
      "distributionMode": "PRIVATE"
    },
    "apis": {
      "custom": {
        "endpoint": {
          "uri": "<LambdaのARN>"
        }
      }
    },
    "manifestVersion": "1.0"
  }
}
$ ask api update-skill -s ${skill_id} -f skill.json

4. 公開情報とプライバシーとコンプライアンスの設定

以下Amazon開発者コンソールで作業します。

ちょっと面倒ですが、小アイコン(108×108)と大アイコン(512×512)を用意してアップロードし
他記入できるところを全て記入します。

5. 申請とスキルの有効化

1. 申請する

$ ask api submit -s ${skill_id}

 
2. ステータスがLiveになるまで、数時間待つ
 
3. Alexa for Businessオーガナイゼーションに配布する

$ account_id=arn:aws:iam::<AWSアカウントID>:root
$ ask api add-private-distribution-account -s ${skill_id} --stage live --account-id ${account_id}

 
4. スキルを有効にする

AWSマネジメントコンソール:Alexa for Businessのprivate skillsの一覧にスキルが表示されます。

skill.png

[Review]をクリックして[Enable]を選択しスキルを有効にします。
 
5. ここまで
東京リージョンではまだ利用できないので、残念ながらここまでです。
2017/12/5追記 英語でよければ、利用できるそうです。

雑感

Alexa for Businessの何でもできちゃいそう感がすごい。
Re:Inventのキーノートを見たら「ちょっといい会議システム」みたいなものと勘違いしちゃいそうですが、もっと全然奥が深そう。
自分の中では2017年のRe:Inventの2番目の目玉です。
早く東京リージョンにこないかな。

続きを読む

boto3でS3にファイルアップロードする方法

Pythonで画像ファイルをS3にアップロードしてくなったので、以下方法でやってみて簡単にできました。

環境

python 3.5.1
Mac OS 10.11.4

credentialの設定を忘れずに

AWS cliコマンドをまずは使えるようにしておきます。

$ pip install awscli

credentialを設定しておく

$ aws configure

aws configureコマンドを打つと、accesskey,secret key,regionの設定ができるので、
AWSコンソール画面にログインし、事前に確認しておき、ここで設定しておきます。

すると .aws/credentialsにprofileが設定されるので、下準備はOK

boto3をインストールしておく

Amazonへの操作ができるようboto3をインストールします。

$ pip install boto3

以下のようにファイルコードを記述します。
upload_file APIを使います。

import json
import boto3

bucket_name = "my-bucket-name"
s3 = boto3.resource('s3')

s3.Bucket(bucket_name).upload_file('/Users/tottu22/Downloads/local.jpg', 'server.jpg')

※ダウンロードフォルダにあるlocal.jpgファイルをS3のmy-bucket-nameフォルダにserver.jpgファイルとしてアップロードする例

これでファイルをアップロードは完了

指定したバケットにファイルがアップロードされていることを確認してみてください。

続きを読む

まだ彼女作りで消耗してるの?/v2

カノジョできないエンジニア Advent Calendar 2017 【初夜】

https://adventar.org/calendars/2467

今年もやってきました!!
彼女できないエンジニアの諸君!!

エンジニアの仕事とは何か!!
今一度問いたい。
君たちの仕事はなんなのかと!!

  • プログラムを書くこと?
  • 仕様の通りにシステムを作る事?

ばかやろう!お客さんや社会、世の中の課題を解決することじゃねぇか!!

「彼女できない」という課題を解決せずして何を解決できるというんだ!!!

行くぞまずは分析からだ!!

 男と女の好み分析

以下は僕のbigdataをグラフ化したもの

男はどんな女性が好きなのか

スクリーンショット 2017-12-01 21.07.48.png

そう!!男なんて顔が可愛けりゃ正直誰でもいいと思ってるだろ!!
(俺はそうは思わない)

グラフにも現れてるけど、ブスを好んで好きになる人なんていない!!
(俺はそうは思わないけどね)

性格とか、フィーリングとかから徐々に好きになることはあっても、ファーストイメージが占める割合が高い!
(そんな単純じゃないと思う俺は)

一方女子はどうだろうか

スクリーンショット 2017-12-01 21.12.42.png

お分り頂けただろうか。
男はかわいい女の子が好きというシンプルなものだったが(僕は例外)、
女子の好みは細分化されている!!!!

決してイケメンがモテるワケではない!!!!

つまり、自分がどのタイプに属するか理解すること。
これがお前の武器になるんだ!!!

ではどうするか

自分の顔がこれらのどの部類に入るか分析しよう。

まず膨大な教師データを集めて。。

スクリーンショット 2017-12-01 21.20.55.png

特徴点をちゅうsh。。。。

クリスマスまで時間がないのでRekognitionを使おう

S3バケットに自分の画像つっこんで、Lambdaでこんな感じに買いて

print('Loading function')

import boto3

def lambda_handler(event, context):
    reko_client = boto3.client('rekognition','us-east-1')
    r = reko_client.detect_faces(
        Image={
            'S3Object': {
                'Bucket': 'onda-test-1234',
                'Name': '俺の顔.jpg'
            }
        },
        Attributes=['ALL']
        )
    print(r)

実行すればでデータが取れる!

{
    "FaceDetails": [{
        "BoundingBox": {
            "Width": 0.10499999672174454,
            "Height": 0.18666666746139526,
            "Left": 0.4312500059604645,
            "Top": 0.057777777314186096
        },
        "AgeRange": {
            "Low": 23,
            "High": 38
        },
        "Smile": {
            "Value": True,
            "Confidence": 57.108856201171875
        },
        "Eyeglasses": {
            "Value": False,
            "Confidence": 99.96912384033203
        },
        "Sunglasses": {
            "Value": False,
            "Confidence": 99.87023162841797
        },
        "Gender": {
            "Value": "Female",
            "Confidence": 100.0
        },
        "Beard": {
            "Value": False,
            "Confidence": 99.96233367919922
        },
        "Mustache": {
            "Value": False,
            "Confidence": 99.77082824707031
        },
        "EyesOpen": {
            "Value": True,
            "Confidence": 99.99246215820312
        },
        "MouthOpen": {
            "Value": False,
            "Confidence": 99.9146728515625
        },
        "Emotions": [{
            "Type": "CALM",
            "Confidence": 82.83368682861328
        }, {
            "Type": "HAPPY",
            "Confidence": 15.994048118591309
        }, {
            "Type": "SAD",
            "Confidence": 3.979891538619995
        }],
        "Landmarks": [{
            "Type": "eyeLeft",
            "X": 0.4645110070705414,
            "Y": 0.1398819535970688
        }, {
            "Type": "eyeRight",
            "X": 0.5005732178688049,
            "Y": 0.13559408485889435
        }, {
            "Type": "nose",
            "X": 0.48266592621803284,
            "Y": 0.17239990830421448
        }, {
            "Type": "mouthLeft",
            "X": 0.47122448682785034,
            "Y": 0.20362448692321777
        }, {
            "Type": "mouthRight",
            "X": 0.4979352056980133,
            "Y": 0.1986924409866333
        }, {
            "Type": "leftPupil",
            "X": 0.46497994661331177,
            "Y": 0.13870377838611603
        }, {
            "Type": "rightPupil",
            "X": 0.5003837943077087,
            "Y": 0.13453131914138794
        }, {
            "Type": "leftEyeBrowLeft",
            "X": 0.45106732845306396,
            "Y": 0.12510329484939575
        }, {
            "Type": "leftEyeBrowUp",
            "X": 0.46162354946136475,
            "Y": 0.1195397600531578
        }, {
            "Type": "leftEyeBrowRight",
            "X": 0.4728100895881653,
            "Y": 0.12344556301832199
        }, {
            "Type": "rightEyeBrowLeft",
            "X": 0.4925210475921631,
            "Y": 0.1216742992401123
        }, {
            "Type": "rightEyeBrowUp",
            "X": 0.5035019516944885,
            "Y": 0.11563356220722198
        }, {
            "Type": "rightEyeBrowRight",
            "X": 0.5140520334243774,
            "Y": 0.11972268670797348
        }, {
            "Type": "leftEyeLeft",
            "X": 0.4573439657688141,
            "Y": 0.1410721391439438
        }, {
            "Type": "leftEyeRight",
            "X": 0.4719408452510834,
            "Y": 0.14020104706287384
        }, {
            "Type": "leftEyeUp",
            "X": 0.46427008509635925,
            "Y": 0.1344168335199356
        }, {
            "Type": "leftEyeDown",
            "X": 0.46462056040763855,
            "Y": 0.14459241926670074
        }, {
            "Type": "rightEyeLeft",
            "X": 0.4931148886680603,
            "Y": 0.13775958120822906
        }, {
            "Type": "rightEyeRight",
            "X": 0.5079139471054077,
            "Y": 0.1350509226322174
        }, {
            "Type": "rightEyeUp",
            "X": 0.5002275705337524,
            "Y": 0.1299990564584732
        }, {
            "Type": "rightEyeDown",
            "X": 0.5009776949882507,
            "Y": 0.140377938747406
        }, {
            "Type": "noseLeft",
            "X": 0.47677117586135864,
            "Y": 0.1822502166032791
        }, {
            "Type": "noseRight",
            "X": 0.48997604846954346,
            "Y": 0.17978978157043457
        }, {
            "Type": "mouthUp",
            "X": 0.4844018518924713,
            "Y": 0.19528257846832275
        }, {
            "Type": "mouthDown",
            "X": 0.48532187938690186,
            "Y": 0.21053189039230347
        }],
        "Pose": {
            "Roll": -4.027588367462158,
            "Yaw": -3.790987253189087,
            "Pitch": -4.375948905944824
        },
        "Quality": {
            "Brightness": 37.62503433227539,
            "Sharpness": 99.9980239868164
        },
        "Confidence": 99.99703979492188
    }],
    "OrientationCorrection": "ROTATE_0",
    "ResponseMetadata": {
        "RequestId": "7272b9b3-d692-11e7-a999-51ef28f1d5b1",
        "HTTPStatusCode": 200,
        "HTTPHeaders": {
            "content-type": "application/x-amz-json-1.1",
            "date": "Fri, 01 Dec 2017 12:23:34 GMT",
            "x-amzn-requestid": "7272b9b3-d692-11e7-a999-51ef28f1d5b1",
            "content-length": "2766",
            "connection": "keep-alive"
        },
        "RetryAttempts": 0
    }
}

注)なぜかFemaleの値が100になってるけど、これは僕がエマワトソンが好きすぎて、俺の顔.jpgといつわってエマワトソンの画像を使ったからである。

こんな感じで簡単にデータが取れる!!

幸せとか悲しいとか穏やかとかそんな感じだけどなんとかなる!!
気になる女の子のタイプの顔を分析し、自分と比較しよう!!
もし、数値が近ければワンチャンあるぞ!!

女の子の好きなタイプとか調査する術もない場合はもうね。
直接聞こう!!

deftextlarge#1{%
  {rmLarge #1}
}
deftextsmall#1{%
  {rmscriptsize #1}
}
​​​​​​​​
textlarge{どんなタイプの人が好きですか?}
textlarge{JSONで教えて??}

続きを読む

AWS LambdaからAmazon Elasticsearch Serviceにつないでみる

この投稿は、AWS Lambda Advent Calendar 2017の初日の投稿になります。

初日なので簡単なのをば!

Elasticsearchは今までローカルやプライベートネットワーク上にインストールして使うことが多く、マネージドサービスを使ったことがなかったので、今回Amazon Elsaticsearch Serviceを使ってみました。

で、色々なサイトを参考させてもらい、設定〜インデックス作成まで実施したので、その備忘録を載せます。

準備

準備ですが、クラスタを作成するだけです。当然t2.smallで作成。Elasticsearchのバージョンは5.5を選択しました。他はデフォルトの設定です。

スクリーンショット 2017-11-19 18.52.14.png

Lambdaからアクセスしてインデックスを作成してみる

AWS Lambdaに適当な関数を作成して、とりあえずアクセスしてみます。こちらを参考にさせてもらいました。

Lambda から elasticsearch service に何かする [cloudpack OSAKA blog]

このコードでアクセスするためには、STSのAssumeRolests:AssumeRoleがIAM Roleに付与されている必要があります。ポイントとなるポリシーの設定はこんな感じです。(もちろんテスト的な設定なので、本来はきちんとアクセス権限を設計しましょう)

        {
            "Action": [
                "sts:AssumeRole"
            ],
            "Resource": [
                "*"
            ],
            "Effect": "Allow"
        }

ソースコードはこんな感じで動きました。ENDPOINT、REGION、ROLE_ARN、S3_BUCKET、S3_OBJECTは環境変数からの定義となります。

import os
import sys

import boto3

sys.path.append(os.path.join(
    os.path.abspath(os.path.dirname(__file__)), 'lib'))
from elasticsearch import Elasticsearch, RequestsHttpConnection
from requests_aws4auth import AWS4Auth

ENDPOINT = os.environ['ES_ENDPOINT']
REGION = os.environ['REGION']
ROLE_ARN = os.environ['ROLE_ARN']
S3_BUCKET = os.environ['S3_BUCKET']
S3_OBJECT = os.environ['S3_OBJECT']


def run(event, context):
    es_client = connect_es(ENDPOINT)

    if event['method'] == "create-index":
        s3 = boto3.client('s3')
        index_doc = s3.get_object(Bucket=S3_BUCKET, Key=S3_OBJECT)['Body'].read()
        print(index_doc)
        create_index(es_client, event['index'], index_doc)
        return "Success"
    return es_client.info()

def connect_es(es_endpoint):

    print('Connecting to the ES Endpoint {0}'.format(es_endpoint))
    credentials = get_credential()
    awsauth = AWS4Auth(credentials['access_key'], credentials['secret_key'], REGION, 'es', session_token=credentials['token'])

    try:
        es_client = Elasticsearch(
            hosts=[{'host': es_endpoint, 'port': 443}],
            http_auth=awsauth,
            use_ssl=True,
            verify_certs=True,
            connection_class=RequestsHttpConnection)
        return es_client
    except Exception as E:
        print("Unable to connect to {0}".format(es_endpoint))
        print(E)
        exit(3)

def create_index(es_client, index_name, index_doc):
 try:
  res = es_client.indices.exists(index_name)
  print("Index Exists ... {}".format(res))
  if res is False:
   es_client.indices.create(index_name, body=index_doc)
 except Exception as E:
  print("Unable to Create Index {0}".format("metadata-store"))
  print(E)
  exit(4)

def get_credential():
    client = boto3.client('sts')
    assumedRoleObject = client.assume_role(
        RoleArn=ROLE_ARN,
        RoleSessionName="Access_to_ES_from_lambda"
    )
    credentials = assumedRoleObject['Credentials']
    return { 'access_key': credentials['AccessKeyId'],
             'secret_key': credentials['SecretAccessKey'],
             'token': credentials['SessionToken'] }

コードを見れば分かる通り、S3にアクセスするため、権限の設定が必要になります。

S3に配置するJSONデータの準備

今回はS3にインデックスの設定を保持します。こんな感じのdata.jsonを配置します。この辺はAWS Solution Architectのブログを参考にします。

【AWS Database Blog】AWS Lambda と Pythonを使ってメタデータをAmazon Elasticsearch Serviceにインデクシング

{
    "dataRecord": {
        "properties": {
            "createdDate": {
                "type": "date",
                "format": "dateOptionalTime"
            },
            "objectKey": {
                "type": "string",
                "format": "dateOptionalTime"
            },
            "content_type": {
                "type": "string"
            },
            "content_length": {
                "type": "long"
            },
            "metadata": {
                "type": "string"
            }
        }
    },
    "settings": {
        "number_of_shards": 1,
        "number_of_replicas": 0
    }
}

アクセスしてみる

アクセスしたところ、無事インデックスが作成されました。sample-indexが増えているのがわかります。

スクリーンショット 2017-12-01 18.29.10.png

まとめ

今回初めてAmazon Elasticsearch Serviceを利用しましたが、準備段階の手間が省けているのはありがたいですね。またAWS上のサービスなので、Lambdaなどの別サービスからのアクセスも簡単でした。

おまけ

ちなみに最初、受け取ったElasticsearch ClientをそのままReturnしていたのですが、その時のエラーメッセージがこんな感じ。一見するとJSONで保持できない=パラメータが異なった?と受け取れます。

{
  "errorMessage": "<Elasticsearch([{'host': 'HOSTNAME.us-east-1.es.amazonaws.com', 'port': 443}])> is not JSON serializable",
  "errorType": "TypeError",
  "stackTrace": [
    [
      "/var/lang/lib/python3.6/json/__init__.py",
      238,
      "dumps",
      "**kw).encode(obj)"
    ],
    [
      "/var/lang/lib/python3.6/json/encoder.py",
      199,
      "encode",
      "chunks = self.iterencode(o, _one_shot=True)"
    ],
    [
      "/var/lang/lib/python3.6/json/encoder.py",
      257,
      "iterencode",
      "return _iterencode(o, 0)"
    ],
    [
      "/var/runtime/awslambda/bootstrap.py",
      110,
      "decimal_serializer",
      "raise TypeError(repr(o) + " is not JSON serializable")"
    ]
  ]
}

これ、原因が

    return es_client

es_clientをそのまま返していたから。

    return es_client.info()

とすれば動くのですが、こんなの気づけんって。。。せめてピンポイントでエラー発生行番号が出力されれば話は別ですが、不親切すぎなエラーメッセージ(^^;どこかで改善されることを期待っす。

続きを読む