The source definition

One YAML file per source, and it carries the whole stream: identity, where the records come from, whether they are transformed, the table they land in, and the standard views they expose. What a Source IS lives in source.md; this is the file itself.

A full source

# Example: sources/filebeat.yaml
source: filebeat                        # The _source label -- immutable identifier
display_name: Filebeat
description: Elastic Filebeat log collector
# from_catalogue_item: okta             # Catalogue entry name. Omitted when the source was written by hand
# from_catalogue_intake: beats          # Intake selected for that entry: beats | receiver | fetcher
state: active                           # active | dormant | disabled (see source-registry.md)

# --- Flow (optional; see source-flow.md) ---
transport: bus                          # bus (a broker holds records between stages) | direct (gRPC,
                                        # nothing stored). Omit to take the deployment default
archive: false                          # Keep the raw record as it arrived. On the bus the archiver
                                        # reads the landing topic; on direct it is sent the record
                                        # by the receiver, so a fetched record, in its own source or
                                        # routed to another, is archived only on the bus

# --- Header ---
# Common schema header. A new source starts as JUST this header -- the common
# fields for the selected type and version. Source-specific fields are added
# incrementally.
header:
  type: timeseries                      # Profile name (timeseries, minimal, passthrough)
  version: 1.0.0                        # Common header version (semver)

# --- Origin: receiver match or when, OR fetcher, OR external (exactly one) ---
# A receiver-based source is identified by the always-present receiver pool:
# the receiver evaluates match rules and sets _source in the JSON payload.
match:
  field: tags.collector.type            # Dotted path into the incoming JSON payload
  operator: equals                      # equals (default) | not_equals | exists | includes | starts_with | ends_with
  value: filebeat                       # Operand (unused when operator is exists or always)
  # `always` matches every record and is reserved for the `main` source,
  # which is how the main flow is defined - see source-flow.md.
# A match may instead group tests, nested up to four deep, and negate any of them:
# match:
#   all:
#     - {field: data_stream.dataset, value: cisco_ios.log}
#     - any:
#         - {field: agent.type, value: filebeat}
#         - {field: agent.type, value: elastic-agent}
#     - {field: event.dataset, operator: exists, negate: true}
# Or carry a CEL Fast condition in place of match:
# when: "data_stream.dataset == 'cisco_ios.log' && agent.type == 'filebeat'"

# --- Compatibility profiles (optional) ---
# Unset binds the deployment's list, [] binds none.
# compat_profiles:
#   - profile: elastic-agent
#     opt_in: [org-namespace]           # Rows the profile leaves off until a binding names them
# aliases:                              # Per profile, the values its lookup maps to this source
#   elastic-agent: [cisco_ios.log]

# A fetcher-based source has no match rule. The engine deploys one dfe-fetcher
# instance named for the source, with this stanza compiled into it, when the
# source is active and deployed, and removes it when the source is not.
# A source is ONE connector type across as many connections as it likes -- one
# schema per source is what keeps its table one shape -- so a second type is
# refused on the write and again when the fetcher config is composed.
# `topic: own` needs a schema (meta_schema, derived_schema or additional_fields),
# because a source with none can never deploy. The write is refused until the
# topic is `main` or a schema is defined.
# fetcher:
#   source_type: okta                   # A family the deployed fetcher ships (apps.yaml source_types)
#   topic: own                          # own: this source's topic and table | main: the shared main topic and table
#   config:                             # The fetcher's own per-type stanza, verbatim
#     tenant_url: https://example.okta.com
#     credential_secret: vault:kv/dfe/okta:token       # never a literal. vault:/bao:/openbao: take <mount>/<path>:<key>, first segment the mount; env: takes a variable name and file: a path, neither with a :key
#     services:
#       - name: system_log
#     interval_secs: 300
#   routes:                             # Fetched records that belong to ANOTHER source
#     - match: {field: eventType, operator: starts_with, value: policy.}
#       source: okta-policy             # Their landing, table and transform, not this source's
#   org: acme                           # The registered org every fetched record belongs to, unset is the default org

