Compare commits

...

7 Commits

15 changed files with 382 additions and 77 deletions

95
exegroup/eghttp/actor.go Normal file
View File

@@ -0,0 +1,95 @@
package eghttp
import (
"context"
"net"
"net/http"
"strconv"
"sync/atomic"
"time"
)
const (
DefaultAddr = ":8080"
DefaultReadHeaderTimeout = time.Second
)
type Option interface {
apply(cfg *config)
}
func WithPort(port int) Option {
return optionFunc(func(cfg *config) {
cfg.server.Addr = net.JoinHostPort("", strconv.Itoa(port))
})
}
func WithHandler(handler http.Handler) Option {
return optionFunc(func(cfg *config) {
cfg.server.Handler = handler
})
}
func WithServer(server *http.Server) Option {
return optionFunc(func(cfg *config) {
cfg.server = server
})
}
func WithStartFn(fn func(server *http.Server) error) Option {
return optionFunc(func(cfg *config) {
cfg.startFn = fn
})
}
// WithServerOption 使用 fn 配置 [http.Server];
func WithServerOption(fn func(server *http.Server)) Option {
return optionFunc(func(cfg *config) {
fn(cfg.server)
})
}
type config struct {
server *http.Server
startFn func(server *http.Server) error
}
func newDefaultConfig() *config {
return &config{
server: &http.Server{
Addr: DefaultAddr,
ReadHeaderTimeout: DefaultReadHeaderTimeout,
},
startFn: func(server *http.Server) error {
return server.ListenAndServe()
},
}
}
type optionFunc func(cfg *config)
func (fn optionFunc) apply(cfg *config) {
fn(cfg)
}
// HTTPListenAndServe 创建 [http.Server] 并提供启动和停止函数;
func HTTPListenAndServe(opts ...Option) (func(ctx context.Context) error, func(ctx context.Context)) {
cfg := newDefaultConfig()
for _, opt := range opts {
opt.apply(cfg)
}
inShutdown := &atomic.Bool{}
c := make(chan error, 1)
goFunc := func(_ context.Context) error {
err := cfg.startFn(cfg.server)
if inShutdown.Load() {
err = <-c
}
return err
}
stopFunc := func(ctx context.Context) {
inShutdown.Store(true)
c <- cfg.server.Shutdown(ctx)
}
return goFunc, stopFunc
}

View File

@@ -1,33 +0,0 @@
package exegroup
import (
"context"
"net"
"net/http"
"strconv"
"sync/atomic"
"time"
)
// HTTPListenAndServe 提供 [http.Server] 的启动和停止函数;
func HTTPListenAndServe(port int, handler http.Handler) (func(ctx context.Context) error, func(ctx context.Context)) {
server := &http.Server{
Addr: net.JoinHostPort("", strconv.Itoa(port)),
Handler: handler,
ReadHeaderTimeout: time.Second,
}
inShutdown := &atomic.Bool{}
c := make(chan error, 1)
goFunc := func(_ context.Context) error {
err := server.ListenAndServe()
if inShutdown.Load() {
err = <-c
}
return err
}
stopFunc := func(ctx context.Context) {
inShutdown.Store(true)
c <- server.Shutdown(ctx)
}
return goFunc, stopFunc
}

141
httpdata/.golangci.yaml Normal file
View File

