Hinweis:

Benachrichtigungen für AIDP-Workflowjobs und benutzerdefinierte Szenarios mit OCI-Benachrichtigungen konfigurieren

Einführung

In diesem Tutorial wird erläutert, wie Sie Benachrichtigungen für AIDP-Workflowjobs mit dem OCI Notifications and Events-Service konfigurieren. Außerdem lernen Sie, wie Sie benutzerdefinierte Benachrichtigungslogik mit einem Python-Programm in einem AIDP-Notizbuch implementieren.

Der OCI Notifications-Service verwendet ein Publish-Subscribe-Modell, um Nachrichten zuverlässig und sicher zuzustellen.

Ziele

Am Ende dieses Tutorials haben Sie folgende Möglichkeiten:

Voraussetzungen

Aufgabe 1: Benachrichtigungsthema und Abonnement erstellen

  1. Melden Sie sich bei OCI Console an.
  2. Navigieren Sie zu Entwicklerservices → Benachrichtigungen.
  3. Klicken Sie auf Thema erstellen, und geben Sie Folgendes an:
    • Name
    • Beschreibung

    Thema erstellen

  4. Abonnement erstellen:
    • Protokoll: E-Mail
    • E-Mail: Ihre E-Mail-Adresse

    Abonnement erstellen

  5. Überprüfen Sie Ihre E-Mail-Adresse und bestätigen Sie das Abonnement.

    Subscription bestätigen

Hinweis: Der Abonnementstatus wird nach der Bestätigung von Ausstehend in Aktiv geändert.

Aufgabe 2: Ereignisregel für AIDP-Workflow konfigurieren

  1. Navigieren Sie zu Observability and Management → Events Service → Rules.
  2. Klicken Sie auf Regel erstellen.
  3. Konfigurieren Sie:
    • Service: Intelligent Data Lake
    • Ereignistyp: Job ausführen - Ende
    • Attributname 1: jobKey
    • Attributwerte 1: Hilfs-Workflow-ID beibehalten
    • Attributname 2: jobStatus
    • Attributwerte2: Nicht erfolgreich
  4. Aktion festlegen:
    • Benachrichtigungsthema auswählen, das zuvor erstellt wurde

    Subscription bestätigen

Tipp: Sie können Regeln für andere Statusarten wie "Erfolgreich" oder "Wird ausgeführt" anpassen.

Aufgabe 3: E-Mail-Benachrichtigung validieren

  1. Workflowjob auslösen oder ausführen
  2. Erzwingen Sie bei Bedarf ein Fehlerszenario.

  3. Prüfen Sie Ihren E-Mail-Posteingang.

    Benachrichtigung über Jobfalsch

Hinweis: Die Benachrichtigung enthält Jobdetails wie Jobname und Status.

Aufgabe 4: Benutzerdefinierte Benachrichtigungen mit Python implementieren

Im Folgenden finden Sie ein Python-Beispielprogramm zum Senden von Benachrichtigungen mit dem OCI-SDK.

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)

Aufgabe 5: Erweiterte benutzerdefinierte Benachrichtigungsklasse

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}"
    )

Aufgabe 6: Benutzerdefinierte Benachrichtigung aus AIDP-Notizbuch aufrufen

Sie können die Benachrichtigungslogik in Ihrem Workflow oder Notizbuch aufrufen:

from notifier import notify_failure

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

Fehlerbehebung und Tipps

Tipp: Stellen Sie sicher, dass der Pfad der OCI-Konfigurationsdatei korrekt ist.

Hinweis: Prüfen Sie die Themen-OCID und den Abonnementstatus.

Tipp: Verwenden Sie Wiederholungen im benutzerdefinierten Notifier, um die Zuverlässigkeit zu gewährleisten.

Nächste Schritte

Danksagungen

Weitere Lernressourcen

Sehen Sie sich weitere Übungen auf docs.oracle.com/learn an, oder greifen Sie auf weitere kostenlose Lerninhalte im YouTube-Kanal von Oracle Learning zu. Besuchen Sie außerdem education.oracle.com/learning-explorer, um ein Oracle Learning Explorer zu werden.

Die Produktdokumentation finden Sie im Oracle Help Center.