DEV Community: Gurudev Prasad Teketi

Day 4: Talking About Work, Hobbies, and Daily Activities

Gurudev Prasad Teketi — Mon, 08 Jun 2026 13:01:49 +0000

¡Hola!

This class focused on expanding vocabulary around daily activities, work, free time, and practicing some of the most useful sentence patterns in Spanish.

The Four Sentence Patterns We Keep Using

One thing I’m noticing is that Spanish becomes much easier when I master a few sentence structures and reuse them in different situations.

Tengo que… (I have to…)
Used for obligations and responsibilities.

Examples:

Tengo que trabajar.
Tengo que reparar mi carro.
Tengo que asistir a muchas reuniones.
Tengo que ir al parque con mi hijo.

Voy a… (I’m going to…)
Used for future plans.

Examples:

Voy a limpiar mi casa.
Voy a cocinar comida india.
Voy a viajar.
Voy a estudiar español.

Quiero… (I want…)

Used to express desires and goals.

Examples:

Quiero hablar español.
Quiero viajar a México.
Quiero aprender más vocabulario.

Me gusta… (I like…)

One of the most useful expressions for conversations.

Me gusta cocinar.
Me gusta correr.
Me gusta meditar.
Me gusta viajar.

Vocabulary: Activities at Home

We learned several verbs commonly used around the house.

Activities at Home

Trabajar — To work
Limpiar — To clean
Leer — To read
Hacer — To do / make
Comer — To eat
Dormir — To sleep
Ver — To watch
Hablar — To speak
Cocinar — To cook

Activities at Work

This section felt especially relevant because many of these activities describe a typical workday.

Asistir a reuniones — To attend meetings
Asignar trabajos — To assign tasks
Dibujar diagramas arquitectónicos — To draw architectural diagrams
Planear proyectos — To plan projects
Solucionar problemas — To solve problems
Documentar proyectos — To document projects
Hablar con mi jefe — To talk with my boss
Discutir con mi equipo — To discuss with my team

Free Time Activities

We also covered hobbies and leisure activities.

Jugar — To play
Viajar — To travel
Pasar tiempo — To spend time
Ir de compras — To go shopping
Correr — To run
Meditar — To meditate
Estudiar — To study
Tocar la guitarra — To play the guitar
Ver deportes — To watch sports

A Useful Grammar Observation

Spanish verbs are grouped into three major infinitive endings:

-ar
-er
-ir

Examples:

Trabajar
Comer
Dormir

This may seem like a small detail now, but it becomes very important once verb conjugations start getting more advanced.

Core Expressions

Me gusta... — I like...
Tengo que... — I have to...
Voy a... — I am going to...
Quiero... — I want...
¿Te gusta...? — Do you like...?
¿Qué te gusta? — What do you like?
Sí, me gusta. — Yes, I like it.
No, no me gusta. — No, I don't like it.
Dime... — Tell me...
¿Cómo se dice...? — How do you say...?

Common Nouns

La reunión — The meeting
Las reuniones — The meetings
El proyecto — The project
Los proyectos — The projects
El jefe — The boss
El equipo — The team
La casa — The house
El trabajo — The work / job
El problema — The problem
Los problemas — The problems
El día — The day
La clase de Español — The Spanish class

Verb Endings (Infinitives)

Spanish verbs are grouped into three categories based on their endings:

-AR
- Trabajar — To work
- Limpiar — To clean
- Hablar — To speak
- Cocinar — To cook
- Viajar — To travel
-ER
- Comer — To eat
- Beber — To drink
- Leer — To read
- Correr — To run
-IR
- Dormir — To sleep
- Asistir — To attend
- Discutir — To discuss

Day 3: Talking About Days, Months, and Daily Expressions

Gurudev Prasad Teketi — Mon, 08 Jun 2026 02:06:29 +0000

Hola, ¿cómo estás?

Today we focused on daily expressions, days of the week, months of the year, and Spanish articles.

📅 Days of the Week

One interesting thing I learned is that days of the week are not capitalized in Spanish.

Spanish   English

lunes     Monday

martes    Tuesday

miércoles Wednesday

jueves    Thursday

viernes   Friday

sábado    Saturday

domingo   Sunday

Examples:

Hoy es miércoles.
(Today is Wednesday.)

Mañana es jueves.
(Tomorrow is Thursday.)

Ayer fue martes.
(Yesterday was Tuesday.)

I also learned how to talk about recurring activities:

Tengo clase de Español los lunes y los miércoles.
(I have Spanish class on Mondays and Wednesdays.)

