Real-time data processing, Part 1: AWS ECS App Mesh and retry strategies

In today's fast-paced data-driven world, real-time data processing has become indispensable for businesses across various sectors.

From monitoring system performance to analyzing customer behavior, the ability to process data in real-time offers invaluable insights for timely decision-making.

However, one critical aspect often overlooked is handling failures gracefully in real-time data processing pipelines.

In this part of our series, we will explore the importance of implementing robust retry strategies.

Important Note: AWS has announced that AWS App Mesh will be discontinued in September 2026. While App Mesh remains fully supported until then, you should consider AWS Service Connect instead.

Part 1: AWS ECS App Mesh and retry strategies

Amazon Elastic Container Service (ECS) is a fully managed container orchestration service that allows you to run, stop, and manage Docker containers on a cluster. It eliminates the need to install, operate, and scale a cluster management infrastructure.

AWS App Mesh belongs to the category of service meshes, which are specialized infrastructure layers designed to manage communication between services within a distributed application architecture.

Essentially, AWS App Mesh simplifies the networking aspect of your applications by offering features like service discovery, load balancing, encryption, authentication, and observability.

In essence, AWS App Mesh streamlines communication between microservices, allowing developers to focus on building application logic rather than worrying about networking setup.

Importance of retry strategies

In real-time data processing, failures are inevitable due to network issues, service disruptions, or transient errors. Therefore, implementing effective retry strategies becomes crucial. Here are some key aspects to consider:

1. Determining Retry Attempts: Deciding the number of retry attempts depends on factors like the criticality of the data, the likelihood of transient failures, and the impact on downstream processes.

2. Exponential Backoff: Adopting exponential backoff strategies can prevent overwhelming downstream systems during high-load scenarios.

3. Dead Letter Queues (DLQ): Implementing DLQs allows you to capture failed messages for further analysis and manual intervention.

Testing App Mesh retry policy

Let's put the App Mesh retry policy to the test by simulating failure scenarios in a real-time data processing pipeline.

We'll set up two services: a Data Ingestion service and a Data Processing service using Terraform.

Terraform configuration

We'll use Terraform to define the infrastructure. Here's a high-level overview of the key components:

Service Mesh: The logical boundary for the services:

resource "aws_appmesh_mesh" "app-mesh" {
  name = "${var.env}-${var.project}-app-mesh"

  spec {
    egress_filter {
      type = "DROP_ALL"
    }
  }
}

Virtual Service: Abstract representation of the services:

resource "aws_appmesh_virtual_service" "data-processing-service" {
  name      = "${local.services.data-processing}.${var.env}.${var.internal_domain}"
  mesh_name = aws_appmesh_mesh.app-mesh.id

  spec {
    provider {
      virtual_router {
        virtual_router_name = aws_appmesh_virtual_router.data-processing-service.name
      }
    }
  }
}

Virtual Node: Concrete implementations behind the virtual services:

resource "aws_appmesh_virtual_node" "data-ingestion-service" {
  name      = "${var.env}-${var.project}-data-ingestion-service"
  mesh_name = aws_appmesh_mesh.app-mesh.id

  spec {
    backend {
      virtual_service {
        virtual_service_name = aws_appmesh_virtual_service.data-processing-service.name
      }
    }

    listener {
      port_mapping {
        port     = var.ecs_services["data-ingestion"].app_port
        protocol = "http"
      }

      timeout {
        http {
          per_request {
            value = var.ecs_services["data-ingestion"].app_mesh_timeout.value
            unit  = var.ecs_services["data-ingestion"].app_mesh_timeout.unit
          }
        }
      }
    }

    service_discovery {
      aws_cloud_map {
        service_name   = aws_appmesh_virtual_service.data-processing-service.name
        namespace_name = aws_service_discovery_private_dns_namespace.internal.name
      }
    }

    logging {
      access_log {
        file {
          path = "/dev/stdout"
        }
      }
    }
  }
}

Route with Retry Policy:

resource "aws_appmesh_route" "data-processing-service" {
  name                = "${var.env}-${var.project}-data-processing-service"
  mesh_name           = aws_appmesh_mesh.app-mesh.id
  virtual_router_name = aws_appmesh_virtual_router.data-processing-service.name

  spec {
    http_route {
      match {
        prefix = "/"
      }

      retry_policy {
        http_retry_events = ["gateway-error"]
        max_retries = 12
        per_retry_timeout {
          unit  = "s"
          value = 5
        }
        tcp_retry_events = ["connection-error"]
      }

      action {
        weighted_target {
          virtual_node = aws_appmesh_virtual_node.data-processing-service.name
          weight       = 1
        }
      }
    }

    priority = 1
  }
}

