Ejemplo de una aplicación impulsada por eventos basada en webhooks en el almacenamiento S3 de Mail.ru Cloud Solutions

Máquina de café de Rube Goldberg

La arquitectura orientada a eventos mejora la eficiencia de costos de los recursos utilizados, ya que solo se activan en el momento en que son necesarios. Hay muchas maneras de implementar esto sin crear entidades en la nube adicionales como aplicaciones worker. Hoy no hablaré de FaaS, sino de webhooks. Mostraré un ejemplo práctico de procesamiento de eventos utilizando webhooks de almacenamiento de objetos.

Unas palabras sobre el almacenamiento de objetos y los webhooks. Los almacenamientos de objetos permiten almacenar cualquier dato en la nube en forma de objetos, accesibles a través de S3 u otra API (dependiendo de la implementación) mediante HTTP/HTTPS. Los webhooks (webhooks) son, en general, callbacks personalizados por HTTP. Normalmente se activan por un evento, como el envío de código a un repositorio o un comentario publicado en un blog. Cuando ocurre un evento, el sitio de origen envía una solicitud HTTP a la URL especificada para el webhook. Como resultado, se puede hacer que los eventos en un sitio desencadenen acciones en otro. En el caso de que el sitio de origen sea un almacenamiento de objetos, los cambios en su contenido actúan como eventos.wiki.

Ejemplos de casos simples donde se puede usar tal automatización:

1. Creación de copias de todos los objetos en otro almacenamiento en la nube. Las copias deben realizarse "en tiempo real", con cada adición o modificación de archivos.
2. Creación automática de series de miniaturas de archivos gráficos, adición de marcas de agua a fotografías y otras modificaciones de imágenes.
3. Notificación sobre la llegada de nuevos documentos (por ejemplo, un servicio de contabilidad distribuido sube informes a la nube, y el monitoreo financiero recibe notificaciones sobre nuevos informes, los verifica y los analiza).
4. Casos un poco más complejos implican, por ejemplo, la generación de una petición a Kubernetes, que crea un pod con los contenedores necesarios, le pasa parámetros de tarea y, tras su procesamiento, desmantela el contenedor.

Como ejemplo, realizaremos una variante de la tarea 1, donde los cambios en el bucket del almacenamiento de objetos de Mail.ru Cloud Solutions (MCS) se sincronizan mediante webhooks en el almacenamiento de objetos de AWS. En un caso real con carga, se debe prever el trabajo asincrónico mediante el registro de webhooks en una cola, pero para esta tarea de aprendizaje haremos la implementación sin ello.

Esquema de operación

El protocolo de interacción se describe en detalle en la guía de webhooks S3 en MCS. El esquema de trabajo incluye los siguientes elementos:

• Servicio de publicación, que se encuentra del lado del almacenamiento S3 y publica solicitudes HTTP al activarse el webhook.
• Servidor de recepción de webhooks, que escucha las solicitudes del servicio de publicación a través de HTTP y ejecuta las acciones correspondientes. El servidor se puede escribir en cualquier lenguaje, en nuestro ejemplo escribiremos el servidor en Go.

Una característica de la implementación de webhooks en la API de S3 es el registro del servidor de recepción de webhooks en el servicio de publicación. En particular, el servidor de recepción de webhooks debe confirmar la suscripción a los mensajes del servicio de publicación (en otras implementaciones de webhooks, generalmente no se requiere confirmar la suscripción).

Por lo tanto, el servidor de recepción de webhooks debe admitir dos operaciones principales:

• responder a la solicitud del servicio de publicación para confirmar el registro,
• procesar los eventos entrantes.

Configuración del servidor de recepción de webhooks

Para poner en marcha el servidor de recepción de webhooks se necesita un servidor Linux. En este artículo, utilizamos como ejemplo una instancia virtual que desplegamos en MCS.

Instalaremos el software necesario y pondremos en marcha el servidor de recepción de webhooks.

ubuntu@ubuntu-basic-1-2-10gb:~$ sudo apt-get install git
Leyendo listas de paquetes... Listo
Construyendo árbol de dependencias
Leyendo la información del estado... Listo
Los siguientes paquetes se instalaron automáticamente y ya no son necesarios:
 bc dns-root-data dnsmasq-base ebtables landscape-common liblxc-common 
liblxc1 libuv1 lxcfs lxd lxd-client python3-attr python3-automat 
python3-click python3-constantly python3-hyperlink
 python3-incremental python3-pam python3-pyasn1-modules 