No trabajo los domingos.
(I don't work on Sundays.)

📆 Months of the Year

Just like the days of the week, months are written in lowercase.

Spanish     English

enero       January

febrero     February

marzo       March

abril       April

mayo        May

junio       June

julio       July

agosto      August

septiembre  September

octubre     October

noviembre   November

diciembre   December

Examples:

Tengo mi clase de Español el 8 de junio.
(I have my Spanish class on June 8.)

Mi cumpleaños es el 19 de agosto.
(My birthday is August 19.)

Everyday Expressions

These four expressions came up repeatedly throughout the lesson:

Me gusta…

Used to say what you like.

Me gusta viajar con mi familia.
(I like traveling with my family.)

Me gusta beber té con mi esposa.
(I like drinking tea with my wife.)

Voy a…

Used for future plans (“I’m going to…”).

Voy a practicar español.
(I am going to practice Spanish.)

Voy a leer un libro.
(I am going to read a book.)

Tengo que…

Used when you have to do something.

Tengo que trabajar hoy.
(I have to work today.)

Tengo que estudiar Español.
(I have to study Spanish.)

Quiero…

Used to express what you want.

Quiero viajar a Japón.
(I want to travel to Japan.)

Quiero hablar en español.
(I want to speak in Spanish.)

📖 Learning About Articles

Spanish articles change depending on gender and number.

Definite Articles (“The”)

Singular               Plural

El                      Los

La                      Las

Examples:

El café
Los cafés

La casa
Las casas

Indefinite Articles (“A” / “An”)

Singular

Un

Una

Examples:

Un día
Una clase de Español

Dia 2 : Tu Identidad (Your Identity)

Gurudev Prasad Teketi — Mon, 08 Jun 2026 01:54:21 +0000

## Vocaubulario

Convivir = to live together / to coexist / to get along with others
demás = others / the rest
Saludar = to greet / to say hello
y = and
despedirse = to say goodbye / to take leave
presentarse = to introduce oneself
otras = other
personas = people/persons
llamarse = to be called / to name oneself
identidad = identity
edad = age
pais = country
busca = look for / search for 
deporte = sport
equipo = team
nombre = first name
apellido = last name/surname
siguientes = following
Mire = this

¿Cómo se llama tu profesor o profesora?
(What is your teacher’s name?)

¿Cómo se escribe su nombre y apellido?
(How do you spell their first and last name?)

¿Cómo te llamas tú?
(What is your name?)

Mire, este es mi novio, Lucas.
(Look, this is my boyfriend, Lucas.)

En México se habla español.
(Spanish is spoken in Mexico.)

Frida Kahlo es una pintora mexicana conocida en todo el mundo.
(Frida Kahlo is a Mexican painter known all over the world.)

Los mejores tenistas de la historia:
(he best tennis players in history)
Rafa Nadal es español,
(Rafa Nadal is Spanish)
Roger Federer es suizo
(Roger Federer is Swiss.)
y Novak Djokovic es serbio.
(Novak Djokovic is Serbian.)

Nationality Adjective Patterns

Masculine Singular Feminine Singular Masculine Plural Feminine Plural

italiano          italiana           italianos          italianas

japonés           japonesa           japoneses          japonesas

canadiense        canadiense         canadienses        canadienses

marroquí          marroquí           marroquíes          marroquíes

Dia 1 : Las palabras importantes, verbos en presente

Gurudev Prasad Teketi — Mon, 01 Jun 2026 02:41:06 +0000

Hola.

¿Cómo estás?

I’ve been practicing Spanish on Duolingo for more than 350 days and currently have a 340-day streak. But even with all that consistency, I realized I still lacked confidence when speaking with real people.

So I decided to take the next step and joined an online Spanish course with live instructors to practice real conversations and improve my fluency.

Here’s what I learned on my very first day.

Las palabras importantes de la clase

(The important words of the class)

- ¿Qué significa ...? = what does mean?
- Significa... = it means
- ¿Cómo se dice...? = how do you say ...
- Se dice... = we say
- Tengo una pregunta = I have a question
- ¿Tienes preguntas? = You have a question?
- No tengo preguntas = I do not have questions
- Dime = tell me
- ¿Entiendes? = Understood?
- Entiendo =  I understand
- No entiendo = No I do not understand
- Repite, por favor = Repeat, Please 
- ¿Puedes repetir? = Can you repeat
- Vamos a... = we are going to / lets...

## Vocaubulario

- Tiempo = time
- Hace = ago
- Empleado/a = employees
- De = of

## Verbos

- Hablar = to talk
- Ayudar = to help
- Enseñar = to teach
- Entender = to understand
- Ver = to see

## Presentacion

- Yo soy Gurudev
  (I am Gurudev)

- Me dicen Guru
  (They call me Guru)

- Soy de India
  (I am from India)

- Soy ingeniero de Software
  (I am a software Engineer)

- Vivo en Austin, Tx
  (I live in Texas)

- Me gusta jugar Badminton y pasar tiempo con mi familia
  (I like to play badminton and spend time with my family)

- ¿Cuánto tiempo hace que vives en Austin?
  (How long you live in Austin)

- Hace 5 años que vivo en Austin
  (I have been living in Austin for 5 years)

- ¿Por qué quieres estudiar español?
  (Why do you want to learn spanish)

- Porque tengo empleados en mi casa que hablan español
  (because I have employees in my house who speaks spanish)

- Para hablar con mis empleados
  (to talk to my employees)

- Para ayudar a otros ingenieros de software
  (To help other software engineers)

- Para enseñar
  (To teach)

- ¿Cuál o cuáles son tus lenguas madre?
  (what is your mother tongue(language))

- Mi lengua madre es Telugu
  (My mother tounge is Telugu)

- ¿El inglés es tu segunda lengua?
   (Is English your second language?)

- No, el hindi es mi segunda lengua y el inglés es mi tercera
  (No Hindi is my second language and Englihs is my third)

- ¿Hablas otras lenguas?
  (How many languages do you speak)
- Sí. (yes)

- ¿qué significa "enseñar"?
  (What do you mean by "enseñar)

- "Enseñar" significa "to teach"
   ("Enseñar" means "to teach")

- '¿Cómo se dice "house" en español?
   (How do you say "house" in spanish)

- se dice casa
  (we say "casa")

AWS Cost Reporting: Why Parquet Data Doesn't Match Your Console (And How to Fix It)

Gurudev Prasad Teketi — Mon, 13 Oct 2025 20:46:18 +0000

1. The Problem

Real production issue: Lambda showing 10x higher costs than console
User expectation vs reality ($1,262 console vs $12,918 Lambda)

2. Initial Investigation

Athena/parquet approach seemed logical
Adjustment factors didn't solve service-level discrepancies
Individual service ratios were inconsistent (4.0x to 5.7x)

3. Research & Discovery

AWS documentation deep-dive
Key differences between Cost Explorer API vs Cost and Usage Reports
Data freshness, calculation methods, service groupings

4. The Solution

Cost Explorer API implementation
Exact console matching with proper error handling
99.999% accuracy achieved

5. Lessons Learned

When to use CUR vs Cost Explorer API
AWS billing reconciliation best practices
Real-world cost considerations ($0.30/month vs engineering time)

Technical Value:

Practical code examples in Python/boto3
AWS architecture decisions with reasoning
Performance optimization and cost trade-offs
Enterprise-grade error handling

This would be valuable for:

DevOps/FinOps teams building cost monitoring
AWS architects choosing between cost data sources
Developers debugging billing discrepancies

Key Features

99.999% console accuracy using Cost Explorer API
Enterprise error handling with proper logging
Cost-effective: ~$0.30/month for daily reports
No adjustment factors needed
Production-ready with comprehensive validation

import json
  import boto3
  import logging
  import os
  from datetime import datetime, timedelta
  from typing import Dict, List
  import urllib3

  logger = logging.getLogger()
  logger.setLevel(os.getenv('LOG_LEVEL', 'INFO'))

  def lambda_handler(event, context):

      try:
          logger.info("Starting daily cost usage report processing with Cost Explorer API")

          # Initialize Cost Explorer client (must be us-east-1)
          ce_client = boto3.client('ce', region_name='us-east-1')

          # Get environment variables
          slack_webhook_url = os.getenv('SLACK_WEBHOOK_URL')

          if not slack_webhook_url:
              raise ValueError("SLACK_WEBHOOK_URL environment variable is required")

          # Get current month date range for Cost Explorer API
          today = datetime.now()
          month_start = datetime(today.year, today.month, 1)

          # Format dates for Cost Explorer API (YYYY-MM-DD format)
          start_date = month_start.strftime('%Y-%m-%d')
          end_date = (today + timedelta(days=1)).strftime('%Y-%m-%d')  # API end date is exclusive

          logger.info(f"Processing costs from {start_date} to {end_date} using Cost Explorer API")

          # Get cost data using Cost Explorer API
          cost_data = get_cost_data_from_cost_explorer(
              ce_client,
              start_date,
              end_date
          )

          # Format and send Slack message
          slack_message = format_slack_message(cost_data, month_start, today)
          send_slack_message(slack_webhook_url, slack_message)

          logger.info("Successfully completed daily cost usage report")

          return {
              'statusCode': 200,
              'body': json.dumps({
                  'message': 'Cost report sent successfully',
                  'total_cost': cost_data['total_cost'],
                  'services_count': len(cost_data['services'])
              })
          }

      except Exception as e:
          logger.error(f"Error processing cost report: {str(e)}")
          return {
              'statusCode': 500,
              'body': json.dumps({
                  'error': str(e)
              })
          }
  def get_cost_data_from_cost_explorer(ce_client, start_date: str, end_date: str) -> Dict:
      logger.info(f"Querying Cost Explorer API from {start_date} to {end_date}")

      try:
          # Cost Explorer API call - matches exactly what AWS console uses
          response = ce_client.get_cost_and_usage(
              TimePeriod={
                  'Start': start_date,
                  'End': end_date
              },
              Granularity='MONTHLY',
              Metrics=['UnblendedCost'],
              GroupBy=[
                  {
                      'Type': 'DIMENSION',
                      'Key': 'SERVICE'
                  }
              ]
          )

          # Parse Cost Explorer API response
          services = []
          total_cost = 0.0

          # Validate response structure
          if not response.get('ResultsByTime') or len(response['ResultsByTime']) == 0:
              logger.warning("No cost data in Cost Explorer API response")
              return {'services': [], 'total_cost': 0.0, 'currency': 'USD'}

          # Get the first (and only) time period result
          result = response['ResultsByTime'][0]

          # Process each service group
          for group in result.get('Groups', []):
              try:
                  service_name = group['Keys'][0] if group.get('Keys') else 'Unknown'

                  # Get unblended cost amount
                  cost_data = group.get('Metrics', {}).get('UnblendedCost', {})
                  cost_amount_str = cost_data.get('Amount', '0')
                  currency = cost_data.get('Unit', 'USD')

                  # Safely parse cost amount
                  try:
                      cost_amount = float(cost_amount_str)
                  except (ValueError, TypeError):
                      logger.warning(f"Invalid cost amount for service {service_name}: {cost_amount_str}")
                      continue

                  if cost_amount > 0.01:  # Only include services with meaningful cost
                      services.append({
                          'name': service_name,
                          'cost': round(cost_amount, 2),
                          'currency': currency
                      })
                      total_cost += cost_amount

              except Exception as service_error:
                  logger.warning(f"Error processing service group: {str(service_error)}")
                  continue

          # Sort by cost (highest first)
          services.sort(key=lambda x: x['cost'], reverse=True)

          logger.info(f"Retrieved {len(services)} services with total cost: ${total_cost:.2f}")

          return {
              'services': services,
              'total_cost': round(total_cost, 2),
              'currency': 'USD'
          }

      except Exception as e:
          logger.error(f"Error calling Cost Explorer API: {str(e)}")
          raise Exception(f"Cost Explorer API call failed: {str(e)}")

  def format_slack_message(cost_data: Dict, start_date: datetime, end_date: datetime) -> Dict:

      total_cost = cost_data['total_cost']
      services = cost_data['services']
      currency = cost_data.get('currency', 'USD')

      # Header
      date_range = f"{start_date.strftime('%b %d')} - {end_date.strftime('%b %d, %Y')}"
      header = f"💰 *AWS Cost Report: {date_range}*\n"
      header += f"💳 *Total Cost: ${total_cost:,.2f} {currency}*\n\n"

      # Services table
      if services:
          services_text = "```

\n"
          services_text += f"{'AWS Service':<30} {'Cost':<12}\n"
          services_text += f"{'-' * 30} {'-' * 12}\n"

          # Show top 15 services
          for service in services[:15]:
              cost = service['cost']
              name = service['name']

              # Format service name for readability
              display_name = name.replace('Amazon', '').replace('AWS', '').strip()
              if not display_name:
                  display_name = name

              # Truncate long service names
              if len(display_name) > 29:
                  display_name = display_name[:26] + "..."

              services_text += f"{display_name:<30} ${cost:>11,.2f}\n"

          services_text += f"{'-' * 30} {'-' * 12}\n"
          services_text += f"{'TOTAL':<30} ${total_cost:>11,.2f}\n"
          services_text += "

```"

          if len(services) > 15:
              remaining = len(services) - 15
              services_text += f"\n_+ {remaining} more services with smaller costs_"
      else:
          services_text = "No cost data available for this period."

      message_text = header + services_text

      return {
          "text": f"AWS Daily Cost Report - ${total_cost:,.2f}",
          "blocks": [
              {
                  "type": "section",
                  "text": {
                      "type": "mrkdwn",
                      "text": message_text
                  }
              }
          ]
      }

  def send_slack_message(webhook_url: str, message: Dict):

      http = urllib3.PoolManager()

      logger.info("Sending message to Slack channel")

      response = http.request(
          'POST',
          webhook_url,
          body=json.dumps(message),
          headers={'Content-Type': 'application/json'}
      )

      if response.status == 200:
          logger.info("Successfully sent message to Slack")
      else:
          logger.error(f"Failed to send Slack message: {response.status}")
          raise Exception(f"Slack webhook failed with status {response.status}")

  Required IAM Policy:

  {
      "Version": "2012-10-17",
      "Statement": [
          {
              "Effect": "Allow",
              "Action": [
                  "ce:GetCostAndUsage",
                  "ce:GetUsageReport"
              ],
              "Resource": "*"
          }
      ]
  }

Environment Variables:

SLACK_WEBHOOK_URL: Your Slack webhook URL
LOG_LEVEL: INFO (optional)`

Building a True Dual-Destination Analytics Pipeline: Real-Time Streaming with S3 Backup and Recovery

Gurudev Prasad Teketi — Fri, 10 Oct 2025 19:26:36 +0000

Executive Summary

This article details the implementation of a sophisticated dual-destination analytics pipeline that combines real-time data streaming to AWS Redshift Serverless with robust S3 backup and recovery capabilities. The architecture provides the best of both worlds: immediate analytics insights with comprehensive data protection and quality assurance mechanisms.

Project Objective

Primary Goal: Design and implement a resilient dual-destination data pipeline that delivers real-time analytics capabilities while maintaining comprehensive data backup, recovery, and quality assurance processes.

Key Requirements:

Real-time analytics with sub-5-minute data availability
Comprehensive data backup in S3 for recovery and compliance
Zero data loss guarantee during failures
Data quality assurance with validation and reconciliation
Cost-effective architecture balancing performance and expenses
Scalable solution supporting growing data volumes
Operational simplicity with automated monitoring and recovery

Architecture Overview

True Dual-Destination Pattern

📱 Reading App
    ↓
🔄 Enhanced Lambda Function
    ↓                    ↓
🔥 Redshift Firehose   🔥 S3 Firehose
    ↓                    ↓
🏭 Redshift (Real-time) 📦 S3 (Backup)
    ↑                    ↓
    ←── ETL Reconciliation ──

Core Principle: Write Twice, Query Once

Every data record flows to both destinations simultaneously
Real-time analytics from Redshift
Backup and recovery from S3
Periodic reconciliation ensures data consistency

Enhanced Lambda Function Design

Dual-Destination Lambda Architecture

import json
import boto3
from datetime import datetime, timezone

firehose_client = boto3.client("firehose")

# Dual-destination configuration
REDSHIFT_FIREHOSE = "analytics-realtime-delivery-dev"
S3_FIREHOSE_BOOK_ACTIVITY = "analytics-backup-s3-dev-book-activity"
S3_FIREHOSE_DYNAMIC = "analytics-backup-s3-dev-dynamic"

def lambda_handler(event, context):
    if "body" not in event:
        return {"statusCode": 400, "body": json.dumps({"error": "No request body provided"})}

    try:
        data = json.loads(event["body"])
    except json.JSONDecodeError:
        return {"statusCode": 400, "body": json.dumps({"error": "Invalid JSON format"})}

    records = [data] if isinstance(data, dict) else data if isinstance(data, list) else None
    if records is None:
        return {"statusCode": 400, "body": json.dumps({"error": "Invalid data format"})}

    # Prepare dual-destination records
    s3_records = {"<S3 bucket>": [], "dynamic": []}
    redshift_records = []

    for item in records:
        # Validation
        if "recordId" not in item or "ledgerCategory" not in item:
            return {"statusCode": 400, "body": json.dumps({"error": "Missing required fields"})}

        # Add server timestamp
        utc_now = datetime.now(timezone.utc)
        item["serverTimestamp"] = utc_now.strftime("%Y-%m-%dT%H:%M:%SZ")
        item["pipelineTimestamp"] = utc_now.strftime("%Y-%m-%dT%H:%M:%SZ")

        # S3 backup records (original format)
        s3_target = "<S3 Bucket>" if item["Category"] == "<S3 Bucket>" else "dynamic"
        s3_record = {"Data": (json.dumps(item) + "\n").encode("utf-8")}
        s3_records[s3_target].append(s3_record)

        # Redshift real-time records (optimized format)
        redshift_record = {
            "Data": (json.dumps({
                "recordId": item["recordId"],
                "ledgerCategory": item["ledgerCategory"],
                "userId": item.get("userId"),
                "eventData": json.dumps(item),
                "processedAt": utc_now.strftime("%Y-%m-%dT%H:%M:%SZ")
            }) + "\n").encode("utf-8")
        }
        redshift_records.append(redshift_record)

    try:
        # Dual-destination delivery
        delivery_results = {}

        # 1. Send to S3 Firehose (backup)
        for target, firehose_name in [
            ("<S3 Bucket>", S3_FIREHOSE_BOOK_ACTIVITY),
            ("dynamic", S3_FIREHOSE_DYNAMIC)
        ]:
            if s3_records[target]:
                s3_response = firehose_client.put_record_batch(
                    DeliveryStreamName=firehose_name,
                    Records=s3_records[target]
                )
                delivery_results[f"s3_{target}"] = {
                    "delivered": len(s3_records[target]) - s3_response["FailedPutCount"],
                    "failed": s3_response["FailedPutCount"]
                }

        # 2. Send to Redshift Firehose (real-time)
        if redshift_records:
            redshift_response = firehose_client.put_record_batch(
                DeliveryStreamName=REDSHIFT_FIREHOSE,
                Records=redshift_records
            )
            delivery_results["redshift_realtime"] = {
                "delivered": len(redshift_records) - redshift_response["FailedPutCount"],
                "failed": redshift_response["FailedPutCount"]
            }

        # Check for failures
        total_failed = sum(result["failed"] for result in delivery_results.values())
        if total_failed > 0:
            return {"statusCode": 500, "body": json.dumps({
                "error": "Partial delivery failure",
                "details": delivery_results
            })}

        return {"statusCode": 200, "body": json.dumps({
            "status": "success",
            "delivery_results": delivery_results,
            "dual_destination_enabled": True
        })}

    except Exception as e:
        return {"statusCode": 500, "body": json.dumps({
            "error": "Dual-destination delivery failed",
            "message": str(e)[:200]
        })}

Redshift Direct Delivery Configuration

Real-Time Firehose to Redshift Setup

{
    "DeliveryStreamName": "analytics-realtime-delivery-dev",
    "RedshiftDestinationConfiguration": {
        "RoleARN": "arn:aws:iam::123456789012:role/AnalyticsFirehoseRedshiftRole",
        "ClusterJDBCURL": "jdbc:redshift://analytics-cluster.us-east-2.redshift.amazonaws.com:5439/analytics",
        "CopyCommand": {
            "DataTableName": "events.realtime_staging",
            "DataTableColumns": "record_id,ledger_category,user_id,event_data,processed_at",
            "CopyOptions": "FORMAT AS JSON 'auto' TIMEFORMAT 'auto' TRUNCATECOLUMNS ACCEPTINVCHARS"
        },
        "Username": "firehose_user",
        "Password": "SecurePassword123",
        "S3Configuration": {
            "RoleARN": "arn:aws:iam::123456789012:role/AnalyticsFirehoseS3Role",
            "BucketARN": "arn:aws:s3:::analytics-backup-dev",
            "Prefix": "redshift-backup/year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/hour=!{timestamp:HH}/",
            "BufferingHints": {
                "SizeInMBs": 5,
                "IntervalInSeconds": 60
            },
            "CompressionFormat": "GZIP"
        },
        "ProcessingConfiguration": {
            "Enabled": true,
            "Processors": [
                {
                    "Type": "Lambda",
                    "Parameters": [
                        {
                            "ParameterName": "LambdaArn",
                            "ParameterValue": "arn:aws:lambda:us-east-2:123456789012:function:analytics-data-transformer"
                        }
                    ]
                }
            ]
        },
        "RetryOptions": {
            "DurationInSeconds": 3600
        }
    }
}

Real-Time Staging Table Design

-- Real-time staging table for immediate analytics
CREATE TABLE events.realtime_staging (
    record_id VARCHAR(256) NOT NULL,
    ledger_category VARCHAR(128) NOT NULL,
    user_id INTEGER,
    event_data VARCHAR(65535),
    processed_at TIMESTAMP NOT NULL,
    load_timestamp TIMESTAMP DEFAULT GETDATE()
)
DISTKEY(user_id)
SORTKEY(processed_at, ledger_category);

-- Production analytics views
CREATE VIEW events.realtime_activity AS
SELECT
    record_id,
    user_id,
    JSON_EXTRACT_PATH_TEXT(event_data, 'bookID') as book_id,
    JSON_EXTRACT_PATH_TEXT(event_data, 'eventType') as event_type,
    JSON_EXTRACT_PATH_TEXT(event_data, 'pageNumber') as page_number,
    processed_at,
    load_timestamp
FROM events.realtime_staging
WHERE ledger_category = 'book_activity'
AND processed_at >= DATEADD(hour, -24, GETDATE());

Monitoring and Alerting

CloudWatch Dashboard Configuration**

{
    "widgets": [
        {
            "type": "metric",
            "properties": {
                "metrics": [
                    ["AWS/Kinesis/Firehose", "DeliveryToRedshift.Records", "DeliveryStreamName", "analytics-realtime-delivery-dev"],
                    ["AWS/Kinesis/Firehose", "DeliveryToS3.Records", "DeliveryStreamName", "analytics-backup-s3-dev-book-activity"],
                    ["AWS/Lambda", "Invocations", "FunctionName", "analytics-dual-destination-processor"],
                    ["AWS/Lambda", "Errors", "FunctionName", "analytics-reconciliation-processor"]
                ],
                "period": 300,
                "stat": "Sum",
                "region": "us-east-2",
                "title": "Dual-Destination Pipeline Health"
            }
        },
        {
            "type": "log",
            "properties": {
                "query": "SOURCE '/aws/lambda/analytics-dual-destination-processor'\n| fields @timestamp, @message\n| filter @message like /ERROR/\n| sort @timestamp desc\n| limit 100",
                "region": "us-east-2",
                "title": "Recent Pipeline Errors"
            }
        }
    ]
}

Automated Alerting

# CloudWatch alarm for dual-destination failures
aws cloudwatch put-metric-alarm \
    --alarm-name "Dual-Destination-Pipeline-Failures" \
    --alarm-description "Alert on dual-destination delivery failures" \
    --metric-name "Errors" \
    --namespace "AWS/Lambda" \
    --statistic "Sum" \
    --period 300 \
    --threshold 5 \
    --comparison-operator "GreaterThanThreshold" \
    --dimensions Name=FunctionName,Value=analytics-dual-destination-processor \
    --alarm-actions "arn:aws:sns:us-east-2:123456789012:analytics-alerts"

# Data discrepancy alarm
aws cloudwatch put-metric-alarm \
    --alarm-name "Data-Reconciliation-Discrepancy" \
    --alarm-description "Alert on high data discrepancy between S3 and Redshift" \
    --metric-name "DiscrepancyPercentage" \
    --namespace "Analytics/Reconciliation" \
    --statistic "Average" \
    --period 3600 \
    --threshold 10 \
    --comparison-operator "GreaterThanThreshold"

Deployment Architecture

Infrastructure as Code (CloudFormation)

AWSTemplateFormatVersion: '2010-09-09'
Description: 'Dual-Destination Analytics Pipeline'

Parameters:
  Environment:
    Type: String
    Default: dev
    AllowedValues: [dev, qa, prod]

Resources:
  # Redshift Serverless Namespace
  RedshiftNamespace:
    Type: AWS::RedshiftServerless::Namespace
    Properties:
      NamespaceName: !Sub 'analytics-ledger-${Environment}'
      AdminUsername: admin
      AdminUserPassword: !Ref RedshiftPassword
      DbName: analytics

  # Redshift Serverless Workgroup
  RedshiftWorkgroup:
    Type: AWS::RedshiftServerless::Workgroup
    Properties:
      WorkgroupName: !Sub 'analytics-datalake-${Environment}'
      NamespaceName: !Ref RedshiftNamespace
      PubliclyAccessible: true
      BaseCapacity: 8

  # Dual-Destination Lambda
  DualDestinationLambda:
    Type: AWS::Lambda::Function
    Properties:
      FunctionName: !Sub 'analytics-dual-destination-${Environment}'
      Runtime: python3.9
      Handler: dual_destination.lambda_handler
      Role: !GetAtt LambdaExecutionRole.Arn
      Code:
        ZipFile: !Sub |
          # Lambda function code here
      Environment:
        Variables:
          REDSHIFT_FIREHOSE: !Ref RedshiftFirehose
          S3_FIREHOSE_BOOK: !Ref S3FirehoseBook
          S3_FIREHOSE_DYNAMIC: !Ref S3FirehoseDynamic

  # Redshift Firehose Stream
  RedshiftFirehose:
    Type: AWS::KinesisFirehose::DeliveryStream
    Properties:
      DeliveryStreamName: !Sub 'analytics-realtime-delivery-${Environment}'
      DeliveryStreamType: DirectPut
      RedshiftDestinationConfiguration:
        RoleARN: !GetAtt FirehoseRole.Arn
        ClusterJDBCURL: !Sub
          - 'jdbc:redshift://${RedshiftEndpoint}:5439/analytics'
          - RedshiftEndpoint: !GetAtt RedshiftWorkgroup.Workgroup.Endpoint.Address
        CopyCommand:
          DataTableName: events.realtime_staging
          CopyOptions: "FORMAT AS JSON 'auto' TIMEFORMAT 'auto'"
        Username: admin
        Password: !Ref RedshiftPassword

  # EventBridge Rule for Reconciliation
  ReconciliationSchedule:
    Type: AWS::Events::Rule
    Properties:
      Name: !Sub 'analytics-reconciliation-${Environment}'
      Description: 'Daily reconciliation between S3 and Redshift'
      ScheduleExpression: 'cron(0 6 * * ? *)'  # Daily at 6 AM
      State: ENABLED
      Targets:
        - Arn: !GetAtt ReconciliationLambda.Arn
          Id: ReconciliationTarget

Outputs:
  RedshiftEndpoint:
    Description: 'Redshift Serverless Endpoint'
    Value: !GetAtt RedshiftWorkgroup.Workgroup.Endpoint.Address
    Export:
      Name: !Sub '${AWS::StackName}-RedshiftEndpoint'

Performance Optimization

Redshift Table Design Best Practices

-- Optimized real-time staging table
CREATE TABLE events.realtime_staging (
    record_id VARCHAR(256) NOT NULL,
    ledger_category VARCHAR(128) NOT NULL,
    user_id INTEGER,
    event_data VARCHAR(65535),
    processed_at TIMESTAMP NOT NULL,
    load_timestamp TIMESTAMP DEFAULT GETDATE(),
    partition_date DATE GENERATED ALWAYS AS (DATE(processed_at))
)
DISTKEY(user_id)
SORTKEY(partition_date, processed_at, ledger_category);

-- Automatic table maintenance
CREATE EVENT analytics_table_maintenance
ON SCHEDULE 'cron(0 2 * * ? *)'  -- Daily at 2 AM
AS $$
  -- Vacuum and analyze tables
  VACUUM events.realtime_staging;
  ANALYZE events.realtime_staging;

  -- Drop old partitions (older than 90 days)
  DELETE FROM events.realtime_staging
  WHERE partition_date < DATEADD(day, -90, GETDATE());
$$;

Firehose Optimization Settings**

{
    "BufferingHints": {
        "SizeInMBs": 128,
        "IntervalInSeconds": 60
    },
    "CompressionFormat": "GZIP",
    "DataTransformation": {
        "ProcessorType": "Lambda",
        "Parameters": {
            "LambdaArn": "arn:aws:lambda:us-east-2:123456789012:function:analytics-data-optimizer"
        }
    },
    "DynamicPartitioning": {
        "Enabled": true,
        "RetryOptions": {
            "DurationInSeconds": 3600
        }
    }
}

Cost Analysis and Optimization**

Cost Breakdown

Component	Monthly Cost (Dev)	Monthly Cost (Prod)
Redshift Serverless	$45-75	$200-400
Kinesis Firehose	$15-25	$50-100
Lambda Executions	$5-10	$20-40
S3 Storage	$10-20	$50-100
Data Transfer	$5-15	$25-50
Total	$80-145	$345-690

Cost Optimization Strategies

Redshift Serverless Scaling:

   -- Set appropriate base capacity
   ALTER WORKGROUP analytics-datalake-dev
   SET base_capacity = 8;  -- Start small, auto-scale up

S3 Lifecycle Policies:

   {
       "Rules": [
           {
               "Status": "Enabled",
               "Transitions": [
                   {
                       "Days": 30,
                       "StorageClass": "STANDARD_IA"
                   },
                   {
                       "Days": 90,
                       "StorageClass": "GLACIER"
                   }
               ]
           }
       ]
   }

Firehose Buffer Optimization:
- Increase buffer size to reduce frequency
- Use compression to reduce storage costs
- Optimize batch sizes for Redshift COPY operations

Advantages of Dual-Destination Architecture

✅ Real-Time Analytics with Safety Net

Sub-5-minute latency for business-critical dashboards
Complete data backup in S3 for recovery scenarios
Zero data loss guarantee through dual writes
Automatic reconciliation ensures data consistency

✅ Operational Resilience

Fault tolerance: Failure in one stream doesn't affect the other
Recovery capabilities: Can rebuild Redshift from S3 if needed
Disaster recovery: Geographic backup through S3 replication
Version control: S3 maintains historical data versions

✅ Performance Benefits

Optimized queries: Purpose-built Redshift tables for analytics
Concurrent workloads: S3 and Redshift serve different use cases
Reduced load: Redshift not used for data lake operations
Scalable architecture: Each component scales independently

✅ Business Value

Immediate insights: Real-time user behavior analysis
Historical analysis: Deep-dive analytics using S3 data
Compliance ready: Audit trails and data lineage tracking
Future-proof: Ready for ML and advanced analytics

Considerations and Trade-offs

❌ Increased Complexity

More components: Additional Firehose streams and monitoring
Dual maintenance: Both S3 and Redshift schemas need updates
Error handling: More failure scenarios to handle
Cost overhead: ~30-50% increase over single-destination

❌ Operational Overhead

Monitoring complexity: Track multiple data flows
Reconciliation requirements: Daily consistency checks needed
Schema coordination: Changes must be synchronized
Debugging challenges: Multiple data paths to troubleshoot

❌ Resource Utilization

Compute costs: Additional Lambda executions
Storage duplication: Data exists in both S3 and Redshift
Network bandwidth: Higher data transfer volumes
Always-on costs: Redshift Serverless base capacity charges

When to Choose Dual-Destination

Ideal Use Cases

Mission-critical analytics requiring real-time insights
High-value data where loss is unacceptable
Compliance requirements for data retention and auditability
Mixed workloads requiring both real-time and batch processing
Growing organizations planning to scale analytics capabilities

Not Recommended When

Simple analytics needs satisfied by batch processing
Cost-sensitive environments where latency is acceptable
Small data volumes that don't justify complexity
Limited operational resources for monitoring multiple systems

Future Enhancements and Roadmap

Phase 2: Advanced Analytics

Machine Learning Integration: Real-time ML inference on streaming data
Complex Event Processing: Multi-stream correlation and pattern detection
Predictive Analytics: User behavior prediction models
Anomaly Detection: Real-time fraud and abuse detection

Phase 3: Multi-Region Architecture

Global data replication for disaster recovery
Regional analytics clusters for performance optimization
Cross-region reconciliation for data consistency
Compliance-specific regions for data sovereignty

Phase 4: Advanced Data Governance

Data lineage tracking across all pipeline components
Automated data quality monitoring and alerting
PII detection and masking for privacy compliance
Automated schema evolution and compatibility checking

Conclusion

The dual-destination analytics pipeline represents a sophisticated approach to modern data architecture, balancing the immediate needs for real-time analytics with the long-term requirements for data durability, compliance, and recovery. This architecture pattern is particularly valuable for organizations that:

Cannot afford data loss in their analytics pipeline
Need both real-time and historical analytics capabilities
Plan to scale their data and analytics operations significantly
Value operational resilience over architectural simplicity

Key Success Factors:

Comprehensive monitoring across all pipeline components
Automated reconciliation to ensure data consistency
Proper cost management to justify the additional complexity
Clear operational procedures for failure scenarios and recovery

Expected Outcomes:

99.9% data availability with sub-5-minute analytics latency
Zero data loss through redundant storage and processing
Scalable foundation supporting future analytics and ML workloads
Operational confidence through comprehensive backup and recovery

This architecture serves as a robust foundation for organizations transitioning from batch-oriented data lakes to real-time analytics platforms while maintaining the safety and compliance benefits of traditional data lake approaches.

Automating Publisher Reporting on AWS: A Serverless Architecture with Slack Alerts

Gurudev Prasad Teketi — Wed, 03 Sep 2025 03:27:51 +0000

Overview

In this article, I walk you through building an automated reporting pipeline using AWS services. The goal was to generate daily summary reports on publisher readership, detect missing metadata (like publishers), store the results as CSVs in S3, and deliver structured Slack notifications to internal stakeholders — all without manual intervention.

Architecture Diagram

Workflow:
1.Amazon EventBridge (Scheduler) triggers the workflow daily.

2.Lambda Function #1: Report Generator

Runs a named query on Amazon Athena that calculates reading activity data.
Stores results in an S3 bucket using a structured naming convention.
Sends a Slack message once the report is ready.

3.Lambda Function #2: Publisher Summary Aggregator

Fetches the latest CSV report from S3.
Aggregates per-publisher read counts by quarter.
Posts a clean, readable table to Slack showing publisher performance.

4.Lambda Function #3: Missing Publisher Detector (optional)

Runs a separate Athena query to find books with missing publisher info.
Sends a notification to Slack with a direct link to the generated file in S3.

Project Structure

publisher-reporting/
│
├── deploy.sh                         # Infra provisioning and Lambda packaging
├── config.yaml                       # Runtime config (bucket names, cron schedules)
│
├── lambda/
│   ├── report_generator/
│   │   └── handler.py                # Runs Athena query and stores report in S3
│   ├── summary_report_notifier/
│   │   └── handler.py                # Aggregates data and posts to Slack
│   └── missing_publisher_report/
│       └── handler.py                # Detects and reports missing metadata
│
└── terraform/                        # Infra as Code (Lambda, EventBridge, IAM, etc.)

Code Snippets

Lambda: Report Generator

exec_response = athena.start_query_execution(
    QueryString=query_string,
    QueryExecutionContext={"Database": ATHENA_DATABASE},
    ResultConfiguration={
        "OutputLocation": f"s3://{ATHENA_OUTPUT_BUCKET}/temporary-athena-query-results/"
    }
)
...
s3.copy_object(
    Bucket=TARGET_REPORT_BUCKET,
    CopySource={"Bucket": source_bucket, "Key": source_key},
    Key=final_key
)

Lambda: Slack Summary Formatter

for row in reader:
    if row["book_read_counts"].strip().upper() == "TRUE":
        yq = row["year_quarter"].strip()
        pub = row["publisher"].strip() or "Unknown"
        counts[yq][pub] += 1

Scheduling

+------------------------------+------------------------------------------+-------------------------+
| Lambda Function              | Purpose                                  | Schedule (Cron Format)  |
+------------------------------+------------------------------------------+-------------------------+
| report_generator             | Run Athena query and save CSV to S3      | cron(0 0 * * ? *)       |
| summary_report_notifier      | Read latest CSV and post Slack summary   | cron(10 0 * * ? *)      |
| missing_publisher_report     | Detect books without publisher info      | cron(15 0 * * ? *)      |
+------------------------------+------------------------------------------+-------------------------+

# Notes:
# - Times are in UTC
# - 10–15 min stagger prevents overlap and race conditions

Challenges & Fixes

1.Slack showed same data daily

Issue:
Slack message was posting the same publisher data every day.

Fix:

EventBridge schedule was set to run only on the 1st of each month.
Updated to run daily using: cron(0 0 * * ? *)

2.No new publisher summary files after Aug 1

Issue:
S3 bucket had no updated files after August 1st.

Fix:

Found that the publisher report generator Lambda was not running.
Corrected the EventBridge schedule to trigger daily.

3.Lambda race condition

Issue:
The Slack posting Lambda was sometimes reading an older CSV file instead of the one just generated.

Fix:

Introduced a 10-minute delay between the generator Lambda and the Slack reporter Lambda using separate schedules.

4.Slack output was hard to read

Issue:
The publisher read counts in Slack were misaligned and difficult to follow.

Fix:

Formatted the message using Slack-compatible triple backticks (```

...

```) to show preformatted blocks.

5.S3 bucket getting cluttered

Issue:
Temporary Athena result files were crowding the output bucket.

Fix:

Moved results to a 'temporary-athena-query-results/' folder prefix
Added a lifecycle policy to auto-delete them after 3 days.

Impact

Eliminated all manual report generation
Improved team visibility into reader engagement
Ensured scalable, serverless infrastructure using AWS best practices
Automated alerts improved issue tracking and data consistency

This project was a great example of combining Athena, Lambda, S3, and EventBridge into a cost-efficient, automated reporting pipeline. If you’re working with serverless data workflows, this pattern is easily adaptable to product analytics, user activity tracking, sales dashboards, and more.

Kickstarting Generative AI with Amazon Bedrock: Building a No-Code Product Description Generator

Gurudev Prasad Teketi — Sat, 31 May 2025 05:03:22 +0000

Introduction: What and Why

In today’s fast-moving digital world, the ability to generate personalized, high-quality content at scale is a game-changer for businesses across industries. Whether it’s marketing copy, product descriptions, chatbot responses, or internal documentation, Generative AI has the potential to significantly reduce content creation time while improving consistency and creativity.

That’s where Amazon Bedrock comes in.

Amazon Bedrock is a fully managed service that lets developers build and scale generative AI applications using foundation models (FMs) from leading AI providers like Anthropic (Claude), AI21, and Amazon Titan—without needing to manage any infrastructure.

This project explores how non-developers and beginners can quickly prototype a generative AI solution directly from the AWS Console. Our use case: generating fun, compelling product descriptions with no code using the Claude model via Bedrock.

Project Overview

Use Case:
Generate AI-powered product descriptions for e-commerce products (e.g., a waterproof smartwatch for kids).

Tech Stack:

Amazon Bedrock (for text generation)
Claude model (Anthropic's foundation model)
AWS Console (no programming required)

Step-by-Step Instructions

Go to Amazon Bedrock → Model Access → Enable Claude

Use Claude 3 Sonnet

It offers a great balance between speed, cost, and reasoning power.
Perfect for text generation tasks like product descriptions.

Open Playground
Use the “Text Generation” option
Prompt: "Generate a fun, engaging product description for a waterproof smartwatch designed for children ages 6 to 12."

Explore variations with different Prompts and their results

Generate a fun, engaging product description for a waterproof smartwatch designed for children ages 6 to 12.

`Here's a fun, engaging product description for a waterproof smartwatch designed for children ages 6 to 12:

Dive into Adventure with the Aqua Kiddo Smartwatch!

Calling all little adventurers! Get ready to make a splash with the Aqua Kiddo Smartwatch – the ultimate waterproof companion for your wild and wacky explorations. Whether you're splashing in the pool, braving the sprinklers, or caught in a surprise rain shower, this smartwatch is built to keep up with your aquatic antics.

With its vibrant, customizable watchfaces and fun games, the Aqua Kiddo is more than just a timepiece – it's a portable playground strapped to your wrist! Challenge your friends to epic underwater dance-offs, or race against the clock in exciting tap-tastic games. And when you're not making waves, stay connected with your crew through fun emoji messages and reminders.

But the fun doesn't stop there! This smartwatch is also a trusty sidekick for your land adventures. Track your steps, set fitness goals, and even unlock cool virtual rewards for your active accomplishments. And with its durable, water-resistant design, you don't have to worry about a little mud, sand, or spills slowing you down.

So what are you waiting for? Dive headfirst into endless fun and excitement with the Aqua Kiddo Smartwatch – the ultimate aquatic accessory for kids who love to make a splash!
`

“Write a playful and humorous product description for a kids’ waterproof smartwatch.”

`Model: Here's a playful and humorous product description for a kids' waterproof smartwatch:

Unleash the Aqua Adventure with the Splish-Splash SmartKid Watch!

Are your little ones always making a splash, whether in the pool, at the beach, or even in the bathtub? Fear not, because the Splish-Splash SmartKid Watch is here to ensure their aquatic escapades are both fun and functional! This waterproof wonder is designed to withstand even the most epic water battles, complete with cannonball dives and whirlpool whirls.

But wait, there's more! This smartwatch isn't just a sidekick for your child's underwater antics; it's a multi-talented marvel that'll keep them entertained and engaged, even when they're back on dry land. With its vibrant display and kid-friendly interface, they can track their steps, set reminders for playdates and chores (good luck with that one!), and even challenge their friends to virtual pet-raising competitions.

And let's not forget the coolest feature of all – the built-in water detector! That's right, this watch knows when your little fish has taken the plunge, and it'll automatically switch to "aqua mode," complete with underwater games and challenges to keep them occupied while they're making waves.

So, whether your kiddo is a budding marine biologist or just a professional puddle-jumper, the Splish-Splash SmartKid Watch is the perfect companion for their wet and wild adventures. Just be prepared for them to never want to leave the water (or your bathtub) again!`

Other Options to Try

Temperature - Controls creativity. Lower = focused, Higher = diverse/random - Try 0.7 (default), 0.3 (precise), 0.9 (creative)

Top P - Controls randomness, like temperature but in a probabilistic way - Keep it at 1.0 unless experimenting

Max Tokens - Controls length of output - Start with 300–500

Key Takeaways

No-code start: You don’t need to write any code to start experimenting with generative AI on AWS.
Prompt engineering matters: The way you phrase your input prompt greatly affects the quality and tone of the output.
Great for rapid prototyping: Bedrock is ideal for trying different use cases like product content, customer support scripts, or email generation.

This mini-project demonstrates how easy and powerful it is to prototype generative AI use cases using Amazon Bedrock and Claude 3 Sonnet—without writing a single line of code. It's a perfect starting point for anyone exploring content automation, prompt engineering, or AI-driven creativity on AWS.

Automate AWS Cost Monitoring & Explanations using EventBridge + Bedrock

Gurudev Prasad Teketi — Fri, 25 Apr 2025 13:54:43 +0000

Key concepts that we are going to explore

AWS Cost Tools - Cost Explorer, Budgets, Trusted Advisor, Compute Optimizer
EventBridge - Rule-based automation, Event pattern filtering
Bedrock + LangChain (light) - Trigger LLM to explain cost anomalies
Lambda - Glue logic between AWS services
SNS - Cost alerts via email/SMS are going to explore.

🛠️ Architecture:

Step 1:

Enable AWS Cost Explorer

Step 2:

Configure AWS Budgets for service cost threshold (e.g., EC2 > $20/day)

Step 3:

Create an SNS Topic (e.g., CostAlertsTopic) and subscribe via email

Step 4:

Enable Amazon Bedrock which is available in:

us-east-1 (N. Virginia)
us-west-2 (Oregon)
eu-central-1 (Frankfurt)
ap-southeast-1 (Singapore)

Recommended Model: Anthropic Claude v2 (via Amazon Bedrock)

Why Claude v2?

Excellent at summarization and natural explanations
Fast and reliable via Bedrock API
Supports secure inference with no fine-tuning needed
Default model in many Bedrock-based examples

Model Details for Bedrock Invocation

Model ID: "anthropic.claude-v2"
Provider: Anthropic
API Method: InvokeModel
Region: us-east-1 or other supported region
Input format: Anthropic-specific prompt (not plain OpenAI style)

Step 5:

Setup Event bridge rule - Target Lambda Function

{
  "source": ["aws.budgets"],
  "detail-type": ["Budget Threshold Breached"]
}

Step 6:

Lambda Function: Analyze & Explain Cost Spike

import boto3
import json
import os

def lambda_handler(event, context):
    bedrock_runtime = boto3.client("bedrock-runtime", region_name="us-east-1")

    anomaly_detail = json.dumps(event['detail'], indent=2)
    prompt = f"Explain the following AWS cost anomaly in plain English:\n\n{anomaly_detail}"

    response = bedrock_runtime.invoke_model(
        modelId="anthropic.claude-v2",  # or other Bedrock-supported LLM
        body=json.dumps({"prompt": prompt, "max_tokens_to_sample": 500}),
        contentType="application/json"
    )

    output = json.loads(response['body'].read())
    explanation = output.get("completion", "No explanation generated.")

    sns = boto3.client('sns')
    sns.publish(
        TopicArn=os.environ["SNS_TOPIC_ARN"],
        Subject="AWS Cost Alert 📈",
        Message=explanation
    )

    return {"statusCode": 200, "body": explanation}

Note - Set SNS_TOPIC_ARN as an environment variable in Lambda

In this hands-on project, we combined the power of AWS automation and Generative AI to create an intelligent, event-driven cost monitoring system.
By leveraging AWS Budgets, EventBridge, Lambda, and Amazon Bedrock, we not only detected cost anomalies in real-time but also generated human-readable explanations using LLMs — making cloud spend analysis more accessible and actionable.

This is just the beginning of what’s possible when AI meets Cloud FinOps.
Imagine extending this to proactive optimization suggestions, ticket creation, or even Slack alerts — all powered by serverless workflows and GenAI.

Design and Manage Pub/Sub Messaging with Amazon Simple Notification Service (SNS)

Gurudev Prasad Teketi — Fri, 21 Mar 2025 03:14:14 +0000

Amazon Simple Notification Service (Amazon SNS)

Amazon Simple Notification Service (Amazon SNS) is a fully managed messaging service that enables the coordination of message delivery from publishers to subscribers.
It provides a highly available, secure, and scalable messaging infrastructure, allowing applications to send notifications to various endpoints, such as email addresses, mobile devices, or AWS services.
With Amazon SNS, you can decouple message producers from consumers, enabling asynchronous communication and event-driven architectures.

Objectives

Configure message filtering policies in Amazon SNS to control message delivery to subscribers.
Set up CloudWatch metrics to monitor SNS message activity.
Create SNS topic subscriptions with multi-attribute filtering rules.
Use CloudWatch metrics to monitor message delivery and filtering effectiveness.

Task 1: Subscribe to an SNS topic

Set up multiple subscriptions to the Amazon SNS topic to simulate different teams receiving notifications. You’ll start with a basic subscription and then create additional subscriptions for the customer support and inventory management teams.

Task 1.1: Configure Initial Subscription

Open Simple Notification Service.
select Topics.
Select ecommerce-notifications.
On the ecommerce-notifications topic page, choose Create subscription.
On the Create subscription page, configure the following settings:
Topic ARN: The ARN of the ecommerce-notifications topic is already filled in. Leave this unchanged.
Protocol: Choose Email.
Endpoint: Enter an email address where you can receive notifications during this lab.
Note: You can create multiple subscriptions for a single topic, and each subscription can use different protocols including email, SMS (text messages), or other AWS services. This allows you to send the same message to multiple destinations simultaneously.
Choose Create subscription.
Check your email for a message from AWS Notifications.
In the email body, choose the Confirm subscription link.
Note: Now that your email address is subscribed to the topic, you will receive a notification whenever a new message is published to it.

Task 1.2: Create Team-Specific Subscriptions

From the breadcrumbs at the top of the screen, choose the ecommerce-notifications link.
On the ecommerce-notifications topic page, choose Create subscription.
Configure the following settings:
- Protocol: Choose Email
- Endpoint: Enter your email address with +support added before the @ symbol. Example: student+support@example.com
Choose Create subscription.
Repeat the process for the inventory management team:
Return to the ecommerce-notifications page.
Choose Create subscription
- Protocol to Email
- Endpoint, enter your email address with +inventory added before the @ symbol. Example: student+inventory@example.com
Choose Create subscription

Task 1.3: Verify All Subscriptions

Check your email inbox for two additional confirmation messages from AWS Notifications.
Confirm both new subscriptions by clicking the Confirm subscription link in each email.
Return to the SNS console and select Subscriptions from the left navigation menu.
Verify that you see the following three confirmed subscriptions:
- Your original email address
- Your email address with +support
- Your email address with +inventory

Task 2: Monitor SNS Performance with CloudWatch

_- Amazon CloudWatch provides valuable insights into the performance and health of your SNS topics.

Lets create a dashboard to monitor key metrics, including delivery and failures rates.
This visibility helps you ensure your notification system is working efficiently and allows you to quickly identify any potential issues.
By tracking metrics like the number of messages published, successful deliveries, and failed notifications, you can:
1. Verify that messages are being delivered to subscribers
2. Identify potential delivery problems
3. Monitor the overall health of your SNS topic
4. Make data-driven decisions about your notification system_
Open CloudWatch.
Under Metrics, choose All metrics.
In the Metrics search bar, enter SNS and press Enter.
In the filtered lit of SNS Metrics, select SNS > Topic Metrics.
A list of topics and available metrics appears.
Identify the metrics for the ecommerce-notifications topic and select the checkboxes next to the following metrics:
- NumberOfMessagesPublished
- NumberOfNotificationsDelivered
- NumberOfNotificationsFailed - WARNING: Do not select the PublishSize metric.
Now that you’ve selected metrics to track, you create a new CloudWatch Dashboard and add them to it.
Open the Actions menu at the top of the screen and select Add to dashboard.
In the Add to dashboard popup window, choose Create new.
Configure the following for your dashboard:
- Create new dashboard: Enter SNSMetricsDashboard.
- Widget type: Choose Stacked area.
- Customize widget title: Enter SNS Performance and Health Metrics.
Choose the Create button.
Choose Add to dashboard button.
Save

Task 3: Implement Message Filtering

Implementing message filtering to ensure that different teams receive only the notifications relevant to their responsibilities. You work with an existing Lambda function that publishes various types of messages to your SNS topic:

High-value order notifications for the sales team
Support tickets for the customer service team
Inventory alerts for the warehouse team

By configuring message filtering policies, you’ll ensure that each team receives only the notifications they need to act upon, reducing noise and improving operational efficiency.

Task 3.1: Configure Filtering Policies

Open Simple Notification Service.
Choose Subscriptions.
You start by updating the main subscription (your original email address).
Select the radio next to the subscription with your original email address and choose the Edit button at the top of the screen.
On the Edit subscription screen, expand the Subscription filter policy - optional section.
Use the Subscription filter policy toggle button to enable filtering for this subscription.
Note: The filter policy scope in Amazon SNS determines whether filter conditions are applied to message attributes or the message body. Choosing Message attributes applies filtering based on metadata, whereas Message body filters based on content, allowing precise control over message delivery to subscribers.
Choose Message attributes.
In the JSON editor text box, enter the following policy:

{
     "category": ["high_value"],
     "message_type": ["order_notification"]
}

Save
Repeat the same for support and inventory subscriptions.

{
  "message_type": ["support_ticket"],
  "category": ["customer_service"]
}

{
  "message_type": ["inventory_alert"],
  "category": ["inventory"]
}

Task 3.2: Test Message Filtering

Return to the Topics page and select your topic.
Choose Publish message.
Start by confirming that your main email address receives notifications when high value orders are placed.
Choose the Publish message button at the top of the screen and create a test message for a high value order:
- Subject: Enter Test - High value order alert
- Message body: Enter An order for $750 has been placed
- Add message attribute:
- Type: Choose String
- Name: Enter category
- Value: Enter high_value
Choose Add another attribute
- Add message attribute:
- Type: Choose String
- Name: Enter message_type
- Value: Enter order_notification
- Choose Publish message
Check your email inbox:
Look at the email header and confirm that it was sent to your original email address The +support and +inventory aliases should not receive this message Note: When using multiple attributes in a filter policy, SNS applies an AND operation between them. This means a subscription will only receive a message if ALL attributes in its filter policy match the message’s attributes. For example, given the policy you just created:

{
     "category": ["high_value"],
     "message_type": ["order_notification"]
}

Your email address will only receive notification that have both:
category = high_value AND
message_type = order_notification

This powerful filtering capability has many applications beyond e-commerce. For example:

A weather alert system could filter by both severity (critical) AND location (west-coast)
A security monitoring system could filter by both threat level (high) AND asset type (database)
A manufacturing system could filter by both machine status (maintenance-required) AND facility location (plant-A)
This allows you to create very specific routing rules for your messages, ensuring that subscribers only receive the exact subset of messages they need to act upon.

Now test that your support team subscription receives customer service tickets.
Choose the Publish message button at the top of the screen and create the following test message for support tickets:
- Subject: Enter Test - Customer support ticket
- Message body: Enter New support ticket opened
- Add message attribute:
- Type: Choose String
- Name: Enter category
- Value: Enter customer_service
- Choose Add another attribute
- Add message attribute:
- Type: Choose String
- Name: Enter message_type
- Value: Enter support_ticket
Choose Publish message
Check your email inbox:
Look at the email header and confirm that it was sent to your +support email address
Your original email and +inventory alias should not receive this message
Create a test message for inventory alerts:
- Subject: Enter Test - Inventory alert
- Message body: Enter Low stock warning
- Add message attribute:
- Type: Choose String
- Name: Enter category
- Value: Enter inventory
Choose Add another attribute
- Add message attribute:
- Type: Choose String
- Name: Enter message_type
- Value: Enter inventory_alert
- Choose Publish message
Check your email inbox:
Look at the email header and confirm that it was sent to your +inventory email address
Your original email and +support alias should not receive this message

The following table shows a summary of your filtering configuration:

Task 4: Analyze SNS Performance Using CloudWatch

Task 4.1: Review and update CloudWatch Dashboard

Choose CloudWatch.
choose Dashboards.
Select the dashboard named SNSMetricsDashboard that you created earlier.
Note: To adjust the dashboard size, look for the resize handle in the bottom-right corner. You can drag this handle to increase the viewing area of your dashboard.
In the time series bar above the dashboard, select Custom and then choose 30 minutes.
In the dashboard, observe the following metrics:
- NumberOfMessagesPublished: Shows how many messages the Lambda function is publishing
- NumberOfNotificationsDelivered: Shows successful message deliveries to subscribers
- NumberOfNotificationsFailed: Shows failed delivery attempts
Note: You should see:
The number of MessagesPublished should be largely consistent, as this is controlled by a Lambda function triggering at regular intervals
Few or no NotificationsFailed (indicating successful delivery)
Hover over any data point in the graphs to see exact values at that time.
Now, let’s investigate how filtering has affected your SNS topic.
In the upper-right corner of your dashboard, select the icon with three vertical dots.
From the menu that appears on screen, choose Edit.
Choose the Browse tab.
Enter NumberOfNotificationsFilteredOut-MessageAttributes in the search bar and press Enter.
Choose SNS > Topic Metrics.
Select the checkbox next to ecommerce-notifications and then choose Update widget.
Choose Save.
With the additional metric added to your dashboard, you can now see how filtering has affected your SNS topic.

Task 4.2: Create CloudWatch Alarms

In the CloudWatch console, expand the navigation menu by choosing the menu icon in the upper-left corner, then open the Alarms menu and select All alarms.
Choose Create alarm.
Choose Select metric.
Select the Browse tab.
Enter NumberOfNotificationsFailed in the search bar and press Enter.
Choose SNS > Topic Metrics.
Select the checkbox next to ecommerce-notifications and then choose Select metric.
Configure the following alarm conditions:
Under Metric, leave the default settings
Under Conditions:
- For Threshold type, choose Static
- For Whenever NumberOfNotificationsFailed is…, choose Greater
- For than…, enter 0
Note: This alarm will trigger whenever any message fails to deliver. Setting the threshold to 0 ensures you’re notified of even a single failure, which is important for maintaining reliable message delivery in a production environment.
Choose Next.
Configure the notification:
Under Notification, choose Create new topic
- For Create a new topice, enter sns-alarms
- For Email endpoints, enter your email address
Choose Create topic
Choose Next.
For Alarm name, enter SNSFailedNotifications.
Choose Next.
Review your alarm configuration and choose Create alarm.
Check your email and confirm the subscription to the new SNS topic.

Review

Configured message filtering policies in Amazon SNS to control message delivery to subscribers.
Set up CloudWatch metrics to monitor SNS message activity.
Created SNS topic subscriptions with multi-attribute filtering rules.
Used CloudWatch metrics to monitor message delivery and filtering effectiveness.

Zero Trust Architecture for Service-to-Service Workload in AWS

Gurudev Prasad Teketi — Thu, 20 Mar 2025 10:25:13 +0000

Zero Trust Architecture in AWS

is a security framework that implements Zero Trust principles across AWS services and workloads. It’s designed to verify every request for access to resources—regardless of its origin—based on identity, device state, and security posture.

This hands-on lab focuses on applying Zero Trust principles to the service-to-service architectures used in many micro-architecture or distributed environments today.

High-level architecture of the lab

The scenario is comprised of two services:

The calling services component has multiple compute nodes which make API calls to the target services. The API calls are made by a fleet of Amazon Elastic Compute Cloud (Amazon EC2) instances and AWS Lambda functions. These compute nodes resides in different subnets within multiple Amazon VPCs.
The target services component has multiple REST APIs hosted by the API Gateway, multiple backend Lambda functions and other backend services like Amazon DynamoDB tables.
Because the REST APIs are set as private, VPC Endpoints are configured within the calling services subnets to provide a network path for the API calls from the calling services using AWS PrivateLink.

Task 1: Review the lab scenario current state service-to-service architecture

Recommendation service (ServiceA) : Generates recommendations based on previous orders retrieved from the Order History service (ServiceB). ServiceA valid compute nodes are represented by the EC2 instance denoted as Expected Caller.

Order History service (ServiceB) : Provides access to customers past orders. This is represented by the ServiceB API. The API communicates at the backend with a Lambda function and a DynamoDB table to store the order table.

In addition to the above, there are other unwanted callers and unknown APIs which perform other functions but should not be able to interact with the main two services above.

There are many conditions that you can use to allow or block the API call using different security controls:

Does the caller need to make an API call to the target service?
Does the API call has a valid path to the target service?
Is the API call sourced from a valid CIDR IP range?
Does the caller have the correct IAM permissions to make the API call via IAM role?
Is the API call authenticated (signed) using AWS Signature V4 (SigV4) algorithm?
Note: AWS Signature V4 is an algorithm used to sign the API calls. When enforced, it requires that the caller uses valid IAM credentials to sign the API calls.

Task 2: Review the existing security controls

where you can apply security controls in the solution:

API Gateway:

IAM authorization on the API level: This is where you can require that the caller has valid IAM credentials and the API call is signed using SigV4. If turned off, the API can be called by any entity without having valid IAM credentials.
Resource policy: This is where you can apply an IAM resource policy with specific conditions to restrict making calls to the API.
API Endpoint Type: You can control where the API can be called from depending on the type you choose. In the lab scenario, both the ServiceB API and Unknown API are set to Private. This means that calls must be initiated from a API Gateway VPC endpoint from within a VPC.

VPC Endpoint:

Resource policy: This is where you can apply an IAM resource policy with specific conditions to restrict making calls to the API. This is similar to the API Gateway resource policy but configured closer to the source of the API call.
Security Group: You can define inbound/outbound rules to control the traffic traversing through the VPC endpoint.

Step 1: Open API Gateway
Step 2: In the list of APIs, under the name column, choose the ServiceBAPI link.
Step 3: In the Resources pane, under /orders, choose GET.
Step 4: In the /orders-GET-Method Execution section, choose the Method request tab.
Step 5: In the Settings page, verify that Authorization is set to NONE.
Step 6: In the navigation pane at the left of the page, in the API:ServiceBAPI section, choose Resource policy.
Step 7: Check the existing IAM resource policy configured on the ServiceBAPI. The policy allows any call sourced from an IP address within the 10.199.0.0/24 range. This range is similar to the range of Private subnet 1 in the ServiceA VPC and also the Private subnet 2 in the Other VPC. You can verify the ranges using the Detailed architecture diagram and/or the Callers table presented earlier.

Step 8: Open VPC
Step 9: In the navigation pane at the left of the page, in the Virtual private cloud section, choose Endpoints
Step 10: In the Endpoints page, select the endpoint with a VPC endpoint ID that matches the APIGWVPCEndpointID value that is listed to the left of these instructions.
Step 11: In the VPC endpoint Details section under the list of Endpoints, choose the Tags tab, then choose Manage tags.
Step 12: Choose Add new tag.
Step 13: On the Manage tags page, in the Tags section:
For Key, enter Name.
For Value - optional, enter ServiceA-APIGW-EP.
Step 14: Save and While the ServiceA-APIGW-EP endpoint name is selected, choose the Policy tab.
Note that the existing policy allows all traffic to traverse through the endpoint to make a call to any Private API in the same region with no restrictions.

Task 3: Run an assessment to evaluate your current security posture

Step 1: Connect to Service A EC2 instance.
Step 2: run the command "runscanner" to run the scan.

Based on the assessment results above, only two out of six API calls patterns comply with desired security outcome. This really raises an alarm and does not fit well with the Zero Trust principles that you need comply with.

Task 4: Improve your security posture using IAM authorization on the API Gateway

Step 1: choose API Gateway.
Step 2: In the list of APIs, under the name column, choose the ServiceBAPI link.
Step 3: In the Resources panel, under the /orders, choose GET.
Step 4: In the /orders-GET-Method Execution section, choose the Method request tab.
Step 5: In the Method request settings section, choose Edit, and then from the Authorization drop-down menu, select AWS IAM.
Step 6: To save your selection, choose Save.
Step 7: After saving your selection, you need to deploy the API for your changes to take effect.
Step 8: In the Resources panel, choose Deploy API.

For Stage, select api.
Choose Deploy.

Task 5: Improve your security posture using API Gateway resource policy

Step 1: Return to the API Gateway Console browser tab.
Step 2: In the ServiceBAPI left panel, choose Resource policy.
Step 3: Choose Edit, and then delete the entire policy.
Step 4: In the Policy details pane, add the following policy after making these two changes:
Replace the INSERT_AWS_ACCOUNT_ID placeholder value with the AWSAccountID value that is listed to the left of these instructions (keep the quotes).
Replace the INSERT_APIGW_VPC_ENDPOINT_ID placeholder value with the APIGWVPCEndpointID value that is listed to the left of these instructions (keep the quotes).

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Principal": "*",
            "Action": "execute-api:Invoke",
            "Resource": "arn:aws:execute-api:*:*:*"
        },
        {
            "Effect": "Deny",
            "Principal": "*",
            "Action": "execute-api:Invoke",
            "Resource": "arn:aws:execute-api:*:*:*",
            "Condition": {
                "StringNotEquals": {
                    "aws:PrincipalAccount": "INSERT_AWS_ACCOUNT_ID"
                }
            }
        },
        {
            "Effect": "Deny",
            "Principal": "*",
            "Action": "execute-api:Invoke",
            "Resource": "arn:aws:execute-api:*:*:*",
            "Condition": {
                "StringNotEquals": {
                    "aws:SourceVpce": "INSERT_APIGW_VPC_ENDPOINT_ID"
                }
            }
        }
    ]
}

