From a51b41507dc2dd178beed3c804575e124cd9f988 Mon Sep 17 00:00:00 2001 From: AnshumanTripathi Date: Fri, 15 Sep 2017 13:24:07 -0700 Subject: [PATCH 01/11] Implemented querying concurrent schema --- go/cmd/ccql/main.go | 16 ++++++++++------ go/logic/ccql.go | 45 ++++++++++++++++++++++++++++++++++++++++++++- 2 files changed, 54 insertions(+), 7 deletions(-) diff --git a/go/cmd/ccql/main.go b/go/cmd/ccql/main.go index ec2c02a..80dc18c 100644 --- a/go/cmd/ccql/main.go +++ b/go/cmd/ccql/main.go @@ -15,6 +15,7 @@ import ( golib_log "github.com/outbrain/golib/log" "gopkg.in/gcfg.v1" + "strings" ) var AppVersion string @@ -34,12 +35,13 @@ func main() { osUser = usr.Username } + osUser = "root" help := flag.Bool("help", false, "Display usage") user := flag.String("u", osUser, "MySQL username") password := flag.String("p", "", "MySQL password") askPassword := flag.Bool("ask-pass", false, "prompt for MySQL password") credentialsFile := flag.String("C", "", "Credentials file, expecting [client] scope, with 'user', 'password' fields. Overrides -u and -p") - defaultSchema := flag.String("d", "information_schema", "Default schema to use") + schemaList := flag.String("s", "information_schema", "List of Schema to query from.") hostsList := flag.String("h", "", "Comma or space delimited list of hosts in hostname[:port] format. If not given, hosts read from stdin") hostsFile := flag.String("H", "", "Hosts file, hostname[:port] comma or space or newline delimited format. If not given, hosts read from stdin") queriesText := flag.String("q", "", "Query/queries to execute") @@ -94,9 +96,9 @@ func main() { if *credentialsFile != "" { mySQLConfig := struct { Client struct { - User string - Password string - } + User string + Password string + } }{} gcfg.RelaxedParserMode = true err := gcfg.ReadFileInto(&mySQLConfig, *credentialsFile) @@ -117,7 +119,9 @@ func main() { *password = string(passwd) } - if err := logic.QueryHosts(hosts, *user, *password, *defaultSchema, queries, *maxConcurrency, *timeout); err != nil { + schemas := strings.Split(*schemaList, ",") + + if err := logic.QuerySchemas(hosts, *user, *password, schemas, queries, *maxConcurrency, *timeout); err != nil { os.Exit(1) } -} +} \ No newline at end of file diff --git a/go/logic/ccql.go b/go/logic/ccql.go index 9e0a39a..710c9cc 100644 --- a/go/logic/ccql.go +++ b/go/logic/ccql.go @@ -11,6 +11,7 @@ import ( // queryHost connects to a given host, issues the given set of queries, and outputs the results // line per row in tab delimited format func queryHost(host string, user string, password string, defaultSchema string, queries []string, timeout float64) error { + log.Println("Running for schema", defaultSchema) mysqlURI := fmt.Sprintf("%s:%s@tcp(%s)/%s?timeout=%fs", user, password, host, defaultSchema, timeout) db, _, err := sqlutils.GetDB(mysqlURI) if err != nil { @@ -28,7 +29,7 @@ func queryHost(host string, user string, password string, defaultSchema string, output = append(output, rowCell.String) } rowOutput := strings.Join(output, "\t") - fmt.Println(rowOutput) + fmt.Println(defaultSchema, rowOutput) } } return nil @@ -57,3 +58,45 @@ func QueryHosts(hosts []string, user string, password string, defaultSchema stri } return anyError } + +func QuerySchemas(hosts []string, user string, password string, schemas []string, queries []string, maxConcurrency uint, timeout float64) (anyError error) { + concurrentHosts := make(chan bool, maxConcurrency) + completedHosts := make(chan bool) + + concurrentSchemas := make(chan bool, maxConcurrency) + completedSchemas := make(chan bool) + + for _, host := range hosts { + go func(host string) { + concurrentHosts <- true + //For each host, run all queries for the respective schema + for _, schema := range schemas { + go func(schema string) { + concurrentSchemas <- true + if err := queryHost(host, user, password, schema, queries, timeout); err != nil { + anyError = err + log.Printf("%s %s", host, err.Error()) + } + <-concurrentSchemas + completedSchemas <- true + }(schema) + } + + // Barrier. Wait for all to complete + for range schemas { + <-completedSchemas + } + + <-concurrentHosts + + completedHosts <- true + }(host) + } + // Barrier. Wait for all to complete + for range hosts { + <-completedHosts + } + + return anyError +} + From 2e4edf78b3e695c106cf591bcfb511a93d267c22 Mon Sep 17 00:00:00 2001 From: AnshumanTripathi Date: Wed, 20 Sep 2017 15:27:12 -0700 Subject: [PATCH 02/11] Updated formatting and QuerySchema with WaitGroup --- go/cmd/ccql/main.go | 12 +++++----- go/logic/ccql.go | 57 +++++++++++---------------------------------- 2 files changed, 19 insertions(+), 50 deletions(-) diff --git a/go/cmd/ccql/main.go b/go/cmd/ccql/main.go index 80dc18c..e7fb38a 100644 --- a/go/cmd/ccql/main.go +++ b/go/cmd/ccql/main.go @@ -41,7 +41,7 @@ func main() { password := flag.String("p", "", "MySQL password") askPassword := flag.Bool("ask-pass", false, "prompt for MySQL password") credentialsFile := flag.String("C", "", "Credentials file, expecting [client] scope, with 'user', 'password' fields. Overrides -u and -p") - schemaList := flag.String("s", "information_schema", "List of Schema to query from.") + databases := flag.String("d", "information_schema", "List of databases to query from.") hostsList := flag.String("h", "", "Comma or space delimited list of hosts in hostname[:port] format. If not given, hosts read from stdin") hostsFile := flag.String("H", "", "Hosts file, hostname[:port] comma or space or newline delimited format. If not given, hosts read from stdin") queriesText := flag.String("q", "", "Query/queries to execute") @@ -96,9 +96,9 @@ func main() { if *credentialsFile != "" { mySQLConfig := struct { Client struct { - User string - Password string - } + User string + Password string + } }{} gcfg.RelaxedParserMode = true err := gcfg.ReadFileInto(&mySQLConfig, *credentialsFile) @@ -119,9 +119,9 @@ func main() { *password = string(passwd) } - schemas := strings.Split(*schemaList, ",") + schemas := strings.Split(*databases, ",") if err := logic.QuerySchemas(hosts, *user, *password, schemas, queries, *maxConcurrency, *timeout); err != nil { os.Exit(1) } -} \ No newline at end of file +} diff --git a/go/logic/ccql.go b/go/logic/ccql.go index 710c9cc..5a6a56e 100644 --- a/go/logic/ccql.go +++ b/go/logic/ccql.go @@ -6,13 +6,14 @@ import ( "strings" "github.com/outbrain/golib/sqlutils" + "sync" ) // queryHost connects to a given host, issues the given set of queries, and outputs the results // line per row in tab delimited format -func queryHost(host string, user string, password string, defaultSchema string, queries []string, timeout float64) error { - log.Println("Running for schema", defaultSchema) - mysqlURI := fmt.Sprintf("%s:%s@tcp(%s)/%s?timeout=%fs", user, password, host, defaultSchema, timeout) +func queryHost(host string, user string, password string, schema string, queries []string, timeout float64) error { + mysqlURI := fmt.Sprintf("%s:%s@tcp(%s)/%s?timeout=%fs", user, password, host, schema, timeout) + fmt.Println(mysqlURI) db, _, err := sqlutils.GetDB(mysqlURI) if err != nil { return err @@ -24,74 +25,43 @@ func queryHost(host string, user string, password string, defaultSchema string, return err } for _, row := range resultData { - output := []string{host} + output := []string{host, schema} for _, rowCell := range row { output = append(output, rowCell.String) } rowOutput := strings.Join(output, "\t") - fmt.Println(defaultSchema, rowOutput) + fmt.Println(rowOutput) } } return nil } -// QueryHosts will issue concurrent queries on given list of hosts -func QueryHosts(hosts []string, user string, password string, defaultSchema string, queries []string, maxConcurrency uint, timeout float64) (anyError error) { - concurrentHosts := make(chan bool, maxConcurrency) - completedHosts := make(chan bool) - - for _, host := range hosts { - go func(host string) { - concurrentHosts <- true - if err := queryHost(host, user, password, defaultSchema, queries, timeout); err != nil { - anyError = err - log.Printf("%s %s", host, err.Error()) - } - <-concurrentHosts - - completedHosts <- true - }(host) - } - // Barrier. Wait for all to complete - for range hosts { - <-completedHosts - } - return anyError -} - func QuerySchemas(hosts []string, user string, password string, schemas []string, queries []string, maxConcurrency uint, timeout float64) (anyError error) { concurrentHosts := make(chan bool, maxConcurrency) completedHosts := make(chan bool) - - concurrentSchemas := make(chan bool, maxConcurrency) - completedSchemas := make(chan bool) + var wg sync.WaitGroup for _, host := range hosts { go func(host string) { + wg.Add(len(schemas)) concurrentHosts <- true - //For each host, run all queries for the respective schema + // For each host, run all queries for the respective schema for _, schema := range schemas { go func(schema string) { - concurrentSchemas <- true if err := queryHost(host, user, password, schema, queries, timeout); err != nil { anyError = err log.Printf("%s %s", host, err.Error()) } - <-concurrentSchemas - completedSchemas <- true + defer wg.Done() }(schema) } - - // Barrier. Wait for all to complete - for range schemas { - <-completedSchemas - } - + wg.Wait() <-concurrentHosts - completedHosts <- true }(host) + } + // Barrier. Wait for all to complete for range hosts { <-completedHosts @@ -99,4 +69,3 @@ func QuerySchemas(hosts []string, user string, password string, schemas []string return anyError } - From 7900f1f1caf1c1d85139a3bcf29e0171f13bca5f Mon Sep 17 00:00:00 2001 From: AnshumanTripathi Date: Wed, 20 Sep 2017 18:07:23 -0700 Subject: [PATCH 03/11] Readme update. Removed default osuser --- README.md | 2 +- go/cmd/ccql/main.go | 1 - 2 files changed, 1 insertion(+), 2 deletions(-) diff --git a/README.md b/README.md index 5a18c05..7e06486 100644 --- a/README.md +++ b/README.md @@ -23,7 +23,7 @@ Usage of ccql: -Q string Query/queries input file -d string - Default schema to use (default "information_schema") + Schemas to use (default "information_schema") -h string Comma or space delimited list of hosts in hostname[:port] format. If not given, hosts read from stdin -m uint diff --git a/go/cmd/ccql/main.go b/go/cmd/ccql/main.go index e7fb38a..94c377d 100644 --- a/go/cmd/ccql/main.go +++ b/go/cmd/ccql/main.go @@ -35,7 +35,6 @@ func main() { osUser = usr.Username } - osUser = "root" help := flag.Bool("help", false, "Display usage") user := flag.String("u", osUser, "MySQL username") password := flag.String("p", "", "MySQL password") From 77c17f67477f94393bd019b9c0c9b8b3bc8f5d4f Mon Sep 17 00:00:00 2001 From: AnshumanTripathi Date: Thu, 21 Sep 2017 16:04:32 -0700 Subject: [PATCH 04/11] Removed log where connect uri was displayed --- go/logic/ccql.go | 2 -- 1 file changed, 2 deletions(-) diff --git a/go/logic/ccql.go b/go/logic/ccql.go index 5a6a56e..7f009d8 100644 --- a/go/logic/ccql.go +++ b/go/logic/ccql.go @@ -4,7 +4,6 @@ import ( "fmt" "log" "strings" - "github.com/outbrain/golib/sqlutils" "sync" ) @@ -13,7 +12,6 @@ import ( // line per row in tab delimited format func queryHost(host string, user string, password string, schema string, queries []string, timeout float64) error { mysqlURI := fmt.Sprintf("%s:%s@tcp(%s)/%s?timeout=%fs", user, password, host, schema, timeout) - fmt.Println(mysqlURI) db, _, err := sqlutils.GetDB(mysqlURI) if err != nil { return err From bddf30d0a624026768b70d2500dd11914bc9c5db Mon Sep 17 00:00:00 2001 From: AnshumanTripathi Date: Mon, 25 Sep 2017 15:22:35 -0700 Subject: [PATCH 05/11] Added view source schema flag to view the source schema in the output --- go/cmd/ccql/main.go | 4 +++- go/logic/ccql.go | 17 +++++++++++------ 2 files changed, 14 insertions(+), 7 deletions(-) diff --git a/go/cmd/ccql/main.go b/go/cmd/ccql/main.go index 94c377d..51369a4 100644 --- a/go/cmd/ccql/main.go +++ b/go/cmd/ccql/main.go @@ -47,6 +47,8 @@ func main() { queriesFile := flag.String("Q", "", "Query/queries input file") timeout := flag.Float64("t", 0, "Connect timeout seconds") maxConcurrency := flag.Uint("m", 32, "Max concurrent connections") + viewSourceSchema := flag.Bool("v", false, "View the source schema of the results") + flag.Parse() if AppVersion == "" { @@ -120,7 +122,7 @@ func main() { schemas := strings.Split(*databases, ",") - if err := logic.QuerySchemas(hosts, *user, *password, schemas, queries, *maxConcurrency, *timeout); err != nil { + if err := logic.QuerySchemas(hosts, *user, *password, schemas, queries, *maxConcurrency, *timeout, *viewSourceSchema); err != nil { os.Exit(1) } } diff --git a/go/logic/ccql.go b/go/logic/ccql.go index 7f009d8..51b4dc9 100644 --- a/go/logic/ccql.go +++ b/go/logic/ccql.go @@ -2,21 +2,21 @@ package logic import ( "fmt" + "github.com/outbrain/golib/sqlutils" "log" "strings" - "github.com/outbrain/golib/sqlutils" "sync" ) // queryHost connects to a given host, issues the given set of queries, and outputs the results // line per row in tab delimited format -func queryHost(host string, user string, password string, schema string, queries []string, timeout float64) error { +func queryHost(host string, user string, password string, schema string, queries []string, timeout float64, viewSourceSchema bool) error { mysqlURI := fmt.Sprintf("%s:%s@tcp(%s)/%s?timeout=%fs", user, password, host, schema, timeout) db, _, err := sqlutils.GetDB(mysqlURI) if err != nil { return err } - + fmt.Println() for _, query := range queries { resultData, err := sqlutils.QueryResultData(db, query) if err != nil { @@ -24,6 +24,10 @@ func queryHost(host string, user string, password string, schema string, queries } for _, row := range resultData { output := []string{host, schema} + if !viewSourceSchema { + output = append([]string(nil), output[:1]...) + output[0] = host + } for _, rowCell := range row { output = append(output, rowCell.String) } @@ -34,7 +38,8 @@ func queryHost(host string, user string, password string, schema string, queries return nil } -func QuerySchemas(hosts []string, user string, password string, schemas []string, queries []string, maxConcurrency uint, timeout float64) (anyError error) { +func QuerySchemas(hosts []string, user string, password string, schemas []string, queries []string, maxConcurrency uint, + timeout float64, viewSourceSchema bool) (anyError error) { concurrentHosts := make(chan bool, maxConcurrency) completedHosts := make(chan bool) var wg sync.WaitGroup @@ -46,11 +51,11 @@ func QuerySchemas(hosts []string, user string, password string, schemas []string // For each host, run all queries for the respective schema for _, schema := range schemas { go func(schema string) { - if err := queryHost(host, user, password, schema, queries, timeout); err != nil { + defer wg.Done() + if err := queryHost(host, user, password, schema, queries, timeout, viewSourceSchema); err != nil { anyError = err log.Printf("%s %s", host, err.Error()) } - defer wg.Done() }(schema) } wg.Wait() From 1a4af286eee1574c5b27753a0bd69fc023f03153 Mon Sep 17 00:00:00 2001 From: AnshumanTripathi Date: Fri, 29 Sep 2017 08:16:34 -0700 Subject: [PATCH 06/11] Ask pass input in the next line --- go/cmd/ccql/main.go | 2 +- go/logic/ccql.go | 3 --- 2 files changed, 1 insertion(+), 4 deletions(-) diff --git a/go/cmd/ccql/main.go b/go/cmd/ccql/main.go index 51369a4..af5f6ec 100644 --- a/go/cmd/ccql/main.go +++ b/go/cmd/ccql/main.go @@ -112,7 +112,7 @@ func main() { } if *askPassword { - fmt.Print("Mysql password: ") + fmt.Println("Mysql password: ") passwd, err := terminal.ReadPassword(int(syscall.Stdin)) if err != nil { log.Fatalf("\nError while get password:", err) diff --git a/go/logic/ccql.go b/go/logic/ccql.go index 51b4dc9..0bb3f94 100644 --- a/go/logic/ccql.go +++ b/go/logic/ccql.go @@ -16,7 +16,6 @@ func queryHost(host string, user string, password string, schema string, queries if err != nil { return err } - fmt.Println() for _, query := range queries { resultData, err := sqlutils.QueryResultData(db, query) if err != nil { @@ -43,7 +42,6 @@ func QuerySchemas(hosts []string, user string, password string, schemas []string concurrentHosts := make(chan bool, maxConcurrency) completedHosts := make(chan bool) var wg sync.WaitGroup - for _, host := range hosts { go func(host string) { wg.Add(len(schemas)) @@ -62,7 +60,6 @@ func QuerySchemas(hosts []string, user string, password string, schemas []string <-concurrentHosts completedHosts <- true }(host) - } // Barrier. Wait for all to complete From cc15a21f592ddc2333a45532e4db93eeb45cff07 Mon Sep 17 00:00:00 2001 From: AnshumanTripathi Date: Sun, 1 Oct 2017 01:56:25 -0700 Subject: [PATCH 07/11] better handling of logging of schema --- go/logic/ccql.go | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/go/logic/ccql.go b/go/logic/ccql.go index 0bb3f94..c8ce148 100644 --- a/go/logic/ccql.go +++ b/go/logic/ccql.go @@ -22,10 +22,10 @@ func queryHost(host string, user string, password string, schema string, queries return err } for _, row := range resultData { - output := []string{host, schema} - if !viewSourceSchema { - output = append([]string(nil), output[:1]...) - output[0] = host + output := []string{host} + if viewSourceSchema { + outputSchema := []string{schema} + output = append(output, outputSchema...) } for _, rowCell := range row { output = append(output, rowCell.String) From be414e35a2fb1d77bc271ff98cffca63e34fd78e Mon Sep 17 00:00:00 2001 From: AnshumanTripathi Date: Sun, 1 Oct 2017 10:49:06 -0700 Subject: [PATCH 08/11] removed appending of schema output --- go/logic/ccql.go | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/go/logic/ccql.go b/go/logic/ccql.go index c8ce148..2592491 100644 --- a/go/logic/ccql.go +++ b/go/logic/ccql.go @@ -24,8 +24,7 @@ func queryHost(host string, user string, password string, schema string, queries for _, row := range resultData { output := []string{host} if viewSourceSchema { - outputSchema := []string{schema} - output = append(output, outputSchema...) + output = append(output, schema) } for _, rowCell := range row { output = append(output, rowCell.String) From 1ecdc37237973abde0103c710cb9bd789886d4df Mon Sep 17 00:00:00 2001 From: Shlomi Noach Date: Tue, 24 Oct 2017 14:12:28 +0300 Subject: [PATCH 09/11] support for multiple schemas --- README.md | 4 +++- go/cmd/ccql/main.go | 11 ++++----- go/logic/ccql.go | 55 ++++++++++++++++++++++--------------------- go/text/hosts.go | 10 ++++++++ go/text/hosts_test.go | 14 +++++++++++ 5 files changed, 60 insertions(+), 34 deletions(-) diff --git a/README.md b/README.md index 7e06486..544e195 100644 --- a/README.md +++ b/README.md @@ -23,7 +23,9 @@ Usage of ccql: -Q string Query/queries input file -d string - Schemas to use (default "information_schema") + Default schema to use (default "information_schema") + -s string + Comma separated list of schemas, overrides '-d'. Implies printing schema name to output -h string Comma or space delimited list of hosts in hostname[:port] format. If not given, hosts read from stdin -m uint diff --git a/go/cmd/ccql/main.go b/go/cmd/ccql/main.go index af5f6ec..124e3af 100644 --- a/go/cmd/ccql/main.go +++ b/go/cmd/ccql/main.go @@ -15,7 +15,6 @@ import ( golib_log "github.com/outbrain/golib/log" "gopkg.in/gcfg.v1" - "strings" ) var AppVersion string @@ -40,14 +39,14 @@ func main() { password := flag.String("p", "", "MySQL password") askPassword := flag.Bool("ask-pass", false, "prompt for MySQL password") credentialsFile := flag.String("C", "", "Credentials file, expecting [client] scope, with 'user', 'password' fields. Overrides -u and -p") - databases := flag.String("d", "information_schema", "List of databases to query from.") + defaultSchema := flag.String("d", "information_schema", "Default schema to use") + schemasList := flag.String("s", "", "List of databases to query from; overrides -d, prints schema name to output") hostsList := flag.String("h", "", "Comma or space delimited list of hosts in hostname[:port] format. If not given, hosts read from stdin") hostsFile := flag.String("H", "", "Hosts file, hostname[:port] comma or space or newline delimited format. If not given, hosts read from stdin") queriesText := flag.String("q", "", "Query/queries to execute") queriesFile := flag.String("Q", "", "Query/queries input file") timeout := flag.Float64("t", 0, "Connect timeout seconds") maxConcurrency := flag.Uint("m", 32, "Max concurrent connections") - viewSourceSchema := flag.Bool("v", false, "View the source schema of the results") flag.Parse() @@ -112,7 +111,7 @@ func main() { } if *askPassword { - fmt.Println("Mysql password: ") + fmt.Print("Mysql password: ") passwd, err := terminal.ReadPassword(int(syscall.Stdin)) if err != nil { log.Fatalf("\nError while get password:", err) @@ -120,9 +119,9 @@ func main() { *password = string(passwd) } - schemas := strings.Split(*databases, ",") + schemas := text.SplitNonEmpty(*schemasList, ",") - if err := logic.QuerySchemas(hosts, *user, *password, schemas, queries, *maxConcurrency, *timeout, *viewSourceSchema); err != nil { + if err := logic.QueryHosts(hosts, *user, *password, *defaultSchema, schemas, queries, *maxConcurrency, *timeout); err != nil { os.Exit(1) } } diff --git a/go/logic/ccql.go b/go/logic/ccql.go index 2592491..1fd3303 100644 --- a/go/logic/ccql.go +++ b/go/logic/ccql.go @@ -2,15 +2,16 @@ package logic import ( "fmt" - "github.com/outbrain/golib/sqlutils" "log" "strings" "sync" + + "github.com/outbrain/golib/sqlutils" ) // queryHost connects to a given host, issues the given set of queries, and outputs the results // line per row in tab delimited format -func queryHost(host string, user string, password string, schema string, queries []string, timeout float64, viewSourceSchema bool) error { +func queryHost(host string, user string, password string, schema string, queries []string, timeout float64, printSchema bool) error { mysqlURI := fmt.Sprintf("%s:%s@tcp(%s)/%s?timeout=%fs", user, password, host, schema, timeout) db, _, err := sqlutils.GetDB(mysqlURI) if err != nil { @@ -23,7 +24,7 @@ func queryHost(host string, user string, password string, schema string, queries } for _, row := range resultData { output := []string{host} - if viewSourceSchema { + if printSchema { output = append(output, schema) } for _, rowCell := range row { @@ -36,35 +37,35 @@ func queryHost(host string, user string, password string, schema string, queries return nil } -func QuerySchemas(hosts []string, user string, password string, schemas []string, queries []string, maxConcurrency uint, - timeout float64, viewSourceSchema bool) (anyError error) { - concurrentHosts := make(chan bool, maxConcurrency) - completedHosts := make(chan bool) +// QueryHosts will issue concurrent queries on given list of hosts +func QueryHosts(hosts []string, user string, password string, + defaultSchema string, schemas []string, queries []string, + maxConcurrency uint, timeout float64, +) (anyError error) { + concurrentQueries := make(chan bool, maxConcurrency) + printSchema := len(schemas) > 0 + if len(schemas) == 0 { + schemas = []string{defaultSchema} + } var wg sync.WaitGroup for _, host := range hosts { - go func(host string) { - wg.Add(len(schemas)) - concurrentHosts <- true - // For each host, run all queries for the respective schema - for _, schema := range schemas { - go func(schema string) { - defer wg.Done() - if err := queryHost(host, user, password, schema, queries, timeout, viewSourceSchema); err != nil { - anyError = err - log.Printf("%s %s", host, err.Error()) - } - }(schema) - } - wg.Wait() - <-concurrentHosts - completedHosts <- true - }(host) + // For each host, run all queries for the respective schema + for _, schema := range schemas { + wg.Add(1) + go func(host, schema string) { + concurrentQueries <- true + defer func() { <-concurrentQueries }() + defer wg.Done() + if err := queryHost(host, user, password, schema, queries, timeout, printSchema); err != nil { + anyError = err + log.Printf("%s %s", host, err.Error()) + } + }(host, schema) + } } // Barrier. Wait for all to complete - for range hosts { - <-completedHosts - } + wg.Wait() return anyError } diff --git a/go/text/hosts.go b/go/text/hosts.go index da4299b..d6948d6 100644 --- a/go/text/hosts.go +++ b/go/text/hosts.go @@ -43,3 +43,13 @@ func ParseHosts(hostsList string, hostsFile string) (hosts []string, err error) return hosts, err } + +func SplitNonEmpty(s string, sep string) (result []string) { + tokens := strings.Split(s, sep) + for _, token := range tokens { + if token != "" { + result = append(result, strings.TrimSpace(token)) + } + } + return result +} diff --git a/go/text/hosts_test.go b/go/text/hosts_test.go index 7dafdbb..c292d3b 100644 --- a/go/text/hosts_test.go +++ b/go/text/hosts_test.go @@ -48,3 +48,17 @@ func TestParseHostsMulti(t *testing.T) { } } } + +func TestSplitNonEmpty(t *testing.T) { + s := "the, quick,, brown,fox ,," + splits := SplitNonEmpty(s, ",") + + if len(splits) != 4 { + t.Errorf("expected 4 tokens; got %+v", len(splits)) + } + join := strings.Join(splits, ";") + expected := "the;quick;brown;fox" + if join != expected { + t.Errorf("expected tokens: `%+v`. Got: `%+v`", expected, join) + } +} From 5610d89db75944ae095c6489061edeefe47cf7bb Mon Sep 17 00:00:00 2001 From: Shlomi Noach Date: Tue, 24 Oct 2017 14:29:58 +0300 Subject: [PATCH 10/11] Updated readme --- README.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 544e195..57cecbb 100644 --- a/README.md +++ b/README.md @@ -24,8 +24,6 @@ Usage of ccql: Query/queries input file -d string Default schema to use (default "information_schema") - -s string - Comma separated list of schemas, overrides '-d'. Implies printing schema name to output -h string Comma or space delimited list of hosts in hostname[:port] format. If not given, hosts read from stdin -m uint @@ -34,6 +32,8 @@ Usage of ccql: MySQL password -q string Query/queries to execute + -s string + List of databases to query from; overrides -d, prints schema name to output -t float Connect timeout seconds -u string From 59799c0ec3f9dac8abdf5272654aeb352958165a Mon Sep 17 00:00:00 2001 From: Shlomi Noach Date: Tue, 24 Oct 2017 14:34:09 +0300 Subject: [PATCH 11/11] readme: multiple schemas --- README.md | 24 ++++++++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/README.md b/README.md index 57cecbb..dd7b5c9 100644 --- a/README.md +++ b/README.md @@ -60,6 +60,16 @@ You may provide a query or a list of queries in the following ways: Queries are delimited by a semicolon (`;`). The last query may, but does not have to, be terminated by a semicolon. Quotes are respected, up to a reasonable level. It is valid to include a semicolon in a quoted text, as in `select 'single;query'`. However `ccql` does not employ a full blown parser, so please don't overdo it. For example, the following may not be parsed correctly: `select '\';\''`. You get it. +#### Schemas + +You may either provide: + +- An implicit, default schema via `-d schema_name` + - Schema name is not visible on output. +- Or explicit list of schemas via `-s "schema_1,schema_2[,schema_3...]"` (overrides `-d`) + - Queries are executed per host, per schema. + - Schema name printed as output column. + #### Credentials input You may provide credentials in the following ways: @@ -146,6 +156,20 @@ Set `sync_binlog=0` on all intermediate masters: cat /tmp/hosts.txt | ccql -q "show slave status;" | awk -F $'\t' '{print $3 ":" $5}' | sort | uniq | ccql -q "show slave status" | awk '{print $1}' | ccql -q "set global sync_binlog=0" ``` +Multiple schemas: + +```shell +$ cat /tmp/hosts.txt | ccql -t 0.5 -s "test,meta" -q "select uuid() from dual" | column -t +host3:3306 test d0d95311-b8ad-11e7-81e7-008cfa542442 +host2:3306 meta d0d95311-b8ad-11e7-a16c-a0369fb3dc94 +host2:3306 test d0d95fd6-b8ad-11e7-9a23-008cfa544064 +host1:3306 meta d0d95311-b8ad-11e7-9a15-a0369fb5fdd0 +host3:3306 meta d0d95311-b8ad-11e7-bd26-a0369fb5f3d8 +host4:3306 meta d0d95311-b8ad-11e7-a16c-a0369fb3dc94 +host1:3306 test d0d96924-b8ad-11e7-9bde-008cfa5440e4 +host4:3306 test d0d99a9d-b8ad-11e7-a680-008cfa542c9e +``` + ## LICENSE See [LICENSE](LICENSE). _ccql_ imports and includes 3rd party libraries, which have their own license. These are found under [vendor](vendor).