MOTO+4 Applocations のブログ

AWS IoTにてサーバリクエストを行う「MQTT」。比較的簡易ですが、エラー処理面でやや留意点がありました。必ずしもではないかとおもいますが、記録しておきます。機器はラズパイ、言語はPythonです。

1 . connect()でエラーになりやすい場合

インターネットおよびLANも生物ですから不安定な場合もあります。そのせいか当方では、MQTTクライアントのconnect()メソッドでエラー一時的に多発しました。

よくよく調べると、connect()メソッドのタイムアウトのデフォルト値は5秒です。

これではすぐに引っかかりそうです。AWSサーバで生成した「接続キット」のサンプルコードのままではNGです。30秒は欲しいところです。以下のように修正しました。

    mqttc = AWSIoTMQTTClient(clientId)
            ( 中 略 )
           
    mqttc.configureConnectDisconnectTimeout(30)  
    mqttc.configureMQTTOperationTimeout(30)
    mqttc.connect( )

ついでに、publish()メソッドのタイムアウト値も延ばしておきます。

2 . publish()がエラーにならずスルーする

publish()メソッドのエラー検出のテストのため、テストコードで先にmqttc.closeメソッドし、publish()したところ、エラーにならずスルーしました。

これは疑似的なので仕方ないと、実機テストに踏み切ったところ、たまたまネットワークが切れ、サーバ受信にできなかったのですが、publish()側では正常スルーなりました。ネットワーク本体が切れると、スルーしてしまうのかな?

しかなく、Pythonスクリプトとは別に、シェルスクリプトでネットワークデバイスを、数秒置きに監視しするようにしました。当方では、4Gによるppp接続だったため以下のようにした。

while [ 1 ]
do
    ifconfig ppp0 > /dev/null
    RET_PPP=$? 

    if [ $RET_PPP -eq 0 ];then
	PPP_ON=1
    else
    	# flash disk cash
	sync
	sync
	sync
	shutdown -r now
    fi

    sleep 30
done

少し乱暴ですが、pppが切れる要因によっては復帰できない場合もあるため、一律リブートとしました。

AWSで、ユーザがダウンロードするファイルは、暗号化したいところです。まず「CloudShell」で、「S3」のファイルを暗号化Zipする練習してみました。

以下のシェルでできました。

#!/usr/bin/bash
MY_BUCKET=S3対象のバケット
MY_CSV=対象ファイル
MY_ZIP=$MY_CSV.zip
MY_PASS=パスワード
cd /tmp

if test -z $MY_CSV; then
        exit 1
fi

if test -z $MY_PASS; then
        exit 2
fi

rm -f $MY_CSV $MY_ZIP

aws s3 cp s3://$MY_BUCKET/$MY_CSV  .
if test $? -ne 0; then
        exit 3
fi

zip -e --password $MY_PASS $MY_ZIP $MY_CSV
if test $? -ne 0; then
        exit 4
fi

aws s3 mv $MY_ZIP s3://$MY_BUCKET
if test $? -ne 0; then
        exit 5
fi

exit 0

これを「Lamdba」から呼出せばと安直にかんがえていましたが、「Lamdba」と「CloudShell」は、全く別空間なので呼び出せません。

「Lamdba」の「レイヤー」でシェルを試したりしましたが、結局「Lamdba」でUNIXシェルは接動かせず、「Lamdba」の上でZipコマンドは未サポートでした。この作戦はNG。「EC2」なら呼べるらしいですが、今回は常に仮想サーバが必要でシステムではないので控えました。

Python側の「zipfile」に頼ります。ヘルプではいちおうパスワードが指定できるようになっていますね。

ZipFile.setpassword(pwd)
Set pwd (a bytes object) as default password to extract encrypted files.

しかしこれは読込時のみでした。書込み時は、pythonから未サポートの旨のエラーが出ます。なんかzipの暗号化はライセンス上の制約があるとか、昔なにかで聞いたことがある気がします。

ネットで探すと、暗号化zipのpythonライブラリ「pyminizip」がありますね。「EC2」にもっていってためします。(EC2用に再Buildが必要だったかもしたかもしれません)

とりあえず「Lamdba」の「レイヤー」に追加するため、「pyminizip」をzipにまとめてから、PCローカルに「scp」でコピーします。windowsから行う場合、少し事前準備が必要でした。