# An external source is written by the customer's own pipeline, which produces to
# <source>_land past the receiver. Bus only. See external-pipelines.md.
# external:
#   credential_ref: vault:kv/acme/kafka/vector   # A reference only: vault:, bao:, openbao:, env: or file:
#   org: acme                           # The registered org its records belong to, unset is the default org
#   allowed_org_ids: []                 # Further tenant ids a record may keep in _org_id, any other is replaced
#   allowed_sources: []                 # Further sources a record may keep in _source, any other is replaced

# --- Topics (derived, not configured) ---
# topic_land: filebeat_land             # Auto: {_source}_land
# topic_load: filebeat_load             # Auto: {_source}_load (only if transform exists)

# --- Transform (optional) ---
# If present, records flow _land -> transform -> _load; if absent, _land -> loader.
transform:
  engine: vector                        # A catalogued transform app, by engine name (apps.yaml)
  variant: filebeat.okta.default        # The compiled-in program, where the app offers a catalogue
  # mode: agentless                     # elastic only: detect each event's program, take no variant
  # detect:                             # agentless only, all optional
  #   unmatched: fallback               # passthrough (default), fallback or stop
  #   fallback: filebeat.cisco_asa.default   # registry name, required exactly when unmatched is fallback
  #   candidates: []                    # registry names detection may name, empty means every source
  config_file: /etc/vector/filebeat.yaml
  env: {}                               # Per-transform ENV overrides
  files: []                             # Enrichment files (CSV, MMDB)
# engine: external reads _land in dfe-external-<source> and writes _load with the
# customer's own pipeline. It takes none of the keys above, only the four that
# `external` takes (credential_ref, org, allowed_org_ids, allowed_sources).

# --- Views (optional) ---
# Naming-standard views this source exposes (sigma, ecs, cim, ocsf).
# One entry per standard; inline custom_mappings WIN over the field_map pin.
views:
  - standard: sigma
    field_map: sigma/windows            # FieldMap registry pin ("standard/name" or bare name)
    taxonomy: windows                   # Sigma logsource product binding (sigma views only)
    custom_mappings:                    # Per-source overrides (win over field_map)
      CommandLine: command_line

# --- Schema (mandatory) ---
# The source owns its ClickHouse table schema: one source = one table = one schema.
schema:
  meta_schema: logs_beats_filebeat      # Base field definitions (YAML)
  meta_schema_version: 1.0.0
  derived_schema: derived/beats/filebeat_auth   # Column SELECTION from the meta schema (optional)
  derived_schema_version: 1.0.0                 # Unset follows the derived schema's own current
  additional_fields: filebeat/add       # Extra fields, indexes (optional)
  ttl_days: 90                          # Data retention
  engine: MergeTree                     # Base variant only (MergeTree, ReplacingMergeTree(...), ...)
                                        # The Replicated prefix is added at DDL time

match.field is a dotted path into the record as it arrives, which is what the receiver's router splits on. It is not a ClickHouse column reference: a _json. prefix would be read as a first path segment that no payload has.

