How to make two EC2 instance (installed Elasticsearch created by AMI) as multi-node using cloudformation template?

Try to add a macro to your Cloudformation template. This is an example of lambda that can be invoked by a macro. This function use SSM RunCommand to send bash commands to your instance, at this case the instances are filtered by autoscaling groups, but you can filter your instances by tags or any other attribute. Also you need to add a IAM role to the instance with the AmazonEC2RoleforSSM permission.

import json
import boto3
import time
import os

ssm = boto3.client('ssm')
ec2 = boto3.client('ec2')
autoscalingg = boto3.client('autoscaling')

def lambda_handler(event, context):
    # TODO implement
    env = event['environment']
    kafka = event['kafka']
    print (env, kafka)
    status = check_autoscaling_group(env)
    if status:
        add_IP(env, kafka)
        return {
            'body': json.dumps('Function executed, please see the logs!')
        }
    else:
        return {
            'body': json.dumps('The function was not executed, please see the logs!')
        }

def check_autoscaling_group(env):
    response = autoscalingg.describe_auto_scaling_groups(
        AutoScalingGroupNames=[
            'elasticsearch-master-'+str(env)
        ]
    )
    stable = True
    for instance in response['AutoScalingGroups'][0]['Instances']:
        if instance['LifecycleState'] != "InService":
            stable = False
            message = "At least 1 master instance is not in Running status, please wait until all the masters nodes are stable."
            print (message)
            break
    return stable

def add_IP(env, kafka):
    response = ec2.describe_instances(
        Filters=[
            {
                'Name': 'tag:Name',
                'Values': [
                    'elasticsearch-masters-server-'+str(env)
                ]
            },
            {
                'Name': 'instance-state-name',
                'Values': [
                    'running'
                ]
            },
            {
                'Name': 'tag:role',
                'Values': [
                    'master'
                ]
            }
        ]
    )

    for instances in response['Reservations']:
        id = instances['Instances'][0]['InstanceId']
        ip = instances['Instances'][0]['PrivateIpAddress']
        
        #Add ES server IP to Kibana config
        command_to_execute = 'sed -i "s/ELASTICSEARCH_HOSTS: http:\/\/.*/ELASTICSEARCH_HOSTS: http:\/\/'+str(ip)+':9200/g" /home/ubuntu/Kibana/docker-compose.yml'
        execute_in_master(command_to_execute, id, env)
        
        #Add kafka server IP to logstash conf
        kafka = '\\"'+str(kafka)+':9092\\"'
        command_to_execute = 'sed -i "s/bootstrap_servers => .*/bootstrap_servers => ['+str(kafka)+']/g" /home/ubuntu/Logstash/config/logstash/pipeline/my_pipeline.conf'
        execute_in_master(command_to_execute, id, env)
        
        #Add ES server IP to logstash config
        ip = '\\"'+str(ip)+':9200\\"'
        command_to_execute = 'sed -i "s/hosts => .*/hosts => '+str(ip)+'/g" /home/ubuntu/Logstash/config/logstash/pipeline/my_pipeline.conf'
        execute_in_master(command_to_execute, id, env)
        
        #Restart services
        command_to_execute = 'cd /home/ubuntu/Kibana/ && docker-compose up -d'
        execute_in_master(command_to_execute, id, env)
        
        command_to_execute = 'cd /home/ubuntu/Logstash/ && docker-compose up -d'
        execute_in_master(command_to_execute, id, env)
        
        #Just select one master instance
        break
        
def execute_in_master(command_to_execute, id, env):
    print (command_to_execute)
    response = ec2.describe_instances(
        Filters=[
            {
                'Name': 'tag:Name',
                'Values': [
                    'elasticsearch-masters-server-'+str(env)
                ]
            },
            {
                'Name': 'instance-state-name',
                'Values': [
                    'running'
                ]
            },
            {
                'Name': 'tag:role',
                'Values': [
                    'master'
                ]
            }
        ]
    )
    
    for instances in response['Reservations']:
        instance_id = instances['Instances'][0]['InstanceId']
        if instance_id == id:
            response = ssm.send_command(
                InstanceIds=[instance_id],
                DocumentName="AWS-RunShellScript",
                Parameters={'commands': [command_to_execute]}
            )
            time.sleep(1)   

To call a macro inside the cloudformation template you need to add this:

  ModifyInstances:
    Fn::Transform:
        Name: MacroSetUpCluster
        Parameters:
          env: !Ref MyEnv
          kafka: !Ref MyKafkaIP

The stack MacroSetUpCluster needs to have been previously deployed with the lambda function.