SMB Share Ingestion Stall and Quarantine Sentinel

id: smb-share-ingestion-stall-and-quarantine-sentinel
namespace: company.storage

inputs:
  - id: smb_host
    type: STRING
    defaults: localhost
    description: Hostname or IP address of the SMB file server.

  - id: smb_port
    type: STRING
    defaults: "445"
    description: Port of the SMB file service (default 445).

  - id: smb_username
    type: STRING
    defaults: smbuser
    description: Service account username authorized to read and move share files.

  - id: smb_share_path
    type: STRING
    defaults: /shared/incoming/
    description: Directory URI on the SMB share monitored for incoming data files.

  - id: quarantine_share_path
    type: STRING
    defaults: /shared/quarantine/
    description: Destination directory URI where stalled or malformed drops are moved.

  - id: file_pattern
    type: STRING
    defaults: ".*\\.csv$"
    description: Regular expression filter applied against file names in the target share.

  - id: min_expected_files
    type: INT
    defaults: 1
    description: Minimum number of batch files expected before raising a stall alert.

  - id: simulate_local_mode
    type: BOOL
    defaults: true
    description: When true, runs synthetic inspection for zero-dependency local testing.

variables:
  default_channel: "#storage-ops"

triggers:
  - id: schedule_periodic_probe
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "*/15 * * * *"

tasks:
  - id: probe_mode_router
    type: io.kestra.plugin.core.flow.If
    condition: "{{ inputs.simulate_local_mode }}"
    then:
      - id: simulate_smb_inspection
        type: io.kestra.plugin.core.execution.SetVariables
        variables:
          detected_files_count: 0
          corrupt_files_count: 1
          quarantine_target_name: corrupt_zero_byte_feed.csv
    else:
      - id: list_smb_incoming_files
        type: io.kestra.plugin.fs.smb.List
        host: "{{ inputs.smb_host }}"
        port: "{{ inputs.smb_port }}"
        username: "{{ inputs.smb_username }}"
        password: "{{ secret('SMB_PASSWORD') }}"
        from: "{{ inputs.smb_share_path }}"
        regExp: "{{ inputs.file_pattern }}"
        sort: LAST_MODIFIED_ASC

  - id: classify_ingestion_health
    type: io.kestra.plugin.core.execution.Labels
    labels:
      health_status: >-
        {{ (inputs.simulate_local_mode ?
        (outputs.simulate_smb_inspection.vars.corrupt_files_count > 0 ?
        'WARNING_CORRUPT' :
        (outputs.simulate_smb_inspection.vars.detected_files_count <
        inputs.min_expected_files ? 'CRITICAL_STALL' : 'HEALTHY')) :
        ((outputs.list_smb_incoming_files.files | default([])) | length <
        inputs.min_expected_files ? 'CRITICAL_STALL' : 'HEALTHY')) }}

  - id: handle_health_tier
    type: io.kestra.plugin.core.flow.Switch
    value: "{{ taskrun.labels.health_status }}"
    cases:
      CRITICAL_STALL:
        - id: alert_critical_stall
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {
              "channel": "{{ vars.default_channel }}",
              "text": "CRITICAL: SMB Share Ingestion Stall detected on host {{ inputs.smb_host }} share {{ inputs.smb_share_path }}. Zero or insufficient files matched pattern {{ inputs.file_pattern }}. Execution ID: {{ execution.id }}"
            }
      WARNING_CORRUPT:
        - id: alert_warning_corrupt
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {
              "channel": "{{ vars.default_channel }}",
              "text": "WARNING: Corrupted or zero-byte drop detected on SMB share {{ inputs.smb_share_path }}. Initiating automated quarantine move. Execution ID: {{ execution.id }}"
            }
        - id: quarantine_corrupt_files
          type: io.kestra.plugin.core.flow.If
          condition: "{{ not inputs.simulate_local_mode }}"
          then:
            - id: move_to_quarantine
              type: io.kestra.plugin.fs.smb.Move
              host: "{{ inputs.smb_host }}"
              port: "{{ inputs.smb_port }}"
              username: "{{ inputs.smb_username }}"
              password: "{{ secret('SMB_PASSWORD') }}"
              from: "{{ inputs.smb_share_path }}corrupt_zero_byte_feed.csv"
              to: "{{ inputs.quarantine_share_path }}corrupt_zero_byte_feed.csv"
      HEALTHY:
        - id: notify_healthy_status
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {
              "channel": "{{ vars.default_channel }}",
              "text": "HEALTHY: SMB Share Ingestion operating normally on {{ inputs.smb_host }}. Active files verified. Execution ID: {{ execution.id }}"
            }

  - id: log_audit_summary
    type: io.kestra.plugin.core.log.Log
    message: |
      SMB Ingestion Sentinel Audit Summary:
      Host: {{ inputs.smb_host }}:{{ inputs.smb_port }}
      Share Path: {{ inputs.smb_share_path }}
      Quarantine Path: {{ inputs.quarantine_share_path }}
      Pattern: {{ inputs.file_pattern }}
      Health Status: {{ taskrun.labels.health_status }}
      Execution ID: {{ execution.id }}
      Timestamp: {{ execution.startDate }}