利用AWS EKS CA结合AWS GameLift与Agones能力，实现游戏房间容器化调度

aws autoscaling create-or-update-tags --tags ResourceId=gamelift-gameservergroup-${GSGNAME},ResourceType=auto-scaling-group,Key=kubernetes.io/cluster/agones,Value=owned,PropagateAtLaunch=true ResourceId=gamelift-gameservergroup-${GSGNAME},ResourceType=auto-scaling-group,Key=k8s.io/cluster-autoscaler/agones,Value=enabled,PropagateAtLaunch=true ResourceId=gamelift-gameservergroup-${GSGNAME},ResourceType=auto-scaling-group,Key=Name,Value=FleetIQ,PropagateAtLaunch=true ResourceId=gamelift-gameservergroup-${GSGNAME},ResourceType=auto-scaling-group,Key=k8s.io/cluster-autoscaler/enabled,Value=true,PropagateAtLaunch=true --region ${AWS_REGION}

# 利用 --node 参数
command:
            - ./cluster-autoscaler
            - --v=4
            - --stderrthreshold=info
            - --cloud-provider=aws
            - --skip-nodes-with-local-storage=false
            - --expander=priority
            - --nodes=0:10:gamelift-gameservergroup-${GSGNAME}
            - --balance-similar-node-groups
            - --skip-nodes-with-system-pods=false
            
# 优先选择 gamelift-gameservergroup         
apiVersion: v1
kind: ConfigMap
metadata:
  name: cluster-autoscaler-priority-expander
  namespace: kube-system
data:
  priorities: |-
    10:
      - .*-non-existing-entry.*
    20:
      - gamelift-gameservergroup-${GSGNAME}

# 主动发现集群中的节点状态，推送到SQS中
...
"""
    Current status of the game server instance.
        ACTIVE -- The instance is viable for hosting game servers.
        DRAINING -- The instance is not viable for hosting game servers. Existing game servers are in the process of ending, and new game servers are not started on this instance unless no other resources are available. When the instance is put in DRAINING, a new instance is started up to replace it. Once the instance has no UTILIZED game servers, it will be terminated in favor of the new instance.
        SPOT_TERMINATING -- The instance is in the process of shutting down due to a Spot instance interruption. No new game servers are started on this instance.
"""          
def lambda_handler(event, context):
 
    queue = sqs.get_queue_by_name(QueueName=sqsName)
    
    gamelift = boto3.client('gamelift', region_name=os.getenv('AWS_REGION'))
    paginator = gamelift.get_paginator('describe_game_server_instances')
    #TODO get GAME_SERVER_GROUP_NAME from a ConfigMap and loop through the values
    groups = json.loads(os.getenv('CONFIG_TXT'))
    for group in groups['GameServerGroups']:
        pages = paginator.paginate(GameServerGroupName=group)
        for page in pages:
            game_servers = []
            for game_server in page['GameServerInstances']:
                
                game_server = decribe_instance(game_server)
                if 'PrivateDnsName' in game_server and len(game_server['PrivateDnsName']) > 0:
                    print(f'Publishing status on channel {game_server["InstanceId"]}', flush=True)
                    game_servers.append(game_server)

            if len(game_servers) == 0:
                continue
            randomsuffix = str(random.randint(1,1000))
            response = queue.send_message(
                MessageBody=json.dumps(game_servers), 
                MessageGroupId=group, 
                MessageDeduplicationId=group + randomsuffix)
                
            print(f'response = {response}')        
    return {
        'statusCode': 200,
        'body': json.dumps('Hello from Lambda!')
    }
...

# 主流程，根据是否有redis记录决定是否要初始化GameServer
def main_loop(messageId, message):
    set_messageId(messageId, message['InstanceId'])
    print_messageId(message)
    global r
    r = redis_conn()
    rkey = '{}.{}'.format(message['GameServerGroupName'], message['InstanceId'])
    if r.setnx(rkey, '1') == True:
        print_messageId(f'access {rkey}')
        initialize_game_server(message)
    
    update_health_status(message)
    get_health_status(message)
    pass

