Compare commits

..

2 Commits

Author SHA1 Message Date
b6d2d459c5 fixed metadata hooks
Some checks failed
lint / lint (pull_request) Failing after 51s
test / test (pull_request) Successful in 2m27s
coverage / build (pull_request) Failing after 6m20s
2025-04-23 20:50:44 +03:00
c2d5dd4ffd move hooks
Some checks failed
lint / lint (pull_request) Failing after 51s
coverage / build (pull_request) Failing after 1m17s
test / test (pull_request) Successful in 3m31s
2025-04-22 18:22:34 +03:00
66 changed files with 750 additions and 23155 deletions

View File

@@ -3,16 +3,14 @@ name: coverage
on:
push:
branches: [ main, v3, v4 ]
paths-ignore:
- '.github/**'
- '.gitea/**'
pull_request:
branches: [ main, v3, v4 ]
# Allows you to run this workflow manually from the Actions tab
workflow_dispatch:
jobs:
build:
if: github.server_url != 'https://github.com'
runs-on: ubuntu-latest
steps:
- name: checkout code
@@ -24,7 +22,7 @@ jobs:
uses: actions/setup-go@v5
with:
cache-dependency-path: "**/*.sum"
go-version: 'stable'
go-version: 'stable'
- name: test coverage
run: |
@@ -41,8 +39,8 @@ jobs:
name: autocommit
with:
commit_message: Apply Code Coverage Badge
skip_fetch: false
skip_checkout: false
skip_fetch: true
skip_checkout: true
file_pattern: ./README.md
- name: push
@@ -50,4 +48,4 @@ jobs:
uses: ad-m/github-push-action@master
with:
github_token: ${{ github.token }}
branch: ${{ github.ref }}
branch: ${{ github.ref }}

View File

@@ -3,10 +3,10 @@ name: lint
on:
pull_request:
types: [opened, reopened, synchronize]
branches: [ master, v3, v4 ]
paths-ignore:
- '.github/**'
- '.gitea/**'
branches:
- master
- v3
- v4
jobs:
lint:
@@ -20,10 +20,10 @@ jobs:
uses: actions/setup-go@v5
with:
cache-dependency-path: "**/*.sum"
go-version: 'stable'
go-version: 'stable'
- name: setup deps
run: go get -v ./...
- name: run lint
uses: golangci/golangci-lint-action@v6
uses: https://github.com/golangci/golangci-lint-action@v6
with:
version: 'latest'

View File

@@ -3,12 +3,15 @@ name: test
on:
pull_request:
types: [opened, reopened, synchronize]
branches: [ master, v3, v4 ]
branches:
- master
- v3
- v4
push:
branches: [ master, v3, v4 ]
paths-ignore:
- '.github/**'
- '.gitea/**'
branches:
- master
- v3
- v4
jobs:
test:

View File

@@ -3,12 +3,15 @@ name: test
on:
pull_request:
types: [opened, reopened, synchronize]
branches: [ master, v3, v4 ]
branches:
- master
- v3
- v4
push:
branches: [ master, v3, v4 ]
paths-ignore:
- '.github/**'
- '.gitea/**'
branches:
- master
- v3
- v4
jobs:
test:
@@ -32,19 +35,19 @@ jobs:
go-version: 'stable'
- name: setup go work
env:
GOWORK: ${{ github.workspace }}/go.work
GOWORK: /workspace/${{ github.repository_owner }}/go.work
run: |
go work init
go work use .
go work use micro-tests
- name: setup deps
env:
GOWORK: ${{ github.workspace }}/go.work
GOWORK: /workspace/${{ github.repository_owner }}/go.work
run: go get -v ./...
- name: run tests
env:
INTEGRATION_TESTS: yes
GOWORK: ${{ github.workspace }}/go.work
GOWORK: /workspace/${{ github.repository_owner }}/go.work
run: |
cd micro-tests
go test -mod readonly -v ./... || true

View File

