Panoramica delle funzionalità della pipeline in Amazon Ingestion OpenSearch

Amazon OpenSearch Ingestion fornisce pipeline costituite da una fonte, un buffer, zero o più processori e uno o più sink. Le pipeline di ingestione sono alimentate da Data Prepper come motore di dati. Per una panoramica dei vari componenti di una pipeline, vedere. Concetti chiave

Le seguenti sezioni forniscono una panoramica di alcune delle funzionalità più comunemente utilizzate in Amazon OpenSearch Ingestion.

Nota

Questo non è un elenco esaustivo delle funzionalità disponibili per le pipeline. Per una documentazione completa di tutte le funzionalità della pipeline disponibili, consulta la documentazione di Data Prepper. Tieni presente che OpenSearch Ingestion impone alcuni vincoli ai plugin e alle opzioni che puoi utilizzare. Per ulteriori informazioni, consulta Plugin e opzioni supportati per le pipeline di Amazon OpenSearch Ingestion.

Buffering persistente

Un buffer persistente archivia i dati in un buffer basato su disco su più zone di disponibilità per aggiungere durabilità ai dati. È possibile utilizzare il buffering persistente per importare dati da tutte le fonti basate su push supportate senza la necessità di configurare un buffer autonomo. Queste includono OpenTelemetry fonti per log, tracce HTTP e metriche.

Per abilitare il buffering persistente, scegliete Abilita buffer persistente durante la creazione o l'aggiornamento di una pipeline. Per ulteriori informazioni, vedere. Creazione di pipeline Amazon OpenSearch Ingestion OpenSearch Ingestion determina automaticamente la capacità di buffering richiesta in base alle Ingestion OpenSearch Compute Units (IngestionOCUs) specificate per la pipeline.

Se abilitate il buffering persistente per una pipeline, la dimensione massima del payload della richiesta è di 1 MB per le sorgenti e 4 MB per le fonti. HTTP OpenTelemetry Se non abiliti il buffering persistente, la dimensione massima del payload è di 20 MB. Non è possibile modificare il payload massimo della richiesta.

Per impostazione predefinita, le pipeline utilizzano an Chiave di proprietà di AWS per crittografare i dati del buffer. Queste pipeline non necessitano di autorizzazioni aggiuntive per il ruolo della pipeline. In alternativa, puoi specificare una chiave gestita dal cliente e aggiungere le seguenti IAM autorizzazioni al ruolo pipeline:

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "KeyAccess",
            "Effect": "Allow",
            "Action": [
              "kms:Decrypt",
              "kms:GenerateDataKeyWithoutPlaintext"
            ],
            "Resource": "arn:aws:kms:{region}:{aws-account-id}:key/1234abcd-12ab-34cd-56ef-1234567890ab"
        }
    ]
}

Per ulteriori informazioni, consulta Customer managed keys nella Guida per sviluppatori AWS Key Management Service .

Nota

Se disabiliti il buffering persistente, la pipeline verrà aggiornata per essere eseguita interamente con il buffering in memoria.

Divisione

È possibile configurare una pipeline di OpenSearch ingestione per suddividere gli eventi in entrata in una sottopipeline, in modo da eseguire diversi tipi di elaborazione sullo stesso evento in entrata.

La pipeline di esempio seguente divide gli eventi in entrata in due sotto-pipeline. Ogni sottopipeline utilizza il proprio processore per arricchire e manipolare i dati, quindi invia i dati a indici diversi. OpenSearch

version: "2"
log-pipeline:
  source:
    http:
    ...
  sink:
    - pipeline:
        name: "logs_enriched_one_pipeline"
    - pipeline:
        name: "logs_enriched_two_pipeline"

logs_enriched_one_pipeline:
  source:
    log-pipeline
  processor:
   ...
  sink:
    - opensearch:
        # Provide a domain or collection endpoint
        # Enable the 'serverless' flag if the sink is an OpenSearch Serverless collection
        aws:
          ...
        index: "enriched_one_logs"

logs_enriched_two_pipeline:
  source:
    log-pipeline
  processor:
   ...
  sink:
    - opensearch:
        # Provide a domain or collection endpoint
        # Enable the 'serverless' flag if the sink is an OpenSearch Serverless collection
        aws:
          ...
          index: "enriched_two_logs"

Concatenamento

