Workspaces の最終ログイン日時を一覧取得するのが意外に面倒

<H1>はじめに</H1>
筆者はWorkspacesを運用していますが、運用の中で見つけたちょっとした工夫を紹介します。

<H1>Workspacesとは</H1>
この記事では実際にWorkspacesを運用されている方を対象とするので、VDIとは？Workspacesとは？は省略します。詳しくは公式参照。
[Amazon WorkSpaces](https://aws.amazon.com/jp/workspaces/)

<H1>なぜ最終ログイン日時が必要なのか</H1>
AWSのWorkspacesを運用していると、インスタンス作ったけど本当は使っていない人はいるかな、が知りたくなります。あなたの最終ログイン日はXヶ月前ですけどもう使わないですか？と言う棚卸しを行いたい訳です。

<H1>最終ログイン日時はどうやって取得できるか</H1>
マネジメントコンソールのインスタンス一覧表示には最終ログイン日時がありません。1台づつ詳細を見れば出てくるのですが。。。
ここから調査開始。まずはAPI仕様を調べてみると、確かにインスタンス一覧を取得する [DescribeWorkspaces](https://docs.aws.amazon.com/workspaces/latest/api/API_DescribeWorkspaces.html)  には最終ログイン日がない。。。

よくよくAPIドキュメントを見てみると [DescribeWorkspacesConnectionStatus](https://docs.aws.amazon.com/workspaces/latest/api/API_DescribeWorkspacesConnectionStatus.html)  になら最終ログイン日時 `LastKnownUserConnectionTimestamp` が見つかる！


``` javascript:DescribeWorkspacesConnectionStatus
//Response Syntax
{
   "NextToken": "string",
   "WorkspacesConnectionStatus": [ 
      { 
         "ConnectionState": "string",
         "ConnectionStateCheckTimestamp": number,
         "LastKnownUserConnectionTimestamp": number, //<-コレ！
         "WorkspaceId": "string"
      }
   ]
}
```

ただこのAPI、リクエストにWorkspaceIdを入れた上で、しかも25インスタンスまでしかリクエスト出来ないらしい。25台以上あると全量一括取得は出来ないみたい。
> WorkspaceIds
> The identifiers of the WorkSpaces. You can specify up to 25 WorkSpaces.




<H1>一覧取得にPythonを使う</H1>
API仕様は分かったので、Pythonでプログラム作成。必要な権限は環境変数に持っている前提で、CSV形式でインスタンス一覧＋最終ログイン日時を吐き出すプログラムを作りました。
ちょっと工夫したのは、DirecotryIDがCSVで表示されても大体何だったか分からなく鳴るので、DirecotryNameを表示するように調整。
`DirecotryIds` に運用しているDirectoryIDを配列で入れれば動くはず。Python 2.7.10です。

```python:getInstances.py
import boto3
import csv

# Initialize SDK client for AWS Workspaces
client = boto3.client('workspaces')
paginator_describe = client.get_paginator('describe_workspaces')
paginator_connection = client.get_paginator('describe_workspaces_connection_status')

# Initialize output CSV header
data = [["UserName", "WorkspaceId", "Directory", "State", "ComputerType", "LastLogin"]]


# Initialize Directory ID list
DirectoryIds = [
    'd-XXXXXXXX',
    'd-YYYYYYYY'
]

# Initialize mapping between Directory ID and Name
def addDirectoryMapping(direcotry):
    directoryMapping[direcotry["DirectoryId"]] = direcotry["Alias"]

directoryMapping = {}
paginator = client.get_paginator('describe_workspace_directories')
page_iterator = paginator.paginate()
for page in page_iterator:
    map(addDirectoryMapping, page["Directories"])
print("Direcotry mapping is DONE")


# Find Directory Name from ID
# @param  workspace ID
# @return Name of directory
def findDirectoryName(direcotyrId):
    if direcotyrId in directoryMapping:
        return directoryMapping[direcotyrId]
    else:
        return "Other"

# Find last login time by SingleId
# @param  workspace ID
# @return LastLogin Time
def findLatestLoginTimeById(workspaceId):
    response = client.describe_workspaces_connection_status(
        WorkspaceIds=[
            workspaceId
        ]
    )
    if not len(response["WorkspacesConnectionStatus"])==0:
        connectionStatus = response["WorkspacesConnectionStatus"][0]
        if "LastKnownUserConnectionTimestamp" in connectionStatus:
            return connectionStatus["LastKnownUserConnectionTimestamp"]
        else:
            return "unknown"
    else:
        return "Error"

# Get necessary information from Workspace Instance
# @param workspaces
# @return list
def parse_list(workspaces):
    list=[]
    list.append(workspaces["UserName"])
    list.append(workspaces["WorkspaceId"])
    list.append(findDirectoryName(workspaces["DirectoryId"]))
    list.append(workspaces["State"])
    list.append(workspaces["WorkspaceProperties"]["ComputeTypeName"])
    list.append(findLatestLoginTimeById(workspaces["WorkspaceId"]))
    return list

# Get necessary information from Workspace Connection Status
# @param WorkspacesConnectionStatus
# @return list
def parse_list_connection(WorkspacesConnectionStatus):
    list=[]
    list.append(WorkspacesConnectionStatus["WorkspaceId"])
    if "LastKnownUserConnectionTimestamp" in WorkspacesConnectionStatus:
        list.append(WorkspacesConnectionStatus["LastKnownUserConnectionTimestamp"])
    else:
        list.append("unknown")
    return list

# Export CSV file
def exportCSV(data):
    with open("VDI_LIST.csv", "w") as f:
        writer = csv.writer(f, lineterminator="\n")
        writer.writerows(data)
        print("Create File is DONE")

# main function
def describeWorkspacesByDirectoryId (Id):  
    try:
        workspacesDescribeInDirectory = []
        print(Id + " is started")
        # describe workspaces
        page_iterator_describe = paginator_describe.paginate(DirectoryId=Id,PaginationConfig={'PageSize': 10})
        for page in page_iterator_describe:
            workspacesDescribeInDirectory = map(parse_list, page["Workspaces"])
            data.extend(workspacesDescribeInDirectory)
        print(Id + " describe_workspaces is DONE")
    except Exception as e:
        print(e.message + " :: impacted Directory ID is " + Id)

map(describeWorkspacesByDirectoryId, DirectoryIds)
exportCSV(data)
```

<H1>もっと高度な利用状況分析</H1>
ちなみに、キチンとした利用分析をするサービスの[Workspaces Cost Optimizer](
https://docs.aws.amazon.com/solutions/latest/workspaces-cost-optimizer/welcome.html)がAWSでは用意されているみたいなので、本格的な運用ではコレを利用するのが良いみたい。

<H1>おわりに</H1>
月に１回バッチ実行するようにすれば良いのだろうけど、筆者は手動運用（笑）申請あったからインスタンス作ったのに一度もログインしたことがないユーザがそこそこいたので、意外と役に立ったプログラムでした。