Step 5: Save
After saving the policy, you need to deploy the API for your changes to take effect.
Step 6: In the navigation pane at the left of the page, in the API:ServiceBAPI section, choose Resources.
Step 7: In the Resources panel, choose Deploy API.

For Stage, select api.
Choose Deploy.

Step 8: Open the Session Manager terminal session to the ServiceA-Instance.
Step 9: Run command "runscanner"

Task 6: Improve the security posture using VPC endpoint policy

Step 1: Return to the API Gateway Console browser tab.
Step 2: At the top of the AWS Management Console, in the search bar, search for and choose VPC.
Step 3: In the navigation pane at the left of the page, in the Virtual private cloud section, choose Endpoints.
Step 4: In the list of Endpoints, under the Name column, locate the ServiceA-APIGW-EP and select it.
Step 5: In the VPC endpoint Details section under the list of Endpoints, choose the Policy tab.
Step 6: In the Policy pane, choose Edit Policy.
Step 7: In the Edit policy page, ensure that Custom is selected.
Step 8: Select the existing policy and delete it.
Step 9: In the Policy pane, add the following policy after making these two changes:

Replace the INSERT_SERVICEA_INSTANCE_ROLE_ARN placeholder value with the ServiceAInstanceRoleARN value that is listed to the left of these instructions (keep the quotes).
Replace the INSERT_API_METHOD_ARN placeholder value with the APIMethodARN value that is listed to the left of these instructions (keep the quotes).