È possibile concatenare più sotto-pipeline per eseguire l'elaborazione e l'arricchimento dei dati in blocchi. In altre parole, è possibile arricchire un evento in entrata con determinate funzionalità di elaborazione in una sottopipeline, quindi inviarlo a un'altra sottopipeline per un ulteriore arricchimento con un processore diverso e infine inviarlo al relativo sink. OpenSearch

Nell'esempio seguente, la log_pipeline sub-pipeline arricchisce un evento di registro in entrata con un set di processori, quindi invia l'evento a un indice denominato. OpenSearch enriched_logs La pipeline invia lo stesso evento alla log_advanced_pipeline pipeline secondaria, che lo elabora e lo invia a un indice diverso denominato. OpenSearch enriched_advanced_logs

version: "2"
log-pipeline:
  source:
    http:
    ...
  processor:
    ...
  sink:
    - opensearch:
        # Provide a domain or collection endpoint
        # Enable the 'serverless' flag if the sink is an OpenSearch Serverless collection
        aws:
          ...
          index: "enriched_logs"
    - pipeline:
        name: "log_advanced_pipeline"

log_advanced_pipeline:
  source:
    log-pipeline
  processor:
   ...
  sink:
    - opensearch:
        # Provide a domain or collection endpoint
        # Enable the 'serverless' flag if the sink is an OpenSearch Serverless collection
        aws:
          ...
          index: "enriched_advanced_logs"

Code DLQ

Dead-letter queues (DLQs) sono destinazioni per eventi che una pipeline non riesce a scrivere in un sink. In OpenSearch Ingestion, è necessario specificare un bucket Amazon S3 con autorizzazioni di scrittura appropriate da utilizzare come. DLQ Puoi aggiungere una DLQ configurazione a ogni sink all'interno di una pipeline. Quando una pipeline incontra errori di scrittura, crea DLQ oggetti nel bucket S3 configurato. DLQgli oggetti esistono all'interno di un JSON file come matrice di eventi non riusciti.

Una pipeline scrive eventi su DLQ quando viene soddisfatta una delle seguenti condizioni:

• I max_retries rubinetti per il OpenSearch lavandino sono esauriti. OpenSearch L'ingestione richiede un minimo di 16 per questa opzione.

• Gli eventi vengono rifiutati dal sink a causa di una condizione di errore.

Configurazione

Per configurare una coda di lettere non scritte per una pipeline secondaria, specifica l'dlqopzione all'interno della configurazione sink: opensearch

apache-log-pipeline:
  ...
  sink:
    opensearch:
      dlq:
        s3:
          bucket: "my-dlq-bucket"
          key_path_prefix: "dlq-files"
          region: "us-west-2"
          sts_role_arn: "arn:aws:iam::123456789012:role/dlq-role"

I file scritti su questo S3 DLQ avranno il seguente schema di denominazione:

dlq-v${version}-${pipelineName}-${pluginId}-${timestampIso8601}-${uniqueId}

Per ulteriori informazioni, vedete Dead-Letter Queues (). DLQ

Per istruzioni su come configurare il sts_role_arn ruolo, vedere. Scrittura in una coda di lettere non scritte

Esempio

Considerate il seguente DLQ file di esempio:

dlq-v2-apache-log-pipeline-opensearch-2023-04-05T15:26:19.152938Z-e7eb675a-f558-4048-8566-dac15a4f8343

Ecco un esempio di dati che non sono stati scritti nel sink e che vengono inviati al bucket DLQ S3 per ulteriori analisi:

Record_0	
pluginId            "opensearch"
pluginName          "opensearch"
pipelineName        "apache-log-pipeline"
failedData	
index		  "logs"
indexId		 null
status		  0
message		"Number of retries reached the limit of max retries (configured value 15)"
document	
log		    "sample log"
timestamp	    "2023-04-14T10:36:01.070Z"

Record_1	
pluginId            "opensearch"
pluginName          "opensearch"
pipelineName        "apache-log-pipeline"
failedData	
index               "logs"
indexId		 null
status		  0
message		"Number of retries reached the limit of max retries (configured value 15)"
document	
log                 "another sample log"
timestamp           "2023-04-14T10:36:01.071Z"

Gestione degli indici

Amazon OpenSearch Ingestion offre molte funzionalità di gestione degli indici, tra cui le seguenti.