@@ -1,94 +0,0 @@
name: sync
on:
schedule:
- cron: '*/5 * * * *'
# Allows you to run this workflow manually from the Actions tab
workflow_dispatch:
jobs:
sync:
if: github.server_url != 'https://github.com'
runs-on: ubuntu-latest
steps:
- name: init
run: |
git config --global user.email "vtolstov <vtolstov@users.noreply.github.com>"
git config --global user.name "github-actions[bot]"
echo "machine git.unistack.org login vtolstov password ${{ secrets.TOKEN_GITEA }}" >> /root/.netrc
echo "machine github.com login vtolstov password ${{ secrets.TOKEN_GITHUB }}" >> /root/.netrc
- name: check master
id: check_master
run: |
src_hash=$(git ls-remote https://github.com/${GITHUB_REPOSITORY} refs/heads/master | cut -f1)
dst_hash=$(git ls-remote ${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY} refs/heads/master | cut -f1)
echo "src_hash=$src_hash"
echo "dst_hash=$dst_hash"
if [ "$src_hash" != "$dst_hash" ]; then
echo "sync_needed=true" >> $GITHUB_OUTPUT
else
echo "sync_needed=false" >> $GITHUB_OUTPUT
fi
- name: sync master
if: steps.check_master.outputs.sync_needed == 'true'
run: |
git clone --filter=blob:none --filter=tree:0 --branch master --single-branch ${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY} repo
cd repo
git remote add --no-tags --fetch --track master upstream https://github.com/${GITHUB_REPOSITORY}
git pull --rebase upstream master
git push upstream master --progress
git push origin master --progress
cd ../
rm -rf repo
- name: check v3
id: check_v3
run: |
src_hash=$(git ls-remote https://github.com/${GITHUB_REPOSITORY} refs/heads/v3 | cut -f1)
dst_hash=$(git ls-remote ${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY} refs/heads/v3 | cut -f1)
echo "src_hash=$src_hash"
echo "dst_hash=$dst_hash"
if [ "$src_hash" != "$dst_hash" ]; then
echo "sync_needed=true" >> $GITHUB_OUTPUT
else
echo "sync_needed=false" >> $GITHUB_OUTPUT
fi
- name: sync v3
if: steps.check_v3.outputs.sync_needed == 'true'
run: |
git clone --filter=blob:none --filter=tree:0 --branch v3 --single-branch ${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY} repo
cd repo
git remote add --no-tags --fetch --track v3 upstream https://github.com/${GITHUB_REPOSITORY}
git pull --rebase upstream v3
git push upstream v3 --progress
git push origin v3 --progress
cd ../
rm -rf repo
- name: check v4
id: check_v4
run: |
src_hash=$(git ls-remote https://github.com/${GITHUB_REPOSITORY} refs/heads/v4 | cut -f1)
dst_hash=$(git ls-remote ${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY} refs/heads/v4 | cut -f1)
echo "src_hash=$src_hash"
echo "dst_hash=$dst_hash"
if [ "$src_hash" != "$dst_hash" ]; then
echo "sync_needed=true" >> $GITHUB_OUTPUT
else
echo "sync_needed=false" >> $GITHUB_OUTPUT
fi
- name: sync v4
if: steps.check_v4.outputs.sync_needed == 'true'
run: |
git clone --filter=blob:none --filter=tree:0 --branch v4 --single-branch ${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY} repo
cd repo
git remote add --no-tags --fetch --track v4 upstream https://github.com/${GITHUB_REPOSITORY}
git pull --rebase upstream v4
git push upstream v4 --progress
git push origin v4 --progress
cd ../
rm -rf repo

View File

@@ -1,5 +1,5 @@
run:
concurrency: 8
timeout: 5m
deadline: 5m
issues-exit-code: 1
tests: true

View File

@@ -1,5 +1,5 @@
# Micro
![Coverage](https://img.shields.io/badge/Coverage-33.6%25-yellow)
![Coverage](https://img.shields.io/badge/Coverage-44.3%25-yellow)
[![License](https://img.shields.io/:license-apache-blue.svg)](https://opensource.org/licenses/Apache-2.0)
[![Doc](https://img.shields.io/badge/go.dev-reference-007d9c?logo=go&logoColor=white&style=flat-square)](https://pkg.go.dev/go.unistack.org/micro/v4?tab=overview)
[![Status](https://git.unistack.org/unistack-org/micro/actions/workflows/job_tests.yml/badge.svg?branch=v4)](https://git.unistack.org/unistack-org/micro/actions?query=workflow%3Abuild+branch%3Av4+event%3Apush)

15
SECURITY.md Normal file
View File

@@ -0,0 +1,15 @@
# Security Policy
## Supported Versions
Use this section to tell people about which versions of your project are
currently being supported with security updates.
| Version | Supported |
| ------- | ------------------ |
| 3.7.x | :white_check_mark: |
| < 3.7.0 | :x: |
## Reporting a Vulnerability
If you find any issue, please create github issue in this repo

View File

@@ -22,8 +22,8 @@ type Broker struct {
subscribers map[string][]*Subscriber
addr string
opts broker.Options
mu sync.RWMutex
connected bool
sync.RWMutex
connected bool
}
type memoryMessage struct {
@@ -72,9 +72,9 @@ func (b *Broker) newCodec(ct string) (codec.Codec, error) {
if idx := strings.IndexRune(ct, ';'); idx >= 0 {
ct = ct[:idx]
}
b.mu.RLock()
b.RLock()
c, ok := b.opts.Codecs[ct]
b.mu.RUnlock()
b.RUnlock()
if ok {
return c, nil
}
@@ -96,8 +96,8 @@ func (b *Broker) Connect(ctx context.Context) error {
default:
}
b.mu.Lock()
defer b.mu.Unlock()
b.Lock()
defer b.Unlock()
if b.connected {
return nil
@@ -126,8 +126,8 @@ func (b *Broker) Disconnect(ctx context.Context) error {
default:
}
b.mu.Lock()
defer b.mu.Unlock()
b.Lock()
defer b.Unlock()
if !b.connected {
return nil
@@ -159,9 +159,6 @@ func (b *Broker) Init(opts ...broker.Option) error {
func (b *Broker) NewMessage(ctx context.Context, hdr metadata.Metadata, body interface{}, opts ...broker.PublishOption) (broker.Message, error) {
options := broker.NewPublishOptions(opts...)
if options.ContentType == "" {
options.ContentType = b.opts.ContentType
}
m := &memoryMessage{ctx: ctx, hdr: hdr, opts: options}
c, err := b.newCodec(m.opts.ContentType)
if err == nil {
@@ -183,12 +180,12 @@ func (b *Broker) fnPublish(ctx context.Context, topic string, messages ...broker
}
func (b *Broker) publish(ctx context.Context, topic string, messages ...broker.Message) error {
b.mu.RLock()
b.RLock()
if !b.connected {
b.mu.RUnlock()
b.RUnlock()
return broker.ErrNotConnected
}
b.mu.RUnlock()
b.RUnlock()
select {
case <-ctx.Done():
@@ -196,9 +193,9 @@ func (b *Broker) publish(ctx context.Context, topic string, messages ...broker.M
default:
}
b.mu.RLock()
b.RLock()
subs, ok := b.subscribers[topic]
b.mu.RUnlock()
b.RUnlock()
if !ok {
return nil
}
@@ -255,12 +252,12 @@ func (b *Broker) fnSubscribe(ctx context.Context, topic string, handler interfac
return nil, err
}
b.mu.RLock()
b.RLock()
if !b.connected {
b.mu.RUnlock()
b.RUnlock()
return nil, broker.ErrNotConnected
}
b.mu.RUnlock()
b.RUnlock()
sid, err := id.New()
if err != nil {
@@ -278,13 +275,13 @@ func (b *Broker) fnSubscribe(ctx context.Context, topic string, handler interfac
ctx: ctx,
}
b.mu.Lock()
b.Lock()
b.subscribers[topic] = append(b.subscribers[topic], sub)
b.mu.Unlock()
b.Unlock()
go func() {
<-sub.exit
b.mu.Lock()
b.Lock()
newSubscribers := make([]*Subscriber, 0, len(b.subscribers)-1)
for _, sb := range b.subscribers[topic] {
if sb.id == sub.id {
@@ -293,7 +290,7 @@ func (b *Broker) fnSubscribe(ctx context.Context, topic string, handler interfac
newSubscribers = append(newSubscribers, sb)
}
b.subscribers[topic] = newSubscribers
b.mu.Unlock()
b.Unlock()
}()
return sub, nil

View File

@@ -14,16 +14,16 @@ type NoopBroker struct {
funcPublish FuncPublish
funcSubscribe FuncSubscribe
opts Options
mu sync.RWMutex
sync.RWMutex
}
func (b *NoopBroker) newCodec(ct string) (codec.Codec, error) {
if idx := strings.IndexRune(ct, ';'); idx >= 0 {
ct = ct[:idx]
}
b.mu.RLock()
b.RLock()
c, ok := b.opts.Codecs[ct]
b.mu.RUnlock()
b.RUnlock()
if ok {
return c, nil
}
@@ -128,9 +128,6 @@ func (m *noopMessage) Unmarshal(dst interface{}, opts ...codec.Option) error {
func (b *NoopBroker) NewMessage(ctx context.Context, hdr metadata.Metadata, body interface{}, opts ...PublishOption) (Message, error) {
options := NewPublishOptions(opts...)
if options.ContentType == "" {
options.ContentType = b.opts.ContentType
}
m := &noopMessage{ctx: ctx, hdr: hdr, opts: options}
c, err := b.newCodec(m.opts.ContentType)
if err == nil {

View File

@@ -45,9 +45,6 @@ type Options struct {
// GracefulTimeout contains time to wait to finish in flight requests
GracefulTimeout time.Duration
// ContentType will be used if no content-type set when creating message
ContentType string
}
// NewOptions create new Options
@@ -60,19 +57,14 @@ func NewOptions(opts ...Option) Options {
Codecs: make(map[string]codec.Codec),
Tracer: tracer.DefaultTracer,
GracefulTimeout: DefaultGracefulTimeout,
ContentType: DefaultContentType,
}
for _, o := range opts {
o(&options)
}
return options
}
// DefaultContentType is the default content-type if not specified
var DefaultContentType = ""
// Context sets the context option
func Context(ctx context.Context) Option {
return func(o *Options) {
@@ -80,13 +72,6 @@ func Context(ctx context.Context) Option {
}
}
// ContentType used by default if not specified
func ContentType(ct string) Option {
return func(o *Options) {
o.ContentType = ct
}
}
// PublishOptions struct
type PublishOptions struct {
// ContentType for message body

View File

@@ -3,6 +3,8 @@ package codec
import (
"errors"
"gopkg.in/yaml.v3"
)
var (
@@ -66,10 +68,10 @@ func (m *RawMessage) MarshalYAML() ([]byte, error) {
}
// UnmarshalYAML sets *m to a copy of data.
func (m *RawMessage) UnmarshalYAML(data []byte) error {
func (m *RawMessage) UnmarshalYAML(n *yaml.Node) error {
if m == nil {
return errors.New("RawMessage UnmarshalYAML on nil pointer")
}
*m = append((*m)[0:0], data...)
*m = append((*m)[0:0], []byte(n.Value)...)
return nil
}

View File

@@ -1,5 +1,7 @@
package codec
import "gopkg.in/yaml.v3"
// Frame gives us the ability to define raw data to send over the pipes
type Frame struct {
Data []byte
@@ -26,8 +28,8 @@ func (m *Frame) MarshalYAML() ([]byte, error) {
}
// UnmarshalYAML set frame data
func (m *Frame) UnmarshalYAML(data []byte) error {
m.Data = append((m.Data)[0:0], data...)
func (m *Frame) UnmarshalYAML(n *yaml.Node) error {
m.Data = []byte(n.Value)
return nil
}

6
go.mod
View File

@@ -6,19 +6,19 @@ require (
dario.cat/mergo v1.0.1
github.com/DATA-DOG/go-sqlmock v1.5.2
github.com/KimMachineGun/automemlimit v0.7.0
github.com/goccy/go-yaml v1.17.1
github.com/ash3in/uuidv8 v1.2.0
github.com/google/uuid v1.6.0
github.com/matoous/go-nanoid v1.5.1
github.com/patrickmn/go-cache v2.1.0+incompatible
github.com/silas/dag v0.0.0-20220518035006-a7e85ada93c5
github.com/spf13/cast v1.7.1
github.com/stretchr/testify v1.10.0
go.uber.org/atomic v1.11.0
go.uber.org/automaxprocs v1.6.0
go.unistack.org/micro-proto/v4 v4.1.0
golang.org/x/sync v0.10.0
google.golang.org/grpc v1.69.4
google.golang.org/protobuf v1.36.3
gopkg.in/yaml.v3 v3.0.1
)
require (
@@ -26,9 +26,9 @@ require (
github.com/pbnjay/memory v0.0.0-20210728143218-7b4eea64cf58 // indirect
github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
github.com/rogpeppe/go-internal v1.13.1 // indirect
github.com/stretchr/testify v1.10.0 // indirect
golang.org/x/net v0.34.0 // indirect
golang.org/x/sys v0.29.0 // indirect
google.golang.org/genproto/googleapis/rpc v0.0.0-20241216192217-9240e9c98484 // indirect
gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c // indirect
gopkg.in/yaml.v3 v3.0.1 // indirect
)

4
go.sum
View File

@@ -4,12 +4,12 @@ github.com/DATA-DOG/go-sqlmock v1.5.2 h1:OcvFkGmslmlZibjAjaHm3L//6LiuBgolP7Oputl
github.com/DATA-DOG/go-sqlmock v1.5.2/go.mod h1:88MAG/4G7SMwSE3CeA0ZKzrT5CiOU3OJ+JlNzwDqpNU=
github.com/KimMachineGun/automemlimit v0.7.0 h1:7G06p/dMSf7G8E6oq+f2uOPuVncFyIlDI/pBWK49u88=
github.com/KimMachineGun/automemlimit v0.7.0/go.mod h1:QZxpHaGOQoYvFhv/r4u3U0JTC2ZcOwbSr11UZF46UBM=
github.com/ash3in/uuidv8 v1.2.0 h1:2oogGdtCPwaVtyvPPGin4TfZLtOGE5F+W++E880G6SI=
github.com/ash3in/uuidv8 v1.2.0/go.mod h1:BnU0wJBxnzdEKmVg4xckBkD+VZuecTFTUP3M0dWgyY4=
github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM=
github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/frankban/quicktest v1.14.6 h1:7Xjx+VpznH+oBnejlPUj8oUpdxnVs4f8XU8WnHkI4W8=
github.com/frankban/quicktest v1.14.6/go.mod h1:4ptaffx2x8+WTWXmUCuVU6aPUX1/Mz7zb5vbUoiM6w0=
github.com/goccy/go-yaml v1.17.1 h1:LI34wktB2xEE3ONG/2Ar54+/HJVBriAGJ55PHls4YuY=
github.com/goccy/go-yaml v1.17.1/go.mod h1:XBurs7gK8ATbW4ZPGKgcbrY1Br56PdM69F7LkFRi1kA=
github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek=
github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps=
github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=

View File

@@ -36,8 +36,8 @@ func NewClientCallWrapper(keys ...string) client.CallWrapper {
omd = metadata.New(len(imd))
}
for _, k := range keys {
if v := imd.Get(k); v != nil {
omd.Set(k, v...)
if v, ok := imd.Get(k); ok {
omd.Add(k, v...)
}
}
if !ook {
@@ -59,8 +59,8 @@ func (w *wrapper) Call(ctx context.Context, req client.Request, rsp interface{},
omd = metadata.New(len(imd))
}
for _, k := range w.keys {
if v := imd.Get(k); v != nil {
omd.Set(k, v...)
if v, ok := imd.Get(k); ok {
omd.Add(k, v...)
}
}
if !ook {
@@ -80,8 +80,8 @@ func (w *wrapper) Stream(ctx context.Context, req client.Request, opts ...client
omd = metadata.New(len(imd))
}
for _, k := range w.keys {
if v := imd.Get(k); v != nil {
omd.Set(k, v...)
if v, ok := imd.Get(k); ok {
omd.Add(k, v...)
}
}
if !ook {
@@ -103,8 +103,8 @@ func NewServerHandlerWrapper(keys ...string) server.HandlerWrapper {
omd = metadata.New(len(imd))
}
for _, k := range keys {
if v := imd.Get(k); v != nil {
omd.Set(k, v...)
if v, ok := imd.Get(k); ok {
omd.Add(k, v...)
}
}
if !ook {

View File

@@ -37,9 +37,20 @@ var DefaultMetadataFunc = func(ctx context.Context) (context.Context, error) {
}
if xid == "" {
xid = imd.GetJoined(DefaultMetadataKey)
if xid == "" {
xid = omd.GetJoined(DefaultMetadataKey)
var ids []string
if ids, iok = imd.Get(DefaultMetadataKey); iok {
for i := range ids {
if ids[i] != "" {
xid = ids[i]
}
}
}
if ids, ook = omd.Get(DefaultMetadataKey); ook {
for i := range ids {
if ids[i] != "" {
xid = ids[i]
}
}
}
}

View File

@@ -2,7 +2,6 @@ package requestid
import (
"context"
"slices"
"testing"
"go.unistack.org/micro/v4/metadata"
@@ -25,10 +24,10 @@ func TestDefaultMetadataFunc(t *testing.T) {
t.Fatalf("md missing in outgoing context")
}
iv := imd.Get(DefaultMetadataKey)
ov := omd.Get(DefaultMetadataKey)
_, iok := imd.Get(DefaultMetadataKey)
_, ook := omd.Get(DefaultMetadataKey)
if !slices.Equal(iv, ov) {
t.Fatalf("missing metadata key value %v != %v", iv, ov)
if !iok || !ook {
t.Fatalf("missing metadata key value")
}
}

View File

@@ -1,51 +0,0 @@
package sql
import (
"database/sql/driver"
"errors"
"fmt"
"runtime"
)
//go:generate sh -c "go run gen.go > wrap_gen.go"
// namedValueToValue converts driver arguments of NamedValue format to Value format. Implemented in the same way as in
// database/sql ctxutil.go.
func namedValueToValue(named []driver.NamedValue) ([]driver.Value, error) {
dargs := make([]driver.Value, len(named))
for n, param := range named {
if len(param.Name) > 0 {
return nil, errors.New("sql: driver does not support the use of Named Parameters")
}
dargs[n] = param.Value
}
return dargs, nil
}
// namedValueToLabels convert driver arguments to interface{} slice
func namedValueToLabels(named []driver.NamedValue) []interface{} {
largs := make([]interface{}, 0, len(named)*2)
var name string
for _, param := range named {
if param.Name != "" {
name = param.Name
} else {
name = fmt.Sprintf("$%d", param.Ordinal)
}
largs = append(largs, fmt.Sprintf("%s=%v", name, param.Value))
}
return largs
}
// getCallerName get the name of the function A where A() -> B() -> GetFunctionCallerName()
func getCallerName() string {
pc, _, _, ok := runtime.Caller(3)
details := runtime.FuncForPC(pc)
var callerName string
if ok && details != nil {
callerName = details.Name()
} else {
callerName = labelUnknown
}
return callerName
}

View File

@@ -1,467 +0,0 @@
package sql
import (
"context"
"database/sql/driver"
"fmt"
"time"
"go.unistack.org/micro/v4/hooks/requestid"
"go.unistack.org/micro/v4/tracer"
)
var (
_ driver.Conn = (*wrapperConn)(nil)
_ driver.ConnBeginTx = (*wrapperConn)(nil)
_ driver.ConnPrepareContext = (*wrapperConn)(nil)
_ driver.Pinger = (*wrapperConn)(nil)
_ driver.Validator = (*wrapperConn)(nil)
_ driver.Queryer = (*wrapperConn)(nil) // nolint:staticcheck
_ driver.QueryerContext = (*wrapperConn)(nil)
_ driver.Execer = (*wrapperConn)(nil) // nolint:staticcheck
_ driver.ExecerContext = (*wrapperConn)(nil)
// _ driver.Connector
// _ driver.Driver
// _ driver.DriverContext
)
// wrapperConn defines a wrapper for driver.Conn
type wrapperConn struct {
d *wrapperDriver
dname string
conn driver.Conn
opts Options
ctx context.Context
//span tracer.Span
}
// Close implements driver.Conn Close
func (w *wrapperConn) Close() error {
var ctx context.Context
if w.ctx != nil {
ctx = w.ctx
} else {
ctx = context.Background()
}
_ = ctx
labels := []string{labelMethod, "Close"}
ts := time.Now()
err := w.conn.Close()
td := time.Since(ts)
te := td.Seconds()
if err != nil {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelFailure)...).Inc()
} else {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelSuccess)...).Inc()
}
w.opts.Meter.Summary(meterRequestLatencyMicroseconds, labels...).Update(te)
w.opts.Meter.Histogram(meterRequestDurationSeconds, labels...).Update(te)
/*
if w.opts.LoggerEnabled && w.opts.Logger.V(w.opts.LoggerLevel) {
w.opts.Logger.Log(ctx, w.opts.LoggerLevel, w.opts.LoggerObserver(ctx, "Close", getCallerName(), td, err)...)
}
*/
return err
}
// Begin implements driver.Conn Begin
func (w *wrapperConn) Begin() (driver.Tx, error) {
var ctx context.Context
if w.ctx != nil {
ctx = w.ctx
} else {
ctx = context.Background()
}
labels := []string{labelMethod, "Begin"}
ts := time.Now()
tx, err := w.conn.Begin() // nolint:staticcheck
td := time.Since(ts)
te := td.Seconds()
if err != nil {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelFailure)...).Inc()
w.opts.Meter.Summary(meterRequestLatencyMicroseconds, labels...).Update(te)
w.opts.Meter.Histogram(meterRequestDurationSeconds, labels...).Update(te)
/*
if w.opts.LoggerEnabled && w.opts.Logger.V(w.opts.LoggerLevel) {
w.opts.Logger.Log(ctx, w.opts.LoggerLevel, w.opts.LoggerObserver(ctx, "Begin", getCallerName(), td, err)...)
}
*/
return nil, err
}
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelSuccess)...).Inc()
w.opts.Meter.Summary(meterRequestLatencyMicroseconds, labels...).Update(te)
w.opts.Meter.Histogram(meterRequestDurationSeconds, labels...).Update(te)
/*
if w.opts.LoggerEnabled && w.opts.Logger.V(w.opts.LoggerLevel) {
w.opts.Logger.Log(ctx, w.opts.LoggerLevel, w.opts.LoggerObserver(ctx, "Begin", getCallerName(), td, err)...)
}
*/
return &wrapperTx{tx: tx, opts: w.opts, ctx: ctx}, nil
}
// BeginTx implements driver.ConnBeginTx BeginTx
func (w *wrapperConn) BeginTx(ctx context.Context, opts driver.TxOptions) (driver.Tx, error) {
name := getQueryName(ctx)
nctx, span := w.opts.Tracer.Start(ctx, "sdk.database", tracer.WithSpanKind(tracer.SpanKindClient))
span.AddLabels("db.method", "BeginTx")
span.AddLabels("db.statement", name)
if id, ok := ctx.Value(requestid.XRequestIDKey{}).(string); ok {
span.AddLabels("x-request-id", id)
}
labels := []string{labelMethod, "BeginTx", labelQuery, name}
connBeginTx, ok := w.conn.(driver.ConnBeginTx)
if !ok {
return w.Begin()
}
ts := time.Now()
tx, err := connBeginTx.BeginTx(nctx, opts)
td := time.Since(ts)
te := td.Seconds()
if err != nil {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelFailure)...).Inc()
w.opts.Meter.Summary(meterRequestLatencyMicroseconds, labels...).Update(te)
w.opts.Meter.Histogram(meterRequestDurationSeconds, labels...).Update(te)
span.SetStatus(tracer.SpanStatusError, err.Error())
/*
if w.opts.LoggerEnabled && w.opts.Logger.V(w.opts.LoggerLevel) {
w.opts.Logger.Log(ctx, w.opts.LoggerLevel, w.opts.LoggerObserver(ctx, "BeginTx", getCallerName(), td, err)...)
}
*/
return nil, err
}
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelSuccess)...).Inc()
/*
if w.opts.LoggerEnabled && w.opts.Logger.V(w.opts.LoggerLevel) {
w.opts.Logger.Log(ctx, w.opts.LoggerLevel, w.opts.LoggerObserver(ctx, "BeginTx", getCallerName(), td, err)...)
}
*/
return &wrapperTx{tx: tx, opts: w.opts, ctx: ctx, span: span}, nil
}
// Prepare implements driver.Conn Prepare
func (w *wrapperConn) Prepare(query string) (driver.Stmt, error) {
var ctx context.Context
if w.ctx != nil {
ctx = w.ctx
} else {
ctx = context.Background()
}
_ = ctx
labels := []string{labelMethod, "Prepare", labelQuery, getCallerName()}
ts := time.Now()
stmt, err := w.conn.Prepare(query)
td := time.Since(ts)
te := td.Seconds()
if err != nil {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelFailure)...).Inc()
w.opts.Meter.Summary(meterRequestLatencyMicroseconds, labels...).Update(te)
w.opts.Meter.Histogram(meterRequestDurationSeconds, labels...).Update(te)
/*
if w.opts.LoggerEnabled && w.opts.Logger.V(w.opts.LoggerLevel) {
w.opts.Logger.Log(ctx, w.opts.LoggerLevel, w.opts.LoggerObserver(ctx, "Prepare", getCallerName(), td, err)...)
}
*/
return nil, err
}
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelSuccess)...).Inc()
w.opts.Meter.Summary(meterRequestLatencyMicroseconds, labels...).Update(te)
w.opts.Meter.Histogram(meterRequestDurationSeconds, labels...).Update(te)
/*
if w.opts.LoggerEnabled && w.opts.Logger.V(w.opts.LoggerLevel) {
w.opts.Logger.Log(ctx, w.opts.LoggerLevel, w.opts.LoggerObserver(ctx, "Prepare", getCallerName(), td, err)...)
}
*/
return wrapStmt(stmt, query, w.opts), nil
}
// PrepareContext implements driver.ConnPrepareContext PrepareContext
func (w *wrapperConn) PrepareContext(ctx context.Context, query string) (driver.Stmt, error) {
var nctx context.Context
var span tracer.Span
name := getQueryName(ctx)
if w.ctx != nil {
nctx, span = w.opts.Tracer.Start(w.ctx, "sdk.database", tracer.WithSpanKind(tracer.SpanKindClient))
} else {
nctx, span = w.opts.Tracer.Start(ctx, "sdk.database", tracer.WithSpanKind(tracer.SpanKindClient))
}
span.AddLabels("db.method", "PrepareContext")
span.AddLabels("db.statement", name)
if id, ok := ctx.Value(requestid.XRequestIDKey{}).(string); ok {
span.AddLabels("x-request-id", id)
}
labels := []string{labelMethod, "PrepareContext", labelQuery, name}
conn, ok := w.conn.(driver.ConnPrepareContext)
if !ok {
return w.Prepare(query)
}
ts := time.Now()
stmt, err := conn.PrepareContext(nctx, query)
td := time.Since(ts)
te := td.Seconds()
if err != nil {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelFailure)...).Inc()
w.opts.Meter.Summary(meterRequestLatencyMicroseconds, labels...).Update(te)
w.opts.Meter.Histogram(meterRequestDurationSeconds, labels...).Update(te)
span.SetStatus(tracer.SpanStatusError, err.Error())
/*
if w.opts.LoggerEnabled && w.opts.Logger.V(w.opts.LoggerLevel) {
w.opts.Logger.Log(ctx, w.opts.LoggerLevel, w.opts.LoggerObserver(ctx, "PrepareContext", getCallerName(), td, err)...)
}
*/
return nil, err
}
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelSuccess)...).Inc()
w.opts.Meter.Summary(meterRequestLatencyMicroseconds, labels...).Update(te)
w.opts.Meter.Histogram(meterRequestDurationSeconds, labels...).Update(te)
/*
if w.opts.LoggerEnabled && w.opts.Logger.V(w.opts.LoggerLevel) {
w.opts.Logger.Log(ctx, w.opts.LoggerLevel, w.opts.LoggerObserver(ctx, "PrepareContext", getCallerName(), td, err)...)
}
*/
return wrapStmt(stmt, query, w.opts), nil
}
// Exec implements driver.Execer Exec
func (w *wrapperConn) Exec(query string, args []driver.Value) (driver.Result, error) {
var ctx context.Context
if w.ctx != nil {
ctx = w.ctx
} else {
ctx = context.Background()
}
_ = ctx
labels := []string{labelMethod, "Exec", labelQuery, getCallerName()}
// nolint:staticcheck
conn, ok := w.conn.(driver.Execer)
if !ok {
return nil, driver.ErrSkip
}
ts := time.Now()
res, err := conn.Exec(query, args)
td := time.Since(ts)
te := td.Seconds()
if err != nil {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelFailure)...).Inc()
} else {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelSuccess)...).Inc()
}
w.opts.Meter.Summary(meterRequestLatencyMicroseconds, labels...).Update(te)
w.opts.Meter.Histogram(meterRequestDurationSeconds, labels...).Update(te)
/*
if w.opts.LoggerEnabled && w.opts.Logger.V(w.opts.LoggerLevel) {
w.opts.Logger.Log(ctx, w.opts.LoggerLevel, w.opts.LoggerObserver(ctx, "Exec", getCallerName(), td, err)...)
}
*/
return res, err
}
// Exec implements driver.StmtExecContext ExecContext
func (w *wrapperConn) ExecContext(ctx context.Context, query string, args []driver.NamedValue) (driver.Result, error) {
var nctx context.Context
var span tracer.Span
name := getQueryName(ctx)
if w.ctx != nil {
nctx, span = w.opts.Tracer.Start(w.ctx, "sdk.database", tracer.WithSpanKind(tracer.SpanKindClient))
} else {
nctx, span = w.opts.Tracer.Start(ctx, "sdk.database", tracer.WithSpanKind(tracer.SpanKindClient))
}
span.AddLabels("db.method", "ExecContext")
span.AddLabels("db.statement", name)
if id, ok := ctx.Value(requestid.XRequestIDKey{}).(string); ok {
span.AddLabels("x-request-id", id)
}
defer span.Finish()
if len(args) > 0 {
span.AddLabels("db.args", fmt.Sprintf("%v", namedValueToLabels(args)))
}
labels := []string{labelMethod, "ExecContext", labelQuery, name}
conn, ok := w.conn.(driver.ExecerContext)
if !ok {
// nolint:staticcheck
return nil, driver.ErrSkip
}
ts := time.Now()
res, err := conn.ExecContext(nctx, query, args)
td := time.Since(ts)
te := td.Seconds()
if err != nil {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelFailure)...).Inc()
span.SetStatus(tracer.SpanStatusError, err.Error())
} else {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelSuccess)...).Inc()
}
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelSuccess)...).Inc()
w.opts.Meter.Summary(meterRequestLatencyMicroseconds, labels...).Update(te)
w.opts.Meter.Histogram(meterRequestDurationSeconds, labels...).Update(te)
/*
if w.opts.LoggerEnabled && w.opts.Logger.V(w.opts.LoggerLevel) {
w.opts.Logger.Log(ctx, w.opts.LoggerLevel, w.opts.LoggerObserver(ctx, "ExecContext", getCallerName(), td, err)...)
}
*/
return res, err
}
// Ping implements driver.Pinger Ping
func (w *wrapperConn) Ping(ctx context.Context) error {
conn, ok := w.conn.(driver.Pinger)
if !ok {
// fallback path to check db alive
pc, err := w.d.Open(w.dname)
if err != nil {
return err
}
return pc.Close()
}
var nctx context.Context //nolint:gosimple
nctx = ctx
/*
var span tracer.Span
if w.ctx != nil {
nctx, span = w.opts.Tracer.Start(w.ctx, "sdk.database", tracer.WithSpanKind(tracer.SpanKindClient))
} else {
nctx, span = w.opts.Tracer.Start(ctx, "sdk.database", tracer.WithSpanKind(tracer.SpanKindClient))
}
span.AddLabels("db.method", "Ping")
defer span.Finish()
*/
labels := []string{labelMethod, "Ping"}
ts := time.Now()
err := conn.Ping(nctx)
td := time.Since(ts)
te := td.Seconds()
if err != nil {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelFailure)...).Inc()
// span.SetStatus(tracer.SpanStatusError, err.Error())
/*
if w.opts.LoggerEnabled && w.opts.Logger.V(w.opts.LoggerLevel) {
w.opts.Logger.Log(ctx, w.opts.LoggerLevel, w.opts.LoggerObserver(ctx, "Ping", getCallerName(), td, err)...)
}
*/
return err
} else {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelSuccess)...).Inc()
}
w.opts.Meter.Summary(meterRequestLatencyMicroseconds, labels...).Update(te)
w.opts.Meter.Histogram(meterRequestDurationSeconds, labels...).Update(te)
return nil
}
// Query implements driver.Queryer Query
func (w *wrapperConn) Query(query string, args []driver.Value) (driver.Rows, error) {
var ctx context.Context
if w.ctx != nil {
ctx = w.ctx
} else {
ctx = context.Background()
}
_ = ctx
// nolint:staticcheck
conn, ok := w.conn.(driver.Queryer)
if !ok {
return nil, driver.ErrSkip
}
labels := []string{labelMethod, "Query", labelQuery, getCallerName()}
ts := time.Now()
rows, err := conn.Query(query, args)
td := time.Since(ts)
te := td.Seconds()
if err != nil {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelFailure)...).Inc()
} else {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelSuccess)...).Inc()
}
w.opts.Meter.Summary(meterRequestLatencyMicroseconds, labels...).Update(te)
w.opts.Meter.Histogram(meterRequestDurationSeconds, labels...).Update(te)
/*
if w.opts.LoggerEnabled && w.opts.Logger.V(w.opts.LoggerLevel) {
w.opts.Logger.Log(ctx, w.opts.LoggerLevel, w.opts.LoggerObserver(ctx, "Query", getCallerName(), td, err)...)
}
*/
return rows, err
}
// QueryContext implements Driver.QueryerContext QueryContext
func (w *wrapperConn) QueryContext(ctx context.Context, query string, args []driver.NamedValue) (driver.Rows, error) {
var nctx context.Context
var span tracer.Span
name := getQueryName(ctx)
if w.ctx != nil {
nctx, span = w.opts.Tracer.Start(w.ctx, "sdk.database", tracer.WithSpanKind(tracer.SpanKindClient))
} else {
nctx, span = w.opts.Tracer.Start(ctx, "sdk.database", tracer.WithSpanKind(tracer.SpanKindClient))
}
span.AddLabels("db.method", "QueryContext")
span.AddLabels("db.statement", name)
if id, ok := ctx.Value(requestid.XRequestIDKey{}).(string); ok {
span.AddLabels("x-request-id", id)
}
defer span.Finish()
if len(args) > 0 {
span.AddLabels("db.args", fmt.Sprintf("%v", namedValueToLabels(args)))
}
labels := []string{labelMethod, "QueryContext", labelQuery, name}
conn, ok := w.conn.(driver.QueryerContext)
if !ok {
return nil, driver.ErrSkip
}
ts := time.Now()
rows, err := conn.QueryContext(nctx, query, args)
td := time.Since(ts)
te := td.Seconds()
if err != nil {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelFailure)...).Inc()
span.SetStatus(tracer.SpanStatusError, err.Error())
} else {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelSuccess)...).Inc()
}
w.opts.Meter.Summary(meterRequestLatencyMicroseconds, labels...).Update(te)
w.opts.Meter.Histogram(meterRequestDurationSeconds, labels...).Update(te)
/*
if w.opts.LoggerEnabled && w.opts.Logger.V(w.opts.LoggerLevel) {
w.opts.Logger.Log(ctx, w.opts.LoggerLevel, w.opts.LoggerObserver(ctx, "QueryContext", getCallerName(), td, err)...)
}
*/
return rows, err
}
// CheckNamedValue implements driver.NamedValueChecker
func (w *wrapperConn) CheckNamedValue(v *driver.NamedValue) error {
s, ok := w.conn.(driver.NamedValueChecker)
if !ok {
return driver.ErrSkip
}
return s.CheckNamedValue(v)
}
// IsValid implements driver.Validator
func (w *wrapperConn) IsValid() bool {
v, ok := w.conn.(driver.Validator)
if !ok {
return w.conn != nil
}
return v.IsValid()
}
func (w *wrapperConn) ResetSession(ctx context.Context) error {
s, ok := w.conn.(driver.SessionResetter)
if !ok {
return driver.ErrSkip
}
return s.ResetSession(ctx)
}

View File

@@ -1,94 +0,0 @@
package sql
import (
"context"
"database/sql/driver"
"time"
)
var (
// _ driver.DriverContext = (*wrapperDriver)(nil)
// _ driver.Connector = (*wrapperDriver)(nil)
)
/*
type conn interface {
driver.Pinger
driver.Execer
driver.ExecerContext
driver.Queryer
driver.QueryerContext
driver.Conn
driver.ConnPrepareContext
driver.ConnBeginTx
}
*/
// wrapperDriver defines a wrapper for driver.Driver
type wrapperDriver struct {
driver driver.Driver
opts Options
ctx context.Context
}
// NewWrapper creates and returns a new SQL driver with passed capabilities
func NewWrapper(d driver.Driver, opts ...Option) driver.Driver {
return &wrapperDriver{driver: d, opts: NewOptions(opts...), ctx: context.Background()}
}
type wrappedConnector struct {
connector driver.Connector
// name string
opts Options
ctx context.Context
}
func NewWrapperConnector(c driver.Connector, opts ...Option) driver.Connector {
return &wrappedConnector{connector: c, opts: NewOptions(opts...), ctx: context.Background()}
}
// Connect implements driver.Driver Connect
func (w *wrappedConnector) Connect(ctx context.Context) (driver.Conn, error) {
return w.connector.Connect(ctx)
}
// Driver implements driver.Driver Driver
func (w *wrappedConnector) Driver() driver.Driver {
return w.connector.Driver()
}
/*
// Connect implements driver.Driver OpenConnector
func (w *wrapperDriver) OpenConnector(name string) (driver.Conn, error) {
return &wrapperConnector{driver: w.driver, name: name, opts: w.opts}, nil
}
*/
// Open implements driver.Driver Open
func (w *wrapperDriver) Open(name string) (driver.Conn, error) {
// ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) // Ensure eventual timeout
// defer cancel()
/*
connector, err := w.OpenConnector(name)
if err != nil {
return nil, err
}
return connector.Connect(ctx)
*/
ts := time.Now()
c, err := w.driver.Open(name)
td := time.Since(ts)
/*
if w.opts.LoggerEnabled {
w.opts.Logger.Log(w.ctx, w.opts.LoggerLevel, w.opts.LoggerObserver(w.ctx, "Open", getCallerName(), td, err)...)
}
*/
_ = td
if err != nil {
return nil, err
}
return wrapConn(c, w.opts), nil
}

View File

@@ -1,167 +0,0 @@
//go:build ignore
package main
import (
"bytes"
"crypto/md5"
"fmt"
"io"
"sort"
"strings"
)
var connIfaces = []string{
"driver.ConnBeginTx",
"driver.ConnPrepareContext",
"driver.Execer",
"driver.ExecerContext",
"driver.NamedValueChecker",
"driver.Pinger",
"driver.Queryer",
"driver.QueryerContext",
"driver.SessionResetter",
"driver.Validator",
}
var stmtIfaces = []string{
"driver.StmtExecContext",
"driver.StmtQueryContext",
"driver.ColumnConverter",
"driver.NamedValueChecker",
}
func getHash(s []string) string {
h := md5.New()
io.WriteString(h, strings.Join(s, "|"))
return fmt.Sprintf("%x", h.Sum(nil))
}
func main() {
comboConn := all(connIfaces)
sort.Slice(comboConn, func(i, j int) bool {
return len(comboConn[i]) < len(comboConn[j])
})
comboStmt := all(stmtIfaces)
sort.Slice(comboStmt, func(i, j int) bool {
return len(comboStmt[i]) < len(comboStmt[j])
})
b := bytes.NewBuffer(nil)
b.WriteString("// Code generated. DO NOT EDIT.\n\n")
b.WriteString("package sql\n\n")
b.WriteString(`import "database/sql/driver"`)
b.WriteString("\n\n")
b.WriteString("func wrapConn(dc driver.Conn, opts Options) driver.Conn {\n")
b.WriteString("\tc := &wrapperConn{conn: dc, opts: opts}\n")
for idx := len(comboConn) - 1; idx >= 0; idx-- {
ifaces := comboConn[idx]
n := len(ifaces)
if n == 0 {
continue
}
h := getHash(ifaces)
b.WriteString(fmt.Sprintf("\tif _, ok := dc.(wrapConn%04d_%s); ok {\n", n, h))
b.WriteString("\treturn struct {\n")
b.WriteString("\t\tdriver.Conn\n")
b.WriteString(fmt.Sprintf("\t\t\t%s", strings.Join(ifaces, "\n\t\t\t")))
b.WriteString("\t\t\n}{")
for idx := range ifaces {
if idx > 0 {
b.WriteString(", ")
b.WriteString("c")
} else if idx == 0 {
b.WriteString("c")
} else {
b.WriteString("c")
}
}
b.WriteString(", c}\n")
b.WriteString("}\n\n")
}
b.WriteString("return c\n")
b.WriteString("}\n")
for idx := len(comboConn) - 1; idx >= 0; idx-- {
ifaces := comboConn[idx]
n := len(ifaces)
if n == 0 {
continue
}
h := getHash(ifaces)
b.WriteString(fmt.Sprintf("// %s\n", strings.Join(ifaces, "|")))
b.WriteString(fmt.Sprintf("type wrapConn%04d_%s interface {\n", n, h))
for _, iface := range ifaces {
b.WriteString(fmt.Sprintf("\t%s\n", iface))
}
b.WriteString("}\n\n")
}
b.WriteString("func wrapStmt(stmt driver.Stmt, query string, opts Options) driver.Stmt {\n")
b.WriteString("\tc := &wrapperStmt{stmt: stmt, query: query, opts: opts}\n")
for idx := len(comboStmt) - 1; idx >= 0; idx-- {
ifaces := comboStmt[idx]
n := len(ifaces)
if n == 0 {
continue
}
h := getHash(ifaces)
b.WriteString(fmt.Sprintf("\tif _, ok := stmt.(wrapStmt%04d_%s); ok {\n", n, h))
b.WriteString("\treturn struct {\n")
b.WriteString("\t\tdriver.Stmt\n")
b.WriteString(fmt.Sprintf("\t\t\t%s", strings.Join(ifaces, "\n\t\t\t")))
b.WriteString("\t\t\n}{")
for idx := range ifaces {
if idx > 0 {
b.WriteString(", ")
b.WriteString("c")
} else if idx == 0 {
b.WriteString("c")
} else {
b.WriteString("c")
}
}
b.WriteString(", c}\n")
b.WriteString("}\n\n")
}
b.WriteString("return c\n")
b.WriteString("}\n")
for idx := len(comboStmt) - 1; idx >= 0; idx-- {
ifaces := comboStmt[idx]
n := len(ifaces)
if n == 0 {
continue
}
h := getHash(ifaces)
b.WriteString(fmt.Sprintf("// %s\n", strings.Join(ifaces, "|")))
b.WriteString(fmt.Sprintf("type wrapStmt%04d_%s interface {\n", n, h))
for _, iface := range ifaces {
b.WriteString(fmt.Sprintf("\t%s\n", iface))
}
b.WriteString("}\n\n")
}
fmt.Printf("%s\n", b.String())
}
// all returns all combinations for a given string array.
func all[T any](set []T) (subsets [][]T) {
length := uint(len(set))
for subsetBits := 1; subsetBits < (1 << length); subsetBits++ {
var subset []T
for object := uint(0); object < length; object++ {
if (subsetBits>>object)&1 == 1 {
subset = append(subset, set[object])
}
}
subsets = append(subsets, subset)
}
return subsets
}

View File

@@ -1,172 +0,0 @@
package sql
import (
"context"
"fmt"
"time"
"go.unistack.org/micro/v4/logger"
"go.unistack.org/micro/v4/meter"
"go.unistack.org/micro/v4/tracer"
)
var (
// DefaultMeterStatsInterval holds default stats interval
DefaultMeterStatsInterval = 5 * time.Second
// DefaultLoggerObserver used to prepare labels for logger
DefaultLoggerObserver = func(ctx context.Context, method string, query string, td time.Duration, err error) []interface{} {
labels := []interface{}{"db.method", method, "took", fmt.Sprintf("%v", td)}
if err != nil {
labels = append(labels, "error", err.Error())
}
if query != labelUnknown {
labels = append(labels, "query", query)
}
return labels
}
)
var (
MaxOpenConnections = "micro_sql_max_open_conn"
OpenConnections = "micro_sql_open_conn"
InuseConnections = "micro_sql_inuse_conn"
IdleConnections = "micro_sql_idle_conn"
WaitConnections = "micro_sql_waited_conn"
BlockedSeconds = "micro_sql_blocked_seconds"
MaxIdleClosed = "micro_sql_max_idle_closed"
MaxIdletimeClosed = "micro_sql_closed_max_idle"
MaxLifetimeClosed = "micro_sql_closed_max_lifetime"
meterRequestTotal = "micro_sql_request_total"
meterRequestLatencyMicroseconds = "micro_sql_latency_microseconds"
meterRequestDurationSeconds = "micro_sql_request_duration_seconds"
labelUnknown = "unknown"
labelQuery = "db_statement"
labelMethod = "db_method"
labelStatus = "status"
labelSuccess = "success"
labelFailure = "failure"
labelHost = "db_host"
labelDatabase = "db_name"
)
// Options struct holds wrapper options
type Options struct {
Logger logger.Logger
Meter meter.Meter
Tracer tracer.Tracer
DatabaseHost string
DatabaseName string
MeterStatsInterval time.Duration
LoggerLevel logger.Level
LoggerEnabled bool
LoggerObserver func(ctx context.Context, method string, name string, td time.Duration, err error) []interface{}
}
// Option func signature
type Option func(*Options)
// NewOptions create new Options struct from provided option slice
func NewOptions(opts ...Option) Options {
options := Options{
Logger: logger.DefaultLogger,
Meter: meter.DefaultMeter,
Tracer: tracer.DefaultTracer,
MeterStatsInterval: DefaultMeterStatsInterval,
LoggerLevel: logger.ErrorLevel,
LoggerObserver: DefaultLoggerObserver,
}
for _, o := range opts {
o(&options)
}
options.Meter = options.Meter.Clone(
meter.Labels(
labelHost, options.DatabaseHost,
labelDatabase, options.DatabaseName,
),
)
options.Logger = options.Logger.Clone(logger.WithAddCallerSkipCount(1))
return options
}
// MetricInterval specifies stats interval for *sql.DB
func MetricInterval(td time.Duration) Option {
return func(o *Options) {
o.MeterStatsInterval = td
}
}
func DatabaseHost(host string) Option {
return func(o *Options) {
o.DatabaseHost = host
}
}
func DatabaseName(name string) Option {
return func(o *Options) {
o.DatabaseName = name
}
}
// Meter passes meter.Meter to wrapper
func Meter(m meter.Meter) Option {
return func(o *Options) {
o.Meter = m
}
}
// Logger passes logger.Logger to wrapper
func Logger(l logger.Logger) Option {
return func(o *Options) {
o.Logger = l
}
}
// LoggerEnabled enable sql logging
func LoggerEnabled(b bool) Option {
return func(o *Options) {
o.LoggerEnabled = b
}
}
// LoggerLevel passes logger.Level option
func LoggerLevel(lvl logger.Level) Option {
return func(o *Options) {
o.LoggerLevel = lvl
}
}
// LoggerObserver passes observer to fill logger fields
func LoggerObserver(obs func(context.Context, string, string, time.Duration, error) []interface{}) Option {
return func(o *Options) {
o.LoggerObserver = obs
}
}
// Tracer passes tracer.Tracer to wrapper
func Tracer(t tracer.Tracer) Option {
return func(o *Options) {
o.Tracer = t
}
}
type queryNameKey struct{}
// QueryName passes query name to wrapper func
func QueryName(ctx context.Context, name string) context.Context {
if ctx == nil {
ctx = context.Background()
}
return context.WithValue(ctx, queryNameKey{}, name)
}
func getQueryName(ctx context.Context) string {
if v, ok := ctx.Value(queryNameKey{}).(string); ok && v != labelUnknown {
return v
}
return getCallerName()
}

View File

@@ -1,41 +0,0 @@
package sql
import (
"context"
"database/sql"
"time"
)
type Statser interface {
Stats() sql.DBStats
}
func NewStatsMeter(ctx context.Context, db Statser, opts ...Option) {
options := NewOptions(opts...)
go func() {
ticker := time.NewTicker(options.MeterStatsInterval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
if db == nil {
return
}
stats := db.Stats()
options.Meter.Counter(MaxOpenConnections).Set(uint64(stats.MaxOpenConnections))
options.Meter.Counter(OpenConnections).Set(uint64(stats.OpenConnections))
options.Meter.Counter(InuseConnections).Set(uint64(stats.InUse))
options.Meter.Counter(IdleConnections).Set(uint64(stats.Idle))
options.Meter.Counter(WaitConnections).Set(uint64(stats.WaitCount))
options.Meter.FloatCounter(BlockedSeconds).Set(stats.WaitDuration.Seconds())
options.Meter.Counter(MaxIdleClosed).Set(uint64(stats.MaxIdleClosed))
options.Meter.Counter(MaxIdletimeClosed).Set(uint64(stats.MaxIdleTimeClosed))
options.Meter.Counter(MaxLifetimeClosed).Set(uint64(stats.MaxLifetimeClosed))
}
}
}()
}

