fix bug get goroutine for audit timeout

Signed-off-by: Nwanjunlei <wanjunlei@yunify.com>

pkg/apiserver/apiserver.go

@@ -281,7 +281,7 @@ func (s *APIServer) buildHandlerChain(stopCh <-chan struct{}) {
 
 	if s.Config.AuditingOptions.Enable {
 		handler = filters.WithAuditing(handler,
-			audit.NewAuditing(s.InformerFactory, s.Config.AuditingOptions.WebhookUrl, stopCh))
+			audit.NewAuditing(s.InformerFactory, s.Config.AuditingOptions, stopCh))
 	}
 
 	var authorizers authorizer.Authorizer

pkg/apiserver/auditing/backend.go

@@ -23,42 +23,62 @@ import (
 	"encoding/json"
 	"k8s.io/klog"
 	"kubesphere.io/kubesphere/pkg/apiserver/auditing/v1alpha1"
+	options "kubesphere.io/kubesphere/pkg/simple/client/auditing/elasticsearch"
 	"net/http"
 	"time"
 )
 
 const (
-	WaitTimeout = time.Second
-	WebhookURL  = "https://kube-auditing-webhook-svc.kubesphere-logging-system.svc:443/audit/webhook/event"
+	WaitTimeout          = time.Second
+	SendTimeout          = time.Second * 3
+	DefaultGoroutinesNum = 100
+	DefaultBatchSize     = 100
+	DefaultBatchWait     = time.Second * 3
+	WebhookURL           = "https://kube-auditing-webhook-svc.kubesphere-logging-system.svc:443/audit/webhook/event"
 )
 
 type Backend struct {
-	url             string
-	channelCapacity int
-	semCh           chan interface{}
-	cache           chan *v1alpha1.EventList
-	client          http.Client
-	sendTimeout     time.Duration
-	waitTimeout     time.Duration
-	stopCh          <-chan struct{}
+	url          string
+	semCh        chan interface{}
+	cache        chan *v1alpha1.Event
+	client       http.Client
+	sendTimeout  time.Duration
+	waitTimeout  time.Duration
+	maxBatchSize int
+	maxBatchWait time.Duration
+	stopCh       <-chan struct{}
 }
 
