Aller au contenu principal

Ingérer des événements via l'API Streaming

Introduction​

Utilisez cette API pour envoyer des données d'événements dans un pipeline à des fins de traitement, de filtrage, d'enrichissement et de routage.

Ce point de terminaison prend en charge l'ingestion en temps réel à haut débit et les événements JSON structurés et semi-structurés. Il est idéal pour les applications cloud, les microservices, les appareils IoT ou tout système émettant des données télémétriques.

Fonctions de l'API​

  • Envoyez les journaux et les métriques directement dans un pipeline
  • Normalisez et enrichissez les données avant de les envoyer en aval
  • Réduisez les coûts en filtrant ou en échantillonnant tôt
  • Créez des pipelines flexibles sans dépendance vis-à-vis d'un fournisseur

Avant de commencer​

Assurez-vous de disposer des éléments suivants :

  • Un jeton API avec des autorisations d'ingestion
  • Un identifiant de flux
  • Des charges utiles d'événements au format JSON
  • curl ou Postman installé

Présentation du pipeline​

flowchart LR
    A["Application cloud / Microservice"] --> B["Streaming Ingestion API"]
    B --> C["Couche de traitement<br/> • Analyse<br/> • Enrichissement<br/> • Masquage"]
    C --> D["Moteur de routage<br/> • Filtres<br/> • Échantillonnage<br/> • Règles"]
    D --> E{{"Destinations :<br/>S3 / SIEM / Elastic / Snowflake"}}

Point de terminaison d'ingestion​

POST /v1/streams/ingest

En-têtes de requête​

En-têteValeurObligatoireDescription
AuthorizationBearer <token>OuiJeton API
Content-Typeapplication/jsonOuiCharge utile JSON
X-Stream-Id<your-stream-id>OuiPipeline cible

Corps de la requête​

{
  "timestamp": "2025-02-12T14:32:15Z",
  "source": "payments-app",
  "event": {
    "transactionId": "txn_30921",
    "status": "approved",
    "latency_ms": 182,
    "amount": 499,
    "currency": "USD"
  },
  "metadata": {
    "env": "prod",
    "team": "payments"
  }
}

Remarques​

  • timestamp : format ISO 8601
  • event : Données d'événement principales (paires clé-valeur)
  • Taille maximale de la charge utile : 1 MB

Réponse​

200 OK​

{
  "status": "accepted",
  "ingested_bytes": 527,
  "stream_id": "a0c3b4de-1923-4ff1-8dba-f92ad912d20f"
}

400 Bad request​

Problèmes courants :

  • JSON non valide
  • Format d'horodatage incorrect
  • Champs source ou événement manquants
  • Charge utile dépassant la taille maximale
{
  "error": "invalid_format",
  "detail": "The 'timestamp' field must be a valid ISO 8601 string."
}

401 Unauthorized​

  • Jeton manquant ou expiré.

429 Rate limited​

  • Réduisez le débit des requêtes ou regroupez les événements.

Exemple cURL​

curl -X POST "https://api.example.com/v1/streams/ingest" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -H "X-Stream-Id: $STREAM_ID" \
  -d @event.json

Cas d'utilisation courants​

  • Ingestion de journaux à partir de microservices
  • Diffusion de données télémétriques à partir d'appareils IoT
  • Normalisation des événements provenant de systèmes distribués
  • Routage des événements enrichis vers SIEM, S3, Elastic ou Snowflake
  • Pipelines de détection des fraudes ou des anomalies en temps réel

Dépannage​

SymptômeCause probableSolution
Aucun événement n'apparaît dans le pipelineID de flux incorrectCopiez l'ID de flux correct
Charge utile rejetéeJSON invalideValidez à l'aide de jq
Champs manquantsIncohérence de la structure imbriquéeVérifiez les règles d'analyse
Latence élevéeCharges utiles ou lots volumineuxRéduisez la taille des lots

Étapes suivantes​

  • Ajoutez des règles d'analyse, de masquage ou d'enrichissement
  • Acheminez les données vers plusieurs destinations
  • Réduisez les coûts grâce à l'échantillonnage ou au filtrage
  • Créez des métriques et des tableaux de bord pour le pipeline