shishengliang/lingniu-vehicle-ingest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(platform): harden telemetry pipeline and unify Semi UI workspaces

Browse Source
This commit is contained in:
lingniu
2026-07-18 00:26:36 +08:00
parent 65b4e4f055
commit 159c80b0ae
136 changed files with 21616 additions and 1785 deletions
Download Patch File Download Diff File Expand all files Collapse all files

173
go/vehicle-gateway/cmd/stat-writer/main.go
View File

@@ -8,6 +8,7 @@ import (
"fmt"
"os"
"os/signal"
"path/filepath"
"sort"
"strconv"
"strings"
@@ -54,6 +55,7 @@ func main() {
registry.SetGauge("vehicle_stat_cache_retention_seconds", nil, cfg.CacheRetention.Seconds())
registry.SetGauge("vehicle_stat_cache_cleanup_interval_seconds", nil, cfg.CacheCleanupInterval.Seconds())
registry.SetGauge("vehicle_stat_baseline_miss_ttl_seconds", nil, cfg.BaselineMissTTL.Seconds())
registry.SetGauge("vehicle_stat_baseline_hit_ttl_seconds", nil, cfg.BaselineHitTTL.Seconds())
registry.SetGauge("vehicle_stat_config", metrics.Labels{"setting": "workers"}, float64(cfg.Workers))
health.Start(ctx, logger, health.NewServer(env("HEALTH_ADDR", ""), "vehicle-stat-writer", []health.Check{
{Name: "mysql", Check: db.PingContext},
@@ -65,6 +67,7 @@ func main() {
writer.SetCacheRetention(cfg.CacheRetention)
writer.SetCacheCleanupInterval(cfg.CacheCleanupInterval)
writer.SetBaselineMissTTL(cfg.BaselineMissTTL)
writer.SetBaselineHitTTL(cfg.BaselineHitTTL)
writer.SetMaxCacheEntries(cfg.CacheMaxEntries)
if cfg.EnsureSchema {
if err := writer.EnsureSchema(ctx); err != nil {
@@ -89,14 +92,15 @@ func main() {
delay: cfg.RetryDelay,
registry: registry,
}
quarantiner := fileStatMessageQuarantiner{dir: cfg.QuarantineDir}
logger.Info("stat writer started", "group", cfg.KafkaGroup, "topics", strings.Join(cfg.KafkaTopics, ","), "workers", cfg.Workers, "project_interval_seconds", cfg.ProjectInterval.Seconds(), "source_touch_interval_seconds", cfg.SourceTouchInterval.Seconds(), "cache_retention_seconds", cfg.CacheRetention.Seconds(), "cache_cleanup_interval_seconds", cfg.CacheCleanupInterval.Seconds(), "baseline_miss_ttl_seconds", cfg.BaselineMissTTL.Seconds(), "cache_max_entries", cfg.CacheMaxEntries, "batch_size", cfg.BatchSize, "batch_wait_ms", cfg.BatchWait, "retry_attempts", cfg.RetryAttempts, "retry_delay_ms", cfg.RetryDelay.Milliseconds())
logger.Info("stat writer started", "group", cfg.KafkaGroup, "topics", strings.Join(cfg.KafkaTopics, ","), "workers", cfg.Workers, "project_interval_seconds", cfg.ProjectInterval.Seconds(), "source_touch_interval_seconds", cfg.SourceTouchInterval.Seconds(), "cache_retention_seconds", cfg.CacheRetention.Seconds(), "cache_cleanup_interval_seconds", cfg.CacheCleanupInterval.Seconds(), "baseline_miss_ttl_seconds", cfg.BaselineMissTTL.Seconds(), "baseline_hit_ttl_seconds", cfg.BaselineHitTTL.Seconds(), "cache_max_entries", cfg.CacheMaxEntries, "batch_size", cfg.BatchSize, "batch_wait_ms", cfg.BatchWait, "retry_attempts", cfg.RetryAttempts, "retry_delay_ms", cfg.RetryDelay.Milliseconds(), "quarantine_dir", cfg.QuarantineDir)
var workers sync.WaitGroup
for workerID := 1; workerID <= cfg.Workers; workerID++ {
workers.Add(1)
go func(id int) {
defer workers.Done()
runStatConsumer(ctx, logger, registry, appender, cfg, id)
runStatConsumer(ctx, logger, registry, appender, quarantiner, cfg, id)
}(workerID)
}
workers.Wait()
@@ -105,7 +109,7 @@ func main() {
func runStatConsumer(ctx context.Context, logger interface {
Error(string, ...any)
Warn(string, ...any)
}, registry *metrics.Registry, appender statAppender, cfg config, workerID int) {
}, registry *metrics.Registry, appender statAppender, quarantiner statMessageQuarantiner, cfg config, workerID int) {
reader := kafka.NewReader(kafka.ReaderConfig{
Brokers: cfg.KafkaBrokers,
GroupID: cfg.KafkaGroup,
@@ -129,7 +133,7 @@ func runStatConsumer(ctx context.Context, logger interface {
continue
}
batch := collectStatBatch(ctx, reader, message, cfg.BatchSize, time.Duration(cfg.BatchWait)*time.Millisecond)
processStatBatchReliablyForWorker(ctx, logger, registry, appender, reader, batch, cfg.RetryDelay, workerLabels)
processStatBatchReliablyForWorker(ctx, logger, registry, appender, reader, quarantiner, batch, cfg.RetryDelay, workerLabels)
}
}
@@ -155,6 +159,10 @@ type kafkaMessageFetcher interface {
FetchMessage(context.Context) (kafka.Message, error)
}
type statMessageQuarantiner interface {
Quarantine(context.Context, kafka.Message, error) error
}
type statBatchItem struct {
message kafka.Message
processed bool
@@ -163,6 +171,8 @@ type statBatchItem struct {
type statBatchOutcome struct {
commitMessages []kafka.Message
retryMessages []kafka.Message
failedMessage *kafka.Message
writeErr error
}
var statWriteDurationBucketsMS = []float64{1, 5, 10, 25, 50, 100, 250, 500, 1000, 5000}
@@ -251,7 +261,12 @@ func processStatBatchForWorker(ctx context.Context, logger interface {
addStatMetric(registry, "vehicle_stat_writes_total", message, "error")
logger.Error("mysql append failed", "topic", message.Topic, "partition", message.Partition, "offset", message.Offset, "event_id", env.StableEventID(), "error", err)
committed, commitErr := commitStatProcessedPrefixAfterFailure(messageCtx, logger, registry, committer, items)
outcome := statBatchOutcome{retryMessages: eventbus.MessagesAfterCommittedPrefixes(messages, committed)}
failedMessage := message
outcome := statBatchOutcome{
retryMessages: eventbus.MessagesAfterCommittedPrefixes(messages, committed),
failedMessage: &failedMessage,
writeErr: err,
}
if commitErr != nil {
outcome.commitMessages = committed
}
@@ -279,13 +294,13 @@ func processStatBatchReliably(ctx context.Context, logger interface {
Error(string, ...any)
Warn(string, ...any)
}, registry *metrics.Registry, appender statAppender, committer kafkaMessageCommitter, messages []kafka.Message, retryDelay time.Duration) {
processStatBatchReliablyForWorker(ctx, logger, registry, appender, committer, messages, retryDelay, nil)
processStatBatchReliablyForWorker(ctx, logger, registry, appender, committer, nil, messages, retryDelay, nil)
}
func processStatBatchReliablyForWorker(ctx context.Context, logger interface {
Error(string, ...any)
Warn(string, ...any)
}, registry *metrics.Registry, appender statAppender, committer kafkaMessageCommitter, messages []kafka.Message, retryDelay time.Duration, workerLabels metrics.Labels) {
}, registry *metrics.Registry, appender statAppender, committer kafkaMessageCommitter, quarantiner statMessageQuarantiner, messages []kafka.Message, retryDelay time.Duration, workerLabels metrics.Labels) {
defer registry.SetGauge("vehicle_stat_retry_pending_messages", workerLabels, 0)
pending := messages
for len(pending) > 0 {
@@ -297,6 +312,19 @@ func processStatBatchReliablyForWorker(ctx context.Context, logger interface {
}
}
pending = outcome.retryMessages
if outcome.writeErr != nil && outcome.failedMessage != nil && isQuarantinableStatError(outcome.writeErr) && quarantiner != nil {
failed := *outcome.failedMessage
if err := quarantiner.Quarantine(ctx, failed, outcome.writeErr); err != nil {
logger.Error("stat message quarantine failed", "topic", failed.Topic, "partition", failed.Partition, "offset", failed.Offset, "error", err)
registry.IncCounter("vehicle_stat_quarantine_total", metrics.Labels{"status": "error", "topic": failed.Topic})
} else {
registry.IncCounter("vehicle_stat_quarantine_total", metrics.Labels{"status": "ok", "topic": failed.Topic})
logger.Warn("quarantined permanent stat write failure", "topic", failed.Topic, "partition", failed.Partition, "offset", failed.Offset, "error", outcome.writeErr)
if retryStatCommit(ctx, logger, registry, committer, []kafka.Message{failed}, retryDelay) {
pending = messagesExceptStatMessage(pending, failed)
}
}
}
registry.SetGauge("vehicle_stat_retry_pending_messages", workerLabels, float64(len(pending)))
if len(pending) == 0 {
return
@@ -308,6 +336,105 @@ func processStatBatchReliablyForWorker(ctx context.Context, logger interface {
}
}
type quarantinedStatMessage struct {
QuarantinedAt string `json:"quarantined_at"`
Topic string `json:"topic"`
Partition int `json:"partition"`
Offset int64 `json:"offset"`
Key []byte `json:"key,omitempty"`
Value []byte `json:"value"`
Error string `json:"error"`
}
type fileStatMessageQuarantiner struct {
dir string
}
func (q fileStatMessageQuarantiner) Quarantine(ctx context.Context, message kafka.Message, cause error) error {
if err := ctx.Err(); err != nil {
return err
}
dir := strings.TrimSpace(q.dir)
if dir == "" {
return errors.New("stat quarantine directory is empty")
}
if err := os.MkdirAll(dir, 0o750); err != nil {
return fmt.Errorf("create quarantine directory: %w", err)
}
record := quarantinedStatMessage{
QuarantinedAt: time.Now().UTC().Format(time.RFC3339Nano),
Topic: message.Topic,
Partition: message.Partition,
Offset: message.Offset,
Key: message.Key,
Value: message.Value,
Error: cause.Error(),
}
payload, err := json.Marshal(record)
if err != nil {
return fmt.Errorf("marshal quarantine record: %w", err)
}
name := fmt.Sprintf("%s-%d-%d.json", sanitizeStatQuarantineName(message.Topic), message.Partition, message.Offset)
temporary, err := os.CreateTemp(dir, "."+name+"-*")
if err != nil {
return fmt.Errorf("create quarantine record: %w", err)
}
temporaryName := temporary.Name()
removeTemporary := true
defer func() {
_ = temporary.Close()
if removeTemporary {
_ = os.Remove(temporaryName)
}
}()
if err := temporary.Chmod(0o600); err != nil {
return fmt.Errorf("secure quarantine record: %w", err)
}
if _, err := temporary.Write(payload); err != nil {
return fmt.Errorf("write quarantine record: %w", err)
}
if err := temporary.Sync(); err != nil {
return fmt.Errorf("sync quarantine record: %w", err)
}
if err := temporary.Close(); err != nil {
return fmt.Errorf("close quarantine record: %w", err)
}
if err := os.Rename(temporaryName, filepath.Join(dir, name)); err != nil {
return fmt.Errorf("publish quarantine record: %w", err)
}
removeTemporary = false
return nil
}
func sanitizeStatQuarantineName(value string) string {
value = strings.TrimSpace(value)
if value == "" {
return "unknown-topic"
}
var out strings.Builder
for _, r := range value {
if r >= 'a' && r <= 'z' || r >= 'A' && r <= 'Z' || r >= '0' && r <= '9' || r == '-' || r == '_' || r == '.' {
out.WriteRune(r)
} else {
out.WriteByte('_')
}
}
return out.String()
}
func messagesExceptStatMessage(messages []kafka.Message, excluded kafka.Message) []kafka.Message {
out := make([]kafka.Message, 0, len(messages))
removed := false
for _, message := range messages {
if !removed && message.Topic == excluded.Topic && message.Partition == excluded.Partition && message.Offset == excluded.Offset {
removed = true
continue
}
out = append(out, message)
}
return out
}
func retryStatCommit(ctx context.Context, logger interface {
Error(string, ...any)
Warn(string, ...any)
@@ -447,7 +574,7 @@ func (a retryStatAppender) AppendWithResult(ctx context.Context, env envelope.Fr
var err error
for attempt := 1; attempt <= attempts; attempt++ {
result, err = appendStatEnvelope(ctx, a.delegate, env)
if err == nil || !isTransientMySQLStatError(err) {
if err == nil || (!isTransientMySQLStatError(err) && !isQuarantinableStatError(err)) {
return result, err
}
if attempt == attempts {
@@ -512,6 +639,32 @@ func isTransientMySQLStatError(err error) bool {
strings.Contains(text, "no route to host")
}
func isQuarantinableStatError(err error) bool {
if err == nil || errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
return false
}
text := strings.ToLower(strings.TrimSpace(err.Error()))
// Only deterministic row-data failures may advance past a Kafka message
// after the configured finite retries. Schema, permission and unknown
// application errors remain failure-closed so a systemic outage cannot
// silently quarantine an entire stream.
return strings.Contains(text, "error 1048") ||
strings.Contains(text, "cannot be null") ||
strings.Contains(text, "error 1264") ||
strings.Contains(text, "out of range value") ||
strings.Contains(text, "error 1265") ||
strings.Contains(text, "data truncated") ||
strings.Contains(text, "error 1292") ||
strings.Contains(text, "incorrect datetime value") ||
strings.Contains(text, "error 1366") ||
strings.Contains(text, "incorrect decimal value") ||
strings.Contains(text, "incorrect integer value") ||
strings.Contains(text, "error 1406") ||
strings.Contains(text, "data too long for column") ||
strings.Contains(text, "error 3819") ||
strings.Contains(text, "check constraint")
}
func addStatMetric(registry *metrics.Registry, name string, message kafka.Message, status string) {
if registry == nil {
return
@@ -686,6 +839,7 @@ type config struct {
CacheRetention time.Duration
CacheCleanupInterval time.Duration
BaselineMissTTL time.Duration
BaselineHitTTL time.Duration
CacheMaxEntries int
Workers int
BatchSize int
@@ -694,6 +848,7 @@ type config struct {
RetryDelay time.Duration
NormalizePlatformSourcesOnStart bool
NormalizePlatformSourcesTimeout time.Duration
QuarantineDir string
}
func (c config) Validate() error {
@@ -728,6 +883,7 @@ func loadConfig() config {
CacheRetention: time.Duration(envInt("STATS_CACHE_RETENTION_HOURS", 72)) * time.Hour,
CacheCleanupInterval: time.Duration(envInt("STATS_CACHE_CLEANUP_INTERVAL_SECONDS", 600)) * time.Second,
BaselineMissTTL: time.Duration(envInt("STATS_BASELINE_MISS_TTL_SECONDS", 60)) * time.Second,
BaselineHitTTL: time.Duration(envInt("STATS_BASELINE_HIT_TTL_SECONDS", 300)) * time.Second,
CacheMaxEntries: envInt("STATS_CACHE_MAX_ENTRIES", 1000000),
Workers: envInt("STATS_WORKERS", 3),
BatchSize: envInt("STATS_BATCH_SIZE", 200),
@@ -736,6 +892,7 @@ func loadConfig() config {
RetryDelay: time.Duration(envInt("STATS_RETRY_DELAY_MS", 20)) * time.Millisecond,
NormalizePlatformSourcesOnStart: env("STATS_NORMALIZE_PLATFORM_SOURCES_ON_START", "true") != "false",
NormalizePlatformSourcesTimeout: time.Duration(envInt("STATS_NORMALIZE_PLATFORM_SOURCES_TIMEOUT_SECONDS", 30)) * time.Second,
QuarantineDir: env("STATS_QUARANTINE_DIR", "/var/lib/lingniu-go-native/stat-writer-quarantine"),
}
}

143
go/vehicle-gateway/cmd/stat-writer/main_test.go
View File

@@ -4,6 +4,8 @@ import (
"context"
"encoding/json"
"errors"
"os"
"path/filepath"
"strings"
"testing"
"time"
@@ -664,6 +666,72 @@ func TestProcessStatBatchReliablyRetriesFailedSuffix(t *testing.T) {
}
}
func TestProcessStatBatchReliablyQuarantinesPermanentFailureAndContinues(t *testing.T) {
env := envelope.FrameEnvelope{Protocol: envelope.ProtocolJT808, Phone: "13307795425", MessageID: "0x0200"}
payload := marshalStatTestFieldsEnvelope(t, env)
appender := &contextCheckingStatAppender{failOnCount: 1, err: errors.New("Error 1048 (23000): Column 'daily_mileage_km' cannot be null")}
committer := &contextCheckingStatCommitter{}
quarantiner := &recordingStatQuarantiner{}
registry := metrics.NewRegistry()
messages := []kafka.Message{
{Topic: "vehicle.fields.go.jt808.v1", Partition: 1, Offset: 10, Value: payload},
{Topic: "vehicle.fields.go.jt808.v1", Partition: 1, Offset: 11, Value: payload},
}
processStatBatchReliablyForWorker(
context.Background(),
discardStatLogger{},
registry,
appender,
committer,
quarantiner,
messages,
time.Nanosecond,
nil,
)
if appender.count != 2 {
t.Fatalf("append attempts = %d, want failed message once and following message once", appender.count)
}
if quarantiner.count != 1 || quarantiner.message.Offset != 10 {
t.Fatalf("quarantine = %d/%d, want one message at offset 10", quarantiner.count, quarantiner.message.Offset)
}
if got := committedOffsets(committer.messages); len(got) != 2 || got[0] != 10 || got[1] != 11 {
t.Fatalf("committed offsets = %#v, want [10 11]", got)
}
text := registry.Render()
if !strings.Contains(text, `vehicle_stat_quarantine_total{status="ok",topic="vehicle.fields.go.jt808.v1"} 1`) {
t.Fatalf("quarantine success metric missing:\n%s", text)
}
}
func TestFileStatMessageQuarantinerPersistsOriginalKafkaMessage(t *testing.T) {
dir := t.TempDir()
message := kafka.Message{
Topic: "vehicle.fields.go.jt808.v1",
Partition: 2,
Offset: 5094052,
Key: []byte("vehicle-key"),
Value: []byte(`{"event_id":"poison:fields"}`),
}
wantErr := errors.New("permanent mysql error")
if err := (fileStatMessageQuarantiner{dir: dir}).Quarantine(context.Background(), message, wantErr); err != nil {
t.Fatalf("Quarantine() error = %v", err)
}
payload, err := os.ReadFile(filepath.Join(dir, "vehicle.fields.go.jt808.v1-2-5094052.json"))
if err != nil {
t.Fatalf("read quarantine record: %v", err)
}
var record quarantinedStatMessage
if err := json.Unmarshal(payload, &record); err != nil {
t.Fatalf("unmarshal quarantine record: %v", err)
}
if record.Topic != message.Topic || record.Partition != message.Partition || record.Offset != message.Offset || string(record.Key) != string(message.Key) || string(record.Value) != string(message.Value) || record.Error != wantErr.Error() {
t.Fatalf("quarantine record = %#v, want original message and error", record)
}
}
func TestProcessStatBatchReliablyRetriesCommitWithoutReappending(t *testing.T) {
env := envelope.FrameEnvelope{Protocol: envelope.ProtocolJT808, Phone: "13307795425", MessageID: "0x0200"}
payload := marshalStatTestFieldsEnvelope(t, env)
@@ -826,6 +894,30 @@ func TestRetryStatAppenderRecordsExhaustedTransientError(t *testing.T) {
}
}
func TestRetryStatAppenderRetriesQuarantinableDataErrorBeforeIsolation(t *testing.T) {
registry := metrics.NewRegistry()
delegate := &contextCheckingStatAppender{err: errors.New("Error 1048 (23000): Column 'daily_mileage_km' cannot be null")}
appender := retryStatAppender{delegate: delegate, attempts: 3, registry: registry}
_, err := appender.AppendWithResult(context.Background(), envelope.FrameEnvelope{Protocol: envelope.ProtocolJT808})
if err == nil {
t.Fatal("AppendWithResult() error = nil, want exhausted data error")
}
if delegate.count != 3 {
t.Fatalf("append attempts = %d, want configured finite retries", delegate.count)
}
text := registry.Render()
for _, want := range []string{
`vehicle_stat_write_retries_total{operation="single",status="retry"} 2`,
`vehicle_stat_write_retries_total{operation="single",status="exhausted"} 1`,
} {
if !strings.Contains(text, want) {
t.Fatalf("retry metric missing %s:\n%s", want, text)
}
}
}
func TestRetryStatAppenderDoesNotRetryNonTransientError(t *testing.T) {
wantErr := errors.New("validation failed")
delegate := &contextCheckingStatAppender{err: wantErr}
@@ -841,6 +933,34 @@ func TestRetryStatAppenderDoesNotRetryNonTransientError(t *testing.T) {
}
}
func TestIsQuarantinableStatErrorOnlyAcceptsDeterministicRowDataFailures(t *testing.T) {
for _, err := range []error{
errors.New("Error 1048 (23000): Column 'daily_mileage_km' cannot be null"),
errors.New("Error 1264: Out of range value for column 'daily_mileage_km'"),
errors.New("Error 1265: Data truncated for column 'daily_mileage_km'"),
errors.New("Error 1292: Incorrect datetime value"),
errors.New("Error 1366: Incorrect decimal value"),
errors.New("Error 1406: Data too long for column"),
errors.New("Error 3819: Check constraint is violated"),
} {
if !isQuarantinableStatError(err) {
t.Fatalf("isQuarantinableStatError(%q) = false, want true", err.Error())
}
}
for _, err := range []error{
errors.New("Error 1054: Unknown column 'daily_mileage_km'"),
errors.New("Error 1142: INSERT command denied"),
errors.New("validation failed"),
context.Canceled,
context.DeadlineExceeded,
nil,
} {
if isQuarantinableStatError(err) {
t.Fatalf("isQuarantinableStatError(%v) = true, want false", err)
}
}
}
func TestIsTransientMySQLStatError(t *testing.T) {
for _, err := range []error{
errors.New("dial tcp 127.0.0.1:3306: connection refused"),
@@ -903,6 +1023,9 @@ func TestLoadConfigDefaultsToGoFieldsTopics(t *testing.T) {
if cfg.BaselineMissTTL != time.Minute {
t.Fatalf("BaselineMissTTL = %s, want 1m", cfg.BaselineMissTTL)
}
if cfg.BaselineHitTTL != 5*time.Minute {
t.Fatalf("BaselineHitTTL = %s, want 5m", cfg.BaselineHitTTL)
}
if cfg.RetryAttempts != 3 {
t.Fatalf("RetryAttempts = %d, want 3", cfg.RetryAttempts)
}
@@ -965,6 +1088,7 @@ func TestLoadConfigSetsStatWriteIntervals(t *testing.T) {
t.Setenv("STATS_CACHE_RETENTION_HOURS", "48")
t.Setenv("STATS_CACHE_CLEANUP_INTERVAL_SECONDS", "300")
t.Setenv("STATS_BASELINE_MISS_TTL_SECONDS", "45")
t.Setenv("STATS_BASELINE_HIT_TTL_SECONDS", "180")
t.Setenv("STATS_CACHE_MAX_ENTRIES", "12345")
cfg := loadConfig()
@@ -984,6 +1108,9 @@ func TestLoadConfigSetsStatWriteIntervals(t *testing.T) {
if cfg.BaselineMissTTL != 45*time.Second {
t.Fatalf("BaselineMissTTL = %s, want 45s", cfg.BaselineMissTTL)
}
if cfg.BaselineHitTTL != 3*time.Minute {
t.Fatalf("BaselineHitTTL = %s, want 3m", cfg.BaselineHitTTL)
}
if cfg.CacheMaxEntries != 12345 {
t.Fatalf("CacheMaxEntries = %d, want 12345", cfg.CacheMaxEntries)
}
@@ -997,6 +1124,7 @@ func TestLoadConfigReadsStatBatchSettings(t *testing.T) {
t.Setenv("STATS_RETRY_DELAY_MS", "33")
t.Setenv("STATS_NORMALIZE_PLATFORM_SOURCES_ON_START", "false")
t.Setenv("STATS_NORMALIZE_PLATFORM_SOURCES_TIMEOUT_SECONDS", "17")
t.Setenv("STATS_QUARANTINE_DIR", "/tmp/stat-writer-quarantine-test")
cfg := loadConfig()
@@ -1021,6 +1149,9 @@ func TestLoadConfigReadsStatBatchSettings(t *testing.T) {
if cfg.NormalizePlatformSourcesTimeout != 17*time.Second {
t.Fatalf("NormalizePlatformSourcesTimeout = %s, want 17s", cfg.NormalizePlatformSourcesTimeout)
}
if cfg.QuarantineDir != "/tmp/stat-writer-quarantine-test" {
t.Fatalf("QuarantineDir = %q, want env override", cfg.QuarantineDir)
}
}
type contextCheckingStatAppender struct {
@@ -1070,6 +1201,18 @@ type contextCheckingStatCommitter struct {
failOnCount int
}
type recordingStatQuarantiner struct {
count int
message kafka.Message
err error
}
func (q *recordingStatQuarantiner) Quarantine(_ context.Context, message kafka.Message, _ error) error {
q.count++
q.message = message
return q.err
}
func (c *contextCheckingStatCommitter) CommitMessages(ctx context.Context, messages ...kafka.Message) error {
c.ctxErr = ctx.Err()
c.count++