View File

@@ -1,287 +0,0 @@
package sql
import (
"context"
"database/sql/driver"
"fmt"
"time"
requestid "go.unistack.org/micro/v4/hooks/requestid"
"go.unistack.org/micro/v4/tracer"
)
var (
_ driver.Stmt = (*wrapperStmt)(nil)
_ driver.StmtQueryContext = (*wrapperStmt)(nil)
_ driver.StmtExecContext = (*wrapperStmt)(nil)
_ driver.NamedValueChecker = (*wrapperStmt)(nil)
)
// wrapperStmt defines a wrapper for driver.Stmt
type wrapperStmt struct {
stmt driver.Stmt
opts Options
query string
ctx context.Context
}
// Close implements driver.Stmt Close
func (w *wrapperStmt) Close() error {
var ctx context.Context
if w.ctx != nil {
ctx = w.ctx
} else {
ctx = context.Background()
}
_ = ctx
labels := []string{labelMethod, "Close"}
ts := time.Now()
err := w.stmt.Close()
td := time.Since(ts)
te := td.Seconds()
if err != nil {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelFailure)...).Inc()
} else {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelSuccess)...).Inc()
}
w.opts.Meter.Summary(meterRequestLatencyMicroseconds, labels...).Update(te)
w.opts.Meter.Histogram(meterRequestDurationSeconds, labels...).Update(te)
/*
if w.opts.LoggerEnabled && w.opts.Logger.V(w.opts.LoggerLevel) {
w.opts.Logger.Log(ctx, w.opts.LoggerLevel, w.opts.LoggerObserver(ctx, "Close", getCallerName(), td, err)...)
}
*/
return err
}
// NumInput implements driver.Stmt NumInput
func (w *wrapperStmt) NumInput() int {
return w.stmt.NumInput()
}
// CheckNamedValue implements driver.NamedValueChecker
func (w *wrapperStmt) CheckNamedValue(v *driver.NamedValue) error {
s, ok := w.stmt.(driver.NamedValueChecker)
if !ok {
return driver.ErrSkip
}
return s.CheckNamedValue(v)
}
// Exec implements driver.Stmt Exec
func (w *wrapperStmt) Exec(args []driver.Value) (driver.Result, error) {
var ctx context.Context
if w.ctx != nil {
ctx = w.ctx
} else {
ctx = context.Background()
}
_ = ctx
labels := []string{labelMethod, "Exec"}
ts := time.Now()
res, err := w.stmt.Exec(args) // nolint:staticcheck
td := time.Since(ts)
te := td.Seconds()
if err != nil {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelFailure)...).Inc()
} else {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelSuccess)...).Inc()
}
w.opts.Meter.Summary(meterRequestLatencyMicroseconds, labels...).Update(te)
w.opts.Meter.Histogram(meterRequestDurationSeconds, labels...).Update(te)
/*
if w.opts.LoggerEnabled && w.opts.Logger.V(w.opts.LoggerLevel) {
w.opts.Logger.Log(ctx, w.opts.LoggerLevel, w.opts.LoggerObserver(ctx, "Exec", getCallerName(), td, err)...)
}
*/
return res, err
}
// Query implements driver.Stmt Query
func (w *wrapperStmt) Query(args []driver.Value) (driver.Rows, error) {
var ctx context.Context
if w.ctx != nil {
ctx = w.ctx
} else {
ctx = context.Background()
}
_ = ctx
labels := []string{labelMethod, "Query"}
ts := time.Now()
rows, err := w.stmt.Query(args) // nolint:staticcheck
td := time.Since(ts)
te := td.Seconds()
if err != nil {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelFailure)...).Inc()
} else {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelSuccess)...).Inc()
}
w.opts.Meter.Summary(meterRequestLatencyMicroseconds, labels...).Update(te)
w.opts.Meter.Histogram(meterRequestDurationSeconds, labels...).Update(te)
/*
if w.opts.LoggerEnabled && w.opts.Logger.V(w.opts.LoggerLevel) {
w.opts.Logger.Log(ctx, w.opts.LoggerLevel, w.opts.LoggerObserver(ctx, "Query", getCallerName(), td, err)...)
}
*/
return rows, err
}
// ColumnConverter implements driver.ColumnConverter
func (w *wrapperStmt) ColumnConverter(idx int) driver.ValueConverter {
s, ok := w.stmt.(driver.ColumnConverter) // nolint:staticcheck
if !ok {
return nil
}
return s.ColumnConverter(idx)
}
// ExecContext implements driver.StmtExecContext ExecContext
func (w *wrapperStmt) ExecContext(ctx context.Context, args []driver.NamedValue) (driver.Result, error) {
var nctx context.Context
var span tracer.Span
name := getQueryName(ctx)
if w.ctx != nil {
nctx, span = w.opts.Tracer.Start(w.ctx, "sdk.database", tracer.WithSpanKind(tracer.SpanKindClient))
} else {
nctx, span = w.opts.Tracer.Start(ctx, "sdk.database", tracer.WithSpanKind(tracer.SpanKindClient))
}
span.AddLabels("db.method", "ExecContext")
span.AddLabels("db.statement", name)
defer span.Finish()
if len(args) > 0 {
span.AddLabels("db.args", fmt.Sprintf("%v", namedValueToLabels(args)))
}
if id, ok := ctx.Value(requestid.XRequestIDKey{}).(string); ok {
span.AddLabels("x-request-id", id)
}
labels := []string{labelMethod, "ExecContext", labelQuery, name}
if conn, ok := w.stmt.(driver.StmtExecContext); ok {
ts := time.Now()
res, err := conn.ExecContext(nctx, args)
td := time.Since(ts)
te := td.Seconds()
if err != nil {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelFailure)...).Inc()
span.SetStatus(tracer.SpanStatusError, err.Error())
} else {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelSuccess)...).Inc()
}
w.opts.Meter.Summary(meterRequestLatencyMicroseconds, labels...).Update(te)
w.opts.Meter.Histogram(meterRequestDurationSeconds, labels...).Update(te)
/*
if w.opts.LoggerEnabled && w.opts.Logger.V(w.opts.LoggerLevel) {
w.opts.Logger.Log(ctx, w.opts.LoggerLevel, w.opts.LoggerObserver(ctx, "ExecContext", name, td, err)...)
}
*/
return res, err
}
values, err := namedValueToValue(args)
if err != nil {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelFailure)...).Inc()
span.SetStatus(tracer.SpanStatusError, err.Error())
/*
if w.opts.LoggerEnabled && w.opts.Logger.V(w.opts.LoggerLevel) {
w.opts.Logger.Log(ctx, w.opts.LoggerLevel, w.opts.LoggerObserver(ctx, "ExecContext", name, 0, err)...)
}
*/
return nil, err
}
ts := time.Now()
res, err := w.Exec(values) // nolint:staticcheck
td := time.Since(ts)
te := td.Seconds()
if err != nil {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelFailure)...).Inc()
span.SetStatus(tracer.SpanStatusError, err.Error())
} else {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelSuccess)...).Inc()
}
w.opts.Meter.Summary(meterRequestLatencyMicroseconds, labels...).Update(te)
w.opts.Meter.Histogram(meterRequestDurationSeconds, labels...).Update(te)
/*
if w.opts.LoggerEnabled && w.opts.Logger.V(w.opts.LoggerLevel) {
w.opts.Logger.Log(ctx, w.opts.LoggerLevel, w.opts.LoggerObserver(ctx, "ExecContext", name, td, err)...)
}
*/
return res, err
}
// QueryContext implements driver.StmtQueryContext StmtQueryContext
func (w *wrapperStmt) QueryContext(ctx context.Context, args []driver.NamedValue) (driver.Rows, error) {
var nctx context.Context
var span tracer.Span
name := getQueryName(ctx)
if w.ctx != nil {
nctx, span = w.opts.Tracer.Start(w.ctx, "sdk.database", tracer.WithSpanKind(tracer.SpanKindClient))
} else {
nctx, span = w.opts.Tracer.Start(ctx, "sdk.database", tracer.WithSpanKind(tracer.SpanKindClient))
}
span.AddLabels("db.method", "QueryContext")
span.AddLabels("db.statement", name)
defer span.Finish()
if len(args) > 0 {
span.AddLabels("db.args", fmt.Sprintf("%v", namedValueToLabels(args)))
}
if id, ok := ctx.Value(requestid.XRequestIDKey{}).(string); ok {
span.AddLabels("x-request-id", id)
}
labels := []string{labelMethod, "QueryContext", labelQuery, name}
if conn, ok := w.stmt.(driver.StmtQueryContext); ok {
ts := time.Now()
rows, err := conn.QueryContext(nctx, args)
td := time.Since(ts)
te := td.Seconds()
if err != nil {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelFailure)...).Inc()
span.SetStatus(tracer.SpanStatusError, err.Error())
} else {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelSuccess)...).Inc()
}
w.opts.Meter.Summary(meterRequestLatencyMicroseconds, labels...).Update(te)
w.opts.Meter.Histogram(meterRequestDurationSeconds, labels...).Update(te)
/*
if w.opts.LoggerEnabled && w.opts.Logger.V(w.opts.LoggerLevel) {
w.opts.Logger.Log(ctx, w.opts.LoggerLevel, w.opts.LoggerObserver(ctx, "QueryContext", name, td, err)...)
}
*/
return rows, err
}
values, err := namedValueToValue(args)
if err != nil {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelFailure)...).Inc()
span.SetStatus(tracer.SpanStatusError, err.Error())
/*
if w.opts.LoggerEnabled && w.opts.Logger.V(w.opts.LoggerLevel) {
w.opts.Logger.Log(ctx, w.opts.LoggerLevel, w.opts.LoggerObserver(ctx, "QueryContext", name, 0, err)...)
}
*/
return nil, err
}
ts := time.Now()
rows, err := w.Query(values) // nolint:staticcheck
td := time.Since(ts)
te := td.Seconds()
if err != nil {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelFailure)...).Inc()
span.SetStatus(tracer.SpanStatusError, err.Error())
} else {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelSuccess)...).Inc()
}
w.opts.Meter.Summary(meterRequestLatencyMicroseconds, labels...).Update(te)
w.opts.Meter.Histogram(meterRequestDurationSeconds, labels...).Update(te)
/*
if w.opts.LoggerEnabled && w.opts.Logger.V(w.opts.LoggerLevel) {
w.opts.Logger.Log(ctx, w.opts.LoggerLevel, w.opts.LoggerObserver(ctx, "QueryContext", name, td, err)...)
}
*/
return rows, err
}

