> ## Documentation Index
> Fetch the complete documentation index at: https://docs.siftstack.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Ingestion config streaming reference

> Field definitions, protobuf schemas, and examples for ingestion-config-based streaming

export const MintTable = ({columns = [], rows = [], columnWidths = []}) => {
  const pushTextWithLineBreaks = (parts, text, keyBase) => {
    const segments = String(text).split(/\\n|\n/);
    segments.forEach((segment, idx) => {
      if (segment) {
        parts.push(<span key={`${keyBase}-text-${idx}`}>{segment}</span>);
      }
      if (idx < segments.length - 1) {
        parts.push(<br key={`${keyBase}-br-${idx}`} />);
      }
    });
  };
  const parseMarkdown = text => {
    if (text === null || text === undefined) return "";
    const str = String(text);
    const parts = [];
    let lastIndex = 0;
    const pattern = /(`[^`]+`|\*\*[^*]+\*\*|\*[^*]+\*|\[([^\]]+)\]\(([^)]+)\))/g;
    let match;
    while (true) {
      match = pattern.exec(str);
      if (match === null) {
        break;
      }
      if (match.index > lastIndex) {
        pushTextWithLineBreaks(parts, str.substring(lastIndex, match.index), `before-${lastIndex}`);
      }
      const fullMatch = match[0];
      if (fullMatch.startsWith("`") && fullMatch.endsWith("`")) {
        parts.push(<code key={match.index}>{fullMatch.slice(1, -1)}</code>);
      } else if (fullMatch.startsWith("**") && fullMatch.endsWith("**")) {
        parts.push(<strong key={match.index}>{fullMatch.slice(2, -2)}</strong>);
      } else if (fullMatch.startsWith("*") && fullMatch.endsWith("*")) {
        parts.push(<em key={match.index}>{fullMatch.slice(1, -1)}</em>);
      } else if (fullMatch.startsWith("[")) {
        const linkText = match[2];
        const linkUrl = match[3];
        parts.push(<a key={match.index} href={linkUrl} className="text-black-600 dark:text-black-400">
            {linkText}
          </a>);
      }
      lastIndex = pattern.lastIndex;
    }
    if (lastIndex < str.length) {
      pushTextWithLineBreaks(parts, str.substring(lastIndex), `tail-${lastIndex}`);
    }
    if (parts.length > 0) {
      return parts;
    }
    const plainParts = [];
    pushTextWithLineBreaks(plainParts, str, "plain");
    return plainParts.length ? plainParts : str;
  };
  const safeColumns = Array.isArray(columns) ? columns : [];
  const safeRows = Array.isArray(rows) ? rows : [];
  const safeColumnWidths = Array.isArray(columnWidths) ? columnWidths : [];
  const hasColumnWidths = safeColumnWidths.some(w => w !== null && w !== undefined && w !== "");
  const toCssWidth = width => typeof width === "number" ? `${width}px` : String(width);
  const getColumnStyle = idx => {
    const rawWidth = safeColumnWidths[idx];
    if (rawWidth === null || rawWidth === undefined || rawWidth === "") {
      return undefined;
    }
    const width = toCssWidth(rawWidth);
    return {
      width,
      minWidth: width
    };
  };
  const containerStyle = hasColumnWidths ? undefined : {
    overflowX: "auto"
  };
  const tableStyle = hasColumnWidths ? {
    tableLayout: "fixed",
    width: "100%"
  } : {
    width: "max-content",
    minWidth: "100%"
  };
  if (!Array.isArray(columns) || !Array.isArray(rows) || !Array.isArray(columnWidths)) {
    console.warn("MintTable received invalid props:", {
      columns,
      rows,
      columnWidths
    });
  }
  if (!safeColumns.length && !safeRows.length) {
    return null;
  }
  return <div className="mint-table-container" style={containerStyle}>
      <table style={tableStyle}>
        {hasColumnWidths && <colgroup>
            {safeColumns.map((_, idx) => {
    const style = getColumnStyle(idx);
    return <col key={idx} style={style} />;
  })}
          </colgroup>}
        <thead>
          <tr>
            {safeColumns.map((col, idx) => <th key={idx} className="text-left" style={getColumnStyle(idx)}>
                <b>{parseMarkdown(col)}</b>
              </th>)}
          </tr>
        </thead>
        <tbody>
          {safeRows.map((row, rIdx) => {
    const safeRow = Array.isArray(row) ? row : [];
    return <tr key={rIdx}>
                {safeRow.map((cell, cIdx) => <td key={cIdx} style={getColumnStyle(cIdx)}>
                    {parseMarkdown(cell)}
                  </td>)}
              </tr>;
  })}
        </tbody>
      </table>
    </div>;
};

Ingestion-config-based streaming uses `IngestionConfigService.CreateIngestionConfig` to register the schema of your telemetry, then streams compact structured messages to Sift over gRPC. This approach reduces packet size and improves throughput compared to schemaless methods.

## CreateIngestionConfigRequest

```protobuf theme={null}
message CreateIngestionConfigRequest {
  string asset_name = 1 [(google.api.field_behavior) = REQUIRED];
  repeated FlowConfig flows = 2;
  string organization_id = 3 [(google.api.field_behavior) = OPTIONAL];
  string client_key = 4 [(google.api.field_behavior) = OPTIONAL];
}
```

<MintTable
  columns={['Field', 'Description']}
  columnWidths={['30%', '70%']}
  rows={[
['`asset_name`', 'The name of the asset to create. Required.'],
['`flows`', 'One or more flow configurations defining the schema. See [FlowConfig](#flowconfig).'],
['`organization_id`', 'Your organization ID. Only required if you belong to multiple organizations.'],
['`client_key`', 'An arbitrary string you choose to uniquely identify this ingestion config. Optional but strongly recommended; it simplifies lookups.'],
]}
/>

## FlowConfig

A flow is a named group of channels whose values are sent together in one request.

```protobuf theme={null}
message FlowConfig {
  string name = 1 [(google.api.field_behavior) = REQUIRED];
  repeated ChannelConfig channels = 2;
}
```

## ChannelConfig

```protobuf theme={null}
message ChannelConfig {
  string name = 1 [(google.api.field_behavior) = REQUIRED];
  string component = 2;
  string unit = 3;
  string description = 4;
  sift.common.type.v1.ChannelDataType data_type = 5 [(google.api.field_behavior) = REQUIRED];
  repeated sift.common.type.v1.ChannelEnumType enum_types = 6;
  repeated sift.common.type.v1.ChannelBitFieldElement bit_field_elements = 7;
}
```

<MintTable
  columns={['Field', 'Description']}
  columnWidths={['30%', '70%']}
  rows={[
['`name`', 'The Channel name. Required.'],
['`component`', 'Optional component label.'],
['`unit`', 'Optional unit string (for example, `km/hr`).'],
['`description`', 'Optional human-readable description.'],
['`data_type`', 'The data type of the Channel. Required.'],
['`enum_types`', 'Enum type definitions for enum-typed Channels.'],
['`bit_field_elements`', 'Bit field element definitions for bit-field-typed Channels.'],
]}
/>

### Channel ordering

The order of Channels in a `FlowConfig` must be preserved exactly when sending values in an `IngestWithConfigDataStreamRequest`. Sift attributes each value to a Channel by its position in the list.

If data is available for some Channels in a flow but not others, send `google.protobuf.Empty` in the position of the missing Channel to maintain correct ordering.

<Note>
  **Channel structure mismatch**: If the ingestion client logs `encountered a message that doesn't match any cached flows`, the Channel data sent for that flow doesn't match the flow's Channel structure as registered in `CreateIngestionConfig`. Verify that the Channel names, order, and data types you send match the flow's configured Channels.
</Note>

#### Example

Given a flow `reading` with two Channels, a `double` Channel followed by a `string` Channel:

```python theme={null}
from sift_stream_bindings import (
    ChannelConfigPy,
    ChannelDataTypePy,
    FlowConfigPy,
)

flow_config = FlowConfigPy(
    name="reading",
    channels=[
        ChannelConfigPy(
            name="mainmotor.velocity",
            unit="km/hr",
            description="vehicle speed",
            data_type=ChannelDataTypePy.Double,
            enum_types=[],
            bit_field_elements=[],
        ),
        ChannelConfigPy(
            name="log",
            description="logs",
            data_type=ChannelDataTypePy.String,
            enum_types=[],
            bit_field_elements=[],
        ),
    ],
)
```

The corresponding send call must list values in the same order:

```python theme={null}
from sift_stream_bindings import ChannelValuePy, FlowPy, TimeValuePy, ValuePy
from datetime import datetime, timezone

now = datetime.now(timezone.utc)

await ingest_client.send(
    FlowPy(
        flow_name="reading",
        timestamp=TimeValuePy.from_timestamp_millis(int(now.timestamp() * 1000)),
        values=[
            # velocity channel (position 0)
            ChannelValuePy(name="mainmotor.velocity", value=ValuePy.Double(10.0)),
            # log channel (position 1)
            ChannelValuePy(name="log", value=ValuePy.String("example log")),
        ],
    )
)
```

## IngestWithConfigDataStreamRequest

```protobuf theme={null}
message IngestWithConfigDataStreamRequest {
  string ingestion_config_id = 1;
  string flow = 2;
  google.protobuf.Timestamp timestamp = 3;
  repeated IngestWithConfigDataChannelValue channel_values = 4;
  string run_id = 5;
  bool end_stream_on_validation_error = 6;
  string organization_id = 7;
}
```

<MintTable
  columns={['Field', 'Description']}
  columnWidths={['30%', '70%']}
  rows={[
['`ingestion_config_id`', 'The ID returned when you created the ingestion config.'],
['`flow`', 'The name of the flow this request sends data for.'],
['`timestamp`', 'The timestamp for all Channel values in this request.'],
['`channel_values`', 'Ordered list of values. Order must match the Channel order in the flow definition.'],
['`run_id`', 'Optional. Must be included if this data belongs to a Run.'],
['`end_stream_on_validation_error`', 'When `true`, the stream terminates if a server-side error occurs. Use only during development; this flag severely impacts production performance.'],
['`organization_id`', 'Optional unless your user belongs to multiple organizations.'],
]}
/>

Errors that occur when `end_stream_on_validation_error` is `false` appear in Sift's Data Processing dashboard (`https://app.siftstack.com/manage/data-processing`).

<Note>
  **Duplicate timestamps**: If a Channel receives two values at the exact same timestamp, Sift keeps only the most recently written value and discards the earlier one. This deduplication happens at write time, not read time.
</Note>

## Retrieving an ingestion config by client key

<CodeGroup>
  ```bash curl theme={null}
  curl -G -H "Authorization: Bearer $API_TOKEN" -d "filter=client_key=='example_client_key'" $SIFT_REST_URL/api/v1/ingestion-configs
  ```

  ```bash grpcurl theme={null}
  grpcurl -H "authorization: Bearer $API_TOKEN" -d @ $SIFT_GRPC_URL:$PORT_NUM sift.ingestion_configs.v1.IngestionConfigService/ListIngestionConfigs <<EOM
  {
    "filter": "client_key == 'example_client_key'"
  }
  EOM
  ```
</CodeGroup>

## Updating an ingestion config

To add new flows after creation, send a `CreateIngestionConfigFlowRequest` to `IngestionConfigService`. Adding flows and channels is backwards compatible. Modifying or removing existing flows or channels is not.

## Related topics

* [Stream telemetry from a running application](/documentation/ingest/stream/stream-telemetry-from-a-running-application)
* [Organize streamed data into Assets and Runs](/documentation/ingest/stream/organize-streamed-data-into-assets-and-runs)
* [IngestionConfigService API](/api/reference/protocol-buffers/ingestion_configs#ingestionconfigservice)
* [IngestService API](/api/reference/protocol-buffers/ingest#ingestservice)