CloudWatch logsをgolangで操作してみる

CloudWatch logsをgolangで操作するには？

CloudWatchはかなりAWSの中でもかなり便利なサービスで大好きなんですが、いかんせんSDKにクセがあるので、
自分用のメモがてらラッパーライブラリを実装してみました。

作った機能

業務で使っている限り、大部分の時間はLogs Insightで検索したりしているのですが、
これが結構もっさりしてたり、そもそもコンソール開くのが煩わしいので、CLI化したい、というのがモチベーションでした。

ただ、APを新規で作っているときはメッセージ保存とかも当然やるので、必要そうな機能は実装しました。

1. LogGroup
   1. LogGroup一覧を返す
   2. メッセージ検索
2. LogStream
   1. 新規作成
   2. LogStream一覧を返す
3. LogEvent
   1. 新規作成
   2. 保存

リポジトリ

install

go get gitlab.com/kuritayu/logs

Getting Started

初期化

セッション情報を設定し、logs.Newを呼び出します。

cmd/main.go

func main() {

	region := "ap-northeast-1"
	sess := session.Must(
		session.NewSession(&aws.Config{
			Region: aws.String(region),
		}))
	cloudwatch := logs.New(sess)  // ココ

LogGroup

一覧

FindAllでLogGroupの一覧を取得します。Lambdaとか大量に使うシステムだと画面では1ページにおさまらないので、
一覧はほしくなりました。

cmd/main.go

	// read log group
	log.Println("ALL Log Group")
	log.Println(cloudwatch.LogGroup().FindAll())

検索

メインにやりたかったところです。LogGroup、キーワード、レンジ(From/To)を指定し、全ストリームを検索しています。
GrepByMessageはLogs Insightsで使うクエリを内部的に使っています。
ポイントは、AWS SDKのStartQueryが投げるとクエリIDを返してバックグラウンドで動くところでしょうか。
何らかの待ちを加えないと、Queryが終わる前に結果を取りに行く→結果なしになってしまいます。
この例では、time.sleepでやりましたが、ループ処理をする等、結果の拾い方は検討の余地がありそうです。

cmd/main.go

	// query
	log.Println("query")
	query, err := cloudwatch.LogGroup().GrepByMessage(
		"/aws/lambda/get-sqs",
		"test",
		"2021-09-22T00:00:00+09:00",
		"2021-09-22T23:59:59+09:00",
	)

	if err != nil {
		log.Println(err)
	}
	log.Println(query)

lib/logs/LogGroup.go

func (lg LogGroup) GrepByMessage(logGroup string, keyword string, from string, to string) (string, error) {

	// クエリの構築
	query := fmt.Sprintf("fields @timestamp, @message, @logStream | filter @message like /%v/", keyword)

	// クエリ投入定義構築
	input := &cloudwatchlogs.StartQueryInput{
		EndTime:      aws.Int64(UnixMillisecond(ParsedTo)),
		LogGroupName: aws.String(logGroup),
		QueryString:  aws.String(query),
		StartTime:    aws.Int64(UnixMillisecond(ParsedFrom)),
	}

	// クエリ実行
	out, err := lg.Logs.client.StartQuery(input)
	if err != nil {
		return "", err
	}

	return aws.StringValue(out.QueryId), nil

}

結果の受け取りの部分です。クエリIDを指定すると、結果を取得できるようにしています。
ここでのポイントは、@timestampフィールドがUTCで返ってくることでしょうか。見づらいので、JST変換をしています。
また、過去の体験上、Logs Insigihtsでメッセージ全検索→Stream名取得→該当Streamの付近を見る、のケースがほとんどだったので、
今回のクエリはquery := fmt.Sprintf("fields @timestamp, @message, @logStream | filter @message like /%v/", keyword)で十分と思い、ハードコーディングしました。

cmd/main.go

	time.Sleep(5 * time.Second)
	result, err := cloudwatch.LogGroup().Result(query)
	if err != nil {
		log.Println(err)
	}

	for _, line := range result {
		fmt.Printf("%v %v %v\n", logs.Jst(line.Timestamp), line.LogStream, line.Message)
	}

LogStream

LogGroupとほぼ実装は同じでした。

新規作成

main.go

	// create log stream
	log.Println("create log stream")
	err := s.Save("testStream")
	if err != nil {
		log.Println(err)
	}

一覧

main.go

	// read log stream
	log.Println("ALL Log Stream")
	s := cloudwatch.LogStream("/aws/lambda/get-sqs")
	log.Println(s.FindAll())

LogEvent

全件取得

LogGroupとLogStreamを指定することで全件取得できるようにしました。

main.go

	// read log event
	log.Println("FindAll Log Event")
	e := cloudwatch.LogEvent(
		"/aws/lambda/get-sqs",
		"2021/09/12/[$LATEST]7d7fe71b3aac4a649003f10d9085752b")
	events, _ := e.FindAll()
	for _, event := range events {
		fmt.Print(event.Message)
	}

ライブラリ側の実装としては、Tokenを使ってTokenがある限りループする実装を組む必要があるところがポイントでした。

lib/logs/LogEvent.go

func (le LogEvent) FindAll() ([]LogContent, error) {
	var events []LogContent

	for {
		items, err := le.Logs.client.GetLogEvents(&le.Input)
		if err != nil {
			return nil, err
		}

		for _, item := range items.Events {
			event := &LogContent{
				TimeStamp: aws.Int64Value(item.Timestamp),
				Message:   aws.StringValue(item.Message),
			}

			events = append(events, *event)
		}

		if aws.StringValue(items.NextBackwardToken) == aws.StringValue(le.Input.NextToken) {
			break
		}

		le.Input.NextToken = items.NextBackwardToken

	}

	return events, nil
}

保存

main.go

	// create log event
	log.Println("create log event")
	e.LogStreamName = "testStream"
	err = e.Save("test message")
	if err != nil {
		log.Println(err)
	}

こちらも、Tokenに気をつけないとうまく保存ができないです。
特に、Streamに１件でもデータがある場合はTokenがないとエラーになります。

lib/logs/LogEvent.go

func (le LogEvent) Save(message string) error {

	// トークンを取得する
	token, err := le.fetchToken()
	if err != nil {
		return err
	}

	// 時刻をミリ秒に変換する
	eventTime := now()

	var inputLogEvents []*cloudwatchlogs.InputLogEvent

	inputLogEvent := &cloudwatchlogs.InputLogEvent{
		Message:   aws.String(message),
		Timestamp: aws.Int64(eventTime),
	}

	inputLogEvents = append(inputLogEvents, inputLogEvent)
	return le.doSave(inputLogEvents, token)

}

func (le LogEvent) doSave(inputLogEvent []*cloudwatchlogs.InputLogEvent, token *string) error {
	input := &cloudwatchlogs.PutLogEventsInput{
		LogEvents:     inputLogEvent,
		LogGroupName:  aws.String(le.LogGroupName),
		LogStreamName: aws.String(le.LogStreamName),
		SequenceToken: token,
	}

	_, err := le.Logs.client.PutLogEvents(input)
	if err != nil {
		return err
	}

	return nil

}