logstream 로그 추가

opensearch/client.go

@@ -19,6 +19,8 @@ import (
 	"repositories.action2quare.com/ayo/gocommon/logger"
 )
 
+const logbulksize = 512 * 1024
+
 type Config struct {
 	osg.Config  `json:",inline"`
 	IndexPrefix string `json:"IndexPrefix"`
@@ -157,23 +159,11 @@ func (c *Client) sendLoop(ctx context.Context) {
 	var logMarshallers []*singleLogMarshaller
 	sendTick := time.After(time.Minute)
 
-	sendfunc := func() {
+	sendfunc := func(logs []*singleLogMarshaller) {
-		// 2mb가 넘지 않게 조절.
+		if len(logs) == 0 {
-		// 실패한 로그가 다시 되돌아 오면 contents가 커질 수 있다.
+			return
-		sendingSize := 0
-		cut := 0
-		for ; cut < len(logMarshallers); cut++ {
-			// 2메가가 넘더라도 최소한 하나는 보내자.
-			if cut > 0 && sendingSize+logMarshallers[cut].length > 2*1024*1024 {
-				break
 		}
-			sendingSize += logMarshallers[cut].length
-		}
-		sending := logMarshallers[:cut]
-		logMarshallers = logMarshallers[cut:]
-		sendTick = time.After(time.Minute)
 
-		go func(sending []*singleLogMarshaller) {
 		defer func() {
 			r := recover()
 			if r != nil {
@@ -181,7 +171,7 @@ func (c *Client) sendLoop(ctx context.Context) {
 			}
 		}()
 
-			reader := &stringSliceReader{src: sending, cursor: 0}
+		reader := &stringSliceReader{src: logs, cursor: 0}
 		req := osapi.BulkReq{
 			Body:   reader,
 			Header: c.bulkHeader,
@@ -196,7 +186,7 @@ func (c *Client) sendLoop(ctx context.Context) {
 			} else {
 				// 재시도
 				logger.Println("[LogStream] send bulk failed. retry :", err)
-					failChan <- sending
+				failChan <- logs
 			}
 			return
 		}
@@ -213,12 +203,18 @@ func (c *Client) sendLoop(ctx context.Context) {
 				} `json:"create"`
 			} `json:"items"`
 		}
-			if err := json.NewDecoder(resp.Body).Decode(&respbody); err != nil {
+
-				logger.Println("[LogStream] decode response body failed :", err)
+		decoder := json.NewDecoder(resp.Body)
+		if err := decoder.Decode(&respbody); err != nil {
+			errbody, _ := io.ReadAll(decoder.Buffered())
+			logger.Println("[LogStream] decode response body failed and retry :", err, string(errbody), len(logs))
+			// 전체 재시도 필요
+			failChan <- logs
 			return
 		}
 
 		if !respbody.Errors {
+			// 성공
 			return
 		}
 
@@ -226,15 +222,28 @@ func (c *Client) sendLoop(ctx context.Context) {
 		for i, item := range respbody.Items {
 			if item.Create.Status < 400 {
 				// 재시도
-					retry = append(retry, sending[i])
+				retry = append(retry, logs[i])
 			}
 		}
 
-			logger.Println("[LogStream] send bulk failed. retry :", len(retry))
+		logger.Println("[LogStream] send bulk failed. retry :", len(retry), len(logs), respbody.Items)
 		if len(retry) > 0 {
 			failChan <- retry
 		}
-		}(sending)
+	}
+
+	totalsize := 0
+	appendLog := func(newlog *singleLogMarshaller) bool {
+		if totalsize+newlog.length > logbulksize {
+			go sendfunc(logMarshallers)
+			totalsize = newlog.length
+			logMarshallers = []*singleLogMarshaller{newlog}
+			return true
+		}
+
+		totalsize += newlog.length
+		logMarshallers = append(logMarshallers, newlog)
+		return false
 	}
 
 	for {
@@ -244,12 +253,20 @@ func (c *Client) sendLoop(ctx context.Context) {
 
 		case ret := <-failChan:
 			// 순서는 중요하지 않음.
-			logMarshallers = append(logMarshallers, ret...)
+			sent := false
-			sendfunc()
+			for _, newlog := range ret {
+				sent = sent || appendLog(newlog)
+			}
+
+			if sent {
+				sendTick = time.After(time.Minute)
+			}
 
 		case <-sendTick:
 			if len(logMarshallers) > 0 {
-				sendfunc()
+				go sendfunc(logMarshallers)
+				totalsize = 0
+				logMarshallers = nil
 			} else {
 				sendTick = time.After(time.Minute)
 			}
@@ -257,14 +274,17 @@ func (c *Client) sendLoop(ctx context.Context) {
 		case logDoc := <-c.bulkChan:
 			b, _ := json.Marshal(logDoc)
 			logtype := []byte(logDoc.Type)
-			logMarshallers = append(logMarshallers, &singleLogMarshaller{
+
+			if appendLog(&singleLogMarshaller{
 				singleLogPrepend: c.singleLogPrepend,
 				singleLogMidpend: c.singleLogMidpend,
 				singleLogAppend:  c.singleLogAppend,
 				logtype:          logtype,
 				content:          b,
 				length:           len(logtype) + len(b) + c.singleLogFixedSize,
-			})
+			}) {
+				sendTick = time.After(time.Minute)
+			}
 		}
 	}
 }