Observability & Monitoring

OpenTelemetry Built-in OpenTelemetry integration for comprehensive tracing.

Metrics

Track key workflow metrics for performance and reliability.

Logging Detailed logging capabilities for debugging and auditing.

OpenTelemetry Integration

import { defineWorkflow } from '@identity-flow/sdk';
import { trace } from '@identity-flow/sdk/telemetry';

export default defineWorkflow('payment-processing', async (flow) => {
// Each step automatically creates a span
await flow.do('validate payment', async ({ span }) => {
// Add custom attributes to the span
span.setAttribute('payment.amount', flow.params.amount);

    return validatePayment(flow.params);

});
});

export default defineWorkflow('order-processing', async (flow) => {
  await flow.do('process order', async ({ span }) => {
    // Create nested spans for detailed tracing
    const validateSpan = span.createSpan('validate order');
    try {
      await validateOrder(flow.params);
      validateSpan.setStatus({ code: SpanStatusCode.OK });
    } catch (error) {
      validateSpan.setStatus({
        code: SpanStatusCode.ERROR,
        message: error.message
      });
      throw error;
    } finally {
      validateSpan.end();
    }
  });
});

export default defineWorkflow('user-flow', async (flow) => {
  await flow.do('process user', async ({ span }) => {
    // Add context that flows with the trace
    span.setBaggage('user.id', flow.params.userId);
    span.setBaggage('tenant.id', flow.params.tenantId);

    return processUser(flow.params);

});
});

Metrics

Execution Time

Track workflow and step execution duration.

Success Rates

Monitor success and failure rates.

Resource Usage

Track system resource utilization.

Custom Metrics

Define and track business-specific metrics.

Metric Types

import { metrics } from '@identity-flow/sdk/telemetry';

const executionCounter = metrics.createCounter('workflow_executions_total', {
  description: 'Total number of workflow executions',
  labels: ['workflow', 'status']
});

export default defineWorkflow('monitored-process', async (flow) => {
  try {
    await flow.do('process', async () => {
      // Process logic
    });
    executionCounter.add(1, {
      workflow: flow.definition.name,
      status: 'success'
    });
  } catch (error) {
    executionCounter.add(1, {
      workflow: flow.definition.name,
      status: 'error'
    });
    throw error;
  }
});

const executionTime = metrics.createHistogram('workflow_execution_time', {
  description: 'Workflow execution duration',
  buckets: [0.1, 0.5, 1, 2, 5, 10], // seconds
  labels: ['workflow']
});

export default defineWorkflow('timed-process', async (flow) => {
const start = Date.now();
try {
await flow.do('process', async () => {
// Process logic
});
} finally {
executionTime.record((Date.now() - start) / 1000, {
workflow: flow.definition.name
});
}
});

const activeWorkflows = metrics.createGauge('workflow_active_count', {
  description: 'Number of currently active workflows',
  labels: ['type']
});

export default defineWorkflow('tracked-process', async (flow) => {
  activeWorkflows.inc({ type: flow.definition.name });
  try {
    await flow.do('process', async () => {
      // Process logic
    });
  } finally {
    activeWorkflows.dec({ type: flow.definition.name });
  }
});

Detailed logging is essential for debugging and understanding workflow execution. IdentityFlow provides standard logging methods (flow.log, flow.info, flow.warn, flow.error, etc.) that automatically associate log messages with the current workflow instance and step. See the Error Tracing feature page for how logs contribute to debugging.

export default defineWorkflow('order-processing', async (flow) => {
  flow.debug('Starting order processing', {
    orderId: flow.params.orderId
  });

try {
await flow.do('process order', async () => {
flow.info('Processing order details', {
items: flow.params.items.length
});
return processOrder(flow.params);
});
} catch (error) {
flow.error('Order processing failed', {
error: error.message
});
throw error;
}
});

export default defineWorkflow('payment-processing', async (flow) => {
  await flow.do('process payment', async ({ log }) => {
    log.info('Processing payment', {
      amount: flow.params.amount,
      currency: flow.params.currency,
      method: flow.params.method,
      timestamp: new Date().toISOString()
    });

    try {
      const result = await processPayment(flow.params);
      log.info('Payment processed', {
        transactionId: result.id,
        status: result.status
      });
      return result;
    } catch (error) {
      log.error('Payment failed', {
        error: error.message,
        code: error.code,
        retryable: error.retryable
      });
      throw error;
    }
  });
});

export default defineWorkflow('user-management', async (flow) => {
  // Set context for all logs in this workflow
  flow.setLogContext({
    userId: flow.params.userId,
    tenantId: flow.params.tenantId,
    environment: process.env.NODE_ENV
  });

await flow.do('update user', async ({ log }) => {
// Context is automatically included in all logs
log.info('Updating user profile');
return updateUser(flow.params);
});
});

Monitoring Integration

Prometheus Integration

import { metrics } from '@identity-flow/sdk/telemetry';

// Define metrics
const executionTime = metrics.createHistogram('workflow_execution_time', {
  description: 'Workflow execution duration',
  buckets: [0.1, 0.5, 1, 2, 5, 10]
});

const errorCount = metrics.createCounter('workflow_errors_total', {
  description: 'Total number of workflow errors'
});

// Use in workflow
export default defineWorkflow('monitored-process', async (flow) => {
  const start = Date.now();

  try {
    await flow.do('process', async () => {
      // Process logic
    });
  } catch (error) {
    errorCount.add(1, {
      workflow: flow.definition.name,
      error: error.code
    });
    throw error;
  } finally {
    executionTime.record((Date.now() - start) / 1000, {
      workflow: flow.definition.name
    });
  }
});

Alert Configuration

// Define alerts
metrics.createGauge('workflow_duration_seconds', {
  description: 'Workflow execution duration',
  alerting: {
    rules: [
      {
        name: 'LongRunningWorkflow',
        condition: 'workflow_duration_seconds > 300',
        duration: '5m',
        severity: 'warning',
        annotations: { summary: 'Workflow running longer than 5 minutes' },
      },
    ],
  },
});

// Error rate alerting
metrics.createGauge('workflow_error_rate', {
  description: 'Workflow error rate',
  alerting: {
    rules: [
      {
        name: 'HighErrorRate',
        condition: 'rate(workflow_errors_total[5m]) > 0.1',
        severity: 'critical',
        annotations: { summary: 'High workflow error rate detected' },
      },
    ],
  },
});