Lab 63 — Ingestão em streaming: shard, ordem e reprocesso

#!/usr/bin/env bash
# medir-shard-quente.sh — prova o hot shard com numero, nao com suspeita
set -euo pipefail

STREAM="${STREAM:?defina STREAM}"
INICIO=$(date -u -d '5 minutes ago' +%Y-%m-%dT%H:%M:%S) # GNU date; em macOS use -v-5M

for SHARD_ID in $(aws kinesis list-shards --stream-name "$STREAM" \
                    --query 'Shards[].ShardId' --output text); do
  BYTES=$(aws cloudwatch get-metric-statistics \
    --namespace AWS/Kinesis --metric-name IncomingBytes \
    --dimensions Name=StreamName,Value="$STREAM" Name=ShardId,Value="$SHARD_ID" \
    --start-time "$INICIO" --end-time "$(date -u +%Y-%m-%dT%H:%M:%S)" \
    --period 300 --statistics Sum --query 'Datapoints[0].Sum' --output text)
  echo "$SHARD_ID: ${BYTES:-0} bytes em 5 min (teto do shard: 300.000.000 bytes)"
done
# Esperado ANTES da correcao: um shard perto do teto (1 MiB/s x 300s ~ 300 MB),
# os outros dois em uma fracao pequena disso. Depois de trocar a chave para o
# id do pedido, os oito shards do desenho de producao ficam dentro de 15% um
# do outro — e essa e a prova de que a distribuicao mudou, nao so a contagem.

// O payload que o Checkout API publica a cada transicao de estado do pedido.
// IdPedido e a CHAVE DE PARTICAO — nao Tipo, que foi o desenho minimo.
{
  "idPedido": "6f2a9e10-8b3c-4a71-9e2d-1c9f7a44b201",
  "tipo": "pedido_cancelado",
  "instante": "2026-08-08T14:03:12.418Z",
  "valor": 349.90,
  // Motivo so existe em eventos de cancelamento; campos que nao se aplicam ao
  // tipo simplesmente nao aparecem, em vez de irem como null.
  "motivo": "estorno_solicitado_pelo_cliente"
}

# kinesis.tf — o stream, com a lente do shard

resource "aws_kinesis_stream" "eventos_pedido" {
  name = "${var.projeto}-eventos-pedido"

  # Provisionado, nao on-demand: sob chave enviesada, o on-demand reparte o
  # trafego igualmente ao escalar e NAO isola a chave quente. So o modo
  # provisionado da o controle granular (SplitShard mirado) para corrigir UM
  # shard especifico sem recriar o stream inteiro.
  stream_mode_details {
    stream_mode = "PROVISIONED"
  }
  shard_count = 8 # dimensionado com margem sobre o pico de 3x o volume normal

  # Retencao alem do padrao de 24h: e o que HABILITA o reprocesso depois de
  # corrigir um bug no consumidor. 168h = 7 dias cobre o horizonte real de
  # deteccao de um bug de logica; o teto documentado e 8760h (365 dias).
  retention_period = 168

  encryption_type = "KMS"
  kms_key_id      = aws_kms_key.eventos.key_id

  # Sem isto, CloudWatch so mostra metricas agregadas do STREAM. E a metrica
  # POR SHARD que torna um shard quente visivel em vez de suspeita.
  shard_level_metrics = [
    "IncomingBytes",
    "IncomingRecords",
    "WriteProvisionedThroughputExceeded",
    "ReadProvisionedThroughputExceeded",
    "IteratorAgeMilliseconds",
  ]
}

# O produtor so PRECISA escrever. Ler e responsabilidade do consumidor, com
# um papel proprio — e e por isso que as duas policies abaixo sao distintas.
data "aws_iam_policy_document" "publicar_eventos" {
  statement {
    effect    = "Allow"
    actions   = ["kinesis:PutRecords"]
    resources = [aws_kinesis_stream.eventos_pedido.arn]
  }
  statement {
    effect    = "Allow"
    actions   = ["kms:GenerateDataKey", "kms:Decrypt"]
    resources = [aws_kms_key.eventos.arn]
  }
}

