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 WAFをSIEM on Amazon OpenSearch Serviceで可視化
SIEM on Amazon OpenSearch Serviceを構築の環境に …
-
IAMユーザーのマネジメントコンソールログインパスワードを変更してDynamoDBに記録するLambda(Python)
やりたいこと IAMユーザーのマネジメントコンソールへのサインインパスワードを動 …
-
Amazon CloudWatch Anomaly Detectionをダッシュボードに設定
Amazon CloudWatch クロスリージョンクロスアカウントダッシュボー …
-
特定のIAMロールをLambda(Python)で削除する
やりたいこと 特定アカウントの特定の名前が含まれるIAMロールをまとめて削除した …
-
ALBのヘルスチェックでPHPとMySQL接続をチェック
当ブログで504エラーが発生して、オートスケーリングにより自動でインスタンスが置 …
-
Amazon Location Service入門ワークショップ-ジオコーディング
ジオコーディングにより住所を緯度経度に変換したり、逆に緯度経度から住所を求めたり …
-
GoogleForm,GASからAPI Gateway, Lambdaで入力情報をDynamoDBに格納する
vol.26 AWS認定試験テキスト認定クラウドプラクティショナーのデモ(Dyn …
-
AWS Certificate Manager証明書とAmazon Route 53でAmazon API GatewayのAPIのカスタムドメイン名前解決する
AWS Certificate ManagerとAmazon Route 53と …
-
AWS EC2 Amazon Linuxにmauticをインストールする(Sendgrid, Nginx, MySQL on RDS)
mauticをAmazon LinuxとNginx,MySQL on RDSにイ …
-
AWSアカウント内のLambda関数を削除するLambda(Python)
やりたいこと 特定のAWSアカウントの特定リージョンのAWS Lambda関数を …