# 初始化GameServer: 将EC2注册到GameLift，更改状态到UTILIZED
def initialize_game_server(message):
    """This method registers the instance as a game server with Gamelift FleetIQ using the instance's Id as game server name.
    After registering the instance, it looks at result of DescribeAutoscalingInstances to see whether the instance is HEALTHY. 
    When HEALTHY, the instance is CLAIMED and its status is changed to UTILIZED. Finally, the taint gamelift.aws/status:ACTIVE,NoExecute
    is added to the node. Agones game servers need to have a toleration for this taint before they can run on this instance."""
    status = message['InstanceStatus']
    GameServerGroupName = message['GameServerGroupName']
    GameServerId = message['InstanceId']
    InstanceId = message['InstanceId']
    PrivateDnsName = message['PrivateDnsName']

    try:
        # Register game server instance
        print_messageId(f'Registering game server {GameServerId} PrivateDnsName: {PrivateDnsName}', flush=True)  
        gamelift.register_game_server(
            GameServerGroupName=GameServerGroupName,
            GameServerId=GameServerId,
            InstanceId=InstanceId
        )
    except gamelift.exceptions.ConflictException as error:
        print_messageId(f'The game server {GameServerId} is already registered', flush=True)
        #return None
    except Exception as e: 
        print_messageId(f'{e}', flush=True)
        deregister_game_server(GameServerGroupName, GameServerId)
        return None
    
    # Update the game server status to healthy
    # TODO Change this to use the new FleetIQ API DescribeGameServerInstances
    # TODO Consider using a decorator and backoff library to implement the backoff
    trytime = 0
    backoff = random.randint(1,5)
    while is_healthy(InstanceId) != 'HEALTHY':
        print_messageId(f'Instance is not healthy, re-trying in {backoff}', flush=True)
        sleep(backoff)
        trytime = trytime + 1
        # tried 3 times, let's return here
        if trytime == 3:
            return None
    
    update_health_status(message)
    
    # Claim the game server
    print_messageId(f'Claiming game server {GameServerId}', flush=True)
    try: 
        gamelift.claim_game_server(
            GameServerGroupName=GameServerGroupName,
            GameServerId=GameServerId
        )
    except gamelift.exceptions.ConflictException as error: 
        print_messageId('The instance has already been claimed', flush=True)
        
    # Update game server status 
    print_messageId(f'Changing server {GameServerId} status to utilized', flush=True)
    gamelift.update_game_server(
        GameServerGroupName=GameServerGroupName,
        GameServerId=GameServerId,
        UtilizationStatus='UTILIZED'
    )
    
# 通过 update_health_status 打上标签gamelift.aws/status:ACTIVE 
def update_health_status(message):
    status = message['InstanceStatus']
    GameServerGroupName = message['GameServerGroupName']
    GameServerId = message['InstanceId']
    InstanceId = message['InstanceId']
    PrivateDnsName = message['PrivateDnsName']

    try: 
        gamelift.update_game_server(
            GameServerGroupName=GameServerGroupName,
            GameServerId=GameServerId,
            HealthCheck='HEALTHY'
        )
        print_messageId(f'Updated Gamelift game server {GameServerId} health', flush=True)
    except gamelift.exceptions.NotFoundException as e:
        print_messageId(f'Skipping healthcheck, the node {GameServerId} is not registered', flush=True)
    
    # Adding taint ACTIVE to node
    taint = {
        "key": "gamelift.status/active",
        "value": "true",
        "effect": "NoExecute"
    }
    kubernetes_tools.taint_node(GameServerId, PrivateDnsName, json.dumps(taint))
    pass

# 异步检查节点的状态，提前处理DRAINING的节点

