ayo/gocommon
8
0
Fork 1
Code Issues Pull Requests Projects Releases Wiki Activity

로그 추가

Browse Source
This commit is contained in:
mountain
2025-09-10 11:42:57 +09:00
parent 59fa0cc2ee
commit 085d0bb740
1 changed files with 14 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
opensearch/client.go
View File

@ -154,19 +154,22 @@ func (c *Client) sendLoop(ctx context.Context) {
} }
resp, err := c.Do(context.Background(), req, nil) resp, err := c.Do(context.Background(), req, nil)
logger.Println("[LogStream] process BulkReq :", err)
if err != nil { if err != nil {
if netoperr, ok := err.(*net.OpError); ok && netoperr.Op == "dial" { if netoperr, ok := err.(*net.OpError); ok && netoperr.Op == "dial" {
// 접속 안됨. 재시도 안함 // 접속 안됨. 재시도 안함
logger.Println("log send bulk failed. no retry :", err) logger.Println("[LogStream] send bulk failed. no retry :", err)
reader.printSent() reader.printSent()
} else { } else {
// 재시도 // 재시도
logger.Println("log send bulk failed. retry :", err) logger.Println("[LogStream] send bulk failed. retry :", err)
failChan <- sending failChan <- sending
} }
return return
} }
if resp.Body == nil { if resp.Body == nil {
logger.Println("[LogStream] send bulk failed. empty response")
return return
} }
defer resp.Body.Close() defer resp.Body.Close()
@ -179,12 +182,15 @@ func (c *Client) sendLoop(ctx context.Context) {
} `json:"create"` } `json:"create"`
} `json:"items"` } `json:"items"`
} }
json.NewDecoder(resp.Body).Decode(&respbody) if err := json.NewDecoder(resp.Body).Decode(&respbody); err != nil {
if !respbody.Errors { logger.Println("[LogStream] decode response body failed :", err)
return return
} }
logger.Println("log send bulk failed. retry :", respbody.Items) if !respbody.Errors {
logger.Println("[LogStream] process BulkReq success", respbody)
return
}
var retry [][]byte var retry [][]byte
for i, item := range respbody.Items { for i, item := range respbody.Items {
@ -193,6 +199,8 @@ func (c *Client) sendLoop(ctx context.Context) {
retry = append(retry, sending[i]) retry = append(retry, sending[i])
} }
} }
logger.Println("[LogStream] send bulk failed. retry :", len(retry))
if len(retry) > 0 { if len(retry) > 0 {
failChan <- retry failChan <- retry
} }
@ -330,6 +338,7 @@ func NewClient(ctx context.Context, cfg Config) (Client, error) {
indexPrefix = "ds-logs-" + indexPrefix indexPrefix = "ds-logs-" + indexPrefix
} }
logger.Println("[LogStream] stream indexPrefix :", indexPrefix)
bulkHeader := make(http.Header) bulkHeader := make(http.Header)
singleHeader := make(http.Header) singleHeader := make(http.Header)
if len(cfg.Username) > 0 && len(cfg.Password) > 0 { if len(cfg.Username) > 0 && len(cfg.Password) > 0 {