Note:

Este tutorial requiere acceso a Oracle Cloud. Para registrarse para obtener una cuenta gratuita, consulte Introducción a la cuenta gratuita de Oracle Cloud Infrastructure.
Utiliza valores de ejemplo para credenciales, arrendamiento y compartimentos de Oracle Cloud Infrastructure. Al finalizar el laboratorio, sustituya estos valores por otros específicos de su entorno en la nube.

Configuración de notificaciones para trabajos de flujo de trabajo de AIDP y escenarios personalizados mediante OCI Notifications

Introducción

En este tutorial se explica cómo configurar notificaciones para trabajos de flujo de trabajo de AIDP mediante el servicio OCI Notifications and Events. También aprenderá a implementar la lógica de notificación personalizada mediante un programa Python en un bloc de notas de AIDP.

El servicio OCI Notifications utiliza un modelo de publicación-suscripción para entregar mensajes de forma fiable y segura.

Objetivos

Al final de este tutorial, podrá:

Creación de un tema y una suscripción de OCI Notification
Configurar reglas de eventos para el estado del trabajo de flujo de trabajo de AIDP
Recibir notificaciones por correo electrónico de fallos de trabajo
Implantar lógica de notificación personalizada mediante Python
Disparar notificaciones mediante programación

Requisitos

Acceso a la consola de OCI
Espacio de trabajo de AIDP con trabajos de flujo de trabajo
Permisos para crear notificaciones y eventos
Configuración de OCI para el uso del SDK de Python

Tarea 1: Crear tema de notificación y suscripción

Conéctese a la consola de OCI.
Vaya a Servicios para desarrolladores → Notificaciones.
Haga clic en Crear tema y proporcione:
- Nombre
- Descripción
Cree una suscripción:
- Protocolo: correo electrónico
- Correo electrónico: su dirección de correo electrónico
Compruebe su correo electrónico y confirme la suscripción.

Nota: El estado de la suscripción cambia de Pendiente a Activo después de la confirmación.

Tarea 2: Configurar regla de evento para flujo de trabajo de AIDP

Vaya a Observability & Management → Events Service → Rules.
Haga clic en Create Rule.
Configure:
- Servicio: lago de datos inteligente
- Tipo de evento: Ejecutar trabajo - Finalizar
- Nombre de atributo 1: jobKey
- Valores de atributo 1: Mantener ID de flujo de trabajo de ayuda
- Nombre de atributo 2: jobStatus
- Valores de atributo 2: Fallo
Definir Acción:
- Seleccionar tema de notificación creado anteriormente

Consejo: puede personalizar reglas para otros estados, como Correcto o En ejecución.

Tarea 3: Validar notificación por correo electrónico

Dispara o ejecuta un trabajo de flujo de trabajo.
Fuerce un escenario de fallo si es necesario.
Compruebe el buzón de correo electrónico.

Nota: La notificación incluye detalles del trabajo, como el nombre y el estado del trabajo.

Tarea 4: Implantación de notificaciones personalizadas mediante Python

A continuación se muestra un programa Python de ejemplo para enviar notificaciones mediante el SDK de OCI.

import oci
from oci.ons.models import MessageDetails

config = oci.config.from_file("~/.oci/config", "DEFAULT")
client = oci.ons.NotificationDataPlaneClient(config)

topic_id = "<your_topic_ocid>"

message_details = MessageDetails(
    title="Test Notification",
    body="Hello from AIDP!"
)

response = client.publish_message(
    topic_id=topic_id,
    message_details=message_details
)

print("Message sent:", response.data.message_id)

Tarea 5: Clase de notificación personalizada avanzada

import oci
import logging
import time
from typing import List, Optional, Dict

logger = logging.getLogger("OCI_Notifier")
logger.setLevel(logging.INFO)

handler = logging.StreamHandler()
formatter = logging.Formatter(
    "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
handler.setFormatter(formatter)
logger.addHandler(handler)

class OCINotifier:
    def __init__(
        self,
        topic_id: str,
        config_file: str = "~/.oci/config",
        profile: str = "DEFAULT",
        max_retries: int = 3,
        retry_delay: int = 2,
    ):
        """
        Initialize OCI Notification client

        Args:
            topic_id: OCI Notification Topic OCID
            config_file: Path to OCI config
            profile: Config profile name
            max_retries: Retry attempts
            retry_delay: Delay between retries (seconds)
        """
        self.topic_id = topic_id
        self.max_retries = max_retries
        self.retry_delay = retry_delay

        try:
            self.config = oci.config.from_file(config_file, profile)
            self.client = oci.ons.NotificationDataPlaneClient(self.config)
            logger.info("OCI Notifier initialized successfully")
        except Exception as e:
            logger.error(f"Failed to initialize OCI client: {e}")
            raise

    
    def send(
        self,
        title: str,
        message: str,
        metadata: Optional[Dict] = None,
    ) -> bool:
        """
        Send notification message

        Args:
            title: Notification subject
            message: Notification body
            metadata: Optional metadata dictionary

        Returns:
            bool: True if success, False otherwise
        """

        payload = {
            "title": title,
            "body": message,
        }

        if metadata:
            payload["metadata"] = metadata

        for attempt in range(1, self.max_retries + 1):
            try:
                response = self.client.publish_message(
                    self.topic_id,
                    payload
                )
                logger.info(
                    f"Notification sent successfully | "
                    f"Message ID: {response.data.message_id}"
                )
                return True

            except Exception as e:
                logger.error(
                    f"Attempt {attempt} failed: {str(e)}"
                )
                if attempt < self.max_retries:
                    time.sleep(self.retry_delay)
                else:
                    logger.error("Max retries reached. Notification failed.")
                    return False



def notify_success(notifier: OCINotifier, job_name: str):
    notifier.send(
        title=f"{job_name} SUCCESS",
        message=f"Job '{job_name}' completed successfully."
    )


def notify_failure(notifier: OCINotifier, job_name: str, error: str):
    notifier.send(
        title=f"{job_name} FAILED",
        message=f"Job '{job_name}' failed.\nError: {error}"
    )

Tarea 6: Llamar a notificación personalizada desde el bloc de notas de AIDP

Puede llamar a la lógica de notificación dentro del flujo de trabajo o el bloc de notas:

from notifier import notify_failure

notify_failure(notifier, "SampleJob", "Error details here")

Solución de problemas y consejos

Consejo: asegúrese de que la ruta del archivo de configuración de OCI es correcta.

Nota: Verifique el OCID del tema y el estado de la suscripción.

Consejo: utilice reintentos en el notificador personalizado para garantizar la fiabilidad.

Pasos Siguientes

Agregar suscripciones de Slack o HTTPS
Ampliar las notificaciones para eventos de éxito
Integración con paneles de control de supervisión

Agradecimientos

Autor: Pavan Upadhyay (ingeniero principal de la nube), Saket Bihari (ingeniero principal de la nube)

Más recursos de aprendizaje

Explore otros laboratorios en docs.oracle.com/learn o acceda a más contenido de aprendizaje gratuito en el canal de YouTube de Oracle Learning. Además, visite education.oracle.com/learning-explorer para convertirse en un explorador de Oracle Learning.

Para obtener documentación sobre el producto, visite Oracle Help Center.

Título e Información de Copyright

Configure Notifications for AIDP Workflow Jobs and Custom Scenarios Using OCI Notifications

G56825-01

Oracle y/o sus filiales.