# Anexada ao MESMO task role da Checkout API do L01/L03 — nao criamos um
# papel novo para o publicador, porque a identidade que escreve continua
# sendo a mesma aplicacao.
resource "aws_iam_role_policy" "checkout_publica_eventos" {
  name   = "${var.projeto}-publica-eventos"
  role   = aws_iam_role.checkout_task_role.id
  policy = data.aws_iam_policy_document.publicar_eventos.json
}

output "stream_eventos_pedido_arn" {
  value = aws_kinesis_stream.eventos_pedido.arn
}

// PublicadorEventoPedido.cs — le a resposta de PutRecords antes de dar por certo

using Amazon.Kinesis;
using Amazon.Kinesis.Model;

public sealed class PublicadorEventoPedido
{
    private readonly IAmazonKinesis _kinesis;
    private readonly string _stream;
    private readonly ILogger<PublicadorEventoPedido> _log;

    public PublicadorEventoPedido(IAmazonKinesis kinesis, string stream,
        ILogger<PublicadorEventoPedido> log)
    {
        _kinesis = kinesis;
        _stream = stream;
        _log = log;
    }

    // A CHAVE e o id do pedido — nao o tipo do evento. E a linha que resolve
    // os dois sintomas do laboratorio: distribui a escrita (cardinalidade
    // alta) e garante que os eventos de UM pedido caiam no MESMO shard
    // (ordem). Trocar esta linha de volta para "evento.Tipo" reintroduz os
    // dois problemas ao mesmo tempo, porque tem a MESMA causa.
    private static string ChaveDeParticao(EventoPedido evento) => evento.IdPedido.ToString();

    public async Task PublicarLoteAsync(IReadOnlyList<EventoPedido> eventos,
        CancellationToken ct = default)
    {
        var pendentes = eventos
            .Select(e => new PutRecordsRequestEntry
            {
                PartitionKey = ChaveDeParticao(e),
                Data = new MemoryStream(JsonSerializer.SerializeToUtf8Bytes(e)),
            })
            .ToList();

        // Ate 4 tentativas: a primeira e as chamadas originais, as demais so
        // reenviam quem falhou. Backoff exponencial com jitter evita que o
        // proprio retry sincronizado piore o throttling que o causou.
        for (var tentativa = 1; tentativa <= 4 && pendentes.Count > 0; tentativa++)
        {
            var resposta = await _kinesis.PutRecordsAsync(new PutRecordsRequest
            {
                StreamName = _stream,
                Records = pendentes,
            }, ct);

            // O PONTO CENTRAL do laboratorio: a chamada acima NAO lanca
            // excecao por falha parcial. HTTP 200 e devolvido mesmo com
            // FailedRecordCount > 0 — quem so confere o status conclui
            // sucesso com registros que nunca entraram no stream.
            if (resposta.FailedRecordCount == 0)
                return;

            var proximaRodada = new List<PutRecordsRequestEntry>();
            for (var i = 0; i < resposta.Records.Count; i++)
            {
                var item = resposta.Records[i];
                if (item.ErrorCode is null) continue; // este teve sucesso

                _log.LogWarning(
                    "registro {Indice} falhou na tentativa {Tentativa}: {Codigo} — {Mensagem}",
                    i, tentativa, item.ErrorCode, item.ErrorMessage);
                proximaRodada.Add(pendentes[i]);
            }

            pendentes = proximaRodada;
            if (pendentes.Count > 0)
                await Task.Delay(BackoffComJitter(tentativa), ct);
        }

        if (pendentes.Count > 0)
        {
            // Esgotadas as tentativas do PRODUTOR — nao do stream. Isto e
            // perda real e precisa ser tratado como incidente, nunca como
            // log silencioso: metrica customizada + alarme, nao so LogError.
            _log.LogError(
                "{Quantidade} eventos descartados apos {Tentativas} tentativas — dado perdido",
                pendentes.Count, 4);
            throw new EventosPedidoNaoPublicadosException(pendentes.Count);
        }
    }

