실패 로그 재전송 로직 수정

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,88 +169,108 @@ func (c *Client) sendLoop(ctx context.Context) {
 		}
 	}()
 
-	failChan := make(chan []*singleLogMarshaller)
+	failChan := make(chan []singleLogMarshaller)
-	var logMarshallers []*singleLogMarshaller
+	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() {
-			defer func() {
+			r := recover()
-				r := recover()
+			if r != nil {
-				if r != nil {
+				logger.Println(r)
-					logger.Println(r)
-				}
-			}()
-
-			reader := &stringSliceReader{src: sending, cursor: 0}
-			req := osapi.BulkReq{
-				Body:   reader,
-				Header: c.bulkHeader,
 			}
-			resp, err := c.Do(context.Background(), req, nil)
+		}()
 
-			if err != nil {
+		reader := newLogSliceReader(logs)
-				if netoperr, ok := err.(*net.OpError); ok && netoperr.Op == "dial" {
+		req := osapi.BulkReq{
-					// 접속 안됨. 재시도 안함
+			Body:   reader,
-					logger.Println("[LogStream] send bulk failed. no retry :", err)
+			Header: c.bulkHeader,
-					reader.printSent()
+		}
+		resp, err := c.Do(context.Background(), req, nil)
+
+		if err != nil {
+			if netoperr, ok := err.(*net.OpError); ok && netoperr.Op == "dial" {
+				// 접속 안됨. 재시도 안함
+				logger.Println("[LogStream] send bulk failed. no retry :", err)
+				reader.printSent()
+			} else {
+				// 재시도
+				logger.Println("[LogStream] send bulk failed. retry :", err)
+				failChan <- logs
+			}
+			return
+		}
+		if resp.Body == nil {
+			return
+		}
+		defer resp.Body.Close()
+
+		var respbody struct {
+			Errors bool `json:"errors"`
+			Items  []struct {
+				Create struct {
+					Status int `json:"status"`
+				} `json:"create"`
+			} `json:"items"`
+		}
+
+		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
+		for i, item := range respbody.Items {
+			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))
-					logger.Println("[LogStream] send bulk failed. retry :", err)
-					failChan <- sending
 				}
-				return
+			} else {
+				// 일단 로그만
+				logger.Println("[LogStream] send bulk failed but no retry. status :", item.Create.Status)
 			}
-			if resp.Body == nil {
+		}
-				return
-			}
-			defer resp.Body.Close()
 
-			var respbody struct {
+		if len(retry) > 0 {
-				Errors bool `json:"errors"`
+			failChan <- retry
-				Items  []struct {
+		}
-					Create struct {
+	}
-						Status int `json:"status"`
-					} `json:"create"`
-				} `json:"items"`
-			}
-			if err := json.NewDecoder(resp.Body).Decode(&respbody); err != nil {
-				logger.Println("[LogStream] decode response body failed :", err)
-				return
-			}
 
-			if !respbody.Errors {
+	totalsize := 0
-				return
+	appendLog := func(newlog singleLogMarshaller) bool {
-			}
+		if totalsize+newlog.length > logbulksize {
+			go sendfunc(logMarshallers)
+			totalsize = newlog.length
+			logMarshallers = []singleLogMarshaller{newlog}
+			return true
+		}
 
-			var retry []*singleLogMarshaller
+		totalsize += newlog.length
-			for i, item := range respbody.Items {
+		logMarshallers = append(logMarshallers, newlog)
-				if item.Create.Status < 400 {
+		return false
-					// 재시도
-					retry = append(retry, sending[i])
-				}
-			}
-
-			logger.Println("[LogStream] send bulk failed. retry :", len(retry))
-			if len(retry) > 0 {
-				failChan <- retry
-			}
-		}(sending)
 	}
 
 	for {
@@ -244,12 +280,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 +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)
+			}
 		}
 	}
 }