Permalink
Browse files

Minor cleanups after reviewing the code in this branch

* When `skeema` exits, gracefully close all connection pools, to avoid aborted
  connection counter/logging in some versions of MySQL

* `skeema diff`: If the only differences for a dir are schema-level DDL, the
  exit code now reflects this as a difference

* applier.TargetGroupChanForDir: skipCount return value is no longer a pointer

* cmd_init.go: Remove unnecessary createOptionFile() function

* cmd_pull.go: Track skipCount by return value, rather than a pointer arg
  • Loading branch information...
evanelias committed Oct 29, 2018
1 parent 2ba95f9 commit 135162a6741b41bfa1b7ee0ea23c4fff6b0418ca

Some generated files are not rendered by default. Learn more.

Oops, something went wrong.
@@ -47,6 +47,7 @@ func Worker(ctx context.Context, targetGroups <-chan TargetGroup, results chan<-
if diff.SchemaDDL != "" {
printer.syncPrintf(t.Instance, "", "%s;\n", diff.SchemaDDL)
targetStmtCount++
result.Differences = true
if !dryRun {
var schemaDDLErr error
if strings.HasPrefix(diff.SchemaDDL, "CREATE DATABASE") && t.SchemaFromInstance == nil {
@@ -177,14 +177,12 @@ func targetsForIdealSchema(idealSchema *fs.IdealSchema, dir *fs.Dir, instances [
}
// TargetGroupChanForDir returns a channel for obtaining TargetGroups for this
// dir and its subdirs, and a pointer to an int that will contain the count of
// skipped directories once the channel is closed.
func TargetGroupChanForDir(dir *fs.Dir) (<-chan TargetGroup, *int) {
// dir and its subdirs, and count of directories that were skipped due to non-
// fatal errors.
func TargetGroupChanForDir(dir *fs.Dir) (<-chan TargetGroup, int) {
targets, skipCount := TargetsForDir(dir, 5)
groups := make(chan TargetGroup)
var skipCount int
go func() {
var targets []*Target
targets, skipCount = TargetsForDir(dir, 5)
byInst := make(map[string]TargetGroup)
for _, t := range targets {
key := t.Instance.String()
@@ -195,5 +193,5 @@ func TargetGroupChanForDir(dir *fs.Dir) (<-chan TargetGroup, *int) {
}
close(groups)
}()
return groups, &skipCount
return groups, skipCount
}
@@ -187,7 +187,10 @@ func (s ApplierIntegrationSuite) TestTargetGroupChanForDir(t *testing.T) {
// Parent dir maps to 2 instances, and schema dir maps to 2 schemas, so expect
// 4 targets split into 2 groups (by instance)
dir := getDir(t, "../testdata/applier/multi", "")
tgchan, skipCountPtr := TargetGroupChanForDir(dir)
tgchan, skipCount := TargetGroupChanForDir(dir)
if skipCount != 0 {
t.Errorf("Expected skip count of 0, instead found %d", skipCount)
}
seen := make(map[string]bool, 2)
for tg := range tgchan {
if len(tg) != 2 || tg[0].Instance != tg[1].Instance {
@@ -203,16 +206,16 @@ func (s ApplierIntegrationSuite) TestTargetGroupChanForDir(t *testing.T) {
if len(seen) != 2 {
t.Errorf("Expected to see 2 target groups, instead found %d", len(seen))
}
if *skipCountPtr != 0 {
t.Errorf("Expected skip count of 0, instead found %d", *skipCountPtr)
}
// SQL syntax error in testdata/applier/sqlerror/one/bad.sql should cause one/
// dir to be skipped entirely for both hosts, so skipCount of 2. But other
// dir two/ has no errors and should successfully yield 2 targets (1 per host,
// and put into different targetgroups)
dir = getDir(t, "../testdata/applier/sqlerror", "")
tgchan, skipCountPtr = TargetGroupChanForDir(dir)
tgchan, skipCount = TargetGroupChanForDir(dir)
if skipCount != 2 {
t.Errorf("Expected skip count of 2, instead found %d", skipCount)
}
seen = make(map[string]bool, 2)
for tg := range tgchan {
if len(tg) != 1 {
@@ -228,9 +231,6 @@ func (s ApplierIntegrationSuite) TestTargetGroupChanForDir(t *testing.T) {
if len(seen) != 2 {
t.Errorf("Expected to see 2 target groups, instead found %d", len(seen))
}
if *skipCountPtr != 2 {
t.Errorf("Expected skip count of 2, instead found %d", *skipCountPtr)
}
}
func getBaseConfig(t *testing.T, cliFlags string) *mybase.Config {
@@ -129,8 +129,8 @@ func InitHandler(cfg *mybase.Config) error {
}
// Write the option file
if err := createOptionFile(hostDir.Path, hostOptionFile); err != nil {
return NewExitValue(CodeCantCreate, err.Error())
if err := hostOptionFile.Write(false); err != nil {
return NewExitValue(CodeCantCreate, "Unable to use directory %s: Unable to write to %s: %s", hostDir.Path, hostOptionFile.Path(), err)
}
verb := "Using"
@@ -182,12 +182,12 @@ func PopulateSchemaDir(s *tengo.Schema, parentDir *fs.Dir, makeSubdir bool) erro
// Put a .skeema file with the schema name in it. This is placed outside of
// any named section/environment since the default assumption is that schema
// names match between environments.
optionFile := mybase.NewFile(".skeema")
optionFile := mybase.NewFile(subPath, ".skeema")
optionFile.SetOptionValue("", "schema", s.Name)
optionFile.SetOptionValue("", "default-character-set", s.CharSet)
optionFile.SetOptionValue("", "default-collation", s.Collation)
if err := createOptionFile(subPath, optionFile); err != nil {
return NewExitValue(CodeCantCreate, "Unable to use directory %s for schema %s: %s", subPath, s.Name, err)
if err := optionFile.Write(false); err != nil {
return NewExitValue(CodeCantCreate, "Unable to use directory %s for schema %s: Unable to write to %s: %s", subPath, s.Name, optionFile.Path(), err)
}
} else {
subPath = parentDir.Path
@@ -269,12 +269,3 @@ func preparePath(dirPath string, globalConfig *mybase.Config) (created bool, err
return false, nil
}
func createOptionFile(dirPath string, file *mybase.File) error {
file.Dir = dirPath
err := file.Write(false)
if err == nil {
return nil
}
return fmt.Errorf("Unable to write to %s: %s", file.Path(), err)
}
@@ -43,30 +43,31 @@ func PullHandler(cfg *mybase.Config) error {
return err
}
var errCount int
if _, err = pullWalker(dir, &errCount, 5); err != nil {
var skipCount int
if _, skipCount, err = pullWalker(dir, 5); err != nil {
return err
}
if errCount == 0 {
if skipCount == 0 {
return nil
}
var plural string
if errCount > 1 {
if skipCount > 1 {
plural = "s"
}
return NewExitValue(CodePartialError, "Skipped %d operation%s due to error%s", errCount, plural, plural)
return NewExitValue(CodePartialError, "Skipped %d operation%s due to error%s", skipCount, plural, plural)
}
// pullWalker processes dir, and recursively calls itself on any subdirs. An
// error is only returned if something fatal occurs; otherwise, *errCount is
// incremented if some operation is skipped but it isn't fatal.
func pullWalker(dir *fs.Dir, errCount *int, maxDepth int) (handledSchemaNames []string, err error) {
// error is only returned if something fatal occurs. skipCount reflects the
// number of non-fatal failed operations that were skipped for dir and its
// subdirectories.
func pullWalker(dir *fs.Dir, maxDepth int) (handledSchemaNames []string, skipCount int, err error) {
var instance *tengo.Instance
if dir.Config.Changed("host") {
instance, err = dir.FirstInstance()
if err != nil {
*errCount++
return nil, nil
log.Warnf("Skipping %s: %s", dir, err)
return nil, 1, nil
}
}
@@ -83,7 +84,7 @@ func pullWalker(dir *fs.Dir, errCount *int, maxDepth int) (handledSchemaNames []
var schemaNames []string
if schemaNames, err = dir.SchemaNames(instance); err != nil {
return nil, fmt.Errorf("%s: Unable to fetch schema names mapped by this dir: %s", dir, err)
return nil, skipCount, fmt.Errorf("%s: Unable to fetch schema names mapped by this dir: %s", dir, err)
}
if len(schemaNames) == 0 {
log.Warnf("Ignoring directory %s -- did not map to any schema names\n", dir)
@@ -94,38 +95,39 @@ func pullWalker(dir *fs.Dir, errCount *int, maxDepth int) (handledSchemaNames []
if err == sql.ErrNoRows {
log.Infof("Deleted directory %s -- schema %s no longer exists\n", dir, handledSchemaNames[0])
// Explicitly return here to prevent later attempt at subdir traversal
return nil, dir.Delete()
return nil, skipCount, dir.Delete()
} else if err != nil {
return nil, fmt.Errorf("%s: Unable to fetch schema %s from %s: %s", dir, handledSchemaNames[0], instance, err)
return nil, skipCount, fmt.Errorf("%s: Unable to fetch schema %s from %s: %s", dir, handledSchemaNames[0], instance, err)
}
if err = pullSchemaDir(dir, instance, instSchema, idealSchema); err != nil {
return nil, err
return nil, skipCount, err
}
}
}
if subdirs, badCount, err := dir.Subdirs(); err != nil {
log.Errorf("Cannot list subdirs of %s: %s", dir, err)
*errCount++
skipCount++
} else if len(subdirs) > 0 && maxDepth <= 0 {
log.Warnf("Not walking subdirs of %s: max depth reached", dir)
*errCount += len(subdirs)
skipCount += len(subdirs)
} else {
*errCount += badCount
skipCount += badCount
allSubSchemaNames := make([]string, 0)
for _, sub := range subdirs {
subSchemaNames, walkErr := pullWalker(sub, errCount, maxDepth-1)
subSchemaNames, subSkipCount, walkErr := pullWalker(sub, maxDepth-1)
skipCount += subSkipCount
if walkErr != nil {
return nil, walkErr
return nil, skipCount, walkErr
}
allSubSchemaNames = append(allSubSchemaNames, subSchemaNames...)
}
if instance != nil && !dir.Config.Changed("schema") {
updateFlavor(dir, instance)
return nil, findNewSchemas(dir, instance, allSubSchemaNames)
return nil, skipCount, findNewSchemas(dir, instance, allSubSchemaNames)
}
}
return handledSchemaNames, nil
return handledSchemaNames, skipCount, nil
}
// pullSchemaDir performs appropriate pull logic on a dir that maps to one or
@@ -54,7 +54,7 @@ func PushHandler(cfg *mybase.Config) error {
briefMode := dir.Config.GetBool("dry-run") && dir.Config.GetBool("brief")
printer := applier.NewPrinter(briefMode)
g, ctx := errgroup.WithContext(context.Background())
tgchan, skipCountPtr := applier.TargetGroupChanForDir(dir)
tgchan, skipCount := applier.TargetGroupChanForDir(dir)
results := make(chan applier.Result)
workerCount, err := dir.Config.GetInt("concurrent-instances")
@@ -85,7 +85,7 @@ func PushHandler(cfg *mybase.Config) error {
return err
}
sum := applier.SumResults(allResults)
sum.SkipCount += *skipCountPtr
sum.SkipCount += skipCount
if sum.SkipCount+sum.UnsupportedCount == 0 {
if dir.Config.GetBool("dry-run") && sum.Differences {
@@ -5,6 +5,7 @@ import (
"os"
log "github.com/sirupsen/logrus"
"github.com/skeema/skeema/util"
)
// ExitValue represents an exit code for an operation. It satisfies the Error
@@ -76,5 +77,10 @@ func Exit(err error) {
}
log.Debugf("Exit code %d", exitCode)
}
// Gracefully close all connection pools, to avoid aborted connection counter/
// logging in some versions of MySQL
util.CloseCachedConnectionPools()
os.Exit(exitCode)
}
@@ -306,8 +306,8 @@ func (dir *Dir) SchemaNames(instance *tengo.Instance) ([]string, error) {
}
// HasSchema returns true if this dir maps to at least one schema, either by
// stating a "schema" option in the dir's config, and/or by having *.sql files
// that explicitly mention a schema name
// stating a "schema" option in the dir's config for the current environment,
// and/or by having *.sql files that explicitly mention a schema name.
func (dir *Dir) HasSchema() bool {
if dir.Config.Changed("schema") {
return true
@@ -0,0 +1,4 @@
CREATE TABLE four (
name varchar(30) NOT NULL,
PRIMARY KEY (name)
) ENGINE=InnoDB DEFAULT CHARSET=latin1;
@@ -0,0 +1,4 @@
CREATE TABLE three (
name varchar(30) NOT NULL,
PRIMARY KEY (name)
) ENGINE=InnoDB DEFAULT CHARSET=latin1;
@@ -34,3 +34,11 @@ func NewInstance(driver, dsn string) (*tengo.Instance, error) {
instanceCache.instanceMap[key] = instance
return instance, nil
}
// CloseCachedConnectionPools closes all connection pools in all cached
// Instances that were created via NewInstance.
func CloseCachedConnectionPools() {
for _, inst := range instanceCache.instanceMap {
inst.CloseAll()
}
}

Some generated files are not rendered by default. Learn more.

Oops, something went wrong.

0 comments on commit 135162a

Please sign in to comment.