    private static TimeSpan BackoffComJitter(int tentativa)
    {
        var baseMs = Math.Pow(2, tentativa) * 100;
        var jitter = Random.Shared.Next(0, 100);
        return TimeSpan.FromMilliseconds(baseMs + jitter);
    }
}

public sealed record EventoPedido(Guid IdPedido, string Tipo, DateTimeOffset Instante, decimal Valor);

public sealed class EventosPedidoNaoPublicadosException(int quantidade)
    : Exception($"{quantidade} eventos de pedido nao foram publicados apos esgotar as tentativas");

# consumidor.tf — a Lambda em tempo real, o destino de falha, os alarmes

resource "aws_sqs_queue" "eventos_pedido_dlq" {
  name = "${var.projeto}-eventos-pedido-dlq"

  # 14 dias: maior que a retencao do stream, para dar tempo de investigar um
  # lote que falhou sem que a fila em si vire novo ponto de perda.
  message_retention_seconds = 1209600
  kms_master_key_id         = aws_kms_key.eventos.key_id
}

resource "aws_lambda_function" "alerta_fraude" {
  function_name = "${var.projeto}-alerta-fraude"
  role          = aws_iam_role.lambda_alerta_fraude.arn
  handler       = "FraudeFunction::FraudeFunction.Function::FunctionHandler"
  runtime       = "dotnet8"
  timeout       = 30
  memory_size   = 256

  environment {
    variables = {
      SNS_TOPIC_ARN = aws_sns_topic.alerta_risco.arn
    }
  }
}

resource "aws_lambda_event_source_mapping" "eventos_pedido_tempo_real" {
  event_source_arn  = aws_kinesis_stream.eventos_pedido.arn
  function_name     = aws_lambda_function.alerta_fraude.arn
  starting_position = "LATEST" # so o consumo normal comeca do fim; o replay usa outro mapping

  batch_size                        = 100
  maximum_batching_window_in_seconds = 2

  # Ate 10 lotes concorrentes POR SHARD. O Lambda continua garantindo ordem
  # por CHAVE de particao mesmo com paralelizacao > 1 — a concorrencia e
  # entre chaves diferentes do mesmo shard, nao dentro da mesma chave.
  parallelization_factor = 2

  # Um erro no lote e re-tentado ate 3 vezes, bissectando o lote pela metade
  # a cada tentativa — isola o registro ruim em vez de re-tentar o lote
  # inteiro do mesmo jeito repetidamente.
  bisect_batch_on_function_error = true
  maximum_retry_attempts         = 3
  maximum_record_age_in_seconds  = 3600

  destination_config {
    on_failure {
      # Esgotadas as tentativas, o lote vai para ca — e o shard SEGUE lendo
      # os proximos registros. Sem isto, um registro ruim travaria o shard
      # inteiro, e essa e a diferenca entre "um pedido com problema" e
      # "todo o stream parado".
      destination_arn = aws_sqs_queue.eventos_pedido_dlq.arn
    }
  }
}

resource "aws_sns_topic" "alerta_risco" {
  name              = "${var.projeto}-alerta-risco"
  kms_master_key_id = aws_kms_key.eventos.key_id
}

data "aws_iam_policy_document" "lambda_le_stream" {
  statement {
    effect = "Allow"
    actions = [
      "kinesis:GetRecords",
      "kinesis:GetShardIterator",
      "kinesis:DescribeStreamSummary",
      "kinesis:ListShards",
    ]
    resources = [aws_kinesis_stream.eventos_pedido.arn]
  }
  statement {
    effect    = "Allow"
    actions   = ["kms:Decrypt"]
    resources = [aws_kms_key.eventos.arn]
  }
  statement {
    effect    = "Allow"
    actions   = ["sqs:SendMessage"]
    resources = [aws_sqs_queue.eventos_pedido_dlq.arn]
  }
  statement {
    effect    = "Allow"
    actions   = ["sns:Publish"]
    resources = [aws_sns_topic.alerta_risco.arn]
  }
}

