실패 로그 재전송 로직 수정

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"`
@@ -88,12 +90,26 @@ type singleLogMarshaller struct {
 	length  int
 }
 
-type stringSliceReader struct {
+type logSliceReader struct {
 	src    []*singleLogMarshaller
 	cursor int
 }
 
-func (b *stringSliceReader) Read(p []byte) (n int, err error) {
+func newLogSliceReader(in []singleLogMarshaller) *logSliceReader {
+	src := make([]*singleLogMarshaller, len(in))
+	for i, v := range in {
+		copylog := new(singleLogMarshaller)
+		*copylog = v
+		src[i] = copylog
+	}
+
+	return &logSliceReader{
+		src:    src,
+		cursor: 0,
+	}
+}
+
+func (b *logSliceReader) Read(p []byte) (n int, err error) {
 	n = 0
 	err = nil
 
@@ -138,7 +154,7 @@ func (b *stringSliceReader) Read(p []byte) (n int, err error) {
 	return
 }
 
-func (b *stringSliceReader) printSent() {
+func (b *logSliceReader) printSent() {
 	for _, r := range b.src {
 		fmt.Print(string(r.content))
 	}
@@ -153,27 +169,15 @@ func (c *Client) sendLoop(ctx context.Context) {
 		}
 	}()
 
-	failChan := make(chan []*singleLogMarshaller)
-	var logMarshallers []*singleLogMarshaller
+	failChan := make(chan []singleLogMarshaller)
+	var logMarshallers []singleLogMarshaller
 	sendTick := time.After(time.Minute)
 
-	sendfunc := func() {
-		// 2mb가 넘지 않게 조절.
-		// 실패한 로그가 다시 되돌아 오면 contents가 커질 수 있다.
-		sendingSize := 0
-		cut := 0
-		for ; cut < len(logMarshallers); cut++ {
-			// 2메가가 넘더라도 최소한 하나는 보내자.
-			if cut > 0 && sendingSize+logMarshallers[cut].length > 2*1024*1024 {
-				break
+	sendfunc := func(logs []singleLogMarshaller) {
+		if len(logs) == 0 {
+			return
 		}
-			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 +185,7 @@ func (c *Client) sendLoop(ctx context.Context) {
 			}
 		}()
 
-			reader := &stringSliceReader{src: sending, cursor: 0}
+		reader := newLogSliceReader(logs)
 		req := osapi.BulkReq{
 			Body:   reader,
 			Header: c.bulkHeader,
@@ -196,7 +200,7 @@ func (c *Client) sendLoop(ctx context.Context) {
 			} else {
 				// 재시도
 				logger.Println("[LogStream] send bulk failed. retry :", err)
-					failChan <- sending
+				failChan <- logs
 			}
 			return
 		}
@@ -213,28 +217,60 @@ 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
 		}
 
-			var retry []*singleLogMarshaller
+		var retry []singleLogMarshaller
 		for i, item := range respbody.Items {
-				if item.Create.Status < 400 {
-					// 재시도
-					retry = append(retry, sending[i])
+			if item.Create.Status < 300 {
+				continue
+			}
+
+			if item.Create.Status == 429 || item.Create.Status >= 500 {
+				logger.Println("[LogStream] send bulk failed but retry. status :", item.Create.Status)
+				retry = append(retry, logs[i])
+			} else if item.Create.Status == 400 {
+				// 구문 오류. 재시도 불가
+				if i < len(logs) {
+					logger.Println("[LogStream] send bulk failed. status 400 :", string(logs[i].content))
+				} else {
+					logger.Println("[LogStream] send bulk failed. status 400 but out of index :", i, len(logs))
+				}
+			} else {
+				// 일단 로그만
+				logger.Println("[LogStream] send bulk failed but no retry. status :", item.Create.Status)
 			}
 		}
 
-			logger.Println("[LogStream] send bulk failed. retry :", len(retry))
 		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 +280,20 @@ func (c *Client) sendLoop(ctx context.Context) {
 
 		case ret := <-failChan:
 			// 순서는 중요하지 않음.
-			logMarshallers = append(logMarshallers, ret...)
-			sendfunc()
+			sent := false
+			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 +301,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)
+			}
 		}
 	}
 }