Acceso a datos de telemetría en tiempo real para extensiones mediante la API de telemetría

La API de telemetría de Lambda les permite a sus extensiones recibir datos de telemetría directamente de Lambda. Durante la inicialización e invocación de la función, Lambda captura automáticamente la telemetría, como los registros, las métricas de la plataforma y los seguimientos de la plataforma. La API de telemetría les permite a las extensiones obtener estos datos de telemetría directamente de Lambda casi en tiempo real.

Dentro del entorno de ejecución de Lambda, puede suscribir las extensiones de Lambda a flujos de telemetría. Tras suscribirse, Lambda envía automáticamente todos los datos de telemetría a sus extensiones. A continuación, tiene flexibilidad para procesar, filtrar y despachar los datos a su destino preferido, como un bucket de Amazon Simple Storage Service (Amazon S3) o un proveedor de herramientas de observabilidad de terceros.

El siguiente diagrama muestra cómo la API de extensiones y la API de telemetría enlazan las extensiones a Lambda desde el entorno de ejecución. Además, la API de tiempo de ejecución conecta el tiempo de ejecución y la función a Lambda.

Las extensiones pueden utilizar la API de telemetría para suscribirse a tres flujos de telemetría diferentes:

Creación de extensiones mediante la API de telemetría

Las extensiones de Lambda se ejecutan como un proceso independiente en el entorno de ejecución. Las extensiones se pueden seguir ejecutando una vez que finaliza la invocación de la función. Dado que las extensiones son procesos separados, puede escribirlas en un idioma diferente al del código de la función. Recomendamos al usuario que escriba extensiones mediante un lenguaje compilado como Golang o Rust. De esta manera, la extensión es un binario autónomo compatible con cualquier tiempo de ejecución admitido.

El siguiente diagrama ilustra un proceso de cuatro pasos para crear una extensión que reciba y procese datos de telemetría mediante la API de telemetría.

A continuación, se muestra cada paso en mayor profundidad:

Registro de sus extensiones

Antes de poder suscribirse para recibir datos de telemetría, debe registrar su extensión de Lambda. El registro se produce durante la fase de inicialización de la extensión. En el siguiente ejemplo se muestra una solicitud HTTP para registrar una extensión.

POST http://${AWS_LAMBDA_RUNTIME_API}/2020-01-01/extension/register
 Lambda-Extension-Name: lambda_extension_name
{
    'events': [ 'INVOKE', 'SHUTDOWN']
}

Si la solicitud se realiza correctamente, el suscriptor recibe una respuesta correcta HTTP 200. El encabezado de la respuesta contiene el Lambda-Extension-Identifier. El cuerpo de respuesta contiene otras propiedades de la función.

HTTP/1.1 200 OK
Lambda-Extension-Identifier: a1b2c3d4-5678-90ab-cdef-EXAMPLE11111
{
    "functionName": "lambda_function",
    "functionVersion": "$LATEST",
    "handler": "lambda_handler",
    "accountId": "123456789012"
}

Para obtener más información, consulte Referencia de la API de extensiones.

Creación de un oyente de telemetría

La extensión de Lambda debe tener un oyente que gestione las solicitudes entrantes de la API de telemetría. En el siguiente código se muestra un ejemplo de implementación de un oyente de telemetría en Golang:

// Starts the server in a goroutine where the log events will be sent
func (s *TelemetryApiListener) Start() (string, error) {
	address := listenOnAddress()
	l.Info("[listener:Start] Starting on address", address)
	s.httpServer = &http.Server{Addr: address}
	http.HandleFunc("/", s.http_handler)
	go func() {
		err := s.httpServer.ListenAndServe()
		if err != http.ErrServerClosed {
			l.Error("[listener:goroutine] Unexpected stop on Http Server:", err)
			s.Shutdown()
		} else {
			l.Info("[listener:goroutine] Http Server closed:", err)
		}
	}()
	return fmt.Sprintf("http://%s/", address), nil
}

