InfluxElixir.Flight.Client (InfluxElixir v0.1.53)

Copy Markdown View Source

Arrow Flight gRPC client for high-throughput query transport.

Connects to an InfluxDB v3 Flight endpoint, encodes SQL queries as JSON-bearing Ticket messages, streams back FlightData chunks via the DoGet RPC, and delegates binary decoding to InfluxElixir.Flight.Reader.

Transport

InfluxDB v3 exposes its Flight service on the same host as the HTTP API, typically port 443 (TLS). The host in the connection map must be the plain hostname (no scheme); TLS is configured via the :tls option.

Authentication

Bearer-token auth is passed as gRPC metadata on every call:

Authorization: Bearer <token>

An empty token ("", a server started without auth) sends no authorization metadata, as the HTTP transport sends no header.

Example

conn = %{host: "us-east-1.influxdb.io", token: "my-token",
         database: "mydb", port: 443}

{:ok, rows} = InfluxElixir.Flight.Client.query(conn, "SELECT * FROM cpu LIMIT 10")

Options

  • :timeout — wall-clock bound in milliseconds on the call and the streaming of its whole result (default: 30_000; through InfluxElixir.query_sql/3 the call's, then the connection's :timeout). A server that accepts the connection and stays silent, or that stops sending, is {:error, :timeout} when it is spent.
  • :connect_timeout — bound on the gRPC channel establishment phase (default: same value as :timeout). Without this bound, a stuck TLS handshake against an unresponsive host can hang far longer than the in-stream :timeout would suggest.
  • :tls — true to use TLS (default: true when port is 443)

Summary

Types

A connection map with the keys used by this client.

Functions

Builds a Ticket struct for the given database and SQL query.

Builds the JSON-encoded ticket payload for an InfluxDB v3 SQL query.

Executes a SQL query against InfluxDB v3 via Arrow Flight DoGet.

Types

connection()

@type connection() :: %{
  :host => binary(),
  :token => binary(),
  :database => binary(),
  optional(:port) => non_neg_integer()
}

A connection map with the keys used by this client.

  • :host — hostname of the InfluxDB Flight endpoint (required)
  • :token — bearer token for authentication (required)
  • :database — InfluxDB database / bucket name (required)
  • :port — gRPC port (default: 443)

Functions

build_ticket(database, sql)

@spec build_ticket(binary(), binary()) :: InfluxElixir.Flight.Proto.Ticket.t()

Builds a Ticket struct for the given database and SQL query.

Useful for constructing tickets before calling do_get directly or for inspecting the wire format in tests.

Parameters

  • database — target InfluxDB database name
  • sql — SQL query string

Example

iex> t = InfluxElixir.Flight.Client.build_ticket("mydb", "SELECT 1")
iex> t.ticket |> Jason.decode!() |> Map.fetch!("database")
"mydb"

build_ticket_payload(database, sql)

@spec build_ticket_payload(binary(), binary()) :: binary()

Builds the JSON-encoded ticket payload for an InfluxDB v3 SQL query.

Exposed for testing and introspection purposes. Text JSON cannot carry (a string that is not UTF-8) raises Jason.EncodeError; query/3 answers it as an error.

Parameters

  • database — target InfluxDB database name
  • sql — SQL query string

Example

iex> payload = InfluxElixir.Flight.Client.build_ticket_payload("mydb", "SELECT 1")
iex> Jason.decode!(payload)
%{"database" => "mydb", "sql_query" => "SELECT 1", "query_type" => "sql"}

query(connection, sql, opts \\ [])

@spec query(connection(), binary(), keyword()) :: {:ok, [map()]} | {:error, term()}

Executes a SQL query against InfluxDB v3 via Arrow Flight DoGet.

Builds a Ticket with a JSON payload understood by InfluxDB v3, opens a gRPC channel, streams FlightData messages, and decodes them into a list of row maps.

Parameters

  • connection — map with :host, :token, :database, and optional :port
  • sql — SQL query string
  • opts — keyword options

Options

  • :timeout — milliseconds the call and its whole stream may take (default: 30_000); the call runs in a monitored process that is killed at the bound
  • :connect_timeout — milliseconds to establish the channel (default: :timeout)
  • :tls — force TLS on/off; inferred from port when omitted

Returns

  • {:ok, [map()]} — list of row maps (column name → value)

  • {:error, {:ipv6_unsupported, host}} — the host is an IPv6 address: the gRPC client cannot connect to one (verified against InfluxDB 3 Core, which answers on [::1] over HTTP); use the HTTP transport for it

  • {:error, {:invalid_host, host}} — the host is not a host name (a port belongs in :port, not "host:8181")

  • {:error, {:invalid_token, token}} — the token is not a string

  • {:error, {:unencodable_body, message}} — the database or the statement is not what JSON can carry (text that is not UTF-8, a tuple)

  • {:error, :timeout} — the call and its stream were not finished in :timeout milliseconds, or the server's own deadline expired (gRPC status 4)

  • {:error, :connect_timeout} — the channel was not open in :connect_timeout

  • {:error, %{status: http_status, body: message}} — the server refused the query, in the shape the HTTP transport gives. The gRPC status becomes the nearest HTTP one: 3 (INVALID_ARGUMENT) 400, 16 (UNAUTHENTICATED) 401, 7 (PERMISSION_DENIED) 403, 5 (NOT_FOUND) 404, 6 409, 8 429, 12 501, 14 503, any other 500. The statuses 3 and 13 were checked against InfluxDB 3 Core 3.10.1 with the same failing statements over both transports; the others are the standard mapping, not observed. The statuses do not always agree with HTTP's (a schema error, select nosuch from m, is 500 over HTTP and 400 here; a database that does not exist is 404 over HTTP and 500 here, as Flight answers it INTERNAL), and the Flight message keeps the engine's prefix (Error while logically planning query: ...).

  • {:error, term()} — any other transport or decode error

A connection without :host, :token or :database raises KeyError: that is a programming error, not an answer.

Example

conn = %{host: "cloud2.influxdata.com", token: "my-tok", database: "sensors"}
{:ok, rows} = InfluxElixir.Flight.Client.query(conn, "SELECT * FROM cpu LIMIT 5")