resource "aws_iam_role_policy" "lambda_le_stream" {
  name   = "${var.projeto}-lambda-le-stream"
  role   = aws_iam_role.lambda_alerta_fraude.id
  policy = data.aws_iam_policy_document.lambda_le_stream.json
}

# Os dois alarmes que decidem se o consumidor acompanha o pico, e nao a
# capacidade de escrita — que ja tem os seus proprios (nao mostrados aqui,
# ja cobertos pelo shard_level_metrics do kinesis.tf).
resource "aws_cloudwatch_metric_alarm" "iterator_age" {
  alarm_name          = "${var.projeto}-iterator-age-alta"
  namespace           = "AWS/Lambda"
  metric_name         = "IteratorAge"
  statistic           = "Maximum"
  period              = 60
  evaluation_periods  = 3
  # 30s de atraso e tolerado; acima disso o consumidor esta perdendo o pico,
  # nao so processando devagar.
  threshold           = 30000
  comparison_operator = "GreaterThanThreshold"
  treat_missing_data  = "notBreaching"

  dimensions = {
    FunctionName = aws_lambda_function.alerta_fraude.function_name
  }
  alarm_actions = [aws_sns_topic.alerta_operacao.arn]
}

resource "aws_cloudwatch_metric_alarm" "throughput_excedido" {
  alarm_name          = "${var.projeto}-shard-throughput-excedido"
  namespace           = "AWS/Kinesis"
  metric_name         = "WriteProvisionedThroughputExceeded"
  statistic           = "Sum"
  period              = 60
  evaluation_periods  = 2
  threshold           = 0
  comparison_operator = "GreaterThanThreshold"
  treat_missing_data  = "notBreaching"

  dimensions = {
    StreamName = aws_kinesis_stream.eventos_pedido.name
  }
  alarm_actions = [aws_sns_topic.alerta_operacao.arn]
}

// FraudeFunction.cs — le em ordem por pedido, e admite o limite do que sabe

using Amazon.Lambda.Core;
using Amazon.Lambda.KinesisEvents;
using Amazon.SimpleNotificationService;

namespace FraudeFunction;

public class Function
{
    private readonly IAmazonSimpleNotificationService _sns;
    private readonly string _topicoArn = Environment.GetEnvironmentVariable("SNS_TOPIC_ARN")!;

    // Cache em memoria, valido so ENQUANTO a execucao ficar quente. Nao e um
    // substituto para estado persistido: se o ambiente reciclar entre dois
    // eventos do mesmo pedido, o ultimo estado visto se perde. Para producao
    // de verdade isso pede uma tabela (DynamoDB), e essa decisao fica FORA
    // do escopo deste laboratorio, que e sobre ordem de ENTREGA, nao sobre
    // onde guardar estado de negocio — misturar os dois esconderia qual dos
    // dois esta sendo provado.
    private static readonly Dictionary<Guid, string> UltimoEventoVisto = new();

    public Function(IAmazonSimpleNotificationService sns) => _sns = sns;

