Protobuf Binary Contract and Schema Compliance Gate

id: protobuf-binary-contract-and-schema-compliance-gate
namespace: company.data

labels:
  env: production
  domain: data-engineering
  compliance: schema-governance

inputs:
  - id: schema_type_name
    type: STRING
    defaults: company.telemetry.SensorMetricBatch
    description: Fully-qualified Protobuf message type to decode and validate
      against the descriptor.

  - id: min_expected_records
    type: INT
    defaults: 1
    description: Minimum number of valid Protobuf records required to pass the
      ingestion compliance gate.

tasks:
  - id: stage_schema_and_payload
    type: io.kestra.plugin.scripts.shell.Commands
    description: Synthesize the canonical Protobuf contract and compile a binary
      descriptor set using protoc.
    containerImage: python:3.11-slim
    taskRunner:
      type: io.kestra.plugin.core.runner.Process
    outputFiles:
      - schema.desc
      - payload.pb
    commands:
      - pip install --quiet protobuf grpcio-tools
      - |
        cat << 'EOF' > schema.proto
        syntax = "proto3";

        package company.telemetry;

        message SensorReading {
          string device_id = 1;
          string location = 2;
          double temperature_celsius = 3;
          double humidity_percent = 4;
          int64 recorded_at_epoch_ms = 5;
        }

        message SensorMetricBatch {
          string batch_id = 1;
          int64 generated_at_epoch_ms = 2;
          repeated SensorReading readings = 3;
        }
        EOF
      - python -m grpc_tools.protoc --proto_path=.
        --descriptor_set_out=schema.desc schema.proto
      - |
        python3 -c "
        import schema_pb2, time

        batch = schema_pb2.SensorMetricBatch()
        batch.batch_id = 'batch-2026-prod-001'
        batch.generated_at_epoch_ms = int(time.time() * 1000)

        for i in range(1, 4):
            reading = batch.readings.add()
            reading.device_id = f'sensor-edge-{i:03d}'
            reading.location = 'datacenter-iad-rack-04'
            reading.temperature_celsius = 21.4 + (i * 0.3)
            reading.humidity_percent = 44.5 + (i * 0.2)
            reading.recorded_at_epoch_ms = int(time.time() * 1000)

        with open('payload.pb', 'wb') as f:
            f.write(batch.SerializeToString())
        "

  - id: decode_protobuf_stream
    type: io.kestra.plugin.serdes.protobuf.ProtobufToIon
    description: Deserialize binary Protobuf payload using the compiled descriptor
      with strict schema enforcement.
    from: "{{ outputs.stage_schema_and_payload.outputFiles['payload.pb'] }}"
    descriptorFile: "{{ outputs.stage_schema_and_payload.outputFiles['schema.desc'] }}"
    typeName: "{{ inputs.schema_type_name }}"
    delimited: false
    errorOnUnknownFields: true

  - id: evaluate_contract_compliance
    type: io.kestra.plugin.core.flow.If
    description: Verify decoded record count against minimum threshold and route to
      alert or warehouse landing.
    condition: "{{ outputs.decode_protobuf_stream.size < inputs.min_expected_records }}"
    then:
      - id: alert_contract_breach
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Dispatch incident notification to Slack channel when payload is
          empty or violates contract.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "CRITICAL: Protobuf schema compliance breach in flow {{ flow.id }} (execution {{ execution.id }}). Expected at least {{ inputs.min_expected_records }} records, but decoded {{ outputs.decode_protobuf_stream.size }} records for message type {{ inputs.schema_type_name }}."
          }

      - id: fail_compliance_gate
        type: io.kestra.plugin.core.execution.Fail
        description: Terminate execution with failure status to prevent downstream
          consumption of invalid binary data.
        errorMessage: "Protobuf ingestion breached contract: expected at least {{
          inputs.min_expected_records }} records, but received {{
          outputs.decode_protobuf_stream.size }}."
    else:
      - id: convert_ion_to_json
        type: io.kestra.plugin.serdes.json.IonToJson
        description: Convert validated internal Amazon Ion records to newline-delimited
          JSON for analytical landing.
        from: "{{ outputs.decode_protobuf_stream.uri }}"

  - id: audit_compliance_certification
    type: io.kestra.plugin.core.log.Log
    description: Record compliance audit trail with message count, schema
      definition, and storage destination.
    message: "Protobuf contract certified: message_type={{ inputs.schema_type_name
      }}, records_decoded={{ outputs.decode_protobuf_stream.size }},
      ion_storage_uri={{ outputs.decode_protobuf_stream.uri }},
      status=COMPLIANT"

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