featurebase/dax/controller/snapping_turtle.go

package controller

import (
	"context"
	"time"

	"github.com/featurebasedb/featurebase/v3/dax"
	"github.com/featurebasedb/featurebase/v3/logger"
)

func (c *Controller) snappingTurtleRoutine(period time.Duration, control chan struct{}, log logger.Logger) error {
	if period == 0 {
		return nil
	}
	ticker := time.NewTicker(period)
	for {
		select {
		case <-c.stopping:
			ticker.Stop()
			log.Debugf("Stopping Snapping Turtle")
			return nil
		case <-ticker.C:
			c.snapAll(log)
		case <-control:
			c.snapAll(log)
		}
	}
}

func (c *Controller) snapAll(log logger.Logger) {
	start := time.Now()
	defer func() {
		log.Printf("full snapshot took: %v", time.Since(start))
	}()
	ctx := context.Background()

	tx, err := c.Transactor.BeginTx(ctx, false)
	if err != nil {
		log.Printf("Error getting transaction for snapping turtle: %v", err)
		return
	}
	defer tx.Rollback()

	qdbs, err := c.Schemar.Databases(tx, "")
	if err != nil {
		log.Printf("couldn't get databases: %v", err)
	}

	for _, qdb := range qdbs {
		c.snapAllForDatabase(tx, qdb.QualifiedID(), log)
	}
}

func (c *Controller) snapAllForDatabase(tx dax.Transaction, qdbid dax.QualifiedDatabaseID, log logger.Logger) {
	log.Debugf("snapAllForDatabase: %s", qdbid)
	computeNodes, err := c.Balancer.CurrentState(tx, dax.RoleTypeCompute, qdbid)
	if err != nil {
		log.Printf("Error getting compute balancer state for snapping turtle: %v", err)
	}

	// Weird nested loop for snapshotting shard data. The reason for
	// this is to avoid hotspotting each node in turn and spread the
	// snapshotting load across all nodes rather than snapshotting all
	// jobs on one node and then moving onto the next one.
	i := 0
	stillWorking := true
	for stillWorking {
		stillWorking = false
		for _, workerInfo := range computeNodes {
			if len(workerInfo.Jobs) <= i {
				continue
			}
			stillWorking = true
			j, err := decodeShard(workerInfo.Jobs[i])
			if err != nil {
				log.Printf("couldn't decode a shard out of the job: '%s', err: %v", workerInfo.Jobs[i], err)
			}
			if err := c.snapshotShardData(tx, j.t.QualifiedTableID(), j.shardNum()); err != nil {
				log.Printf("Couldn't snapshot table: %s, shard: %d, error: %v", j.t, j.shardNum(), err)
			}
		}
		i++
	}

	// Get all tables across all orgs/dbs so we can snapshot all keyed
	// fields and look up whether a table is keyed to snapshot its
	// partitions.
	tables, err := c.Schemar.Tables(tx, dax.QualifiedDatabaseID{})
	if err != nil {
		log.Printf("Couldn't get schema for snapshotting keys: %v", err)
		return
	}
	// snapshot keyed fields
	tableMap := make(map[dax.TableKey]*dax.QualifiedTable)
	for _, table := range tables {
		tableMap[table.Key()] = table
		for _, f := range table.Fields {
			if f.StringKeys() && !f.IsPrimaryKey() {
				if err := c.snapshotFieldKeys(tx, table.QualifiedID(), f.Name); err != nil {
					log.Printf("Couldn't snapshot table: %s, field: %s, error: %v", table, f.Name, err)
				}
			}
		}
	}

	// Get all partition jobs from balancer and snapshot table keys
	// for any partition that goes with a keyed table. Doing the same
	// weird nested loop thing to avoid doing all jobs on one node
	// back to back.
	translateNodes, err := c.Balancer.CurrentState(tx, dax.RoleTypeTranslate, qdbid)
	if err != nil {
		log.Printf("Error getting translate balancer state for snapping turtle: %v", err)
	}

	i = 0
	stillWorking = true
	for stillWorking {
		stillWorking = false
		for _, workerInfo := range translateNodes {
			if len(workerInfo.Jobs) <= i {
				continue
			}
			stillWorking = true
			j, err := decodePartition(workerInfo.Jobs[i])
			if err != nil {
				table := tableMap[j.table()]
				if table.StringKeys() {
					if err := c.snapshotTableKeys(tx, table.QualifiedID(), j.partitionNum()); err != nil {
						log.Printf("Couldn't snapshot table: %s, partition: %d, error: %v", table, j.partitionNum(), err)
					}
				}
				log.Printf("couldn't decode a partition out of the job: '%s', err: %v", workerInfo.Jobs[i], err)
			}
		}
		i++
	}
	log.Debugf("snapAllForDatabase complete: %s", qdbid)
}