    public async Task FunctionHandler(KinesisEvent evento, ILambdaContext contexto)
    {
        // Os registros deste lote pertencem TODOS ao mesmo shard, e chegam
        // na ordem em que o shard os recebeu — e ela so é confiável porque
        // a chave de particao é o id do pedido: todo evento de um pedido
        // sempre cai neste MESMO shard.
        foreach (var registro in evento.Records)
        {
            var dado = JsonSerializer.Deserialize<EventoPedidoLido>(
                registro.Kinesis.Data);
            if (dado is null) continue;

            if (UltimoEventoVisto.TryGetValue(dado.IdPedido, out var anterior)
                && anterior == "pedido_cancelado" && dado.Tipo == "pedido_pago")
            {
                // Deveria ser logicamente impossivel com a ordem preservada:
                // pago depois de cancelado. Se isto disparar, o problema nao
                // e o consumidor — e a garantia de shard sendo violada em
                // algum ponto anterior (ex.: reprocesso escrevendo no
                // caminho de tempo real por engano).
                contexto.Logger.LogWarning(
                    $"ordem suspeita no pedido {dado.IdPedido}: pago apos cancelado");
            }

            if (anterior == "pedido_pago" && dado.Tipo == "pedido_cancelado")
            {
                await _sns.PublishAsync(_topicoArn,
                    $"pedido {dado.IdPedido}: cancelado {SegundosDesde(anterior)}s apos o pagamento — revisar estorno");
            }

            UltimoEventoVisto[dado.IdPedido] = dado.Tipo;
        }
    }

    private static long SegundosDesde(string _) => 0; // ilustrativo: o calculo real usa o instante do evento anterior
}

public sealed record EventoPedidoLido(Guid IdPedido, string Tipo, DateTimeOffset Instante, decimal Valor);

# provas.sh — cinco medicoes; nenhuma conclusao vem de "parece que chegou"

PROJETO=ffv-lab-kinesis
STREAM="${PROJETO}-eventos-pedido"

# ── Prova 1: o hot shard, com numero (ver medir-shard-quente.sh acima) ───────

# ── Prova 2: PutRecords tem falha parcial mesmo devolvendo 200 ───────────────
# Gera trafego concentrado numa unica chave, o suficiente para passar de
# 1.000 registros/s NUM shard, e confere o corpo da resposta.
python3 - <<'PY'
import boto3, json, time
k = boto3.client("kinesis")
lote = [{"Data": json.dumps({"i": i}).encode(), "PartitionKey": "pedido_criado"}
        for i in range(500)]
falhas_totais = 0
for _ in range(6):  # 6 x 500 = 3.000 registros contra um shard de 1.000/s
    r = k.put_records(StreamName="ffv-lab-kinesis-eventos-pedido", Records=lote)
    falhas_totais += r["FailedRecordCount"]
    print("HTTP 200, FailedRecordCount =", r["FailedRecordCount"])
print("total de falhas parciais:", falhas_totais)
PY
# Esperado: pelo menos uma das seis chamadas mostra FailedRecordCount > 0,
# e TODAS retornam sem lancar excecao — a falha mora no corpo, nao no status.

# ── Prova 3: os eventos do MESMO pedido chegam no MESMO shard ────────────────
aws kinesis list-shards --stream-name "$STREAM" \
  --query 'Shards[].{Id:ShardId,Faixa:HashKeyRange}' --output table
# Compare o ShardId reportado nos logs da Lambda para dois eventos do mesmo
# IdPedido: com chave = id do pedido, tem de ser o MESMO ShardId sempre.

# ── Prova 4: a retenção realmente guarda o dado além de 24h ──────────────────
aws kinesis describe-stream-summary --stream-name "$STREAM" \
  --query 'StreamDescriptionSummary.RetentionPeriodHours' --output text
# Esperado: 168 (7 dias). Se voltar 24, a mudança de retenção não foi aplicada
# e o reprocesso da prova 5 vai falhar por falta de dado.

# ── Prova 5: reprocesso a partir de TRIM_HORIZON encontra o evento antigo ────
SHARD=$(aws kinesis list-shards --stream-name "$STREAM" \
  --query 'Shards[0].ShardId' --output text)
ITERADOR=$(aws kinesis get-shard-iterator --stream-name "$STREAM" \
  --shard-id "$SHARD" --shard-iterator-type TRIM_HORIZON \
  --query 'ShardIterator' --output text)
aws kinesis get-records --shard-iterator "$ITERADOR" \
  --query 'Records[0].{Sequencia:SequenceNumber,Chave:PartitionKey}' --output table