def get_health_status(message):
    """This is another asynchronous method that checks the health of the viability of the instance according to FleetIQ.
    It is subscribing to a Redis channel for updates because the DescribeGameServerInstances API has a low throttling rate.
    For this to work, a separate application has to be deployed onto the cluster.  This application gets the viability of 
    each instance and publishes it on a separate channel for each instance.  When the viability changes to DRAINING, the node
    is cordoned and tainted, preventing Agones from scheduling new game servers on the instance.  Game servers in the non-Allocated
    state will be rescheduled onto other instances."""

    status = message['InstanceStatus']
    GameServerGroupName = message['GameServerGroupName']
    GameServerId = message['InstanceId']
    InstanceId = message['InstanceId']
    PrivateDnsName = message['PrivateDnsName']

    if status == 'ACTIVE':
        return
    global r
    is_cordoned = (True, False)[r.hget(name="{InstanceId}", key="is_cordoned") == 1]
    is_ready_shutdown = (True, False)[r.hget(name="{InstanceId}", key="is_cordoned") == 1]
    is_waiting_for_termination = (True, False)[r.hget(name="{InstanceId}", key="is_cordoned") == 1]

    # is_cordoned = False
    # is_ready_shutdown = False
    # is_waiting_for_termination = False

    print_messageId('Starting message loop', flush=True)
    print_messageId(f'is_cordoned: {is_cordoned}, is_ready_shutdown: {is_ready_shutdown}, is_waiting_for_termination: {is_waiting_for_termination}', flush=True)
    print_messageId(f"Instance {GameServerId}/{PrivateDnsName} status is: {status}", flush=True)

    try:
        if status == 'DRAINING':
            print_messageId(f'Instance is no longer viable', flush=True)
            is_cordoned = cordon_and_protect(PrivateDnsName, GameServerId)
            is_cordoned = True
            r.hset(name="{InstanceId}", key="is_cordoned", value=(1, 0)[is_cordoned])

            print_messageId(f'Instance should have no allocted session', flush=True)
            is_ready_shutdown = is_ready_shutdown_check(PrivateDnsName)
            r.hset(name="{InstanceId}", key="is_ready_shutdown", value=(1, 0)[is_ready_shutdown])

            print_messageId(f'Instance should be deregistered', flush=True)
            deregister_game_server(GameServerGroupName, GameServerId)
            is_waiting_for_termination = True
            r.hset(name="{InstanceId}", key="is_waiting_for_termination", value=(1, 0)[is_waiting_for_termination])
            
            if is_waiting_for_termination == True:
                print_messageId(f'Waiting for termination signal', flush=True)
                
        elif status == 'SPOT_TERMINATING':
            if is_waiting_for_termination == True:
                # This is never invoked because the status never equals SPOT_TERMINATING  
                print_messageId(f'Received termination signal', flush=True)
                kubernetes_tools.drain_pods(InstanceId, PrivateDnsName)  
        else:
            pass     

    except Exception as e:
        print_messageId(f'get_health_status->error: {e}', flush=True)   

game_server["PrivateDnsName"] = instance.get("PrivateDnsName")
game_server["PrivateIpAddress"] = instance.get("PrivateIpAddress")
game_server["PublicDnsName"] = instance.get("PublicDnsName")
game_server["PublicIpAddress"] = instance.get("PublicIpAddress")

# 以instanceID为hash名字：
# is_cordoned: non-viable 节点已经被污染保护，等待被关闭，不接受新的pod
# is_ready_shutdown： 节点已经关闭
# is_waiting_for_termination:  节点已经关闭，等待注销

r.hset(name="{InstanceId}", key="is_cordoned", value=(1, 0)[is_cordoned])
r.hset(name="{InstanceId}", key="is_ready_shutdown", value=(1, 0)[is_ready_shutdown])
r.hset(name="{InstanceId}", key="is_waiting_for_termination", value=(1, 0)[is_waiting_for_termination])

kubectl describe --namespace agones-system pods

kubectl get pods --namespace agones-system

kubectl get fleets

kubectl get gameservers

watch kubectl describe gameserver

kubectl scale fleet stk-fleet --replicas=100watch kubectl get fleets
kubectl get nodes

kubectl scale fleet stk-fleet --replicas=1
watch kubectl get fleets
kubectl get nodes

# gameserverfleetautoscaler.yaml

apiVersion: "autoscaling.agones.dev/v1"
kind: FleetAutoscaler
metadata:
  name: minetest-autoscaler
spec:
  fleetName: minetest
  policy:
    type: Buffer
    buffer:
      bufferSize: 5
      minReplicas: 0
      maxReplicas: 20
      
# gameserverallocation.yaml      
apiVersion: "allocation.agones.dev/v1"
kind: GameServerAllocation
spec:
  required:
    matchLabels:
      agones.dev/fleet: minetest