feat: support webhook proxy configuration

alert/dispatch/dispatch.go

@@ -293,9 +293,9 @@ func (e *Dispatch) Send(rule *models.AlertRule, event *models.AlertCurEvent, not
 	// handle global webhooks
 	if !event.OverrideGlobalWebhook() {
 		if e.alerting.WebhookBatchSend {
-			sender.BatchSendWebhooks(e.ctx, notifyTarget.ToWebhookMap(), event, e.Astats)
+			sender.BatchSendWebhooks(e.ctx, notifyTarget.ToWebhookList(), event, e.Astats)
 		} else {
-			sender.SingleSendWebhooks(e.ctx, notifyTarget.ToWebhookMap(), event, e.Astats)
+			sender.SingleSendWebhooks(e.ctx, notifyTarget.ToWebhookList(), event, e.Astats)
 		}
 	}
 

alert/dispatch/notify_target.go

@@ -76,8 +76,52 @@ func (s *NotifyTarget) ToCallbackList() []string {
 	return callbacks
 }
 
+func (s *NotifyTarget) ToWebhookList() []*models.Webhook {
+	webhooks := make([]*models.Webhook, 0, len(s.webhooks))
+	for _, wh := range s.webhooks {
+		if wh.Batch == 0 {
+			wh.Batch = 1000
+		}
+
+		if wh.Timeout == 0 {
+			wh.Timeout = 10
+		}
+
+		if wh.RetryCount == 0 {
+			wh.RetryCount = 10
+		}
+
+		if wh.RetryInterval == 0 {
+			wh.RetryInterval = 10
+		}
+
+		webhooks = append(webhooks, wh)
+	}
+	return webhooks
+}
+
 func (s *NotifyTarget) ToWebhookMap() map[string]*models.Webhook {
-	return s.webhooks
+	webhookMap := make(map[string]*models.Webhook, len(s.webhooks))
+	for _, wh := range s.webhooks {
+		if wh.Batch == 0 {
+			wh.Batch = 1000
+		}
+
+		if wh.Timeout == 0 {
+			wh.Timeout = 10
+		}
+
+		if wh.RetryCount == 0 {
+			wh.RetryCount = 10
+		}
+
+		if wh.RetryInterval == 0 {
+			wh.RetryInterval = 10
+		}
+
+		webhookMap[wh.Url] = wh
+	}
+	return webhookMap
 }
 
 func (s *NotifyTarget) ToUidList() []int64 {

alert/sender/webhook.go

@@ -13,6 +13,7 @@ import (
 	"github.com/ccfos/nightingale/v6/alert/astats"
 	"github.com/ccfos/nightingale/v6/models"
 	"github.com/ccfos/nightingale/v6/pkg/ctx"
+	"github.com/ccfos/nightingale/v6/pkg/poster"
 
 	"github.com/toolkits/pkg/logger"
 )
@@ -60,20 +61,22 @@ func sendWebhook(webhook *models.Webhook, event interface{}, stats *astats.Stats
 		insecureSkipVerify = webhook.SkipVerify
 	}
 
-	if conf.Client == nil {
-		logger.Warningf("event_%s, event:%s, url: [%s], error: [%s]", channel, string(bs), conf.Url, "client is nil")
-		conf.Client = &http.Client{
-			Timeout: time.Duration(conf.Timeout) * time.Second,
-			Transport: &http.Transport{
-				TLSClientConfig: &tls.Config{InsecureSkipVerify: insecureSkipVerify},
-			},
-		}
+	transport := &http.Transport{
+		TLSClientConfig: &tls.Config{InsecureSkipVerify: insecureSkipVerify},
+	}
+	if poster.UseProxy(conf.Url) {
+		transport.Proxy = http.ProxyFromEnvironment
+	}
+
+	client := http.Client{
+		Timeout:   time.Duration(conf.Timeout) * time.Second,
+		Transport: transport,
 	}
 
 	stats.AlertNotifyTotal.WithLabelValues(channel).Inc()
 	var resp *http.Response
 	var body []byte
-	resp, err = conf.Client.Do(req)
+	resp, err = client.Do(req)
 
 	if err != nil {
 		stats.AlertNotifyErrorTotal.WithLabelValues(channel).Inc()
@@ -95,7 +98,7 @@ func sendWebhook(webhook *models.Webhook, event interface{}, stats *astats.Stats
 	return false, string(body), nil
 }
 
-func SingleSendWebhooks(ctx *ctx.Context, webhooks map[string]*models.Webhook, event *models.AlertCurEvent, stats *astats.Stats) {
+func SingleSendWebhooks(ctx *ctx.Context, webhooks []*models.Webhook, event *models.AlertCurEvent, stats *astats.Stats) {
 	for _, conf := range webhooks {
 		retryCount := 0
 		for retryCount < 3 {
@@ -110,7 +113,7 @@ func SingleSendWebhooks(ctx *ctx.Context, webhooks map[string]*models.Webhook, e
 	}
 }
 
-func BatchSendWebhooks(ctx *ctx.Context, webhooks map[string]*models.Webhook, event *models.AlertCurEvent, stats *astats.Stats) {
+func BatchSendWebhooks(ctx *ctx.Context, webhooks []*models.Webhook, event *models.AlertCurEvent, stats *astats.Stats) {
 	for _, conf := range webhooks {
 		logger.Infof("push event:%+v to queue:%v", event, conf)
 		PushEvent(ctx, conf, event, stats)

etc/config.toml

@@ -138,9 +138,6 @@ ForceUseServerTS = true
 # [Pushgw.WriterOpt]
 # QueueMaxSize = 1000000
 # QueuePopSize = 1000
-# AllQueueMaxSize = 1000000
-# fresh time, unit ms
-# AllQueueMaxSizeInterval = 200
 
 [[Pushgw.Writers]] 
 # Url = "http://127.0.0.1:8480/insert/0/prometheus/api/v1/write"

memsto/notify_config.go

@@ -1,9 +1,7 @@
 package memsto
 
 import (
-	"crypto/tls"
 	"encoding/json"
-	"net/http"
 	"strings"
 	"sync"
 	"time"
@@ -21,7 +19,7 @@ import (
 type NotifyConfigCacheType struct {
 	ctx         *ctx.Context
 	ConfigCache *ConfigCache
-	webhooks    map[string]*models.Webhook
+	webhooks    []*models.Webhook
 	smtp        aconf.SMTPConfig
 	script      models.NotifyScript
 
@@ -49,7 +47,6 @@ func NewNotifyConfigCache(ctx *ctx.Context, configCache *ConfigCache) *NotifyCon
 	w := &NotifyConfigCacheType{
 		ctx:         ctx,
 		ConfigCache: configCache,
-		webhooks:    make(map[string]*models.Webhook),
 	}
 	w.SyncNotifyConfigs()
 	return w
@@ -88,55 +85,11 @@ func (w *NotifyConfigCacheType) syncNotifyConfigs() error {
 	}
 
 	if strings.TrimSpace(cval) != "" {
-		var webhooks []*models.Webhook
-		err = json.Unmarshal([]byte(cval), &webhooks)
+		err = json.Unmarshal([]byte(cval), &w.webhooks)
 		if err != nil {
 			dumper.PutSyncRecord("webhooks", start.Unix(), -1, -1, "failed to unmarshal configs.webhook: "+err.Error())
 			logger.Errorf("failed to unmarshal webhooks:%s error:%v", cval, err)
 		}
-
-		newWebhooks := make(map[string]*models.Webhook, len(webhooks))
-		for i := 0; i < len(webhooks); i++ {
-			if webhooks[i].Batch == 0 {
-				webhooks[i].Batch = 1000
-			}
-
-			if webhooks[i].Timeout == 0 {
-				webhooks[i].Timeout = 10
-			}
-
-			if webhooks[i].RetryCount == 0 {
-				webhooks[i].RetryCount = 10
-			}
-
-			if webhooks[i].RetryInterval == 0 {
-				webhooks[i].RetryInterval = 10
-			}
-
-			if webhooks[i].Client == nil {
-				webhooks[i].Client = &http.Client{
-					Timeout: time.Second * time.Duration(webhooks[i].Timeout),
-					Transport: &http.Transport{
-						TLSClientConfig: &tls.Config{InsecureSkipVerify: webhooks[i].SkipVerify},
-					},
-				}
-			}
-			newWebhooks[webhooks[i].Url] = webhooks[i]
-		}
-
-		for url, wh := range newWebhooks {
-			if oldWh, has := w.webhooks[url]; has && oldWh.Hash() != wh.Hash() {
-				w.webhooks[url] = wh
-			} else {
-				w.webhooks[url] = wh
-			}
-		}
-
-		for url := range w.webhooks {
-			if _, has := newWebhooks[url]; !has {
-				delete(w.webhooks, url)
-			}
-		}
 	}
 
 	dumper.PutSyncRecord("webhooks", start.Unix(), time.Since(start).Milliseconds(), len(w.webhooks), "success, webhooks:\n"+cval)
@@ -180,7 +133,7 @@ func (w *NotifyConfigCacheType) syncNotifyConfigs() error {
 	return nil
 }
 
-func (w *NotifyConfigCacheType) GetWebhooks() map[string]*models.Webhook {
+func (w *NotifyConfigCacheType) GetWebhooks() []*models.Webhook {
 	w.RWMutex.RLock()
 	defer w.RWMutex.RUnlock()
 	return w.webhooks

models/busi_group.go

@@ -115,54 +115,68 @@ func BusiGroupExists(ctx *ctx.Context, where string, args ...interface{}) (bool,
 	return num > 0, err
 }
 
-var entries = []struct {
-	entry        interface{}
-	errorMessage string
-}{
-	{
-		entry:        &AlertRule{},
-		errorMessage: "Some alert rules still in the BusiGroup",
-	},
-	{
-		entry:        &AlertMute{},
-		errorMessage: "Some alert mutes still in the BusiGroup",
-	},
-	{
-		entry:        &AlertSubscribe{},
-		errorMessage: "Some alert subscribes still in the BusiGroup",
-	},
-	{
-		entry:        &Target{},
-		errorMessage: "Some targets still in the BusiGroup",
-	},
-	{
-		entry:        &RecordingRule{},
-		errorMessage: "Some recording rules still in the BusiGroup",
-	},
-	{
-		entry:        &TaskTpl{},
-		errorMessage: "Some recovery scripts still in the BusiGroup",
-	},
-	{
-		entry:        &TaskRecord{},
-		errorMessage: "Some Task Record records still in the BusiGroup",
-	},
-	{
-		entry:        &TargetBusiGroup{},
-		errorMessage: "Some target busigroups still in the BusiGroup",
-	},
-}
-
 func (bg *BusiGroup) Del(ctx *ctx.Context) error {
-	for _, e := range entries {
-		has, err := Exists(DB(ctx).Model(e.entry).Where("group_id=?", bg.Id))
-		if err != nil {
-			return err
-		}
+	has, err := Exists(DB(ctx).Model(&AlertMute{}).Where("group_id=?", bg.Id))
+	if err != nil {
+		return err
+	}
 
-		if has {
-			return errors.New(e.errorMessage)
-		}
+	if has {
+		return errors.New("Some alert mutes still in the BusiGroup")
+	}
+
+	has, err = Exists(DB(ctx).Model(&AlertSubscribe{}).Where("group_id=?", bg.Id))
+	if err != nil {
+		return err
+	}
+
+	if has {
+		return errors.New("Some alert subscribes still in the BusiGroup")
+	}
+
+	has, err = Exists(DB(ctx).Model(&TargetBusiGroup{}).Where("group_id=?", bg.Id))
+	if err != nil {
+		return err
+	}
+
+	if has {
+		return errors.New("Some targets still in the BusiGroup")
+	}
+
+	has, err = Exists(DB(ctx).Model(&Board{}).Where("group_id=?", bg.Id))
+	if err != nil {
+		return err
+	}
+
+	if has {
+		return errors.New("Some dashboards still in the BusiGroup")
+	}
+
+	has, err = Exists(DB(ctx).Model(&TaskTpl{}).Where("group_id=?", bg.Id))
+	if err != nil {
+		return err
+	}
+
+	if has {
+		return errors.New("Some recovery scripts still in the BusiGroup")
+	}
+
+	// hasCR, err := Exists(DB(ctx).Table("collect_rule").Where("group_id=?", bg.Id))
+	// if err != nil {
+	// 	return err
+	// }
+
+	// if hasCR {
+	// 	return errors.New("Some collect rules still in the BusiGroup")
+	// }
+
+	has, err = Exists(DB(ctx).Model(&AlertRule{}).Where("group_id=?", bg.Id))
+	if err != nil {
+		return err
+	}
+
+	if has {
+		return errors.New("Some alert rules still in the BusiGroup")
 	}
 
 	return DB(ctx).Transaction(func(tx *gorm.DB) error {

models/notify_config.go

@@ -1,12 +1,5 @@
 package models
 
-import (
-	"fmt"
-	"net/http"
-
-	"github.com/toolkits/pkg/str"
-)
-
 const WEBHOOKKEY = "webhook"
 const NOTIFYSCRIPT = "notify_script"
 const NOTIFYCHANNEL = "notify_channel"
@@ -31,11 +24,6 @@ type Webhook struct {
 	RetryCount    int               `json:"retry_count"`
 	RetryInterval int               `json:"retry_interval"`
 	Batch         int               `json:"batch"`
-	Client        *http.Client      `json:"-"`
-}
-
-func (w *Webhook) Hash() string {
-	return str.MD5(fmt.Sprintf("%d_%t_%s_%s_%s_%d_%v_%t_%s_%d_%d_%d", w.Type, w.Enable, w.Url, w.BasicAuthUser, w.BasicAuthPass, w.Timeout, w.HeaderMap, w.SkipVerify, w.Note, w.RetryCount, w.RetryInterval, w.Batch))
 }
 
 type NotifyScript struct {

pkg/poster/post.go

@@ -63,7 +63,7 @@ func GetByUrl[T any](url string, cfg conf.CenterApi) (T, error) {
 		Timeout: time.Duration(cfg.Timeout) * time.Millisecond,
 	}
 
-	if useProxy(url) {
+	if UseProxy(url) {
 		client.Transport = ProxyTransporter
 	}
 
@@ -147,7 +147,7 @@ func PostByUrl[T any](url string, cfg conf.CenterApi, v interface{}) (t T, err e
 		Timeout: time.Duration(cfg.Timeout) * time.Millisecond,
 	}
 
-	if useProxy(url) {
+	if UseProxy(url) {
 		client.Transport = ProxyTransporter
 	}
 
@@ -195,7 +195,7 @@ var ProxyTransporter = &http.Transport{
 	Proxy: http.ProxyFromEnvironment,
 }
 
-func useProxy(url string) bool {
+func UseProxy(url string) bool {
 	// N9E_PROXY_URL=oapi.dingtalk.com,feishu.com
 	patterns := os.Getenv("N9E_PROXY_URL")
 	if patterns != "" {
@@ -228,7 +228,7 @@ func PostJSON(url string, timeout time.Duration, v interface{}, retries ...int)
 		Timeout: timeout,
 	}
 
-	if useProxy(url) {
+	if UseProxy(url) {
 		client.Transport = ProxyTransporter
 	}
 

pushgw/pconf/conf.go

@@ -24,10 +24,8 @@ type Pushgw struct {
 }
 
 type WriterGlobalOpt struct {
-	QueueMaxSize            int
-	QueuePopSize            int
-	AllQueueMaxSize         int
-	AllQueueMaxSizeInterval int
+	QueueMaxSize int
+	QueuePopSize int
 }
 
 type WriterOptions struct {
@@ -79,14 +77,6 @@ func (p *Pushgw) PreCheck() {
 		p.WriterOpt.QueuePopSize = 1000
 	}
 
-	if p.WriterOpt.AllQueueMaxSize <= 0 {
-		p.WriterOpt.AllQueueMaxSize = 10000000
-	}
-
-	if p.WriterOpt.AllQueueMaxSizeInterval <= 0 {
-		p.WriterOpt.AllQueueMaxSizeInterval = 200
-	}
-
 	if p.WriteConcurrency <= 0 {
 		p.WriteConcurrency = 5000
 	}

pushgw/writer/writer.go

@@ -8,7 +8,6 @@ import (
 	"net/http"
 	"strings"
 	"sync"
-	"sync/atomic"
 	"time"
 
 	"github.com/ccfos/nightingale/v6/pkg/fasttime"
@@ -139,10 +138,9 @@ func (w WriterType) Post(req []byte, headers ...map[string]string) error {
 }
 
 type WritersType struct {
-	pushgw      pconf.Pushgw
-	backends    map[string]WriterType
-	queues      map[string]*IdentQueue
-	allQueueLen atomic.Value
+	pushgw   pconf.Pushgw
+	backends map[string]WriterType
+	queues   map[string]*IdentQueue
 	sync.RWMutex
 }
 
@@ -162,30 +160,14 @@ func (ws *WritersType) ReportQueueStats(ident string, identQueue *IdentQueue) (i
 	}
 }
 
-func (ws *WritersType) SetAllQueueLen() {
-	for {
-		curMetricLen := 0
-		ws.RLock()
-		for _, q := range ws.queues {
-			curMetricLen += q.list.Len()
-		}
-		ws.RUnlock()
-		ws.allQueueLen.Store(curMetricLen)
-		time.Sleep(time.Duration(ws.pushgw.WriterOpt.AllQueueMaxSizeInterval) * time.Millisecond)
-	}
-}
-
 func NewWriters(pushgwConfig pconf.Pushgw) *WritersType {
 	writers := &WritersType{
-		backends:    make(map[string]WriterType),
-		queues:      make(map[string]*IdentQueue),
-		pushgw:      pushgwConfig,
-		allQueueLen: atomic.Value{},
+		backends: make(map[string]WriterType),
+		queues:   make(map[string]*IdentQueue),
+		pushgw:   pushgwConfig,
 	}
 
 	writers.Init()
-
-	go writers.SetAllQueueLen()
 	go writers.CleanExpQueue()
 	return writers
 }
@@ -235,13 +217,6 @@ func (ws *WritersType) PushSample(ident string, v interface{}) {
 	}
 
 	identQueue.ts = time.Now().Unix()
-	curLen := ws.allQueueLen.Load().(int)
-	if curLen > ws.pushgw.WriterOpt.AllQueueMaxSize {
-		logger.Warningf("Write %+v full, metric count over limit: %d", v, curLen)
-		CounterPushQueueErrorTotal.WithLabelValues(ident).Inc()
-		return
-	}
-
 	succ := identQueue.list.PushFront(v)
 	if !succ {
 		logger.Warningf("Write channel(%s) full, current channel size: %d", ident, identQueue.list.Len())
@@ -270,7 +245,6 @@ func (ws *WritersType) StartConsumer(identQueue *IdentQueue) {
 
 func (ws *WritersType) Init() error {
 	opts := ws.pushgw.Writers
-	ws.allQueueLen.Store(0)
 
 	for i := 0; i < len(opts); i++ {
 		tlsConf, err := opts[i].ClientConfig.TLSConfig()