Purge old schema versions from memory in historian (vitessio#13056)

* purge old schema versions from memory in historian

Signed-off-by: austenLacy <[email protected]>

* fix vttablet arg e2e test

Signed-off-by: Austen Lacy <[email protected]>

* only reassign historian schemas if necessary

Signed-off-by: Austen Lacy <[email protected]>

* dont use from_unixtime in query because already in unix

Signed-off-by: Austen Lacy <[email protected]>

---------

Signed-off-by: austenLacy <[email protected]>
Signed-off-by: Austen Lacy <[email protected]>
Co-authored-by: Austen Lacy <[email protected]>

Author: austenLacy

go/flags/endtoend/vttablet.txt

@@ -256,6 +256,7 @@ Usage of vttablet:
       --s3_backup_storage_root string                                    root prefix for all backup-related object names.
       --s3_backup_tls_skip_verify_cert                                   skip the 'certificate is valid' check for SSL connections.
       --sanitize_log_messages                                            Remove potentially sensitive information in tablet INFO, WARNING, and ERROR log messages such as query parameters.
+      --schema-version-max-age-seconds int                               max age of schema version records to kept in memory by the vreplication historian
       --security_policy string                                           the name of a registered security policy to use for controlling access to URLs - empty means allow all for anyone (built-in policies: deny-all, read-only)
       --service_map strings                                              comma separated list of services to enable (or disable if prefixed with '-') Example: grpc-queryservice
       --serving_state_grace_period duration                              how long to pause after broadcasting health to vtgate, before enforcing a new serving state

go/vt/vttablet/tabletserver/schema/engine.go

@@ -115,7 +115,7 @@ func NewEngine(env tabletenv.Env) *Engine {
 
 		schemazHandler(se.GetSchema(), w, r)
 	})
-	se.historian = newHistorian(env.Config().TrackSchemaVersions, se.conns)
+	se.historian = newHistorian(env.Config().TrackSchemaVersions, env.Config().SchemaVersionMaxAgeSeconds, se.conns)
 	return se
 }
 
@@ -627,8 +627,9 @@ func (se *Engine) handleHTTPSchema(response http.ResponseWriter) {
 // doesn't reload.  Use SetTableForTests to set table schema.
 func NewEngineForTests() *Engine {
 	se := &Engine{
-		isOpen: true,
-		tables: make(map[string]*Table),
+		isOpen:    true,
+		tables:    make(map[string]*Table),
+		historian: newHistorian(false, 0, nil),
 	}
 	return se
 }

go/vt/vttablet/tabletserver/schema/engine_test.go

@@ -75,7 +75,7 @@ func TestOpenAndReload(t *testing.T) {
 	))
 	firstReadRowsValue := 12
 	AddFakeInnoDBReadRowsResult(db, firstReadRowsValue)
-	se := newEngine(10, 10*time.Second, 10*time.Second, db)
+	se := newEngine(10, 10*time.Second, 10*time.Second, 0, db)
 	se.Open()
 	defer se.Close()
 
