Skip to main content

Pipeline engine

A stored pipeline is: name, input datasource, reader, and ordered steps.

Source → reader → steps → sink​

  1. Discover items from the datasource source
  2. Reader turns item bytes into records (nmea; json — JSON Lines or a JSON array; or binary — length-prefixed or fixed-size frames emitted as a hex/base64 field without decoding)
  3. Steps: first must be a classifier; classifiers cannot be adjacent (put a transformer between them)
  4. Classifier validates against a registered schema id
  5. Transformer maps/decodes/expands (passthrough, pick, ais, custom)
  6. Idle 5 seconds, then write the sink. TCP/UDP listen mode instead streams: each datagram/connection is processed as it arrives and records are written to the sink every 500 records or 2 seconds until the listen window ends or the run is cancelled.

Implemented sources: s3, kafka, http, tcp, udp, local (needs a path the pod can read). Implemented sinks: s3 and kafka only. The console may show SFTP and a local sink; those types fail at run time.

HTTP source is a poller (GET/POST, body fingerprint, rate limit, 10 MiB cap). It is not a webhook listener.

TCP and UDP sources are either dial or listen (mode, default client).

Dial connects to host:port and reads until the socket is idle (default 2s). The engine's 5s item drain then finishes the job. TCP framing is newline (one item per line) or raw (one burst). UDP emits one item per datagram from that peer only.

Listen binds an ephemeral port on the worker for listen_timeout seconds (0 until cancel). The engine does not use the 5s idle drain: it processes items as they arrive and flushes to the sink every 500 records or 2 seconds, updating records_out on each flush. On cancel, records already received are flushed before the run stops. With listener_services enabled, the worker creates ClusterIP Service ft-exec-<execution id> (no pod selector) plus an Endpoints object for this pod's IP and the ephemeral port. The Service port is the datasource port. Peers use ft-exec-<id>.<namespace>.svc.cluster.local:<port>. Concurrent runs each get their own Service. The Service is deleted when the run ends; startup deletes leftovers labeled for this pod. The Service and Endpoints are owned by the worker pod, so Kubernetes garbage-collects them if the pod is deleted before the worker can clean up.

Custom transformers​

Pipelines store type: custom plus an id. At run time the step is expanded to Python with the stored script. The child:

  • Must define transform(record) -> dict
  • Default timeout 5s per record, max 60s; batch wall-clock cap 5 minutes
  • 256 MiB memory cap

Interpreter: FALCONTALON_PYTHON_BIN (/usr/local/bin/python3 in the image).

Safe URLs​

Loopback and RFC1918 HTTP/S3 endpoints are rejected unless FALCONTALON_ALLOW_PRIVATE_URLS=true. Keep that false.

TCP/UDP hosts use a different policy: RFC1918 and loopback are allowed (on-prem AIS receivers). Metadata, link-local, and multicast addresses are still rejected. Enabling FALCONTALON_ALLOW_PRIVATE_URLS is not required for private socket feeds.