DEV Community

Cover image for Custom CloudWatch Events
馃殌  Vu Dao 馃殌
馃殌 Vu Dao 馃殌

Posted on

Custom CloudWatch Events

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

Use event source with customize an Event Pattern

What鈥檚 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)