Step 10: Save
Step 11: Open TerminalA and run command "runscanner"

Task 7: Improve the security posture by tuning the VPC endpoint security group

Step 1: In the list of Endpoints, select ServiceA-APIGW-EP endpoint if not already selected.
Step 2: Choose the Security Groups tab. There is one security group applied to the VPC endpoint.
Step 3: Under the Group ID, choose the security group link.
Step 4: On the Security Groups page, choose the security group link.
Step 5: In the Security Group Details section, choose the Inbound rules tab and examine the existing inbound rule.
Step 6: The current rule allows all HTTPS (TCP:443) traffic inbound to the VPC endpoint from the whole ServiceA VPC CIDR range of 10.199.0.0/16. This represents any subnet in the ServiceA VPC even though the expected caller resides only in Private subnet 1. This is a broad range of IP addresses which you need to narrow down.
Step 7: First, you delete the current rule and then add a new rule.
Step 8: Choose Edit inbound rules.
Step 9: To delete the existing rule, choose Delete.
Step 10: To add a new rule, choose Add rule.
Step 11: For the new rule, configure the following settings:

For Type, choose HTTPS
For Source, choose Custom, then in the search field, select the security group that has ServiceASecurityGroup in its name.
For Description - optional, enter From ServiceA Instances Security Group Step 12: Choose Save rules.