View File

@@ -1,63 +0,0 @@
package sql
import (
"context"
"database/sql/driver"
"time"
"go.unistack.org/micro/v4/tracer"
)
var _ driver.Tx = (*wrapperTx)(nil)
// wrapperTx defines a wrapper for driver.Tx
type wrapperTx struct {
tx driver.Tx
span tracer.Span
opts Options
ctx context.Context
}
// Commit implements driver.Tx Commit
func (w *wrapperTx) Commit() error {
ts := time.Now()
err := w.tx.Commit()
td := time.Since(ts)
_ = td
if w.span != nil {
if err != nil {
w.span.SetStatus(tracer.SpanStatusError, err.Error())
}
w.span.Finish()
}
/*
if w.opts.LoggerEnabled && w.opts.Logger.V(w.opts.LoggerLevel) {
w.opts.Logger.Log(w.ctx, w.opts.LoggerLevel, w.opts.LoggerObserver(w.ctx, "Commit", getCallerName(), td, err)...)
}
*/
w.ctx = nil
return err
}
// Rollback implements driver.Tx Rollback
func (w *wrapperTx) Rollback() error {
ts := time.Now()
err := w.tx.Rollback()
td := time.Since(ts)
_ = td
if w.span != nil {
if err != nil {
w.span.SetStatus(tracer.SpanStatusError, err.Error())
}
w.span.Finish()
}
/*
if w.opts.LoggerEnabled && w.opts.Logger.V(w.opts.LoggerLevel) {
w.opts.Logger.Log(w.ctx, w.opts.LoggerLevel, w.opts.LoggerObserver(w.ctx, "Rollback", getCallerName(), td, err)...)
}
*/
w.ctx = nil
return err
}

View File

@@ -1,19 +0,0 @@
package sql
import (
"database/sql/driver"
)
/*
func wrapDriver(d driver.Driver, opts Options) driver.Driver {
if _, ok := d.(driver.DriverContext); ok {
return &wrapperDriver{driver: d, opts: opts}
}
return struct{ driver.Driver }{&wrapperDriver{driver: d, opts: opts}}
}
*/
// WrapConn allows an existing driver.Conn to be wrapped.
func WrapConn(c driver.Conn, opts ...Option) driver.Conn {
return wrapConn(c, NewOptions(opts...))
}

File diff suppressed because it is too large Load Diff

View File

@@ -4,20 +4,18 @@ package logger
type Level int8
const (
// TraceLevel usually used to find bugs, very verbose
// TraceLevel level usually used to find bugs, very verbose
TraceLevel Level = iota - 2
// DebugLevel used only when enabled debugging
// DebugLevel level used only when enabled debugging
DebugLevel
// InfoLevel used for general info about what's going on inside the application
// InfoLevel level used for general info about what's going on inside the application
InfoLevel
// WarnLevel used for non-critical entries
// WarnLevel level used for non-critical entries
WarnLevel
// ErrorLevel used for errors that should definitely be noted
// ErrorLevel level used for errors that should definitely be noted
ErrorLevel
// FatalLevel used for critical errors and then calls `os.Exit(1)`
// FatalLevel level used for critical errors and then calls `os.Exit(1)`
FatalLevel
// NoneLevel used to disable logging
NoneLevel
)
// String returns logger level string representation
@@ -35,8 +33,6 @@ func (l Level) String() string {
return "error"
case FatalLevel:
return "fatal"
case NoneLevel:
return "none"
}
return "info"
}
@@ -62,8 +58,6 @@ func ParseLevel(lvl string) Level {
return ErrorLevel
case FatalLevel.String():
return FatalLevel
case NoneLevel.String():
return NoneLevel
}
return InfoLevel
}

View File

