Documentation
¶
Index ¶
Constants ¶
This section is empty.
Variables ¶
var ErrBufferFull = errors.New("batchwriter: ingestion channel is full, record dropped")
ErrBufferFull is returned by Enqueue when the ingestion channel is at capacity.
Functions ¶
This section is empty.
Types ¶
type BatchInserter ¶
type BatchInserter interface {
CopyFrom(ctx context.Context, tableName pgx.Identifier, columnNames []string, rowSrc pgx.CopyFromSource) (int64, error)
}
BatchInserter defines the interface for executing bulk CopyFrom operations.
type BatchWriter ¶
type BatchWriter struct {
// contains filtered or unexported fields
}
BatchWriter buffers incoming TelemetryRecords and periodically flushes them to TimescaleDB via pgx.CopyFrom.
func NewBatchWriter ¶
func NewBatchWriter(inserter BatchInserter, cfg config.IngestionConfig, tableName string) *BatchWriter
NewBatchWriter creates a new BatchWriter instance from validated IngestionConfig and target table name.
func (*BatchWriter) DroppedCount ¶
func (w *BatchWriter) DroppedCount() int64
DroppedCount returns the number of records dropped due to a full ingestion channel.
func (*BatchWriter) Enqueue ¶
func (w *BatchWriter) Enqueue(record TelemetryRecord) error
Enqueue attempts a non-blocking send of a TelemetryRecord into the ingestion channel. If the channel is at capacity, the record is dropped and ErrBufferFull is returned. Callers should log or count dropped records at their discretion.
func (*BatchWriter) Flush ¶
func (w *BatchWriter) Flush(ctx context.Context) error
Flush writes the current in-memory buffer to TimescaleDB using pgx.CopyFrom.
func (*BatchWriter) Start ¶
func (w *BatchWriter) Start(ctx context.Context)
Start launches the background worker that processes incoming records and periodic timer flushes.
func (*BatchWriter) Stats ¶
func (w *BatchWriter) Stats() (int64, int64)
Stats returns total inserted count, total batch count, and total dropped record count.
func (*BatchWriter) Stop ¶
func (w *BatchWriter) Stop()
Stop signals the writer to flush pending records and wait for worker completion.
type DB ¶
type DB struct {
// contains filtered or unexported fields
}
DB wraps the TimescaleDB connection pool.
func NewDB ¶
NewDB creates and tests a connection pool to TimescaleDB/PostgreSQL using validated configuration parameters.
func (*DB) AsBatchInserter ¶
func (d *DB) AsBatchInserter() BatchInserter
AsBatchInserter returns the underlying pool as a BatchInserter interface, allowing the BatchWriter to use it without exposing the full pgxpool.Pool.
Source Files