From a51b41507dc2dd178beed3c804575e124cd9f988 Mon Sep 17 00:00:00 2001 From: AnshumanTripathi Date: Fri, 15 Sep 2017 13:24:07 -0700 Subject: [PATCH 1/8] 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 2/8] 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 3/8] 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 4/8] 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 5/8] 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 6/8] 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 7/8] 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 8/8] 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)