Add support for pt-kill. Fixes #34

Author: AMecea

hack/docker/mysql-helper/docker-entrypoint.sh

@@ -60,6 +60,7 @@ case "$1" in
     *)
         echo "Usage: $0 {files-config|clone|config-and-serve}"
         echo "Now runs your command."
+        echo "$@"
 
         exec "$@"
 esac

pkg/apis/mysql/v1alpha1/cluster.go

@@ -19,6 +19,7 @@ package v1alpha1
 import (
 	"fmt"
 	"strconv"
+	"strings"
 
 	"github.com/golang/glog"
 	apiv1 "k8s.io/api/core/v1"
@@ -275,3 +276,43 @@ func (c *MysqlCluster) GetLabels() map[string]string {
 		"mysql_cluster": c.Name,
 	}
 }
+
+func (ql *QueryLimits) GetOptions() []string {
+	options := []string{
+		"--print",
+	}
+	if ql.MaxIdleTime != nil {
+		options = append(options, "--idle-time", fmt.Sprintf("%d", *ql.MaxIdleTime))
+	}
+
+	if ql.MaxQueryTime != nil {
+		options = append(options, "--busy-time", fmt.Sprintf("%d", *ql.MaxQueryTime))
+	}
+
+	if ql.Kill != nil {
+		options = append(options, "--victims", *ql.Kill)
+	}
+
+	switch ql.KillMode {
+	case "connection":
+		options = append(options, "--kill")
+	case "query":
+		options = append(options, "--kill-query")
+	default:
+		options = append(options, "--kill-query")
+	}
+
+	if len(ql.IgnoreDb) > 0 {
+		options = append(options, "--ignore-db", strings.Join(ql.IgnoreDb, "|"))
+	}
+
+	if len(ql.IgnoreCommand) > 0 {
+		options = append(options, "--ignore-command", strings.Join(ql.IgnoreCommand, "|"))
+	}
+
+	if len(ql.IgnoreUser) > 0 {
+		options = append(options, "--ignore-user", strings.Join(ql.IgnoreUser, "|"))
+	}
+
+	return options
+}

pkg/apis/mysql/v1alpha1/types.go

@@ -85,13 +85,17 @@ type ClusterSpec struct {
 
 	// PVC extra specifiaction
 	// +optional
-	VolumeSpec `json:"volumeSpec,omitempty"`
+	VolumeSpec VolumeSpec `json:"volumeSpec,omitempty"`
 
 	// MaxSlaveLatency represents the allowed latency for a slave node in
 	// seconds. If set then the node with a latency grater than this is removed
 	// from service.
 	// +optional
 	MaxSlaveLatency *int64 `json:"maxSlaveLatency,omitempty"`
+
+	// QueryLimits represents limits for a query
+	// +optional
+	QueryLimits *QueryLimits `json:"queryLimits,omitempty"`
 }
 
 type MysqlConf map[string]string
@@ -161,6 +165,40 @@ type VolumeSpec struct {
 	core.PersistentVolumeClaimSpec `json:",inline"`
 }
 
+type QueryLimits struct {
+	// MaxIdleTime match queries that have benn idle for longer then this time.
+	// (--idle-time flag)
+	// + optional
+	MaxIdleTime *int `json:"maxIdleTime,omitempty"`
+
+	// MaxQueryTime match queries that have been running for longer then this
+	// time. (--busy-time flag)
+	// +optional
+	MaxQueryTime *int `json:"maxQueryTime,omitempty"`
+
+	// Kill represents the mode of which the matching queries in each class will
+	// be killed, (the --victims flag). Can be one of oldest|all|all-but-oldest
+	// +optional
+	Kill *string `json:"kill,omitempty"`
+
+	// KillMode can be: `connection` or `query`, when it's used `connection`
+	// means that when a query is matched the connection is killed (using --kill
+	// flag) and if it's used `query` means that the query is killed (using
+	// --kill-query flag)
+	KillMode string `json:"killMode,omitempty"`
+
+	// IgnoreDb is the lost of tatabase that are ignored by pt-kill (--ignore-db
+	// flag)
+	// +optional
+	IgnoreDb []string `json:"ignoreDb,omitempty"`
+
+	// IgnoreCommands
+	IgnoreCommand []string `json:"ignoreCommands,omitempty"`
+
+	//IgnoreUser
+	IgnoreUser []string `json:"ignoreUser,omitempty"`
+}
+
 // +genclient
 // +k8s:openapi-gen=true
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object

pkg/apis/mysql/v1alpha1/zz_generated.deepcopy.go

@@ -119,6 +119,15 @@ func (in *ClusterSpec) DeepCopyInto(out *ClusterSpec) {
 			**out = **in
 		}
 	}
