diff --git a/gr26/job/ingest_batch.go b/gr26/job/ingest_batch.go index a97626aa..17ea21bd 100644 --- a/gr26/job/ingest_batch.go +++ b/gr26/job/ingest_batch.go @@ -19,9 +19,12 @@ import ( ulid "github.com/gaucho-racing/ulid-go" gr26config "github.com/gaucho-racing/mapache/gr26/config" + "github.com/gaucho-racing/mapache/gr26/model" "github.com/gaucho-racing/mapache/gr26/pkg/foreman" "github.com/gaucho-racing/mapache/gr26/pkg/logger" "github.com/gaucho-racing/mapache/gr26/service" + + mapache "github.com/gaucho-racing/mapache/mapache-go/v3" ) // ─── producer side: gr26.ingest_batch ─────────────────────────────────────── @@ -97,12 +100,15 @@ func IngestBatchHandler(ctx context.Context, job foreman.Job, progress *foreman. return nil, err } - res, err := processFile(ctx, client, shelterKey(p.VehicleID, p.FileULID), progress) + // Report stats on both paths — foreman records the result on a failed run + // too, so a partial-failure attempt still says what landed and what didn't. + res, procErr := processFile(ctx, client, shelterKey(p.VehicleID, p.FileULID), progress) + res.FileULID = p.FileULID + payload, err := json.Marshal(res) if err != nil { - return nil, err + return nil, errors.Join(procErr, fmt.Errorf("encode result: %w", err)) } - res.FileULID = p.FileULID - return json.Marshal(res) + return payload, procErr } func processFile(ctx context.Context, client *s3.Client, key string, progress *foreman.Progress) (ingestResult, error) { @@ -133,87 +139,120 @@ func processFile(ctx context.Context, client *s3.Client, key string, progress *f stats := newIngestStats() const chunk = 4096 rows := make([]shelterRow, chunk) + cans := make([]model.CAN, 0, chunk) + signals := make([]mapache.Signal, 0, chunk) total := 0 + chunkIdx := 0 + var flushErr error for { n, readErr := pr.Read(rows) + cans, signals = cans[:0], signals[:0] for i := 0; i < n; i++ { - dispatchRow(rows[i], stats) + can, sigs, ok := dispatchRow(rows[i], stats) + if !ok { + continue + } + cans = append(cans, can) + signals = append(signals, sigs...) + } + // One flush per chunk. Keep going past a failed insert so the rest of + // the file still lands, but hold onto the first error and fail the job + // below — a dropped chunk is up to 4,096 frames, so it has to burn a + // foreman attempt rather than report success (RMT dedup makes the + // re-processed chunks harmless). + if err := service.CreateCANs(cans); err != nil { + logger.SugarLogger.Errorf("[SHELTER] %s: chunk %d: failed to insert %d CAN records: %s", key, chunkIdx, len(cans), err) + stats.recordInsertFailure(insertTargetCANs, chunkIdx, len(cans), err) + if flushErr == nil { + flushErr = fmt.Errorf("insert cans: %w", err) + } + } + if err := service.CreateSignals(signals); err != nil { + logger.SugarLogger.Errorf("[SHELTER] %s: chunk %d: failed to insert %d signals: %s", key, chunkIdx, len(signals), err) + stats.recordInsertFailure(insertTargetSignals, chunkIdx, len(signals), err) + if flushErr == nil { + flushErr = fmt.Errorf("insert signals: %w", err) + } } total += n + chunkIdx++ progress.Set(int64(total), totalRows, "decoding parquet rows") if errors.Is(readErr, io.EOF) { break } if readErr != nil { - return ingestResult{}, fmt.Errorf("parquet read: %w", readErr) + return stats.result(total, time.Since(start)), fmt.Errorf("parquet read: %w", readErr) } } duration := time.Since(start) + if flushErr != nil { + // Every row decoded, so the counters legitimately read 100% — don't + // claim "complete" though, the inserts didn't all land. Best-effort: + // only lands if a heartbeat tick catches it before the handler returns. + progress.Set(int64(total), totalRows, fmt.Sprintf("%d chunk insert(s) failed", stats.failedInserts)) + logger.SugarLogger.Errorf("[SHELTER] %s: %d rows in %s, %d chunk insert(s) failed (cans_dropped=%d signals_dropped=%d)", + key, total, duration, stats.failedInserts, stats.cansDropped, stats.signalsDropped) + return stats.result(total, duration), flushErr + } + // Pin to (total, total) so the final heartbeat stores a clean 100%. progress.Set(totalRows, totalRows, "complete") logger.SugarLogger.Infof("[SHELTER] %s: %d rows in %s (decoded=%d unknown=%d errors=%d invalid_timestamp=%d)", key, total, duration, stats.decoded, stats.unknown, stats.decodeError, stats.invalidTimestamp) - return ingestResult{ - TotalRows: total, - Decoded: stats.decoded, - UnknownCanID: stats.unknown, - DecodeError: stats.decodeError, - InvalidTimestamp: stats.invalidTimestamp, - DurationMs: duration.Milliseconds(), - UnknownBreakdown: topUnknown(stats.unknownByCanID, 10), - DecodeErrorBreakdown: topErrors(stats.errorByCanID, 10), - InvalidTimestampSamples: stats.invalidTimestampSamples, - }, nil + return stats.result(total, duration), nil } -func dispatchRow(r shelterRow, stats *ingestStats) { +// dispatchRow is the cold-storage decode path — persistence happens in +// per-chunk flushes upstream. UploadKey stays 0 (bucket access is the +// trust boundary), and no WS/side-channel firing — historical data, and +// we don't want to re-enqueue shelter batches. +func dispatchRow(r shelterRow, stats *ingestStats) (model.CAN, []mapache.Signal, bool) { // Topic format: gr26/{vehicle}/{node}/0x{can_id_hex} parts := strings.Split(r.Topic, "/") if len(parts) != 4 { - return + return model.CAN{}, nil, false } nodeID := parts[2] canIDStr := strings.TrimPrefix(parts[3], "0x") canIDInt, err := strconv.ParseInt(canIDStr, 16, 64) if err != nil { - return - } - replayFrame(r.VehicleID, nodeID, int(canIDInt), int(r.Timestamp), r.Data, stats) -} - -// replayFrame is the cold-storage decode-and-persist path. UploadKey -// stays 0 (bucket access is the trust boundary), and no WS/side-channel -// firing — historical data, and we don't want to re-enqueue shelter batches. -func replayFrame(vehicleID, nodeID string, canID, ts int, data []byte, stats *ingestStats) { - can, signals := service.ProcessFrame(vehicleID, nodeID, canID, ts, data) - stats.record(canID, nodeID, ts, can.Metadata) - - if _, err := service.CreateCAN(can); err != nil { - logger.SugarLogger.Infof("Error creating CAN record: %s", err) - } - if len(signals) > 0 { - if err := service.CreateSignals(signals); err != nil { - logger.SugarLogger.Infof("Error creating signals: %s", err) - } + return model.CAN{}, nil, false } + can, signals := service.ProcessFrame(r.VehicleID, nodeID, int(canIDInt), int(r.Timestamp), r.Data) + stats.record(int(canIDInt), nodeID, int(r.Timestamp), can.Metadata) + return can, signals, true } // ─── result reporting ─────────────────────────────────────────────────────── const invalidTimestampSampleLimit = 10 +const insertFailureSampleLimit = 10 type ingestStats struct { decoded int unknown int decodeError int invalidTimestamp int + failedInserts int + cansDropped int + signalsDropped int unknownByCanID map[int]int errorByCanID map[int]decodeErrorSample invalidTimestampSamples []invalidTimestampSample + insertFailureSamples []insertFailureSample } +// insertTarget names which per-chunk flush failed, and lands verbatim in +// the job result. +type insertTarget string + +const ( + insertTargetCANs insertTarget = "cans" + insertTargetSignals insertTarget = "signals" +) + type decodeErrorSample struct { count int sample string @@ -226,6 +265,13 @@ type invalidTimestampSample struct { NodeID string `json:"node_id"` } +type insertFailureSample struct { + Target insertTarget `json:"target"` + Chunk int `json:"chunk"` + Rows int `json:"rows"` + Error string `json:"error"` +} + func newIngestStats() *ingestStats { return &ingestStats{ unknownByCanID: make(map[int]int), @@ -233,6 +279,26 @@ func newIngestStats() *ingestStats { } } +// recordInsertFailure notes a chunk that decoded fine but never landed in +// ClickHouse, so the result can distinguish "decoded" from "persisted". +func (s *ingestStats) recordInsertFailure(target insertTarget, chunk, rows int, err error) { + s.failedInserts++ + switch target { + case insertTargetCANs: + s.cansDropped += rows + case insertTargetSignals: + s.signalsDropped += rows + } + if len(s.insertFailureSamples) < insertFailureSampleLimit { + s.insertFailureSamples = append(s.insertFailureSamples, insertFailureSample{ + Target: target, + Chunk: chunk, + Rows: rows, + Error: err.Error(), + }) + } +} + func (s *ingestStats) record(canID int, nodeID string, ts int, metadata []byte) { var meta struct { Status string `json:"status"` @@ -279,7 +345,10 @@ type errorBreakdownEntry struct { SampleError string `json:"sample_error,omitempty"` } -// ingestResult lands on foreman.job.result. Top-N caps bound payload size. +// ingestResult lands on foreman.job.result — on failed attempts too, so a +// partial-failure run still reports what landed. Top-N caps bound payload +// size. Decoded counts what decoded, not what persisted: subtract +// CANsDropped / SignalsDropped for that. type ingestResult struct { FileULID string `json:"file_ulid,omitempty"` TotalRows int `json:"total_rows"` @@ -287,10 +356,34 @@ type ingestResult struct { UnknownCanID int `json:"unknown_can_id"` DecodeError int `json:"decode_error"` InvalidTimestamp int `json:"invalid_timestamp,omitempty"` + FailedInserts int `json:"failed_inserts,omitempty"` + CANsDropped int `json:"cans_dropped,omitempty"` + SignalsDropped int `json:"signals_dropped,omitempty"` DurationMs int64 `json:"duration_ms"` UnknownBreakdown []breakdownEntry `json:"unknown_breakdown,omitempty"` DecodeErrorBreakdown []errorBreakdownEntry `json:"decode_error_breakdown,omitempty"` InvalidTimestampSamples []invalidTimestampSample `json:"invalid_timestamp_samples,omitempty"` + InsertFailureSamples []insertFailureSample `json:"insert_failure_samples,omitempty"` +} + +// result snapshots the stats into the foreman result payload. Called on the +// success and failure paths alike so a failed attempt still reports coverage. +func (s *ingestStats) result(totalRows int, duration time.Duration) ingestResult { + return ingestResult{ + TotalRows: totalRows, + Decoded: s.decoded, + UnknownCanID: s.unknown, + DecodeError: s.decodeError, + InvalidTimestamp: s.invalidTimestamp, + FailedInserts: s.failedInserts, + CANsDropped: s.cansDropped, + SignalsDropped: s.signalsDropped, + DurationMs: duration.Milliseconds(), + UnknownBreakdown: topUnknown(s.unknownByCanID, 10), + DecodeErrorBreakdown: topErrors(s.errorByCanID, 10), + InvalidTimestampSamples: s.invalidTimestampSamples, + InsertFailureSamples: s.insertFailureSamples, + } } func topUnknown(m map[int]int, n int) []breakdownEntry { diff --git a/gr26/pkg/foreman/worker.go b/gr26/pkg/foreman/worker.go index 5e3b7e96..0c9c8a63 100644 --- a/gr26/pkg/foreman/worker.go +++ b/gr26/pkg/foreman/worker.go @@ -67,6 +67,10 @@ import ( // Returning an error → Fail. By default it's retryable with the // Worker's DefaultBackoffSec; return a *FailError to override, or // errors.Is(err, ErrPermanent) for a non-retryable terminalization. +// +// Bytes and an error together are valid: the run still fails, but the +// result is recorded on it. Partial-failure handlers should use this to +// report what they did land before giving up. type Handler func(ctx context.Context, job Job, progress *Progress) (json.RawMessage, error) // ErrPermanent marks a handler error as non-retryable. The job @@ -308,6 +312,7 @@ func (w *Worker) handleOne(parent context.Context, claimed Claimed, onErr func(e Error: msg, Retryable: retryable, BackoffSec: backoff, + Result: result, }); err != nil { onErr(fmt.Errorf("fail: %w", err)) } diff --git a/gr26/service/can.go b/gr26/service/can.go index d241cd77..ef7794de 100644 --- a/gr26/service/can.go +++ b/gr26/service/can.go @@ -89,6 +89,7 @@ func GetSignalsForCAN(canMessageID string) ([]mapache.Signal, error) { // Dedup on (vehicle_id, node_id, timestamp) is handled by the // ReplacingMergeTree engine — latest created_at wins on merge. const insertCANSQL = `INSERT INTO gr26_can (id, vehicle_id, node_id, timestamp, can_id, bytes, upload_key, metadata) VALUES (?, ?, ?, ?, ?, ?, ?, ?)` +const insertCANBatchSQL = `INSERT INTO gr26_can (id, vehicle_id, node_id, timestamp, can_id, bytes, upload_key, metadata)` func CreateCAN(can model.CAN) (model.CAN, error) { can.ID = ulid.Make().Prefixed("can") @@ -105,6 +106,32 @@ func CreateCAN(can model.CAN) (model.CAN, error) { return can, nil } +// CreateCANs is the bulk-ingest counterpart of CreateCAN: one buffered block +// insert per call — a single round trip regardless of frame count. async_insert +// (InsertCtx) doesn't apply to native block inserts, so plain ctx here. +func CreateCANs(cans []model.CAN) error { + for i := range cans { + cans[i].ID = ulid.Make().Prefixed("can") + } + if !config.ClickhouseEnabled() || len(cans) == 0 { + return nil + } + batch, err := database.Conn.PrepareBatch(context.Background(), insertCANBatchSQL) + if err != nil { + return err + } + defer batch.Close() + for _, c := range cans { + if err := batch.Append( + c.ID, c.VehicleID, c.NodeID, int64(c.Timestamp), int32(c.CANID), + string(c.Bytes), int32(c.UploadKey), string(c.Metadata), + ); err != nil { + return err + } + } + return batch.Send() +} + func scanCANRow(row driver.Row) (model.CAN, error) { var ( id, vehicleID, nodeID string