prom query retry

alert/aconf/conf.go

@@ -60,10 +60,6 @@ func (a *Alert) PreCheck(configDir string) {
 		a.Heartbeat.Interval = 1000
 	}
 
-	if a.Heartbeat.EngineName == "" {
-		a.Heartbeat.EngineName = "default"
-	}
-
 	if a.EngineDelay == 0 {
 		a.EngineDelay = 30
 	}

alert/eval/eval.go

@@ -453,6 +453,10 @@ func GetAnomalyPoint(ruleId int64, ruleQuery models.RuleQuery, seriesTagIndexes
 		return points, recoverPoints
 	}
 
+	if len(seriesTagIndexes) == 0 {
+		return points, recoverPoints
+	}
+
 	for _, trigger := range ruleQuery.Triggers {
 		// seriesTagIndex 的 key 仅做分组使用，value 为每组 series 的 hash
 		seriesTagIndex := make(map[uint64][]uint64)

alert/process/process.go

@@ -170,7 +170,9 @@ func (p *Processor) Handle(anomalyPoints []common.AnomalyPoint, from string, inh
 		p.handleEvent(events)
 	}
 
-	p.HandleRecover(alertingKeys, now, inhibit)
+	if from == "inner" {
+		p.HandleRecover(alertingKeys, now, inhibit)
+	}
 }
 
 func (p *Processor) BuildEvent(anomalyPoint common.AnomalyPoint, from string, now int64) *models.AlertCurEvent {

alert/sender/ibex.go

@@ -185,6 +185,11 @@ func canDoIbex(username string, tpl *models.TaskTpl, host string, targetCache *m
 }
 
 func TaskAdd(f models.TaskForm, authUser string, isCenter bool) (int64, error) {
+	if storage.Cache == nil {
+		logger.Warning("event_callback_ibex: redis cache is nil")
+		return 0, fmt.Errorf("redis cache is nil")
+	}
+
 	hosts := cleanHosts(f.Hosts)
 	if len(hosts) == 0 {
 		return 0, fmt.Errorf("arg(hosts) empty")

models/migrate/migrate.go

@@ -28,7 +28,7 @@ func MigrateIbexTables(db *gorm.DB) {
 		db = db.Set("gorm:table_options", tableOptions)
 	}
 
-	dts := []interface{}{&imodels.TaskMeta{}, &imodels.TaskScheduler{}, &imodels.TaskSchedulerHealth{}, &imodels.TaskHostDoing{}, &imodels.TaskAction{}}
+	dts := []interface{}{&imodels.TaskMeta{}, &imodels.TaskScheduler{}, &imodels.TaskSchedulerHealth{}, &TaskHostDoing{}, &imodels.TaskAction{}}
 	for _, dt := range dts {
 		err := db.AutoMigrate(dt)
 		if err != nil {
@@ -280,3 +280,15 @@ type BuiltinPayloads struct {
 	UUID        int64 `json:"uuid" gorm:"type:bigint;not null;index:idx_uuid;comment:'uuid of payload'"`
 	ComponentID int64 `json:"component_id" gorm:"type:bigint;index:idx_component,sort:asc;not null;default:0;comment:'component_id of payload'"`
 }
+
+type TaskHostDoing struct {
+	Id             int64  `gorm:"column:id;index;primaryKey:false"`
+	Host           string `gorm:"column:host;size:128;not null;index"`
+	Clock          int64  `gorm:"column:clock;not null;default:0"`
+	Action         string `gorm:"column:action;size:16;not null"`
+	AlertTriggered bool   `gorm:"-"`
+}
+
+func (TaskHostDoing) TableName() string {
+	return "task_host_doing"
+}

models/target.go

@@ -121,7 +121,18 @@ func TargetDel(ctx *ctx.Context, idents []string) error {
 	if len(idents) == 0 {
 		panic("idents empty")
 	}
-	return DB(ctx).Where("ident in ?", idents).Delete(new(Target)).Error
+
+	return DB(ctx).Transaction(func(tx *gorm.DB) error {
+		txErr := tx.Where("ident in ?", idents).Delete(new(Target)).Error
+		if txErr != nil {
+			return txErr
+		}
+		txErr = TargetDeleteBgids(ctx, idents)
+		if txErr != nil {
+			return txErr
+		}
+		return nil
+	})
 }
 
 type BuildTargetWhereOption func(session *gorm.DB) *gorm.DB

pkg/oauth2x/oauth2x.go

@@ -208,7 +208,7 @@ func (s *SsoClient) exchangeUser(code string) (*CallbackOutput, error) {
 	if err != nil {
 		return nil, fmt.Errorf("failed to exchange token: %s", err)
 	}
-	userInfo, err := s.getUserInfo(s.UserInfoAddr, oauth2Token.AccessToken, s.TranTokenMethod)
+	userInfo, err := s.getUserInfo(s.Config.ClientID, s.UserInfoAddr, oauth2Token.AccessToken, s.TranTokenMethod)
 	if err != nil {
 		logger.Errorf("failed to get user info: %s", err)
 		return nil, fmt.Errorf("failed to get user info: %s", err)
@@ -223,10 +223,10 @@ func (s *SsoClient) exchangeUser(code string) (*CallbackOutput, error) {
 	}, nil
 }
 
-func (s *SsoClient) getUserInfo(UserInfoAddr, accessToken string, TranTokenMethod string) ([]byte, error) {
+func (s *SsoClient) getUserInfo(ClientId, UserInfoAddr, accessToken string, TranTokenMethod string) ([]byte, error) {
 	var req *http.Request
 	if TranTokenMethod == "formdata" {
-		body := bytes.NewBuffer([]byte("access_token=" + accessToken))
+		body := bytes.NewBuffer([]byte("access_token=" + accessToken + "&client_id=" + ClientId))
 		r, err := http.NewRequest("POST", UserInfoAddr, body)
 		if err != nil {
 			return nil, err
@@ -234,7 +234,8 @@ func (s *SsoClient) getUserInfo(UserInfoAddr, accessToken string, TranTokenMetho
 		r.Header.Add("Content-Type", "application/x-www-form-urlencoded")
 		req = r
 	} else if TranTokenMethod == "querystring" {
-		r, err := http.NewRequest("GET", UserInfoAddr+"?access_token="+accessToken, nil)
+
+		r, err := http.NewRequest("GET", UserInfoAddr+"?access_token="+accessToken+"&client_id="+ClientId, nil)
 		if err != nil {
 			return nil, err
 		}
@@ -246,6 +247,7 @@ func (s *SsoClient) getUserInfo(UserInfoAddr, accessToken string, TranTokenMetho
 			return nil, err
 		}
 		r.Header.Add("Authorization", "Bearer "+accessToken)
+		r.Header.Add("client_id", ClientId)
 		req = r
 	}
 
@@ -262,6 +264,7 @@ func (s *SsoClient) getUserInfo(UserInfoAddr, accessToken string, TranTokenMetho
 
 	body, err := ioutil.ReadAll(resp.Body)
 	resp.Body.Close()
+	logger.Debugf("getUserInfo req:%+v resp: %+v", req, string(body))
 	return body, err
 }
 

pkg/prom/reader.go

@@ -697,6 +697,17 @@ func (h *httpAPI) LabelValues(ctx context.Context, label string, matchs []string
 }
 
 func (h *httpAPI) Query(ctx context.Context, query string, ts time.Time) (model.Value, Warnings, error) {
+	for i := 0; i < 3; i++ {
+		value, warnings, err := h.query(ctx, query, ts)
+		if err == nil {
+			return value, warnings, nil
+		}
+		time.Sleep(1 * time.Second)
+	}
+	return nil, nil, errors.New("query failed")
+}
+
+func (h *httpAPI) query(ctx context.Context, query string, ts time.Time) (model.Value, Warnings, error) {
 	u := h.client.URL(epQuery, nil)
 	q := u.Query()