120 lines
3.5 KiB
Go
120 lines
3.5 KiB
Go
package wshub
|
|
|
|
import (
|
|
"fmt"
|
|
"net/http"
|
|
"sort"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/parquet-go/parquet-go"
|
|
)
|
|
|
|
// ExportSample is one row of the binary export: a single stored sample, in
|
|
// long ("tidy") form, keyed by source and signal with its own timestamp.
|
|
//
|
|
// Keeping each signal's samples as its own rows — rather than resampling onto a
|
|
// shared time grid — is what makes the export hole-free: per-signal streams of
|
|
// different lengths export exactly as stored, nothing is fabricated, and
|
|
// nothing is dropped.
|
|
type ExportSample struct {
|
|
Source string `parquet:"source"`
|
|
Signal string `parquet:"signal"`
|
|
Time float64 `parquet:"time"`
|
|
Value float64 `parquet:"value"`
|
|
}
|
|
|
|
// exportChunkRows bounds each batched write and, via MaxRowsPerRowGroup, the
|
|
// size of each parquet row group: memory stays bounded however large the
|
|
// export is, because a finished row group is flushed to the HTTP stream.
|
|
const exportChunkRows = 65536
|
|
|
|
// exportWriteBuffer is the parquet writer's output buffer: larger than the
|
|
// 32KiB default means fewer writes on the HTTP stream for a multi-GB export.
|
|
const exportWriteBuffer = 1 << 20
|
|
|
|
// HandleExport serves GET /api/export?t0=..&t1=..[&signals=a,b] as a Parquet
|
|
// file containing every stored sample of the named signals in [t0, t1].
|
|
//
|
|
// Unlike /api/zoom there is no decimation: the file holds the full contents of
|
|
// the rings. At rates above the ring budget those contents are min/max buckets
|
|
// (the finest resolution the hub retains); at lower rates they are verbatim.
|
|
func (h *Hub) HandleExport(w http.ResponseWriter, r *http.Request) {
|
|
q := r.URL.Query()
|
|
t0, err0 := strconv.ParseFloat(q.Get("t0"), 64)
|
|
t1, err1 := strconv.ParseFloat(q.Get("t1"), 64)
|
|
if err0 != nil || err1 != nil || t1 <= t0 {
|
|
http.Error(w, "invalid t0/t1", http.StatusBadRequest)
|
|
return
|
|
}
|
|
var keys []string
|
|
if s := strings.TrimSpace(q.Get("signals")); s != "" {
|
|
keys = strings.Split(s, ",")
|
|
for i := range keys {
|
|
keys[i] = strings.TrimSpace(keys[i])
|
|
}
|
|
}
|
|
|
|
// Snapshot the rings we will read. A signal removed mid-export must not
|
|
// silently drop rows from the file.
|
|
h.ringsMu.RLock()
|
|
refs := make(map[string]*sigRing)
|
|
if keys == nil {
|
|
for k, rb := range h.rings {
|
|
refs[k] = rb
|
|
}
|
|
} else {
|
|
for _, k := range keys {
|
|
if rb, ok := h.rings[k]; ok {
|
|
refs[k] = rb
|
|
}
|
|
}
|
|
}
|
|
h.ringsMu.RUnlock()
|
|
if len(refs) == 0 {
|
|
http.Error(w, "no signals", http.StatusNotFound)
|
|
return
|
|
}
|
|
|
|
// Deterministic column order.
|
|
names := make([]string, 0, len(refs))
|
|
for k := range refs {
|
|
names = append(names, k)
|
|
}
|
|
sort.Strings(names)
|
|
|
|
w.Header().Set("Content-Type", "application/vnd.apache.parquet")
|
|
w.Header().Set("Content-Disposition",
|
|
fmt.Sprintf("attachment; filename=\"signals_%d.parquet\"", time.Now().Unix()))
|
|
|
|
writer := parquet.NewGenericWriter[ExportSample](w,
|
|
parquet.MaxRowsPerRowGroup(exportChunkRows),
|
|
parquet.WriteBufferSize(exportWriteBuffer),
|
|
)
|
|
batch := make([]ExportSample, 0, exportChunkRows)
|
|
for _, key := range names {
|
|
st, sv := refs[key].slice(t0, t1)
|
|
colon := strings.IndexByte(key, ':')
|
|
source, signal := key, key
|
|
if colon >= 0 {
|
|
source = key[:colon]
|
|
signal = key[colon+1:]
|
|
}
|
|
for i := range st {
|
|
batch = append(batch, ExportSample{Source: source, Signal: signal, Time: st[i], Value: sv[i]})
|
|
if len(batch) >= exportChunkRows {
|
|
if _, err := writer.Write(batch); err != nil {
|
|
// Client went away or the stream broke; stop writing.
|
|
return
|
|
}
|
|
batch = batch[:0]
|
|
}
|
|
}
|
|
}
|
|
if len(batch) > 0 {
|
|
_, _ = writer.Write(batch)
|
|
}
|
|
_ = writer.Close()
|
|
}
|