Amazon Timestream でのマルチメジャーレコード、マグネティックストレージへの書き込み、スケジュールドクエリを利用した時系列データの格納と分析

AWSTemplateFormatVersion: '2010-09-09'
Description: Create a Timestream database and a table
Resources:
  MyDatabase:
    Type: AWS::Timestream::Database
  MyTable:
    Type: AWS::Timestream::Table
    Properties: 
      DatabaseName: !Ref rDatabase
      RetentionProperties:
        MemoryStoreRetentionPeriodInHours: "24"
        MagneticStoreRetentionPeriodInDays: "7"
Outputs:
  DatabaseName:
    Description: Timestream Database Name
    Value: !Ref MyDatabase
  TableName:
    Description: Timestream Table Name
    Value: !GetAtt MyTable.Name

aws cloudformation create-stack \
    --stack-name devops-time-series \
    --template-body file://template.yaml

import time
import boto3
import psutil
import os

from botocore.config import Config

DATABASE_NAME = os.environ['DATABASE_NAME']
TABLE_NAME = os.environ['TABLE_NAME']

COUNTRY = "UK"
CITY = "London"
HOSTNAME = "MyHostname" # You can make it dynamic using socket.gethostname()

INTERVAL = 1 # Seconds

def prepare_common_attributes():
    common_attributes = {
        'Dimensions': [
            {'Name': 'country', 'Value': COUNTRY},
            {'Name': 'city', 'Value': CITY},
            {'Name': 'hostname', 'Value': HOSTNAME}
        ],
        'MeasureName': 'utilization',
        'MeasureValueType': 'MULTI'
    }
    return common_attributes


def prepare_record(current_time):
    record = {
        'Time': str(current_time),
        'MeasureValues': []
    }
    return record


def prepare_measure(measure_name, measure_value):
    measure = {
        'Name': measure_name,
        'Value': str(measure_value),
        'Type': 'DOUBLE'
    }
    return measure


def write_records(records, common_attributes):
    try:
        result = write_client.write_records(DatabaseName=DATABASE_NAME,
                                            TableName=TABLE_NAME,
                                            CommonAttributes=common_attributes,
                                            Records=records)
        status = result['ResponseMetadata']['HTTPStatusCode']
        print("Processed %d records. WriteRecords HTTPStatusCode: %s" %
              (len(records), status))
    except Exception as err:
        print("Error:", err)


if __name__ == '__main__':

    print("writing data to database {} table {}".format(
        DATABASE_NAME, TABLE_NAME))

    session = boto3.Session()
    write_client = session.client('timestream-write', config=Config(
        read_timeout=20, max_pool_connections=5000, retries={'max_attempts': 10}))
    query_client = session.client('timestream-query') # Not used

    common_attributes = prepare_common_attributes()

    records = []

    while True:

        current_time = int(time.time() * 1000)
        cpu_utilization = psutil.cpu_percent()
        memory_utilization = psutil.virtual_memory().percent
        swap_utilization = psutil.swap_memory().percent
        disk_utilization = psutil.disk_usage('/').percent

        record = prepare_record(current_time)
        record['MeasureValues'].append(prepare_measure('cpu', cpu_utilization))
        record['MeasureValues'].append(prepare_measure('memory', memory_utilization))
        record['MeasureValues'].append(prepare_measure('swap', swap_utilization))
        record['MeasureValues'].append(prepare_measure('disk', disk_utilization))

        records.append(record)

        print("records {} - cpu {} - memory {} - swap {} - disk {}".format(
            len(records), cpu_utilization, memory_utilization,
            swap_utilization, disk_utilization))

        if len(records) == 100:
            write_records(records, common_attributes)
            records = []

        time.sleep(INTERVAL)

export TABLE_NAME=`aws cloudformation describe-stacks --stack-name devops-time-series --query 'Stacks[0].Outputs' --output text | grep TableName | cut -f 3`
export DATABASE_NAME=`aws cloudformation describe-stacks --stack-name devops-time-series --query 'Stacks[0].Outputs' --output text | grep DatabaseName | cut -f 3`
python3 collect.py

