add LoggerOpenObserve feature

CHANGELOG.md

@@ -2,6 +2,26 @@
 
 All notable changes to paopao-ce are documented in this file.
 ## 0.5.0+dev ([`dev`](https://github.com/rocboss/paopao-ce/tree/dev))
+### Added
+- add `LoggerOpenObserve` feature use OpenObserve to collect log.[#370](https://github.com/rocboss/paopao-ce/pull/370)    
+  add `LoggerOpenObserve` to `conf.yaml` 's `Features` section to enable this feature like below:
+  ```yaml
+  # file config.yaml
+  ...
+  Features:
+    Default: ["Base", "Postgres", "Meili", "LocalOSS", "LoggerOpenObserve", "BigCacheIndex", "web"]
+  LoggerOpenObserve: # 使用OpenObserve写日志
+  Host: 127.0.0.1:5080
+  Organization: paopao-ce
+  Stream: default
+  User: root@paopao.info
+  Password: tiFEI8UeJWuYA7kN
+  Secure: False
+  MinWorker: 5               # 最小后台工作者, 设置范围[5, 100], 默认5
+  MaxLogBuffer: 100          # 最大log缓存条数, 设置范围[10, 10000], 默认100
+  ...
+  ```
+
 ## 0.4.1
 ### Changed
 - infinite scrolling instead of pagination for Home/User/Profile page
@@ -19,7 +39,7 @@ All notable changes to paopao-ce are documented in this file.
   ```
 - add user highlight tweet support include custom tweet set to highlight and list in user/profile page.
 - add cli subcommand to start paopao-ce serve or other task. [#354](https://github.com/rocboss/paopao-ce/pull/354)  
-- add `Friendship` feature . [#355](https://github.com/rocboss/paopao-ce/pull/355)
+- add `Followship` feature . [#355](https://github.com/rocboss/paopao-ce/pull/355)
   migration database first(sql ddl file in `scripts/migration/**/*_user_following.up.sql`):
   ```sql
   DROP TABLE IF EXISTS p_following; 

README.md

@@ -373,6 +373,7 @@ release/paopao serve --no-default-features --features sqlite3,localoss,loggerfil
 |`LoggerFile` | 日志 | 稳定 | 使用文件写日志 |
 |`LoggerZinc` | 日志 | 稳定(推荐) | 使用[Zinc](https://github.com/zinclabs/zinc)写日志 |
 |`LoggerMeili` | 日志 | 内测 | 使用[Meilisearch](https://github.com/meilisearch/meilisearch)写日志 |
+|`LoggerOpenObserve` | 日志 | 内测 | 使用[OpenObserve](https://github.com/openobserve/openobserve)写日志 |
 |[`Friendship`](docs/proposal/22110410-关于Friendship功能项的设计.md) | 关系模式 | 内置 Builtin | 弱关系好友模式，类似微信朋友圈 |
 |[`Followship`](docs/proposal/22110409-关于Followship功能项的设计.md) | 关系模式 | 内置 Builtin | 关注者模式，类似Twitter的Follow模式 |
 |[`Lightship`](docs/proposal/22121409-关于Lightship功能项的设计.md) | 关系模式 | 弃用 Deprecated | 开放模式，所有推文都公开可见 |

features-status.md

@@ -137,6 +137,10 @@
     * [ ] 提按文档  
     * [x] 接口定义
     * [x] 业务逻辑实现 
+* `LoggerOpenObserve` 使用[OpenObserve](https://github.com/openobserve/openobserve)写日志(目前状态: 内测阶段);  
+    * [ ] 提按文档  
+    * [x] 接口定义
+    * [x] 业务逻辑实现 
 
 #### 监控:
 * `Sentry` 使用Sentry进行错误跟踪与性能监控(目前状态: 内测); 

internal/conf/conf.go

@@ -16,6 +16,7 @@ var (
 	loggerFileSetting        *loggerFileConf
 	loggerZincSetting        *loggerZincConf
 	loggerMeiliSetting       *loggerMeiliConf
+	loggerOpenObserveSetting *loggerOpenObserveConf
 	sentrySetting            *sentryConf
 	redisSetting             *redisConf
 
@@ -89,6 +90,7 @@ func setupSetting(suite []string, noDefault bool) error {
 		"LoggerFile":        &loggerFileSetting,
 		"LoggerZinc":        &loggerZincSetting,
 		"LoggerMeili":       &loggerMeiliSetting,
+		"LoggerOpenObserve": &loggerOpenObserveSetting,
 		"Database":          &DatabaseSetting,
 		"MySQL":             &MysqlSetting,
 		"Postgres":          &PostgresSetting,

internal/conf/config.yaml

@@ -114,6 +114,15 @@ LoggerMeili: # 使用Meili写日志
   Secure: False
   MinWorker: 5               # 最小后台工作者, 设置范围[5, 100], 默认5
   MaxLogBuffer: 100          # 最大log缓存条数, 设置范围[10, 10000], 默认100
+LoggerOpenObserve: # 使用OpenObserve写日志
+  Host: 127.0.0.1:5080
+  Organization: paopao-ce
+  Stream: default
+  User: root@paopao.info
+  Password: tiFEI8UeJWuYA7kN
+  Secure: False
+  MinWorker: 5               # 最小后台工作者, 设置范围[5, 100], 默认5
+  MaxLogBuffer: 100          # 最大log缓存条数, 设置范围[10, 10000], 默认100
 JWT: # 鉴权加密
   Secret: 18a6413dc4fe394c66345ebe501b2f26
   Issuer: paopao-api

internal/conf/logger.go

@@ -28,7 +28,7 @@ func setupLogger() {
 	logrus.SetFormatter(&logrus.JSONFormatter{})
 	logrus.SetLevel(loggerSetting.logLevel())
 
-	cfg.In(cfg.Actions{
+	cfg.On(cfg.Actions{
 		"LoggerFile": func() {
 			out := newFileLogger()
 			logrus.SetOutput(out)
@@ -43,6 +43,11 @@ func setupLogger() {
 			logrus.SetOutput(io.Discard)
 			logrus.AddHook(hook)
 		},
+		"LoggerOpenObserve": func() {
+			hook := newObserveLogHook()
+			logrus.SetOutput(io.Discard)
+			logrus.AddHook(hook)
+		},
 	})
 }
 

internal/conf/logger_observe.go

@@ -0,0 +1,71 @@
+// Copyright 2023 ROC. All rights reserved.
+// Use of this source code is governed by a MIT style
+// license that can be found in the LICENSE file.
+
+package conf
+
+import (
+	"log"
+	"net/http"
+	"time"
+
+	hx "github.com/rocboss/paopao-ce/pkg/http"
+	"github.com/rocboss/paopao-ce/pkg/json"
+	"github.com/rocboss/paopao-ce/pkg/obx"
+	"github.com/sirupsen/logrus"
+)
+
+type observeLogData struct {
+	Time    time.Time     `json:"time"`
+	Level   logrus.Level  `json:"level"`
+	Message string        `json:"message"`
+	Data    logrus.Fields `json:"data"`
+}
+
+type observeLogHook struct {
+	client obx.OpenObserveClient
+}
+
+func (h *observeLogHook) Fire(entry *logrus.Entry) error {
+	info := []observeLogData{{
+		Time:    entry.Time,
+		Level:   entry.Level,
+		Message: entry.Message,
+		Data:    entry.Data,
+	}}
+	data, _ := json.Marshal(info)
+	h.client.LogJson(data)
+	return nil
+}
+
+func (h *observeLogHook) Levels() []logrus.Level {
+	return logrus.AllLevels
+}
+
+func newObserveLogHook() *observeLogHook {
+	s := loggerOpenObserveSetting
+	obc := &obx.Config{
+		Host:         s.Host,
+		User:         s.User,
+		Password:     s.Password,
+		Organization: s.Organization,
+		Stream:       s.Stream,
+		Secure:       s.Secure,
+	}
+	acc := &hx.AsyncClientConf{
+		MinWorker:          s.MinWorker,
+		MaxRequestInCh:     s.MaxLogBuffer,
+		MaxRequestInTempCh: 100,
+		MaxTickCount:       60,
+		TickWaitTime:       time.Second,
+	}
+	return &observeLogHook{
+		client: obx.NewClient(obc, acc, func(req *http.Request, resp *http.Response, err error) {
+			if err == nil && resp != nil && resp.Body != nil {
+				resp.Body.Close()
+			} else if err != nil {
+				log.Printf("logrus use observe do LogJson error: %s", err)
+			}
+		}),
+	}
+}

internal/conf/setting.go

@@ -63,6 +63,17 @@ type loggerMeiliConf struct {
 	MinWorker    int
 }
 
+type loggerOpenObserveConf struct {
+	Host         string
+	Organization string
+	Stream       string
+	User         string
+	Password     string
+	Secure       bool
+	MaxLogBuffer int
+	MinWorker    int
+}
+
 type httpServerConf struct {
 	RunMode      string
 	HttpIp       string

internal/dao/search/bridge.go

@@ -65,7 +65,7 @@ func (s *bridgeTweetSearchServant) updateDocs(doc *documents) {
 
 				// watch updateDocsTempch to continue handle update if needed.
 				// cancel loop if no item had watched in 1 minute.
-				for count := 0; count > 60; count++ {
+				for count := 0; count < 60; count++ {
 					select {
 					case item := <-s.updateDocsTempCh:
 						// reset count to continue handle docs update

pkg/http/client.go

@@ -0,0 +1,120 @@
+// Copyright 2023 Michael Li <alimy@gility.net>. All rights reserved.
+// Use of this source code is governed by Apache License 2.0 that
+// can be found in the LICENSE file.
+
+package http
+
+import (
+	"net/http"
+	"time"
+
+	"github.com/sirupsen/logrus"
+)
+
+var (
+	_ AsyncClient = (*wormClient)(nil)
+)
+
+const (
+	_minRequestInCh    = 10
+	_minRequestInTmpCh = 10
+	_minWorker         = 5
+)
+
+// ResponseFn a function used handle the response of http.Client.Do
+type ResponseFn = func(req *http.Request, resp *http.Response, err error)
+
+// AsyncClient asynchronous client interface
+type AsyncClient interface {
+	Do(req *http.Request, fn ResponseFn)
+}
+
+// AsyncClientConf client configure used to create an AsynClient instance
+type AsyncClientConf struct {
+	MinWorker          int
+	MaxRequestInCh     int
+	MaxRequestInTempCh int
+	MaxTickCount       int
+	TickWaitTime       time.Duration
+}
+
+type requestItem struct {
+	request *http.Request
+	fn      ResponseFn
+}
+
+type wormClient struct {
+	client        *http.Client
+	requestCh     chan *requestItem // 正式工 缓存通道
+	requestTempCh chan *requestItem // 临时工 缓存通道
+	maxTickCount  int
+	tickWaitTime  time.Duration
+}
+
+func (s *wormClient) Do(req *http.Request, fn ResponseFn) {
+	item := &requestItem{req, fn}
+	select {
+	case s.requestCh <- item:
+		logrus.Debugln("simepleClient.Do send request item by requestCh chan")
+	default:
+		select {
+		case s.requestTempCh <- item:
+			logrus.Debugln("simepleClient.Do send request item by requestTempCh chan")
+		default:
+			go func() {
+				s.do(item)
+				// watch requestTempCh to continue do work if needed.
+				// cancel loop if no item had watched in s.maxCyle * s.maxWaitTime.
+				for count := 0; count < s.maxTickCount; count++ {
+					select {
+					case item := <-s.requestTempCh:
+						// reset count to continue do work
+						count = 0
+						s.do(item)
+					default:
+						// sleeping to wait request item pass over to do work
+						time.Sleep(s.tickWaitTime)
+					}
+				}
+			}()
+		}
+	}
+}
+
+func (s *wormClient) starDotWork() {
+	for item := range s.requestCh {
+		s.do(item)
+	}
+}
+
+func (s *wormClient) do(req *requestItem) {
+	resp, err := s.client.Do(req.request)
+	req.fn(req.request, resp, err)
+}
+
+// NewAsyncClient create an AsyncClient instance
+func NewAsyncClient(client *http.Client, conf *AsyncClientConf) AsyncClient {
+	maxRequestInCh := _minRequestInCh
+	maxRequestInTempCh := _minRequestInTmpCh
+	if conf.MaxRequestInCh > _minRequestInCh {
+		maxRequestInCh = conf.MaxRequestInCh
+	}
+	if conf.MaxRequestInTempCh > _minRequestInTmpCh {
+		maxRequestInTempCh = conf.MaxRequestInTempCh
+	}
+	wc := &wormClient{
+		client:        client,
+		requestCh:     make(chan *requestItem, maxRequestInCh),
+		requestTempCh: make(chan *requestItem, maxRequestInTempCh),
+	}
+	numWorker := conf.MinWorker
+	if numWorker < _minWorker {
+		numWorker = _minWorker
+	}
+	logrus.Debugf("use %d backend worker to do the http request", numWorker)
+	// 启动 do work 正式工
+	for ; numWorker > 0; numWorker-- {
+		go wc.starDotWork()
+	}
+	return wc
+}

pkg/http/http.go

@@ -2,4 +2,6 @@
 // Use of this source code is governed by Apache License 2.0 that
 // can be found in the LICENSE file.
 
+// package http contain some custom help function for std http library.
+
 package http

pkg/obx/obx.go

@@ -0,0 +1,80 @@
+// Copyright 2023 ROC. All rights reserved.
+// Use of this source code is governed by a MIT style
+// license that can be found in the LICENSE file.
+
+// package obx contain some help function for OpenObserve.
+package obx
+
+import (
+	"bytes"
+	"net/http"
+
+	hx "github.com/rocboss/paopao-ce/pkg/http"
+)
+
+var (
+	_ OpenObserveClient = (*obxClient)(nil)
+)
+
+const (
+	_userAgent = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_4) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/81.0.4044.138 Safari/537.36"
+)
+
+// OpenObserveClient OpenObserve client interface
+type OpenObserveClient interface {
+	LogJson(data []byte)
+}
+
+// Config confiugre used for create a OpenObserveClient instance
+type Config struct {
+	Host         string
+	User         string
+	Password     string
+	Organization string
+	Stream       string
+	UserAgent    string
+	Secure       bool
+}
+
+type obxClient struct {
+	endpoint  string
+	user      string
+	password  string
+	userAgent string
+	respFn    hx.ResponseFn
+	client    hx.AsyncClient
+}
+
+func (c *Config) Endpoint() string {
+	schema := "http"
+	if c.Secure {
+		schema = "https"
+	}
+	return schema + "://" + c.Host + "/api/" + c.Organization + "/" + c.Stream + "/_json"
+}
+
+func (s *obxClient) LogJson(data []byte) {
+	req, err := http.NewRequest("POST", s.endpoint, bytes.NewReader(data))
+	if err != nil {
+		s.respFn(nil, nil, err)
+	}
+	req.SetBasicAuth(s.user, s.password)
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("User-Agent", s.userAgent)
+	s.client.Do(req, s.respFn)
+}
+
+// NewClient create OpenObserve client instance
+func NewClient(conf *Config, acc *hx.AsyncClientConf, fn hx.ResponseFn) OpenObserveClient {
+	userAgent := _userAgent
+	if conf.UserAgent != "" {
+		userAgent = conf.UserAgent
+	}
+	return &obxClient{
+		endpoint:  conf.Endpoint(),
+		user:      conf.User,
+		password:  conf.Password,
+		userAgent: userAgent,
+		client:    hx.NewAsyncClient(http.DefaultClient, acc),
+	}
+}