Add attack to Patroni PostgreSQL cluster (#229)

Signed-off-by: Nikita Savchenko <nikisavchenko@ozon.ru>

cmd/attack/patroni.go

@@ -0,0 +1,97 @@
+// Copyright 2020 Chaos Mesh Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package attack
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/spf13/cobra"
+	"go.uber.org/fx"
+
+	"github.com/chaos-mesh/chaosd/cmd/server"
+	"github.com/chaos-mesh/chaosd/pkg/core"
+	"github.com/chaos-mesh/chaosd/pkg/server/chaosd"
+	"github.com/chaos-mesh/chaosd/pkg/utils"
+)
+
+func NewPatroniAttackCommand(uid *string) *cobra.Command {
+	options := core.NewPatroniCommand()
+	dep := fx.Options(
+		server.Module,
+		fx.Provide(func() *core.PatroniCommand {
+			options.UID = *uid
+			return options
+		}),
+	)
+
+	cmd := &cobra.Command{
+		Use:   "patroni <subcommand>",
+		Short: "Patroni attack related commands",
+	}
+
+	cmd.AddCommand(
+		NewPatroniSwitchoverCommand(dep, options),
+		NewPatroniFailoverCommand(dep, options),
+	)
+
+	cmd.PersistentFlags().StringVarP(&options.User, "user", "u", "patroni", "patroni cluster user")
+	cmd.PersistentFlags().StringVar(&options.Password, "password", "p", "patroni cluster password")
+
+	return cmd
+}
+
+func NewPatroniSwitchoverCommand(dep fx.Option, options *core.PatroniCommand) *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "switchover",
+		Short: "exec switchover, default without another attack. Warning! Command is not recover!",
+		Run: func(*cobra.Command, []string) {
+			options.Action = core.SwitchoverAction
+			utils.FxNewAppWithoutLog(dep, fx.Invoke(PatroniAttackF)).Run()
+		},
+	}
+	cmd.Flags().StringVarP(&options.Address, "address", "a", "", "patroni cluster address, any of available hosts")
+	cmd.Flags().StringVarP(&options.Candidate, "candidate", "c", "", "switchover candidate, default random unit for replicas")
+	cmd.Flags().StringVarP(&options.Scheduled_at, "scheduled_at", "d", fmt.Sprintln(time.Now().Add(time.Second*60).Format(time.RFC3339)), "scheduled switchover, default now()+1 minute")
+
+	return cmd
+}
+
+func NewPatroniFailoverCommand(dep fx.Option, options *core.PatroniCommand) *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "failover",
+		Short: "exec failover, default without another attack",
+		Run: func(*cobra.Command, []string) {
+			options.Action = core.FailoverAction
+			utils.FxNewAppWithoutLog(dep, fx.Invoke(PatroniAttackF)).Run()
+		},
+	}
+
+	cmd.Flags().StringVarP(&options.Address, "address", "a", "", "patroni cluster address, any of available hosts")
+	cmd.Flags().StringVarP(&options.Candidate, "leader", "c", "", "failover new leader, default random unit for replicas")
+	return cmd
+}
+
+func PatroniAttackF(options *core.PatroniCommand, chaos *chaosd.Server) {
+	if err := options.Validate(); err != nil {
+		utils.ExitWithError(utils.ExitBadArgs, err)
+	}
+
+	uid, err := chaos.ExecuteAttack(chaosd.PatroniAttack, options, core.CommandMode)
+	if err != nil {
+		utils.ExitWithError(utils.ExitError, err)
+	}
+
+	utils.NormalExit(fmt.Sprintf("Attack %s successfully to patroni address %s, uid: %s", options.Action, options.Address, uid))
+}

go.mod