+	if in.QueryLimits != nil {
+		in, out := &in.QueryLimits, &out.QueryLimits
+		if *in == nil {
+			*out = nil
+		} else {
+			*out = new(QueryLimits)
+			(*in).DeepCopyInto(*out)
+		}
+	}
 	return
 }
 
@@ -372,6 +381,64 @@ func (in *PodSpec) DeepCopy() *PodSpec {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *QueryLimits) DeepCopyInto(out *QueryLimits) {
+	*out = *in
+	if in.MaxIdleTime != nil {
+		in, out := &in.MaxIdleTime, &out.MaxIdleTime
+		if *in == nil {
+			*out = nil
+		} else {
+			*out = new(int)
+			**out = **in
+		}
+	}
+	if in.MaxQueryTime != nil {
+		in, out := &in.MaxQueryTime, &out.MaxQueryTime
+		if *in == nil {
+			*out = nil
+		} else {
+			*out = new(int)
+			**out = **in
+		}
+	}
+	if in.Kill != nil {
+		in, out := &in.Kill, &out.Kill
+		if *in == nil {
+			*out = nil
+		} else {
+			*out = new(string)
+			**out = **in
+		}
+	}
+	if in.IgnoreDb != nil {
+		in, out := &in.IgnoreDb, &out.IgnoreDb
+		*out = make([]string, len(*in))
+		copy(*out, *in)
+	}
+	if in.IgnoreCommand != nil {
+		in, out := &in.IgnoreCommand, &out.IgnoreCommand
+		*out = make([]string, len(*in))
+		copy(*out, *in)
+	}
+	if in.IgnoreUser != nil {
+		in, out := &in.IgnoreUser, &out.IgnoreUser
+		*out = make([]string, len(*in))
+		copy(*out, *in)
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new QueryLimits.
+func (in *QueryLimits) DeepCopy() *QueryLimits {
+	if in == nil {
+		return nil
+	}
+	out := new(QueryLimits)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *VolumeSpec) DeepCopyInto(out *VolumeSpec) {
 	*out = *in

pkg/mysqlcluster/statefullset.go

@@ -116,6 +116,7 @@ const (
 	containerMysqlName     = "mysql"
 	containerExporterName  = "metrics-exporter"
 	containerHeartBeatName = "pt-heartbeat"
+	containerKillerName    = "pt-kill"
 )
 
 func (f *cFactory) ensureContainer(in core.Container, name, image string, args []string) core.Container {
@@ -287,6 +288,10 @@ func (f *cFactory) ensureInitContainersSpec(in []core.Container) []core.Containe
 
 func (f *cFactory) ensureContainersSpec(in []core.Container) []core.Container {
 	noContainers := 4
+	if f.cluster.Spec.QueryLimits != nil {
+		noContainers += 1
+	}
+
 	if len(in) != noContainers {
 		in = make([]core.Container, noContainers)
 	}
@@ -382,6 +387,23 @@ func (f *cFactory) ensureContainersSpec(in []core.Container) []core.Container {
 
 	in[3] = heartbeat
 
+	if f.cluster.Spec.QueryLimits != nil {
+		command := []string{
+			"pt-kill",
+			"--wait-after-kill=1",
+			// host need to be specified, see pt-kill bug: https://jira.percona.com/browse/PT-1223
+			"--host=127.0.0.1",
+			fmt.Sprintf("--defaults-file=%s/client.cnf", ConfVolumeMountPath),
+		}
+		command = append(command, f.cluster.Spec.QueryLimits.GetOptions()...)
+
+		killer := f.ensureContainer(in[4], containerKillerName,
+			f.cluster.Spec.GetHelperImage(),
+			command,
+		)
+		in[4] = killer
+	}
+
 	return in
 }
 
@@ -477,7 +499,7 @@ func (f *cFactory) getVolumeMountsFor(name string) []core.VolumeMount {
 			},
 		}
 
-	case containerHeartBeatName:
+	case containerHeartBeatName, containerKillerName:
 		return []core.VolumeMount{
 			core.VolumeMount{
 				Name:      confVolumeName,

pkg/openapi/openapi_generated.go

@@ -41,6 +41,7 @@ func GetOpenAPIDefinitions(ref common.ReferenceCallback) map[string]common.OpenA
 		"github.com/presslabs/mysql-operator/pkg/apis/mysql/v1alpha1.NodeCondition":    schema_pkg_apis_mysql_v1alpha1_NodeCondition(ref),
 		"github.com/presslabs/mysql-operator/pkg/apis/mysql/v1alpha1.NodeStatus":       schema_pkg_apis_mysql_v1alpha1_NodeStatus(ref),
 		"github.com/presslabs/mysql-operator/pkg/apis/mysql/v1alpha1.PodSpec":          schema_pkg_apis_mysql_v1alpha1_PodSpec(ref),
+		"github.com/presslabs/mysql-operator/pkg/apis/mysql/v1alpha1.QueryLimits":      schema_pkg_apis_mysql_v1alpha1_QueryLimits(ref),
 		"github.com/presslabs/mysql-operator/pkg/apis/mysql/v1alpha1.VolumeSpec":       schema_pkg_apis_mysql_v1alpha1_VolumeSpec(ref),
 		"k8s.io/apimachinery/pkg/apis/meta/v1.APIGroup":                                schema_pkg_apis_meta_v1_APIGroup(ref),
 		"k8s.io/apimachinery/pkg/apis/meta/v1.APIGroupList":                            schema_pkg_apis_meta_v1_APIGroupList(ref),
@@ -343,11 +344,17 @@ func schema_pkg_apis_mysql_v1alpha1_ClusterSpec(ref common.ReferenceCallback) co
 							Format:      "int64",
 						},
 					},
+					"queryLimits": {
+						SchemaProps: spec.SchemaProps{
+							Description: "QueryLimits represents limits for a query",
+							Ref:         ref("github.com/presslabs/mysql-operator/pkg/apis/mysql/v1alpha1.QueryLimits"),
+						},
+					},
 				},
 			},
 		},
 		Dependencies: []string{
-			"github.com/presslabs/mysql-operator/pkg/apis/mysql/v1alpha1.PodSpec", "github.com/presslabs/mysql-operator/pkg/apis/mysql/v1alpha1.VolumeSpec"},
+			"github.com/presslabs/mysql-operator/pkg/apis/mysql/v1alpha1.PodSpec", "github.com/presslabs/mysql-operator/pkg/apis/mysql/v1alpha1.QueryLimits", "github.com/presslabs/mysql-operator/pkg/apis/mysql/v1alpha1.VolumeSpec"},
 	}
 }
 
