はじめに
筆者はWorkspacesを運用していますが、運用の中で見つけたちょっとした工夫を紹介します。Workspacesとは
この記事では実際にWorkspacesを運用されている方を対象とするので、VDIとは？Workspacesとは？は省略します。詳しくは公式参照。 [Amazon WorkSpaces](https://aws.amazon.com/jp/workspaces/)なぜ最終ログイン日時が必要なのか
AWSのWorkspacesを運用していると、インスタンス作ったけど本当は使っていない人はいるかな、が知りたくなります。あなたの最終ログイン日はXヶ月前ですけどもう使わないですか？と言う棚卸しを行いたい訳です。最終ログイン日時はどうやって取得できるか
マネジメントコンソールのインスタンス一覧表示には最終ログイン日時がありません。1台づつ詳細を見れば出てくるのですが。。。 ここから調査開始。まずはAPI仕様を調べてみると、確かにインスタンス一覧を取得する [DescribeWorkspaces](https://docs.aws.amazon.com/workspaces/latest/api/API_DescribeWorkspaces.html) には最終ログイン日がない。。。よくよくAPIドキュメントを見てみると DescribeWorkspacesConnectionStatus になら最終ログイン日時 LastKnownUserConnectionTimestamp が見つかる！
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.
一覧取得にPythonを使う
API仕様は分かったので、Pythonでプログラム作成。必要な権限は環境変数に持っている前提で、CSV形式でインスタンス一覧＋最終ログイン日時を吐き出すプログラムを作りました。 ちょっと工夫したのは、DirecotryIDがCSVで表示されても大体何だったか分からなく鳴るので、DirecotryNameを表示するように調整。 `DirecotryIds` に運用しているDirectoryIDを配列で入れれば動くはず。Python 2.7.10です。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)