DEV Community

loading...
Cover image for Custom CloudWatch Events

Custom CloudWatch Events

Vu Dao
Awesome Devops || AWS SA || CloudOpz
・3 min read

Create rules to invoke Targets based on Events happening in your AWS environment.

Use event source with customize an Event Pattern

What’s In This Document

πŸš€ Create custom cloudwatch event rule

  • Build custom event pattern
{
  "source": [
    "com.test.ssm.to.target"
  ]
}
Enter fullscreen mode Exit fullscreen mode
  • Target: SSM Run Command Alt Text

πŸš€ Create AWS Systems Manager Document

  • JSON Content: Write {{Message}} content to {{workingDirectory}}/testSSM.txt"
{
  "schemaVersion": "2.2",
  "description": "Run SSM command",
  "parameters": {
    "Message": {
      "type": "String",
      "description": "Parameter of SSM script",
      "default": ""
    },
    "workingDirectory": {
      "type": "String",
      "description": "Working dir",
      "default": "/tmp/"
    }
  },
  "mainSteps": [
    {
      "action": "aws:runShellScript",
      "name": "runSSMCommand",
      "inputs": {
        "runCommand": [
          "echo {{Message}} > {{workingDirectory}}/testSSM.txt"
        ]
      }
    }
  ]
}
Enter fullscreen mode Exit fullscreen mode
  • Target type: /AWS::EC2::Instance Alt Text

πŸš€ Update IAM role to run SSM document from cloudwatch

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Action": "ssm:SendCommand",
            "Effect": "Allow",
            "Resource": [
                "arn:aws:ec2:ap-northeast-1:111111111111:instance/i-0f4a1c3c2ca0a7dee",
                "arn:aws:ssm:ap-northeast-1:111111111111:document/testSSM"
            ]
        }
    ]
}
Enter fullscreen mode Exit fullscreen mode

πŸš€ Put cloudwatch event to test

  • Use python script to put event to cloudwatch rule
import boto3
import json
from datetime import datetime


def put_cloudwatch_event():
    try:
        client = boto3.client('events', region_name='ap-northeast-1')
        json_input = {"data": "{0} {1}".format('my-source', 'my-target')}
        response = client.put_events(
            Entries=[
                {
                    'Time': datetime.now(),
                    'Source': 'com.test.ssm.to.target',
                    'DetailType': 'MyDetailType',
                    'Resources': ['resource1', 'resource2'],
                    'Detail': json.dumps(json_input)
                }
            ]
        )
        if response['FailedEntryCount'] == 0:
            print(f"Result {json.dumps(json_input)} is in progress")
    except ValueError as err:
        print(str(err))


put_cloudwatch_event()
Enter fullscreen mode Exit fullscreen mode
  • Run script
Result {"data": "my-source my-target"} is in progress

Process finished with exit code 0
Enter fullscreen mode Exit fullscreen mode
  • Check result: Access to target instance
# cat /tmp/testSSM.txt 
my-source my-target
Enter fullscreen mode Exit fullscreen mode

Mirror:

Read More

🌠 Blog · Web · Linkedin · Group · Page · Twitter 🌠

Discussion (0)