-
Notifications
You must be signed in to change notification settings - Fork 2.1k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #1776 from michael-berlin/vtctl_throttler_vtctl_cmds
vtctl: Added throttler commands. vtworker: Enabled throttler RPC server.
- Loading branch information
Showing
14 changed files
with
427 additions
and
8 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,11 @@ | ||
| // Copyright 2016, Google Inc. All rights reserved. | ||
| // Use of this source code is governed by a BSD-style | ||
| // license that can be found in the LICENSE file. | ||
|
|
||
| package main | ||
|
|
||
| // Imports and register the gRPC throttler client. | ||
|
|
||
| import ( | ||
| _ "github.com/youtube/vitess/go/vt/throttler/grpcthrottlerclient" | ||
| ) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,11 @@ | ||
| // Copyright 2016, Google Inc. All rights reserved. | ||
| // Use of this source code is governed by a BSD-style | ||
| // license that can be found in the LICENSE file. | ||
|
|
||
| package main | ||
|
|
||
| // Imports and register the gRPC throttler client. | ||
|
|
||
| import ( | ||
| _ "github.com/youtube/vitess/go/vt/throttler/grpcthrottlerclient" | ||
| ) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,11 @@ | ||
| // Copyright 2016, Google Inc. All rights reserved. | ||
| // Use of this source code is governed by a BSD-style | ||
| // license that can be found in the LICENSE file. | ||
|
|
||
| package main | ||
|
|
||
| // Imports and register the gRPC throttler server. | ||
|
|
||
| import ( | ||
| _ "github.com/youtube/vitess/go/vt/throttler/grpcthrottlerserver" | ||
| ) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,132 @@ | ||
| // Copyright 2016, Google Inc. All rights reserved. | ||
| // Use of this source code is governed by a BSD-style | ||
| // license that can be found in the LICENSE file. | ||
|
|
||
| package vtctl | ||
|
|
||
| import ( | ||
| "flag" | ||
| "fmt" | ||
| "strconv" | ||
| "strings" | ||
| "time" | ||
|
|
||
| "github.com/olekukonko/tablewriter" | ||
| "github.com/youtube/vitess/go/vt/throttler" | ||
| "github.com/youtube/vitess/go/vt/throttler/throttlerclient" | ||
| "github.com/youtube/vitess/go/vt/wrangler" | ||
| "golang.org/x/net/context" | ||
| ) | ||
|
|
||
| // This file contains the commands to control the throttler which is used during | ||
| // resharding (vtworker) and by filtered replication (vttablet). | ||
|
|
||
| const throttlerGroupName = "Resharding Throttler" | ||
| const shortTimeout = 15 * time.Second | ||
|
|
||
| func init() { | ||
| addCommandGroup(throttlerGroupName) | ||
|
|
||
| addCommand(throttlerGroupName, command{ | ||
| "ThrottlerMaxRates", | ||
| commandThrottlerMaxRates, | ||
| "-server <vtworker or vttablet>", | ||
| "Returns the current max rate of all active resharding throttlers on the server."}) | ||
| addCommand(throttlerGroupName, command{ | ||
| "ThrottlerSetMaxRate", | ||
| commandThrottlerSetMaxRate, | ||
| "-server <vtworker or vttablet> <rate>", | ||
| "Sets the max rate for all active resharding throttlers on the server."}) | ||
| } | ||
|
|
||
| func commandThrottlerMaxRates(ctx context.Context, wr *wrangler.Wrangler, subFlags *flag.FlagSet, args []string) error { | ||
| server := subFlags.String("server", "", "vtworker or vttablet to connect to") | ||
| if err := subFlags.Parse(args); err != nil { | ||
| return err | ||
| } | ||
| if subFlags.NArg() != 0 { | ||
| return fmt.Errorf("the ThrottlerSetMaxRate command does not accept any positional parameters") | ||
| } | ||
|
|
||
| // Connect to the server. | ||
| ctx, cancel := context.WithTimeout(ctx, shortTimeout) | ||
| defer cancel() | ||
| client, err := throttlerclient.New(*server) | ||
| if err != nil { | ||
| return fmt.Errorf("error creating a throttler client for server '%v': %v", *server, err) | ||
| } | ||
| defer client.Close() | ||
|
|
||
| rates, err := client.MaxRates(ctx) | ||
| if err != nil { | ||
| return fmt.Errorf("failed to get the throttler rate from server '%v': %v", *server, err) | ||
| } | ||
|
|
||
| if len(rates) == 0 { | ||
| wr.Logger().Printf("There are no active throttlers on server '%v'.\n", *server) | ||
| return nil | ||
| } | ||
|
|
||
| table := tablewriter.NewWriter(loggerWriter{wr.Logger()}) | ||
| table.SetAutoFormatHeaders(false) | ||
| table.SetHeader([]string{"Name", "Rate"}) | ||
| for name, rate := range rates { | ||
| rateText := strconv.FormatInt(rate, 10) | ||
| if rate == throttler.MaxRateModuleDisabled { | ||
| rateText = "unlimited" | ||
| } | ||
| table.Append([]string{name, rateText}) | ||
| } | ||
| table.Render() | ||
| wr.Logger().Printf("%d active throttler(s) on server '%v'.\n", len(rates), *server) | ||
| return nil | ||
| } | ||
|
|
||
| func commandThrottlerSetMaxRate(ctx context.Context, wr *wrangler.Wrangler, subFlags *flag.FlagSet, args []string) error { | ||
| server := subFlags.String("server", "", "vtworker or vttablet to connect to") | ||
| if err := subFlags.Parse(args); err != nil { | ||
| return err | ||
| } | ||
| if subFlags.NArg() != 1 { | ||
| return fmt.Errorf("the <rate> argument is required for the ThrottlerSetMaxRate command") | ||
| } | ||
| var rate int64 | ||
| if strings.ToLower(subFlags.Arg(0)) == "unlimited" { | ||
| rate = throttler.MaxRateModuleDisabled | ||
| } else { | ||
| var err error | ||
| rate, err = strconv.ParseInt(subFlags.Arg(0), 0, 64) | ||
| if err != nil { | ||
| return fmt.Errorf("failed to parse rate '%v' as integer value: %v", subFlags.Arg(0), err) | ||
| } | ||
| } | ||
|
|
||
| // Connect to the server. | ||
| ctx, cancel := context.WithTimeout(ctx, shortTimeout) | ||
| defer cancel() | ||
| client, err := throttlerclient.New(*server) | ||
| if err != nil { | ||
| return fmt.Errorf("error creating a throttler client for server '%v': %v", *server, err) | ||
| } | ||
| defer client.Close() | ||
|
|
||
| names, err := client.SetMaxRate(ctx, rate) | ||
| if err != nil { | ||
| return fmt.Errorf("failed to set the throttler rate on server '%v': %v", *server, err) | ||
| } | ||
|
|
||
| if len(names) == 0 { | ||
| wr.Logger().Printf("ThrottlerSetMaxRate did nothing because server '%v' has no active throttlers.\n", *server) | ||
| return nil | ||
| } | ||
|
|
||
| table := tablewriter.NewWriter(loggerWriter{wr.Logger()}) | ||
| table.SetAutoFormatHeaders(false) | ||
| table.SetHeader([]string{"Name"}) | ||
| for _, name := range names { | ||
| table.Append([]string{name}) | ||
| } | ||
| table.Render() | ||
| wr.Logger().Printf("%d active throttler(s) on server '%v' were updated.\n", len(names), *server) | ||
| return nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,106 @@ | ||
| // Copyright 2016, Google Inc. All rights reserved. | ||
| // Use of this source code is governed by a BSD-style | ||
| // license that can be found in the LICENSE file. | ||
|
|
||
| package testlib | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "net" | ||
| "strings" | ||
| "testing" | ||
|
|
||
| "google.golang.org/grpc" | ||
|
|
||
| "github.com/youtube/vitess/go/vt/throttler" | ||
| "github.com/youtube/vitess/go/vt/throttler/grpcthrottlerserver" | ||
| "github.com/youtube/vitess/go/vt/zktopo/zktestserver" | ||
|
|
||
| // The test uses the gRPC throttler client and server implementations. | ||
| _ "github.com/youtube/vitess/go/vt/throttler/grpcthrottlerclient" | ||
| ) | ||
|
|
||
| // TestVtctlThrottlerCommands tests all vtctl commands from the | ||
| // "Resharding Throttler" group. | ||
| func TestVtctlThrottlerCommands(t *testing.T) { | ||
| // Run a throttler server using the default process throttle manager. | ||
| listener, err := net.Listen("tcp", ":0") | ||
| if err != nil { | ||
| t.Fatalf("Cannot listen: %v", err) | ||
| } | ||
| s := grpc.NewServer() | ||
| go s.Serve(listener) | ||
| grpcthrottlerserver.StartServer(s, throttler.GlobalManager) | ||
|
|
||
| addr := fmt.Sprintf("localhost:%v", listener.Addr().(*net.TCPAddr).Port) | ||
|
|
||
| ts := zktestserver.New(t, []string{"cell1", "cell2"}) | ||
| vp := NewVtctlPipe(t, ts) | ||
| defer vp.Close() | ||
|
|
||
| // Get and set rate commands do not fail when no throttler is registered. | ||
| { | ||
| got, err := vp.RunAndOutput([]string{"ThrottlerMaxRates", "-server", addr}) | ||
| if err != nil { | ||
| t.Fatalf("VtctlPipe.RunAndStreamOutput() failed: %v", err) | ||
| } | ||
| want := "no active throttlers" | ||
| if !strings.Contains(got, want) { | ||
| t.Fatalf("ThrottlerMaxRates() = %v, want substring = %v", got, want) | ||
| } | ||
| } | ||
|
|
||
| { | ||
| got, err := vp.RunAndOutput([]string{"ThrottlerSetMaxRate", "-server", addr, "23"}) | ||
| if err != nil { | ||
| t.Fatalf("VtctlPipe.RunAndStreamOutput() failed: %v", err) | ||
| } | ||
| want := "no active throttlers" | ||
| if !strings.Contains(got, want) { | ||
| t.Fatalf("ThrottlerSetMaxRate(23) = %v, want substring = %v", got, want) | ||
| } | ||
| } | ||
|
|
||
| // Add a throttler and check the commands again. | ||
| t1, err := throttler.NewThrottler("t1", "TPS", 1 /* threadCount */, 2323, throttler.ReplicationLagModuleDisabled) | ||
| if err != nil { | ||
| t.Fatal(err) | ||
| } | ||
| defer t1.Close() | ||
| // MaxRates() will return the initial rate. | ||
| expectRate(t, vp, addr, "2323") | ||
|
|
||
| // Disable the module by setting the rate to 'unlimited'. | ||
| setRate(t, vp, addr, "unlimited") | ||
| expectRate(t, vp, addr, "unlimited") | ||
|
|
||
| // Re-enable it by setting a limit. | ||
| setRate(t, vp, addr, "9999") | ||
| expectRate(t, vp, addr, "9999") | ||
| } | ||
|
|
||
| func setRate(t *testing.T, vp *VtctlPipe, addr, rateStr string) { | ||
| got, err := vp.RunAndOutput([]string{"ThrottlerSetMaxRate", "-server", addr, rateStr}) | ||
| if err != nil { | ||
| t.Fatalf("VtctlPipe.RunAndStreamOutput() failed: %v", err) | ||
| } | ||
| want := "t1" | ||
| if !strings.Contains(got, want) { | ||
| t.Fatalf("ThrottlerSetMaxRate(%v) = %v, want substring = %v", rateStr, got, want) | ||
| } | ||
| } | ||
|
|
||
| func expectRate(t *testing.T, vp *VtctlPipe, addr, rateStr string) { | ||
| got, err := vp.RunAndOutput([]string{"ThrottlerMaxRates", "-server", addr}) | ||
| if err != nil { | ||
| t.Fatalf("VtctlPipe.RunAndStreamOutput() failed: %v", err) | ||
| } | ||
| want := "1 active throttler" | ||
| if !strings.Contains(got, want) { | ||
| t.Fatalf("ThrottlerMaxRates() = %v, want substring = %v", got, want) | ||
| } | ||
| want2 := fmt.Sprintf("| %v |", rateStr) | ||
| if !strings.Contains(got, want2) { | ||
| t.Fatalf("ThrottlerMaxRates() = %v, want substring = %v", got, want2) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.