./run.sh
writing data to database MyDatabase-0XM3OgacnHQR table MyTable-XAX9sYiaV5Ig
records 1 - cpu 40.0 - memory 67.6 - swap 62.7 - disk 26.2
records 2 - cpu 14.4 - memory 67.6 - swap 62.7 - disk 26.2
records 3 - cpu 13.1 - memory 67.6 - swap 62.7 - disk 26.2
records 4 - cpu 12.1 - memory 67.7 - swap 62.7 - disk 26.2
records 5 - cpu 10.0 - memory 66.9 - swap 62.7 - disk 26.2
...
records 95 - cpu 9.2 - memory 66.2 - swap 62.7 - disk 26.2
records 96 - cpu 10.3 - memory 66.2 - swap 62.7 - disk 26.2
records 97 - cpu 11.4 - memory 66.2 - swap 62.7 - disk 26.2
records 98 - cpu 11.7 - memory 66.2 - swap 62.7 - disk 26.2
records 99 - cpu 13.7 - memory 66.2 - swap 62.7 - disk 26.2
records 100 - cpu 9.3 - memory 66.2 - swap 62.7 - disk 26.2
Processed 100 records. WriteRecords HTTPStatusCode: 200

SELECT country, hostname, city, avg(cpu) AS avg_cpu,
       avg(memory) AS avg_memory, avg(disk) AS avg_disk,
       avg(swap) AS avg_swap, bin(time, 10m) AS binned_time
  FROM $__database.$__table
 GROUP BY country, hostname, city, bin(time, 10m)
 ORDER BY binned_time

  MyAggregatedTable:
    Type: AWS::Timestream::Table
    Properties: 
      DatabaseName: !Ref MyDatabase
      RetentionProperties:
        MemoryStoreRetentionPeriodInHours: "24"
        MagneticStoreRetentionPeriodInDays: "7"

aws cloudformation update-stack --stack-name devops-time-series --template-body file://template.yaml

SELECT country, hostname, city,
  avg(cpu) AS avg_cpu,
  avg(memory) AS avg_memory,
  avg(disk) AS avg_disk,
  avg(swap) AS avg_swap,
  bin(time, 10m) AS binned_timestamp
 FROM "MyDatabase-0XM3OgacnHQR"."MyTable-XAX9sYiaV5Ig"
GROUP BY country, hostname, city, bin(time, 10m)

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "timestream:Select",
                "timestream:DescribeTable",
                "timestream:ListMeasures"
            ],
            "Resource": "arn:aws:timestream:eu-west-1:123412341234:database/MyDatabase-0XM3OgacnHQR/table/MyTable-XAX9sYiaV5Ig"
        },
        {
            "Effect": "Allow",
            "Action": [
                "timestream:DescribeEndpoints",
                "timestream:SelectValues",
                "timestream:CancelQuery"
            ],
            "Resource": ""
        },
        {
            "Action": [
                "timestream:WriteRecords"
            ],
            "Resource": [
                "arn:aws:timestream:eu-west-1:123412341234:database/MyDatabase-0XM3OgacnHQR/table/MyAggregatedTable-SWLoLKjEXrDC"
            ],
            "Effect": "Allow"
        },
        {
            "Action": [
                "timestream:DescribeEndpoints"
            ],
            "Resource": "",
            "Effect": "Allow"
        }
    ]
}

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "VisualEditor0",
            "Effect": "Allow",
            "Action": "sns:Publish",
            "Resource": "arn:aws:sns:eu-west-1:123412341234:timestream-scheduled-query"
        }
    ]
}

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "VisualEditor0",
            "Effect": "Allow",
            "Action": "s3:PutObject",
            "Resource": "arn:aws:s3:::my-bucket/*"
        }
    ]
}

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "Service": "timestream.amazonaws.com"
      },
      "Action": "sts:AssumeRole"
    }
  ]
}

aws cloudformation delete-stack —stack-name devops-time-series