Step 13: Open TerminalA and run command "runscanner"

From the assessment results, you can see that all security requirements are still met. However, Unwanted Callers #1, #2, and #3 are now blocked by the VPC endpoint security group level which is closer to the source.

https://aws.amazon.com/blogs/security/zero-trust-architectures-an-aws-perspective/

Hands-on Lab - Event Driven Architecture with Amazon API Gateway, Amazon EventBridge and AWS Lambda

Gurudev Prasad Teketi — Tue, 11 Mar 2025 04:54:44 +0000

Event-driven architecture

is a design pattern where events are utilized to trigger and facilitate communication between decoupled services, making it a common choice for modern applications built with microservices. Serverless architecture, on the other hand, provides a model for building and running applications and services without the need to provision or manage underlying infrastructure.

Objective of the Lab

Configure an HTTP API on API Gateway to redirect requests to EventBridge.
Create event bus rules that match incoming requests and route events to Lambda functions.
Process events with Lambda functions and send the processed events back to the event bus as new events.
Publish events to the event bus, triggering separate Lambda functions.
Lambda functions receive events and post them back to the client application via a WebSocket connection hosted on API Gateway.

High Level Architecture Diagram

Step 1

Implement five Lambda functions using below configurations and code and deploy each lambda

Function name, enter: make_pizza
- For Runtime, select: Python 3.12
- For Change default execution role, choose - lab_lambda_make_pizza
- Environment variable: Key: EVENT_BUS, Value: lab_event_bus
Function name, enter: cook_pizza
- For Runtime, select: Python 3.12
- For Change default execution role, choose - lab_lambda_cook_pizza
- Environment variable: Key: EVENT_BUS, Value: lab_event_bus
Function name, enter: deliver_pizza
- For Runtime, select: Python 3.12
- For Change default execution role,choose - lab_lambda_deliver_pizza
- Environment variable: Key: EVENT_BUS, Value: lab_event_bus
Function name, enter: websocket_connect
- For Runtime, select: Python 3.12
- For Change default execution role,choose - lab_lambda_websocket_connection
- Environment variable: Key: TABLENAME, Value: websocket_connections
Function name, enter: receive_events
- For Runtime, select: Python 3.12
- For Change default execution role,choose - lab_lambda_receive_events
- Environment variable: Key: TABLENAME, Value: receive_events