Creazione di indici

È possibile specificare un nome di indice in un sink di pipeline e OpenSearch Ingestion crea l'indice quando effettua il provisioning della pipeline. Se esiste già un indice, la pipeline lo utilizza per indicizzare gli eventi in entrata. Se si arresta e si riavvia una pipeline o se si aggiorna la sua YAML configurazione, la pipeline tenta di creare nuovi indici se non esistono già. Una pipeline non può mai eliminare un indice.

I sinks di esempio seguenti creano due indici quando viene eseguito il provisioning della pipeline:

sink:
  - opensearch:
      index: apache_logs
  - opensearch:
      index: nginx_logs

Generazione di nomi e modelli di indici

È possibile generare nomi di indice dinamici utilizzando variabili dai campi degli eventi in arrivo. Nella configurazione sink, utilizzate il formato string${} per segnalare l'interpolazione delle stringhe e utilizzate un JSON puntatore per estrarre i campi dagli eventi. Le opzioni per index_type sono o. custom management_disabled Poiché l'index_typeimpostazione predefinita è per i OpenSearch domini e custom management_disabled per le raccolte OpenSearch Serverless, può essere lasciata non impostata.

Ad esempio, la seguente pipeline seleziona il metadataType campo dagli eventi in entrata per generare i nomi degli indici.

pipeline:
  ...
  sink:
    opensearch:
      index: "metadata-${metadataType}"

La seguente configurazione continua a generare un nuovo indice ogni giorno o ogni ora.

pipeline:
  ...
  sink:
    opensearch:
      index: "metadata-${metadataType}-%{yyyy.MM.dd}"

pipeline:
  ...
  sink:
    opensearch:
      index: "metadata-${metadataType}-%{yyyy.MM.dd.HH}"

Il nome dell'indice può anche essere una stringa semplice con un modello data-ora come suffisso, ad esempio. my-index-%{yyyy.MM.dd} Quando il sink invia dati a OpenSearch, sostituisce il modello data-ora con l'UTCora e crea un nuovo indice per ogni giorno, ad esempio. my-index-2022.01.25 Per ulteriori informazioni, consultate la classe. DateTimeFormatter

Questo nome di indice può anche essere una stringa formattata (con o senza un suffisso del modello data-ora), ad esempio. my-${index}-name Quando il sink invia dati a OpenSearch, sostituisce la "${index}" parte con il valore dell'evento in fase di elaborazione. Se il formato è"${index1/index2/index3}", sostituisce il campo index1/index2/index3 con il relativo valore nell'evento.

Generazione di un documento IDs

Una pipeline può generare un ID di documento durante l'indicizzazione dei documenti su. OpenSearch Può dedurre questi documenti IDs dai campi all'interno degli eventi in arrivo.

Questo esempio utilizza il uuid campo di un evento in entrata per generare un ID del documento.

pipeline:
  ...
  sink:
    opensearch:
      index_type: custom
      index: "metadata-${metadataType}-%{yyyy.MM.dd}" 
      document_id_field: "uuid"   

Nell'esempio seguente, il processore Add entries unisce i campi uuid e other_field l'evento in entrata per generare un ID del documento.

L'createazione garantisce che i documenti identici IDs non vengano sovrascritti. La pipeline elimina i documenti duplicati senza alcun nuovo tentativo o evento. DLQ Si tratta di un'aspettativa ragionevole per gli autori della pipeline che utilizzano questa azione, poiché l'obiettivo è evitare l'aggiornamento dei documenti esistenti.

pipeline:
  ...
  processor:
   - add_entries:
      entries:
        - key: "my_doc_id_field"
          format: "${uuid}-${other_field}"
  sink:
    - opensearch:
       ...
       action: "create"
       document_id_field: "my_doc_id_field"

Potresti voler impostare l'ID del documento di un evento su un campo di un oggetto secondario. Nell'esempio seguente, il plugin OpenSearch sink utilizza l'oggetto secondario info/id per generare un ID del documento.

sink:
  - opensearch:
       ...
       document_id_field: info/id

Dato il seguente evento, la pipeline genererà un documento con il _id campo impostato su: json001

{
   "fieldA":"arbitrary value",
   "info":{
      "id":"json001",
      "fieldA":"xyz",
      "fieldB":"def"
   }
}