@@ -32,6 +32,7 @@ require (
 	github.com/swaggo/files v0.0.0-20220610200504-28940afbdbfe
 	github.com/swaggo/gin-swagger v1.5.0
 	github.com/swaggo/swag v1.8.3
+	github.com/tidwall/gjson v1.14.4
 	go.uber.org/fx v1.17.1
 	go.uber.org/zap v1.21.0
 	google.golang.org/grpc v1.40.0
@@ -122,6 +123,8 @@ require (
 	github.com/romana/ipset v1.0.0 // indirect
 	github.com/romana/rlog v0.0.0-20171115192701-f018bc92e7d7 // indirect
 	github.com/sirupsen/logrus v1.8.1 // indirect
+	github.com/tidwall/match v1.1.1 // indirect
+	github.com/tidwall/pretty v1.2.0 // indirect
 	github.com/tklauser/go-sysconf v0.3.10 // indirect
 	github.com/tklauser/numcpus v0.4.0 // indirect
 	github.com/ugorji/go/codec v1.2.7 // indirect

go.sum

@@ -1029,6 +1029,12 @@ github.com/syndtr/gocapability v0.0.0-20200815063812-42c35b437635/go.mod h1:hkRG
 github.com/syndtr/goleveldb v1.0.1-0.20200815110645-5c35d600f0ca/go.mod h1:u2MKkTVTVJWe5D1rCvame8WqhBd88EuIwODJZ1VHCPM=
 github.com/syndtr/goleveldb v1.0.1-0.20210305035536-64b5b1c73954/go.mod h1:u2MKkTVTVJWe5D1rCvame8WqhBd88EuIwODJZ1VHCPM=
 github.com/tchap/go-patricia v2.2.6+incompatible/go.mod h1:bmLyhP68RS6kStMGxByiQ23RP/odRBOTVjwp2cDyi6I=
+github.com/tidwall/gjson v1.14.4 h1:uo0p8EbA09J7RQaflQ1aBRffTR7xedD2bcIVSYxLnkM=
+github.com/tidwall/gjson v1.14.4/go.mod h1:/wbyibRr2FHMks5tjHJ5F8dMZh3AcwJEMf5vlfC0lxk=
+github.com/tidwall/match v1.1.1 h1:+Ho715JplO36QYgwN9PGYNhgZvoUSc9X2c80KVTi+GA=
+github.com/tidwall/match v1.1.1/go.mod h1:eRSPERbgtNPcGhD8UCthc6PmLEQXEWd3PRB5JTxsfmM=
+github.com/tidwall/pretty v1.2.0 h1:RWIZEg2iJ8/g6fDDYzMpobmaoGh5OLl4AXtGUGPcqCs=
+github.com/tidwall/pretty v1.2.0/go.mod h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhsoaGGjNU=
 github.com/tinylib/msgp v1.0.2/go.mod h1:+d+yLhGm8mzTaHzB+wgMYrodPfmZrzkirds8fDWklFE=
 github.com/tklauser/go-sysconf v0.3.10 h1:IJ1AZGZRWbY8T5Vfk04D9WOA5WSejdflXxP03OUqALw=
 github.com/tklauser/go-sysconf v0.3.10/go.mod h1:C8XykCvCb+Gn0oNCWPIlcb0RuglQTYaQ2hGm7jmxEFk=

pkg/core/experiment.go

@@ -43,6 +43,7 @@ const (
 	FileAttack        = "file"
 	HTTPAttack        = "http"
 	VMAttack          = "vm"
+	PatroniAttack     = "patroni"
 	UserDefinedAttack = "userDefined"
 )
 

pkg/core/patroni.go

@@ -0,0 +1,66 @@
+// Copyright 2020 Chaos Mesh Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package core
+
+import (
+	"encoding/json"
+
+	"github.com/pingcap/errors"
+)
+
+const (
+	SwitchoverAction = "switchover"
+	FailoverAction   = "failover"
+)
+
+var _ AttackConfig = &PatroniCommand{}
+
+type PatroniCommand struct {
+	CommonAttackConfig
+
+	Address      string `json:"address,omitempty"`
+	Candidate    string `json:"candidate,omitempty"`
+	Leader       string `json:"leader,omitempty"`
+	User         string `json:"user,omitempty"`
+	Password     string `json:"password,omitempty"`
+	Scheduled_at string `json:"scheduled_at,omitempty"`
+	RecoverCmd   string `json:"recoverCmd,omitempty"`
+}
+
+func (p *PatroniCommand) Validate() error {
+	if err := p.CommonAttackConfig.Validate(); err != nil {
+		return err
+	}
+	if len(p.Address) == 0 {
+		return errors.New("address not provided")
+	}
+
+	// TODO: validate signal
+
+	return nil
+}
+
+func (p PatroniCommand) RecoverData() string {
+	data, _ := json.Marshal(p)
+
+	return string(data)
+}
+
+func NewPatroniCommand() *PatroniCommand {
+	return &PatroniCommand{
+		CommonAttackConfig: CommonAttackConfig{
+			Kind: PatroniAttack,
+		},
+	}
+}

pkg/server/chaosd/patroni.go

@@ -0,0 +1,128 @@
+package chaosd
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io"
+	"math/rand"
+	"net/http"
+
+	"github.com/chaos-mesh/chaosd/pkg/core"
+	"github.com/chaos-mesh/chaosd/pkg/server/utils"
+	"github.com/pingcap/errors"
+	"github.com/pingcap/log"
+)
+
+type patroniAttack struct{}
+
+var PatroniAttack AttackType = patroniAttack{}
+
+func (patroniAttack) Attack(options core.AttackConfig, _ Environment) error {
+	attack := options.(*core.PatroniCommand)
+
+	candidate := attack.Candidate
+
+	leader := attack.Leader
+
+	var scheduled_at string
+
+	var url string
+
+	var availableReplicas []string
+
+	values := make(map[string]string)
+
+	patroniInfo, err := utils.GetPatroniInfo(attack.Address)
+	if err != nil {
+		err = errors.Errorf("failed to get patroni info for : %v", options.String(), err)
+		return errors.WithStack(err)
+	}
+
+	for _, replica := range patroniInfo.Replicas {
+		if replica != attack.Address {
+			availableReplicas = append(availableReplicas, replica)
+		}
+	}
+
+	if len(availableReplicas) == 0 {
+		err = errors.Errorf("failed to get available replics. Please, choose another host")
+		return errors.WithStack(err)
+	}
+
+	if candidate == "" {
+
+		candidate = availableReplicas[rand.Intn(len(availableReplicas))]
+
+	}
+
+	if leader == "" {
+		leader = patroniInfo.Master
+	}
+
+	switch options.String() {
+	case "switchover":
+
+		scheduled_at = attack.Scheduled_at
+
+		values = map[string]string{"leader": leader, "scheduled_at": scheduled_at}
+
+		log.Info(fmt.Sprintf("Switchover will be done from %v to another available replica in %v", patroniInfo.Master, scheduled_at))
+
+	case "failover":
+
+		values = map[string]string{"candidate": candidate}
+
+		log.Info(fmt.Sprintf("Failover will be done from %v to %v", patroniInfo.Master, candidate))
+
+	}
+
+	patroniAddr := attack.Address
+
+	cmd := options.String()
+
+	data, err := json.Marshal(values)
+	if err != nil {
+		err = errors.Errorf("failed to marshal data: %v", values)
+		return errors.WithStack(err)
+	}
+
+	url = fmt.Sprintf("http://%v:8008/%v", patroniAddr, cmd)
+
+	request, err := http.NewRequest("POST", url, bytes.NewBuffer(data))
+	if err != nil {
+		err = errors.Errorf("failed to %v: %v", cmd, err)
+		return errors.WithStack(err)
+	}
+
+	request.Header.Set("Content-Type", "application/json")
+	request.SetBasicAuth(attack.User, attack.Password)
+
+	client := &http.Client{}
+	resp, error := client.Do(request)
+	if error != nil {
+		err = errors.Errorf("failed to %v: %v", cmd, err)
+		return errors.WithStack(err)
+	}
+
+	defer resp.Body.Close()
+
+	buf, err := io.ReadAll(resp.Body)
+	if err != nil {
+		err = errors.Errorf("failed to read %v responce: %v", cmd, err)
+		return errors.WithStack(err)
+	}
+
+	if resp.StatusCode != 200 && resp.StatusCode != 202 {
+		err = errors.Errorf("failed to %v: status code %v, responce %v", cmd, resp.StatusCode, string(buf))
+		return errors.WithStack(err)
+	}
+
+	log.S().Infof("Execute %v successfully: %v", cmd, string(buf))
+
+	return nil
+}
+
+func (patroniAttack) Recover(exp core.Experiment, _ Environment) error {
+	return nil
+}

pkg/server/utils/status.go

@@ -0,0 +1,54 @@
+package utils
+
+import (
+	"fmt"
+	"io"
+	"net/http"
+
+	"github.com/pingcap/log"
+	"github.com/pkg/errors"
+	"github.com/tidwall/gjson"
+)
+
+type PatroniInfo struct {
+	Master   string
+	Replicas []string
+	Status   []string
+}
+
+func GetPatroniInfo(address string) (PatroniInfo, error) {
+	res, err := http.Get(fmt.Sprintf("http://%v:8008/cluster", address))
+	if err != nil {
+		err = errors.Errorf("failed to get patroni status: %v", err)
+		return PatroniInfo{}, errors.WithStack(err)
+	}
+
+	defer res.Body.Close()
+
+	buf, err := io.ReadAll(res.Body)
+	if err != nil {
+		err = errors.Errorf("failed to read responce: %v", err)
+		return PatroniInfo{}, errors.WithStack(err)
+	}
+
+	data := string(buf)
+
+	patroniInfo := PatroniInfo{}
+
+	members := gjson.Get(data, "members")
+
+	for _, member := range members.Array() {
+		if member.Get("role").Str == "leader" {
+			patroniInfo.Master = member.Get("name").Str
+			patroniInfo.Status = append(patroniInfo.Status, member.Get("state").Str)
+		} else if member.Get("role").Str == "replica" || member.Get("role").Str == "sync_standby" {
+			patroniInfo.Replicas = append(patroniInfo.Replicas, member.Get("name").Str)
+			patroniInfo.Status = append(patroniInfo.Status, member.Get("state").Str)
+		}
+	}
+
+	log.Info(fmt.Sprintf("patroni info: master %v, replicas %v, statuses %v\n", patroniInfo.Master, patroniInfo.Replicas, patroniInfo.Status))
+
+	return patroniInfo, nil
+
+}