python3-service-identity python3-twisted python3-twisted-bin 
python3-zope.interface uidmap xdelta3
Utiliza 'sudo apt autoremove' para eliminarlos.
Paquetes sugeridos:
 git-daemon-run | git-daemon-sysvinit git-doc git-el git-email git-gui 
gitk gitweb git-cvs git-mediawiki git-svn
Los siguientes NUEVOS paquetes se instalarán:
 git
0 actualizados, 1 instalado nuevo, 0 para eliminar y 46 no actualizados.
Necesita obtener 3915 kB de archivos.
Después de esta operación, se utilizarán 32.3 MB de espacio adicional en disco.
Obteniendo:1 http://MS1.clouds.archive.ubuntu.com/ubuntu bionic-updates/main 
amd64 git amd64 1:2.17.1-1ubuntu0.7 [3915 kB]
Se obtuvieron 3915 kB en 1s (5639 kB/s)
Seleccionando el paquete git previamente no seleccionado.
(Leyendo base de datos ... 53932 archivos y directorios actualmente instalados.)
Preparándose para desempaquetar .../git_12.17.1-1ubuntu0.7_amd64.deb ...
Desempaquetando git (1:2.17.1-1ubuntu0.7) ...
Configurando git (1:2.17.1-1ubuntu0.7) ...

Clonamos la carpeta del servidor de recepción de webhooks:

ubuntu@ubuntu-basic-1-2-10gb:~$ git clone
https://github.com/RomanenkoDenys/s3-webhook.git
Clonando en 's3-webhook'...
remoto: Enumerando objetos: 48, hecho.
remoto: Contando objetos: 100% (48/48), hecho.
remoto: Comprimiendo objetos: 100% (27/27), hecho.
remoto: Total 114 (delta 20), reutilizados 45 (delta 18), paquete reutilizado 66
Recibiendo objetos: 100% (114/114), 23.77 MiB | 20.25 MiB/s, hecho.
Resolviendo deltas: 100% (49/49), hecho.

Pondremos en marcha el servidor:

ubuntu@ubuntu-basic-1-2-10gb:~$ cd s3-webhook/
ubuntu@ubuntu-basic-1-2-10gb:~/s3-webhook$ sudo ./s3-webhook -port 80

Suscripción al servicio de publicación

Puedes registrar tu servidor de recepción de webhooks a través de la API o la interfaz web. Para simplificar, lo registraremos a través de la interfaz web:

1. Vamos a la sección de buckets en el panel de control.
2. Entramos en el bucket para el que vamos a configurar los webhooks y hacemos clic en el engranaje:

Vamos a la pestaña Webhooks y hacemos clic en Agregar:

Llenamos los campos:

ID — nombre del webhook.

Evento — qué eventos enviar. Hemos configurado el envío de todos los eventos que ocurren al trabajar con archivos (adición y eliminación).

URL — dirección del servidor receptor de webhooks.

Filtro de prefijo/sufijo — filtro que permite generar webhooks solo para objetos cuyos nombres cumplen determinadas reglas. Por ejemplo, para que el webhook se active únicamente para archivos con la extensión .png, en Filtro de sufijo hay que escribir «png».

En este momento, solo se admiten los puertos 80 y 443 para la comunicación con el servidor receptor de webhooks.

Hagamos clic en Agregar hook y veremos lo siguiente:

Hook agregado.

El servidor receptor de webhooks muestra en los registros el proceso de registro del hook:

ubuntu@ubuntu-basic-1-2-10gb:~/s3-webhook$ sudo ./s3-webhook -port 80
2020/06/15 12:01:14 [POST] solicitud HTTP entrante de 
95.163.216.92:42530
2020/06/15 12:01:14 Obtuve la marca de tiempo: 2020-06-15T15:01:13+03:00 TopicArn: 
mcs5259999770|myfiles-ash|s3:ObjectCreated:*,s3:ObjectRemoved:* Token: 
E2itMqAMUVVZc51pUhFWSp13DoxezvRxkUh5P7LEuk1dEe9y URL: 
http://89.208.199.220/webhook
2020/06/15 12:01:14 Generar firma de respuesta: 
3754ce36636f80dfd606c5254d64ecb2fd8d555c27962b70b4f759f32c76b66d

Registro completado. En la siguiente sección, examinaremos más detalladamente el algoritmo de funcionamiento del servidor receptor de webhooks.

