fix recorders
This commit is contained in:
+19
-6
@@ -4,24 +4,34 @@ import (
|
||||
"context"
|
||||
"io"
|
||||
|
||||
"github.com/go-gost/core/metrics"
|
||||
"github.com/go-gost/core/recorder"
|
||||
xmetrics "github.com/go-gost/x/metrics"
|
||||
)
|
||||
|
||||
type fileRecorderOptions struct {
|
||||
sep string
|
||||
recorder string
|
||||
sep string
|
||||
}
|
||||
|
||||
type FileRecorderOption func(opts *fileRecorderOptions)
|
||||
|
||||
func SepRecorderOption(sep string) FileRecorderOption {
|
||||
func RecorderFileRecorderOption(recorder string) FileRecorderOption {
|
||||
return func(opts *fileRecorderOptions) {
|
||||
opts.recorder = recorder
|
||||
}
|
||||
}
|
||||
|
||||
func SepFileRecorderOption(sep string) FileRecorderOption {
|
||||
return func(opts *fileRecorderOptions) {
|
||||
opts.sep = sep
|
||||
}
|
||||
}
|
||||
|
||||
type fileRecorder struct {
|
||||
out io.WriteCloser
|
||||
sep string
|
||||
recorder string
|
||||
out io.WriteCloser
|
||||
sep string
|
||||
}
|
||||
|
||||
// FileRecorder records data to file.
|
||||
@@ -32,12 +42,15 @@ func FileRecorder(out io.WriteCloser, opts ...FileRecorderOption) recorder.Recor
|
||||
}
|
||||
|
||||
return &fileRecorder{
|
||||
out: out,
|
||||
sep: options.sep,
|
||||
recorder: options.recorder,
|
||||
out: out,
|
||||
sep: options.sep,
|
||||
}
|
||||
}
|
||||
|
||||
func (r *fileRecorder) Record(ctx context.Context, b []byte, opts ...recorder.RecordOption) error {
|
||||
xmetrics.GetCounter(xmetrics.MetricRecorderRecordsCounter, metrics.Labels{"recorder": r.recorder}).Inc()
|
||||
|
||||
if _, err := r.out.Write(b); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
+14
-2
@@ -8,16 +8,25 @@ import (
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/go-gost/core/metrics"
|
||||
"github.com/go-gost/core/recorder"
|
||||
xmetrics "github.com/go-gost/x/metrics"
|
||||
)
|
||||
|
||||
type httpRecorderOptions struct {
|
||||
timeout time.Duration
|
||||
header http.Header
|
||||
recorder string
|
||||
timeout time.Duration
|
||||
header http.Header
|
||||
}
|
||||
|
||||
type HTTPRecorderOption func(opts *httpRecorderOptions)
|
||||
|
||||
func RecorderHTTPRecorderOption(recorder string) HTTPRecorderOption {
|
||||
return func(opts *httpRecorderOptions) {
|
||||
opts.recorder = recorder
|
||||
}
|
||||
}
|
||||
|
||||
func TimeoutHTTPRecorderOption(timeout time.Duration) HTTPRecorderOption {
|
||||
return func(opts *httpRecorderOptions) {
|
||||
opts.timeout = timeout
|
||||
@@ -31,6 +40,7 @@ func HeaderHTTPRecorderOption(header http.Header) HTTPRecorderOption {
|
||||
}
|
||||
|
||||
type httpRecorder struct {
|
||||
recorder string
|
||||
url string
|
||||
httpClient *http.Client
|
||||
header http.Header
|
||||
@@ -60,6 +70,8 @@ func HTTPRecorder(url string, opts ...HTTPRecorderOption) recorder.Recorder {
|
||||
}
|
||||
|
||||
func (r *httpRecorder) Record(ctx context.Context, b []byte, opts ...recorder.RecordOption) error {
|
||||
xmetrics.GetCounter(xmetrics.MetricRecorderRecordsCounter, metrics.Labels{"recorder": r.recorder}).Inc()
|
||||
|
||||
req, err := http.NewRequest(http.MethodPost, r.url, bytes.NewReader(b))
|
||||
if err != nil {
|
||||
return err
|
||||
|
||||
+27
-6
@@ -3,11 +3,14 @@ package recorder
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/go-gost/core/metrics"
|
||||
"github.com/go-gost/core/recorder"
|
||||
xmetrics "github.com/go-gost/x/metrics"
|
||||
"github.com/go-redis/redis/v8"
|
||||
)
|
||||
|
||||
type redisRecorderOptions struct {
|
||||
recorder string
|
||||
db int
|
||||
username string
|
||||
password string
|
||||
@@ -15,6 +18,12 @@ type redisRecorderOptions struct {
|
||||
}
|
||||
type RedisRecorderOption func(opts *redisRecorderOptions)
|
||||
|
||||
func RecorderRedisRecorderOption(recorder string) RedisRecorderOption {
|
||||
return func(opts *redisRecorderOptions) {
|
||||
opts.recorder = recorder
|
||||
}
|
||||
}
|
||||
|
||||
func DBRedisRecorderOption(db int) RedisRecorderOption {
|
||||
return func(opts *redisRecorderOptions) {
|
||||
opts.db = db
|
||||
@@ -39,8 +48,9 @@ func KeyRedisRecorderOption(key string) RedisRecorderOption {
|
||||
}
|
||||
|
||||
type redisSetRecorder struct {
|
||||
client *redis.Client
|
||||
key string
|
||||
recorder string
|
||||
client *redis.Client
|
||||
key string
|
||||
}
|
||||
|
||||
// RedisSetRecorder records data to a redis set.
|
||||
@@ -51,6 +61,7 @@ func RedisSetRecorder(addr string, opts ...RedisRecorderOption) recorder.Recorde
|
||||
}
|
||||
|
||||
return &redisSetRecorder{
|
||||
recorder: options.recorder,
|
||||
client: redis.NewClient(&redis.Options{
|
||||
Addr: addr,
|
||||
Password: options.password,
|
||||
@@ -61,6 +72,8 @@ func RedisSetRecorder(addr string, opts ...RedisRecorderOption) recorder.Recorde
|
||||
}
|
||||
|
||||
func (r *redisSetRecorder) Record(ctx context.Context, b []byte, opts ...recorder.RecordOption) error {
|
||||
xmetrics.GetCounter(xmetrics.MetricRecorderRecordsCounter, metrics.Labels{"recorder": r.recorder}).Inc()
|
||||
|
||||
if r.key == "" {
|
||||
return nil
|
||||
}
|
||||
@@ -73,8 +86,9 @@ func (r *redisSetRecorder) Close() error {
|
||||
}
|
||||
|
||||
type redisListRecorder struct {
|
||||
client *redis.Client
|
||||
key string
|
||||
recorder string
|
||||
client *redis.Client
|
||||
key string
|
||||
}
|
||||
|
||||
// RedisListRecorder records data to a redis list.
|
||||
@@ -85,6 +99,7 @@ func RedisListRecorder(addr string, opts ...RedisRecorderOption) recorder.Record
|
||||
}
|
||||
|
||||
return &redisListRecorder{
|
||||
recorder: options.recorder,
|
||||
client: redis.NewClient(&redis.Options{
|
||||
Addr: addr,
|
||||
Password: options.password,
|
||||
@@ -95,6 +110,8 @@ func RedisListRecorder(addr string, opts ...RedisRecorderOption) recorder.Record
|
||||
}
|
||||
|
||||
func (r *redisListRecorder) Record(ctx context.Context, b []byte, opts ...recorder.RecordOption) error {
|
||||
xmetrics.GetCounter(xmetrics.MetricRecorderRecordsCounter, metrics.Labels{"recorder": r.recorder}).Inc()
|
||||
|
||||
if r.key == "" {
|
||||
return nil
|
||||
}
|
||||
@@ -107,8 +124,9 @@ func (r *redisListRecorder) Close() error {
|
||||
}
|
||||
|
||||
type redisSortedSetRecorder struct {
|
||||
client *redis.Client
|
||||
key string
|
||||
recorder string
|
||||
client *redis.Client
|
||||
key string
|
||||
}
|
||||
|
||||
// RedisSortedSetRecorder records data to a redis sorted set.
|
||||
@@ -119,6 +137,7 @@ func RedisSortedSetRecorder(addr string, opts ...RedisRecorderOption) recorder.R
|
||||
}
|
||||
|
||||
return &redisSortedSetRecorder{
|
||||
recorder: options.recorder,
|
||||
client: redis.NewClient(&redis.Options{
|
||||
Addr: addr,
|
||||
Password: options.password,
|
||||
@@ -129,6 +148,8 @@ func RedisSortedSetRecorder(addr string, opts ...RedisRecorderOption) recorder.R
|
||||
}
|
||||
|
||||
func (r *redisSortedSetRecorder) Record(ctx context.Context, b []byte, opts ...recorder.RecordOption) error {
|
||||
xmetrics.GetCounter(xmetrics.MetricRecorderRecordsCounter, metrics.Labels{"recorder": r.recorder}).Inc()
|
||||
|
||||
if r.key == "" {
|
||||
return nil
|
||||
}
|
||||
|
||||
+19
-6
@@ -6,16 +6,25 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/go-gost/core/logger"
|
||||
"github.com/go-gost/core/metrics"
|
||||
"github.com/go-gost/core/recorder"
|
||||
xmetrics "github.com/go-gost/x/metrics"
|
||||
)
|
||||
|
||||
type tcpRecorderOptions struct {
|
||||
timeout time.Duration
|
||||
log logger.Logger
|
||||
recorder string
|
||||
timeout time.Duration
|
||||
log logger.Logger
|
||||
}
|
||||
|
||||
type TCPRecorderOption func(opts *tcpRecorderOptions)
|
||||
|
||||
func RecorderTCPRecorderOption(recorder string) TCPRecorderOption {
|
||||
return func(opts *tcpRecorderOptions) {
|
||||
opts.recorder = recorder
|
||||
}
|
||||
}
|
||||
|
||||
func TimeoutTCPRecorderOption(timeout time.Duration) TCPRecorderOption {
|
||||
return func(opts *tcpRecorderOptions) {
|
||||
opts.timeout = timeout
|
||||
@@ -29,9 +38,10 @@ func LogTCPRecorderOption(log logger.Logger) TCPRecorderOption {
|
||||
}
|
||||
|
||||
type tcpRecorder struct {
|
||||
addr string
|
||||
dialer *net.Dialer
|
||||
log logger.Logger
|
||||
recorder string
|
||||
addr string
|
||||
dialer *net.Dialer
|
||||
log logger.Logger
|
||||
}
|
||||
|
||||
// TCPRecorder records data to TCP service.
|
||||
@@ -42,7 +52,8 @@ func TCPRecorder(addr string, opts ...TCPRecorderOption) recorder.Recorder {
|
||||
}
|
||||
|
||||
return &tcpRecorder{
|
||||
addr: addr,
|
||||
recorder: options.recorder,
|
||||
addr: addr,
|
||||
dialer: &net.Dialer{
|
||||
Timeout: options.timeout,
|
||||
},
|
||||
@@ -51,6 +62,8 @@ func TCPRecorder(addr string, opts ...TCPRecorderOption) recorder.Recorder {
|
||||
}
|
||||
|
||||
func (r *tcpRecorder) Record(ctx context.Context, b []byte, opts ...recorder.RecordOption) error {
|
||||
xmetrics.GetCounter(xmetrics.MetricRecorderRecordsCounter, metrics.Labels{"recorder": r.recorder}).Inc()
|
||||
|
||||
c, err := r.dialer.DialContext(ctx, "tcp", r.addr)
|
||||
if err != nil {
|
||||
return err
|
||||
|
||||
Reference in New Issue
Block a user