Sample Lambda function code for Make Pizza:

import json
import boto3
import os
from botocore.exceptions import ClientError

client = boto3.client('events')

def lambda_handler(event, context):
  try:
    detail = event["detail"]
    detail["item"]["eventtype"]="cook_pizza"
    response = client.put_events(
      Entries=[
        {
          'DetailType': 'eventtype',
          'Detail': json.dumps(detail),
          'EventBusName': os.environ.get('EVENT_BUS'),
          'Source':"make_pizza"
        },
      ]
    )
    print(response)
  except ClientError as err:
    print(err)

Sample Lambda function code for Cook Pizza:

import json
import boto3
import os
from botocore.exceptions import ClientError

client = boto3.client('events')

def lambda_handler(event, context):
  try:
    detail = event["detail"]
    detail["item"]["eventtype"]="deliver_pizza"
    response = client.put_events(
      Entries=[
        {
          'DetailType': 'eventtype',
          'Detail': json.dumps(detail),
          'EventBusName': os.environ.get('EVENT_BUS'),
          'Source':"cook_pizza"
        },
      ]
    )
    print(response)
  except ClientError as err:
    print(err)

Sample Lambda function code for Deliver Pizza:

import json
import boto3
import os
from botocore.exceptions import ClientError

client = boto3.client('events')

