stability test: async emit annotations

tests/actions.go

@@ -26,13 +26,15 @@ import (
 	"path/filepath"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	_ "github.com/go-sql-driver/mysql"
 	"github.com/golang/glog"
 	pingcapErrors "github.com/pingcap/errors"
 	"github.com/pingcap/kvproto/pkg/metapb"
 	"github.com/pingcap/tidb-operator/tests/pkg/apimachinery"
+	"github.com/pingcap/tidb-operator/tests/pkg/webhook"
 	admissionV1beta1 "k8s.io/api/admissionregistration/v1beta1"
 	"k8s.io/api/apps/v1beta1"
 	batchv1 "k8s.io/api/batch/v1"
@@ -51,7 +53,6 @@ import (
 	"github.com/pingcap/tidb-operator/tests/pkg/blockwriter"
 	"github.com/pingcap/tidb-operator/tests/pkg/metrics"
 	"github.com/pingcap/tidb-operator/tests/pkg/util"
-	"github.com/pingcap/tidb-operator/tests/pkg/webhook"
 )
 
 const (
@@ -65,14 +66,27 @@ const (
 	NodeUnreachablePodReason = "NodeLost"
 )
 
-func NewOperatorActions(cli versioned.Interface, kubeCli kubernetes.Interface, pollInterval time.Duration, cfg *Config) OperatorActions {
-	return &operatorActions{
+func NewOperatorActions(cli versioned.Interface,
+	kubeCli kubernetes.Interface,
+	pollInterval time.Duration,
+	cfg *Config,
+	clusters []*TidbClusterConfig) OperatorActions {
+	oa := &operatorActions{
 		cli:          cli,
 		kubeCli:      kubeCli,
 		pdControl:    controller.NewDefaultPDControl(),
 		pollInterval: pollInterval,
 		cfg:          cfg,
 	}
+	oa.clusterEvents = make(map[string]*clusterEvent)
+	for _, c := range clusters {
+		oa.clusterEvents[c.String()] = &clusterEvent{
+			ns:          c.Namespace,
+			clusterName: c.ClusterName,
+			events:      make([]event, 0),
+		}
+	}
+	return oa
 }
 
 const (
@@ -142,14 +156,29 @@ type OperatorActions interface {
 	RegisterWebHookAndServiceOrDie(info *OperatorConfig)
 	CleanWebHookAndService(info *OperatorConfig) error
 	StartValidatingAdmissionWebhookServerOrDie(info *OperatorConfig)
+	EventWorker()
+	EmitEvent(info *TidbClusterConfig, msg string)
 }
 
 type operatorActions struct {
-	cli          versioned.Interface
-	kubeCli      kubernetes.Interface
-	pdControl    controller.PDControlInterface
-	pollInterval time.Duration
-	cfg          *Config
+	cli           versioned.Interface
+	kubeCli       kubernetes.Interface
+	pdControl     controller.PDControlInterface
+	pollInterval  time.Duration
+	cfg           *Config
+	clusterEvents map[string]*clusterEvent
+	lock          sync.Mutex
+}
+
+type clusterEvent struct {
+	ns          string
+	clusterName string
+	events      []event
+}
+
+type event struct {
+	message string
+	ts      int64
 }
 
 var _ = OperatorActions(&operatorActions{})
@@ -203,6 +232,10 @@ func (oi *OperatorConfig) ConfigTLS() *tls.Config {
 	}
 }
 
+func (tc *TidbClusterConfig) String() string {
+	return fmt.Sprintf("%s/%s", tc.Namespace, tc.ClusterName)
+}
+
 func (tc *TidbClusterConfig) BackupHelmSetString(m map[string]string) string {
 
 	set := map[string]string{
@@ -261,6 +294,7 @@ func (tc *TidbClusterConfig) TidbClusterHelmSetString(m map[string]string) strin
 func (oi *OperatorConfig) OperatorHelmSetString(m map[string]string) string {
 	set := map[string]string{
 		"operatorImage":                    oi.Image,
+		"imagePullPolicy":                  "Always",
 		"controllerManager.autoFailover":   "true",
 		"scheduler.kubeSchedulerImageName": oi.SchedulerImage,
 		"controllerManager.logLevel":       oi.LogLevel,
@@ -351,7 +385,7 @@ func (oa *operatorActions) UpgradeOperator(info *OperatorConfig) error {
 
 func (oa *operatorActions) DeployTidbCluster(info *TidbClusterConfig) error {
 	glog.Infof("deploying tidb cluster [%s/%s]", info.Namespace, info.ClusterName)
-	oa.emitEvent(info, "DeployTidbCluster")
+	oa.EmitEvent(info, "DeployTidbCluster")
 
 	namespace := &corev1.Namespace{
 		ObjectMeta: metav1.ObjectMeta{
@@ -391,7 +425,7 @@ func (oa *operatorActions) DeployTidbClusterOrDie(info *TidbClusterConfig) {
 
 func (oa *operatorActions) CleanTidbCluster(info *TidbClusterConfig) error {
 	glog.Infof("cleaning tidbcluster %s/%s", info.Namespace, info.ClusterName)
-	oa.emitEvent(info, "CleanTidbCluster")
+	oa.EmitEvent(info, "CleanTidbCluster")
 
 	charts := []string{
 		info.ClusterName,
@@ -534,7 +568,7 @@ func (oa *operatorActions) CheckTidbClusterStatusOrDie(info *TidbClusterConfig)
 }
 
 func (oa *operatorActions) BeginInsertDataTo(info *TidbClusterConfig) error {
-	oa.emitEvent(info, fmt.Sprintf("BeginInsertData: concurrency: %d", oa.cfg.BlockWriter.Concurrency))
+	oa.EmitEvent(info, fmt.Sprintf("BeginInsertData: concurrency: %d", oa.cfg.BlockWriter.Concurrency))
 
 	dsn := getDSN(info.Namespace, info.ClusterName, "test", info.Password)
 	if info.blockWriter == nil {
@@ -558,7 +592,7 @@ func (oa *operatorActions) BeginInsertDataToOrDie(info *TidbClusterConfig) {
 }
 
 func (oa *operatorActions) StopInsertDataTo(info *TidbClusterConfig) {
-	oa.emitEvent(info, "StopInsertData")
+	oa.EmitEvent(info, "StopInsertData")
 
 	info.blockWriter.Stop()
 }
@@ -580,7 +614,8 @@ func (oa *operatorActions) backupChartPath(tag string) string {
 }
 
 func (oa *operatorActions) ScaleTidbCluster(info *TidbClusterConfig) error {
-	oa.emitEvent(info, fmt.Sprintf("ScaleTidbCluster"))
+	oa.EmitEvent(info, fmt.Sprintf("ScaleTidbCluster to pd: %s, tikv: %s, tidb: %s",
+		info.Args["pd.replicas"], info.Args["tikv.replicas"], info.Args["tidb.replicas"]))
 
 	cmd := fmt.Sprintf("helm upgrade %s %s --set-string %s",
 		info.ClusterName, oa.tidbClusterChartPath(info.OperatorTag), info.TidbClusterHelmSetString(nil))
@@ -661,7 +696,7 @@ func (oa *operatorActions) UpgradeTidbCluster(info *TidbClusterConfig) error {
 	if err != nil {
 		return err
 	}
-	oa.emitEvent(info, "UpgradeTidbCluster")
+	oa.EmitEvent(info, "UpgradeTidbCluster")
 
 	cmd := fmt.Sprintf("helm upgrade %s %s --set-string %s",
 		info.ClusterName, oa.tidbClusterChartPath(info.OperatorTag), info.TidbClusterHelmSetString(nil))
@@ -1445,7 +1480,7 @@ func (oa *operatorActions) checkoutTag(tagName string) error {
 }
 
 func (oa *operatorActions) DeployAdHocBackup(info *TidbClusterConfig) error {
-	oa.emitEvent(info, "DeployAdHocBackup")
+	oa.EmitEvent(info, "DeployAdHocBackup")
 	glog.Infof("begin to deploy adhoc backup cluster[%s] namespace[%s]", info.ClusterName, info.Namespace)
 
 	sets := map[string]string{
@@ -1497,7 +1532,7 @@ func (oa *operatorActions) CheckAdHocBackup(info *TidbClusterConfig) error {
 }
 
 func (oa *operatorActions) Restore(from *TidbClusterConfig, to *TidbClusterConfig) error {
-	oa.emitEvent(to, fmt.Sprintf("RestoreBackup: source: %s", from.ClusterName))
+	oa.EmitEvent(from, fmt.Sprintf("RestoreBackup: source: %s", from.ClusterName))
 	glog.Infof("deploying restore cluster[%s/%s]", from.Namespace, from.ClusterName)
 
 	sets := map[string]string{
@@ -1642,7 +1677,7 @@ func releaseIsExist(err error) bool {
 }
 
 func (oa *operatorActions) DeployScheduledBackup(info *TidbClusterConfig) error {
-	oa.emitEvent(info, "DeploySchedulerBackup")
+	oa.EmitEvent(info, "DeploySchedulerBackup")
 	glog.Infof("begin to deploy scheduled backup")
 
 	cron := fmt.Sprintf("'*/1 * * * *'")
@@ -1863,7 +1898,7 @@ func (info *TidbClusterConfig) FullName() string {
 }
 
 func (oa *operatorActions) DeployIncrementalBackup(from *TidbClusterConfig, to *TidbClusterConfig) error {
-	oa.emitEvent(from, fmt.Sprintf("DeployIncrementalBackup: slave: %s", to.ClusterName))
+	oa.EmitEvent(from, fmt.Sprintf("DeployIncrementalBackup: slave: %s", to.ClusterName))
 	glog.Infof("begin to deploy incremental backup cluster[%s] namespace[%s]", from.ClusterName, from.Namespace)
 
 	sets := map[string]string{
@@ -2110,30 +2145,71 @@ func (oa *operatorActions) StartValidatingAdmissionWebhookServerOrDie(info *Oper
 	err = server.ListenAndServeTLS("", "")
 	if err != nil {
 		glog.Errorf("fail to start webhook server err %v", err)
+		// TODO use context instead
 		os.Exit(4)
 	}
 }
 
-func (oa *operatorActions) emitEvent(info *TidbClusterConfig, event string) {
-	if info.GrafanaClient == nil {
-		glog.V(4).Infof("cluster:[%s] grafana client not ready, skip recording event %s.",
-			info.ClusterName, event)
+func (oa *operatorActions) EmitEvent(info *TidbClusterConfig, message string) {
+	oa.lock.Lock()
+	defer oa.lock.Unlock()
+
+	ev := event{
+		message: message,
+		ts:      time.Now().UnixNano() / int64(time.Millisecond),
+	}
+
+	if info == nil {
+		for k, _ := range oa.clusterEvents {
+			ce := oa.clusterEvents[k]
+			ce.events = append(ce.events, ev)
+		}
 		return
 	}
 
-	anno := metrics.Annotation{
-		Text:                event,
-		TimestampInMilliSec: time.Now().UnixNano() / int64(time.Millisecond),
+	ce := oa.clusterEvents[info.String()]
+	ce.events = append(ce.events, ev)
 
-		Tags: []string{
-			statbilityTestTag,
-			fmt.Sprintf("cluster-%s", info.ClusterName),
-			fmt.Sprintf("ns-%s", info.Namespace),
-		},
-	}
-	go func(anno metrics.Annotation) {
-		if err := info.GrafanaClient.AddAnnotation(anno); err != nil {
-			glog.Errorf("cluster:[%s] error recording event %s, reason: %v", info.ClusterName, event, err)
+	// sleep a while to avoid overlapping time
+	time.Sleep(10 * time.Second)
+}
+
+func (oa *operatorActions) EventWorker() {
+	oa.lock.Lock()
+	defer oa.lock.Unlock()
+
+	for key, clusterEv := range oa.clusterEvents {
+		retryEvents := make([]event, 0)
+		for _, ev := range clusterEv.events {
+			ns := clusterEv.ns
+			clusterName := clusterEv.clusterName
+			grafanaUrl := fmt.Sprintf("http://%s-grafana.%s:3000", clusterName, ns)
+			client, err := metrics.NewClient(grafanaUrl, grafanaUsername, grafanaPassword, metricsPort)
+			if err != nil {
+				retryEvents = append(retryEvents, ev)
+				glog.V(4).Infof("failed to new grafana client: [%s/%s], %v", ns, clusterName, err)
+				continue
+			}
+
+			anno := metrics.Annotation{
+				Text:                ev.message,
+				TimestampInMilliSec: ev.ts,
+				Tags: []string{
+					statbilityTestTag,
+					fmt.Sprintf("clusterName: %s", clusterName),
+					fmt.Sprintf("namespace: %s", ns),
+				},
+			}
+			if err := client.AddAnnotation(anno); err != nil {
+				glog.V(4).Infof("cluster:[%s/%s] error recording event: %s, reason: %v",
+					ns, clusterName, ev.message, err)
+				retryEvents = append(retryEvents, ev)
+				continue
+			}
+			glog.Infof("cluster: [%s/%s] recoding event: %s", ns, clusterName, ev.message)
 		}
-	}(anno)
+
+		ce := oa.clusterEvents[key]
+		ce.events = retryEvents
+	}
 }

tests/backup/backupcase.go

@@ -40,13 +40,6 @@ func (bc *BackupCase) Run() error {
 
 	// pause write pressure during backup
 	bc.operator.StopInsertDataTo(bc.srcCluster)
-	defer func() {
-		go func() {
-			if err := bc.operator.BeginInsertDataTo(bc.srcCluster); err != nil {
-				glog.Errorf("cluster:[%s] begin insert data failed,error: %v", bc.srcCluster.ClusterName, err)
-			}
-		}()
-	}()
 
 	err := bc.operator.DeployAdHocBackup(bc.srcCluster)
 	if err != nil {
@@ -78,6 +71,8 @@ func (bc *BackupCase) Run() error {
 		return err
 	}
 
+	go bc.operator.BeginInsertDataToOrDie(bc.srcCluster)
+
 	err = bc.operator.DeployScheduledBackup(bc.srcCluster)
 	if err != nil {
 		glog.Errorf("cluster:[%s] scheduler happen error: %v", bc.srcCluster.ClusterName, err)