로그 전송을 비동기로

opensearch/client.go

@@ -8,9 +8,9 @@ import (
 	"encoding/base64"
 	"encoding/json"
 	"fmt"
-	"io"
 	"net/http"
 	"strings"
+	"sync/atomic"
 	"time"
 
 	osg "github.com/opensearch-project/opensearch-go/v4"
@@ -31,6 +31,7 @@ type Client struct {
 	indexTemplatePattern string
 	bulkHeader           http.Header
 	singleHeader         http.Header
+	sendingCount         int32
 }
 
 type LogDocument struct {
@@ -55,38 +56,43 @@ func NewLogDocument(logType string, body any) *LogDocument {
 	}
 }
 
-func (c *Client) Send(ld *LogDocument) error {
+func (c *Client) Send(ld *LogDocument) {
 	if c.Client == nil {
-		return nil
+		return
 	}
 
 	serialized, _ := json.Marshal(ld)
-	reader := bytes.NewBuffer(serialized)
+	go func(serialized []byte) {
-	req := osapi.IndexReq{
+		sending := atomic.AddInt32(&c.sendingCount, 1)
-		Index:  c.indexTemplatePattern + ld.Type,
+		atomic.AddInt32(&c.sendingCount, -1)
-		Body:   reader,
-		Header: c.singleHeader,
-	}
-	logger.Println("LogSend", req)
 
-	resp, err := c.Do(context.Background(), req, nil)
+		if sending > 100 {
-	logger.Println(resp)
+			logger.Println("sending log bottleneck :", sending)
-	if err != nil {
+			logger.Println(string(serialized))
-		return err
+			return
-	}
+		}
-	defer resp.Body.Close()
 
-	r, err2 := io.ReadAll(resp.Body)
+		reader := bytes.NewBuffer(serialized)
-	if err2 != nil {
+		req := osapi.IndexReq{
-		logger.Println("LogSend resp read error :", err2)
+			Index:  c.indexTemplatePattern + ld.Type,
-	} else {
+			Body:   reader,
-		logger.Println("LogSend resp :", string(r))
+			Header: c.singleHeader,
-	}
+		}
 
-	return nil
+		resp, err := c.Do(context.Background(), req, nil)
+		if err != nil {
+			logger.Println("log send failed :", err)
+			return
+		}
+		resp.Body.Close()
+	}(serialized)
 }
 
-func (c *Client) SendBulk(ds map[string]*LogDocument) error {
+func (c *Client) SendBulk(ds map[string]*LogDocument) {
+	if c == nil {
+		return
+	}
+
 	var contents string
 
 	for _, d := range ds {
@@ -94,19 +100,28 @@ func (c *Client) SendBulk(ds map[string]*LogDocument) error {
 		contents += fmt.Sprintf(`{"create":{"_index":"%s%s"}}`+"\n"+`%s`+"\n", c.indexTemplatePattern, d.Type, string(b))
 	}
 
-	reader := bytes.NewBuffer([]byte(contents))
+	go func(contents string) {
-	req := osapi.BulkReq{
+		sending := atomic.AddInt32(&c.sendingCount, 1)
-		Body:   reader,
+		atomic.AddInt32(&c.sendingCount, -1)
-		Header: c.bulkHeader,
-	}
-	resp, err := c.Do(context.Background(), req, nil)
-	if err != nil {
-		return err
-	}
-	logger.Println(resp)
-	defer resp.Body.Close()
 
-	return nil
+		if sending > 100 {
+			logger.Println("sending log bottleneck :", sending)
+			logger.Println(contents)
+			return
+		}
+
+		reader := bytes.NewBuffer([]byte(contents))
+		req := osapi.BulkReq{
+			Body:   reader,
+			Header: c.bulkHeader,
+		}
+		resp, err := c.Do(context.Background(), req, nil)
+		if err != nil {
+			logger.Println("log send bulk failed :", err)
+			return
+		}
+		resp.Body.Close()
+	}(contents)
 }
 
 var jwtHeader string