Descripción del servidor receptor de webhooks

En nuestro ejemplo, el servidor está escrito en Go. Analicemos los principios básicos de su funcionamiento.

package main

// Generar hmac_sha256_hex
func HmacSha256hex(message string, secret string) string {
}

// Generar hmac_sha256
func HmacSha256(message string, secret string) string {
}

// Enviar confirmación de suscripción
func SubscriptionConfirmation(w http.ResponseWriter, req *http.Request, body []byte) {
}

// Enviar confirmación de suscripción
func GotRecords(w http.ResponseWriter, req *http.Request, body []byte) {
}

// Prueba de vida
func Ping(w http.ResponseWriter, req *http.Request) {
    // registrar solicitud
    log.Printf("[%s] solicitud HTTP Ping entrante de %sn", req.Method, req.RemoteAddr)
    fmt.Fprintf(w, "Pongn")
}

//Webhook
func Webhook(w http.ResponseWriter, req *http.Request) {
}

func main() {

    // obtener argumentos de línea de comandos
    bindPort := flag.Int("port", 80, "número entre 1-65535")
    bindAddr := flag.String("address", "", "dirección IP en formato decimal")
    flag.StringVar(&actionScript, "script", "", "script externo para ejecutar")
    flag.Parse()

    http.HandleFunc("/ping", Ping)
    http.HandleFunc("/webhook", Webhook)

log.Fatal(http.ListenAndServe(*bindAddr+":"+strconv.Itoa(*bindPort), nil))
}

Veamos las funciones principales:

• Ping() — ruta que responde en URL/ping, una implementación simple de liveness probe.
• Webhook() — ruta principal, controlador de URL/webhook:
  • confirma el registro en el servicio de publicación (transición a la función SubscriptionConfirmation),
  • procesa los webhooks entrantes (función Gotrecords).
• Las funciones HmacSha256 y HmacSha256hex son implementaciones de los algoritmos de cifrado HMAC-SHA256 y HMAC-SHA256 con salida en forma de cadena de números hexadecimales para calcular la firma.
• main — función principal, procesa los parámetros de la línea de comandos y registra los controladores de URL.

Parámetros de la línea de comandos que el servidor acepta:

• -port — puerto en el que el servidor estará escuchando.
• -address — dirección IP que el servidor estará escuchando.
• -script — programa externo que se invoca con cada webhook entrante.

Analicemos más a fondo algunas funciones:

//Webhook
func Webhook(w http.ResponseWriter, req *http.Request) {

    // Read body
    body, err := ioutil.ReadAll(req.Body)
    defer req.Body.Close()
    if err != nil {
        http.Error(w, err.Error(), 500)
        return
    }

    // log request
    log.Printf("[%s] incoming HTTP request from %sn", req.Method, req.RemoteAddr)
    // check if we got subscription confirmation request
    if strings.Contains(string(body), 
""Type":"SubscriptionConfirmation"") {
        SubscriptionConfirmation(w, req, body)
    } else {
        GotRecords(w, req, body)
    }

}

Esta función determina qué se ha recibido: si es una solicitud de confirmación de registro o un webhook. Como se indica en la documentación, en caso de confirmación de registro, la siguiente estructura Json se envía en la solicitud Post:

POST http://test.com HTTP/1.1
x-amz-sns-messages-type: SubscriptionConfirmation
content-type: application/json

{
    "Timestamp":"2019-12-26T19:29:12+03:00",
    "Type":"SubscriptionConfirmation",
    "Message":"Has elegido suscribirte al tema $topic. Para confirmar la suscripción, necesitas responder con la firma calculada",
    "TopicArn":"mcs2883541269|bucketA|s3:ObjectCreated:Put",
    "SignatureVersion":1,
    "Token":"RPE5UuG94rGgBH6kHXN9FUPugFxj1hs2aUQc99btJp3E49tA"
}

Se debe responder a esta solicitud:

content-type: application/json

{"signature":"ea3fce4bb15c6de4fec365d36bcebbc34ccddf54616d5ca12e1972f82b6d37af"}

Donde la firma se calcula como:

signature = hmac_sha256(url, hmac_sha256(TopicArn, 
hmac_sha256(Timestamp, Token)))

Si llega un webhook, la estructura de la solicitud Post es así:

POST  HTTP/1.1
x-amz-sns-messages-type: SubscriptionConfirmation