Generazione del routing IDs

È possibile utilizzare l'routing_fieldopzione all'interno del plug-in OpenSearch sink per impostare il valore di una proprietà di routing del documento (_routing) su un valore proveniente da un evento in arrivo.

Il routing supporta la sintassi dei JSON puntatori, quindi sono disponibili anche campi annidati, non solo campi di primo livello.

sink:
  - opensearch:
       ...
       routing_field: metadata/id
       document_id_field: id

Dato il seguente evento, il plugin genera un documento con il campo impostato su: _routing abcd

{
   "id":"123",
   "metadata":{
      "id":"abcd",
      "fieldA":"valueA"
   },
   "fieldB":"valueB"
}

Per istruzioni su come creare modelli di indice che le pipeline possono utilizzare durante la creazione dell'indice, vedete Modelli di indice.

End-to-end ) Riconoscimento

OpenSearch Ingestion garantisce la durabilità e l'affidabilità dei dati tracciandone la trasmissione dalla fonte ai pozzi nelle condutture stateless mediante riconoscimento. end-to-end Attualmente, solo il plug-in sorgente S3 supporta il riconoscimento. end-to-end

Con il end-to-end riconoscimento, il plug-in pipeline source crea un set di riconoscimenti per monitorare un batch di eventi. Riceve un riconoscimento positivo quando tali eventi vengono inviati con successo ai rispettivi sink o un riconoscimento negativo quando nessuno degli eventi non può essere inviato ai rispettivi sink.

In caso di guasto o arresto anomalo di un componente della pipeline, o se una fonte non riceve una conferma, la fonte scade e intraprende le azioni necessarie, come riprovare o registrare l'errore. Se nella pipeline sono configurati più sink o più subpipeline, i riconoscimenti a livello di evento vengono inviati solo dopo che l'evento è stato inviato a tutti i sink di tutte le pipeline secondarie. Se un sink ha un file configurato, i riconoscimenti tengono traccia anche degli eventi scritti su. DLQ end-to-end DLQ

Per abilitare la end-to-end conferma, includi l'acknowledgmentsopzione nella configurazione di origine:

s3-pipeline:
  source:
    s3:
      acknowledgments: true
...

Contropressione alla fonte

Una pipeline può subire una contropressione quando è impegnata nell'elaborazione dei dati o se i suoi sink sono temporaneamente inattivi o rallentano l'acquisizione dei dati. OpenSearch Ingestion ha diversi modi di gestire la contropressione a seconda del plug-in di origine utilizzato da una pipeline.

HTTPfonte

Le pipeline che utilizzano il plugin HTTPsource gestiscono la contropressione in modo diverso a seconda del componente della pipeline che è congestionato:

• Buffer: quando i buffer sono pieni, la pipeline inizia a restituire HTTP lo stato REQUEST_TIMEOUT con il codice di errore 408 all'endpoint di origine. Quando i buffer vengono liberati, la pipeline riavvia l'elaborazione degli eventi. HTTP

• Thread di origine: quando tutti i thread di HTTP origine sono impegnati nell'esecuzione di richieste e la dimensione della coda delle richieste non elaborate ha superato il numero massimo consentito di richieste, la pipeline inizia a restituire HTTP lo stato TOO_MANY_REQUESTS con il codice di errore 429 all'endpoint di origine. Quando la coda delle richieste scende al di sotto della dimensione massima consentita, la pipeline riavvia l'elaborazione delle richieste.

OTelfonte

Quando i buffer sono pieni per le pipeline che utilizzano OpenTelemetry sorgenti (OTellog, OTelmetriche e OTeltrace), la pipeline inizia a restituire HTTP lo stato REQUEST_TIMEOUT con il codice di errore 408 all'endpoint di origine. Quando i buffer vengono liberati, la pipeline riavvia l'elaborazione degli eventi.

Fonte S3

Quando i buffer sono pieni per le pipeline con una sorgente S3, le pipeline interrompono l'elaborazione delle notifiche. SQS Man mano che i buffer vengono liberati, le pipeline riprendono a elaborare le notifiche.

Se un sink è inattivo o non è in grado di inserire dati e la end-to-end conferma è abilitata per la fonte, la pipeline interrompe l'elaborazione delle SQS notifiche finché non riceve una conferma corretta da tutti i sink.