Nominal channel writer#
Write data points to Nominal data sources.
OpenAPI file: NominalChannelWriterService.json
Endpoints#
Write batches#
POST /storage/writer/v1
Synchronously writes batches of records to a Nominal data source.
If the request is too large, either due to the number of individual batches (> 10) or the number of points across batches (> 500k), the request may be split up into multiple requests internally when writing to the Nominal data source. Generally, it’s advisable to limit the number of points to 50k.
Body
batchesarray ofRecordsBatchExternal
Fields of
RecordsBatchExternalchannelstring required
tagsmap ofTagNameto string
pointsPointsExternalrequired
Set
typeto"string".stringarray ofStringPoint
Fields of
StringPointtimestampTimestamprequired
Fields of
Timestampsecondsinteger (int64) required
nanosinteger (int64) required
valuestring required
Set
typeto"double".doublearray ofDoublePoint
Fields of
DoublePointtimestampTimestamprequired
valuenumber required
Set
typeto"int".Set
typeto"uint64".uint64array ofUint64Point
Fields of
Uint64PointtimestampTimestamprequired
valueinteger (int64) required
Set
typeto"array".arrayArrayPoints
Set
typeto"double".doublearray ofDoubleArrayPoint
Fields of
DoubleArrayPointtimestampTimestamprequired
valuearray of number
Set
typeto"string".stringarray ofStringArrayPoint
Fields of
StringArrayPointtimestampTimestamprequired
valuearray of string
Set
typeto"struct".structarray ofStructPoint
Fields of
StructPointtimestampTimestamprequired
jsonStringstring required
unitstring
dataSourceRidstring (RID) required
asynchronousInsertboolean deprecatedIs always true - setting this to false will do nothing. Deprecated: Should not be used.
sessionNamestringOptional client-provided session name identifier for tracing the origin of a write request.
Response
No content (HTTP 204).
Request
curl -X POST "https://api.gov.nominal.io/api/storage/writer/v1" \
-H "Authorization: Bearer <api key>" \
-H "Content-Type: application/json" \
--data '{
"dataSourceRid": "<nominal data source or dataset rid>"
}'
import requests
response = requests.post(
"https://api.gov.nominal.io/api/storage/writer/v1",
headers={"Authorization": "Bearer <api key>"},
json={"dataSourceRid": "<nominal data source or dataset rid>"},
)
response.raise_for_status()
import httpx
response = httpx.post(
"https://api.gov.nominal.io/api/storage/writer/v1",
headers={"Authorization": "Bearer <api key>"},
json={"dataSourceRid": "<nominal data source or dataset rid>"},
)
response.raise_for_status()
import asyncio
import aiohttp
async def main() -> None:
async with aiohttp.ClientSession() as session:
async with session.post(
"https://api.gov.nominal.io/api/storage/writer/v1",
headers={"Authorization": "Bearer <api key>"},
json={"dataSourceRid": "<nominal data source or dataset rid>"},
) as response:
response.raise_for_status()
asyncio.run(main())
Response: 204
No content.
Write column batches#
POST /storage/writer/v1/columnar
Synchronously writes batches of columns of data to a Nominal data source.
This is a column-major variant of writeBatches (which is row-major) to optimize serialization and compression time for client applications streaming large numbers of points from a single column at a time. This has the tradeoff of slightly larger sizes post-gzipping of requests, so should be used in the particular case where the main bottleneck is in encoding columnar data into the row-based format found in writeBatches.
Body
batchesarray ofColumnBatchBatches of columnar data to stream to Nominal. Each channel’s data are provided as a column batch.
Fields of
ColumnBatchchannelstring requiredChannel within nominal to stream data to.
tagsmap ofTagNameto stringMapping of key-value pairs to provide as tags to all points within the batch
timestampsarray ofTimestampList of timestamp values that correspond to the provided list of column values. The number of timestamps provided MUST match the number of columnar values provided, otherwise a 400 error will be returned.
Fields of
Timestampsecondsinteger (int64) required
nanosinteger (int64) required
valuesColumnValuesrequiredList of time series values that should be ingested to a single channel. The number of columnar values provided MUST match the number of timestamps provided, otherwise a 400 error will be returned.
Set
typeto"strings".stringsarray of string
Set
typeto"doubles".doublesarray of number
Set
typeto"ints".intsarray of integer
Set
typeto"uint64s".uint64sarray of integer (int64)
Set
typeto"arrays".arraysArraysValues
Set
typeto"strings1d".strings1darray of array of string
Set
typeto"doubles1d".doubles1darray of array of number
Set
typeto"structs".structsarray of string
dataSourceRidstring (RID) requiredRID of the datasource (e.g., for a Connection) or dataset to stream data into.
sessionNamestringOptional client-provided session name identifier for tracing the origin of a write request.
Response
No content (HTTP 204).
Request
curl -X POST "https://api.gov.nominal.io/api/storage/writer/v1/columnar" \
-H "Authorization: Bearer <api key>" \
-H "Content-Type: application/json" \
--data '{
"dataSourceRid": "<nominal data source or dataset rid>"
}'
import requests
response = requests.post(
"https://api.gov.nominal.io/api/storage/writer/v1/columnar",
headers={"Authorization": "Bearer <api key>"},
json={"dataSourceRid": "<nominal data source or dataset rid>"},
)
response.raise_for_status()
import httpx
response = httpx.post(
"https://api.gov.nominal.io/api/storage/writer/v1/columnar",
headers={"Authorization": "Bearer <api key>"},
json={"dataSourceRid": "<nominal data source or dataset rid>"},
)
response.raise_for_status()
import asyncio
import aiohttp
async def main() -> None:
async with aiohttp.ClientSession() as session:
async with session.post(
"https://api.gov.nominal.io/api/storage/writer/v1/columnar",
headers={"Authorization": "Bearer <api key>"},
json={"dataSourceRid": "<nominal data source or dataset rid>"},
) as response:
response.raise_for_status()
asyncio.run(main())
Response: 204
No content.
Write Telegraf batches#
POST /storage/writer/v1/telegraf/{dataSourceRid}
Synchronously writes batches of records to a Nominal data source.
Has the same functionality as writeBatches, but is compatible with the Telegraf output format. Assumes that the Telegraf batch format is used. Timestamp is assumed to be in nanoseconds. The URL in the Telegraf output plugin configuration should be the fully qualified URL, including the dataSourceRid query parameter.
Path parameters
dataSourceRidstring (RID) required
Body
metricsarray ofTelegrafMetric
Fields of
TelegrafMetricfieldsmap ofFieldto any JSON valueThe values are expected to be either numeric or string values
namestring requiredThe measurement name. Measurement name and field are concatenated when creating the Nominal channel name.
tagsmap ofTagNameto string
timestampany JSON value required
Response
No content (HTTP 204).
Request
curl -X POST "https://api.gov.nominal.io/api/storage/writer/v1/telegraf/<data source rid>" \
-H "Authorization: Bearer <api key>" \
-H "Content-Type: application/json" \
--data '{
"metrics": [
{
"name": "<measurement name>",
"timestamp": {}
}
]
}'
import requests
response = requests.post(
"https://api.gov.nominal.io/api/storage/writer/v1/telegraf/<data source rid>",
headers={"Authorization": "Bearer <api key>"},
json={"metrics": [{"name": "<measurement name>", "timestamp": {}}]},
)
response.raise_for_status()
import httpx
response = httpx.post(
"https://api.gov.nominal.io/api/storage/writer/v1/telegraf/<data source rid>",
headers={"Authorization": "Bearer <api key>"},
json={"metrics": [{"name": "<measurement name>", "timestamp": {}}]},
)
response.raise_for_status()
import asyncio
import aiohttp
async def main() -> None:
async with aiohttp.ClientSession() as session:
async with session.post(
"https://api.gov.nominal.io/api/storage/writer/v1/telegraf/<data source rid>",
headers={"Authorization": "Bearer <api key>"},
json={"metrics": [{"name": "<measurement name>", "timestamp": {}}]},
) as response:
response.raise_for_status()
asyncio.run(main())
Response: 204
No content.
Write Prometheus batches#
POST /storage/writer/v1/prometheus/{dataSourceRid}
Synchronously writes batches of records to a Nominal data source.
Has the same functionality as writeBatches, but is encoded using the Prometheus remote write format. We follow the specification defined here: https://prometheus.io/docs/specs/remote_write_spec/ There are a few notable caveats: 1. Must be content encoded as application/x-protobuf 2. Must be compressed using snappy compression
Path parameters
dataSourceRidstring (RID) required
Body: application/octet-stream
The body is the raw bytes.
Response
No content (HTTP 204).
Request
curl -X POST "https://api.gov.nominal.io/api/storage/writer/v1/prometheus/<data source rid>" \
-H "Authorization: Bearer <api key>" \
-H "Content-Type: application/octet-stream" \
--data-binary "@<file>"
from pathlib import Path
import requests
response = requests.post(
"https://api.gov.nominal.io/api/storage/writer/v1/prometheus/<data source rid>",
headers={
"Authorization": "Bearer <api key>",
"Content-Type": "application/octet-stream",
},
data=Path("<file>").read_bytes(),
)
response.raise_for_status()
from pathlib import Path
import httpx
response = httpx.post(
"https://api.gov.nominal.io/api/storage/writer/v1/prometheus/<data source rid>",
headers={
"Authorization": "Bearer <api key>",
"Content-Type": "application/octet-stream",
},
content=Path("<file>").read_bytes(),
)
response.raise_for_status()
import asyncio
from pathlib import Path
import aiohttp
async def main() -> None:
async with aiohttp.ClientSession() as session:
async with session.post(
"https://api.gov.nominal.io/api/storage/writer/v1/prometheus/<data source rid>",
headers={
"Authorization": "Bearer <api key>",
"Content-Type": "application/octet-stream",
},
data=Path("<file>").read_bytes(),
) as response:
response.raise_for_status()
asyncio.run(main())
Response: 204
No content.
Prometheus remote write health check#
GET /storage/writer/v1/prometheus/{dataSourceRid}
Performs a health check for prometheus remote write Vector sink. All this endpoint does is verify if the caller is authenticated and the server is online. Once Vector allows the Prometheus remote write endpoint to configure the healthcheck url, we can remove this endpoint.
See: https://github.com/vectordotdev/vector/issues/8279
Path parameters
dataSourceRidstring (RID) required
Response
The body is a boolean.
Request
curl "https://api.gov.nominal.io/api/storage/writer/v1/prometheus/<data source rid>" \
-H "Authorization: Bearer <api key>"
import requests
response = requests.get(
"https://api.gov.nominal.io/api/storage/writer/v1/prometheus/<data source rid>",
headers={"Authorization": "Bearer <api key>"},
)
response.raise_for_status()
print(response.json())
import httpx
response = httpx.get(
"https://api.gov.nominal.io/api/storage/writer/v1/prometheus/<data source rid>",
headers={"Authorization": "Bearer <api key>"},
)
response.raise_for_status()
print(response.json())
import asyncio
import aiohttp
async def main() -> None:
async with aiohttp.ClientSession() as session:
async with session.get(
"https://api.gov.nominal.io/api/storage/writer/v1/prometheus/<data source rid>",
headers={"Authorization": "Bearer <api key>"},
) as response:
response.raise_for_status()
print(await response.json())
asyncio.run(main())
Response: 200
true
Write Nominal batches#
POST /storage/writer/v1/nominal/{dataSourceRid}
Synchronously writes a Nominal Write Request to a Nominal data source using the NominalWrite Protobuf schema. The request must be Protobuf-encoded and accompanied by the appropriate content encoding headers if compressed.
The request should follow this Protobuf schema: ```proto message WriteRequestNominal { repeated Series series = 1; }
message Series { Channel channel = 1; map<string, string> tags = 2; // Key-value pairs for series tags Points points = 3; // Contains double, string, integer, array, or struct points }
message Channel { string name = 1; }
message Points { oneof points_type { DoublePoints double_points = 1; StringPoints string_points = 2; IntegerPoints integer_points = 3; ArrayPoints array_points = 4; StructPoints struct_points = 5; Uint64Points uint64_points = 6; } }
message DoublePoints { repeated DoublePoint points = 1; }
message StringPoints { repeated StringPoint points = 1; }
message IntegerPoints { repeated IntegerPoint points = 1; }
message Uint64Points { repeated Uint64Point points = 1; }
message ArrayPoints { oneof array_type { DoubleArrayPoints double_array_points = 1; StringArrayPoints string_array_points = 2; } }
message DoubleArrayPoints { repeated DoubleArrayPoint points = 1; }
message StringArrayPoints { repeated StringArrayPoint points = 1; }
message StructPoints { repeated StructPoint points = 1; }
message DoublePoint { google.protobuf.Timestamp timestamp = 1; double value = 2; }
message StringPoint { google.protobuf.Timestamp timestamp = 1; string value = 2; }
message IntegerPoint { google.protobuf.Timestamp timestamp = 1; int64 value = 2; }
message Uint64Point { google.protobuf.Timestamp timestamp = 1; uint64 value = 2; }
message DoubleArrayPoint { google.protobuf.Timestamp timestamp = 1; repeated double value = 2; }
message StringArrayPoint { google.protobuf.Timestamp timestamp = 1; repeated string value = 2; }
message StructPoint { google.protobuf.Timestamp timestamp = 1; string jsonString = 2; } ```
Each request can contain multiple series, where each series consists of: - A channel name - A map of tags (key-value pairs) - A collection of points, which can be double, string, integer, array (double/string), or struct (json) values - Each point includes a timestamp (using google.protobuf.Timestamp) and its value - Array points contain repeated values (array<double> or array<string>) - Struct points are represented using strings that contain the JSON formatted blobs, with all keys being strings
The endpoint requires the Content-Type header to be set to “application/x-protobuf”. If the payload is compressed, the appropriate Content-Encoding header must be included.
Path parameters
dataSourceRidstring (RID) required
Body: application/octet-stream
The body is the raw bytes.
Response
No content (HTTP 204).
Request
curl -X POST "https://api.gov.nominal.io/api/storage/writer/v1/nominal/<data source rid>" \
-H "Authorization: Bearer <api key>" \
-H "Content-Type: application/octet-stream" \
--data-binary "@<file>"
from pathlib import Path
import requests
response = requests.post(
"https://api.gov.nominal.io/api/storage/writer/v1/nominal/<data source rid>",
headers={
"Authorization": "Bearer <api key>",
"Content-Type": "application/octet-stream",
},
data=Path("<file>").read_bytes(),
)
response.raise_for_status()
from pathlib import Path
import httpx
response = httpx.post(
"https://api.gov.nominal.io/api/storage/writer/v1/nominal/<data source rid>",
headers={
"Authorization": "Bearer <api key>",
"Content-Type": "application/octet-stream",
},
content=Path("<file>").read_bytes(),
)
response.raise_for_status()
import asyncio
from pathlib import Path
import aiohttp
async def main() -> None:
async with aiohttp.ClientSession() as session:
async with session.post(
"https://api.gov.nominal.io/api/storage/writer/v1/nominal/<data source rid>",
headers={
"Authorization": "Bearer <api key>",
"Content-Type": "application/octet-stream",
},
data=Path("<file>").read_bytes(),
) as response:
response.raise_for_status()
asyncio.run(main())
Response: 204
No content.
Write Nominal columnar batches#
POST /storage/writer/v1/nominal-columnar
Synchronously writes to a Nominal data source using the columnar WriteBatchesRequest Protobuf schema.
This is the most efficient write format. Unlike the row-oriented writeNominalBatches endpoint, this schema stores timestamps and values as separate packed arrays per channel, which enables protobuf packed encoding for value arrays and significantly better compression.
The data_source_rid field in the protobuf body identifies the target data source or dataset. The request must be Protobuf-encoded and accompanied by the appropriate content encoding headers if compressed.
The request should follow this Protobuf schema: ```proto message WriteBatchesRequest { repeated RecordsBatch batches = 1; string data_source_rid = 2; }
message RecordsBatch { string channel = 1; map<string, string> tags = 2; Points points = 3; }
message Points { repeated Timestamp timestamps = 1; oneof points { DoublePoints double_points = 2; StringPoints string_points = 3; IntPoints int_points = 5; ArrayPoints array_points = 6; StructPoints struct_points = 7; Uint64Points uint64_points = 8; } }
// Nanosecond precision timestamp. message Timestamp { optional int64 seconds = 1; optional int64 nanos = 2; }
message DoublePoints { repeated double points = 1; } message StringPoints { repeated string points = 1; } message IntPoints { repeated int64 points = 1; } message Uint64Points { repeated uint64 points = 1; } message StructPoints { repeated string points = 1; }
message ArrayPoints { oneof array_type { DoubleArrayPoints double_array_points = 1; StringArrayPoints string_array_points = 2; } } message DoubleArrayPoints { repeated DoubleArrayPoint points = 1; } message DoubleArrayPoint { repeated double value = 2; } message StringArrayPoints { repeated StringArrayPoint points = 1; } message StringArrayPoint { repeated string value = 2; } ```
Body: application/octet-stream
The body is the raw bytes.
Response
No content (HTTP 204).
Request
curl -X POST "https://api.gov.nominal.io/api/storage/writer/v1/nominal-columnar" \
-H "Authorization: Bearer <api key>" \
-H "Content-Type: application/octet-stream" \
--data-binary "@<file>"
from pathlib import Path
import requests
response = requests.post(
"https://api.gov.nominal.io/api/storage/writer/v1/nominal-columnar",
headers={
"Authorization": "Bearer <api key>",
"Content-Type": "application/octet-stream",
},
data=Path("<file>").read_bytes(),
)
response.raise_for_status()
from pathlib import Path
import httpx
response = httpx.post(
"https://api.gov.nominal.io/api/storage/writer/v1/nominal-columnar",
headers={
"Authorization": "Bearer <api key>",
"Content-Type": "application/octet-stream",
},
content=Path("<file>").read_bytes(),
)
response.raise_for_status()
import asyncio
from pathlib import Path
import aiohttp
async def main() -> None:
async with aiohttp.ClientSession() as session:
async with session.post(
"https://api.gov.nominal.io/api/storage/writer/v1/nominal-columnar",
headers={
"Authorization": "Bearer <api key>",
"Content-Type": "application/octet-stream",
},
data=Path("<file>").read_bytes(),
) as response:
response.raise_for_status()
asyncio.run(main())
Response: 204
No content.
Write logs#
POST /storage/writer/v1/logs/{dataSourceRid}
Synchronously writes logs to a Nominal data source.
Path parameters
dataSourceRidstring (RID) required
Body
logsarray ofLogPoint
channelstringIf provided, the channel to which to write logs. If not provided, defaults to “logs”
Response
No content (HTTP 204).
Request
curl -X POST "https://api.gov.nominal.io/api/storage/writer/v1/logs/<data source rid>" \
-H "Authorization: Bearer <api key>" \
-H "Content-Type: application/json" \
--data '{
"logs": [
{
"timestamp": {
"seconds": 1,
"nanos": 1
},
"value": {
"message": "<message>"
}
}
],
"channel": "<channel>"
}'
import requests
response = requests.post(
"https://api.gov.nominal.io/api/storage/writer/v1/logs/<data source rid>",
headers={"Authorization": "Bearer <api key>"},
json={
"logs": [
{"timestamp": {"seconds": 1, "nanos": 1}, "value": {"message": "<message>"}},
],
"channel": "<channel>",
},
)
response.raise_for_status()
import httpx
response = httpx.post(
"https://api.gov.nominal.io/api/storage/writer/v1/logs/<data source rid>",
headers={"Authorization": "Bearer <api key>"},
json={
"logs": [
{"timestamp": {"seconds": 1, "nanos": 1}, "value": {"message": "<message>"}},
],
"channel": "<channel>",
},
)
response.raise_for_status()
import asyncio
import aiohttp
async def main() -> None:
async with aiohttp.ClientSession() as session:
async with session.post(
"https://api.gov.nominal.io/api/storage/writer/v1/logs/<data source rid>",
headers={"Authorization": "Bearer <api key>"},
json={
"logs": [
{
"timestamp": {"seconds": 1, "nanos": 1},
"value": {"message": "<message>"},
},
],
"channel": "<channel>",
},
) as response:
response.raise_for_status()
asyncio.run(main())
Response: 204
No content.
Errors#
Errors shows how these errors are sent.
Each error’s errorName is NominalChannelWriter: followed by its name below.
Status |
Name |
Notes |
|---|---|---|
400 |
|
Array size exceeds maximum allowed length. |
400 |
|
Found two different data types for the same series in the request. |
400 |
|
Found two different units for the same channel in the request. |
400 |
|
Multiple ingested columns resolve to the same channel name after applying channelPrefix and channelNameOverrides. Resolved channel names must be unique within a single ingest. |
400 |
|
The channel name is invalid. |
400 |
|
The nominal data source or dataset provided is invalid. |
400 |
|
Ingested Struct is not a valid JSON struct |
400 |
|
|
400 |
|
|
400 |
|
Columnar streaming request contains different numbers of timestamps and values |
400 |
|
The request uses a tag name that is reserved for internal use. Remove the tag from the request. |
400 |
|
Unsupported dataset for streaming. |
400 |
|
The request sets a streaming session rid, which is assigned by Nominal and cannot be provided by the client. |
400 |
|
Struct value exceeds maximum allowed UTF-8 byte size. |
400 |
|
The request specifies more channel name overrides than the maximum allowed. |
400 |
|
The dataset has exceeded the maximum number of distinct active session sources allowed per hour. This is a best-effort limit enforced per channel-writer instance. |
413 |
|
The size of the points being written exceeds the maximum allowed size. |
Types#
ArrayPoints#
Array points for the internal API for directly writing points which supports all points types.
Set type to the name of one of the other fields, and set only that field.
doublearray ofDoubleArrayPoint
stringarray ofStringArrayPoint
ArraysValues#
Each element within the outer list corresponds to an Array point. Subsequent inner lists contain the array values recorded for that point.
Set type to the name of one of the other fields, and set only that field.
strings1darray of array of string
doubles1darray of array of number
ColumnBatch#
Batch of data to stream for a single channel with their associated timestamps.
channelstring requiredChannel within nominal to stream data to.
tagsmap ofTagNameto stringMapping of key-value pairs to provide as tags to all points within the batch
timestampsarray ofTimestampList of timestamp values that correspond to the provided list of column values. The number of timestamps provided MUST match the number of columnar values provided, otherwise a 400 error will be returned.
valuesColumnValuesrequiredList of time series values that should be ingested to a single channel. The number of columnar values provided MUST match the number of timestamps provided, otherwise a 400 error will be returned.
ColumnValues#
List of values for a set of points to ingest from a single channel.
Set type to the name of one of the other fields, and set only that field.
stringsarray of string
doublesarray of number
intsarray of integer
uint64sarray of integer (int64)
arraysArraysValues
structsarray of string
DoubleArrayPoint#
timestampTimestamprequired
valuearray of number
DoublePoint#
timestampTimestamprequired
valuenumber required
IntPoint#
timestampTimestamprequired
valueinteger required
LogPoint#
LogValue#
messagestring required
argsmap of string to string
PointsExternal#
Set type to the name of one of the other fields, and set only that field.
stringarray ofStringPoint
doublearray ofDoublePoint
intarray ofIntPoint
uint64array ofUint64Point
arrayArrayPoints
structarray ofStructPoint
RecordsBatchExternal#
channelstring required
tagsmap ofTagNameto string
pointsPointsExternalrequired
unitstring
StringArrayPoint#
timestampTimestamprequired
valuearray of string
StringPoint#
timestampTimestamprequired
valuestring required
StructPoint#
timestampTimestamprequired
jsonStringstring required
TelegrafMetric#
fieldsmap ofFieldto any JSON valueThe values are expected to be either numeric or string values
namestring requiredThe measurement name. Measurement name and field are concatenated when creating the Nominal channel name.
tagsmap ofTagNameto string
timestampany JSON value required
Timestamp#
Deprecated for new APIs. Use google.protobuf.Timestamp for absolute instants in proto APIs.
This legacy nanosecond precision timestamp type is represented by an epoch time in seconds and a nanosecond offset. The nanosecond offset is from the start of the epoch second, so must be less than 1 billion.
secondsinteger (int64) required
nanosinteger (int64) required
Uint64Point#
timestampTimestamprequired
valueinteger (int64) required
Page design based on Scalar, MIT License.