@@ -266,7 +266,7 @@ func TestReloadWithSwappedTables(t *testing.T) {
 	firstReadRowsValue := 12
 	AddFakeInnoDBReadRowsResult(db, firstReadRowsValue)
 
-	se := newEngine(10, 10*time.Second, 10*time.Second, db)
+	se := newEngine(10, 10*time.Second, 10*time.Second, 0, db)
 	se.Open()
 	defer se.Close()
 	want := initialSchema()
@@ -416,7 +416,7 @@ func TestOpenFailedDueToExecErr(t *testing.T) {
 	schematest.AddDefaultQueries(db)
 	want := "injected error"
 	db.RejectQueryPattern(baseShowTablesPattern, want)
-	se := newEngine(10, 1*time.Second, 1*time.Second, db)
+	se := newEngine(10, 1*time.Second, 1*time.Second, 0, db)
 	err := se.Open()
 	if err == nil || !strings.Contains(err.Error(), want) {
 		t.Errorf("se.Open: %v, want %s", err, want)
@@ -446,7 +446,7 @@ func TestOpenFailedDueToTableErr(t *testing.T) {
 	})
 
 	AddFakeInnoDBReadRowsResult(db, 0)
-	se := newEngine(10, 1*time.Second, 1*time.Second, db)
+	se := newEngine(10, 1*time.Second, 1*time.Second, 0, db)
 	err := se.Open()
 	want := "Row count exceeded"
 	if err == nil || !strings.Contains(err.Error(), want) {
@@ -458,7 +458,7 @@ func TestExportVars(t *testing.T) {
 	db := fakesqldb.New(t)
 	defer db.Close()
 	schematest.AddDefaultQueries(db)
-	se := newEngine(10, 1*time.Second, 1*time.Second, db)
+	se := newEngine(10, 1*time.Second, 1*time.Second, 0, db)
 	se.Open()
 	defer se.Close()
 	expvar.Do(func(kv expvar.KeyValue) {
@@ -470,7 +470,7 @@ func TestStatsURL(t *testing.T) {
 	db := fakesqldb.New(t)
 	defer db.Close()
 	schematest.AddDefaultQueries(db)
-	se := newEngine(10, 1*time.Second, 1*time.Second, db)
+	se := newEngine(10, 1*time.Second, 1*time.Second, 0, db)
 	se.Open()
 	defer se.Close()
 
@@ -500,7 +500,7 @@ func TestSchemaEngineCloseTickRace(t *testing.T) {
 		})
 	AddFakeInnoDBReadRowsResult(db, 12)
 	// Start the engine with a small reload tick
-	se := newEngine(10, 100*time.Millisecond, 1*time.Second, db)
+	se := newEngine(10, 100*time.Millisecond, 1*time.Second, 0, db)
 	err := se.Open()
 	require.NoError(t, err)
 
@@ -527,13 +527,14 @@ func TestSchemaEngineCloseTickRace(t *testing.T) {
 	}
 }
 
-func newEngine(queryCacheSize int, reloadTime time.Duration, idleTimeout time.Duration, db *fakesqldb.DB) *Engine {
+func newEngine(queryCacheSize int, reloadTime time.Duration, idleTimeout time.Duration, schemaMaxAgeSeconds int64, db *fakesqldb.DB) *Engine {
 	config := tabletenv.NewDefaultConfig()
 	config.QueryCacheSize = queryCacheSize
 	config.SchemaReloadIntervalSeconds.Set(reloadTime)
 	config.OltpReadPool.IdleTimeoutSeconds.Set(idleTimeout)
 	config.OlapReadPool.IdleTimeoutSeconds.Set(idleTimeout)
 	config.TxPool.IdleTimeoutSeconds.Set(idleTimeout)
+	config.SchemaVersionMaxAgeSeconds = schemaMaxAgeSeconds
 	se := NewEngine(tabletenv.NewEnv(config, "SchemaTest"))
 	se.InitDBConfig(newDBConfigs(db).DbaWithDB())
 	return se

go/vt/vttablet/tabletserver/schema/historian.go

@@ -18,9 +18,9 @@ package schema
 
 import (
 	"context"
-	"fmt"
 	"sort"
 	"sync"
+	"time"
 
 	"google.golang.org/protobuf/proto"
 
@@ -35,36 +35,40 @@ import (
 	"vitess.io/vitess/go/vt/sqlparser"
 )
 
-const getSchemaVersions = "select id, pos, ddl, time_updated, schemax from _vt.schema_version where id > %d order by id asc"
+const getInitialSchemaVersions = "select id, pos, ddl, time_updated, schemax from _vt.schema_version where time_updated > %d order by id asc"
+const getNextSchemaVersions = "select id, pos, ddl, time_updated, schemax from _vt.schema_version where id > %d order by id asc"
 
 // vl defines the glog verbosity level for the package
 const vl = 10
 
 // trackedSchema has the snapshot of the table at a given pos (reached by ddl)
 type trackedSchema struct {
-	schema map[string]*binlogdatapb.MinimalTable
-	pos    mysql.Position
-	ddl    string
+	schema      map[string]*binlogdatapb.MinimalTable
+	pos         mysql.Position
+	ddl         string
+	timeUpdated int64
 }
 
 // historian implements the Historian interface by calling schema.Engine for the underlying schema
 // and supplying a schema for a specific version by loading the cached values from the schema_version table
 // The schema version table is populated by the Tracker
 type historian struct {
-	conns   *connpool.Pool
-	lastID  int64
-	schemas []*trackedSchema
-	mu      sync.Mutex
-	enabled bool
-	isOpen  bool
+	conns               *connpool.Pool
+	lastID              int64
+	schemas             []*trackedSchema
+	mu                  sync.Mutex
+	enabled             bool
+	isOpen              bool
+	schemaMaxAgeSeconds int64
 }
 
 // newHistorian creates a new historian. It expects a schema.Engine instance
-func newHistorian(enabled bool, conns *connpool.Pool) *historian {
+func newHistorian(enabled bool, schemaMaxAgeSeconds int64, conns *connpool.Pool) *historian {
 	sh := historian{
-		conns:   conns,
-		lastID:  0,
-		enabled: enabled,
+		conns:               conns,
+		lastID:              0,
+		enabled:             enabled,
+		schemaMaxAgeSeconds: schemaMaxAgeSeconds,
 	}
 	return &sh
 }
@@ -164,7 +168,17 @@ func (h *historian) loadFromDB(ctx context.Context) error {
 		return err
 	}
 	defer conn.Recycle()
-	tableData, err := conn.Exec(ctx, fmt.Sprintf(getSchemaVersions, h.lastID), 10000, true)
+
+	var tableData *sqltypes.Result
+	if h.lastID == 0 && h.schemaMaxAgeSeconds > 0 { // only at vttablet start
+		schemaMaxAge := time.Now().UTC().Add(time.Duration(-h.schemaMaxAgeSeconds) * time.Second)
+		tableData, err = conn.Exec(ctx, sqlparser.BuildParsedQuery(getInitialSchemaVersions,
+			schemaMaxAge.Unix()).Query, 10000, true)
+	} else {
+		tableData, err = conn.Exec(ctx, sqlparser.BuildParsedQuery(getNextSchemaVersions,
+			h.lastID).Query, 10000, true)
+	}
+
 	if err != nil {
 		log.Infof("Error reading schema_tracking table %v, will operate with the latest available schema", err)
 		return nil
@@ -177,6 +191,14 @@ func (h *historian) loadFromDB(ctx context.Context) error {
 		h.schemas = append(h.schemas, trackedSchema)
 		h.lastID = id
 	}
+
+	if h.lastID != 0 && h.schemaMaxAgeSeconds > 0 {
+		// To avoid keeping old schemas in memory which can lead to an eventual memory leak
+		// we purge any older than h.schemaMaxAgeSeconds. Only needs to be done when adding
+		// new schema rows.
+		h.purgeOldSchemas()
+	}
+
 	h.sortSchemas()
 	return nil
 }
@@ -217,13 +239,40 @@ func (h *historian) readRow(row []sqltypes.Value) (*trackedSchema, int64, error)
 		tables[t.Name] = t
 	}
 	tSchema := &trackedSchema{
-		schema: tables,
-		pos:    pos,
-		ddl:    ddl,
+		schema:      tables,
+		pos:         pos,
+		ddl:         ddl,
+		timeUpdated: timeUpdated,
 	}
 	return tSchema, id, nil
 }
 
+func (h *historian) purgeOldSchemas() {
+	maxAgeDuration := time.Duration(h.schemaMaxAgeSeconds) * time.Second
+	shouldPurge := false
+
+	// check if we have any schemas we need to purge and only create the filtered
+	// slice if necessary
+	for _, s := range h.schemas {
+		if time.Since(time.Unix(s.timeUpdated, 0)) > maxAgeDuration {
+			shouldPurge = true
+			break
+		}
+	}
+
+	if !shouldPurge {
+		return
+	}
+
+	filtered := make([]*trackedSchema, 0)
+	for _, s := range h.schemas {
+		if time.Since(time.Unix(s.timeUpdated, 0)) < maxAgeDuration {
+			filtered = append(filtered, s)
+		}
+	}
+	h.schemas = filtered
+}
+
 // sortSchemas sorts entries in ascending order of gtid, ex: 40,44,48
 func (h *historian) sortSchemas() {
 	sort.Slice(h.schemas, func(i int, j int) bool {

go/vt/vttablet/tabletserver/schema/historian_test.go

@@ -19,6 +19,7 @@ package schema
 import (
 	"fmt"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/require"
 	"google.golang.org/protobuf/proto"
@@ -71,7 +72,7 @@ func getDbSchemaBlob(t *testing.T, tables map[string]*binlogdatapb.MinimalTable)
 }
 
 func TestHistorian(t *testing.T) {
-	se, db, cancel := getTestSchemaEngine(t)
+	se, db, cancel := getTestSchemaEngine(t, 0)
 	defer cancel()
 
 	se.EnableHistorian(false)
@@ -174,3 +175,77 @@ func TestHistorian(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, exp3, fmt.Sprintf("%v", tab))
 }
+
+func TestHistorianPurgeOldSchemas(t *testing.T) {
+	schemaVersionMaxAgeSeconds := 3600 // 1 hour
+	se, db, cancel := getTestSchemaEngine(t, int64(schemaVersionMaxAgeSeconds))
+	defer cancel()
+
+	gtidPrefix := "MySQL56/7b04699f-f5e9-11e9-bf88-9cb6d089e1c3:"
+	gtid1 := gtidPrefix + "1-10"
+	ddl1 := "create table tracker_test (id int)"
+	// create the first record 1 day ago so it gets purged from memory
+	ts1 := time.Now().Add(time.Duration(-24) * time.Hour)
+	_, _, _ = ddl1, ts1, db
+	se.EnableHistorian(true)
+	_, err := se.GetTableForPos(sqlparser.NewIdentifierCS("t1"), gtid1)
+	require.Equal(t, "table t1 not found in vttablet schema", err.Error())
+	var blob1 string
+
+	fields := []*querypb.Field{{
+		Name: "id",
+		Type: sqltypes.Int32,
+	}, {
+		Name: "pos",
+		Type: sqltypes.VarBinary,
+	}, {
+		Name: "ddl",
+		Type: sqltypes.VarBinary,
+	}, {
+		Name: "time_updated",
+		Type: sqltypes.Int32,
+	}, {
+		Name: "schemax",
+		Type: sqltypes.Blob,
+	}}
+
+	table := getTable("t1", []string{"id1", "id2"}, []querypb.Type{querypb.Type_INT32, querypb.Type_INT32}, []int64{0})
+	tables := make(map[string]*binlogdatapb.MinimalTable)
+	tables["t1"] = table
+	blob1 = getDbSchemaBlob(t, tables)
+	db.AddQueryPattern("select id, pos, ddl, time_updated, schemax from _vt\\.schema_version where time_updated \\>.*", &sqltypes.Result{
+		Fields: fields,
+		Rows: [][]sqltypes.Value{
+			{sqltypes.NewInt32(1), sqltypes.NewVarBinary(gtid1), sqltypes.NewVarBinary(ddl1), sqltypes.NewInt32(int32(ts1.Unix())), sqltypes.NewVarBinary(blob1)},
+		},
+	})
+	require.Nil(t, se.RegisterVersionEvent())
+	_, err = se.GetTableForPos(sqlparser.NewIdentifierCS("t1"), gtid1)
+	// validate the old schema has been purged
+	require.Equal(t, "table t1 not found in vttablet schema", err.Error())
+	require.Equal(t, 0, len(se.historian.schemas))
+
+	// add a second schema record row with a time_updated that won't be purged
+	gtid2 := gtidPrefix + "1-20"
+	_, err = se.GetTableForPos(sqlparser.NewIdentifierCS("t1"), gtid2)
+	require.Equal(t, "table t1 not found in vttablet schema", err.Error())
+
+	table = getTable("t1", []string{"id1", "id2"}, []querypb.Type{querypb.Type_INT32, querypb.Type_VARBINARY}, []int64{0})
+	tables["t1"] = table
+	blob2 := getDbSchemaBlob(t, tables)
+	ddl2 := "alter table t1 modify column id2 varbinary"
+	// set time_updated younger than the cutoff from historian.schemaMaxAgeSeconds
+	ts2 := time.Now().Add(time.Duration(-60) * time.Second)
+	db.AddQuery("select id, pos, ddl, time_updated, schemax from _vt.schema_version where id > 1 order by id asc", &sqltypes.Result{
+		Fields: fields,
+		Rows: [][]sqltypes.Value{
+			{sqltypes.NewInt32(2), sqltypes.NewVarBinary(gtid2), sqltypes.NewVarBinary(ddl2), sqltypes.NewInt32(int32(ts2.Unix())), sqltypes.NewVarBinary(blob2)},
+		},
+	})
+	require.Nil(t, se.RegisterVersionEvent())
+	exp2 := `name:"t1" fields:{name:"id1" type:INT32 table:"t1"} fields:{name:"id2" type:VARBINARY table:"t1"} p_k_columns:0`
+	tab, err := se.GetTableForPos(sqlparser.NewIdentifierCS("t1"), gtid2)
+	require.NoError(t, err)
+	require.Equal(t, exp2, fmt.Sprintf("%v", tab))
+	require.Equal(t, 1, len(se.historian.schemas))
+}

go/vt/vttablet/tabletserver/schema/main_test.go

@@ -27,7 +27,7 @@ import (
 	"vitess.io/vitess/go/sqltypes"
 )
 
-func getTestSchemaEngine(t *testing.T) (*Engine, *fakesqldb.DB, func()) {
+func getTestSchemaEngine(t *testing.T, schemaMaxAgeSeconds int64) (*Engine, *fakesqldb.DB, func()) {
 	db := fakesqldb.New(t)
 	db.AddQuery("select unix_timestamp()", sqltypes.MakeTestResult(sqltypes.MakeTestFields(
 		"t",
@@ -37,7 +37,7 @@ func getTestSchemaEngine(t *testing.T) (*Engine, *fakesqldb.DB, func()) {
 	db.AddQueryPattern(baseShowTablesPattern, &sqltypes.Result{})
 	db.AddQuery(mysql.BaseShowPrimary, &sqltypes.Result{})
 	AddFakeInnoDBReadRowsResult(db, 1)
-	se := newEngine(10, 10*time.Second, 10*time.Second, db)
+	se := newEngine(10, 10*time.Second, 10*time.Second, schemaMaxAgeSeconds, db)
 	require.NoError(t, se.Open())
 	cancel := func() {
 		defer db.Close()

go/vt/vttablet/tabletserver/schema/tracker_test.go

@@ -30,7 +30,7 @@ import (
 
 func TestTracker(t *testing.T) {
 	initialSchemaInserted := false
-	se, db, cancel := getTestSchemaEngine(t)
+	se, db, cancel := getTestSchemaEngine(t, 0)
 	defer cancel()
 	gtid1 := "MySQL56/7b04699f-f5e9-11e9-bf88-9cb6d089e1c3:1-10"
 	ddl1 := "create table tracker_test (id int)"
@@ -91,7 +91,7 @@ func TestTracker(t *testing.T) {
 
 func TestTrackerShouldNotInsertInitialSchema(t *testing.T) {
 	initialSchemaInserted := false
-	se, db, cancel := getTestSchemaEngine(t)
+	se, db, cancel := getTestSchemaEngine(t, 0)
 	gtid1 := "MySQL56/7b04699f-f5e9-11e9-bf88-9cb6d089e1c3:1-10"
 
 	defer cancel()

go/vt/vttablet/tabletserver/tabletenv/config.go

@@ -133,6 +133,7 @@ func registerTabletEnvFlags(fs *pflag.FlagSet) {
 	fs.BoolVar(&currentConfig.AnnotateQueries, "queryserver-config-annotate-queries", defaultConfig.AnnotateQueries, "prefix queries to MySQL backend with comment indicating vtgate principal (user) and target tablet type")
 	fs.BoolVar(&currentConfig.WatchReplication, "watch_replication_stream", false, "When enabled, vttablet will stream the MySQL replication stream from the local server, and use it to update schema when it sees a DDL.")
 	fs.BoolVar(&currentConfig.TrackSchemaVersions, "track_schema_versions", false, "When enabled, vttablet will store versions of schemas at each position that a DDL is applied and allow retrieval of the schema corresponding to a position")
+	fs.Int64Var(&currentConfig.SchemaVersionMaxAgeSeconds, "schema-version-max-age-seconds", 0, "max age of schema version records to kept in memory by the vreplication historian")
 	fs.BoolVar(&currentConfig.TwoPCEnable, "twopc_enable", defaultConfig.TwoPCEnable, "if the flag is on, 2pc is enabled. Other 2pc flags must be supplied.")
 	fs.StringVar(&currentConfig.TwoPCCoordinatorAddress, "twopc_coordinator_address", defaultConfig.TwoPCCoordinatorAddress, "address of the (VTGate) process(es) that will be used to notify of abandoned transactions.")
 	SecondsVar(fs, &currentConfig.TwoPCAbandonAge, "twopc_abandon_age", defaultConfig.TwoPCAbandonAge, "time in seconds. Any unresolved transaction older than this time will be sent to the coordinator to be resolved.")
@@ -289,6 +290,7 @@ type TabletConfig struct {
 	SignalSchemaChangeReloadIntervalSeconds Seconds `json:"signalSchemaChangeReloadIntervalSeconds,omitempty"`
 	WatchReplication                        bool    `json:"watchReplication,omitempty"`
 	TrackSchemaVersions                     bool    `json:"trackSchemaVersions,omitempty"`
+	SchemaVersionMaxAgeSeconds              int64   `json:"schemaVersionMaxAgeSeconds,omitempty"`
 	TerseErrors                             bool    `json:"terseErrors,omitempty"`
 	AnnotateQueries                         bool    `json:"annotateQueries,omitempty"`
 	MessagePostponeParallelism              int     `json:"messagePostponeParallelism,omitempty"`