@@ -669,6 +676,88 @@ func schema_pkg_apis_mysql_v1alpha1_PodSpec(ref common.ReferenceCallback) common
 	}
 }
 
+func schema_pkg_apis_mysql_v1alpha1_QueryLimits(ref common.ReferenceCallback) common.OpenAPIDefinition {
+	return common.OpenAPIDefinition{
+		Schema: spec.Schema{
+			SchemaProps: spec.SchemaProps{
+				Properties: map[string]spec.Schema{
+					"maxIdleTime": {
+						SchemaProps: spec.SchemaProps{
+							Description: "MaxIdleTime match queries that have benn idle for longer then this time. (--idle-time flag)",
+							Type:        []string{"integer"},
+							Format:      "int32",
+						},
+					},
+					"maxQueryTime": {
+						SchemaProps: spec.SchemaProps{
+							Description: "MaxQueryTime match queries that have been running for longer then this time. (--busy-time flag)",
+							Type:        []string{"integer"},
+							Format:      "int32",
+						},
+					},
+					"kill": {
+						SchemaProps: spec.SchemaProps{
+							Description: "Kill represents the mode of which the matching queries in each class will be killed, (the --victims flag). Can be one of oldest|all|all-but-oldest",
+							Type:        []string{"string"},
+							Format:      "",
+						},
+					},
+					"killMode": {
+						SchemaProps: spec.SchemaProps{
+							Description: "KillMode can be: `connection` or `query`, when it's used `connection` means that when a query is matched the connection is killed (using --kill flag) and if it's used `query` means that the query is killed (using --kill-query flag)",
+							Type:        []string{"string"},
+							Format:      "",
+						},
+					},
+					"ignoreDb": {
+						SchemaProps: spec.SchemaProps{
+							Description: "IgnoreDb is the lost of tatabase that are ignored by pt-kill (--ignore-db flag)",
+							Type:        []string{"array"},
+							Items: &spec.SchemaOrArray{
+								Schema: &spec.Schema{
+									SchemaProps: spec.SchemaProps{
+										Type:   []string{"string"},
+										Format: "",
+									},
+								},
+							},
+						},
+					},
+					"ignoreCommands": {
+						SchemaProps: spec.SchemaProps{
+							Description: "IgnoreCommands",
+							Type:        []string{"array"},
+							Items: &spec.SchemaOrArray{
+								Schema: &spec.Schema{
+									SchemaProps: spec.SchemaProps{
+										Type:   []string{"string"},
+										Format: "",
+									},
+								},
+							},
+						},
+					},
+					"ignoreUser": {
+						SchemaProps: spec.SchemaProps{
+							Description: "IgnoreUser",
+							Type:        []string{"array"},
+							Items: &spec.SchemaOrArray{
+								Schema: &spec.Schema{
+									SchemaProps: spec.SchemaProps{
+										Type:   []string{"string"},
+										Format: "",
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+		Dependencies: []string{},
+	}
+}
+
 func schema_pkg_apis_mysql_v1alpha1_VolumeSpec(ref common.ReferenceCallback) common.OpenAPIDefinition {
 	return common.OpenAPIDefinition{
 		Schema: spec.Schema{