-func NewBackend(url string, channelCapacity int, cache chan *v1alpha1.EventList, sendTimeout time.Duration, stopCh <-chan struct{}) *Backend {
+func NewBackend(opts *options.Options, cache chan *v1alpha1.Event, stopCh <-chan struct{}) *Backend {
 
 	b := Backend{
-		url:             url,
-		semCh:           make(chan interface{}, channelCapacity),
-		channelCapacity: channelCapacity,
-		waitTimeout:     WaitTimeout,
-		cache:           cache,
-		sendTimeout:     sendTimeout,
-		stopCh:          stopCh,
+		url:          opts.WebhookUrl,
+		waitTimeout:  WaitTimeout,
+		cache:        cache,
+		sendTimeout:  SendTimeout,
+		maxBatchSize: opts.MaxBatchSize,
+		maxBatchWait: opts.MaxBatchWait,
+		stopCh:       stopCh,
 	}
 
 	if len(b.url) == 0 {
 		b.url = WebhookURL
 	}
 
+	if b.maxBatchWait == 0 {
+		b.maxBatchWait = DefaultBatchWait
+	}
+
+	if b.maxBatchSize == 0 {
+		b.maxBatchSize = DefaultBatchSize
+	}
+
+	goroutinesNum := opts.GoroutinesNum
+	if goroutinesNum == 0 {
+		goroutinesNum = DefaultGoroutinesNum
+	}
+	b.semCh = make(chan interface{}, goroutinesNum)
+
 	b.client = http.Client{
 		Transport: &http.Transport{
 			TLSClientConfig: &tls.Config{
@@ -76,53 +96,97 @@ func NewBackend(url string, channelCapacity int, cache chan *v1alpha1.EventList,
 func (b *Backend) worker() {
 
 	for {
+		events := b.getEvents()
+		if events == nil {
+			break
+		}
 
-		var event *v1alpha1.EventList
+		if len(events.Items) == 0 {
+			continue
+		}
+
+		go b.sendEvents(events)
+	}
+}
+
+func (b *Backend) getEvents() *v1alpha1.EventList {
+
+	ctx, cancel := context.WithTimeout(context.Background(), b.maxBatchWait)
+	defer cancel()
+
+	events := &v1alpha1.EventList{}
+	for {
 		select {
-		case event = <-b.cache:
+		case event := <-b.cache:
 			if event == nil {
 				break
 			}
+			events.Items = append(events.Items, *event)
+			if len(events.Items) >= b.maxBatchSize {
+				return events
+			}
+		case <-ctx.Done():
+			return events
 		case <-b.stopCh:
-			break
+			return nil
 		}
+	}
+}
 
-		send := func(event *v1alpha1.EventList) {
-			ctx, cancel := context.WithTimeout(context.Background(), b.waitTimeout)
-			defer cancel()
+func (b *Backend) sendEvents(events *v1alpha1.EventList) {
 
-			select {
-			case <-ctx.Done():
-				klog.Errorf("get goroutine for audit(%s) timeout", event.Items[0].AuditID)
-				return
-			case b.semCh <- struct{}{}:
-			}
+	ctx, cancel := context.WithTimeout(context.Background(), b.sendTimeout)
+	defer cancel()
 
-			defer func() {
-				<-b.semCh
-			}()
+	stopCh := make(chan struct{})
 
-			bs, err := b.eventToBytes(event)
-			if err != nil {
-				klog.V(6).Infof("json marshal error, %s", err)
-				return
-			}
+	send := func() {
+		ctx, cancel := context.WithTimeout(context.Background(), b.waitTimeout)
+		defer cancel()
 
-			klog.V(8).Infof("%s", string(bs))
+		select {
+		case <-ctx.Done():
+			klog.Error("get goroutine timeout")
+			return
+		case b.semCh <- struct{}{}:
+		}
 
-			response, err := b.client.Post(b.url, "application/json", bytes.NewBuffer(bs))
-			if err != nil {
-				klog.Errorf("send audit event[%s] error, %s", event.Items[0].AuditID, err)
-				return
-			}
+		start := time.Now()
+		defer func() {
+			stopCh <- struct{}{}
+			klog.V(8).Infof("send %d auditing logs used %d", len(events.Items), time.Now().Sub(start).Milliseconds())
+		}()
 
-			if response.StatusCode != http.StatusOK {
-				klog.Errorf("send audit event[%s] error[%d]", event.Items[0].AuditID, response.StatusCode)
-				return
-			}
+		bs, err := b.eventToBytes(events)
+		if err != nil {
+			klog.V(6).Infof("json marshal error, %s", err)
+			return
 		}
 
-		go send(event)
+		klog.V(8).Infof("%s", string(bs))
+
+		response, err := b.client.Post(b.url, "application/json", bytes.NewBuffer(bs))
+		if err != nil {
+			klog.Errorf("send audit events error, %s", err)
+			return
+		}
+
+		if response.StatusCode != http.StatusOK {
+			klog.Errorf("send audit events error[%d]", response.StatusCode)
+			return
+		}
+	}
+
+	go send()
+
+	defer func() {
+		<-b.semCh
+	}()
+
+	select {
+	case <-ctx.Done():
+		klog.Error("send audit events timeout")
+	case <-stopCh:
 	}
 }
 

pkg/apiserver/auditing/types.go

@@ -36,6 +36,7 @@ import (
 	"kubesphere.io/kubesphere/pkg/informers"
 	"kubesphere.io/kubesphere/pkg/models/resources/v1alpha3"
 	"kubesphere.io/kubesphere/pkg/models/resources/v1alpha3/devops"
+	options "kubesphere.io/kubesphere/pkg/simple/client/auditing/elasticsearch"
 	"kubesphere.io/kubesphere/pkg/utils/iputil"
 	"net"
 	"net/http"
@@ -46,8 +47,6 @@ const (
 	DefaultWebhook       = "kube-auditing-webhook"
 	DefaultCacheCapacity = 10000
 	CacheTimeout         = time.Second
-	SendTimeout          = time.Second * 3
-	ChannelCapacity      = 10
 )
 
 type Auditing interface {
@@ -60,19 +59,19 @@ type Auditing interface {
 type auditing struct {
 	webhookLister v1alpha1.WebhookLister
 	devopsGetter  v1alpha3.Interface
-	cache         chan *auditv1alpha1.EventList
+	cache         chan *auditv1alpha1.Event
 	backend       *Backend
 }
 
-func NewAuditing(informers informers.InformerFactory, url string, stopCh <-chan struct{}) Auditing {
+func NewAuditing(informers informers.InformerFactory, opts *options.Options, stopCh <-chan struct{}) Auditing {
 
 	a := &auditing{
 		webhookLister: informers.KubeSphereSharedInformerFactory().Auditing().V1alpha1().Webhooks().Lister(),
 		devopsGetter:  devops.New(informers.KubeSphereSharedInformerFactory()),
-		cache:         make(chan *auditv1alpha1.EventList, DefaultCacheCapacity),
+		cache:         make(chan *auditv1alpha1.Event, DefaultCacheCapacity),
 	}
 
-	a.backend = NewBackend(url, ChannelCapacity, a.cache, SendTimeout, stopCh)
+	a.backend = NewBackend(opts, a.cache, stopCh)
 	return a
 }
 
@@ -226,10 +225,8 @@ func (a *auditing) LogResponseObject(e *auditv1alpha1.Event, resp *ResponseCaptu
 
 func (a *auditing) cacheEvent(e auditv1alpha1.Event) {
 
-	eventList := &auditv1alpha1.EventList{}
-	eventList.Items = append(eventList.Items, e)
 	select {
-	case a.cache <- eventList:
+	case a.cache <- &e:
 		return
 	case <-time.After(CacheTimeout):
 		klog.Errorf("cache audit event %s timeout", e.AuditID)

pkg/simple/client/auditing/elasticsearch/options.go

@@ -19,14 +19,21 @@ package elasticsearch
 import (
 	"github.com/spf13/pflag"
 	"kubesphere.io/kubesphere/pkg/utils/reflectutils"
+	"time"
 )
 
 type Options struct {
-	Enable      bool   `json:"enable" yaml:"enable"`
-	WebhookUrl  string `json:"webhookUrl" yaml:"webhookUrl"`
-	Host        string `json:"host" yaml:"host"`
-	IndexPrefix string `json:"indexPrefix,omitempty" yaml:"indexPrefix"`
-	Version     string `json:"version" yaml:"version"`
+	Enable     bool   `json:"enable" yaml:"enable"`
+	WebhookUrl string `json:"webhookUrl" yaml:"webhookUrl"`
+	// The number of goroutines which send auditing events to webhook.
+	GoroutinesNum int `json:"goroutinesNum" yaml:"goroutinesNum"`
+	// The max size of the auditing event in a batch.
+	MaxBatchSize int `json:"batchSize" yaml:"batchSize"`
+	// MaxBatchWait indicates the maximum interval between two batches.
+	MaxBatchWait time.Duration `json:"batchTimeout" yaml:"batchTimeout"`
+	Host         string        `json:"host" yaml:"host"`
+	IndexPrefix  string        `json:"indexPrefix,omitempty" yaml:"indexPrefix"`
+	Version      string        `json:"version" yaml:"version"`
 }
 
 func NewElasticSearchOptions() *Options {
@@ -52,7 +59,12 @@ func (s *Options) AddFlags(fs *pflag.FlagSet, c *Options) {
 	fs.BoolVar(&s.Enable, "auditing-enabled", c.Enable, "Enable auditing component or not. ")
 
 	fs.StringVar(&s.WebhookUrl, "auditing-webhook-url", c.WebhookUrl, "Auditing wehook url")
-
+	fs.IntVar(&s.GoroutinesNum, "auditing-goroutines-num", c.GoroutinesNum,
+		"The number of goroutines which send auditing events to webhook.")
+	fs.IntVar(&s.MaxBatchSize, "auditing-batch-max-size", c.MaxBatchSize,
+		"The max size of the auditing event in a batch.")
+	fs.DurationVar(&s.MaxBatchWait, "auditing-batch-max-wait", c.MaxBatchWait,
+		"MaxBatchWait indicates the maximum interval between two batches.")
 	fs.StringVar(&s.Host, "auditing-elasticsearch-host", c.Host, ""+
 		"Elasticsearch service host. KubeSphere is using elastic as auditing store, "+
 		"if this filed left blank, KubeSphere will use kubernetes builtin event API instead, and"+