ClickHouse Dictionary Memory Overflow Sentinel

id: clickhouse-dictionary-memory-overflow-sentinel
namespace: company.olap

description: Audits ClickHouse external dictionaries for excessive RAM
  consumption, alerting OLAP teams before MEMORY_LIMIT_EXCEEDED errors degrade
  analytical queries.

concurrency:
  limit: 1

inputs:
  - id: clickhouse_endpoint
    type: STRING
    defaults: "http://clickhouse.internal:8123"
    description: HTTP endpoint URL for the ClickHouse cluster coordinator or node.

  - id: max_dict_memory_mb_warning
    type: FLOAT
    defaults: 1024.0
    description: Maximum permissible RAM footprint in megabytes for any single
      external dictionary.

  - id: slack_channel
    type: STRING
    defaults: "#clickhouse-alerts"
    description: Slack channel destination for OLAP memory governance alerts.

tasks:
  - id: audit_dictionary_memory
    type: io.kestra.plugin.scripts.python.Script
    containerImage: python:3.11-slim
    beforeCommands:
      - pip install requests
    env:
      CLICKHOUSE_URL: "{{ inputs.clickhouse_endpoint }}"
      CLICKHOUSE_USER: "{{ secret('CLICKHOUSE_USER') }}"
      CLICKHOUSE_PASSWORD: "{{ secret('CLICKHOUSE_PASSWORD') }}"
      MEMORY_LIMIT_MB: "{{ inputs.max_dict_memory_mb_warning }}"
    script: |
      import os
      import json
      import requests
      from kestra import Kestra

      endpoint = os.getenv('CLICKHOUSE_URL').rstrip('/')
      user = os.getenv('CLICKHOUSE_USER')
      password = os.getenv('CLICKHOUSE_PASSWORD')
      limit_mb = float(os.getenv('MEMORY_LIMIT_MB', 1024.0))

      headers = {}
      if user:
          headers['X-ClickHouse-User'] = user
      if password:
          headers['X-ClickHouse-Key'] = password

      query = """
      SELECT 
          name,
          type,
          status,
          bytes_allocated,
          element_count,
          round(bytes_allocated / (1024 * 1024), 2) as memory_mb
      FROM system.dictionaries
      FORMAT JSON;
      """

      response = requests.post(
          f"{endpoint}/",
          params={'query': query},
          headers=headers,
          timeout=15
      )
      data = response.json()
      rows = data.get('data', [])

      flagged_dictionaries = []

      for r in rows:
          mem_mb = float(r.get('memory_mb', 0.0))
          status = r.get('status', 'LOADED')

          if mem_mb >= limit_mb or status == 'FAILED':
              flagged_dictionaries.append({
                  'name': r.get('name'),
                  'type': r.get('type'),
                  'status': status,
                  'memory_mb': mem_mb,
                  'element_count': r.get('element_count', 0)
              })

      total_dicts = len(rows)
      flagged_count = len(flagged_dictionaries)
      has_overflow = flagged_count > 0
      max_memory_mb = max([d['memory_mb'] for d in flagged_dictionaries], default=0.0)

      Kestra.outputs({
          'total_dictionaries_audited': total_dicts,
          'flagged_dictionaries_count': flagged_count,
          'has_memory_overflow': has_overflow,
          'max_dictionary_memory_mb': max_memory_mb,
          'flagged_dictionary_names': [d['name'] for d in flagged_dictionaries]
      })

      report = {
          'endpoint': endpoint,
          'total_dictionaries': total_dicts,
          'flagged_count': flagged_count,
          'flagged_dictionaries': flagged_dictionaries
      }

      with open('clickhouse_dictionary_report.json', 'w') as f:
          json.dump(report, f, indent=2)

    outputFiles:
      - clickhouse_dictionary_report.json

  - id: check_memory_threshold
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.audit_dictionary_memory.vars.has_memory_overflow }}"
    then:
      - id: notify_slack_olap
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "channel": "{{ inputs.slack_channel }}",
            "text": "ClickHouse Dictionary Memory Alert: {{ outputs.audit_dictionary_memory.vars.flagged_dictionaries_count }} dictionary(ies) exceeded {{ inputs.max_dict_memory_mb_warning }} MB RAM or failed to load.\n• Flagged Dictionaries: {{ outputs.audit_dictionary_memory.vars.flagged_dictionary_names | join(', ') }}\n• Peak Memory Allocation: {{ outputs.audit_dictionary_memory.vars.max_dictionary_memory_mb }} MB\n• Remediation: Switch memory-heavy dictionaries from `LAYOUT(FLAT())` to `LAYOUT(HASHED_ARRAY())` or `CACHE` layout to mitigate OOM risks."
          }
    else:
      - id: log_dictionaries_healthy
        type: io.kestra.plugin.core.log.Log
        message: "All {{ outputs.audit_dictionary_memory.vars.total_dictionaries_audited
          }} ClickHouse external dictionaries are operating within safe memory
          bounds."

  - id: export_dict_manifest
    type: io.kestra.plugin.core.execution.Labels
    labels:
      total_dictionaries: "{{
        outputs.audit_dictionary_memory.vars.total_dictionaries_audited | string
        }}"
      overflow_count: "{{
        outputs.audit_dictionary_memory.vars.flagged_dictionaries_count | string
        }}"
      olap_health: "{{ outputs.audit_dictionary_memory.vars.has_memory_overflow ?
        'MEMORY_OVERFLOW' : 'HEALTHY' }}"

triggers:
  - id: hourly_dict_memory_audit
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 * * * *"