def lambda_handler(event, context):
  try:
    detail = event["detail"]
    detail["item"]["eventtype"]="delivered"
    response = client.put_events(
      Entries=[
        {
          'DetailType': 'eventtype',
          'Detail': json.dumps(detail),
          'EventBusName': os.environ.get('EVENT_BUS'),
          'Source':"deliver_pizza"
        },
      ]
    )
  except ClientError as err:
    print(err)

Sample Lambda function code for Websocket Connect:

import boto3
import json
import os
from botocore.exceptions import ClientError

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table(os.environ.get('TABLENAME'))

def lambda_handler(event, context):
  try:
    order_id=event['queryStringParameters']['order_id']
    response = table.put_item(
      Item={
      'order_id': order_id,
      'connection_id': event["requestContext"]["connectionId"]
      })

    return {
      "statusCode": 200,
      "headers": {
      "Content-Type": "application/json"
    }
    }
  except ClientError as err:
    print(err)

Sample Lambda function code for Receive Events:

import boto3
import json
import os
from botocore.exceptions import ClientError

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table(os.environ.get('TABLENAME'))
management = boto3.client('apigatewaymanagementapi', endpoint_url=os.getenv('APIGW_ENDPOINT'))

def lambda_handler(event, context):
  try:
    response = table.get_item(
      Key={
      'order_id': event["detail"]['item']['order_id']
      }
    )
    management.post_to_connection(
      Data=json.dumps(event["detail"]),
      ConnectionId=response["Item"]["connection_id"]
    )
  except ClientError as err:
    print(err)

