ClusterCockpit/cc-backend
1
0
Fork 0
mirror of https://github.com/ClusterCockpit/cc-backend synced 2026-01-20 19:41:44 +01:00
Code Issues Packages Releases Activity

Refactor Archive to Cleanup and rename everything

Browse Source
This commit is contained in:
Aditya Ujeniya
2026-01-16 14:09:50 +01:00
parent 1d62ee1e22
commit 1a41629535
6 changed files with 78 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
internal/metricstore/archive.go
View File

@@ -23,19 +23,23 @@ import (
// Worker for either Archiving or Deleting files
func Archiving(wg *sync.WaitGroup, ctx context.Context) {
if Keys.Archive != nil {
func CleanUp(wg *sync.WaitGroup, ctx context.Context) {
if Keys.Cleanup.Mode == "archive" {
// Run as Archiver
runWorker(wg, ctx,
Keys.Archive.ArchiveInterval,
cleanUpWorker(wg, ctx,
Keys.Cleanup.Interval,
"archiving",
Keys.Archive.RootDir,
Keys.Archive.DeleteInstead,
Keys.Cleanup.RootDir,
false,
)
} else {
if Keys.Cleanup.Interval == "" {
Keys.Cleanup.Interval = Keys.RetentionInMemory
}
// Run as Deleter
runWorker(wg, ctx,
Keys.RetentionInMemory,
cleanUpWorker(wg, ctx,
Keys.Cleanup.Interval,
"deleting",
"",
true,
@@ -44,7 +48,7 @@ func Archiving(wg *sync.WaitGroup, ctx context.Context) {
}
// runWorker takes simple values to configure what it does
func runWorker(wg *sync.WaitGroup, ctx context.Context, interval string, mode string, archiveDir string, delete bool) {
func cleanUpWorker(wg *sync.WaitGroup, ctx context.Context, interval string, mode string, cleanupDir string, delete bool) {
go func() {
defer wg.Done()
@@ -67,12 +71,12 @@ func runWorker(wg *sync.WaitGroup, ctx context.Context, interval string, mode st
t := time.Now().Add(-d)
cclog.Infof("[METRICSTORE]> start %s checkpoints (older than %s)...", mode, t.Format(time.RFC3339))
n, err := ArchiveCheckpoints(Keys.Checkpoints.RootDir, archiveDir, t.Unix(), delete)
n, err := CleanupCheckpoints(Keys.Checkpoints.RootDir, cleanupDir, t.Unix(), delete)
if err != nil {
cclog.Errorf("[METRICSTORE]> %s failed: %s", mode, err.Error())
} else {
if delete && archiveDir == "" {
if delete && cleanupDir == "" {
cclog.Infof("[METRICSTORE]> done: %d checkpoints deleted", n)
} else {
cclog.Infof("[METRICSTORE]> done: %d files zipped and moved to archive", n)
@@ -85,9 +89,9 @@ func runWorker(wg *sync.WaitGroup, ctx context.Context, interval string, mode st
var ErrNoNewArchiveData error = errors.New("all data already archived")
// 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, deleteInstead bool) (int, error) {
// Delete or ZIP all checkpoint files older than `from` together and write them to the `cleanupDir`,
// deleting/moving them from the `checkpointsDir`.
func CleanupCheckpoints(checkpointsDir, cleanupDir string, from int64, deleteInstead bool) (int, error) {
entries1, err := os.ReadDir(checkpointsDir)
if err != nil {
return 0, err
@@ -107,7 +111,7 @@ func ArchiveCheckpoints(checkpointsDir, archiveDir string, from int64, deleteIns
go func() {
defer wg.Done()
for workItem := range work {
m, err := archiveCheckpoints(workItem.cdir, workItem.adir, from, deleteInstead)
m, err := cleanupCheckpoints(workItem.cdir, workItem.adir, from, deleteInstead)
if err != nil {
cclog.Errorf("error while archiving %s/%s: %s", workItem.cluster, workItem.host, err.Error())
atomic.AddInt32(&errs, 1)
@@ -125,7 +129,7 @@ func ArchiveCheckpoints(checkpointsDir, archiveDir string, from int64, deleteIns
for _, de2 := range entries2 {
cdir := filepath.Join(checkpointsDir, de1.Name(), de2.Name())
adir := filepath.Join(archiveDir, de1.Name(), de2.Name())
adir := filepath.Join(cleanupDir, de1.Name(), de2.Name())
work <- workItem{
adir: adir, cdir: cdir,
cluster: de1.Name(), host: de2.Name(),
@@ -146,8 +150,8 @@ func ArchiveCheckpoints(checkpointsDir, archiveDir string, from int64, deleteIns
return int(n), nil
}
// Helper function for `ArchiveCheckpoints`.
func archiveCheckpoints(dir string, archiveDir string, from int64, deleteInstead bool) (int, error) {
// Helper function for `CleanupCheckpoints`.
func cleanupCheckpoints(dir string, cleanupDir string, from int64, deleteInstead bool) (int, error) {
entries, err := os.ReadDir(dir)
if err != nil {
return 0, err
@@ -171,10 +175,10 @@ func archiveCheckpoints(dir string, archiveDir string, from int64, deleteInstead
return n, nil
}
filename := filepath.Join(archiveDir, fmt.Sprintf("%d.zip", from))
filename := filepath.Join(cleanupDir, fmt.Sprintf("%d.zip", from))
f, err := os.OpenFile(filename, os.O_CREATE|os.O_WRONLY, CheckpointFilePerms)
if err != nil && os.IsNotExist(err) {
err = os.MkdirAll(archiveDir, CheckpointDirPerms)
err = os.MkdirAll(cleanupDir, CheckpointDirPerms)
if err == nil {
f, err = os.OpenFile(filename, os.O_CREATE|os.O_WRONLY, CheckpointFilePerms)
}