# Esperado: um registro com SequenceNumber baixo (o mais antigo ainda retido),
# nao o mais recente — confirma que TRIM_HORIZON aponta para o inicio, e nao
# para "agora", que seria o comportamento de LATEST.

#!/usr/bin/env bash
# reprocessar.sh — depois de corrigir o bug, relê o stream inteiro sem tocar o produtor
set -euo pipefail

PROJETO="${PROJETO:?defina PROJETO}"
STREAM="${PROJETO}-eventos-pedido"
FUNCAO_REPLAY="${PROJETO}-reprocessamento"

# 1. Cria um mapping TEMPORARIO, separado do consumo em tempo real, que ja
#    continua rodando com starting_position=LATEST sem interrupção.
UUID=$(aws lambda create-event-source-mapping \
  --function-name "$FUNCAO_REPLAY" \
  --event-source-arn "arn:aws:kinesis:us-east-1:$(aws sts get-caller-identity --query Account --output text):stream/$STREAM" \
  --starting-position TRIM_HORIZON \
  --batch-size 200 \
  --query 'UUID' --output text)
echo "mapping de reprocesso criado: $UUID"

# 2. Acompanha o IteratorAge caindo: quando chegar perto de zero, o replay
#    alcançou o tempo real e pode ser desligado.
echo "acompanhe com:"
echo "  aws lambda get-event-source-mapping --uuid $UUID --query 'LastProcessingResult'"
echo "  aws cloudwatch get-metric-statistics --namespace AWS/Lambda --metric-name IteratorAge \"
echo "    --dimensions Name=FunctionName,Value=$FUNCAO_REPLAY --statistics Maximum --period 60 \"
echo "    --start-time \$(date -u -d '10 minutes ago' +%Y-%m-%dT%H:%M:%S) --end-time \$(date -u +%Y-%m-%dT%H:%M:%S)"

read -rp "IteratorAge proximo de zero? apagar o mapping de reprocesso [s/N]: " ok
if [[ "$ok" == "s" ]]; then
  aws lambda delete-event-source-mapping --uuid "$UUID"
  echo "mapping de reprocesso removido — o replay foi temporario, como deveria ser"
fi

#!/usr/bin/env bash
# limpar.sh — o que o destroy não leva
set -euo pipefail

PROJETO=ffv-lab-kinesis

# 1. Derrube o que o Terraform administra.
terraform destroy -auto-approve

# 2. Se houve um mapping de reprocesso e ele nao foi apagado no fim do
#    replay, ele continua lendo e pagando invocacao de Lambda para sempre.
aws lambda list-event-source-mappings --function-name "${PROJETO}-reprocessamento" \
  --query 'EventSourceMappings[].UUID' --output text | \
  xargs -n1 -I{} aws lambda delete-event-source-mapping --uuid {} 2>/dev/null || true

# 3. FILA DE FALHA: guarda ate 14 dias de mensagens mesmo depois do stream
#    sumir. Se houver algo pendente de investigar, salve antes de apagar.
aws sqs receive-message --queue-url "$(aws sqs get-queue-url --queue-name ${PROJETO}-eventos-pedido-dlq --query QueueUrl --output text)" \
  --max-number-of-messages 10 --query 'Messages[].Body' --output json 2>/dev/null || true

# 4. GRUPOS DE LOGS da Lambda: tem ciclo proprio, nao cai com a funcao.
aws logs delete-log-group --log-group-name "/aws/lambda/${PROJETO}-alerta-fraude" 2>/dev/null || true
aws logs delete-log-group --log-group-name "/aws/lambda/${PROJETO}-reprocessamento" 2>/dev/null || true

# 5. Prova final: nada com a tag do projeto de pe.
aws resourcegroupstaggingapi get-resources \
  --tag-filters Key=Projeto,Values=${PROJETO} \
  --query "ResourceTagMappingList[].ResourceARN" --output table