Cloud Map Service Discovery:

resource "aws_service_discovery_private_dns_namespace" "internal" {
  name        = "${var.env}.${var.internal_domain}"
  description = "${var.env}-${var.project}-private-dns-namespace"
  vpc         = var.aws_vpc-vpc-id
}

resource "aws_service_discovery_service" "data-processing" {
  name = local.services.data-processing

  dns_config {
    namespace_id = aws_service_discovery_private_dns_namespace.internal.id
    dns_records {
      ttl  = 10
      type = "A"
    }
    routing_policy = "MULTIVALUE"
  }

  health_check_custom_config {
    failure_threshold = 1
  }
}

Here's a high-level overview of the Terraform configuration:

Backend services

1. Data Ingestion Service (Golang)

func main() {
	router := mux.NewRouter()
	router.HandleFunc("/health-check", HealthCheckHandler).Methods("GET")
	router.HandleFunc("/ev", ElectricVehicleHandler).Methods("POST")
	log.Fatal(http.ListenAndServe(":3000", router))
}

func ElectricVehicleHandler(w http.ResponseWriter, r *http.Request) {
	log.Printf("Request: %s %s", r.Method, r.URL.Path)

	body, err := io.ReadAll(r.Body)
	if err != nil {
		http.Error(w, "Failed to read request body", http.StatusInternalServerError)
		return
	}

	var payload ElectricVehiclePayload
	err = json.Unmarshal(body, &payload)
	if err != nil {
		http.Error(w, "Failed to decode JSON payload", http.StatusBadRequest)
		return
	}

	// Forward to data processing service
	req, _ := http.NewRequest("POST", dataProcessingEndpoint, bytes.NewBuffer(body))
	req.Header = r.Header.Clone()

	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		http.Error(w, "Failed to POST to data processing service", http.StatusInternalServerError)
		return
	}

	if resp.StatusCode != http.StatusOK {
		msg := fmt.Sprintf("Data processing service returned: %d", resp.StatusCode)
		http.Error(w, msg, resp.StatusCode)
		return
	}

	io.Copy(w, resp.Body)
}

2. Data Processing Service (Golang)

func ElectricVehicleDataProcessingHandler(w http.ResponseWriter, r *http.Request) {
	log.Printf("Request: %s %s", r.Method, r.URL.Path)

	// Simulate failure: return 503 if x-503 header is set
	if value, ok := r.Header["X-503"]; ok {
		log.Printf("X-503 header is set with values: %v", value)
		w.WriteHeader(http.StatusServiceUnavailable)
		w.Write([]byte("Service will return 503 — called with x-503 header set."))
		return
	}

	body, err := io.ReadAll(r.Body)
	if err != nil {
		http.Error(w, "Failed to read request body", http.StatusInternalServerError)
		return
	}

	var payload ElectricVehiclePayload
	json.Unmarshal(body, &payload)

	// Process the payload...

	w.WriteHeader(http.StatusOK)
	w.Write([]byte("Payload processed successfully"))
}

After deploying, we can confirm services are running:

And the App Mesh configuration:

Simulating failure scenarios

We confirm calls go through the Envoy proxy:

• server: envoy: request handled by Envoy Proxy
• x-envoy-upstream-service-time: 1: time taken to communicate with upstream

Now we call with the x-503 header set:

curl --location 'http://data-ingestion.dev.smn-app-mesh-ecs.internal:3000/ev' \
     --header 'Content-Type: application/json' \
     --header 'x-503: true' \
     --data '{
      "vehicle_id": "EV-001",
      "timestamp": "2024-02-15T10:30:00Z",
      "battery": { "percentage": 75, "voltage": 390, "temperature": 25.3 },
      "speed": 60,
      "odometer": 12500
    }'

Data Ingestion service logs:

Data Processing service logs:

For a single request, the Data Processing service returned 503 and the request was retried 12 times per the retry policy.

Without App Mesh, this retry logic would need to be implemented manually in application code. App Mesh offers a centralized, infrastructure-level approach.

Conclusion

Incorporating retry strategies is imperative for ensuring the resilience and reliability of real-time data communication between microservices.

With AWS ECS App Mesh, coupled with effective retry policies, organizations can mitigate the impact of failures and uphold data integrity in critical business processes.

In the next part of our series, we will explore additional crucial elements of real-time data processing.