Building a Python Framework to Monitor IDMC Taskflow Execution
Starting an Informatica Intelligent Data Management Cloud (IDMC) taskflow from code is easy. Knowing what happened to it afterward is harder.
In most pipelines I've reviewed, the "monitoring" is a scheduled job that fires the taskflow and hopes for the best. When a run hangs, gets suspended, or quietly fails at 2 a.m., nobody finds out until a downstream report is empty.
This post walks through a small Python framework that fixes that. It:
Authenticates to IDMC and renews the session when it expires
Starts a taskflow and captures its run ID
Polls run status with exponential backoff and a hard timeout
Treats FAILED, SUSPENDED, and timeouts as distinct outcomes
Sends alerts through pluggable notifiers (Slack, Amazon SNS, or anything else)
Emits structured JSON logs you can query in CloudWatch, Datadog, or Splunk
All of it fits in about 300 lines and works the same on a laptop, in a container, or inside AWS Lambda.
Note: IDMC endpoint paths and status values can vary by pod and release. The ones below match the public REST API docs at the time of writing. Check them against your org's API reference before going to production.
The architecture
The framework has four parts:
Step 1: Configuration
Keep credentials out of code. A small dataclass reads them from environment variables, and in AWS those can come from Secrets Manager.
# idmc_monitor/config.py import os from dataclasses import dataclass, field @dataclass(frozen=True) class IDMCConfig: username: str password: str login_url: str = "https://dm-us.informaticacloud.com/saas/public/core/v3/login" poll_initial_seconds: float = 10.0 poll_max_seconds: float = 120.0 timeout_seconds: float = 3600.0 request_timeout: float = 30.0 terminal_failure_states: frozenset = field( default_factory=lambda: frozenset({"FAILED", "SUSPENDED", "FAULTED", "ABORTED"}) ) terminal_success_states: frozenset = field( default_factory=lambda: frozenset({"SUCCESS", "COMPLETED"}) ) @classmethod def from_env(cls) -> "IDMCConfig": return cls( username=os.environ["IDMC_USERNAME"], password=os.environ["IDMC_PASSWORD"], login_url=os.environ.get("IDMC_LOGIN_URL", cls.login_url), timeout_seconds=float(os.environ.get("IDMC_TIMEOUT_SECONDS", 3600)), ) The login URL depends on your region (dm-us, dm-em, dm-ap, and so on), so it's configurable.
Step 2: A session that renews itself
An IDMC session times out after a period of inactivity. Long-running taskflows can outlast it, so the client has to notice a 401 and log in again without failing the run.
The v3 login response returns a sessionId and a baseApiUrl for your pod. Application Integration endpoints (/active-bpel/...) live on the same pod host, without the /saas suffix.
# idmc_monitor/session.py import logging from urllib.parse import urlparse import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry from .config import IDMCConfig log = logging.getLogger(__name__) class IDMCAuthError(RuntimeError): pass class IDMCSession: """Wraps requests.Session with IDMC login and automatic re-login.""" def __init__(self, config: IDMCConfig): self.config = config self._http = requests.Session() retry = Retry( total=5, backoff_factor=1.5, status_forcelist=(429, 500, 502, 503, 504), allowed_methods=frozenset({"GET", "POST"}), ) self._http.mount("https://", HTTPAdapter(max_retries=retry)) self.session_id: str | None = None self.base_api_url: str | None = None self.pod_url: str | None = None def login(self) -> None: resp = self._http.post( self.config.login_url, json={"username": self.config.username, "password": self.config.password}, timeout=self.config.request_timeout, ) if resp.status_code != 200: raise IDMCAuthError(f"Login failed: HTTP {resp.status_code}") body = resp.json() self.session_id = body["userInfo"]["sessionId"] self.base_api_url = body["products"][0]["baseApiUrl"] parsed = urlparse(self.base_api_url) self.pod_url = f"{parsed.scheme}://{parsed.netloc}" log.info("idmc_login_ok", extra={"pod": self.pod_url}) def request(self, method: str, url: str, **kwargs) -> requests.Response: if not self.session_id: self.login() for attempt in (1, 2): headers = kwargs.pop("headers", {}) | { "INFA-SESSION-ID": self.session_id, "icSessionId": self.session_id, "Accept": "application/json", } resp = self._http.request( method, url, headers=headers, timeout=self.config.request_timeout, **kwargs, ) if resp.status_code == 401 and attempt == 1: log.warning("idmc_session_expired_relogin") self.login() continue resp.raise_for_status() return resp raise IDMCAuthError("Request still unauthorized after re-login") Two layers of resilience run here. urllib3.Retry handles transient network errors and 5xx/429 responses. The explicit 401 branch handles session expiry, which a generic retry can't fix.
Step 3: The monitor
This is the core of the framework. It starts the taskflow, then polls until one of three things happens: success, failure, or timeout.
# idmc_monitor/monitor.py import logging import random import time from dataclasses import dataclass, field from enum import Enum from .config import IDMCConfig from .notifiers import Notifier from .session import IDMCSession log = logging.getLogger(__name__) class Outcome(str, Enum): SUCCESS = "SUCCESS" FAILED = "FAILED" TIMEOUT = "TIMEOUT" @dataclass class RunResult: taskflow: str run_id: str outcome: Outcome final_status: str duration_seconds: float details: dict = field(default_factory=dict) class TaskflowMonitor: def __init__(self, session: IDMCSession, config: IDMCConfig, notifiers: list[Notifier] | None = None): self.session = session self.config = config self.notifiers = notifiers or [] # -- API calls ----------------------------------------------------- def start(self, taskflow: str, params: dict | None = None) -> str: if not self.session.pod_url: self.session.login() url = f"{self.session.pod_url}/active-bpel/rt/{taskflow}" resp = self.session.request("POST", url, json=params or {}) run_id = str(resp.json()["RunId"]) log.info("taskflow_started", extra={"taskflow": taskflow, "run_id": run_id}) return run_id def get_status(self, run_id: str) -> dict: url = f"{self.session.pod_url}/active-bpel/services/tf/status/{run_id}" return self.session.request("GET", url).json() # -- Polling loop -------------------------------------------------- def _next_delay(self, current: float) -> float: """Exponential backoff with full jitter, capped at poll_max_seconds.""" grown = min(current * 2, self.config.poll_max_seconds) return random.uniform(self.config.poll_initial_seconds, grown) def wait(self, taskflow: str, run_id: str) -> RunResult: started = time.monotonic() delay = self.config.poll_initial_seconds last_status = "UNKNOWN" payload: dict = {} while True: elapsed = time.monotonic() - started if elapsed > self.config.timeout_seconds: return self._finish(taskflow, run_id, Outcome.TIMEOUT, last_status, elapsed, payload) payload = self.get_status(run_id) last_status = str(payload.get("status", "UNKNOWN")).upper() log.info("taskflow_status", extra={ "taskflow": taskflow, "run_id": run_id, "status": last_status, "elapsed": round(elapsed, 1), }) if last_status in self.config.terminal_success_states: return self._finish(taskflow, run_id, Outcome.SUCCESS, last_status, elapsed, payload) if last_status in self.config.terminal_failure_states: return self._finish(taskflow, run_id, Outcome.FAILED, last_status, elapsed, payload) time.sleep(delay) delay = self._next_delay(delay) def run(self, taskflow: str, params: dict | None = None) -> RunResult: run_id = self.start(taskflow, params) return self.wait(taskflow, run_id) # -- Outcome handling ---------------------------------------------- def _finish(self, taskflow, run_id, outcome, status, elapsed, payload) -> RunResult: result = RunResult(taskflow, run_id, outcome, status, round(elapsed, 1), payload) log.info("taskflow_finished", extra={ "taskflow": taskflow, "run_id": run_id, "outcome": outcome.value, "status": status, "duration_seconds": result.duration_seconds, }) if outcome is not Outcome.SUCCESS: for notifier in self.notifiers: try: notifier.send(result) except Exception: # an alert failure must never mask the run result log.exception("notifier_failed", extra={"notifier": type(notifier).__name__}) return result A few design choices are worth explaining.
Why jittered backoff? If you run 40 taskflows at once and they all poll every 10 seconds, they hit the API in lockstep. Jitter spreads the calls out and keeps you clear of rate limits. The cap at poll_max_seconds keeps a long run from going unchecked for too long.
Why is TIMEOUT its own outcome? A timeout doesn't mean the taskflow failed. It may still be running. Keeping it separate lets on-call engineers decide whether to wait, cancel, or escalate, instead of treating every timeout as a failure.
Why catch notifier exceptions? If Slack is down, you still want the correct RunResult returned and logged. Alerting is best-effort. The run result is the source of truth.
Step 4: Pluggable notifiers
Each channel implements one method. Adding Teams, PagerDuty, or email means adding one small class.
# idmc_monitor/notifiers.py import json from typing import Protocol import requests class Notifier(Protocol): def send(self, result) -> None: ... def _summary(result) -> str: return (f"IDMC taskflow *{result.taskflow}* ended as *{result.outcome.value}* " f"(status={result.final_status}, run_id={result.run_id}, " f"{result.duration_seconds}s)") class SlackNotifier: def __init__(self, webhook_url: str): self.webhook_url = webhook_url def send(self, result) -> None: requests.post(self.webhook_url, json={"text": _summary(result)}, timeout=10) class SNSNotifier: def __init__(self, topic_arn: str, client=None): import boto3 self.topic_arn = topic_arn self.client = client or boto3.client("sns") def send(self, result) -> None: self.client.publish( TopicArn=self.topic_arn, Subject=f"IDMC {result.outcome.value}: {result.taskflow}"[:100], Message=json.dumps({ "taskflow": result.taskflow, "run_id": result.run_id, "outcome": result.outcome.value, "status": result.final_status, "duration_seconds": result.duration_seconds, }), ) SNS fans out to email, SMS, Lambda, or an SQS queue, so one notifier covers a lot of ground in AWS.
Step 5: Structured logs
Plain-text logs are hard to query. A short JSON formatter turns every extra={...} field into a searchable key.
# idmc_monitor/logging_setup.py import json import logging _STANDARD = set(vars(logging.makeLogRecord({}))) class JsonFormatter(logging.Formatter): def format(self, record: logging.LogRecord) -> str: data = {"level": record.levelname, "event": record.getMessage(), "logger": record.name, "ts": self.formatTime(record)} data.update({k: v for k, v in vars(record).items() if k not in _STANDARD}) return json.dumps(data, default=str) def configure(level: int = logging.INFO) -> None: handler = logging.StreamHandler() handler.setFormatter(JsonFormatter()) logging.basicConfig(level=level, handlers=[handler], force=True