From 30a93a868e74c90efb8e2cbc1e4c4aed29e87376 Mon Sep 17 00:00:00 2001 From: David Rinaldo Date: Sat, 4 Jul 2026 17:40:50 -0400 Subject: [PATCH] ingotctl, HTTP layer, and self-instrumentation Add cmd/ingotctl with four subcommands: blocks (list with stats), inspect (series/postings dump), chunks (decode raw samples by ref), and fsck (CRC and index integrity validation across all blocks). Add cmd/ingothttp with JSON query endpoints: /api/v1/query_range (Prometheus-style matrix response), /api/v1/read (matcher-based read requests), and /epi/v1/status (DB stats snapshot). Uses JSON instead of protobuf to maintain zero dependencies. Add self-instrumentation via the normal Appender path so metrics are queryable with the same API: ingot_head_series, ingot_head_chunks_active, ingot_blocks_total, ingot_compactions_total, ingot_wal_fsync_duration_seconds. Supporting changes: block.Validate() and block.ReadMeta() exports, Head.Stats() for series/chunk counts, WAL.LastSyncDuration() with timed fsync tracking, DB.Stats() for the HTTP status endpoint. --- cmd/ingotctl/main.go | 357 ++++++++++++++++++++++++++++++++ cmd/ingotctl/main_test.go | 278 +++++++++++++++++++++++++ cmd/ingothttp/handler.go | 278 +++++++++++++++++++++++++ cmd/ingothttp/handler_test.go | 279 +++++++++++++++++++++++++ cmd/ingothttp/main.go | 58 ++++++ ingot.go | 28 +++ internal/block/validate.go | 148 +++++++++++++ internal/block/validate_test.go | 191 +++++++++++++++++ internal/head/head.go | 26 +++ internal/wal/wal.go | 21 +- metrics.go | 68 ++++++ metrics_test.go | 148 +++++++++++++ 12 files changed, 1878 insertions(+), 2 deletions(-) create mode 100644 cmd/ingotctl/main.go create mode 100644 cmd/ingotctl/main_test.go create mode 100644 cmd/ingothttp/handler.go create mode 100644 cmd/ingothttp/handler_test.go create mode 100644 cmd/ingothttp/main.go create mode 100644 internal/block/validate.go create mode 100644 internal/block/validate_test.go create mode 100644 metrics.go create mode 100644 metrics_test.go diff --git a/cmd/ingotctl/main.go b/cmd/ingotctl/main.go new file mode 100644 index 0000000..7a61ed4 --- /dev/null +++ b/cmd/ingotctl/main.go @@ -0,0 +1,357 @@ +// Command ingotctl provides CLI tools for inspecting and validating ingot blocks. +// +// Usage: +// +// ingotctl blocks — list blocks with metadata +// ingotctl inspect — detailed block dump +// ingotctl chunks — decode and print raw samples +// ingotctl fsck — validate all blocks +package main + +import ( + "encoding/json" + "fmt" + "os" + "path/filepath" + "sort" + "strconv" + "strings" + "text/tabwriter" + "time" + + "git.dvdt.dev/david/ingot/internal/block" + "git.dvdt.dev/david/ingot/internal/index" + "git.dvdt.dev/david/ingot/labels" +) + +func main() { + if len(os.Args) < 2 { + usage() + os.Exit(1) + } + + var err error + switch os.Args[1] { + case "blocks": + err = cmdBlocks(os.Args[2:]) + case "inspect": + err = cmdInspect(os.Args[2:]) + case "chunks": + err = cmdChunks(os.Args[2:]) + case "fsck": + err = cmdFsck(os.Args[2:]) + case "help", "-h", "--help": + usage() + default: + fmt.Fprintf(os.Stderr, "unknown command: %s\n\n", os.Args[1]) + usage() + os.Exit(1) + } + + if err != nil { + fmt.Fprintf(os.Stderr, "error: %v\n", err) + os.Exit(1) + } +} + +func usage() { + fmt.Fprintln(os.Stderr, `Usage: ingotctl [args] + +Commands: + blocks List blocks with ULID, time range, level, stats + inspect Detailed block dump: series, chunks, postings + chunks Decode and print raw samples for a series ref + fsck Validate all blocks: CRC checks, index integrity`) +} + +// cmdBlocks lists all blocks in a data directory. +func cmdBlocks(args []string) error { + if len(args) < 1 { + return fmt.Errorf("usage: ingotctl blocks ") + } + dataDir := args[0] + + entries, err := os.ReadDir(dataDir) + if err != nil { + return err + } + + type blockInfo struct { + meta block.BlockMeta + dir string + } + var blocks []blockInfo + + for _, e := range entries { + if !e.IsDir() || e.Name() == "wal" { + continue + } + dir := filepath.Join(dataDir, e.Name()) + if _, err := os.Stat(filepath.Join(dir, "meta.json")); err != nil { + continue + } + meta, err := block.ReadMeta(dir) + if err != nil { + fmt.Fprintf(os.Stderr, "warning: %s: %v\n", e.Name(), err) + continue + } + blocks = append(blocks, blockInfo{meta: meta, dir: dir}) + } + + sort.Slice(blocks, func(i, j int) bool { + return blocks[i].meta.MinTime < blocks[j].meta.MinTime + }) + + if len(blocks) == 0 { + fmt.Println("no blocks found") + return nil + } + + w := tabwriter.NewWriter(os.Stdout, 0, 4, 2, ' ', 0) + fmt.Fprintln(w, "ULID\tMIN TIME\tMAX TIME\tDURATION\tLEVEL\tSERIES\tSAMPLES\tCHUNKS") + for _, b := range blocks { + m := b.meta + dur := time.Duration(m.MaxTime-m.MinTime) * time.Millisecond + fmt.Fprintf(w, "%s\t%s\t%s\t%s\t%d\t%d\t%d\t%d\n", + m.ULID, + formatTimestamp(m.MinTime), + formatTimestamp(m.MaxTime), + dur.Truncate(time.Second), + m.Compaction.Level, + m.Stats.NumSeries, + m.Stats.NumSamples, + m.Stats.NumChunks, + ) + } + w.Flush() + + fmt.Printf("\n%d block(s) total\n", len(blocks)) + return nil +} + +// cmdInspect dumps detailed information about a single block. +func cmdInspect(args []string) error { + if len(args) < 1 { + return fmt.Errorf("usage: ingotctl inspect ") + } + blockDir := args[0] + + meta, err := block.ReadMeta(blockDir) + if err != nil { + return fmt.Errorf("read meta: %w", err) + } + + // Print meta. + fmt.Println("=== Block Meta ===") + metaJSON, _ := json.MarshalIndent(meta, "", " ") + fmt.Println(string(metaJSON)) + + // Open block for index inspection. + br, err := block.Open(blockDir) + if err != nil { + return fmt.Errorf("open block: %w", err) + } + defer br.Close() + + series := br.Series() + + fmt.Printf("\n=== Series (%d) ===\n", len(series)) + w := tabwriter.NewWriter(os.Stdout, 0, 4, 2, ' ', 0) + fmt.Fprintln(w, "REF\tLABELS\tCHUNKS\tMIN TIME\tMAX TIME") + for _, s := range series { + minT, maxT := seriesTimeRange(s.Chunks) + fmt.Fprintf(w, "%d\t%s\t%d\t%s\t%s\n", + s.Ref, + formatLabels(s.Labels), + len(s.Chunks), + formatTimestamp(minT), + formatTimestamp(maxT), + ) + } + w.Flush() + + // Postings stats. + fmt.Printf("\n=== Postings ===\n") + postingsStats := collectPostingsStats(br, series) + pw := tabwriter.NewWriter(os.Stdout, 0, 4, 2, ' ', 0) + fmt.Fprintln(pw, "LABEL\tVALUES\tPOSTINGS") + for _, ps := range postingsStats { + fmt.Fprintf(pw, "%s\t%d\t%d\n", ps.name, ps.numValues, ps.totalPostings) + } + pw.Flush() + + return nil +} + +// cmdChunks decodes and prints raw samples for a series ref in a block. +func cmdChunks(args []string) error { + if len(args) < 2 { + return fmt.Errorf("usage: ingotctl chunks ") + } + blockDir := args[0] + ref, err := strconv.ParseUint(args[1], 10, 64) + if err != nil { + return fmt.Errorf("invalid series ref %q: %w", args[1], err) + } + + br, err := block.Open(blockDir) + if err != nil { + return fmt.Errorf("open block: %w", err) + } + defer br.Close() + + entry, ok := br.SeriesByRef(ref) + if !ok { + return fmt.Errorf("series ref %d not found in block", ref) + } + + fmt.Printf("Series %d: %s\n", ref, formatLabels(entry.Labels)) + fmt.Printf("Chunks: %d\n\n", len(entry.Chunks)) + + for i, cm := range entry.Chunks { + fmt.Printf("--- Chunk %d [%s .. %s] segment=%d offset=%d ---\n", + i, formatTimestamp(cm.MinT), formatTimestamp(cm.MaxT), + cm.Ref.Segment(), cm.Ref.Offset()) + + it, err := br.ChunkIterator(cm.Ref) + if err != nil { + fmt.Fprintf(os.Stderr, " error reading chunk: %v\n", err) + continue + } + + w := tabwriter.NewWriter(os.Stdout, 0, 4, 2, ' ', 0) + fmt.Fprintln(w, " TIMESTAMP\tVALUE") + count := 0 + for it.Next() { + t, v := it.At() + fmt.Fprintf(w, " %s\t%g\n", formatTimestamp(t), v) + count++ + } + w.Flush() + if it.Err() != nil { + fmt.Fprintf(os.Stderr, " iterator error: %v\n", it.Err()) + } + fmt.Printf(" (%d samples)\n\n", count) + } + + return nil +} + +// cmdFsck validates all blocks in a data directory. +func cmdFsck(args []string) error { + if len(args) < 1 { + return fmt.Errorf("usage: ingotctl fsck ") + } + dataDir := args[0] + + entries, err := os.ReadDir(dataDir) + if err != nil { + return err + } + + var blockDirs []string + for _, e := range entries { + if !e.IsDir() || e.Name() == "wal" { + continue + } + dir := filepath.Join(dataDir, e.Name()) + if _, err := os.Stat(filepath.Join(dir, "meta.json")); err != nil { + continue + } + blockDirs = append(blockDirs, dir) + } + + if len(blockDirs) == 0 { + fmt.Println("no blocks found") + return nil + } + + totalErrors := 0 + for _, dir := range blockDirs { + name := filepath.Base(dir) + errs := block.Validate(dir) + if len(errs) == 0 { + fmt.Printf("%s: ok\n", name) + } else { + for _, e := range errs { + fmt.Printf("%s\n", e.Error()) + totalErrors++ + } + } + } + + fmt.Printf("\n%d block(s) checked, %d error(s)\n", len(blockDirs), totalErrors) + if totalErrors > 0 { + return fmt.Errorf("%d integrity error(s) found", totalErrors) + } + return nil +} + +// --- helpers --- + +func formatTimestamp(ms int64) string { + t := time.UnixMilli(ms) + return t.UTC().Format("2006-01-02T15:04:05Z") +} + +func formatLabels(ls []labels.Label) string { + var parts []string + for _, l := range ls { + parts = append(parts, l.Name+"="+strconv.Quote(l.Value)) + } + return "{" + strings.Join(parts, ", ") + "}" +} + +func seriesTimeRange(chunks []index.ChunkMeta) (int64, int64) { + if len(chunks) == 0 { + return 0, 0 + } + minT := chunks[0].MinT + maxT := chunks[0].MaxT + for _, c := range chunks[1:] { + if c.MinT < minT { + minT = c.MinT + } + if c.MaxT > maxT { + maxT = c.MaxT + } + } + return minT, maxT +} + +type postingsStat struct { + name string + numValues int + totalPostings int +} + +func collectPostingsStats(br *block.Reader, series []index.SeriesEntry) []postingsStat { + // Collect all label names. + nameSet := make(map[string]struct{}) + for _, s := range series { + for _, l := range s.Labels { + nameSet[l.Name] = struct{}{} + } + } + + names := make([]string, 0, len(nameSet)) + for n := range nameSet { + names = append(names, n) + } + sort.Strings(names) + + var stats []postingsStat + for _, name := range names { + values := br.LabelValues(name) + total := 0 + for _, v := range values { + total += len(br.Postings(name, v)) + } + stats = append(stats, postingsStat{ + name: name, + numValues: len(values), + totalPostings: total, + }) + } + return stats +} diff --git a/cmd/ingotctl/main_test.go b/cmd/ingotctl/main_test.go new file mode 100644 index 0000000..03bd6f9 --- /dev/null +++ b/cmd/ingotctl/main_test.go @@ -0,0 +1,278 @@ +package main + +import ( + "os" + "path/filepath" + "strings" + "testing" + + "git.dvdt.dev/david/ingot/internal/block" + "git.dvdt.dev/david/ingot/internal/chunkenc" + "git.dvdt.dev/david/ingot/labels" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func makeChunk(t *testing.T, samples []struct{ t int64; v float64 }) []byte { + t.Helper() + c := chunkenc.NewXORChunk() + a, err := c.Appender() + require.NoError(t, err) + for _, s := range samples { + a.Append(s.t, s.v) + } + return append([]byte(nil), c.Bytes()...) +} + +func setupTestData(t *testing.T) string { + t.Helper() + dir := t.TempDir() + + samples := []struct{ t int64; v float64 }{ + {0, 1.0}, {15000, 2.0}, {30000, 3.0}, + } + chunk := makeChunk(t, samples) + + series := []block.SeriesFlush{ + { + Ref: 1, + Labels: labels.FromStrings("__name__", "temp", "room", "office"), + Chunks: []block.ChunkData{ + {MinT: 0, MaxT: 30000, Data: chunk}, + }, + }, + { + Ref: 2, + Labels: labels.FromStrings("__name__", "humidity", "room", "office"), + Chunks: []block.ChunkData{ + {MinT: 0, MaxT: 30000, Data: chunk}, + }, + }, + } + + _, err := block.Flush(dir, series) + require.NoError(t, err) + return dir +} + +// blockDir returns the first block directory inside dataDir. +func blockDir(t *testing.T, dataDir string) string { + t.Helper() + entries, err := os.ReadDir(dataDir) + require.NoError(t, err) + for _, e := range entries { + if e.IsDir() && e.Name() != "wal" { + return filepath.Join(dataDir, e.Name()) + } + } + t.Fatal("no block directory found") + return "" +} + +// captureStdout calls fn with stdout redirected and returns the output. +func captureStdout(fn func()) string { + old := os.Stdout + r, w, _ := os.Pipe() + os.Stdout = w + fn() + w.Close() + os.Stdout = old + out := make([]byte, 16384) + n, _ := r.Read(out) + return string(out[:n]) +} + +func TestCmdBlocks(t *testing.T) { + tests := []struct { + name string + args []string + wantErr string // "" = no error expected + wantOutputs []string + }{ + { + name: "list_blocks", + args: nil, // replaced in loop with setupTestData result + wantErr: "", + wantOutputs: []string{"ULID", "1 block(s) total"}, + }, + { + name: "no_blocks", + args: nil, // replaced with empty TempDir + wantErr: "", + wantOutputs: []string{"no blocks found"}, + }, + { + name: "missing_args", + args: []string{}, + wantErr: "usage", + wantOutputs: nil, + }, + } + + // Set up args that need dynamic dirs. + dataDir := setupTestData(t) + emptyDir := t.TempDir() + tests[0].args = []string{dataDir} + tests[1].args = []string{emptyDir} + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + var output string + err := func() error { + var cmdErr error + output = captureStdout(func() { cmdErr = cmdBlocks(tc.args) }) + return cmdErr + }() + + assert.Equal(t, tc.wantErr != "", err != nil, "error presence") + assert.Contains(t, errString(err), tc.wantErr) + for _, want := range tc.wantOutputs { + assert.Contains(t, output, want) + } + }) + } +} + +func TestCmdInspect(t *testing.T) { + dataDir := setupTestData(t) + bd := blockDir(t, dataDir) + + tests := []struct { + name string + args []string + wantErr string + wantOutputs []string + }{ + { + name: "valid_block", + args: []string{bd}, + wantErr: "", + wantOutputs: []string{"Block Meta", "__name__", "Postings"}, + }, + { + name: "missing_args", + args: []string{}, + wantErr: "usage", + wantOutputs: nil, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + var output string + err := func() error { + var cmdErr error + output = captureStdout(func() { cmdErr = cmdInspect(tc.args) }) + return cmdErr + }() + + assert.Equal(t, tc.wantErr != "", err != nil, "error presence") + assert.Contains(t, errString(err), tc.wantErr) + for _, want := range tc.wantOutputs { + assert.Contains(t, output, want) + } + }) + } +} + +func TestCmdChunks(t *testing.T) { + dataDir := setupTestData(t) + bd := blockDir(t, dataDir) + + tests := []struct { + name string + args []string + wantErr string + wantOutputs []string + }{ + { + name: "valid_ref", + args: []string{bd, "1"}, + wantErr: "", + wantOutputs: []string{"Series 1", "(3 samples)"}, + }, + { + name: "ref_not_found", + args: []string{bd, "999"}, + wantErr: "not found", + wantOutputs: nil, + }, + { + name: "missing_ref_arg", + args: []string{bd}, + wantErr: "usage", + wantOutputs: nil, + }, + { + name: "missing_all_args", + args: []string{}, + wantErr: "usage", + wantOutputs: nil, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + var output string + err := func() error { + var cmdErr error + output = captureStdout(func() { cmdErr = cmdChunks(tc.args) }) + return cmdErr + }() + + assert.Equal(t, tc.wantErr != "", err != nil, "error presence") + assert.Contains(t, errString(err), tc.wantErr) + for _, want := range tc.wantOutputs { + assert.Contains(t, output, want) + } + }) + } +} + +func TestCmdFsck(t *testing.T) { + dataDir := setupTestData(t) + emptyDir := t.TempDir() + + tests := []struct { + name string + args []string + wantErr string + }{ + { + name: "valid_data", + args: []string{dataDir}, + wantErr: "", + }, + { + name: "no_blocks", + args: []string{emptyDir}, + wantErr: "", + }, + { + name: "missing_args", + args: []string{}, + wantErr: "usage", + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + err := func() error { + var cmdErr error + captureStdout(func() { cmdErr = cmdFsck(tc.args) }) + return cmdErr + }() + + assert.Equal(t, tc.wantErr != "", err != nil, "error presence") + assert.Contains(t, errString(err), tc.wantErr) + }) + } +} + +// errString returns the error message or "" for nil. +func errString(err error) string { + if err == nil { + return "" + } + return strings.ToLower(err.Error()) +} diff --git a/cmd/ingothttp/handler.go b/cmd/ingothttp/handler.go new file mode 100644 index 0000000..2cb2404 --- /dev/null +++ b/cmd/ingothttp/handler.go @@ -0,0 +1,278 @@ +package main + +import ( + "encoding/json" + "fmt" + "math" + "net/http" + "strconv" + + "git.dvdt.dev/david/ingot" + "git.dvdt.dev/david/ingot/labels" +) + +type handler struct { + db *ingot.DB +} + +func newHandler(db *ingot.DB) *handler { + return &handler{db: db} +} + +// --- /api/v1/query_range --- + +// queryRange handles GET /api/v1/query_range?query=&start=&end= +// Returns Prometheus-style JSON matrix results. +func (h *handler) queryRange(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + httpError(w, http.StatusMethodNotAllowed, "method not allowed") + return + } + + query := r.URL.Query().Get("query") + if query == "" { + httpError(w, http.StatusBadRequest, "missing query parameter") + return + } + + startStr := r.URL.Query().Get("start") + endStr := r.URL.Query().Get("end") + + mint := int64(math.MinInt64) + maxt := int64(math.MaxInt64) + + if startStr != "" { + v, err := strconv.ParseInt(startStr, 10, 64) + if err != nil { + httpError(w, http.StatusBadRequest, "invalid start: "+err.Error()) + return + } + mint = v + } + if endStr != "" { + v, err := strconv.ParseInt(endStr, 10, 64) + if err != nil { + httpError(w, http.StatusBadRequest, "invalid end: "+err.Error()) + return + } + maxt = v + } + + matcher := labels.MustNewMatcher(labels.MatchEqual, "__name__", query) + result, err := h.executeQuery(mint, maxt, []*labels.Matcher{matcher}) + if err != nil { + httpError(w, http.StatusInternalServerError, err.Error()) + return + } + + resp := queryRangeResponse{ + Status: "success", + Data: queryRangeData{ + ResultType: "matrix", + Result: result, + }, + } + + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(resp) +} + +// --- /api/v1/read --- + +// ReadRequest is a JSON-encoded query for the read endpoint. +type ReadRequest struct { + Queries []ReadQuery `json:"queries"` +} + +// ReadQuery describes a single read query. +type ReadQuery struct { + StartTimestampMs int64 `json:"startTimestampMs"` + EndTimestampMs int64 `json:"endTimestampMs"` + Matchers []ReadMatcher `json:"matchers"` +} + +// ReadMatcher is a label matcher in a read request. +type ReadMatcher struct { + Type string `json:"type"` // "=", "!=", "=~", "!~" + Name string `json:"name"` + Value string `json:"value"` +} + +// ReadResponse is the JSON response for the read endpoint. +type ReadResponse struct { + Results []ReadResult `json:"results"` +} + +// ReadResult contains the timeseries for a single query. +type ReadResult struct { + Timeseries []timeseriesResult `json:"timeseries"` +} + +// read handles POST /api/v1/read with a JSON ReadRequest body. +func (h *handler) read(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + httpError(w, http.StatusMethodNotAllowed, "method not allowed") + return + } + + var req ReadRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + httpError(w, http.StatusBadRequest, "invalid request body: "+err.Error()) + return + } + + resp := ReadResponse{ + Results: make([]ReadResult, len(req.Queries)), + } + + for i, rq := range req.Queries { + matchers, err := convertMatchers(rq.Matchers) + if err != nil { + httpError(w, http.StatusBadRequest, err.Error()) + return + } + + result, err := h.executeQuery(rq.StartTimestampMs, rq.EndTimestampMs, matchers) + if err != nil { + httpError(w, http.StatusInternalServerError, err.Error()) + return + } + + ts := make([]timeseriesResult, len(result)) + for j, sr := range result { + ts[j] = timeseriesResult{ + Labels: sr.Metric, + Samples: sr.Values, + } + } + resp.Results[i] = ReadResult{Timeseries: ts} + } + + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(resp) +} + +// --- /api/v1/status --- + +type statusResponse struct { + HeadSeries int `json:"headSeries"` + HeadChunks int `json:"headChunks"` + Blocks int `json:"blocks"` + Compactions int `json:"compactions"` +} + +func (h *handler) status(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + httpError(w, http.StatusMethodNotAllowed, "method not allowed") + return + } + + stats := h.db.Stats() + resp := statusResponse{ + HeadSeries: stats.HeadSeries, + HeadChunks: stats.HeadChunks, + Blocks: stats.Blocks, + Compactions: stats.Compactions, + } + + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(resp) +} + +// --- shared helpers --- + +type queryRangeResponse struct { + Status string `json:"status"` + Data queryRangeData `json:"data"` +} + +type queryRangeData struct { + ResultType string `json:"resultType"` + Result []seriesResult `json:"result"` +} + +type seriesResult struct { + Metric map[string]string `json:"metric"` + Values [][2]interface{} `json:"values"` // [timestamp_ms, "value"] +} + +type timeseriesResult struct { + Labels map[string]string `json:"labels"` + Samples [][2]interface{} `json:"samples"` +} + +func (h *handler) executeQuery(mint, maxt int64, matchers []*labels.Matcher) ([]seriesResult, error) { + q, err := h.db.Querier(mint, maxt) + if err != nil { + return nil, err + } + defer q.Close() + + ss := q.Select(matchers...) + var results []seriesResult + + for ss.Next() { + s := ss.At() + ls := s.Labels() + metric := make(map[string]string, len(ls)) + for _, l := range ls { + metric[l.Name] = l.Value + } + + var values [][2]interface{} + it := s.Iterator() + for it.Next() { + t, v := it.At() + values = append(values, [2]interface{}{t, fmt.Sprintf("%g", v)}) + } + if it.Err() != nil { + return nil, it.Err() + } + + if len(values) > 0 { + results = append(results, seriesResult{ + Metric: metric, + Values: values, + }) + } + } + if ss.Err() != nil { + return nil, ss.Err() + } + + return results, nil +} + +func convertMatchers(rms []ReadMatcher) ([]*labels.Matcher, error) { + matchers := make([]*labels.Matcher, len(rms)) + for i, rm := range rms { + var matchType labels.MatchType + switch rm.Type { + case "=": + matchType = labels.MatchEqual + case "!=": + matchType = labels.MatchNotEqual + case "=~": + matchType = labels.MatchRegexp + case "!~": + matchType = labels.MatchNotRegexp + default: + return nil, fmt.Errorf("unsupported matcher type: %q", rm.Type) + } + m, err := labels.NewMatcher(matchType, rm.Name, rm.Value) + if err != nil { + return nil, fmt.Errorf("invalid matcher: %w", err) + } + matchers[i] = m + } + return matchers, nil +} + +func httpError(w http.ResponseWriter, code int, msg string) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(code) + json.NewEncoder(w).Encode(map[string]string{ + "status": "error", + "error": msg, + }) +} diff --git a/cmd/ingothttp/handler_test.go b/cmd/ingothttp/handler_test.go new file mode 100644 index 0000000..cb593ac --- /dev/null +++ b/cmd/ingothttp/handler_test.go @@ -0,0 +1,279 @@ +package main + +import ( + "bytes" + "encoding/json" + "math" + "net/http" + "net/http/httptest" + "testing" + + "git.dvdt.dev/david/ingot" + "git.dvdt.dev/david/ingot/labels" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func openTestDB(t *testing.T) *ingot.DB { + t.Helper() + db, err := ingot.Open(t.TempDir(), ingot.Options{}) + require.NoError(t, err) + t.Cleanup(func() { db.Close() }) + return db +} + +func seedDB(t *testing.T, db *ingot.DB) { + t.Helper() + app := db.Appender() + ref, err := app.Append(0, labels.FromStrings("__name__", "temp", "room", "office"), 1000, 71.3) + require.NoError(t, err) + _, err = app.Append(ref, nil, 2000, 71.4) + require.NoError(t, err) + _, err = app.Append(0, labels.FromStrings("__name__", "humidity", "room", "office"), 1000, 55.0) + require.NoError(t, err) + require.NoError(t, app.Commit()) +} + +func TestQueryRange(t *testing.T) { + tests := []struct { + name string + method string + query string + start string + end string + wantStatus int + wantCount int // number of result series; 0 for error responses + }{ + { + name: "match_temp", + method: http.MethodGet, + query: "temp", + wantStatus: http.StatusOK, + wantCount: 1, + }, + { + name: "match_humidity", + method: http.MethodGet, + query: "humidity", + wantStatus: http.StatusOK, + wantCount: 1, + }, + { + name: "no_match", + method: http.MethodGet, + query: "pressure", + wantStatus: http.StatusOK, + wantCount: 0, + }, + { + name: "missing_query", + method: http.MethodGet, + query: "", + wantStatus: http.StatusBadRequest, + wantCount: 0, + }, + { + name: "with_time_range", + method: http.MethodGet, + query: "temp", + start: "1500", + end: "3000", + wantStatus: http.StatusOK, + wantCount: 1, + }, + { + name: "time_range_miss", + method: http.MethodGet, + query: "temp", + start: "5000", + end: "6000", + wantStatus: http.StatusOK, + wantCount: 0, + }, + { + name: "method_not_allowed", + method: http.MethodPost, + query: "temp", + wantStatus: http.StatusMethodNotAllowed, + wantCount: 0, + }, + } + + db := openTestDB(t) + seedDB(t, db) + h := newHandler(db) + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + url := "/api/v1/query_range?query=" + tc.query + if tc.start != "" { + url += "&start=" + tc.start + } + if tc.end != "" { + url += "&end=" + tc.end + } + + req := httptest.NewRequest(tc.method, url, nil) + rec := httptest.NewRecorder() + h.queryRange(rec, req) + + assert.Equal(t, tc.wantStatus, rec.Code) + + // Always decode — error responses produce zero-valued struct. + var resp queryRangeResponse + json.Unmarshal(rec.Body.Bytes(), &resp) + assert.Equal(t, tc.wantCount, len(resp.Data.Result)) + }) + } +} + +func TestRead(t *testing.T) { + tests := []struct { + name string + method string + request ReadRequest + wantStatus int + wantSeries int // in first result; 0 for error responses + }{ + { + name: "single_query", + method: http.MethodPost, + request: ReadRequest{ + Queries: []ReadQuery{ + { + StartTimestampMs: math.MinInt64, + EndTimestampMs: math.MaxInt64, + Matchers: []ReadMatcher{ + {Type: "=", Name: "__name__", Value: "temp"}, + }, + }, + }, + }, + wantStatus: http.StatusOK, + wantSeries: 1, + }, + { + name: "regex_matcher", + method: http.MethodPost, + request: ReadRequest{ + Queries: []ReadQuery{ + { + StartTimestampMs: math.MinInt64, + EndTimestampMs: math.MaxInt64, + Matchers: []ReadMatcher{ + {Type: "=~", Name: "__name__", Value: "temp|humidity"}, + }, + }, + }, + }, + wantStatus: http.StatusOK, + wantSeries: 2, + }, + { + name: "invalid_matcher_type", + method: http.MethodPost, + request: ReadRequest{ + Queries: []ReadQuery{ + { + Matchers: []ReadMatcher{ + {Type: "??", Name: "a", Value: "b"}, + }, + }, + }, + }, + wantStatus: http.StatusBadRequest, + wantSeries: 0, + }, + { + name: "method_not_allowed", + method: http.MethodGet, + request: ReadRequest{}, + wantStatus: http.StatusMethodNotAllowed, + wantSeries: 0, + }, + } + + db := openTestDB(t) + seedDB(t, db) + h := newHandler(db) + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + body, err := json.Marshal(tc.request) + require.NoError(t, err) + + req := httptest.NewRequest(tc.method, "/api/v1/read", bytes.NewReader(body)) + rec := httptest.NewRecorder() + h.read(rec, req) + + assert.Equal(t, tc.wantStatus, rec.Code) + + // Always decode — error responses produce zero-valued struct. + var resp ReadResponse + json.Unmarshal(rec.Body.Bytes(), &resp) + firstResultLen := 0 + if len(resp.Results) > 0 { + firstResultLen = len(resp.Results[0].Timeseries) + } + assert.Equal(t, tc.wantSeries, firstResultLen) + }) + } +} + +func TestStatus(t *testing.T) { + tests := []struct { + name string + method string + seed bool + wantStatus int + wantHeadSeries int + wantBlocks int + }{ + { + name: "seeded_db", + method: http.MethodGet, + seed: true, + wantStatus: http.StatusOK, + wantHeadSeries: 2, + wantBlocks: 0, + }, + { + name: "empty_db", + method: http.MethodGet, + seed: false, + wantStatus: http.StatusOK, + wantHeadSeries: 0, + wantBlocks: 0, + }, + { + name: "method_not_allowed", + method: http.MethodPost, + seed: false, + wantStatus: http.StatusMethodNotAllowed, + wantHeadSeries: 0, + wantBlocks: 0, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + db := openTestDB(t) + if tc.seed { + seedDB(t, db) + } + h := newHandler(db) + + req := httptest.NewRequest(tc.method, "/api/v1/status", nil) + rec := httptest.NewRecorder() + h.status(rec, req) + + assert.Equal(t, tc.wantStatus, rec.Code) + + // Always decode — error responses produce zero-valued struct. + var resp statusResponse + json.Unmarshal(rec.Body.Bytes(), &resp) + assert.Equal(t, tc.wantHeadSeries, resp.HeadSeries) + assert.Equal(t, tc.wantBlocks, resp.Blocks) + }) + } +} diff --git a/cmd/ingothttp/main.go b/cmd/ingothttp/main.go new file mode 100644 index 0000000..ae2c2d4 --- /dev/null +++ b/cmd/ingothttp/main.go @@ -0,0 +1,58 @@ +// Command ingothttp serves a minimal HTTP query API for an ingot database. +// +// Endpoints: +// +// GET /api/v1/query_range?query={name}&start={unix_ms}&end={unix_ms} +// POST /api/v1/read (JSON ReadRequest body) +// GET /api/v1/status (DB summary stats) +// +// This is a demo/bridge for Grafana integration, not a full PromQL engine. +// The query parameter is a metric name (matched as __name__=). +package main + +import ( + "flag" + "fmt" + "log" + "net/http" + "os" + "os/signal" + "syscall" + + "git.dvdt.dev/david/ingot" +) + +func main() { + var ( + dataDir = flag.String("data", "./data", "ingot data directory") + addr = flag.String("addr", ":9001", "HTTP listen address") + ) + flag.Parse() + + db, err := ingot.Open(*dataDir, ingot.Options{}) + if err != nil { + log.Fatalf("failed to open db: %v", err) + } + defer db.Close() + + h := newHandler(db) + mux := http.NewServeMux() + mux.HandleFunc("/api/v1/query_range", h.queryRange) + mux.HandleFunc("/api/v1/read", h.read) + mux.HandleFunc("/api/v1/status", h.status) + + srv := &http.Server{Addr: *addr, Handler: mux} + + go func() { + sig := make(chan os.Signal, 1) + signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM) + <-sig + fmt.Fprintln(os.Stderr, "\nshutting down...") + srv.Close() + }() + + log.Printf("ingothttp listening on %s (data: %s)", *addr, *dataDir) + if err := srv.ListenAndServe(); err != http.ErrServerClosed { + log.Fatal(err) + } +} diff --git a/ingot.go b/ingot.go index 4beb9f0..71bf2fd 100644 --- a/ingot.go +++ b/ingot.go @@ -8,6 +8,7 @@ import ( "path/filepath" "sort" "sync" + "sync/atomic" "time" "git.dvdt.dev/david/ingot/internal/block" @@ -37,6 +38,9 @@ type DB struct { compactCtx context.Context compactCancel context.CancelFunc compactWg sync.WaitGroup + + compactionCount atomic.Int64 // incremented on each successful compaction + metricsR metricsRefs // cached series refs for self-instrumentation } // Options configures a DB. @@ -237,6 +241,7 @@ func (db *DB) RunCompaction() error { } } + db.compactionCount.Add(1) return nil } @@ -292,6 +297,7 @@ func (db *DB) compactLoop() { case <-db.compactCtx.Done(): return case <-ticker.C: + db.collectMetrics() db.autoFlush() db.RunCompaction() db.ApplyRetention() @@ -306,6 +312,28 @@ func (db *DB) autoFlush() { db.FlushOlderThan(cutoff) } +// DBStats holds summary statistics for the database. +type DBStats struct { + HeadSeries int + HeadChunks int + Blocks int + Compactions int +} + +// Stats returns a snapshot of database statistics. +func (db *DB) Stats() DBStats { + hs := db.head.Stats() + db.mu.RLock() + numBlocks := len(db.blocks) + db.mu.RUnlock() + return DBStats{ + HeadSeries: hs.NumSeries, + HeadChunks: hs.NumActiveChunks, + Blocks: numBlocks, + Compactions: int(db.compactionCount.Load()), + } +} + // Close closes the DB, releasing all resources. func (db *DB) Close() error { db.compactCancel() diff --git a/internal/block/validate.go b/internal/block/validate.go new file mode 100644 index 0000000..cb88f08 --- /dev/null +++ b/internal/block/validate.go @@ -0,0 +1,148 @@ +package block + +import ( + "encoding/binary" + "fmt" + "hash/crc32" + "os" + "path/filepath" + + "git.dvdt.dev/david/ingot/internal/index" +) + +// ValidationError describes a single integrity issue found during validation. +type ValidationError struct { + Block string // ULID or directory name + Section string // "meta", "index", "chunks" + Detail string +} + +func (e ValidationError) Error() string { + return fmt.Sprintf("%s: %s: %s", e.Block, e.Section, e.Detail) +} + +// ReadMeta reads and returns the BlockMeta for a block directory. +// Exported for use by ingotctl. +func ReadMeta(dir string) (BlockMeta, error) { + return readMeta(dir) +} + +// Validate performs integrity checks on a single block directory. +// It checks meta.json consistency, index integrity (magic, version, TOC CRC), +// and CRC validation on every chunk entry. +func Validate(dir string) []ValidationError { + var errs []ValidationError + blockName := filepath.Base(dir) + + // 1. Validate meta.json. + meta, err := readMeta(dir) + if err != nil { + errs = append(errs, ValidationError{blockName, "meta", err.Error()}) + return errs + } + if meta.Version != 1 { + errs = append(errs, ValidationError{blockName, "meta", + fmt.Sprintf("unsupported version %d", meta.Version)}) + } + if meta.MinTime > meta.MaxTime { + errs = append(errs, ValidationError{blockName, "meta", + fmt.Sprintf("minTime %d > maxTime %d", meta.MinTime, meta.MaxTime)}) + } + + // 2. Validate index file. + indexData, err := os.ReadFile(filepath.Join(dir, "index")) + if err != nil { + errs = append(errs, ValidationError{blockName, "index", err.Error()}) + } else { + _, err := index.NewReader(indexData) + if err != nil { + errs = append(errs, ValidationError{blockName, "index", err.Error()}) + } + } + + // 3. Validate all chunk segment files — iterate every entry and check CRC. + chunksDir := filepath.Join(dir, chunksDirName) + entries, err := os.ReadDir(chunksDir) + if err != nil { + errs = append(errs, ValidationError{blockName, "chunks", err.Error()}) + return errs + } + + for _, e := range entries { + if e.IsDir() { + continue + } + segIdx := parseSegmentName(e.Name()) + if segIdx < 0 { + continue + } + + data, err := os.ReadFile(filepath.Join(chunksDir, e.Name())) + if err != nil { + errs = append(errs, ValidationError{blockName, "chunks", + fmt.Sprintf("segment %s: %s", e.Name(), err)}) + continue + } + + segErrs := validateChunkSegment(data, e.Name()) + for _, se := range segErrs { + errs = append(errs, ValidationError{blockName, "chunks", se}) + } + } + + return errs +} + +// validateChunkSegment checks every chunk entry in a segment file for CRC integrity. +func validateChunkSegment(data []byte, name string) []string { + var errs []string + + if len(data) < chunkHeaderLen { + return []string{fmt.Sprintf("segment %s: too short for header", name)} + } + + magic := binary.BigEndian.Uint32(data[:4]) + if magic != chunkMagic { + return []string{fmt.Sprintf("segment %s: invalid magic %#x", name, magic)} + } + if data[4] != chunkVersion { + return []string{fmt.Sprintf("segment %s: unsupported version %d", name, data[4])} + } + + off := chunkHeaderLen + entryIdx := 0 + for off < len(data) { + if off+chunkEntryHeaderLen > len(data) { + errs = append(errs, fmt.Sprintf("segment %s entry %d at offset %d: truncated header", + name, entryIdx, off)) + break + } + + dataLen := int(binary.BigEndian.Uint32(data[off : off+4])) + encoding := data[off+4] + off += chunkEntryHeaderLen + + end := off + dataLen + chunkEntryCRCLen + if end > len(data) { + errs = append(errs, fmt.Sprintf("segment %s entry %d: truncated data (need %d bytes, have %d)", + name, entryIdx, dataLen+chunkEntryCRCLen, len(data)-off)) + break + } + + chunkBytes := data[off : off+dataLen] + off += dataLen + + wantCRC := binary.BigEndian.Uint32(data[off : off+4]) + crc := crc32.New(castagnoliTable) + crc.Write([]byte{encoding}) + crc.Write(chunkBytes) + if crc.Sum32() != wantCRC { + errs = append(errs, fmt.Sprintf("segment %s entry %d at offset %d: CRC mismatch", + name, entryIdx, off-dataLen-chunkEntryHeaderLen)) + } + off += chunkEntryCRCLen + entryIdx++ + } + + return errs +} diff --git a/internal/block/validate_test.go b/internal/block/validate_test.go new file mode 100644 index 0000000..529caf6 --- /dev/null +++ b/internal/block/validate_test.go @@ -0,0 +1,191 @@ +package block + +import ( + "encoding/binary" + "os" + "path/filepath" + "strings" + "testing" + + "git.dvdt.dev/david/ingot/labels" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestValidate(t *testing.T) { + tests := []struct { + name string + setup func(t *testing.T) string // returns block dir + wantErrors int + wantMatch string // substring to find in concatenated errors; "" matches everything + }{ + { + name: "valid_block", + setup: func(t *testing.T) string { + dir := t.TempDir() + series := []SeriesFlush{ + { + Ref: 1, + Labels: labels.FromStrings("__name__", "temp"), + Chunks: []ChunkData{ + {MinT: 0, MaxT: 1000, Data: makeTestChunk(t)}, + }, + }, + } + ulid, err := Flush(dir, series) + require.NoError(t, err) + return filepath.Join(dir, ulid) + }, + wantErrors: 0, + wantMatch: "", + }, + { + name: "missing_meta", + setup: func(t *testing.T) string { + dir := t.TempDir() + series := []SeriesFlush{ + { + Ref: 1, + Labels: labels.FromStrings("__name__", "temp"), + Chunks: []ChunkData{ + {MinT: 0, MaxT: 1000, Data: makeTestChunk(t)}, + }, + }, + } + ulid, err := Flush(dir, series) + require.NoError(t, err) + blockDir := filepath.Join(dir, ulid) + os.Remove(filepath.Join(blockDir, "meta.json")) + return blockDir + }, + wantErrors: 1, + wantMatch: "meta", + }, + { + name: "corrupt_chunk_crc", + setup: func(t *testing.T) string { + dir := t.TempDir() + series := []SeriesFlush{ + { + Ref: 1, + Labels: labels.FromStrings("__name__", "temp"), + Chunks: []ChunkData{ + {MinT: 0, MaxT: 1000, Data: makeTestChunk(t)}, + }, + }, + } + ulid, err := Flush(dir, series) + require.NoError(t, err) + blockDir := filepath.Join(dir, ulid) + + // Corrupt a byte in the chunk data. + chunkPath := filepath.Join(blockDir, "chunks", "000001") + data, err := os.ReadFile(chunkPath) + require.NoError(t, err) + data[chunkHeaderLen+chunkEntryHeaderLen+2] ^= 0xFF + require.NoError(t, os.WriteFile(chunkPath, data, 0644)) + return blockDir + }, + wantErrors: 1, + wantMatch: "CRC mismatch", + }, + { + name: "corrupt_chunk_magic", + setup: func(t *testing.T) string { + dir := t.TempDir() + series := []SeriesFlush{ + { + Ref: 1, + Labels: labels.FromStrings("__name__", "temp"), + Chunks: []ChunkData{ + {MinT: 0, MaxT: 1000, Data: makeTestChunk(t)}, + }, + }, + } + ulid, err := Flush(dir, series) + require.NoError(t, err) + blockDir := filepath.Join(dir, ulid) + + chunkPath := filepath.Join(blockDir, "chunks", "000001") + data, err := os.ReadFile(chunkPath) + require.NoError(t, err) + binary.BigEndian.PutUint32(data[:4], 0xDEADBEEF) + require.NoError(t, os.WriteFile(chunkPath, data, 0644)) + return blockDir + }, + wantErrors: 1, + wantMatch: "invalid magic", + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + blockDir := tc.setup(t) + errs := Validate(blockDir) + assert.Equal(t, tc.wantErrors, len(errs), "error count: %v", errs) + + // Concatenate all error strings; "" is contained in everything. + var combined strings.Builder + for _, e := range errs { + combined.WriteString(e.Error()) + combined.WriteByte('\n') + } + assert.Contains(t, combined.String(), tc.wantMatch) + }) + } +} + +func TestReadMeta(t *testing.T) { + tests := []struct { + name string + setup func(t *testing.T) (string, string) // returns (dir, ulid) + wantULID bool // true = ULID should match + wantMinT int64 + wantMaxT int64 + wantNSer int + wantNChk int + wantErr error + }{ + { + name: "valid_block", + setup: func(t *testing.T) (string, string) { + dir := t.TempDir() + ulid, err := Flush(dir, []SeriesFlush{ + { + Ref: 1, + Labels: labels.FromStrings("__name__", "test"), + Chunks: []ChunkData{ + {MinT: 100, MaxT: 200, Data: makeTestChunk(t)}, + }, + }, + }) + require.NoError(t, err) + return dir, ulid + }, + wantULID: true, + wantMinT: 100, + wantMaxT: 200, + wantNSer: 1, + wantNChk: 1, + wantErr: nil, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + dir, ulid := tc.setup(t) + meta, err := ReadMeta(filepath.Join(dir, ulid)) + assert.Equal(t, tc.wantErr, err) + assert.Equal(t, tc.wantULID, meta.ULID == ulid) + assert.Equal(t, tc.wantMinT, meta.MinTime) + assert.Equal(t, tc.wantMaxT, meta.MaxTime) + assert.Equal(t, tc.wantNSer, meta.Stats.NumSeries) + assert.Equal(t, tc.wantNChk, meta.Stats.NumChunks) + }) + } +} + +func makeTestChunk(t *testing.T) []byte { + t.Helper() + return makeChunk(t, []sample{s(0, 1.0), s(15000, 2.0), s(30000, 3.0)}) +} diff --git a/internal/head/head.go b/internal/head/head.go index 4cae262..3f52c6a 100644 --- a/internal/head/head.go +++ b/internal/head/head.go @@ -272,3 +272,29 @@ func (h *Head) AllPostings() []uint64 { func (h *Head) DataDir() string { return h.dataDir } + +// Stats returns a snapshot of head statistics. +func (h *Head) Stats() HeadStats { + var s HeadStats + h.series.forEach(func(ms *memSeries) { + s.NumSeries++ + ms.mu.Lock() + if ms.chunk != nil && ms.chunk.NumSamples() > 0 { + s.NumActiveChunks++ + } + s.NumActiveChunks += len(ms.sealed) + ms.mu.Unlock() + }) + return s +} + +// HeadStats holds a snapshot of head statistics. +type HeadStats struct { + NumSeries int + NumActiveChunks int +} + +// WALSyncDuration returns the duration of the most recent WAL fsync in seconds. +func (h *Head) WALSyncDuration() float64 { + return h.wal.LastSyncDuration() +} diff --git a/internal/wal/wal.go b/internal/wal/wal.go index e891343..ac1fe62 100644 --- a/internal/wal/wal.go +++ b/internal/wal/wal.go @@ -3,6 +3,7 @@ package wal import ( "os" "sync" + "sync/atomic" "time" ) @@ -46,6 +47,8 @@ type WAL struct { segmentOff int64 buf []byte + lastSyncDur atomic.Int64 // nanoseconds of last fsync + done chan struct{} wg sync.WaitGroup } @@ -164,7 +167,21 @@ func (w *WAL) Replay() (*Reader, error) { func (w *WAL) Sync() error { w.mu.Lock() defer w.mu.Unlock() - return w.segment.Sync() + return w.timedSync() +} + +// LastSyncDuration returns the duration of the most recent fsync in seconds. +func (w *WAL) LastSyncDuration() float64 { + ns := w.lastSyncDur.Load() + return float64(ns) / 1e9 +} + +// timedSync fsyncs the segment and records the duration. Caller must hold w.mu. +func (w *WAL) timedSync() error { + start := time.Now() + err := w.segment.Sync() + w.lastSyncDur.Store(int64(time.Since(start))) + return err } // Truncate deletes all segments with index less than below. @@ -244,7 +261,7 @@ func (w *WAL) syncLoop(interval time.Duration) { return case <-ticker.C: w.mu.Lock() - w.segment.Sync() + w.timedSync() w.mu.Unlock() } } diff --git a/metrics.go b/metrics.go new file mode 100644 index 0000000..9a3c347 --- /dev/null +++ b/metrics.go @@ -0,0 +1,68 @@ +package ingot + +import ( + "git.dvdt.dev/david/ingot/labels" +) + +// Self-instrumentation metric names. +const ( + MetricHeadSeries = "ingot_head_series" + MetricHeadChunksActive = "ingot_head_chunks_active" + MetricBlocksTotal = "ingot_blocks_total" + MetricCompactionsTotal = "ingot_compactions_total" + MetricWALFsyncDurationS = "ingot_wal_fsync_duration_seconds" +) + +// metricsRefs caches series refs for self-instrumentation metrics. +type metricsRefs struct { + headSeries uint64 + headChunksActive uint64 + blocksTotal uint64 + compactionsTotal uint64 + walFsyncDuration uint64 +} + +// collectMetrics snapshots the DB's internal stats and writes them as +// ingot series via the normal Appender path. Called periodically from +// the compact loop. +func (db *DB) collectMetrics() { + now := db.opts.clock()() + + hs := db.head.Stats() + db.mu.RLock() + numBlocks := len(db.blocks) + db.mu.RUnlock() + compactions := db.compactionCount.Load() + walFsync := db.head.WALSyncDuration() + + type metric struct { + name string + value float64 + ref *uint64 + } + metrics := []metric{ + {MetricHeadSeries, float64(hs.NumSeries), &db.metricsR.headSeries}, + {MetricHeadChunksActive, float64(hs.NumActiveChunks), &db.metricsR.headChunksActive}, + {MetricBlocksTotal, float64(numBlocks), &db.metricsR.blocksTotal}, + {MetricCompactionsTotal, float64(compactions), &db.metricsR.compactionsTotal}, + {MetricWALFsyncDurationS, walFsync, &db.metricsR.walFsyncDuration}, + } + + app := db.Appender() + for _, m := range metrics { + ref := *m.ref + var ls []labels.Label + if ref == 0 { + ls = labels.FromStrings("__name__", m.name) + } + newRef, err := app.Append(ref, ls, now, m.value) + if err != nil { + // OOO rejection or other transient error — skip this cycle. + app.Rollback() + return + } + *m.ref = newRef + } + app.Commit() +} + diff --git a/metrics_test.go b/metrics_test.go new file mode 100644 index 0000000..a31c50a --- /dev/null +++ b/metrics_test.go @@ -0,0 +1,148 @@ +package ingot + +import ( + "math" + "testing" + + "git.dvdt.dev/david/ingot/labels" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestCollectMetrics(t *testing.T) { + tests := []struct { + name string + setup func(t *testing.T, db *DB) + collectCalls int + wantHeadSeries float64 // expected value after last collectMetrics call + wantBlocksTotal float64 + wantCompactions float64 + wantMetricCount int // total ingot_* series + }{ + { + name: "empty_db", + setup: func(t *testing.T, db *DB) {}, + // First call snapshots 0 series, writes 5 metric series. + // Second call snapshots 5 series (the metrics themselves). + collectCalls: 2, + wantHeadSeries: 5, + wantBlocksTotal: 0, + wantCompactions: 0, + wantMetricCount: 5, + }, + { + name: "with_user_series", + setup: func(t *testing.T, db *DB) { + app := db.Appender() + _, err := app.Append(0, labels.FromStrings("__name__", "temp", "room", "a"), 1000, 1.0) + require.NoError(t, err) + _, err = app.Append(0, labels.FromStrings("__name__", "temp", "room", "b"), 1000, 2.0) + require.NoError(t, err) + require.NoError(t, app.Commit()) + }, + // 2 user series + 5 metric series = 7 + collectCalls: 2, + wantHeadSeries: 7, + wantBlocksTotal: 0, + wantCompactions: 0, + wantMetricCount: 5, + }, + { + name: "with_block", + setup: func(t *testing.T, db *DB) { + app := db.Appender() + ref, err := app.Append(0, labels.FromStrings("__name__", "temp"), 0, 0) + require.NoError(t, err) + for i := 1; i < 250; i++ { + _, err = app.Append(ref, nil, int64(i*15000), float64(i)) + require.NoError(t, err) + } + require.NoError(t, app.Commit()) + _, err = db.FlushOlderThan(math.MaxInt64) + require.NoError(t, err) + }, + collectCalls: 2, + wantHeadSeries: 6, // 1 user + 5 metric series + wantBlocksTotal: 1, + wantCompactions: 0, + wantMetricCount: 5, + }, + { + name: "idempotent_five_calls", + setup: func(t *testing.T, db *DB) {}, + collectCalls: 5, + wantHeadSeries: 5, + wantBlocksTotal: 0, + wantCompactions: 0, + wantMetricCount: 5, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + clock := &testClock{now: 100_000} + db, err := Open(t.TempDir(), Options{Clock: clock.fn()}) + require.NoError(t, err) + t.Cleanup(func() { db.Close() }) + + tc.setup(t, db) + + for i := 0; i < tc.collectCalls; i++ { + clock.now = int64(200_000 + i*1000) + db.collectMetrics() + } + + // Query each expected metric. + lastValue := func(name string) (float64, bool) { + q, err := db.Querier(math.MinInt64, math.MaxInt64) + require.NoError(t, err) + defer q.Close() + ss := q.Select(labels.MustNewMatcher(labels.MatchEqual, "__name__", name)) + found := false + var last float64 + for ss.Next() { + it := ss.At().Iterator() + for it.Next() { + _, last = it.At() + } + require.NoError(t, it.Err()) + found = true + } + require.NoError(t, ss.Err()) + return last, found + } + + v, found := lastValue(MetricHeadSeries) + assert.True(t, found, "ingot_head_series not found") + assert.Equal(t, tc.wantHeadSeries, v, "ingot_head_series") + + v, found = lastValue(MetricBlocksTotal) + assert.True(t, found, "ingot_blocks_total not found") + assert.Equal(t, tc.wantBlocksTotal, v, "ingot_blocks_total") + + v, found = lastValue(MetricCompactionsTotal) + assert.True(t, found, "ingot_compactions_total not found") + assert.Equal(t, tc.wantCompactions, v, "ingot_compactions_total") + + // Count total ingot_* series. + q, err := db.Querier(math.MinInt64, math.MaxInt64) + require.NoError(t, err) + defer q.Close() + ss := q.Select(labels.MustNewMatcher(labels.MatchRegexp, "__name__", "ingot_.*")) + count := 0 + for ss.Next() { + count++ + } + require.NoError(t, ss.Err()) + assert.Equal(t, tc.wantMetricCount, count, "metric series count") + }) + } +} + +type testClock struct { + now int64 +} + +func (c *testClock) fn() func() int64 { + return func() int64 { return c.now } +}