// http_handler handles the requests coming from the Telemetry API.
// Everytime Telemetry API sends log events, this function will read them from the response body
// and put into a synchronous queue to be dispatched later.
// Logging or printing besides the error cases below is not recommended if you have subscribed to
// receive extension logs. Otherwise, logging here will cause Telemetry API to send new logs for
// the printed lines which may create an infinite loop.
func (s *TelemetryApiListener) http_handler(w http.ResponseWriter, r *http.Request) {
	body, err := ioutil.ReadAll(r.Body)
	if err != nil {
		l.Error("[listener:http_handler] Error reading body:", err)
		return
	}

	// Parse and put the log messages into the queue
	var slice []interface{}
	_ = json.Unmarshal(body, &slice)

	for _, el := range slice {
		s.LogEventsQueue.Put(el)
	}

	l.Info("[listener:http_handler] logEvents received:", len(slice), " LogEventsQueue length:", s.LogEventsQueue.Len())
	slice = nil
}

Especificación de un protocolo de destino

Al suscribirse para recibir telemetría mediante la API de telemetría, puede especificar un protocolo de destino además del URI de destino:

{
    "destination": {
        "protocol": "HTTP",
        "URI": "http://sandbox.localdomain:8080"
    }
}

Lambda acepta dos protocolos para recibir telemetría:

Antes de suscribirse para recibir telemetría, establezca el oyente HTTP local o el puerto TCP. Durante la instalación, tenga en cuenta lo siguiente:

Configuración del uso de memoria y el almacenamiento en búfer

El uso de memoria en un entorno de ejecución aumenta de forma lineal con el número de suscriptores. Las suscripciones consumen recursos de memoria porque cada una abre un nuevo búfer de memoria para almacenar datos de telemetría. El uso de memoria del búfer cuenta para el consumo general de memoria en el entorno de ejecución.

Al suscribirse para recibir telemetría mediante la API de telemetría, tiene la opción de almacenar en búfer los datos de telemetría y entregarlos a los suscriptores en lotes. Para optimizar el uso de memoria, puede especificar la configuración de almacenamiento en búfer:

{
    "buffering": {
        "maxBytes": 256*1024,
        "maxItems": 1000,
        "timeoutMs": 100
    }
}

Cuando configure el almacenamiento en búfer, tenga en cuenta lo siguiente:

Envío de una solicitud de suscripción a la API de telemetría

Las extensiones de Lambda pueden suscribirse para recibir datos de telemetría enviando una solicitud de suscripción a la API de telemetría. La solicitud de suscripción debe contener información sobre los tipos de eventos a los que quiere que se suscriba la extensión. Además, la solicitud puede contener información sobre el destino de la entrega y una configuración de almacenamiento en búfer.

Antes de enviar una solicitud de suscripción, debe tener un ID de extensión (Lambda-Extension-Identifier). Cuando registra su extensión con la API de extensiones, obtiene un ID de extensión de la respuesta de la API.

La suscripción se produce durante la fase de inicialización de la extensión. El siguiente ejemplo muestra una solicitud HTTP para suscribirse a los tres flujos de telemetría: telemetría de plataforma, registros de funciones y registros de extensiones.

PUT http://${AWS_LAMBDA_RUNTIME_API}/2022-07-01/telemetry HTTP/1.1
{
   "schemaVersion": "2022-12-13",
   "types": [
        "platform",
        "function",
        "extension"
   ],
   "buffering": {
        "maxItems": 1000,
        "maxBytes": 256*1024,
        "timeoutMs": 100
   },
   "destination": {
        "protocol": "HTTP",
        "URI": "http://sandbox.localdomain:8080"
   }
}

Si la solicitud se realiza correctamente, entonces el suscriptor recibe una respuesta correcta HTTP 200.

HTTP/1.1 200 OK
"OK"

Mensajes entrantes de la API de telemetría

Tras suscribirse mediante la API de telemetría, una extensión comienza a recibir automáticamente la telemetría de Lambda mediante solicitudes POST. El cuerpo de cada solicitud POST contiene una matriz de objetos Event. Cada Event tiene el siguiente esquema:

{
   time: String,
   type: String,
   record: Object
}

La siguiente tabla resume todos los tipos de objetos de Event y se enlaza con la referencia del Event del esquema de la API de telemetría para cada tipo de evento.