Governance Rules¶

Governance rules enforce standards, naming conventions, and policies across your streaming projects. They're checked during streamt validate.

Overview¶

Rules are defined in stream_project.yml:

rules:
  topics:
    min_partitions: 6
    naming_pattern: "^[a-z]+\\.[a-z]+\\.v[0-9]+$"

  models:
    require_description: true
    require_tests: true

  sources:
    require_schema: true

  security:
    require_classification: true

Topic Rules¶

Control Kafka topic configuration:

rules:
  topics:
    # Partition requirements
    min_partitions: 3
    max_partitions: 128

    # Replication requirements
    min_replication_factor: 2

    # Naming conventions
    naming_pattern: "^[a-z]+\\.[a-z-]+\\.v[0-9]+$"
    forbidden_prefixes:
      - "_"
      - "test"
      - "tmp"
      - "dev"

    # Maximum replication
    max_replication_factor: 3

    # Suffix restrictions
    forbidden_suffixes:
      - "_test"
      - "_temp"
      - "_debug"

    # Required configurations
    required_config:
      - retention.ms
      - min.insync.replicas

Rule Reference¶

Rule	Type	Description
`min_partitions`	int	Minimum partition count
`max_partitions`	int	Maximum partition count
`min_replication_factor`	int	Minimum replication factor
`naming_pattern`	regex	Required topic name pattern
`forbidden_prefixes`	list	Disallowed name prefixes
`max_replication_factor`	int	Maximum replication factor
`forbidden_suffixes`	list	Disallowed name suffixes
`required_config`	list	Configs that must be set

Naming Pattern Examples¶

# Domain.entity.version pattern
naming_pattern: "^[a-z]+\\.[a-z-]+\\.v[0-9]+$"
# Matches: orders.created.v1, payments.processed.v2
# Rejects: Orders.Created, orders_created_v1

# Environment prefix pattern
naming_pattern: "^(prod|staging|dev)\\.[a-z]+\\.[a-z]+$"
# Matches: prod.orders.raw, staging.users.events

# Team namespace pattern
naming_pattern: "^team-[a-z]+\\.[a-z-]+$"
# Matches: team-payments.transactions, team-fraud.alerts

Model Rules¶

Enforce documentation and quality standards:

rules:
  models:
    # Documentation requirements
    require_description: true
    require_owner: true

    # Testing requirements
    require_tests: true

    # Complexity limits
    max_dependencies: 10

Rule Reference¶

Rule	Type	Description
`require_description`	bool	Models must have description
`require_owner`	bool	Models must have owner
`require_tests`	bool	Models must have tests
`max_dependencies`	int	Max upstream dependencies

Source Rules¶

Ensure sources are well-documented:

rules:
  sources:
    # Schema requirements
    require_schema: true

    # Freshness monitoring
    require_freshness: true

Rule Reference¶

Rule	Type	Description
`require_schema`	bool	Must reference Schema Registry
`require_freshness`	bool	Must have freshness SLA

Security Rules¶

Enforce data protection policies:

rules:
  security:
    # Classification requirements
    require_classification: true

    # Masking requirements
    sensitive_columns_require_masking: true

Rule Reference¶

Rule	Type	Description
`require_classification`	bool	Columns must have classification
`sensitive_columns_require_masking`	bool	Sensitive data must be masked

Classification Levels¶

Level	Description	Typical Rules
`public`	Open data	No restrictions
`internal`	Internal use	No external exposure
`confidential`	Business sensitive	Limited access
`sensitive`	PII, personal data	Masking required
`highly_sensitive`	Regulated (PCI, HIPAA)	Encryption + audit

Data Residency Rules¶

Control where data can be processed:

rules:
  data_residency:
    allowed_regions:
      - EU
      - US

Models and sources declare their region:

models:
  - name: eu_orders
    region: EU
    sql: ...

sources:
  - name: eu_events
    region: EU
    topic: eu.events.v1

The validator errors if a model or source declares a region not in allowed_regions.

Rule Reference¶

Rule	Type	Description
`allowed_regions`	list	Regions that models/sources may declare

Schema Versioning¶

Models support a version field for managing schema evolution:

models:
  - name: users
    version: 1
    sql: SELECT id, name FROM {{ source("raw_users") }}

  - name: users
    version: 2
    sql: SELECT id, name, email FROM {{ source("raw_users") }}

The validator checks: - Duplicate versions — error if same model name has duplicate version numbers - Version gaps — warning if version N exists without version N-1

Validation Output¶

When rules are violated:

$ streamt validate

✗ Project 'my-pipeline' has validation errors

Errors:
  ✗ Model 'orders_raw_v2' violates topic naming pattern
    Expected: ^[a-z]+\.[a-z-]+\.v[0-9]+$
    Got: orders_raw_v2

  ✗ Model 'customer_metrics' missing required tests
    Rule: require_tests = true

Warnings:
  ⚠ Source 'events' missing freshness configuration
    Rule: require_freshness = true

  ⚠ Column 'email' in 'users' appears to be PII but lacks classification
    Rule: pii_column_patterns includes 'email'

Summary: 2 errors, 2 warnings

Strict Mode¶

In strict mode, warnings become errors:

$ streamt validate --strict

Use in CI/CD to enforce all rules.

Environment-Specific Rules¶

Use different rules per environment:

stream_project.yml

rules:
  topics:
    min_partitions: 3
    min_replication_factor: 1

Production

export TOPIC_MIN_PARTITIONS=6
export TOPIC_MIN_RF=3
streamt validate

Development

export TOPIC_MIN_PARTITIONS=1
export TOPIC_MIN_RF=1
streamt validate

Best Practices¶

1. Start Permissive, Tighten Over Time¶

# Week 1: Basic requirements
rules:
  models:
    require_description: true

# Month 1: Add testing requirements
rules:
  models:
    require_description: true
    require_tests: true

# Quarter 1: Full governance
rules:
  models:
    require_description: true
    require_tests: true
    require_owner: true
  security:
    require_classification: true

2. Document Rule Rationale¶

rules:
  topics:
    # Minimum 6 partitions for adequate parallelism
    # across our 6-node Kafka cluster
    min_partitions: 6

    # Pattern: domain.entity.version
    # Examples: orders.created.v1, payments.processed.v2
    naming_pattern: "^[a-z]+\\.[a-z-]+\\.v[0-9]+$"

3. Use CI/CD Enforcement¶

.github/workflows/validate.yml

- name: Validate streaming project
  run: |
    streamt validate --strict

4. Regular Audits¶

# Generate compliance report
streamt validate --format json > compliance-report.json

# Check for new violations
streamt validate --strict || notify_team "Governance violations found"