{ "Records":
    [
        {
            "s3": {
                "object": {
                    "eTag":"aed563ecafb4bcc5654c597a421547b2",
                    "sequencer":1577453615,
                    "key":"some-file-to-bucket",
                    "size":100
                },
            "configurationId":"1",
            "bucket": {
                "name": "bucketA",
                "ownerIdentity": {
                    "principalId":"mcs2883541269"}
                },
                "s3SchemaVersion":"1.0"
            },
            "eventVersion":"1.0",
            "requestParameters":{
                "sourceIPAddress":"185.6.245.156"
            },
            "userIdentity": {
                "principalId":"2407013e-cbc1-415f-9102-16fb9bd6946b"
            },
            "eventName":"s3:ObjectCreated:Put",
            "awsRegion":"ru-msk",
            "eventSource":"aws:s3",
            "responseElements": {
                "x-amz-request-id":"VGJR5rtJ"
            }
        }
    ]
}

Por lo tanto, dependiendo de la solicitud, es necesario entender cómo procesar los datos. Elegí como indicador la entrada "Type":"SubscriptionConfirmation", ya que está presente en la solicitud de confirmación de suscripción y no está en el webhook. Según la presencia o ausencia de este registro en la solicitud POST, la ejecución del programa pasa a la función SubscriptionConfirmation, o a la función GotRecords.

No vamos a examinar la función SubscriptionConfirmation en detalle, se implementa según los principios expuestos en la documentación. Puedes estudiar el código fuente de esta función en el repositorio git del proyecto.

La función GotRecords analiza la solicitud entrante y para cada objeto Record llama a un script externo (cuyo nombre fue pasado en el parámetro -script) con los parámetros:

• nombre del bucket
• clave del objeto
• acción:
  • copy — si en la solicitud original EventName = ObjectCreated | PutObject | PutObjectCopy
  • delete — si en la solicitud original EventName = ObjectRemoved | DeleteObject

Por lo tanto, si llega un webhook con la solicitud POST, tal como se describe arriba, y el parámetro -script=script.sh, entonces el script será llamado de la siguiente manera:

script.sh bucketA some-file-to-bucket copy

Debemos entender que este servidor de recepción de webhooks no es una solución de producción completa, sino un ejemplo simplificado de una posible implementación.

Ejemplo de funcionamiento

Haremos la sincronización de archivos del bucket principal en MCS al bucket de respaldo en AWS. El bucket principal se llama myfiles-ash y el de respaldo — myfiles-backup (la configuración del bucket en AWS está fuera del alcance de este artículo). Por lo tanto, cuando un archivo se coloca en el bucket principal, su copia debe aparecer en el de respaldo; cuando se elimina del principal, debe ser eliminada en el de respaldo.

Trabajaremos con los buckets usando la herramienta awscli, que es compatible tanto con el almacenamiento en la nube MCS como con el almacenamiento en la nube AWS.

ubuntu@ubuntu-basic-1-2-10gb:~$ sudo apt-get install awscli
Leyendo listas de paquetes... Hecho
Construyendo árbol de dependencias
Leyendo la información de estado... Hecho
Después de esta operación, se usarán 34.4 MB de espacio adicional en disco.
Desempaquetando awscli (1.14.44-1ubuntu1) ...
Configurando awscli (1.14.44-1ubuntu1) ...

Configuraremos el acceso a la API S3 de MCS:

ubuntu@ubuntu-basic-1-2-10gb:~$ aws configure --profile mcs
AWS Access Key ID [None]: hdywEPtuuJTExxxxxxxxxxxxxx
AWS Secret Access Key [None]: hDz3SgxKwXoxxxxxxxxxxxxxxxxxx
Default region name [None]:
Default output format [None]:

Configuraremos el acceso a la API S3 de AWS:

ubuntu@ubuntu-basic-1-2-10gb:~$ aws configure --profile aws
AWS Access Key ID [None]: AKIAJXXXXXXXXXXXX
AWS Secret Access Key [None]: dfuerphOLQwu0CreP5Z8l5fuXXXXXXXXXXXXXXXX
Default region name [None]:
Default output format [None]:

Verificaremos los accesos:

A AWS:

ubuntu@ubuntu-basic-1-2-10gb:~$ aws s3 ls --profile aws
2020-07-06 08:44:11 myfiles-backup

Para MCS, al ejecutar el comando, se debe añadir —endpoint-url:

