kintoneで設定したスケジュールにあわせてlambda(python)からSQSへメッセージを送る
EC2の起動停止をそろそろ手動でやるのも疲れてきそうなのと、やはり停止するのを忘れてしまってもったいないので。
SQSを挟むほどの規模でもないのですが、慣れておきたかったのでSQSを使ってみています。
今回はSQSへメッセージを送るところまでを記載しています。
SQSからメッセージを受信してEC2の起動停止をlambdaで実行して結果をslackへ通知するのは次回のブログに書きます。
目次
kintoneでアプリ作成
こんな項目を持ったアプリを作成します。
フォームはこんな感じです。
APIトークンを発行します
アプリの設定で詳細設定を展開して[APIトークン]をクリックします。
APIトークンの生成をします。
今回はデータを読むだけなのでアクセス権は閲覧のみにします。
保存した後、[設定完了]をクリックします。
※これをしないと生成したAPIトークンが反映しません。
ターミナルなどで確認します。
1 2 |
$ curl -H "X-Cybozu-API-Token: YOUR_TOKEN" "https://YOUR_DOMAIN.cybozu.com/k/v1/record.json?app=24&id=1" |
YOUR_TOKEN, YOUR_DOMAINは適宜変更します。
データのレスポンスが返ってくればOKです。
Terminalで見づらいのでPOSTMANで見ます。
祝日対応
やりながら祝日も対応しなければと思い、上記のkintoneの曜日に「祝」を足しました。
なんかいいAPIないかなーと探したのですが見つけられなかったのでGoogleカレンダーAPIを使いました。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 |
from apiclient.discovery import build API_KEY = 'XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX' CALENDAR_ID = 'ja.japanese#holiday@group.v.calendar.google.com' def getHolidays(): try: service = build(serviceName='calendar', version='v3', developerKey=API_KEY) events = service.events().list(calendarId=CALENDAR_ID).execute() holidays = [] for item in sorted(events['items'], key=lambda item: item['start']['date']): holidays.append(item['start']['date']) return holidays except Exception as e: return e |
これをlambdaファンクションと同じディレクトリに格納しておきます。
kintoneからデータ取得してSQSへエンキューするlambda
GitHubはこちらです。
kintoneのレコードから起動対象か停止対象かをチェックします。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 |
def checkValidate(type_str,record): field_name = '' field_value = '' if type_str == 'start': field_name = 'c_start_check' field_value = u'開始' elif type_str == 'stop': field_name = 'c_stop_check' field_value = u'停止' check_item = record[field_name]['value'] if len(check_item) > 0: if check_item[0] == field_value: return True return False def checkDate(type_str,record,holidays,date_str,week_day_str): field_name = '' if type_str == 'start': field_name = 'c_start_week_day' elif type_str == 'stop': field_name = 'c_stop_week_day' week_day_array = record[field_name]['value'] if len(week_day_array) > 0: if u'祝' in week_day_array and date_str in holidays: return True elif week_day_str in week_day_array: if u'祝' in week_day_array: return True else: if date_str not in holidays: return True return False def checkItem(type_str,record,time_str): field_name = '' if type_str == 'start': field_name = 't_start_time' elif type_str == 'stop': field_name = 't_stop_time' check_time = record[field_name]['value'] if time_str[0:4] == check_time[0:4]: return True return False |
起動停止対象の場合辞書型のメッセージを配列に格納します。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 |
def getQueueList(): now = datetime.now(pytz.timezone('Asia/Tokyo')) date_str = now.strftime('%Y-%m-%d') time_str = now.strftime("%H:%M") week_day_str = getWeekDayStr(now) headers = {HEADERS_KEY:HEADERS_VALUE} response_record = requests.get(KINTONE_URL,headers=headers) record_data = json.loads(response_record.text) records=record_data['records'] holidays = get_holiday.getHolidays() queue_list = [] for record in records: ec2_name = record['s_ec2_name_tag']['value'] for type_str in ['start','stop']: if checkValidate(type_str,record): if checkDate(type_str,record,holidays,date_str,week_day_str): if checkItem(type_str,record,time_str): queue_list.append({'type':type_str,'ec2_name':ec2_name,'datetime':date_str + ' ' + time_str}) return queue_list |
SQSにメッセージをエンキューします。
1 2 3 4 5 6 7 8 9 10 11 |
def set_enqueue(queue_list): try: sqs_client = boto3.client('sqs') for queue in queue_list: response = sqs_client.send_message( QueueUrl = SQS_URL, MessageBody = json.dumps(queue) ) except Exception as e: logger.info(e) |
実行してます。
1 2 3 4 5 6 7 8 9 10 |
def handler(event, context): try: queue_list = getQueueList() if len(queue_list) > 0: set_enqueue(queue_list) except Exception as e: logger.info(e) raise e |
キューにメッセージが送信出来た事が確認出来ました。
次回はSQSからメッセージを受信してEC2を停止起動してslackへ通知します。
最後までお読みいただきましてありがとうございました!
「AWS認定資格試験テキスト&問題集 AWS認定ソリューションアーキテクト - プロフェッショナル 改訂第2版」という本を書きました。
「AWS認定資格試験テキスト AWS認定クラウドプラクティショナー 改訂第3版」という本を書きました。
「ポケットスタディ AWS認定 デベロッパーアソシエイト [DVA-C02対応] 」という本を書きました。
「要点整理から攻略するAWS認定ソリューションアーキテクト-アソシエイト」という本を書きました。
「AWSではじめるLinux入門ガイド」という本を書きました。
開発ベンダー5年、ユーザ企業システム部門通算9年、ITインストラクター5年目でプロトタイプビルダーもやりだしたSoftware Engineerです。
質問はコメントかSNSなどからお気軽にどうぞ。
出来る限りなるべく答えます。
このブログの内容/発言の一切は個人の見解であり、所属する組織とは関係ありません。
このブログは経験したことなどの共有を目的としており、手順や結果などを保証するものではありません。
ご参考にされる際は、読者様自身のご判断にてご対応をお願いいたします。
また、勉強会やイベントのレポートは自分が気になったことをメモしたり、聞いて思ったことを書いていますので、登壇者の意見や発表内容ではありません。
ad
ad
関連記事
-
AWS Lambda(Python)でDynamoDB テーブルを日次で削除/作成(オートスケーリング付き)
この記事はAWS #2 Advent Calendar 2018に参加した記事で …
-
WordPress W3 Total Cache のDatabaseCacheをAmazon ElastiCacheのmemcachedに格納する
このブログのアーキテクチャは現在こちらです。 データベースは、Amazon Au …
-
VPCピア接続した先のVPCインターフェイスエンドポイントを使用する
VPC1とVPC2でピア接続しています。 VPC2にはKMSのインターフェイスエ …
-
S3バケットのリクエスタ支払い
S3バケットのリクエスタ支払いを試しました。 バケットの設定 検証用のS3バケッ …
-
Lambdaバージョンとエイリアスとトリガー
Lambdaのバージョン、エイリアスにはそれぞれ別のトリガーが設定できます。 上 …
-
AWS Storage Gatewayボリュームゲートウェイを作成してWindowsから使用
ボリュームゲートウェイの作成 Storage Gateway作成メニューからボリ …
-
AWSアカウントルートユーザーのMFAでYubicoセキュリティキーを設定した
先日Yubico セキュリティキーを購入して、USBにささなければならないのがな …
-
AWS DeepLens開封の儀
去年(2019年)7月にamazon.co.jpでDeepLens買えますやんっ …
-
CloudWatch LogsメトリクスフィルタでNginxのaccess_logから転送バイト数をモニタリングする
ユーザーガイドのApache ログからのフィールドの抽出を見てて、これ、Ngin …
-
ENAが有効なEC2インスタンスの帯域幅をiperf3で確認してみた
同じ Amazon VPC 内で Amazon EC2 Linux インスタンス …