Información general

Utiliza el destino Kafka de Observability Pipelines para enviar logs a temas Kafka.

Cuándo utilizar este destino

Escenarios frecuentes en los que puedes utilizar este destino:

• Para enrutar logs a los siguientes destinos:
  • Clickhouse: Sistema abierto de gestión de bases de datos orientado a columnas utilizado para analizar grandes volúmenes de logs.
  • Snowflake: Almacén de datos utilizado para el almacenamiento y la consulta.
    • La integración de la API de Snowflake utiliza Kafka como método de ingesta de logs en tu plataforma.
  • Databricks: Lago de datos para el análisis y el almacenamiento.
  • Azure Event Hub: Servicio de ingesta y procesamiento en el ecosistema de Microsoft y Azure.
• Para enrutar datos a Kafka y utilizar el ecosistema Kafka Connect.
• Para procesar y normalizar tus datos con Observability Pipelines antes de enrutarlos a Apache Spark con Kafka para analizar datos y ejecutar cargas de trabajo de machine learning.

Instalación

Configura el destino Kafka y sus variables de entorno cuando configures un pipeline. La siguiente información se configura en la interfaz de usuario del pipeline.

Configurar el destino

1. Introduce el nombre del tema al que quieres enviar logs.
2. En el menú desplegable Encoding (Codificación), selecciona JSON o Raw message como formato de salida.

Ajustes opcionales

Activar TLS

Alterna al interruptor para activar TLS** (Activar TLS). Se requieren los siguientes archivos de certificados y claves.
Nota: Todas las rutas a los archivos son relativas al directorio de datos de configuración, que es /var/lib/observability-pipelines-worker/config/ por defecto. Consulta Configuraciones avanzadas del worker para obtener más información. El archivo debe ser propiedad del usuario observability-pipelines-worker group y observability-pipelines-worker o al menos debe ser legible por el grupo o el usuario.

