From fdbf94f2a105434ff7c16cd26c98453626da8913 Mon Sep 17 00:00:00 2001 From: Lou Knauer Date: Mon, 14 Mar 2022 08:48:22 +0100 Subject: [PATCH] Optionally delete checkpoints instead of archiving --- archive.go | 24 ++++++++++++++++++------ cc-metric-store.go | 15 ++++++++++----- debug.go | 2 +- memstore.go | 31 +++++++++++++++++++++++++++++++ 4 files changed, 60 insertions(+), 12 deletions(-) diff --git a/archive.go b/archive.go index 313ef4c..fb39cfa 100644 --- a/archive.go +++ b/archive.go @@ -449,7 +449,7 @@ func findFiles(direntries []fs.DirEntry, t int64, findMoreRecentFiles bool) ([]s // ZIP all checkpoint files older than `from` together and write them to the `archiveDir`, // deleting them from the `checkpointsDir`. -func ArchiveCheckpoints(checkpointsDir, archiveDir string, from int64) (int, error) { +func ArchiveCheckpoints(checkpointsDir, archiveDir string, from int64, deleteInstead bool) (int, error) { entries1, err := os.ReadDir(checkpointsDir) if err != nil { return 0, err @@ -469,7 +469,7 @@ func ArchiveCheckpoints(checkpointsDir, archiveDir string, from int64) (int, err go func() { defer wg.Done() for workItem := range work { - m, err := archiveCheckpoints(workItem.cdir, workItem.adir, from) + m, err := archiveCheckpoints(workItem.cdir, workItem.adir, from, deleteInstead) if err != nil { log.Printf("error while archiving %s/%s: %s", workItem.cluster, workItem.host, err.Error()) atomic.AddInt32(&errs, 1) @@ -509,7 +509,7 @@ func ArchiveCheckpoints(checkpointsDir, archiveDir string, from int64) (int, err } // Helper function for `ArchiveCheckpoints`. -func archiveCheckpoints(dir string, archiveDir string, from int64) (int, error) { +func archiveCheckpoints(dir string, archiveDir string, from int64, deleteInstead bool) (int, error) { entries, err := os.ReadDir(dir) if err != nil { return 0, err @@ -520,6 +520,18 @@ func archiveCheckpoints(dir string, archiveDir string, from int64) (int, error) return 0, err } + if deleteInstead { + n := 0 + for _, checkpoint := range files { + filename := filepath.Join(dir, checkpoint) + if err = os.Remove(filename); err != nil { + return n, err + } + n += 1 + } + return n, nil + } + filename := filepath.Join(archiveDir, fmt.Sprintf("%d.zip", from)) f, err := os.OpenFile(filename, os.O_CREATE|os.O_WRONLY, 0644) if err != nil && os.IsNotExist(err) { @@ -538,15 +550,15 @@ func archiveCheckpoints(dir string, archiveDir string, from int64) (int, error) defer zw.Close() n := 0 - for _, jsonFile := range files { - filename := filepath.Join(dir, jsonFile) + for _, checkpoint := range files { + filename := filepath.Join(dir, checkpoint) r, err := os.Open(filename) if err != nil { return n, err } defer r.Close() - w, err := zw.Create(jsonFile) + w, err := zw.Create(checkpoint) if err != nil { return n, err } diff --git a/cc-metric-store.go b/cc-metric-store.go index d514d94..6845b06 100644 --- a/cc-metric-store.go +++ b/cc-metric-store.go @@ -97,8 +97,9 @@ type Config struct { Restore string `json:"restore"` } `json:"checkpoints"` Archive struct { - Interval string `json:"interval"` - RootDir string `json:"directory"` + Interval string `json:"interval"` + RootDir string `json:"directory"` + DeleteInstead bool `json:"delete-instead"` } `json:"archive"` } @@ -200,7 +201,7 @@ func intervals(wg *sync.WaitGroup, ctx context.Context) { case <-ticks: t := time.Now().Add(-d) log.Printf("start archiving checkpoints (older than %s)...\n", t.Format(time.RFC3339)) - n, err := ArchiveCheckpoints(conf.Checkpoints.RootDir, conf.Archive.RootDir, t.Unix()) + n, err := ArchiveCheckpoints(conf.Checkpoints.RootDir, conf.Archive.RootDir, t.Unix(), conf.Archive.DeleteInstead) if err != nil { log.Printf("archiving failed: %s\n", err.Error()) } else { @@ -236,14 +237,18 @@ func main() { restoreFrom := startupTime.Add(-d) log.Printf("Loading checkpoints newer than %s\n", restoreFrom.Format(time.RFC3339)) files, err := memoryStore.FromCheckpoint(conf.Checkpoints.RootDir, restoreFrom.Unix()) + loadedData := memoryStore.SizeInBytes() / 1024 / 1024 // In MB if err != nil { log.Fatalf("Loading checkpoints failed: %s\n", err.Error()) } else { - log.Printf("Checkpoints loaded (%d files, that took %dms)\n", files, time.Since(startupTime).Milliseconds()) + log.Printf("Checkpoints loaded (%d files, %d MB, that took %dms)\n", files, loadedData, time.Since(startupTime).Milliseconds()) } runtime.GC() - debug.SetGCPercent(20) + if loadedData > 1000 { + debug.SetGCPercent(20) + } + ctx, shutdown := context.WithCancel(context.Background()) var wg sync.WaitGroup diff --git a/debug.go b/debug.go index 81bd940..712c928 100644 --- a/debug.go +++ b/debug.go @@ -44,7 +44,7 @@ func (l *level) debugDump(w *bufio.Writer, m *MemoryStore, indent string) error } func (m *MemoryStore) DebugDump(w *bufio.Writer) error { - fmt.Fprintf(w, "MemoryStore:\n") + fmt.Fprintf(w, "MemoryStore (%d MB):\n", m.SizeInBytes()/1024/1024) m.root.debugDump(w, m, " ") return w.Flush() } diff --git a/memstore.go b/memstore.go index 023883b..189a4eb 100644 --- a/memstore.go +++ b/memstore.go @@ -3,6 +3,7 @@ package main import ( "errors" "sync" + "unsafe" ) // Default buffer capacity. @@ -231,6 +232,14 @@ func (b *buffer) iterFromTo(from, to int64, callback func(b *buffer) error) erro return nil } +func (b *buffer) count() int64 { + res := int64(len(b.data)) + if b.prev != nil { + res += b.prev.count() + } + return res +} + // Could also be called "node" as this forms a node in a tree structure. // Called level because "node" might be confusing here. // Can be both a leaf or a inner node. In this tree structue, inner nodes can @@ -320,6 +329,24 @@ func (l *level) free(t int64) (int, error) { return n, nil } +func (l *level) sizeInBytes() int64 { + l.lock.RLock() + defer l.lock.RUnlock() + size := int64(0) + + for _, b := range l.metrics { + if b != nil { + size += b.count() * int64(unsafe.Sizeof(Float(0))) + } + } + + for _, child := range l.children { + size += child.sizeInBytes() + } + + return size +} + type MemoryStore struct { root level // root of the tree structure metrics map[string]MetricConfig @@ -477,6 +504,10 @@ func (m *MemoryStore) FreeAll() error { return nil } +func (m *MemoryStore) SizeInBytes() int64 { + return m.root.sizeInBytes() +} + // Given a selector, return a list of all children of the level selected. func (m *MemoryStore) ListChildren(selector []string) []string { lvl := &m.root