「pyminizip.zip」を、「Lamdba」の「レイヤー」に追加して、Lamdba関数側で、pyminizpをimportします。実行すると、import時点でエラーになってしまいます。

他ネット記事によると、python3用にいじってリビルドするといけるらしいのですが、

業務での使用ですので際どいことは避け、信頼性の面から、控えておきました。

そのうちサポートされるかもしれませんし…

たくさんの機能があるAmazonサーバ。ユーザがダウンロードするファイルは、容量うんぬんというよりはマナー的にzip化したいところです。残念ながら「S3」の機能ではサポートされていないようです。弊方では以下を試しました。

その1：Lambdaから「zipコマンド」を呼出す

まずはUNIXコマンドを使用した方が、処理も早くて信頼性がありそうです。しかし、Lambdaからzipを呼び出したしましたが、lambdaテスト画面で、コマンドなし エラーになりました。

“ls”はたたけるようなので、/usr/binをリストしてみると、基本コマンドしか配置されてないようです。「CloudShell」(何用?)や「EC2」(仮想サーバ)では、zipはつかえるのですけどね。残念。

その2：Lambdaでpythonの「Zipfile」を使用する

以下のように「S3」から/tmpにコピーして、python上でzip化し、「S3」に戻すようにしてみました。

import boto3
import zipfile

def lambda_handler(event, context):
            :
          中  略
            :
        csv_fname = "対象ファイル名"

        zipname = csv_fname + ".zip"

        s3res = boto3.resource('s3')
        myBucket = s3res.Bucket("S3バケット名")
        myBucket.download_file(csv_fname, "/tmp/" + csv_fname)  

        with zipfile.ZipFile("/tmp/" + zipname, 'w',
            compression=zipfile.ZIP_DEFLATED) as zf:
            zf.write("/tmp/" + csv_fname, csv_fname)    

        myBucket.upload_file("/tmp/" + zipname, zipname)
    
        os.remove( "/tmp/" + csv_fname)    
        os.remove( "/tmp/" + zipname) 
        s3clt.delete_object( Bucket=MY_BUCKET, Key=csv_fname )

これで速度、信頼性もそこそこ問題ないようです。むろん本番用では、各fileの実在checkと、try – except を入れます。Lambdaのストレージサイズも必要サイズにUPしておきます。

AWS Lambdaで、予期せぬエラーが発生します。どうみてもコードは間違っていません。もしもコード修正してから、デプロイするまでの時間が短い場合、最後のコード修正が反映されていない可能性があります。

分かりやすい例としては以下のようなエラー。

デプロイしたコードは以下のとおりです。

たしかにデプロイ直前に、BSキーで、”lsp”→”ls”にしました。表示上の修正と、デプロイ内容に時間差があるようです。とりいそぎコメント行を少し変えて、デプロイボタンを有効化し、再デプロイします。

コードが少量で、文字列部などなら気付きやすいですが、コードがおおくて制御コード部だと何が起こったのか気付きにくいです。以下の習慣を身に着ける必要があるようです。

「修正した行から、一旦カーソル移動させて、一呼吸まってデプロイする」

でもすぐに忘れて、同じ過ちを繰り返してしまいます。

AWSのストレージ、Lanbdaのストレージメニューから誘われ「EFS」を使おうとしましたが、「EFS」のための別機能の設定に次ぐ設定の輪廻にはまり、最終的にエラーで続行不能。別のストレージ機能「S3」があることを知り、素直に使えてびっくりです。

「S3」は、WEBからのダウンロードファイルの生成場所に適当です。しかし、ユーザのダウンロード済みを検出して、削除するのは面倒です。

S3には、「ライフサイクル」タブがあり、バケットに以下のような設定を行うと自動削除ができました。ただし削除されるまでの最小時間は1日です。

その他、以下の設定も必要でした。

AWSのサーバ内部プログラムであるLambda。WEBページを構築する場合、通常は以下の手順となります。

1. ブラウザから、Lambda関数に割当てられたAWS Gateway のURLを呼出し。
2. そのLambda内でDynamoDBにアクセス。
3. Lambda内で、HTTPレスポンスを編集し返却。
4. プラウザ側でそのレスポンスを受信。

