[lxc-devel] [lxd/master] Detect and shrink large boltdb files

[freeekanayaka on Github]

A non-text attachment was scrubbed...
Name: not available
Type: text/x-mailbox
Size: 361 bytes
Desc: not available
URL: <http://lists.linuxcontainers.org/pipermail/lxc-devel/attachments/20181102/41fa4a9f/attachment.bin>
-------------- next part --------------
From afc3095159b000d03862c7e064df7cf844f327fb Mon Sep 17 00:00:00 2001
From: Free Ekanayaka <free.ekanayaka at canonical.com>
Date: Fri, 2 Nov 2018 15:11:49 +0100
Subject: [PATCH] Detect and shrink large boltdb files

Signed-off-by: Free Ekanayaka <free.ekanayaka at canonical.com>
---
 lxd/daemon.go  |  33 ++++++++++++---
 lxd/patches.go | 110 +++++++++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 138 insertions(+), 5 deletions(-)

diff --git a/lxd/daemon.go b/lxd/daemon.go
index 750e0585fd..0974ed420b 100644
--- a/lxd/daemon.go
+++ b/lxd/daemon.go
@@ -477,6 +477,34 @@ func (d *Daemon) init() error {
 		return err
 	}
 
+	clustered, err := cluster.Enabled(d.db)
+	if err != nil {
+		return err
+	}
+
+	// If not already applied, run the daemon patch that shrinks the boltdb
+	// file. We can't run this daemon patch later on along with the other
+	// ones because it needs to run before we open the cluster database.
+	appliedPatches, err := d.db.Patches()
+	if err != nil {
+		return errors.Wrap(err, "Fetch applied daemon patches")
+	}
+	if !shared.StringInSlice("shrink_logs_db_file", appliedPatches) {
+		if !clustered {
+			// We actually run the patch only if this lxd daemon is
+			// not clustered.
+			err := patchShrinkLogsDBFile("", d)
+			if err != nil {
+				return errors.Wrap(err, "Shrink logs.db file")
+			}
+		}
+
+		err = d.db.PatchesMarkApplied("shrink_logs_db_file")
+		if err != nil {
+			return err
+		}
+	}
+
 	/* Setup dqlite */
 	clusterLogLevel := "ERROR"
 	if shared.StringInSlice("dqlite", trace) {
@@ -529,11 +557,6 @@ func (d *Daemon) init() error {
 		return err
 	}
 
-	clustered, err := cluster.Enabled(d.db)
-	if err != nil {
-		return err
-	}
-
 	/* Open the cluster database */
 	for {
 		logger.Info("Initializing global database")
diff --git a/lxd/patches.go b/lxd/patches.go
index 3fa3252844..9c6b4bc72c 100644
--- a/lxd/patches.go
+++ b/lxd/patches.go
@@ -3,11 +3,16 @@ package main
 import (
 	"fmt"
 	"io/ioutil"
+	stdlog "log"
 	"os"
 	"path/filepath"
 	"strings"
 	"syscall"
+	"time"
 
+	"github.com/boltdb/bolt"
+	"github.com/hashicorp/raft"
+	"github.com/hashicorp/raft-boltdb"
 	"github.com/lxc/lxd/lxd/cluster"
 	"github.com/lxc/lxd/lxd/db"
 	"github.com/lxc/lxd/lxd/db/query"
@@ -36,6 +41,7 @@ import (
 */
 
 var patches = []patch{
+	{name: "shrink_logs_db_file", run: patchShrinkLogsDBFile},
 	{name: "invalid_profile_names", run: patchInvalidProfileNames},
 	{name: "leftover_profile_config", run: patchLeftoverProfileConfig},
 	{name: "network_permissions", run: patchNetworkPermissions},
@@ -173,6 +179,110 @@ func patchNetworkPermissions(name string, d *Daemon) error {
 	return nil
 }
 
+// Shrink a database/global/logs.db that grew unwildly due to a bug in the 3.6
+// release.
+func patchShrinkLogsDBFile(name string, d *Daemon) error {
+	dir := filepath.Join(d.os.VarDir, "database", "global")
+	info, err := os.Stat(filepath.Join(dir, "logs.db"))
+	if err != nil && !os.IsNotExist(err) {
+		return errors.Wrap(err, "Get the size of the boltdb database")
+	}
+
+	if info.Size() < 1024*1024*100 {
+		// Only try to shrink databases bigger than 100 Megabytes.
+		return nil
+	}
+
+	snaps, err := raft.NewFileSnapshotStoreWithLogger(
+		dir, 2, stdlog.New(ioutil.Discard, "", 0))
+	if err != nil {
+		return errors.Wrap(err, "Open snapshots")
+	}
+
+	metas, err := snaps.List()
+	if err != nil {
+		return errors.Wrap(err, "Fetch snapshots")
+	}
+
+	if len(metas) == 0 {
+		// No snapshot is available, we can't shrink. This should never
+		// happen, in practice.
+		logger.Warnf("Can't shrink boltdb store, no raft snapshot is available")
+		return nil
+	}
+
+	meta := metas[0] // The most recent snapshot.
+
+	// Copy all log entries from the current boltdb file into a new one,
+	// which will be smaller since it excludes all truncated entries that
+	pathCur := filepath.Join(dir, "logs.db")
+	// got allocated before the latest snapshot.
+	logsCur, err := raftboltdb.New(raftboltdb.Options{
+		Path: pathCur,
+		BoltOptions: &bolt.Options{
+			Timeout:  10 * time.Second,
+			ReadOnly: true,
+		},
+	})
+	if err != nil {
+		return errors.Wrap(err, "Open current boltdb store")
+	}
+	defer logsCur.Close()
+
+	pathNew := filepath.Join(dir, "logs.db.new")
+	logsNew, err := raftboltdb.New(raftboltdb.Options{
+		Path:        pathNew,
+		BoltOptions: &bolt.Options{Timeout: 10 * time.Second},
+	})
+	if err != nil {
+		return errors.Wrap(err, "Open new boltdb store")
+	}
+	defer logsNew.Close()
+
+	lastIndex, err := logsCur.LastIndex()
+	if err != nil {
+		return errors.Wrap(err, "Get most recent raft index")
+	}
+
+	for index := meta.Index; index <= lastIndex; index++ {
+		log := &raft.Log{}
+
+		err := logsCur.GetLog(index, log)
+		if err != nil {
+			return errors.Wrapf(err, "Get raft entry at index %d", index)
+		}
+
+		err = logsNew.StoreLog(log)
+		if err != nil {
+			return errors.Wrapf(err, "Store raft entry at index %d", index)
+		}
+	}
+
+	term, err := logsCur.GetUint64([]byte("CurrentTerm"))
+	if err != nil {
+		return errors.Wrap(err, "Get current term")
+	}
+	err = logsNew.SetUint64([]byte("CurrentTerm"), term)
+	if err != nil {
+		return errors.Wrap(err, "Store current term")
+	}
+
+	logsCur.Close()
+	logsNew.Close()
+
+	err = os.Remove(pathCur)
+	if err != nil {
+		return errors.Wrap(err, "Remove current boltdb store")
+	}
+
+	err = os.Rename(pathNew, pathCur)
+	if err != nil {
+		return errors.Wrap(err, "Rename new boltdb store")
+	}
+
+	return nil
+}
+
 func patchStorageApi(name string, d *Daemon) error {
 	var daemonConfig map[string]string
 	err := d.cluster.Transaction(func(tx *db.ClusterTx) error {