Step 2
Configure EventBridge

Configure the rules with the following settings and select Next .
Build event pattern screen, under the Creation method section, select Custom pattern (JSON editor) for the Method.
Copy and paste the json object into the Event pattern editor,
On the Select target(s) page, from the Target 1 section under Target types, choose AWS service.
From the Select a target drop-down list, choose Lambda function.
From the Function drop-down list, choose make_pizza.
Choose the correspondoing role
Select Next and select Next again on the Configure tags - optional screen.
On the Review and create screen, select Create rule.

Configurations for each rule

Name: lab_make_pizza_rule
Event bus: lab_event_bus

{
  "detail": {
    "item": {
      "eventtype": ["make_pizza"]
    }
  }
}

Name: lab_cook_pizza_rule
Event bus: lab_event_bus

{
  "detail": {
    "item": {
      "eventtype": ["cook_pizza"]
    }
  }
}

Name: lab_deliver_pizza_rule
Event bus: lab_event_bus

{
  "detail": {
    "item": {
      "eventtype": ["deliver_pizza"]
    }
  }
}

Name: lab_receive_events_rule
Event bus: lab_event_bus

{
  "detail": {
    "item": {
      "eventtype": ["make_pizza","cook_pizza","deliver_pizza"]
    }
  }
}

Step 4
Configure API Gateway

In the console Choose API Gateway.
In the Choose an API type select Build from the HTTP API section.
In the API Name field, enter lab_http_api
Select Next until you get to the Review and Create screen, then choose Create.
Under the Develop section in the left-hand navigation pane, select Routes, then choose Create .
Select POST from the drop-down list, and select Create.
Under the Develop select Integrations, choose the POST route link and then select Create and attach an integration.
For Integration target, configure the following options.
Integration type: Amazon EventBridge
Integration action: PutEvents
EventBridge - PutEvents, configure the following options.
Detail: $request.body
Detail type: eventtype
Source: lab_http_api
Invocation role, copy the APIExecutionRoleARN
Expand Advanced settings, and configure the remaining fields. Select Create when you are finished.
For Event bus name - optional, enter the event bus ARN you saved in Task 2.
Region - optional.
From the left-hand navigation pane, select CORS and choose Configure in the top right corner.
Access-Control-Allow-Origin enter *.
Access-Control-Allow-Methods select, POST.
Access-Control-Allow-Headers, enter *.
Save.
From the navigation menu on the left-hand side of the screen, navigate to the endpoint details screen by selecting API: lab_http_api. Copy and save the Invoke URL from the Stages for lab_http_api section of the API: lab_api screen. The URL is needed in Task 4.
Now create the web socket API endpoint, which is used to send events back to the web application. Web socket connections are two-way persistent connections allowing bi-directional communication and are often used by chat applications.
At the top of the left-hand navigation pane, select APIs, then select Create API .
In the Choose an API type select **Build from the WebSocket API **section.
Configure the following fields and select Next .
API Name field: lab_websocket_api
Route selection expression: request.body.action
On the Add routes page choose Add $connect route and select Next .
The websocket_connect function invokes when the application establishes the connection. This function saves the connection ID to a DynamoDB table.
Integration type: Lambda.
Lambda function websocket_connect.
Select Next .
On the Add Stages page select Next, followed by Create and deploy .
Select Stages from the left-hand navigation page and choose the Production stage.
Copy and save the WebSocket URL, as it is needed shortly.
Copy the Connection URL up to, but do not include the @connections text.
Navigate back to the Lambda console and open the receive_events Lambda function.
On the configuration tab select Environment variables, choose Edit and select Add environment variable.
Configure the variable with the following settings, and select Save when you are finished.
Key: APIGW_ENDPOINT
Value: the copied Connection URL

Testing the ED Architecture