cactoandClaude Sonnet 5
938700cf86
Subscribes to devices/+/telemetry, validates payloads, buffers writes into ClickHouse (flush on size or interval), and emits a per-reading event on RabbitMQ for rule-engine-service to consume later. Adds the initial Postgres and ClickHouse schema migrations, a Dockerfile, and wires the service into docker-compose. Verified end-to-end via docker compose + mosquitto_pub: row lands in ClickHouse and the event reaches the telemetry.new_reading queue. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
106 lines
2.5 KiB
Go
106 lines
2.5 KiB
Go
// Package batch accumulates telemetry readings and flushes them in bulk,
|
|
// either once a size threshold is hit or on a fixed interval — so ClickHouse
|
|
// gets bulk inserts instead of one round-trip per reading.
|
|
package batch
|
|
|
|
import (
|
|
"context"
|
|
"log/slog"
|
|
"sync"
|
|
"time"
|
|
|
|
"git.cactoz.su/cacto/home_automatization/services/ingest-service/internal/telemetry"
|
|
)
|
|
|
|
// FlushFunc persists one batch. It is called with a context bounded by
|
|
// FlushTimeout, both for interval-triggered and shutdown-triggered flushes.
|
|
type FlushFunc func(ctx context.Context, readings []telemetry.Reading) error
|
|
|
|
// Batcher buffers readings in memory and flushes them via FlushFunc when
|
|
// MaxSize is reached or Interval elapses, whichever comes first.
|
|
type Batcher struct {
|
|
maxSize int
|
|
interval time.Duration
|
|
flushTimeout time.Duration
|
|
flush FlushFunc
|
|
logger *slog.Logger
|
|
|
|
mu sync.Mutex
|
|
buf []telemetry.Reading
|
|
|
|
stop chan struct{}
|
|
done chan struct{}
|
|
}
|
|
|
|
func New(maxSize int, interval, flushTimeout time.Duration, flush FlushFunc, logger *slog.Logger) *Batcher {
|
|
return &Batcher{
|
|
maxSize: maxSize,
|
|
interval: interval,
|
|
flushTimeout: flushTimeout,
|
|
flush: flush,
|
|
logger: logger,
|
|
}
|
|
}
|
|
|
|
// Start begins the interval-flush loop. Call once before the first Add.
|
|
func (b *Batcher) Start() {
|
|
b.stop = make(chan struct{})
|
|
b.done = make(chan struct{})
|
|
go b.loop()
|
|
}
|
|
|
|
func (b *Batcher) loop() {
|
|
defer close(b.done)
|
|
ticker := time.NewTicker(b.interval)
|
|
defer ticker.Stop()
|
|
for {
|
|
select {
|
|
case <-ticker.C:
|
|
b.flushNow("interval")
|
|
case <-b.stop:
|
|
b.flushNow("shutdown")
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
// Add appends a reading, flushing immediately if the buffer just reached
|
|
// maxSize rather than waiting for the next tick.
|
|
func (b *Batcher) Add(r telemetry.Reading) {
|
|
b.mu.Lock()
|
|
b.buf = append(b.buf, r)
|
|
full := len(b.buf) >= b.maxSize
|
|
b.mu.Unlock()
|
|
|
|
if full {
|
|
b.flushNow("size")
|
|
}
|
|
}
|
|
|
|
// Stop flushes any remaining buffered readings and waits for the flush loop
|
|
// to exit. Safe to call once, after Start.
|
|
func (b *Batcher) Stop() {
|
|
close(b.stop)
|
|
<-b.done
|
|
}
|
|
|
|
func (b *Batcher) flushNow(reason string) {
|
|
b.mu.Lock()
|
|
if len(b.buf) == 0 {
|
|
b.mu.Unlock()
|
|
return
|
|
}
|
|
readings := b.buf
|
|
b.buf = nil
|
|
b.mu.Unlock()
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), b.flushTimeout)
|
|
defer cancel()
|
|
|
|
if err := b.flush(ctx, readings); err != nil {
|
|
b.logger.Error("batch flush failed", "reason", reason, "size", len(readings), "error", err)
|
|
return
|
|
}
|
|
b.logger.Debug("batch flushed", "reason", reason, "size", len(readings))
|
|
}
|