@@ -34,7 +34,6 @@ var (
warnValue = slog.StringValue("warn")
errorValue = slog.StringValue("error")
fatalValue = slog.StringValue("fatal")
noneValue = slog.StringValue("none")
)
type wrapper struct {
@@ -86,8 +85,6 @@ func (s *slogLogger) renameAttr(_ []string, a slog.Attr) slog.Attr {
a.Value = errorValue
case lvl >= logger.FatalLevel:
a.Value = fatalValue
case lvl >= logger.NoneLevel:
a.Value = noneValue
default:
a.Value = infoValue
}
@@ -319,8 +316,6 @@ func loggerToSlogLevel(level logger.Level) slog.Level {
return slog.LevelDebug - 1
case logger.FatalLevel:
return slog.LevelError + 1
case logger.NoneLevel:
return slog.LevelError + 2
default:
return slog.LevelInfo
}
@@ -338,8 +333,6 @@ func slogToLoggerLevel(level slog.Level) logger.Level {
return logger.TraceLevel
case slog.LevelError + 1:
return logger.FatalLevel
case slog.LevelError + 2:
return logger.NoneLevel
default:
return logger.InfoLevel
}

View File

@@ -36,24 +36,6 @@ func TestStacktrace(t *testing.T) {
}
}
func TestNoneLevel(t *testing.T) {
ctx := context.TODO()
buf := bytes.NewBuffer(nil)
l := NewLogger(logger.WithLevel(logger.NoneLevel), logger.WithOutput(buf),
WithHandlerFunc(slog.NewTextHandler),
logger.WithAddStacktrace(true),
)
if err := l.Init(logger.WithFields("key1", "val1")); err != nil {
t.Fatal(err)
}
l.Error(ctx, "msg1", errors.New("err"))
if buf.Len() != 0 {
t.Fatalf("logger none level not works, buf contains: %s", buf.Bytes())
}
}
func TestDelayedBuffer(t *testing.T) {
ctx := context.TODO()
buf := bytes.NewBuffer(nil)
@@ -424,7 +406,7 @@ func TestLogger(t *testing.T) {
func Test_WithContextAttrFunc(t *testing.T) {
loggerContextAttrFuncs := []logger.ContextAttrFunc{
func(ctx context.Context) []interface{} {
md, ok := metadata.FromOutgoingContext(ctx)
md, ok := metadata.FromIncomingContext(ctx)
if !ok {
return nil
}
@@ -443,7 +425,7 @@ func Test_WithContextAttrFunc(t *testing.T) {
logger.DefaultContextAttrFuncs = append(logger.DefaultContextAttrFuncs, loggerContextAttrFuncs...)
ctx := context.TODO()
ctx = metadata.AppendOutgoingContext(ctx, "X-Request-Id", uuid.New().String(),
ctx = metadata.AppendIncomingContext(ctx, "X-Request-Id", uuid.New().String(),
"Source-Service", "Test-System")
buf := bytes.NewBuffer(nil)
@@ -463,9 +445,9 @@ func Test_WithContextAttrFunc(t *testing.T) {
t.Fatalf("logger info, buf %s", buf.Bytes())
}
buf.Reset()
omd, _ := metadata.FromOutgoingContext(ctx)
imd, _ := metadata.FromIncomingContext(ctx)
l.Info(ctx, "test message1")
omd.Set("Source-Service", "Test-System2")
imd.Set("Source-Service", "Test-System2")
l.Info(ctx, "test message2")
// t.Logf("xxx %s", buf.Bytes())

View File

@@ -1,294 +0,0 @@
package metadata
import (
"context"
"fmt"
"strings"
)
// In the metadata package, context and metadata are treated as immutable.
// Deep copies of metadata are made to keep things safe and correct.
// If a user takes a map and changes it across threads, it's their responsibility.
//
// 1. Incoming Context
//
// This context is provided by an external system and populated by the server or broker of the micro framework.
// It should not be modified. The idea is to extract all necessary data from it,
// validate the data, and transfer it into the current context.
// After that, only the current context should be used throughout the code.
//
// 2. Current Context
//
// This is the context used during the execution flow.
// You can add any needed metadata to it and pass it through your code.
//
// 3. Outgoing Context
//
// This context is for sending data to external systems.
// You can add what you need before sending it out.
// But its usually better to build and prepare this context right before making the external call,
// instead of changing it in many places.
//
// Execution Flow:
//
// [External System]
// ↓
// [Incoming Context]
// ↓
// [Extract & Validate Metadata from Incoming Context]
// ↓
// [Prepare Current Context]
// ↓
// [Enrich Current Context]
// ↓
// [Business Logic]
// ↓
// [Prepare Outgoing Context]
// ↓
// [External System Call]
type (
metadataCurrentKey struct{}
metadataIncomingKey struct{}
metadataOutgoingKey struct{}
rawMetadata struct {
md Metadata
added [][]string
}
)
// NewContext creates a new context with the provided Metadata attached.
// The Metadata must not be modified after calling this function.
func NewContext(ctx context.Context, md Metadata) context.Context {
return context.WithValue(ctx, metadataCurrentKey{}, rawMetadata{md: md})
}
// NewIncomingContext creates a new context with the provided incoming Metadata attached.
// The Metadata must not be modified after calling this function.
func NewIncomingContext(ctx context.Context, md Metadata) context.Context {
return context.WithValue(ctx, metadataIncomingKey{}, rawMetadata{md: md})
}
// NewOutgoingContext creates a new context with the provided outgoing Metadata attached.
// The Metadata must not be modified after calling this function.
func NewOutgoingContext(ctx context.Context, md Metadata) context.Context {
return context.WithValue(ctx, metadataOutgoingKey{}, rawMetadata{md: md})
}
// AppendContext returns a new context with the provided key-value pairs (kv)
// merged with any existing metadata in the context. For a description of kv,
// please refer to the Pairs documentation.
func AppendContext(ctx context.Context, kv ...string) context.Context {
if len(kv)%2 == 1 {
panic(fmt.Sprintf("metadata: AppendContext got an odd number of input pairs for metadata: %d", len(kv)))
}
md, _ := ctx.Value(metadataCurrentKey{}).(rawMetadata)
added := make([][]string, len(md.added)+1)
copy(added, md.added)
kvCopy := make([]string, 0, len(kv))
for i := 0; i < len(kv); i += 2 {
kvCopy = append(kvCopy, strings.ToLower(kv[i]), kv[i+1])
}
added[len(added)-1] = kvCopy
return context.WithValue(ctx, metadataCurrentKey{}, rawMetadata{md: md.md, added: added})
}
// AppendOutgoingContext returns a new context with the provided key-value pairs (kv)
// merged with any existing metadata in the context. For a description of kv,
// please refer to the Pairs documentation.
func AppendOutgoingContext(ctx context.Context, kv ...string) context.Context {
if len(kv)%2 == 1 {
panic(fmt.Sprintf("metadata: AppendOutgoingContext got an odd number of input pairs for metadata: %d", len(kv)))
}
md, _ := ctx.Value(metadataOutgoingKey{}).(rawMetadata)
added := make([][]string, len(md.added)+1)
copy(added, md.added)
kvCopy := make([]string, 0, len(kv))
for i := 0; i < len(kv); i += 2 {
kvCopy = append(kvCopy, strings.ToLower(kv[i]), kv[i+1])
}
added[len(added)-1] = kvCopy
return context.WithValue(ctx, metadataOutgoingKey{}, rawMetadata{md: md.md, added: added})
}
// FromContext retrieves a deep copy of the metadata from the context and returns it
// with a boolean indicating if it was found.
func FromContext(ctx context.Context) (Metadata, bool) {
raw, ok := ctx.Value(metadataCurrentKey{}).(rawMetadata)
if !ok {
return nil, false
}
metadataSize := len(raw.md)
for i := range raw.added {
metadataSize += len(raw.added[i]) / 2
}
out := make(Metadata, metadataSize)
for k, v := range raw.md {
out[k] = copyOf(v)
}
for _, added := range raw.added {
if len(added)%2 == 1 {
panic(fmt.Sprintf("metadata: FromContext got an odd number of input pairs for metadata: %d", len(added)))
}
for i := 0; i < len(added); i += 2 {
out[added[i]] = append(out[added[i]], added[i+1])
}
}
return out, true
}
// MustContext retrieves a deep copy of the metadata from the context and panics
// if the metadata is not found.
func MustContext(ctx context.Context) Metadata {
md, ok := FromContext(ctx)
if !ok {
panic("missing metadata")
}
return md
}
// FromIncomingContext retrieves a deep copy of the metadata from the context and returns it
// with a boolean indicating if it was found.
func FromIncomingContext(ctx context.Context) (Metadata, bool) {
raw, ok := ctx.Value(metadataIncomingKey{}).(rawMetadata)
if !ok {
return nil, false
}
metadataSize := len(raw.md)
for i := range raw.added {
metadataSize += len(raw.added[i]) / 2
}
out := make(Metadata, metadataSize)
for k, v := range raw.md {
out[k] = copyOf(v)
}
for _, added := range raw.added {
if len(added)%2 == 1 {
panic(fmt.Sprintf("metadata: FromIncomingContext got an odd number of input pairs for metadata: %d", len(added)))
}
for i := 0; i < len(added); i += 2 {
out[added[i]] = append(out[added[i]], added[i+1])
}
}
return out, true
}
// MustIncomingContext retrieves a deep copy of the metadata from the context and panics
// if the metadata is not found.
func MustIncomingContext(ctx context.Context) Metadata {
md, ok := FromIncomingContext(ctx)
if !ok {
panic("missing metadata")
}
return md
}
// FromOutgoingContext retrieves a deep copy of the metadata from the context and returns it
// with a boolean indicating if it was found.
func FromOutgoingContext(ctx context.Context) (Metadata, bool) {
raw, ok := ctx.Value(metadataOutgoingKey{}).(rawMetadata)
if !ok {
return nil, false
}
metadataSize := len(raw.md)
for i := range raw.added {
metadataSize += len(raw.added[i]) / 2
}
out := make(Metadata, metadataSize)
for k, v := range raw.md {
out[k] = copyOf(v)
}
for _, added := range raw.added {
if len(added)%2 == 1 {
panic(fmt.Sprintf("metadata: FromOutgoingContext got an odd number of input pairs for metadata: %d", len(added)))
}
for i := 0; i < len(added); i += 2 {
out[added[i]] = append(out[added[i]], added[i+1])
}
}
return out, ok
}
// MustOutgoingContext retrieves a deep copy of the metadata from the context and panics
// if the metadata is not found.
func MustOutgoingContext(ctx context.Context) Metadata {
md, ok := FromOutgoingContext(ctx)
if !ok {
panic("missing metadata")
}
return md
}
// ValueFromCurrentContext retrieves a deep copy of the metadata for the given key
// from the context, performing a case-insensitive search if needed. Returns nil if not found.
func ValueFromCurrentContext(ctx context.Context, key string) []string {
md, ok := ctx.Value(metadataCurrentKey{}).(rawMetadata)
if !ok {
return nil
}
if v, ok := md.md[key]; ok {
return copyOf(v)
}
for k, v := range md.md {
// Case-insensitive comparison: Metadata is a map, and there's no guarantee
// that the Metadata attached to the context is created using our helper
// functions.
if strings.EqualFold(k, key) {
return copyOf(v)
}
}
return nil
}
// ValueFromIncomingContext retrieves a deep copy of the metadata for the given key
// from the context, performing a case-insensitive search if needed. Returns nil if not found.
func ValueFromIncomingContext(ctx context.Context, key string) []string {
raw, ok := ctx.Value(metadataIncomingKey{}).(rawMetadata)
if !ok {
return nil
}
if v, ok := raw.md[key]; ok {
return copyOf(v)
}
for k, v := range raw.md {
// Case-insensitive comparison: Metadata is a map, and there's no guarantee
// that the Metadata attached to the context is created using our helper
// functions.
if strings.EqualFold(k, key) {
return copyOf(v)
}
}
return nil
}
// ValueFromOutgoingContext retrieves a deep copy of the metadata for the given key
// from the context, performing a case-insensitive search if needed. Returns nil if not found.
func ValueFromOutgoingContext(ctx context.Context, key string) []string {
md, ok := ctx.Value(metadataOutgoingKey{}).(rawMetadata)
if !ok {
return nil
}
if v, ok := md.md[key]; ok {
return copyOf(v)
}
for k, v := range md.md {
// Case-insensitive comparison: Metadata is a map, and there's no guarantee
// that the Metadata attached to the context is created using our helper
// functions.
if strings.EqualFold(k, key) {
return copyOf(v)
}
}
return nil
}

View File

@@ -2,18 +2,18 @@
package metadata
var (
// HeaderTopic is the header name that contains topic name.
// HeaderTopic is the header name that contains topic name
HeaderTopic = "Micro-Topic"
// HeaderContentType specifies content type of message.
// HeaderContentType specifies content type of message
HeaderContentType = "Content-Type"
// HeaderEndpoint specifies endpoint in service.
// HeaderEndpoint specifies endpoint in service
HeaderEndpoint = "Micro-Endpoint"
// HeaderService specifies service.
// HeaderService specifies service
HeaderService = "Micro-Service"
// HeaderTimeout specifies timeout of operation.
// HeaderTimeout specifies timeout of operation
HeaderTimeout = "Micro-Timeout"
// HeaderAuthorization specifies Authorization header.
// HeaderAuthorization specifies Authorization header
HeaderAuthorization = "Authorization"
// HeaderXRequestID specifies request id.
// HeaderXRequestID specifies request id
HeaderXRequestID = "X-Request-Id"
)

View File

@@ -1,7 +0,0 @@
package metadata
func copyOf(v []string) []string {
vals := make([]string, len(v))
copy(vals, v)
return vals
}

View File

@@ -1,37 +0,0 @@
package metadata
import "sort"
type Iterator struct {
md Metadata
keys []string
cur int
cnt int
}
// Next advances the iterator to the next element.
func (iter *Iterator) Next(k *string, v *[]string) bool {
if iter.cur+1 > iter.cnt {
return false
}
if k != nil && v != nil {
*k = iter.keys[iter.cur]
vv := iter.md[*k]
*v = make([]string, len(vv))
copy(*v, vv)
iter.cur++
}
return true
}
// Iterator returns an iterator for iterating over metadata in sorted order.
func (md Metadata) Iterator() *Iterator {
iter := &Iterator{md: md, cnt: len(md)}
iter.keys = make([]string, 0, iter.cnt)
for k := range md {
iter.keys = append(iter.keys, k)
}
sort.Strings(iter.keys)
return iter
}

View File

@@ -1,18 +1,21 @@
package metadata
import (
"context"
"fmt"
"net/textproto"
"sort"
"strings"
)
// defaultMetadataSize is used when initializing new Metadata.
// defaultMetadataSize used when need to init new Metadata
var defaultMetadataSize = 2
// Metadata maps keys to values. Use the New, NewWithMetadata and Pairs functions to create it.
// Metadata is a mapping from metadata keys to values. Users should use the following
// two convenience functions New and Pairs to generate Metadata.
type Metadata map[string][]string
// New creates a zero-value Metadata with the specified size.
// New creates an zero Metadata.
func New(l int) Metadata {
if l == 0 {
l = defaultMetadataSize
@@ -21,7 +24,7 @@ func New(l int) Metadata {
return md
}
// NewWithMetadata creates a Metadata from the provided key-value map.
// NewWithMetadata creates an Metadata from a given key-value map.
func NewWithMetadata(m map[string]string) Metadata {
md := make(Metadata, len(m))
for key, val := range m {
@@ -30,7 +33,8 @@ func NewWithMetadata(m map[string]string) Metadata {
return md
}
// Pairs returns a Metadata formed from the key-value mapping. It panics if the length of kv is odd.
// Pairs returns an Metadata formed by the mapping of key, value ...
// Pairs panics if len(kv) is odd.
func Pairs(kv ...string) Metadata {
if len(kv)%2 == 1 {
panic(fmt.Sprintf("metadata: Pairs got the odd number of input pairs for metadata: %d", len(kv)))
@@ -42,19 +46,12 @@ func Pairs(kv ...string) Metadata {
return md
}
// Join combines multiple Metadatas into a single Metadata.
// The order of values for each key is determined by the order in which the Metadatas are provided to Join.
func Join(mds ...Metadata) Metadata {
out := Metadata{}
for _, md := range mds {
for k, v := range md {
out[k] = append(out[k], v...)
}
}
return out
// Len returns the number of items in Metadata.
func (md Metadata) Len() int {
return len(md)
}
// Copy returns a deep copy of Metadata.
// Copy returns a copy of Metadata.
func Copy(src Metadata) Metadata {
out := make(Metadata, len(src))
for k, v := range src {
@@ -63,7 +60,7 @@ func Copy(src Metadata) Metadata {
return out
}
// Copy returns a deep copy of Metadata.
// Copy returns a copy of Metadata.
func (md Metadata) Copy() Metadata {
out := make(Metadata, len(md))
for k, v := range md {
@@ -72,28 +69,8 @@ func (md Metadata) Copy() Metadata {
return out
}
// CopyTo performs a deep copy of Metadata to the out.
func (md Metadata) CopyTo(out Metadata) {
for k, v := range md {
out[k] = copyOf(v)
}
}
// Len returns the number of items in Metadata.
func (md Metadata) Len() int {
return len(md)
}
// AsMap returns a deep copy of Metadata as a map[string]string
func (md Metadata) AsMap() map[string]string {
out := make(map[string]string, len(md))
for k, v := range md {
out[k] = strings.Join(v, ",")
}
return out
}
// AsHTTP1 returns a deep copy of Metadata with keys converted to canonical MIME header key format.
// AsHTTP1 returns a copy of Metadata
// with CanonicalMIMEHeaderKey.
func (md Metadata) AsHTTP1() map[string][]string {
out := make(map[string][]string, len(md))
for k, v := range md {
@@ -102,7 +79,8 @@ func (md Metadata) AsHTTP1() map[string][]string {
return out
}
// AsHTTP2 returns a deep copy of Metadata with keys converted to lowercase.
// AsHTTP1 returns a copy of Metadata
// with strings.ToLower.
func (md Metadata) AsHTTP2() map[string][]string {
out := make(map[string][]string, len(md))
for k, v := range md {
@@ -111,35 +89,75 @@ func (md Metadata) AsHTTP2() map[string][]string {
return out
}
// Get retrieves the values for a given key, checking the key in three formats:
// - exact case,
// - lower case,
// - canonical MIME header key format.
func (md Metadata) Get(k string) []string {
// CopyTo copies Metadata to out.
func (md Metadata) CopyTo(out Metadata) {
for k, v := range md {
out[k] = copyOf(v)
}
}
// Get obtains the values for a given key.
func (md Metadata) MustGet(k string) []string {
v, ok := md.Get(k)
if !ok {
panic("missing metadata key")
}
return v
}
// Get obtains the values for a given key.
func (md Metadata) Get(k string) ([]string, bool) {
v, ok := md[k]
if !ok {
v, ok = md[strings.ToLower(k)]
}
if !ok {
v = md[textproto.CanonicalMIMEHeaderKey(k)]
v, ok = md[textproto.CanonicalMIMEHeaderKey(k)]
}
return v, ok
}
// MustGetJoined obtains the values for a given key
// with joined values with "," symbol
func (md Metadata) MustGetJoined(k string) string {
v, ok := md.GetJoined(k)
if !ok {
panic("missing metadata key")
}
return v
}
// GetJoined retrieves the values for a given key and joins them into a single string, separated by commas.
func (md Metadata) GetJoined(k string) string {
return strings.Join(md.Get(k), ",")
// GetJoined obtains the values for a given key
// with joined values with "," symbol
func (md Metadata) GetJoined(k string) (string, bool) {
v, ok := md.Get(k)
if !ok {
return "", ok
}
return strings.Join(v, ","), true
}
// Set assigns the values to the given key.
func (md Metadata) Set(key string, vals ...string) {
// Set sets the value of a given key with a slice of values.
func (md Metadata) Add(key string, vals ...string) {
if len(vals) == 0 {
return
}
md[key] = vals
}
// Append adds values to the existing values for the given key.
// Set sets the value of a given key with a slice of values.
func (md Metadata) Set(kvs ...string) {
if len(kvs)%2 == 1 {
panic(fmt.Sprintf("metadata: Set got an odd number of input pairs for metadata: %d", len(kvs)))
}
for i := 0; i < len(kvs); i += 2 {
md[kvs[i]] = append(md[kvs[i]], kvs[i+1])
}
}
// Append adds the values to key k, not overwriting what was already stored at
// that key.
func (md Metadata) Append(key string, vals ...string) {
if len(vals) == 0 {
return
@@ -147,10 +165,7 @@ func (md Metadata) Append(key string, vals ...string) {
md[key] = append(md[key], vals...)
}
// Del removes the values for the given keys k. It checks and removes the keys in the following formats:
// - exact case,
// - lower case,
// - canonical MIME header key format.
// Del removes the values for a given keys k.
func (md Metadata) Del(k ...string) {
for i := range k {
delete(md, k[i])
@@ -158,3 +173,321 @@ func (md Metadata) Del(k ...string) {
delete(md, textproto.CanonicalMIMEHeaderKey(k[i]))
}
}
// Join joins any number of Metadatas into a single Metadata.
//
// The order of values for each key is determined by the order in which the Metadatas
// containing those values are presented to Join.
func Join(mds ...Metadata) Metadata {
out := Metadata{}
for _, Metadata := range mds {
for k, v := range Metadata {
out[k] = append(out[k], v...)
}
}
return out
}
type (
metadataIncomingKey struct{}
metadataOutgoingKey struct{}
metadataCurrentKey struct{}
)
// NewContext creates a new context with Metadata attached. Metadata must
// not be modified after calling this function.
func NewContext(ctx context.Context, md Metadata) context.Context {
return context.WithValue(ctx, metadataCurrentKey{}, rawMetadata{md: md})
}
// NewIncomingContext creates a new context with incoming Metadata attached. Metadata must
// not be modified after calling this function.
func NewIncomingContext(ctx context.Context, md Metadata) context.Context {
return context.WithValue(ctx, metadataIncomingKey{}, rawMetadata{md: md})
}
// NewOutgoingContext creates a new context with outgoing Metadata attached. If used
// in conjunction with AppendOutgoingContext, NewOutgoingContext will
// overwrite any previously-appended metadata. Metadata must not be modified after
// calling this function.
func NewOutgoingContext(ctx context.Context, md Metadata) context.Context {
return context.WithValue(ctx, metadataOutgoingKey{}, rawMetadata{md: md})
}
// AppendContext returns a new context with the provided kv merged
// with any existing metadata in the context. Please refer to the documentation
// of Pairs for a description of kv.
func AppendContext(ctx context.Context, kv ...string) context.Context {
if len(kv)%2 == 1 {
panic(fmt.Sprintf("metadata: AppendContext got an odd number of input pairs for metadata: %d", len(kv)))
}
md, _ := ctx.Value(metadataCurrentKey{}).(rawMetadata)
added := make([][]string, len(md.added)+1)
copy(added, md.added)
kvCopy := make([]string, 0, len(kv))
for i := 0; i < len(kv); i += 2 {
kvCopy = append(kvCopy, strings.ToLower(kv[i]), kv[i+1])
}
added[len(added)-1] = kvCopy
return context.WithValue(ctx, metadataCurrentKey{}, rawMetadata{md: md.md, added: added})
}
// AppendIncomingContext returns a new context with the provided kv merged
// with any existing metadata in the context. Please refer to the documentation
// of Pairs for a description of kv.
func AppendIncomingContext(ctx context.Context, kv ...string) context.Context {
if len(kv)%2 == 1 {
panic(fmt.Sprintf("metadata: AppendIncomingContext got an odd number of input pairs for metadata: %d", len(kv)))
}
md, _ := ctx.Value(metadataIncomingKey{}).(rawMetadata)
added := make([][]string, len(md.added)+1)
copy(added, md.added)
kvCopy := make([]string, 0, len(kv))
for i := 0; i < len(kv); i += 2 {
kvCopy = append(kvCopy, strings.ToLower(kv[i]), kv[i+1])
}
added[len(added)-1] = kvCopy
return context.WithValue(ctx, metadataIncomingKey{}, rawMetadata{md: md.md, added: added})
}
// AppendOutgoingContext returns a new context with the provided kv merged
// with any existing metadata in the context. Please refer to the documentation
// of Pairs for a description of kv.
func AppendOutgoingContext(ctx context.Context, kv ...string) context.Context {
if len(kv)%2 == 1 {
panic(fmt.Sprintf("metadata: AppendOutgoingContext got an odd number of input pairs for metadata: %d", len(kv)))
}
md, _ := ctx.Value(metadataOutgoingKey{}).(rawMetadata)
added := make([][]string, len(md.added)+1)
copy(added, md.added)
kvCopy := make([]string, 0, len(kv))
for i := 0; i < len(kv); i += 2 {
kvCopy = append(kvCopy, strings.ToLower(kv[i]), kv[i+1])
}
added[len(added)-1] = kvCopy
return context.WithValue(ctx, metadataOutgoingKey{}, rawMetadata{md: md.md, added: added})
}
// FromContext returns the metadata in ctx if it exists.
func FromContext(ctx context.Context) (Metadata, bool) {
raw, ok := ctx.Value(metadataCurrentKey{}).(rawMetadata)
if !ok {
return nil, false
}
metadataSize := len(raw.md)
for i := range raw.added {
metadataSize += len(raw.added[i]) / 2
}
out := make(Metadata, metadataSize)
for k, v := range raw.md {
out[k] = copyOf(v)
}
for _, added := range raw.added {
if len(added)%2 == 1 {
panic(fmt.Sprintf("metadata: FromContext got an odd number of input pairs for metadata: %d", len(added)))
}
for i := 0; i < len(added); i += 2 {
out[added[i]] = append(out[added[i]], added[i+1])
}
}
return out, true
}
// MustContext returns the metadata in ctx.
func MustContext(ctx context.Context) Metadata {
md, ok := FromContext(ctx)
if !ok {
panic("missing metadata")
}
return md
}
// FromIncomingContext returns the incoming metadata in ctx if it exists.
func FromIncomingContext(ctx context.Context) (Metadata, bool) {
raw, ok := ctx.Value(metadataIncomingKey{}).(rawMetadata)
if !ok {
return nil, false
}
metadataSize := len(raw.md)
for i := range raw.added {
metadataSize += len(raw.added[i]) / 2
}
out := make(Metadata, metadataSize)
for k, v := range raw.md {
out[k] = copyOf(v)
}
for _, added := range raw.added {
if len(added)%2 == 1 {
panic(fmt.Sprintf("metadata: FromIncomingContext got an odd number of input pairs for metadata: %d", len(added)))
}
for i := 0; i < len(added); i += 2 {
out[added[i]] = append(out[added[i]], added[i+1])
}
}
return out, true
}
// MustIncomingContext returns the incoming metadata in ctx.
func MustIncomingContext(ctx context.Context) Metadata {
md, ok := FromIncomingContext(ctx)
if !ok {
panic("missing metadata")
}
return md
}
// ValueFromIncomingContext returns the metadata value corresponding to the metadata
// key from the incoming metadata if it exists. Keys are matched in a case insensitive
// manner.
func ValueFromIncomingContext(ctx context.Context, key string) []string {
raw, ok := ctx.Value(metadataIncomingKey{}).(rawMetadata)
if !ok {
return nil
}
if v, ok := raw.md[key]; ok {
return copyOf(v)
}
for k, v := range raw.md {
// Case insensitive comparison: Metadata is a map, and there's no guarantee
// that the Metadata attached to the context is created using our helper
// functions.
if strings.EqualFold(k, key) {
return copyOf(v)
}
}
return nil
}
// ValueFromCurrentContext returns the metadata value corresponding to the metadata
// key from the incoming metadata if it exists. Keys are matched in a case insensitive
// manner.
func ValueFromCurrentContext(ctx context.Context, key string) []string {
md, ok := ctx.Value(metadataCurrentKey{}).(rawMetadata)
if !ok {
return nil
}
if v, ok := md.md[key]; ok {
return copyOf(v)
}
for k, v := range md.md {
// Case insensitive comparison: Metadata is a map, and there's no guarantee
// that the Metadata attached to the context is created using our helper
// functions.
if strings.EqualFold(k, key) {
return copyOf(v)
}
}
return nil
}
// MustOutgoingContext returns the outgoing metadata in ctx.
func MustOutgoingContext(ctx context.Context) Metadata {
md, ok := FromOutgoingContext(ctx)
if !ok {
panic("missing metadata")
}
return md
}
// ValueFromOutgoingContext returns the metadata value corresponding to the metadata
// key from the incoming metadata if it exists. Keys are matched in a case insensitive
// manner.
func ValueFromOutgoingContext(ctx context.Context, key string) []string {
md, ok := ctx.Value(metadataOutgoingKey{}).(rawMetadata)
if !ok {
return nil
}
if v, ok := md.md[key]; ok {
return copyOf(v)
}
for k, v := range md.md {
// Case insensitive comparison: Metadata is a map, and there's no guarantee
// that the Metadata attached to the context is created using our helper
// functions.
if strings.EqualFold(k, key) {
return copyOf(v)
}
}
return nil
}
func copyOf(v []string) []string {
vals := make([]string, len(v))
copy(vals, v)
return vals
}
// FromOutgoingContext returns the outgoing metadata in ctx if it exists.
func FromOutgoingContext(ctx context.Context) (Metadata, bool) {
raw, ok := ctx.Value(metadataOutgoingKey{}).(rawMetadata)
if !ok {
return nil, false
}
metadataSize := len(raw.md)
for i := range raw.added {
metadataSize += len(raw.added[i]) / 2
}
out := make(Metadata, metadataSize)
for k, v := range raw.md {
out[k] = copyOf(v)
}
for _, added := range raw.added {
if len(added)%2 == 1 {
panic(fmt.Sprintf("metadata: FromOutgoingContext got an odd number of input pairs for metadata: %d", len(added)))
}
for i := 0; i < len(added); i += 2 {
out[added[i]] = append(out[added[i]], added[i+1])
}
}
return out, ok
}
type rawMetadata struct {
md Metadata
added [][]string
}
// Iterator used to iterate over metadata with order
type Iterator struct {
md Metadata
keys []string
cur int
cnt int
}
// Next advance iterator to next element
func (iter *Iterator) Next(k *string, v *[]string) bool {
if iter.cur+1 > iter.cnt {
return false
}
if k != nil && v != nil {
*k = iter.keys[iter.cur]
vv := iter.md[*k]
*v = make([]string, len(vv))
copy(*v, vv)
iter.cur++
}
return true
}
// Iterator returns the itarator for metadata in sorted order
func (md Metadata) Iterator() *Iterator {
iter := &Iterator{md: md, cnt: len(md)}
iter.keys = make([]string, 0, iter.cnt)
for k := range md {
iter.keys = append(iter.keys, k)
}
sort.Strings(iter.keys)
return iter
}

View File

@@ -5,15 +5,6 @@ import (
"testing"
)
func TesSet(t *testing.T) {
md := Pairs("key1", "val1", "key2", "val2")
md.Set("key1", "val2", "val3")
v := md.GetJoined("X-Request-Id")
if v != "val2, val3" {
t.Fatal("set not works")
}
}
/*
func TestAppendOutgoingContextModify(t *testing.T) {
md := Pairs("key1", "val1")
@@ -28,8 +19,8 @@ func TestAppendOutgoingContextModify(t *testing.T) {
func TestLowercase(t *testing.T) {
md := New(1)
md["x-request-id"] = []string{"12345"}
v := md.GetJoined("X-Request-Id")
if v == "" {
v, ok := md.GetJoined("X-Request-Id")
if !ok || v == "" {
t.Fatalf("metadata invalid %#+v", md)
}
}
@@ -56,9 +47,33 @@ func TestMultipleUsage(t *testing.T) {
_ = omd
}
func TestMetadataSetMultiple(t *testing.T) {
md := New(4)
md.Set("key1", "val1", "key2", "val2")
if v, ok := md.GetJoined("key1"); !ok || v != "val1" {
t.Fatalf("invalid kv %#+v", md)
}
if v, ok := md.GetJoined("key2"); !ok || v != "val2" {
t.Fatalf("invalid kv %#+v", md)
}
}
func TestAppend(t *testing.T) {
ctx := context.Background()
ctx = AppendIncomingContext(ctx, "key1", "val1", "key2", "val2")
md, ok := FromIncomingContext(ctx)
if !ok {
t.Fatal("metadata empty")
}
if _, ok := md.Get("key1"); !ok {
t.Fatal("key1 not found")
}
}
func TestPairs(t *testing.T) {
md := Pairs("key1", "val1", "key2", "val2")
if v := md.Get("key1"); v == nil {
if _, ok := md.Get("key1"); !ok {
t.Fatal("key1 not found")
}
}
@@ -82,7 +97,7 @@ func TestPassing(t *testing.T) {
if !ok {
t.Fatalf("missing metadata from outgoing context")
}
if v := md.Get("Key1"); v == nil || v[0] != "Val1" {
if v, ok := md.Get("Key1"); !ok || v[0] != "Val1" {
t.Fatalf("invalid metadata value %#+v", md)
}
}
@@ -112,21 +127,21 @@ func TestIterator(t *testing.T) {
func TestMedataCanonicalKey(t *testing.T) {
md := New(1)
md.Set("x-request-id", "12345")
v := md.GetJoined("x-request-id")
if v == "" {
v, ok := md.GetJoined("x-request-id")
if !ok {
t.Fatalf("failed to get x-request-id")
} else if v != "12345" {
t.Fatalf("invalid metadata value: %s != %s", "12345", v)
}
v = md.GetJoined("X-Request-Id")
if v == "" {
v, ok = md.GetJoined("X-Request-Id")
if !ok {
t.Fatalf("failed to get x-request-id")
} else if v != "12345" {
t.Fatalf("invalid metadata value: %s != %s", "12345", v)
}
v = md.GetJoined("X-Request-ID")
if v == "" {
v, ok = md.GetJoined("X-Request-ID")
if !ok {
t.Fatalf("failed to get x-request-id")
} else if v != "12345" {
t.Fatalf("invalid metadata value: %s != %s", "12345", v)
@@ -138,8 +153,8 @@ func TestMetadataSet(t *testing.T) {
md.Set("Key", "val")
val := md.GetJoined("Key")
if val == "" {
val, ok := md.GetJoined("Key")
if !ok {
t.Fatal("key Key not found")
}
if val != "val" {
@@ -154,8 +169,8 @@ func TestMetadataDelete(t *testing.T) {
}
md.Del("Baz")
v := md.Get("Baz")
if v != nil {
_, ok := md.Get("Baz")
if ok {
t.Fatal("key Baz not deleted")
}
}
@@ -254,6 +269,20 @@ func TestNewOutgoingContext(t *testing.T) {
}
}
func TestAppendIncomingContext(t *testing.T) {
md := New(1)
md.Set("key1", "val1")
ctx := AppendIncomingContext(context.TODO(), "key2", "val2")
nmd, ok := FromIncomingContext(ctx)
if nmd == nil || !ok {
t.Fatal("AppendIncomingContext not works")
}
if v, ok := nmd.GetJoined("key2"); !ok || v != "val2" {
t.Fatal("AppendIncomingContext not works")
}
}
func TestAppendOutgoingContext(t *testing.T) {
md := New(1)
md.Set("key1", "val1")
@@ -263,7 +292,7 @@ func TestAppendOutgoingContext(t *testing.T) {
if nmd == nil || !ok {
t.Fatal("AppendOutgoingContext not works")
}
if v := nmd.GetJoined("key2"); v != "val2" {
if v, ok := nmd.GetJoined("key2"); !ok || v != "val2" {
t.Fatal("AppendOutgoingContext not works")
}
}

View File

@@ -11,8 +11,8 @@ import (
)
type httpProfile struct {
server *http.Server
mu sync.Mutex
server *http.Server
sync.Mutex
running bool
}
@@ -21,8 +21,8 @@ var DefaultAddress = ":6060"
// Start the profiler
func (h *httpProfile) Start() error {
h.mu.Lock()
defer h.mu.Unlock()
h.Lock()
defer h.Unlock()
if h.running {
return nil
@@ -30,9 +30,9 @@ func (h *httpProfile) Start() error {
go func() {
if err := h.server.ListenAndServe(); err != nil {
h.mu.Lock()
h.Lock()
h.running = false
h.mu.Unlock()
h.Unlock()
}
}()
@@ -43,8 +43,8 @@ func (h *httpProfile) Start() error {
// Stop the profiler
func (h *httpProfile) Stop() error {
h.mu.Lock()
defer h.mu.Unlock()
h.Lock()
defer h.Unlock()
if !h.running {
return nil

View File

@@ -17,7 +17,7 @@ type profiler struct {
cpuFile *os.File
memFile *os.File
opts profile.Options
mu sync.Mutex
sync.Mutex
running bool
}
@@ -39,8 +39,8 @@ func (p *profiler) writeHeap(f *os.File) {
}
func (p *profiler) Start() error {
p.mu.Lock()
defer p.mu.Unlock()
p.Lock()
defer p.Unlock()
if p.running {
return nil
@@ -86,8 +86,8 @@ func (p *profiler) Start() error {
}
func (p *profiler) Stop() error {
p.mu.Lock()
defer p.mu.Unlock()
p.Lock()
defer p.Unlock()
select {
case <-p.exit:

View File

@@ -33,7 +33,7 @@ type memory struct {
records map[string]services
watchers map[string]*watcher
opts register.Options
mu sync.RWMutex
sync.RWMutex
}
// services is a KV map with service name as the key and a map of records as the value
@@ -57,7 +57,7 @@ func (m *memory) ttlPrune() {
defer prune.Stop()
for range prune.C {
m.mu.Lock()
m.Lock()
for namespace, services := range m.records {
for service, versions := range services {
for version, record := range versions {
@@ -72,24 +72,24 @@ func (m *memory) ttlPrune() {
}
}
}
m.mu.Unlock()
m.Unlock()
}
}
func (m *memory) sendEvent(r *register.Result) {
m.mu.RLock()
m.RLock()
watchers := make([]*watcher, 0, len(m.watchers))
for _, w := range m.watchers {
watchers = append(watchers, w)
}
m.mu.RUnlock()
m.RUnlock()
for _, w := range watchers {
select {
case <-w.exit:
m.mu.Lock()
m.Lock()
delete(m.watchers, w.id)
m.mu.Unlock()
m.Unlock()
default:
select {
case w.res <- r:
@@ -113,8 +113,8 @@ func (m *memory) Init(opts ...register.Option) error {
}
// add services
m.mu.Lock()
defer m.mu.Unlock()
m.Lock()
defer m.Unlock()
return nil
}
@@ -124,8 +124,8 @@ func (m *memory) Options() register.Options {
}
func (m *memory) Register(_ context.Context, s *register.Service, opts ...register.RegisterOption) error {
m.mu.Lock()
defer m.mu.Unlock()
m.Lock()
defer m.Unlock()
options := register.NewRegisterOptions(opts...)
@@ -197,8 +197,8 @@ func (m *memory) Register(_ context.Context, s *register.Service, opts ...regist
}
func (m *memory) Deregister(ctx context.Context, s *register.Service, opts ...register.DeregisterOption) error {
m.mu.Lock()
defer m.mu.Unlock()
m.Lock()
defer m.Unlock()
options := register.NewDeregisterOptions(opts...)
@@ -264,9 +264,9 @@ func (m *memory) LookupService(ctx context.Context, name string, opts ...registe
// if it's a wildcard domain, return from all domains
if options.Namespace == register.WildcardNamespace {
m.mu.RLock()
m.RLock()
recs := m.records
m.mu.RUnlock()
m.RUnlock()
var services []*register.Service
@@ -286,8 +286,8 @@ func (m *memory) LookupService(ctx context.Context, name string, opts ...registe
return services, nil
}
m.mu.RLock()
defer m.mu.RUnlock()
m.RLock()
defer m.RUnlock()
// check the domain exists
services, ok := m.records[options.Namespace]
@@ -319,9 +319,9 @@ func (m *memory) ListServices(ctx context.Context, opts ...register.ListOption)
// if it's a wildcard domain, list from all domains
if options.Namespace == register.WildcardNamespace {
m.mu.RLock()
m.RLock()
recs := m.records
m.mu.RUnlock()
m.RUnlock()
var services []*register.Service
@@ -336,8 +336,8 @@ func (m *memory) ListServices(ctx context.Context, opts ...register.ListOption)
return services, nil
}
m.mu.RLock()
defer m.mu.RUnlock()
m.RLock()
defer m.RUnlock()
// ensure the domain exists
services, ok := m.records[options.Namespace]
@@ -371,9 +371,9 @@ func (m *memory) Watch(ctx context.Context, opts ...register.WatchOption) (regis
wo: wo,
}
m.mu.Lock()
m.Lock()
m.watchers[w.id] = w
m.mu.Unlock()
m.Unlock()
return w, nil
}

View File

@@ -69,8 +69,7 @@ type Service struct {
type Node struct {
Metadata metadata.Metadata `json:"metadata,omitempty"`
ID string `json:"id,omitempty"`
// Address also prefixed with scheme like grpc://xx.xx.xx.xx:1234
Address string `json:"address,omitempty"`
Address string `json:"address,omitempty"`
}
// Option func signature

View File

@@ -51,13 +51,13 @@ func (r *rpcHandler) Options() HandlerOptions {
}
type noopServer struct {
h Handler
wg *sync.WaitGroup
rsvc *register.Service
handlers map[string]Handler
exit chan chan error
opts Options
mu sync.RWMutex
h Handler
wg *sync.WaitGroup
rsvc *register.Service
handlers map[string]Handler
exit chan chan error
opts Options
sync.RWMutex
registered bool
started bool
}
@@ -125,10 +125,10 @@ func (n *noopServer) String() string {
//nolint:gocyclo
func (n *noopServer) Register() error {
n.mu.RLock()
n.RLock()
rsvc := n.rsvc
config := n.opts
n.mu.RUnlock()
n.RUnlock()
// if service already filled, reuse it and return early
if rsvc != nil {
@@ -144,9 +144,9 @@ func (n *noopServer) Register() error {
return err
}
n.mu.RLock()
n.RLock()
registered := n.registered
n.mu.RUnlock()
n.RUnlock()
if !registered {
if config.Logger.V(logger.InfoLevel) {
@@ -164,8 +164,8 @@ func (n *noopServer) Register() error {
return nil
}
n.mu.Lock()
defer n.mu.Unlock()
n.Lock()
defer n.Unlock()
n.registered = true
if cacheService {
@@ -178,9 +178,9 @@ func (n *noopServer) Register() error {
func (n *noopServer) Deregister() error {
var err error
n.mu.RLock()
n.RLock()
config := n.opts
n.mu.RUnlock()
n.RUnlock()
service, err := NewRegisterService(n)
if err != nil {
@@ -195,29 +195,29 @@ func (n *noopServer) Deregister() error {
return err
}
n.mu.Lock()
n.Lock()
n.rsvc = nil
if !n.registered {
n.mu.Unlock()
n.Unlock()
return nil
}
n.registered = false
n.mu.Unlock()
n.Unlock()
return nil
}
//nolint:gocyclo
func (n *noopServer) Start() error {
n.mu.RLock()
n.RLock()
if n.started {
n.mu.RUnlock()
n.RUnlock()
return nil
}
config := n.Options()
n.mu.RUnlock()
n.RUnlock()
// use 127.0.0.1 to avoid scan of all network interfaces
addr, err := maddr.Extract("127.0.0.1")
@@ -235,11 +235,11 @@ func (n *noopServer) Start() error {
config.Logger.Info(n.opts.Context, "server [noop] Listening on "+config.Address)
}
n.mu.Lock()
n.Lock()
if len(config.Advertise) == 0 {
config.Advertise = config.Address
}
n.mu.Unlock()
n.Unlock()
// use RegisterCheck func before register
// nolint: nestif
@@ -273,9 +273,9 @@ func (n *noopServer) Start() error {
select {
// register self on interval
case <-t.C:
n.mu.RLock()
n.RLock()
registered := n.registered
n.mu.RUnlock()
n.RUnlock()
rerr := config.RegisterCheck(config.Context)
// nolint: nestif
if rerr != nil && registered {
@@ -332,29 +332,29 @@ func (n *noopServer) Start() error {
}()
// mark the server as started
n.mu.Lock()
n.Lock()
n.started = true
n.mu.Unlock()
n.Unlock()
return nil
}
func (n *noopServer) Stop() error {
n.mu.RLock()
n.RLock()
if !n.started {
n.mu.RUnlock()
n.RUnlock()
return nil
}
n.mu.RUnlock()
n.RUnlock()
ch := make(chan error)
n.exit <- ch
err := <-ch
n.mu.Lock()
n.Lock()
n.rsvc = nil
n.started = false
n.mu.Unlock()
n.Unlock()
return err
}

View File

@@ -96,10 +96,9 @@ func RegisterHandler(s server.Server, h interface{}, opts ...server.HandlerOptio
}
type service struct {
done chan struct{}
opts Options
mu sync.RWMutex
stopped bool
done chan struct{}
opts Options
sync.RWMutex
}
// NewService creates and returns a new Service based on the packages within.
@@ -321,9 +320,9 @@ func (s *service) Health() bool {
func (s *service) Start() error {
var err error
s.mu.RLock()
s.RLock()
config := s.opts
s.mu.RUnlock()
s.RUnlock()
for _, cfg := range s.opts.Configs {
if cfg.Options().Struct == nil {
@@ -380,9 +379,9 @@ func (s *service) Start() error {
}
func (s *service) Stop() error {
s.mu.RLock()
s.RLock()
config := s.opts
s.mu.RUnlock()
s.RUnlock()
if config.Loggers[0].V(logger.InfoLevel) {
config.Loggers[0].Info(s.opts.Context, fmt.Sprintf("stoppping [service] %s", s.Name()))
@@ -425,7 +424,7 @@ func (s *service) Stop() error {
}
}
s.notifyShutdown()
close(s.done)
return nil
}
@@ -449,23 +448,10 @@ func (s *service) Run() error {
return err
}
// wait on context cancel
<-s.done
return nil
}
// notifyShutdown marks the service as stopped and closes the done channel.
// It ensures the channel is closed only once, preventing multiple closures.
func (s *service) notifyShutdown() {
s.mu.Lock()
if s.stopped {
s.mu.Unlock()
return
}
s.stopped = true
s.mu.Unlock()
close(s.done)
return s.Stop()
}
type Namer interface {

View File

@@ -3,9 +3,7 @@ package micro
import (
"reflect"
"testing"
"time"
"github.com/stretchr/testify/require"
"go.unistack.org/micro/v4/broker"
"go.unistack.org/micro/v4/client"
"go.unistack.org/micro/v4/config"
@@ -739,41 +737,3 @@ func Test_getNameIndex(t *testing.T) {
}
}
*/
func TestServiceShutdown(t *testing.T) {
defer func() {
if r := recover(); r != nil {
t.Fatalf("service shutdown failed: %v", r)
}
}()
s, ok := NewService().(*service)
require.NotNil(t, s)
require.True(t, ok)
require.NoError(t, s.Start())
require.False(t, s.stopped)
require.NoError(t, s.Stop())
require.True(t, s.stopped)
}
func TestServiceMultipleShutdowns(t *testing.T) {
defer func() {
if r := recover(); r != nil {
t.Fatalf("service shutdown failed: %v", r)
}
}()
s := NewService()
go func() {
time.Sleep(10 * time.Millisecond)
// first call
require.NoError(t, s.Stop())
// duplicate call
require.NoError(t, s.Stop())
}()
require.NoError(t, s.Run())
}

View File

@@ -139,7 +139,7 @@ func (n *noopStore) fnExists(ctx context.Context, _ string, _ ...ExistsOption) e
return ctx.Err()
default:
}
return ErrNotFound
return nil
}
func (n *noopStore) Write(ctx context.Context, key string, val interface{}, opts ...WriteOption) error {

View File

@@ -9,7 +9,7 @@ type memorySync struct {
locks map[string]*memoryLock
options Options
mu gosync.RWMutex
mtx gosync.RWMutex
}
type memoryLock struct {
@@ -74,7 +74,7 @@ func (m *memorySync) Options() Options {
func (m *memorySync) Lock(id string, opts ...LockOption) error {
// lock our access
m.mu.Lock()
m.mtx.Lock()
var options LockOptions
for _, o := range opts {
@@ -90,11 +90,11 @@ func (m *memorySync) Lock(id string, opts ...LockOption) error {
release: make(chan bool),
}
// unlock
m.mu.Unlock()
m.mtx.Unlock()
return nil
}
m.mu.Unlock()
m.mtx.Unlock()
// set wait time
var wait <-chan time.Time
@@ -124,12 +124,12 @@ lockLoop:
// wait for the lock to be released
select {
case <-lk.release:
m.mu.Lock()
m.mtx.Lock()
// someone locked before us
lk, ok = m.locks[id]
if ok {
m.mu.Unlock()
m.mtx.Unlock()
continue
}
@@ -141,7 +141,7 @@ lockLoop:
release: make(chan bool),
}
m.mu.Unlock()
m.mtx.Unlock()
break lockLoop
case <-ttl:
@@ -160,8 +160,8 @@ lockLoop:
}
func (m *memorySync) Unlock(id string) error {
m.mu.Lock()
defer m.mu.Unlock()
m.mtx.Lock()
defer m.mtx.Unlock()
lk, ok := m.locks[id]
// no lock exists

View File

@@ -46,10 +46,6 @@ func (s memoryStringer) String() string {
return s.s
}
func (t *Tracer) Enabled() bool {
return t.opts.Enabled
}
func (t *Tracer) Flush(_ context.Context) error {
return nil
}

View File

@@ -4,7 +4,7 @@ import (
"context"
"time"
"github.com/google/uuid"
"go.unistack.org/micro/v4/util/id"
)
var _ Tracer = (*noopTracer)(nil)
@@ -18,12 +18,6 @@ func (t *noopTracer) Spans() []Span {
return t.spans
}
var uuidNil = uuid.Nil.String()
func (t *noopTracer) Enabled() bool {
return t.opts.Enabled
}
func (t *noopTracer) Start(ctx context.Context, name string, opts ...SpanOption) (context.Context, Span) {
options := NewSpanOptions(opts...)
span := &noopSpan{
@@ -34,8 +28,8 @@ func (t *noopTracer) Start(ctx context.Context, name string, opts ...SpanOption)
labels: options.Labels,
kind: options.Kind,
}
span.spanID.s = uuidNil
span.traceID.s = uuidNil
span.spanID.s, _ = id.New()
span.traceID.s, _ = id.New()
if span.ctx == nil {
span.ctx = context.Background()
}

View File

@@ -142,8 +142,6 @@ type Options struct {
Name string
// ContextAttrFuncs contains funcs that provides tracing
ContextAttrFuncs []ContextAttrFunc
// Enabled specify trace status
Enabled bool
}
// Option func signature
@@ -183,7 +181,6 @@ func NewOptions(opts ...Option) Options {
Logger: logger.DefaultLogger,
Context: context.Background(),
ContextAttrFuncs: DefaultContextAttrFuncs,
Enabled: true,
}
for _, o := range opts {
o(&options)
@@ -197,10 +194,3 @@ func Name(n string) Option {
o.Name = n
}
}
// Disabled disable tracer
func Disabled(b bool) Option {
return func(o *Options) {
o.Enabled = !b
}
}

View File

@@ -51,8 +51,6 @@ type Tracer interface {
// Extract(ctx context.Context)
// Flush flushes spans
Flush(ctx context.Context) error
// Enabled returns tracer status
Enabled() bool
}
type Span interface {

View File

@@ -67,12 +67,6 @@ func (b *SeekerBuffer) Close() error {
return nil
}
// Reset clears all the data out of the buffer and sets the read position to 0.
func (b *SeekerBuffer) Reset() {
b.data = nil
b.pos = 0
}
// Len returns the length of data remaining to be read.
func (b *SeekerBuffer) Len() int {
return len(b.data[b.pos:])

View File

@@ -137,7 +137,7 @@ type cache struct {
opts Options
mu sync.RWMutex
sync.RWMutex
}
type cacheEntry struct {
@@ -171,7 +171,7 @@ func (c *cache) put(req string, res string) {
ttl = c.opts.MaxCacheTTL
}
c.mu.Lock()
c.Lock()
if c.entries == nil {
c.entries = make(map[string]cacheEntry)
}
@@ -207,7 +207,7 @@ func (c *cache) put(req string, res string) {
}
c.opts.Meter.Counter(semconv.CacheItemsTotal, "type", "dns").Inc()
c.mu.Unlock()
c.Unlock()
}
func (c *cache) get(req string) (res string) {
@@ -219,8 +219,8 @@ func (c *cache) get(req string) (res string) {
return ""
}
c.mu.RLock()
defer c.mu.RUnlock()
c.RLock()
defer c.RUnlock()
if c.entries == nil {
return ""

View File

@@ -20,7 +20,7 @@ type dnsConn struct {
ibuf bytes.Buffer
obuf bytes.Buffer
mu sync.Mutex
sync.Mutex
}
type roundTripper func(ctx context.Context, req string) (res string, err error)
@@ -42,15 +42,15 @@ func (c *dnsConn) Read(b []byte) (n int, err error) {
}
func (c *dnsConn) Write(b []byte) (n int, err error) {
c.mu.Lock()
defer c.mu.Unlock()
c.Lock()
defer c.Unlock()
return c.ibuf.Write(b)
}
func (c *dnsConn) Close() error {
c.mu.Lock()
c.Lock()
cancel := c.cancel
c.mu.Unlock()
c.Unlock()
if cancel != nil {
cancel()
@@ -78,9 +78,9 @@ func (c *dnsConn) SetDeadline(t time.Time) error {
}
func (c *dnsConn) SetReadDeadline(t time.Time) error {
c.mu.Lock()
c.Lock()
c.deadline = t
c.mu.Unlock()
c.Unlock()
return nil
}
@@ -90,8 +90,8 @@ func (c *dnsConn) SetWriteDeadline(_ time.Time) error {
}
func (c *dnsConn) drainBuffers(b []byte) (string, int, error) {
c.mu.Lock()
defer c.mu.Unlock()
c.Lock()
defer c.Unlock()
// drain the output buffer
if c.obuf.Len() > 0 {
@@ -119,8 +119,8 @@ func (c *dnsConn) drainBuffers(b []byte) (string, int, error) {
}
func (c *dnsConn) fillBuffer(b []byte, str string) (int, error) {
c.mu.Lock()
defer c.mu.Unlock()
c.Lock()
defer c.Unlock()
c.obuf.WriteByte(byte(len(str) >> 8))
c.obuf.WriteByte(byte(len(str)))
c.obuf.WriteString(str)
@@ -128,8 +128,8 @@ func (c *dnsConn) fillBuffer(b []byte, str string) (int, error) {
}
func (c *dnsConn) childContext() (context.Context, context.CancelFunc) {
c.mu.Lock()
defer c.mu.Unlock()
c.Lock()
defer c.Unlock()
if c.ctx == nil {
c.ctx, c.cancel = context.WithCancel(context.Background())
}

View File

@@ -52,7 +52,7 @@ type clientTracer struct {
tr tracer.Tracer
activeHooks map[string]context.Context
root tracer.Span
mu sync.Mutex
mtx sync.Mutex
}
func NewClientTrace(ctx context.Context, tr tracer.Tracer) *httptrace.ClientTrace {
@@ -83,8 +83,8 @@ func NewClientTrace(ctx context.Context, tr tracer.Tracer) *httptrace.ClientTrac
}
func (ct *clientTracer) start(hook, spanName string, attrs ...interface{}) {
ct.mu.Lock()
defer ct.mu.Unlock()
ct.mtx.Lock()
defer ct.mtx.Unlock()
if hookCtx, found := ct.activeHooks[hook]; !found {
var sp tracer.Span
@@ -104,8 +104,8 @@ func (ct *clientTracer) start(hook, spanName string, attrs ...interface{}) {
}
func (ct *clientTracer) end(hook string, err error, attrs ...interface{}) {
ct.mu.Lock()
defer ct.mu.Unlock()
ct.mtx.Lock()
defer ct.mtx.Unlock()
if ctx, ok := ct.activeHooks[hook]; ok { // nolint:nestif
if span, ok := tracer.SpanFromContext(ctx); ok {
if err != nil {
@@ -136,8 +136,8 @@ func (ct *clientTracer) getParentContext(hook string) context.Context {
}
func (ct *clientTracer) span(hook string) (tracer.Span, bool) {
ct.mu.Lock()
defer ct.mu.Unlock()
ct.mtx.Lock()
defer ct.mtx.Unlock()
if ctx, ok := ct.activeHooks[hook]; ok {
return tracer.SpanFromContext(ctx)
}

View File

@@ -2,8 +2,12 @@ package id
import (
"crypto/rand"
"encoding/binary"
"errors"
"fmt"
"time"
uuidv8 "github.com/ash3in/uuidv8"
"github.com/google/uuid"
nanoid "github.com/matoous/go-nanoid"
)
@@ -21,7 +25,6 @@ type Type int
const (
TypeUnspecified Type = iota
TypeNanoid
TypeUUIDv7
TypeUUIDv8
)
@@ -55,14 +58,14 @@ func (g *Generator) New() (string, error) {
}
return nanoid.Generate(g.opts.NanoidAlphabet, g.opts.NanoidSize)
case TypeUUIDv7:
uid, err := uuid.NewV7()
if err != nil {
return "", err
}
return uid.String(), nil
case TypeUUIDv8:
return "", errors.New("unsupported uuid version v8")
timestamp := uint64(time.Now().UnixNano())
clockSeq := make([]byte, 2)
if _, err := rand.Read(clockSeq); err != nil {
return "", fmt.Errorf("failed to generate random clock sequence: %w", err)
}
clockSeqValue := binary.BigEndian.Uint16(clockSeq) & 0x0FFF // Mask to 12 bits
return uuidv8.NewWithParams(timestamp, clockSeqValue, g.opts.UUIDNode[:], uuidv8.TimestampBits48)
}
return "", errors.New("invalid option, Type unspecified")
}
@@ -79,15 +82,16 @@ func New(opts ...Option) (string, error) {
if options.NanoidSize <= 0 {
return "", errors.New("invalid option, NanoidSize must be positive integer")
}
return nanoid.Generate(options.NanoidAlphabet, options.NanoidSize)
case TypeUUIDv7:
uid, err := uuid.NewV7()
if err != nil {
return "", err
}
return uid.String(), nil
case TypeUUIDv8:
return "", errors.New("unsupported uuid version v8")
timestamp := uint64(time.Now().UnixNano())
clockSeq := make([]byte, 2)
if _, err := rand.Read(clockSeq); err != nil {
return "", fmt.Errorf("failed to generate random clock sequence: %w", err)
}
clockSeqValue := binary.BigEndian.Uint16(clockSeq) & 0x0FFF // Mask to 12 bits
return uuidv8.NewWithParams(timestamp, clockSeqValue, options.UUIDNode[:], uuidv8.TimestampBits48)
}
return "", errors.New("invalid option, Type unspecified")
@@ -141,7 +145,7 @@ func WithUUIDNode(node [6]byte) Option {
// NewOptions returns new Options struct filled by opts
func NewOptions(opts ...Option) Options {
options := Options{
Type: TypeUUIDv7,
Type: TypeUUIDv8,
NanoidAlphabet: DefaultNanoidAlphabet,
NanoidSize: DefaultNanoidSize,
UUIDNode: generatedNode,

View File

@@ -489,74 +489,35 @@ func URLMap(query string) (map[string]interface{}, error) {
return mp.(map[string]interface{}), nil
}
// FlattenMap flattens a nested map into a single-level map using dot notation for nested keys.
// In case of key conflicts, all nested levels will be discarded in favor of the first-level key.
//
// Example #1:
//
// Input:
// {
// "user.name": "alex",
// "user.document.id": "document_id"
// "user.document.number": "document_number"
// }
// Output:
// {
// "user": {
// "name": "alex",
// "document": {
// "id": "document_id"
// "number": "document_number"
// }
// }
// }
//
// Example #2 (with conflicts):
//
// Input:
// {
// "user": "alex",
// "user.document.id": "document_id"
// "user.document.number": "document_number"
// }
// Output:
// {
// "user": "alex"
// }
func FlattenMap(input map[string]interface{}) map[string]interface{} {
result := make(map[string]interface{})
for k, v := range input {
parts := strings.Split(k, ".")
if len(parts) == 1 {
result[k] = v
// FlattenMap expand key.subkey to nested map
func FlattenMap(a map[string]interface{}) map[string]interface{} {
// preprocess map
nb := make(map[string]interface{}, len(a))
for k, v := range a {
ps := strings.Split(k, ".")
if len(ps) == 1 {
nb[k] = v
continue
}
current := result
for i, part := range parts {
// last element in the path
if i == len(parts)-1 {
current[part] = v
break
}
// initialize map for current level if not exist
if _, ok := current[part]; !ok {
current[part] = make(map[string]interface{})
}
if nested, ok := current[part].(map[string]interface{}); ok {
current = nested // continue to the nested map
} else {
break // if current element is not a map, ignore it
em := make(map[string]interface{})
em[ps[len(ps)-1]] = v
for i := len(ps) - 2; i > 0; i-- {
nm := make(map[string]interface{})
nm[ps[i]] = em
em = nm
}
if vm, ok := nb[ps[0]]; ok {
// nested map
nm := vm.(map[string]interface{})
for vk, vv := range em {
nm[vk] = vv
}
nb[ps[0]] = nm
} else {
nb[ps[0]] = em
}
}
return result
return nb
}
/*

View File

@@ -6,7 +6,6 @@ import (
"testing"
"time"
"github.com/stretchr/testify/require"
rutil "go.unistack.org/micro/v4/util/reflect"
)
@@ -320,140 +319,3 @@ func TestIsZero(t *testing.T) {
// t.Logf("XX %#+v\n", ok)
}
func TestFlattenMap(t *testing.T) {
tests := []struct {
name string
input map[string]interface{}
expected map[string]interface{}
}{
{
name: "empty map",
input: map[string]interface{}{},
expected: map[string]interface{}{},
},
{
name: "nil map",
input: nil,
expected: map[string]interface{}{},
},
{
name: "single level",
input: map[string]interface{}{
"username": "username",
"password": "password",
},
expected: map[string]interface{}{
"username": "username",
"password": "password",
},
},
{
name: "two level",
input: map[string]interface{}{
"order_id": "order_id",
"user.name": "username",
"user.password": "password",
},
expected: map[string]interface{}{
"order_id": "order_id",
"user": map[string]interface{}{
"name": "username",
"password": "password",
},
},
},
{
name: "three level",
input: map[string]interface{}{
"order_id": "order_id",
"user.name": "username",
"user.password": "password",
"user.document.id": "document_id",
"user.document.number": "document_number",
},
expected: map[string]interface{}{
"order_id": "order_id",
"user": map[string]interface{}{
"name": "username",
"password": "password",
"document": map[string]interface{}{
"id": "document_id",
"number": "document_number",
},
},
},
},
{
name: "four level",
input: map[string]interface{}{
"order_id": "order_id",
"user.name": "username",
"user.password": "password",
"user.document.id": "document_id",
"user.document.number": "document_number",
"user.info.permissions.read": "available",
"user.info.permissions.write": "available",
},
expected: map[string]interface{}{
"order_id": "order_id",
"user": map[string]interface{}{
"name": "username",
"password": "password",
"document": map[string]interface{}{
"id": "document_id",
"number": "document_number",
},
"info": map[string]interface{}{
"permissions": map[string]interface{}{
"read": "available",
"write": "available",
},
},
},
},
},
{
name: "key conflicts",
input: map[string]interface{}{
"user": "user",
"user.name": "username",
"user.password": "password",
},
expected: map[string]interface{}{
"user": "user",
},
},
{
name: "overwriting conflicts",
input: map[string]interface{}{
"order_id": "order_id",
"user.document.id": "document_id",
"user.document.number": "document_number",
"user.info.address": "address",
"user.info.phone": "phone",
},
expected: map[string]interface{}{
"order_id": "order_id",
"user": map[string]interface{}{
"document": map[string]interface{}{
"id": "document_id",
"number": "document_number",
},
"info": map[string]interface{}{
"address": "address",
"phone": "phone",
},
},
},
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
for range 100 { // need to exclude the impact of key order in the map on the test.
require.Equal(t, tt.expected, rutil.FlattenMap(tt.input))
}
})
}
}

View File

@@ -14,7 +14,7 @@ type Buffer struct {
vals []*Entry
size int
mu sync.RWMutex
sync.RWMutex
}
// Entry is ring buffer data entry
@@ -35,8 +35,8 @@ type Stream struct {
// Put adds a new value to ring buffer
func (b *Buffer) Put(v interface{}) {
b.mu.Lock()
defer b.mu.Unlock()
b.Lock()
defer b.Unlock()
// append to values
entry := &Entry{
@@ -63,8 +63,8 @@ func (b *Buffer) Put(v interface{}) {
// Get returns the last n entries
func (b *Buffer) Get(n int) []*Entry {
b.mu.RLock()
defer b.mu.RUnlock()
b.RLock()
defer b.RUnlock()
// reset any invalid values
if n > len(b.vals) || n < 0 {
@@ -80,8 +80,8 @@ func (b *Buffer) Get(n int) []*Entry {
// Since returns the entries since a specific time
func (b *Buffer) Since(t time.Time) []*Entry {
b.mu.RLock()
defer b.mu.RUnlock()
b.RLock()
defer b.RUnlock()
// return all the values
if t.IsZero() {
@@ -109,8 +109,8 @@ func (b *Buffer) Since(t time.Time) []*Entry {
// Stream logs from the buffer
// Close the channel when you want to stop
func (b *Buffer) Stream() (<-chan *Entry, chan bool) {
b.mu.Lock()
defer b.mu.Unlock()
b.Lock()
defer b.Unlock()
entries := make(chan *Entry, 128)
id := id.MustNew()

View File

@@ -24,7 +24,7 @@ type stream struct {
err error
request *request
mu sync.RWMutex
sync.RWMutex
}
type request struct {
@@ -57,9 +57,9 @@ func (s *stream) Request() server.Request {
func (s *stream) Send(v interface{}) error {
err := s.Stream.SendMsg(v)
if err != nil {
s.mu.Lock()
s.Lock()
s.err = err
s.mu.Unlock()
s.Unlock()
}
return err
}
@@ -68,17 +68,17 @@ func (s *stream) Send(v interface{}) error {
func (s *stream) Recv(v interface{}) error {
err := s.Stream.RecvMsg(v)
if err != nil {
s.mu.Lock()
s.Lock()
s.err = err
s.mu.Unlock()
s.Unlock()
}
return err
}
// Error returns error that stream holds
func (s *stream) Error() error {
s.mu.RLock()
defer s.mu.RUnlock()
s.RLock()
defer s.RUnlock()
return s.err
}

View File

@@ -6,7 +6,7 @@ import (
"strconv"
"time"
"github.com/goccy/go-yaml"
"gopkg.in/yaml.v3"
)
type Duration int64
@@ -58,9 +58,9 @@ func (d Duration) MarshalYAML() (interface{}, error) {
return time.Duration(d).String(), nil
}
func (d *Duration) UnmarshalYAML(data []byte) error {
func (d *Duration) UnmarshalYAML(n *yaml.Node) error {
var v interface{}
if err := yaml.Unmarshal(data, &v); err != nil {
if err := yaml.Unmarshal([]byte(n.Value), &v); err != nil {
return err
}
switch value := v.(type) {

View File

@@ -6,7 +6,7 @@ import (
"testing"
"time"
"github.com/goccy/go-yaml"
"gopkg.in/yaml.v3"
)
func TestMarshalYAML(t *testing.T) {