ubuntu@ubuntu-basic-1-2-10gb:~$ aws s3 ls --profile mcs --endpoint-url 
https://hb.bizmrg.com
2020-02-04 06:38:05 databasebackups-0cdaaa6402d4424e9676c75a720afa85
2020-05-27 10:08:33 myfiles-ash

Acceso obtenido.

Ahora escribiremos un script para procesar el webhook entrante, lo llamaremos s3_backup_mcs_aws.sh

#!/bin/bash
# Require aws cli
# if file added — copy it to backup bucket
# if file removed — remove it from backup bucket
# Variables
ENDPOINT_MCS="https://hb.bizmrg.com"
AWSCLI_MCS=`which aws`" --endpoint-url ${ENDPOINT_MCS} --profile mcs s3"
AWSCLI_AWS=`which aws`" --profile aws s3"
BACKUP_BUCKET="myfiles-backup"

SOURCE_BUCKET="${1}"
SOURCE_FILE="${2}"
ACTION="${3}"

SOURCE="s3://${SOURCE_BUCKET}/${SOURCE_FILE}"
TARGET="s3://${BACKUP_BUCKET}/${SOURCE_FILE}"
TEMP="/tmp/${SOURCE_BUCKET}/${SOURCE_FILE}"

case ${ACTION} in
    "copy")
    ${AWSCLI_MCS} cp "${SOURCE}" "${TEMP}"
    ${AWSCLI_AWS} cp "${TEMP}" "${TARGET}"
    rm ${TEMP}
    ;;

    "delete")
    ${AWSCLI_AWS} rm ${TARGET}
    ;;

    *)
    echo "Usage: ${0} sourcebucket sourcefile copy/delete"
    exit 1
    ;;
esac

Iniciamos el servidor:

ubuntu@ubuntu-basic-1-2-10gb:~\/s3-webhook$ sudo .\/s3-webhook -port 80 -
scripts/scripts/s3_backup_mcs_aws.sh

Verificamos cómo funcionará. A través de la interfaz web de MCS agregaremos el archivo test.txt al bucket myfiles-ash. En los registros de la consola se puede ver que se realizó una solicitud al servidor de webhooks:

2020\/07\/06 09:43:08 [POST] solicitud HTTP entrante de 
95.163.216.92:56612
descarga: s3:\/\/myfiles-ash\/test.txt a ..\/..\/..\/tmp\/myfiles-ash\/test.txt
subida: ..\/..\/..\/tmp\/myfiles-ash\/test.txt a 
s3:\/\/myfiles-backup\/test.txt

Verificamos el contenido del bucket myfiles-backup en AWS:

ubuntu@ubuntu-basic-1-2-10gb:~\/s3-webhook$ aws s3 --profile aws ls 
myfiles-backup
2020-07-06 09:43:10       1104 test.txt

Ahora eliminaremos el archivo del bucket myfiles-ash a través de la interfaz web.

Registros del servidor:

2020\/07\/06 09:44:46 [POST] solicitud HTTP entrante de 
95.163.216.92:58224
delete: s3:\/\/myfiles-backup\/test.txt

Contenido del bucket:

ubuntu@ubuntu-basic-1-2-10gb:~\/s3-webhook$ aws s3 --profile aws ls 
myfiles-backup
ubuntu@ubuntu-basic-1-2-10gb:~$

El archivo ha sido eliminado, tarea completada.

Conclusión y ToDo

Todo el código utilizado en este artículo se encuentra en mi repositorio. Ahí también hay ejemplos de scripts y ejemplos de cálculo de firmas para registrar webhooks.

Este código es solo un ejemplo de cómo se pueden utilizar los webhooks de S3 en su actividad. Como mencioné al principio, si se planea utilizar un servidor así en producción, es necesario al menos reescribirlo para trabajar de forma asíncrona: registrar los webhooks entrantes en una cola (RabbitMQ o NATS), y desde allí descomponerlos y procesarlos mediante aplicaciones worker. De lo contrario, en caso de un gran volumen de webhooks, se puede enfrentar a la falta de recursos del servidor para ejecutar las tareas. La existencia de colas permite distribuir el servidor y los workers, así como resolver problemas de repetición de tareas en caso de fallos. También sería deseable cambiar el registro a uno más detallado y más estandarizado.

¡Éxitos!

Leer más sobre el tema:

• Trabajando con el almacenamiento de objetos S3 de Mail.ru Cloud Solutions como si fuera un sistema de archivos.
• Cómo utilizan el almacenamiento de objetos para superar a la competencia.

Fuente: habr.com