• Server Certificate Path: La ruta al archivo del certificado que ha sido firmado por el archivo raíz de tu autoridad de certificación (CA) en DER o PEM (X.509).
• CA Certificate Path: La ruta al archivo del certificado que es el archivo raíz de tu autoridad de certificación (CA) en DER o PEM (X.509).
• Private Key Path: la ruta al archivo de clave privada .key que pertenece a la ruta de tu certificado de servidor en formato DER o PEM (PKCS#8).

Activar la autenticación SASL

1. Alterna el interruptor para activar la Autenticación SASL.
2. Selecciona el mecanismo (PLAIN, SCHRAM-SHA-256 o SCHRAM-SHA-512) en el menú desplegable.

Activar la compresión

1. Alterna a la opción Enable Compression (Activar compresión).
2. En el menú desplegable Compression Algorithm (Algoritmo de compresión), selecciona un algoritmo de compresión (gzip, zstd, lz4 o snappy).
3. (Opcional) Selecciona un nivel de compresión en el menú desplegable. Si no se especifica el nivel, se utiliza el nivel por defecto del algoritmo.

Almacenamiento en buffer

Toggle the switch to enable Buffering Options. Enable a configurable buffer on your destination to ensure intermittent latency or an outage at the destination doesn’t create immediate backpressure, and allow events to continue to be ingested from your source. Disk buffers can also increase pipeline durability by writing data to disk, ensuring buffered data persists through a Worker restart. See Destination buffers for more information.

• If left unconfigured, your destination uses a memory buffer with a capacity of 500 events.
• To configure a buffer on your destination:
  1. Select the buffer type you want to set (Memory or Disk).
  2. Enter the buffer size and select the unit.
     1. Maximum memory buffer size is 128 GB.
     2. Maximum disk buffer size is 500 GB.
  3. In the Behavior on full buffer dropdown menu, select whether you want to block events or drop new events when the buffer is full.

Opciones avanzadas

Haz clic en Advanced (Avanzado) si quieres configurar alguno de los siguientes campos:

1. Campo clave del mensaje: Especifica qué campo de log contiene la clave del mensaje para particionar, agrupar y ordenar.
2. Clave de cabecera: Especifica qué campo de log contiene tus cabeceras Kafka. Si se deja en blanco, no se escribe ninguna cabecera.
3. Tiempo de espera del mensaje (ms): Tiempo de espera del mensaje local en milisegundos. Por defecto es 300,000 ms.
4. Tiempo de espera del socket (ms): Tiempo de espera por defecto en milisegundos para solicitudes de red. Por defecto es 60,000 ms.
5. Eventos de límite de frecuencia: Número máximo de solicitudes que el cliente Kafka puede enviar dentro de la ventana de tiempo de límite de frecuencia. Por defecto no hay límite de frecuencia.
6. Ventana de tiempo de límite de frecuencia (segundos): Ventana de tiempo utilizada para la opción de límite de frecuencia.
   • Este ajuste no tiene ningún efecto si no se define el límite de frecuencia para eventos.
   • El valor por defecto es 1 second si Rate Limit Events (Eventos de límite de frecuencia) está configurado, pero Rate Limit Time Window (Ventana de tiempo límite de frecuencia) no está configurado.
7. Para añadir opciones librdkafka adicionales, haz clic en Add Option (Añadir opción) y selecciona una opción en el menú desplegable.
   1. Introduce un valor para esa opción.
   2. Comprueba tus valores en la documentación de librdkafka para asegurarte de que tienen el tipo correcto y están dentro del rango establecido.
   3. Haz clic en Add Option (Añadir opción) para añadir otra opción librdkafka.

Configurar secretos

These are the defaults used for secret identifiers and environment variables.

Note: If you enter secret identifiers and then choose to use environment variables, the environment variable is the identifier entered and prepended with DD_OP. For example, if you entered PASSWORD_1 for a password identifier, the environment variable for that password is DD_OP_PASSWORD_1.

• Identificador de servidores de arranque Kafka:
  • Hace referencia al servidor de arranque que el cliente utiliza para conectarse al clúster Kafka y detectar todos los demás hosts del clúster.
  • En tu gestor de secretos, el host y el puerto deben introducirse en el formato host:port, como 10.14.22.123:9092. Si hay más de un servidor, utiliza comas para separarlos.
  • El identificador por defecto es DESTINATION_KAFKA_BOOTSTRAP_SERVERS.
• Identificador de frase de contraseña TLS de Kafka (cuando TLS está activado):
  • El identificador por defecto es DESTINATION_KAFKA_KEY_PASS.
• Autenticación SASL (si está activada):
  • Identificador de nombre de usuario SASL de Kafka:
    • El identificador por defecto es DESTINATION_KAFKA_SASL_USERNAME.
  • Identificador de contraseña SASL de Kafka:
    • El identificador por defecto es DESTINATION_KAFKA_SASL_PASSWORD.

Kafka bootstrap servers

• The host and port of the Kafka bootstrap servers.
• This is the bootstrap server that the client uses to connect to the Kafka cluster and discover all the other hosts in the cluster. The host and port must be entered in the format of host:port, such as 10.14.22.123:9092. If there is more than one server, use commas to separate them.
• The default environment variable is DD_OP_DESTINATION_KAFKA_BOOTSTRAP_SERVERS.

TLS (when enabled)

• If TLS is enabled, the Kafka TLS passphrase is needed.
• The default environment variable is DD_OP_DESTINATION_KAFKA_KEY_PASS.

SASL (when enabled)

• Kafka SASL username
  • The default environment variable is DD_OP_DESTINATION_KAFKA_SASL_USERNAME.
• Kafka SASL password
  • The default environment variable is DD_OP_DESTINATION_KAFKA_SASL_PASSWORD.

Opciones de librdkafka

Estas son las opciones de librdkafka disponibles:

• client.id
• queue.buffering.max_messages
• transactional.id
• enable.idempotence
• acks

Consulta la documentación de librdkafka para obtener más información y asegurarte de que tus valores tienen el tipo correcto y están dentro del rango.

Métricas

Consulte las métricas de Observability Pipelines para ver una lista completa de métricas de salud disponibles.

Métricas de salud de workers

Métricas de componente

Monitor the health of your destination with the following key metrics:

pipelines.component_sent_events_total

Events successfully delivered.

pipelines.component_discarded_events_total

Events dropped.

pipelines.component_errors_total

Errors in the destination component.

pipelines.component_sent_events_bytes_total

Total event bytes sent.

pipelines.utilization

Worker resource usage.

Métricas de buffer (cuando están activadas)

These metrics are specific to destination buffers, located upstream of a destination. Each destination emits its own respective buffer metrics.

• Use the component_id tag to filter or group by individual components.
• Use the component_type tag to filter or group by the destination type, such as datadog_logs for the Datadog Logs destination.

pipelines.buffer_size_events

Description: Number of events in a destination’s buffer.

Metric type: gauge

pipelines.buffer_size_bytes

Description: Number of bytes in a destination’s buffer.

Metric type: gauge

pipelines.buffer_received_events_total

Description: Events received by a destination’s buffer. Note: This metric represents the count per second and not the cumulative total, even though total is in the metric name.

Metric type: counter

pipelines.buffer_received_bytes_total

Description: Bytes received by a destination’s buffer. Note: This metric represents the count per second and not the cumulative total, even though total is in the metric name.

Metric type: counter

pipelines.buffer_sent_events_total

Description: Events sent downstream by a destination’s buffer. Note: This metric represents the count per second and not the cumulative total, even though total is in the metric name.

Metric type: counter

pipelines.buffer_sent_bytes_total

Description: Bytes sent downstream by a destination’s buffer. Note: This metric represents the count per second and not the cumulative total, even though total is in the metric name.

Metric type: counter

pipelines.buffer_discarded_events_total

Description: Events discarded by the buffer. Note: This metric represents the count per second and not the cumulative total, even though total is in the metric name.

Metric type: counter

Additional tags: intentional:true means an incoming event was dropped because the buffer was configured to drop the newest logs when it’s full. intentional:false means the event was dropped due to an error.

pipelines.buffer_discarded_bytes_total

Description: Bytes discarded by the buffer. Note: This metric represents the count per second and not the cumulative total, even though total is in the metric name.

Metric type: counter

Additional tags: intentional:true means an incoming event was dropped because the buffer was configured to drop the newest logs when it’s full. intentional:false means the event was dropped due to an error.

Métricas de buffer obsoletas

These metrics are still emitted by the Observability Pipelines Worker for backwards compatibility. Datadog recommends using the replacements when possible.

pipelines.buffer_events

Description: Number of events in a destination’s buffer. Use pipelines.buffer_size_events instead.

Metric type: gauge

pipelines.buffer_byte_size

Description: Number of bytes in a destination’s buffer. Use pipelines.buffer_size_bytes instead.

Metric type: gauge

Procesamiento de eventos por lotes

Un lote de eventos se descarga cuando se cumple uno de estos parámetros. Consulta el procesamiento de eventos por lotes para obtener más información.