@@ -0,0 +1,141 @@
## 基于 golangci-lint@v1.52.2
run:
timeout: 1m
build-tags: [ ]
skip-dirs: [ ]
skip-files: [ ]
linters:
disable-all: true
enable:
- errcheck
- gosimple
- govet
- ineffassign
- staticcheck
- typecheck
- unused
- asasalint
- asciicheck
- bidichk
- bodyclose
- containedctx
- cyclop
- dupl
- durationcheck
- errname
- errorlint
- exhaustive
- exportloopref
- funlen
- gocheckcompilerdirectives
- gochecknoinits
- goconst
- gocritic
- gocyclo
- goimports
- gomnd
- goprintffuncname
- gosec
- lll
- loggercheck
- makezero
- nakedret
- nestif
- nilnil
- noctx
- nolintlint
- prealloc
- predeclared
- promlinter
- reassign
- revive
- rowserrcheck
- stylecheck
- tenv
- testableexamples
- testpackage
- tparallel
- unconvert
- unparam
- usestdlibvars
- wastedassign
- whitespace
linters-settings:
errcheck:
check-type-assertions: true
exclude-functions: [ ]
govet:
enable-all: true
disable: [ ]
cyclop:
max-complexity: 10
package-average: 0.0
dupl:
threshold: 150
exhaustive:
check:
- switch
- map
funlen:
lines: 100
statements: 60
gocritic:
disabled-checks:
- commentFormatting
settings:
captLocal:
paramsOnly: false
underef:
skipRecvDeref: false
gocyclo:
min-complexity: 20
gomnd:
ignored-functions:
- os.Chmod
- os.Mkdir
- os.MkdirAll
- os.OpenFile
- os.WriteFile
- prometheus.ExponentialBuckets
- prometheus.ExponentialBucketsRange
- prometheus.LinearBuckets
lll:
line-length: 240
nakedret:
max-func-lines: 10
nestif:
min-complexity: 5
predeclared:
ignore: ""
q: false
reassign:
patterns:
- ".*"
rowserrcheck:
packages:
- github.com/jmoiron/sqlx
tenv:
all: true
usestdlibvars:
time-month: true
time-layout: true
crypto-hash: true
default-rpc-path: true
os-dev-null: true
sql-isolation-level: true
tls-signature-scheme: true
constant-kind: true
syslog-priority: true
issues:
max-same-issues: 10
exclude-rules:
- source: "//noinspection"
linters: [ gocritic ]
- path: "_test\\.go"
linters:
- bodyclose
- dupl
- funlen
- goconst
- gosec
- noctx

14
httpdata/code.go Normal file
View File

@@ -0,0 +1,14 @@
package httpdata
type Code string
const (
CodeOK Code = "ok"
CodeError Code = "error"
CodeBadRequest Code = "bad_request"
CodeInternal Code = "internal"
CodeInternalUpstream Code = "internal.upstream"
CodeUnexpect Code = "unexpect"
CodeUnexpectPanic Code = "unexpect.panic"
CodeUnknown Code = "unknown"
)

22
httpdata/data.go Normal file
View File

@@ -0,0 +1,22 @@
package httpdata
type Response struct {
Data any `json:"data,omitempty"`
Code Code `json:"code"`
Message string `json:"message,omitempty"`
}
type PageData struct {
List []any `json:"list"`
PageIndex int `json:"pageIndex"` // >=1
PageSize int `json:"pageSize"` // >=1
Total int `json:"total"` // maybe 0
}
func NewOkResponse(data any) Response {
return Response{
Code: CodeOK,
Message: "",
Data: data,
}
}

42
httpdata/error.go Normal file
View File

@@ -0,0 +1,42 @@
package httpdata
import (
"errors"
)
type UniverseError struct {
Code Code
Message string
}
func NewUniverseError(code Code, message string) error {
return UniverseError{
Code: code,
Message: message,
}
}
func NewBadRequestError(message string) error {
return UniverseError{
Code: CodeBadRequest,
Message: message,
}
}
func (err UniverseError) Error() string {
if err.Message == "" {
return string(err.Code)
}
return string(err.Code) + ": " + err.Message
}
func IsUniverseError(err error) bool {
return errors.As(err, &UniverseError{})
}
func ToUniverseError(err error, code Code, message string) error {
if err == nil || IsUniverseError(err) {
return err
}
return NewUniverseError(code, message)
}

3
httpdata/go.mod Normal file
View File

@@ -0,0 +1,3 @@
module git.blauwelle.com/go/crate/httpdata
go 1.21.1

View File

@@ -54,8 +54,7 @@ log.Logger()
- `SetReportStack` 设置生成调用栈;
- `SetReportStackLevel` 当日志等级小于设定值时强制生成调用栈;
- `Reset` 把 Logger 恢复到初始状态;
- `AddBeforeExit` 新增调用 `Logger.Exit` 时优先执行的函数, 先增加的后执行;
- `SetExit` 设置 `Logger` 的退出函数(`Logger.Exit`), 当日志等级为 `LevelFatal` 时调用这个函数;
- `AddBeforeExit` 增加 Exit 在调用 [os.Exit] 前执行的函数, 先增加的后执行;
### 日志生成
@@ -85,11 +84,13 @@ log.Logger()
### 其他用法
关闭日志生成 `log.Logger().SetLevel(logsdk.LevelDisabled)`
关闭日志生成, `log.Logger().SetLevel(logsdk.LevelDisabled)`;
关闭强制生成调用栈 `log.Logger().SetReportStackLevel(logsdk.LevelDisabled)`
关闭强制生成调用栈, `log.Logger().SetReportStackLevel(logsdk.LevelDisabled)`;
mock 实现 mock 日志处理器对生成的日志进行处理
记录 panic, 在 `defer v = recover()` 后执行 `log.Z(context.Background(), v)`, 其中 `Z` 需要是附加调用栈的等级;
mock, 实现 mock 日志处理器对生成的日志进行处理;
---