AWS Gatewayを呼び出すには、以下のようにJavascriptのfetch()を使います。(CGIとしてcallも可能でしたがブラウザから警告がでました。)

var myHeaders = new Headers();
myHeaders.append("Content-Type", "application/json");
var raw = JSON.stringify( { 
	 "パラメタ名" : パラメタ値,
          中  略
} );
var requestOptions = {
 	method: 'POST',
        headers: myHeaders,
        body: raw,
        redirect: 'follow'
};
fetch( "https://xxxxxxxxxx.execute-api.ap-northeast-1.amazonaws.com/dev", requestOptions )
    	.then( response => response.text() )
        .then( result => procResponce( result ) )
    	.catch( error => fetch_error( error ) );

fetchのエラーを、fetch_error()関数でキャッチしているのですが、たまに”TypeError: Failed to fetch”が出ます。

うまく呼べるときは呼べるのですが、割とサーバへリクエストする間隔が短いとき発生する感じです。Lambda関数が何処まで実行されているかは、普通の作込みではわかりません。HTTPレスポンス(procResponce()の関数の呼出し)が得られていないで、途中で停止したのでしょうか?

原因は簡単でした。Lambda関数のメモリ容量が少なすぎました。デフォルト128MBはいくら何でも小さすぎます。調整箇所は下図です。

Lambda関数のタイムアウト値は下図のように直ぐエラーがでるので気付きやいすですが、メモリ使用量(しかもまさかの小容量)は気付きにくいです。

弊方は2048までUP。これでレスポンスも含め、すごく調子よくなりました。Lambda関数生成時に、指定項目になってくれるとGoodですね。

癖のあるDynamoDBですが、使いやすい一面もあります。

Backupスケジュールが手軽に設定できます。

テーブルの管理画面から、バックアップのタブを選びます。

「バックアップ作成」→「バックアップのスケジュールを設定」からバックアップのプラン選択句に移ります。

細かい操作は割愛しますが、週一回のバックアップが仕込めました。

バックアップ状況が一覧でみれます。なんか二重にバックアップファイルができますが、これはよくわかりません。

RDBのバックアッブコマンドを、ジョブスケジューラに登録したりしなくてよいです。ストレージも勝手に割り当ててくれるようです。

ちょっとマニアックで癖のあるAWS DynmoDB。

テーブルアクセス時には、まず件数を知りたいとこです。SQL文なら以下のような感じ。

SELECT count(*) from テーブル名

DynmoDBで、pythonなら以下のような感じです。

tbl1 = dynamodb.Table("テーブル名") 
resp = tbl1.scan(   Select="COUNT", 
                    ReturnConsumedCapacity='TOTAL' )
count   = resp['Count']   

簡単できたと思いましたが落とし穴が、、、

まず感覚的に、他RBMSのSELECT count(*) に比べなんか重い。2万件程度、テーブルにデータを追加してみたところ、エラーが発生しました。

原因は、読出し一回1MB制限でNGでした。count値だけなら1MBに達しないと解釈していましたが、内部的にデータも取り出してくるようです。query でもscanでも同じ。

件数取得したいだけなのに、以下のようにloopにしないといけません。ザックリ1万件づつ読出し。

isLoop = True
totalCount = 0
lastKey=""

while( isLoop == True ):
    if ( lastKey == "" ):  # First scan
        resp = tbl1.scan(   Select="COUNT", 
                            ReturnConsumedCapacity='TOTAL',
                            Limit=10000 )
    else                   # Second or later scan
        resp = tbl1.scan(   Select="COUNT", 
                            ExclusiveStartKey=lastKey,
                            ReturnConsumedCapacity='TOTAL',
                            Limit=10000 )
    count   = resp['Count']   
    totalCount += count 
    #ラストキーがあれば、続きあり 
    if 'LastEvaluatedKey' in resp :  
        lastKey = resp['LastEvaluatedKey']
    else:                       # 続きなし 
        isLoop = False          # loop終了
 

おまけにLambdaのタイムアウトやメモリの動作設定の不足でよく止まってしまいます。弊方の構築したWEBサイトで最も重い処理になってしまいました。

スマートに件数取得を行うには、管理用のテーブルを用意して、カウント対象のテーブルの件数を記録しておくしかなさそうです。