API Health and Contract Monitor

id: api-health-contract-monitor
namespace: company.ops

description: |
  Operational API health and contract monitor: executes scheduled probes, validates HTTP status and latency,
  enforces JSON response schema contracts, and alerts on degradation.

triggers:
  - id: inbound_probe_webhook
    type: io.kestra.plugin.core.trigger.Webhook
    key: "{{ secret('API_PROBE_WEBHOOK_KEY') }}"

  - id: scheduled_five_minute_probe
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "*/5 * * * *"
    disabled: true

inputs:
  - id: endpoint_url
    type: STRING
    defaults: "https://api.internal.acme/v1/health"
    description: Target REST API endpoint to probe.

  - id: expected_status_code
    type: INT
    defaults: 200
    description: Expected HTTP response status code.

  - id: max_acceptable_latency_ms
    type: INT
    defaults: 800
    description: Maximum acceptable response latency in milliseconds.

  - id: retry_attempts
    type: INT
    defaults: 3
    description: Number of retry attempts on transient network timeouts before failing.

  - id: environment_tier
    type: SELECT
    values: [ Production, Staging ]
    defaults: Production
    description: Target deployment tier under observation.

tasks:
  - id: init_probe_session
    type: io.kestra.plugin.core.log.Log
    message: |
      Initializing API Health and Contract Monitor:
      Execution ID: {{ execution.id }}
      Endpoint: {{ inputs.endpoint_url }} ({{ inputs.environment_tier }})
      Expected Status: {{ inputs.expected_status_code }}
      Max Latency Threshold: {{ inputs.max_acceptable_latency_ms }} ms
      Retries Configured: {{ inputs.retry_attempts }}

  - id: execute_api_health_probe
    type: io.kestra.plugin.scripts.python.Commands
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    containerImage: python:3.11-slim
    env:
      ENDPOINT: "{{ inputs.endpoint_url }}"
      EXPECTED_STATUS: "{{ inputs.expected_status_code }}"
      MAX_LATENCY: "{{ inputs.max_acceptable_latency_ms }}"
    beforeCommands:
      - pip install --no-cache-dir pydantic
    commands:
      - python probe_endpoint.py
    inputFiles:
      probe_endpoint.py: |
        import json
        import os
        import time
        import sys

        endpoint = os.environ.get("ENDPOINT", "https://api.internal.acme/v1/health")
        expected_status = int(os.environ.get("EXPECTED_STATUS", "200"))
        max_latency = int(os.environ.get("MAX_LATENCY", "800"))

        print(f"Probing API endpoint '{endpoint}' (Max Latency: {max_latency}ms)...")

        # Simulating live HTTP probe response and schema contract evaluation
        start_time = time.time()
        simulated_latency_ms = 245
        time.sleep(0.05)
        response_status = 200

        # Required JSON response schema contract
        sample_response_body = {
            "status": "UP",
            "version": "v3.4.1",
            "services": {
                "database": "CONNECTED",
                "redis_cache": "CONNECTED"
            },
            "timestamp": "2026-10-13T01:50:00Z"
        }

        required_keys = ["status", "version", "services", "timestamp"]
        missing_keys = [k for k in required_keys if k not in sample_response_body]
        schema_valid = len(missing_keys) == 0

        status_ok = response_status == expected_status
        latency_ok = simulated_latency_ms <= max_latency
        is_healthy = status_ok and latency_ok and schema_valid

        probe_report = {
            "endpoint": endpoint,
            "http_status": response_status,
            "latency_ms": simulated_latency_ms,
            "status_ok": status_ok,
            "latency_ok": latency_ok,
            "schema_contract_valid": schema_valid,
            "missing_contract_keys": missing_keys,
            "is_healthy": is_healthy
        }

        with open("probe_report.json", "w") as f:
            json.dump(probe_report, f, indent=2)

        print(f"Probe Result: Status={response_status}, Latency={simulated_latency_ms}ms, Contract Valid={schema_valid} -> Healthy: {is_healthy}")
        print(f"::{{'outputs': {{'http_status': {response_status}, 'latency_ms': {simulated_latency_ms}, 'schema_valid': {str(schema_valid).lower()}, 'is_healthy': {str(is_healthy).lower()}}}}}::")

  - id: evaluate_health_and_contract
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.execute_api_health_probe.vars.is_healthy }}"
    then:
      - id: log_healthy_endpoint
        type: io.kestra.plugin.core.log.Log
        message: |
          API Endpoint {{ inputs.endpoint_url }} is HEALTHY:
          - Response Code: {{ outputs.execute_api_health_probe.vars.http_status }}
          - Latency: {{ outputs.execute_api_health_probe.vars.latency_ms }} ms
          - JSON Schema Contract: Validated

    else:
      - id: notify_slack_endpoint_degraded
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "🚨 *API HEALTH PROBE DEGRADED*\n*Endpoint:* `{{ inputs.endpoint_url }}` ({{ inputs.environment_tier }})\n*Status Code:* `{{ outputs.execute_api_health_probe.vars.http_status }}` (Expected: `{{ inputs.expected_status_code }}`)\n*Latency:* `{{ outputs.execute_api_health_probe.vars.latency_ms }} ms` (Max Allowed: `{{ inputs.max_acceptable_latency_ms }} ms`)\n*Schema Contract Valid:* `{{ outputs.execute_api_health_probe.vars.schema_valid }}`\n*Action:* Service degradation alert paged to on-call engineers."
          }