View File

@@ -145,7 +145,8 @@ func Fatalf(ctx context.Context, format string, args ...any) {
globalLogger.AddCallerSkip(1).Fatalf(ctx, format, args...)
}
// Panicf 格式化输出 LevelPanic 等级的日志
// Panicf 格式化输出 LevelPanic 等级的日志,
// 即使 Logger 日志等级高于 LevelPanic 也会 panic.
func Panicf(ctx context.Context, format string, args ...any) {
globalLogger.AddCallerSkip(1).Panicf(ctx, format, args...)
}

View File

@@ -169,19 +169,19 @@ func (entry Entry) Fatalf(ctx context.Context, format string, args ...any) {
entry.logger.Exit(1)
}
// Panicf 格式化输出 LevelPanic 等级的日志
// Panicf 格式化输出 LevelPanic 等级的日志,
// 即使 Logger 日志等级高于 LevelPanic 也会 panic.
func (entry Entry) Panicf(ctx context.Context, format string, args ...any) {
entry.log(ctx, LevelPanic, fmt.Sprintf(format, args...))
}
func (entry Entry) log(ctx context.Context, level Level, message string) {
newEntry := entry.copy()
defer func() {
if level == LevelPanic {
panic(message)
}
}()
newEntry := entry.copy()
if newEntry.logger.GetLevel() < level {
return
}

View File

@@ -64,14 +64,14 @@ const (
// LevelCount 日志等级的个数
LevelCount = 7
// LevelOffset 是把 LevelInfo 等级的日志值偏移到零值的偏移量
LevelOffset = 4
// levelOffset 是把 LevelInfo 等级的日志值偏移到零值的偏移量
levelOffset = 4
)
const (
// LevelDisabled 表示不处理任何等级的日志,
// 其他日志等级的值越小表示日志严重程度越高.
LevelDisabled Level = iota - LevelOffset - 1
LevelDisabled Level = iota - levelOffset - 1
LevelPanic
LevelFatal
LevelError

View File

@@ -20,11 +20,11 @@ type entry = Entry
// Logger 中保存了日志所需的全局配置,
// 使用 Logger 处理日志.
type Logger struct {
exit func(code int) // protected by lock
beforeExit []func(code int) // protected by lock
levelProcessors levelProcessors // protected by lock
beforeExitFns []func() // protected by lock
levelProcessors levelProcessors // protected by lock
entry
lock sync.RWMutex
exitOnce sync.Once
level atomic.Int32
reportStackLevel atomic.Int32
callerSkip atomic.Int32
@@ -36,7 +36,7 @@ type Logger struct {
func (logger *Logger) AddProcessor(levels []Level, processor EntryProcessor) {
logger.lock.Lock()
for _, level := range levels {
logger.levelProcessors[level+LevelOffset] = append(logger.levelProcessors[level+LevelOffset], processor)
logger.levelProcessors[level+levelOffset] = append(logger.levelProcessors[level+levelOffset], processor)
}
logger.lock.Unlock()
}
@@ -98,8 +98,7 @@ func (logger *Logger) SetReportStackLevel(level Level) {
// Reset 把 Logger 重置到初始状态
func (logger *Logger) Reset() {
logger.lock.Lock()
logger.exit = nil
logger.beforeExit = nil
logger.beforeExitFns = nil
logger.levelProcessors = levelProcessors{}
logger.lock.Unlock()
@@ -110,41 +109,38 @@ func (logger *Logger) Reset() {
logger.SetReportStack(false)
}
// Exit 退出程序, 执行的具体过程可以通过 SetExit 指定
// BeforeExit 按照先添加后执行的顺序执行 AddBeforeExit 添加的函数,
// AddBeforeExit 只会执行1次.
func (logger *Logger) BeforeExit() {
logger.exitOnce.Do(func() {
logger.lock.RLock()
beforeExitFns := make([]func(), len(logger.beforeExitFns))
copy(beforeExitFns, logger.beforeExitFns)
logger.lock.RUnlock()
for i := len(beforeExitFns) - 1; i >= 0; i-- {
beforeExitFns[i]()
}
})
}
// Exit 执行 BeforeExit 后调用 [os.Exit]退出程序.
func (logger *Logger) Exit(code int) {
logger.lock.RLock()
exit := logger.exit
beforeExit := make([]func(int), len(logger.beforeExit))
copy(beforeExit, logger.beforeExit)
logger.lock.RUnlock()
for i := len(beforeExit) - 1; i >= 0; i-- {
beforeExit[i](code)
}
if exit == nil {
exit = os.Exit
}
exit(code)
logger.BeforeExit()
os.Exit(code)
}
// SetExit 指定退出程序时执行的函数
func (logger *Logger) SetExit(fn func(code int)) {
logger.lock.Lock()
logger.exit = fn
logger.lock.Unlock()
}
// AddBeforeExit 增加 Exit 在调用 SetExit 指定的函数前执行的函数,
// AddBeforeExit 增加 Exit 在调用 [os.Exit] 前执行的函数,
// 先增加的后执行.
func (logger *Logger) AddBeforeExit(fn ...func(code int)) {
func (logger *Logger) AddBeforeExit(fn ...func()) {
logger.lock.Lock()
logger.beforeExit = append(logger.beforeExit, fn...)
logger.beforeExitFns = append(logger.beforeExitFns, fn...)
logger.lock.Unlock()
}
func (logger *Logger) getLevelProcessors(level Level) []EntryProcessor {
logger.lock.RLock()
defer logger.lock.RUnlock()
return logger.levelProcessors[level+LevelOffset]
return logger.levelProcessors[level+levelOffset]
}
func (logger *Logger) newEntry() Entry {

View File

@@ -8,6 +8,8 @@ const (
maximumFrames = 32
getCallerSkipOffset = 2
entrySkipOffset = 2
runtimeMain = "runtime.main"
)
// Frame 调用相关信息
@@ -44,6 +46,9 @@ func getStack(skip, maximumFrames int) []Frame {
for {
frame, more := frames.Next()
if frame.PC != 0 {
if frame.Function == runtimeMain {
break
}
stack = append(stack, Frame{
Function: frame.Function,
File: frame.File,

View File

@@ -22,7 +22,8 @@ func newConfig(opts ...Option) *config {
func defaultConfig() *config {
return &config{
hasPool: false,
hasPool: false,
defaultSpan: false,
}
}
@@ -34,6 +35,13 @@ func WithBufferPool(pool logjson.BytesBufferPool) Option {
})
}
// WithDefaultSpan 设置当 span 没有在记录时创建新 span
func WithDefaultSpan(defaultSpan bool) Option {
return optionFunc(func(cfg *config) {
cfg.defaultSpan = defaultSpan
})
}
// Option 配置 Processor
type Option interface {
apply(cfg *config)
@@ -42,6 +50,7 @@ type Option interface {
type config struct {
bytesBufferPool logjson.BytesBufferPool
hasPool bool
defaultSpan bool
}
type optionFunc func(cfg *config)

View File

@@ -9,6 +9,7 @@ import (
"git.blauwelle.com/go/crate/log/logsdk"
"git.blauwelle.com/go/crate/log/logsdk/logjson"
"go.opentelemetry.io/otel"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/codes"
semconv "go.opentelemetry.io/otel/semconv/v1.17.0"
@@ -19,7 +20,8 @@ import (
func New(opts ...Option) *Processor {
cfg := newConfig(opts...)
return &Processor{
bufferPool: cfg.bytesBufferPool,
bufferPool: cfg.bytesBufferPool,
defaultSpan: cfg.defaultSpan,
}
}
@@ -27,11 +29,18 @@ var _ logsdk.EntryProcessor = &Processor{}
// Processor 用于把日志和 opentelemetry 对接
type Processor struct {
bufferPool logjson.BytesBufferPool
bufferPool logjson.BytesBufferPool
defaultSpan bool
}
func (processor *Processor) Process(ctx context.Context, entry logsdk.ReadonlyEntry) {
span := trace.SpanFromContext(ctx)
if !span.IsRecording() {
if processor.defaultSpan {
ctx, span = otel.Tracer("git.blauwelle.com/go/crate/logotel").Start(ctx, "default") //nolint:ineffassign,staticcheck,wastedassign
defer span.End()
}
}
if !span.IsRecording() {
return
}