デバイス別アクセスログの集計処理をLambdaに置き換えました

func lambdaHandler(event events.S3Event) error {
	region := os.Getenv("S3_REGION")
	sess, err := session.NewSession(&aws.Config{
		Region: aws.String(region)},
	)
	if err != nil {
		log.Println(err)
		return err
	}
	client := s3.New(sess)

	config, err := aggregate.NewConfig()
	if err != nil {
		log.Println(err)
		return err
	}

	var accessLog aggregate.AccessLog
	outputPlugin := aggregate.NewOutputPlugin(config)

	// オブジェクト取得
	// Lambdaはファイル毎にイベントがtriggerされるため、forでループはせず、event.Recordsの0番目を取得する
	record := event.Records[0]
	decodedKey, err := url.QueryUnescape(record.S3.Object.Key)
	if err != nil {
		log.Println("Failed to decodeKey(). Key: ", record.S3.Object.Key, ". err: ", err)
		return err
	}
	obj, err := client.GetObject(&s3.GetObjectInput{
		Bucket: aws.String(record.S3.Bucket.Name),
		Key:    aws.String(decodedKey),
	})
	if err != nil {
		log.Println("Failed to client.GetObject(). BucketName: ", record.S3.Bucket.Name, ". Key: ", record.S3.Object.Key, ". err: ", err)
		return err
	}

	rc := obj.Body
	defer rc.Close()

	gz, err := gzip.NewReader(rc)
	if err != nil {
		log.Println("Failed to gzip.NewReader(). BucketName: ", record.S3.Bucket.Name, ". Key: ", record.S3.Object.Key, ". err: ", err)
		return err
	}
	defer gz.Close()
}

// defer gz.Close()の続き
scanner := bufio.NewScanner(gz)
// ファイルの中身を一行ずつ取得し集計する
for scanner.Scan() {
    data := scanner.Text()
	  err = json.Unmarshal([]byte(data), &accessLog)
if err != nil {
		log.Println("Failed to json.Unmarshal(). data: ", data, ". err: ", err)
		return err
	}
	err = outputPlugin.AddEvent(&accessLog)
	if err != nil {
		// エラーになったわけではなく、単にそのレコードを無視しただけなのでログをはくだけで、returnしない
		log.Println(err)
	}
}

type OutputPlugin struct {
	accessCountRepo *AccessCountRepo
	egressEvents    []*EgressEvent
}

func (p *OutputPlugin) AddEvent(accessLog *AccessLog) error {
	egressEvent, err := NewEgressEvent(accessLog)
	if err != nil {
		// ignore: malformed ingress record
		return err
	}
	p.egressEvents = append(p.egressEvents, egressEvent)
	return nil
}

func analyzeAgent(agent string) DeviceType {
	ua := useragent.Parse(agent)
	if ua.Bot {
		return Undefined
	}
	if ua.Mobile {
		return Smartphone
	}
	if ua.Tablet {
		return Tablet
	}
	return PC
}

// 上のmain.goの続き
ctx := context.Background()
ctx, cancel := context.WithTimeout(ctx, Timeout)
defer cancel()
err = outputPlugin.Flush(ctx)
if err != nil {
	log.Println(err)
	return err
}

return nil

{
  "Architectures": [
    "x86_64"
  ],
  "Description": "{{ must_env `DESCRIPTON` }}",
  "Environment": {
    "Variables": {
      "DYNAMODB_REGION": "{{ must_env `DYNAMODB_REGION` }}",
      "DYNAMODB_TABLE_NAME": "{{ must_env `DYNAMODB_TABLE_NAME`}}",
      "S3_REGION": "{{ must_env `S3_REGION` }}"
    }
  },
  "EphemeralStorage": {
    "Size": 512
  },
  "FunctionName": "{{ must_env `FUNCTION_NAME` }}",
  "Handler": "{{ must_env `HANDLER_NAME` }}",
  "MemorySize": 2048,
  "Role": "{{ must_env `FUNCTION_ROLE` }}",
  "Runtime": "go1.x",
  "SnapStart": {
    "ApplyOn": "None"
  },
  "Tags": {
    "Project": "{{ must_env `FUNCTION_NAME` }}"
  },
  "Timeout": 900,
  "TracingConfig": {
    "Mode": "PassThrough"
  }
}

bin/device_access_log: cmd/device_access_log/main.go 
	go build -o bin/device_access_log cmd/device_access_log/main.go

# ステージング環境
.PHONY: deploy_stg_device_access_log
deploy_stg_device_access_log: bin/device_access_log
	DESCRIPTON='stg-description' \
	FUNCTION_NAME=stg-device-access-log \
	HANDLER_NAME='bin/device_access_log' \
	DYNAMODB_REGION='stg-dynamodb-region' \
	DYNAMODB_TABLE_NAME='stg-dynamodb-table-name' \
	S3_REGION='stg-s3-region' \
	FUNCTION_ROLE='stg-function-roll' \
	lambroll deploy --function="function.json"