Menggunakan pipa OpenSearch Ingestion dengan AWS Lambda

Anda dapat menggunakan saluran OpenSearch Ingestion AWS Lambda untuk pengayaan kustom data Anda dari semua sumber dan tujuan yang didukung oleh Ingestion. OpenSearch Lambda sebagai prosesor dapat memperkaya data menggunakan kode kustom dan mengembalikan peristiwa yang diproses kembali ke pipeline integrasi Anda untuk diproses lebih lanjut.

Prasyarat

Sebelum Anda membuat pipeline OpenSearch Ingestion, lakukan langkah-langkah berikut:

1. Buat AWS Lambda fungsi. Untuk melakukan ini, lihat Aliran data AWS Lambda dokumentasi bertindak sebagai sumber. Aliran harus berisi data yang ingin Anda konsumsi ke OpenSearch Layanan.

2. Buat domain OpenSearch Layanan atau koleksi OpenSearch Tanpa Server tempat Anda ingin memigrasikan data. Untuk informasi selengkapnya, lihat Membuat domain OpenSearch Layanan dan Membuat koleksi.

3. Siapkan peran pipeline dengan izin untuk menulis ke domain atau koleksi tujuan. Untuk melakukan ini, lihat peran pipeline.

4. Agar pipeline dapat menulis data ke domain, domain harus memiliki kebijakan akses tingkat domain yang memungkinkan peran pipeline sts_role_arn untuk mengaksesnya. Anda dapat merujuk ke Memberikan akses saluran pipa Amazon OpenSearch Ingestion ke domain untuk memberikan izin domain. Anda dapat merujuk ke Memberikan akses Amazon OpenSearch Ingestion ke koleksi untuk memberikan izin pengumpulan.

   {
       "Version": "2012-10-17",
       "Statement": [
           {
               "Sid": "allowinvokeFunction",
               "Effect": "Allow",
               "Action": [
                   "lambda:invokeFunction",
                   "lambda:InvokeAsync",
                   "lambda:ListFunctions"
               ],
               "Resource": "arn:aws:lambda:{{region}}:{{account-id}}:function:{{function-name}}"
               
           }
       ]
   }

   Untuk membuat peran IAM dengan izin yang benar untuk mengakses data tulis ke koleksi atau domain, lihat Izin yang diperlukan untuk domain dan Izin yang diperlukan untuk koleksi.

   catatan

   Batas ukuran muatan untuk prosesor Lambda atau sink untuk satu acara adalah 5MB. Selain itu, prosesor Lambda hanya mendukung respons dalam format array JSON.

Buat pipeline

Untuk digunakan AWS Lambda sebagai prosesor, Anda harus terlebih dahulu mengkonfigurasi pipeline OpenSearch Ingestion dan menentukan AWS Lambda sebagai prosesor untuk pipeline Anda. Anda juga dapat menggunakan cetak biru prosesor Lambda yang telah dikonfigurasi sebelumnya untuk membuat pipeline ini. Untuk informasi selengkapnya, lihat Menggunakan cetak biru untuk membuat pipeline. Contoh pipeline berikut menerima data dari sumber http, memperkaya data menggunakan prosesor tanggal dan AWS Lambda prosesor, dan menyerap data yang diproses ke Domain. OpenSearch

Untuk mengatur AWS Lambda sebagai prosesor, lihat contoh pipeline berikut:

version: "2"
lambda-processor-pipeline:
  source:
    http:
      path: "/${pipelineName}/logs"
  
  processor:
      - date:
        destination: "@timestamp"
        from_time_received: true

    - aws_lambda:
        function_name: "my-lambda-function"
        
        tags_on_failure: ["lambda_failure"]
        batch:
            key_name: "events"
        aws:
          region: us-east-1
          sts_role_arn: "<<arn:aws:iam::123456789012:role/Example-Role>>"
  sink:
    - opensearch:
        hosts: [ "<<https://search-mydomain-1a2a3a4a5a6a7a8a9a0a9a8a7a.us-east-1.es.amazonaws.com>>" ]
        index: "table-index"
        aws:
          sts_role_arn: "<<arn:aws:iam::123456789012:role/Example-Role>>"
          region: "<<us-east-1>>"
          serverless: false

Contoh berikut adalah AWS Lambda fungsi:

import json

def lambda_handler(event, context):
    input_array = event.get('events', [])
    output = []
    for input in input_arr:
        input["transformed"] = "true";
        output.append(input)
   
    return output

Batching dalam prosesor AWS Lambda

OpenSearch Saluran pipa konsumsi akan mengirimkan acara batch ke prosesor Lambda Anda. OpenSearch Saluran pipa konsumsi akan secara dinamis menentukan ukuran batch dan membatasi ukuran batch hingga di bawah 5MB.

Berikut ini adalah contoh batch pipeline:

batch:
    key_name: "events"
    
input_arrary = event.get('events', [])

catatan

Saat membuat pipeline OpenSearch Ingestion, pastikan key_name dalam konfigurasi lambda pipeline cocok dengan kunci peristiwa di penangan Lambda.

Penyaringan Bersyarat dalam prosesor AWS Lambda

Pemfilteran bersyarat memungkinkan Anda mengontrol kapan AWS Lambda prosesor Anda memanggil fungsi Lambda berdasarkan kondisi tertentu dalam data peristiwa. Ini sangat berguna ketika Anda ingin secara selektif memproses jenis peristiwa tertentu sambil mengabaikan yang lain.

Untuk melakukan fitlering bersyarat, lihat contoh berikut:

processors:
  - aws_lambda:
      function_name: "my-lambda-function"
      aws:
        region: "us-east-1"
        sts_role_arn: "arn:aws:iam::123456789012:role/my-lambda-role"
      lambda_when: "/sourceIp == 10.10.10.10"

Metrik dan Pemantauan

Prosesor Lambda menawarkan dukungan untuk beberapa CloudWatch metrik yang dapat digunakan untuk memantau kinerja saluran pipa OpenSearch Ingestion dan prosesor Lambda Anda. Metrik ini dapat membantu Anda mengidentifikasi jumlah data yang dikirim dan diproses oleh prosesor Lambda Anda. Anda dapat mengonfigurasi CloudWatch alarm untuk melakukan tindakan ketika salah satu metrik ini melebihi nilai yang ditentukan. Berikut ini adalah metrik prosesor yang dapat digunakan untuk memantau kinerja dan kesehatan integrasi Lambda Anda:

• recordsSuccessfullySentToLambda: Penghitung untuk acara yang berhasil diproses

• recordsFailedToSendToLambda: Penghitung untuk acara yang gagal

• lambdafunctionlatency: Timer untuk latensi pemanggilan Lambda

• requestPayloadSize: Gauge untuk ukuran payload permintaan

• numberOfRequestsSucceeded: Jumlah permintaan ke Lambda yang berhasil

• numberOfRequestsFailed: Jumlah permintaan ke Lambda yang gagal

• requestPayloadSize: Lambda meminta ukuran muatan

• responsePayloadSize: Ukuran payload respons Lambda