ttl_days is optional. A source that leaves it unset gets the deployment default (the admin's console value, else DFE_ENGINE_CLICKHOUSE_DEFAULT_TTL_DAYS, 90 days shipped, 0 = none); a value here wins over that default and over the table's dfe-schemas definition. The next schema apply moves an existing table onto a changed value with ALTER TABLE ... MODIFY TTL, and shortening it expires the rows older than the new value.

derived_schema NARROWS. The table's columns become exactly the names its select list carries, in that order, after the common header. A name the base meta schema does not define is a hard error, never an appended column, and index is the only per-column key it may set -- type, expr, comment and attribute all resolve from the base. additional_fields is still the append layer. A derived schema also carries capture_json and capture_raw for this table, both on by default; turning them off tells dfe-loader to stop populating _json and _raw without removing either column. Its CRUD surface is /api/v1/schemas/definitions/derived/<group>/<name>.

The 2.1 keys sigma, mapping_standards and field_mappings were a clean break in 2.2, not a deprecation -- a write carrying one is rejected with a "removed in 2.2" error. Declare naming-standard views under views instead.

The smallest source that works

source: syslog
display_name: Syslog
match:
  field: tags.collector.type
  value: syslog
header:
  type: timeseries
  version: 1.0.0
schema:
  ttl_days: 90

No fetcher, no transform. The receiver matches and routes to syslog_land, the loader picks it up and inserts into {db}.syslog using the common timeseries header schema. Fields are added incrementally later.

Schema ownership

Each source owns exactly one ClickHouse table schema, 1:1:

Source: filebeat
  └── Schema: logs_beats_filebeat
        ├── meta_schema.yaml       (base columns, types, use cases)
        ├── derived_schema.yaml    (which of those columns this table keeps)
        └── additional_fields.yaml (enrichment fields)

The type registry (dfe-schemas registries/types.yaml) maps primitives to ClickHouse types and is global, not per-source. The schema section feeds the SchemaBuilder / ClickHouseSchema pipeline. Create and migrate are triggered from the source's state rather than from a standalone schema config.

# meta_schema.yaml example (excerpt)
columns:
  - name: user_name
    type: string
    use_case: dimension
    comment: "@source: first(user_id/uid/id)"
  - name: source_ip
    type: ip
    use_case: range
  - name: message
    type: text
    use_case: word_search

Loader directives in comment survive into the ClickHouse column comment. The full type system, the primitive-to-ClickHouse mapping and the use cases are in schema.md.

Source event	Schema action
created / active	CREATE TABLE IF NOT EXISTS (idempotent)
schema YAML updated	ALTER TABLE via SchemaModifier (add/modify columns)
dormant	none -- the table is retained, pre-positioned for reactivation
disabled	none -- routing and transforms turn off, the table and its data stay
deleted	none -- the table is retained. Drop it by hand if the data is not wanted

Migrating off Elastic starts from the index template rather than from nothing: the converter maps Elastic types to primitives and analyzers to use cases and emits the schema YAML, which the source then names. The mapping table is in schema.md.

What a write is validated against

What	Rule
_source label	unique, and [a-z]([a-z0-9-]*[a-z0-9])? up to 40 chars
match rule	no conflict across non-disabled sources (an identical match or when). A dormant source HOLDS its match, only disabling releases it
match and when	compile to rows scalo.rules accepts under the receiver's CEL Fast profile, at most 32 of them, else 422 invalid_rule with each problem's field and span. always only on the reserved main source
compatibility profiles	each binding names a stored profile, a version it publishes and only opt-in rows it offers, one version of a profile per deployment, and every alias is unique per profile across non-disabled sources (409 alias_conflict)
fetcher route	must not name its own source, and compares one field with equals
fetcher connector type	one type per source, however many connections it polls; the refusal names the source and the types
flow	runnable here: the deployment offers the transport, and every app in the flow carries it
transform	engine must be one the app manifest catalogues (vrl, vector, elastic)
transform mode and detect	mode and detect only for an app whose manifest declares mode_path and detect_path (elastic). detect only with mode: agentless, which takes no variant. detect.fallback is required when unmatched is fallback and refused otherwise, and candidates holds distinct non-blank names
catalogue entry	the intake must be one the entry arrives by, the variant one it compiles to and not one a mounted contract of the deployed transform leaves out (409 missing_variant), and the source name a legal label
schema	files exist and pass SchemaBuilder validation, and carry _source as a column (injected if missing)

The connector-type rule is the source-schema alignment, enforced: one source is one schema, so it is one type, and the many connections of that type - accounts, tenants, regions, each with its own credential - all land in the one table. DFE_ENGINE_SOURCE_ALLOW_MIXED_FETCHER_TYPES=true (source.allow_mixed_fetcher_types) turns the enforcement off for a deployment that knowingly wants a mixed fetcher, and past that dial you are on your own: no suite support, no schema alignment, and dfe-fetcher's own many-to-many semantics apply to what lands in the table.