nordisk/edda

nornir-flight — guide

How to move RecordBatches over Arrow Flight with this crate.

Serve a transport

use std::sync::Arc;
use nornir_flight::{FlightRoutes, FlightServer, MemFlightStore};

# async fn run() -> nornir_flight::Result<()> {
// One store backs both the sink (do_put) and the source (do_get).
let store = Arc::new(MemFlightStore::new());
let routes = FlightRoutes::shared(store);

// Open (no auth) — dev / single box:
FlightServer::serve(routes.clone(), "127.0.0.1:50051".parse().unwrap()).await?;

// With the shared bearer auth (empty token ⇒ open):
let token = Arc::new("a-long-enough-secret-token".to_string());
FlightServer::serve_with_auth(routes, "127.0.0.1:50051".parse().unwrap(), token).await?;
# Ok(()) }

serve_with_auth composes nornir_service::auth_interceptor + nornir_service::serve — the same auth every nordisk server uses. It is never re-rolled here.

Serve over TLS (feature tls)

Terminate TLS on the Flight bootstrap — the encrypted sibling of the plaintext paths above. Behind the tls feature (--features tls), which turns on nornir-service/tls so the transport reuses the same serve_with_tls loop holger-server uses (never a hand-rolled tls_config tail):

use std::sync::Arc;
use nornir_flight::{FlightConn, FlightRoutes, FlightServer, MemFlightStore};
use tonic::transport::{Certificate, ClientTlsConfig, Identity, ServerTlsConfig};

# async fn run(server_cert_pem: &str, server_key_pem: &str, ca_pem: &str) -> nornir_flight::Result<()> {
let routes = FlightRoutes::shared(Arc::new(MemFlightStore::new()));

// Server identity from PEM cert+key:
let server_tls = ServerTlsConfig::new().identity(Identity::from_pem(server_cert_pem, server_key_pem));
FlightServer::serve_with_tls(routes, "127.0.0.1:50051".parse().unwrap(), server_tls).await?;
// …or layer bearer auth UNDER TLS: FlightServer::serve_with_tls_and_auth(routes, addr, token, server_tls)

// Client trusts the CA (self-signed: the server cert itself), verifying the host:
let client_tls = ClientTlsConfig::new().ca_certificate(Certificate::from_pem(ca_pem)).domain_name("localhost");
let mut c = FlightConn::connect_tls("https://localhost:50051", client_tls).await?;
# Ok(()) }

Talk to it (client)

use nornir_flight::FlightConn;

# async fn run(batches: Vec<nornir_flight::RecordBatch>) -> nornir_flight::Result<()> {
let mut c = FlightConn::connect("http://127.0.0.1:50051").await?
    .with_bearer("a-long-enough-secret-token")?;   // omit when the server is open

c.put("build/42", batches.clone()).await?;          // do_put under a routing key
let got = c.get("build/42").await?;                 // do_get → byte-identical
let echoed = c.exchange("build/42", batches).await?; // do_exchange → round-trip
# Ok(()) }

Plug in your own sink/source

Implement one or both traits — the schema is yours:

use async_trait::async_trait;
use nornir_flight::{FlightSink, RecordBatch, Result};

struct MySink;
#[async_trait]
impl FlightSink for MySink {
    async fn put(&self, key: String, batches: Vec<RecordBatch>) -> Result<()> {
        // seal to znippy, land rows in skade, fan out to viz, …
        Ok(())
    }
}

Then FlightRoutes::new(Arc::new(MySink), source).

Features

• testmatrix — emit functional_status rows (serve/put/get/exchange) into the nornir test-matrix. Off by default (release strips it).

• tls — enable the TLS serve/connect path (FlightServer::serve_with_tls[_and_auth]

  • FlightConn::connect_tls), reusing nornir-service's serve_with_tls loop. Off by default (plaintext consumers never pull in tonic's TLS stack). Test it: cargo test -p nornir-flight --features tls.

• bench — register the nornir-flight.put_get_throughput bencher. Run it: cargo run --release -p nornir-flight --features bench --example nornir-bench.

HEAVY bench (Loki, 12 cores)

The Loki bencher re-runs this on every push; the number above is host-stamped and indicative.