GuanceAmazon ECS ログの収集¶
はじめに¶
Amazon Elastic Container Service（Amazon ECS）は、高度にスケーラブルで高速なコンテナ管理サービスです。これを使用して、クラスター上のコンテナを簡単に実行、停止、管理できます。これらのコンテナは、自身の EC2 サーバー上で実行することも、AWS Fargate が管理するサーバーレスインフラストラクチャ上で実行することもできます。
Fargate 起動タイプを使用するタスクの場合、コンテナの awslogs ログドライバー を起動する必要があります。コンテナ内で動作するアプリケーションが STDOUT および STDERR I/O ストリームで出力するログは、CloudWatch Logs のロググループに送信され、その後 Func によってこれらのログが収集され、Func は EC2 にデプロイされた DataKit を介してログを Guance に書き込みます。
本ドキュメントのログ収集は、AWS Fargate が管理するコンテナを対象としています。
前提条件¶
- 事前に Guance アカウント を作成する必要があります。
- DataKit のインストール
- Func 携帯版のインストール
- ECS で動作する Java アプリケーションがすでに存在していること
ここで使用する ECS クラスター名は cluster-docker です。以下でサンプルのログとロググループを確認します。「AWS」にログインし、「Elastic Container Service」→「クラスター」→「cluster-docker」の順にクリックします。
「サービス名」をクリックします。
タスクに入ります。
詳細情報タブのコンテナの下にあるログ設定を見つけます。
「ログタグ」をクリックします。中にはアプリケーションのログが表示されています。次に、これらのログを収集します。
操作手順¶
Warning
サンプルで使用するバージョンは DataKit 1.4.18 です。
手順 1 AWS 構成¶
1.1 ユーザーキー¶
ECS のデプロイに使用したアカウントを使用します。AWS がこのユーザーを作成した際に提供された Access key ID と Secret access key は後で使用します。
1.2 AWS ユーザー権限の設定¶
AWS の IAM コンソールにログインし、ユーザーの下にある ECS の「ユーザー」を見つけ、「アクセス許可の追加」をクリックします。
「既存のポリシーを直接アタッチ」をクリックし、「ポリシーのフィルタリング」で CloudWatchLogsReadOnlyAccess、CloudWatchEventsReadOnlyAccess を選択し、「次へ：確認」をクリックします。
手順 2 Func 構成¶
2.1 環境変数の設定¶
「Func」にログインし、「開発」→「環境変数」→「環境変数を追加」の順に進みます。ここでは 3 つの環境変数を追加します。
AWS_LOG_KEY値は手順 1.1 の AWS ユーザーのAccess key IDに対応します。AWS_LOG_SECRET_ACCESS_KEY値は手順 1.1 の AWS ユーザーのSecret access keyに対応します。AWS_REGION_NAME値は AWS ユーザーが属するREGIONに対応します。
2.2 コネクターの設定¶
「Func」にログインし、「開発」→「コネクター」→「コネクターを追加」の順に進みます。
ここでは ID を DataKit とし、ホストは DataKit がインストールされているアドレス、ポートは DataKit のポートとします。（本サンプルでは IP を直接使用するため、プロトコルは HTTP とします）
「接続テスト」をクリックし、
が返ってくれば、DataKit が利用可能であることを示します。
2.3 PIP ツールの設定¶
「Func」にログインし、「管理」→「実験的機能」と進み、右側で「PIP ツールモジュールを有効にする」を選択します。
左側の「PIP ツール」をクリックし、「Alibaba Cloud ミラー」を選択し、boto3 を入力し、「インストール」をクリックします。
2.4 スクリプトライブラリ¶
「Func」にログインし、「開発」→「スクリプトライブラリ」→「スクリプトセットを追加」の順に進みます。ID は任意で設定し、「保存」をクリックします。
「AWS ログ収集」を見つけ、「スクリプトを追加」をクリックします。
ID を入力します。サンプルでは aws_ecs__log と定義し、「保存」をクリックします。
「編集」をクリックします。
以下の内容を入力します。
入力内容
import boto3
import json
import time
scope_id='ecs_log'
@DFF.API('aws_ecs log', timeout=500, api_timeout=180)
def run(measurement, logGroupName, interval):
print(measurement, logGroupName, interval)
get_log_data(measurement, logGroupName, interval)
# if data is not None:
# push_log(data)
# else:
# print("None")
def get_cron_time(interval, measurement):
cache = DFF.CACHE.get('last_time_%s' %measurement,scope=scope_id)
if cache == None:
currentTime = int(round(time.time() * 1000))
startTime = currentTime - int(interval) * 1000
endTime = currentTime
else:
currentTime = int(round(time.time() * 1000))
if currentTime - int(cache) > 10 * 60 * 1000:
startTime = currentTime - int(interval) * 1000
endTime = currentTime
else:
startTime = int(cache) + 1
endTime = currentTime
print(startTime, endTime)
return startTime, endTime
def get_log_data(measurement, logGroupName, interval):
logTime = get_cron_time(interval, measurement)
startTime = logTime[0]
endTime = logTime[1]
isPush = False
client = boto3.client(
'logs',
aws_access_key_id=DFF.ENV('AWS_LOG_KEY'),
aws_secret_access_key=DFF.ENV('AWS_LOG_SECRET_ACCESS_KEY'),
region_name=DFF.ENV('AWS_REGION_NAME')
)# print(client.meta.config)
try:
nextToken = 'frist'
logData = []
while nextToken != '':
if nextToken == 'frist':
nextToken = ''
response = client.filter_log_events(
logGroupName=logGroupName,
startTime=startTime,
endTime=endTime,
limit=1000,
#filterPattern="?ERROR ?WARN ?error ?warn",
interleaved=False
)
else:
response = client.filter_log_events(
logGroupName=logGroupName,
startTime=startTime,
endTime=endTime,
nextToken=nextToken,
limit=1000,
#filterPattern="?ERROR ?WARN ?error ?warn",
interleaved=False
)
try:
if len(response['events']) > 0:
data = []
lastTimeList = []
for i in response['events']:
# print("hii", i['logStreamName'])
log = {
'measurement': measurement,
'tags': {
'logGroupName': logGroupName,
'logStreamName': i['logStreamName'],
'host': '127.0.0.1'
},
'fields': {
'message': i['message'],
'time': i['timestamp']
}
}
data.append(log)
lastTimeList.append(i['timestamp'])
push_log(data)
print("max %s" % max(lastTimeList))
DFF.CACHE.set('last_time_%s' % measurement, max(lastTimeList), scope=scope_id, expire=None)
isPush = True
else:
DFF.CACHE.set('last_time_%s' % measurement, endTime , scope=scope_id, expire=None)
nextToken = response['nextToken']
except:
nextToken = ''
except Exception as e:
print('Error: %s' % e )
return None
if not isPush:
DFF.CACHE.set('last_time_%s' % measurement, endTime , scope=scope_id, expire=None)
def push_log(data):
datakit = DFF.SRC('datakit')
status_code, result = datakit.write_logging_many(data=data)
if status_code == 200:
print("total %d" % len(data))
print(status_code, result)
Warning
- 上記の内容の4行目の
ecs_logは、同じ Func 内で一意である必要があります。他の文字に変更しても構いません。 - 6行目の
awc_ecsは、先ほど追加したスクリプトセットの ID です。 - 40、41、42 行目の
AWS_LOG_KEY、AWS_LOG_SECRET_ACCESS_KEY、AWS_REGION_NAMEは手順 2.1 の環境変数名に対応します。環境変数名を変更した場合は、それに合わせて修正する必要があります。
2.5 スクリプトのテスト¶
下図のように「run」を選択し、2 つ目の赤枠の内容では：
measurementの値はecs_log_sourceを入力します。これは Guance ログのログソースに対応します。logGroupNameの値は 前提条件 のログ設定で確認したawslogs-groupに対応します。intervalの値は収集頻度に対応します。サンプルでは 60 秒です。
「実行」をクリックし、total 8 が出力されれば、8 件のログが送信されたことを示します。
「Guance」にログインし、「ログ」モジュールに入り、データソースに「ecs_log_source」を選択すると、ログが表示されます。
右上の「公開」をクリックします。
右上の「編集終了」をクリックします。
2.6 自動ログ収集¶
「Func」にログインし、「管理」→「自動トリガー設定」→「新規作成」の順に進み、パラメータに先ほど実行した内容を入力します。
時間を毎分または 5 分ごとに選択し、「保存」をクリックします。
「自動トリガー設定」リストに「aws_ecs log」のレコードが存在します。
「最近の実行」をクリックして実行状況を確認します。
