Compare commits

..

19 Commits

Author SHA1 Message Date
23e1174f25 Merge pull request 'tracer: improve' (#241) from tracing into v3
Reviewed-on: #241
2023-09-08 13:40:51 +03:00
52bed214cf tracer: improve
Some checks failed
lint / lint (pull_request) Failing after 1m31s
pr / test (pull_request) Failing after 2m44s
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2023-09-08 13:40:01 +03:00
64c4f5f47e Merge pull request 'tracer: tweaks for span tags and naming' (#239) from tracing into v3
Reviewed-on: #239
2023-09-01 14:58:15 +03:00
036c612137 tracer: tweaks for span tags and naming
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2023-09-01 14:58:15 +03:00
ca80e3ecf2 Merge pull request 'tracer: improve tracing info' (#238) from tracing into v3
Reviewed-on: #238
2023-09-01 08:41:46 +03:00
18e7bb41ca tracer: improve tracing info
Some checks failed
lint / lint (pull_request) Failing after 1m29s
pr / test (pull_request) Failing after 2m37s
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2023-09-01 08:41:23 +03:00
8e72fb1c35 Merge pull request 'add util/test' (#235) from util-test into v3
Reviewed-on: #235
2023-08-07 18:35:31 +03:00
17f21a03f4 add util/test
Some checks failed
lint / lint (pull_request) Failing after 1m28s
pr / test (pull_request) Failing after 2m35s
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2023-08-07 18:33:23 +03:00
a076d43a26 add util/test
Some checks failed
lint / lint (pull_request) Failing after 1m31s
pr / test (pull_request) Failing after 2m33s
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2023-08-07 18:32:29 +03:00
de6efaee0b Merge pull request 'config/default: add micro:generate uuid/id' (#232) from config-default-gen into v3
Reviewed-on: #232
2023-07-13 20:27:13 +03:00
9e0e657003 config/default: add micro:generate uuid/id
Some checks failed
lint / lint (pull_request) Failing after 1m28s
pr / test (pull_request) Failing after 2m35s
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2023-07-13 20:26:47 +03:00
be5f9ab77f Merge pull request 'tracer: add Flush method' (#225) from traceimp into v3
Reviewed-on: #225
2023-07-04 00:26:33 +03:00
144dca0cae tracer: add Flush method
Some checks failed
pr / test (pull_request) Failing after 2m42s
lint / lint (pull_request) Failing after 1m29s
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2023-07-04 00:25:41 +03:00
75173560e3 Merge pull request 'util/time: ParseDuration fix' (#222) from timefix into v3
Reviewed-on: #222
2023-05-29 14:04:41 +03:00
9b3bccd1f1 util/time: ParseDuration fix
All checks were successful
lint / lint (pull_request) Successful in 1m0s
pr / test (pull_request) Successful in 58s
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2023-05-29 14:02:06 +03:00
ce125b77c1 Merge pull request 'util/time: fix duration parsing' (#219) from timefeature into v3
Reviewed-on: #219
2023-05-27 23:55:51 +03:00
2ee8d4ed46 util/time: fix duration parsing
Some checks failed
lint / lint (pull_request) Successful in 59s
pr / test (pull_request) Failing after 1m0s
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2023-05-27 23:55:08 +03:00
f58781d076 Merge pull request 'server/noop: fix graceful unsubscribe' (#218) from unsubfix into v3
Reviewed-on: #218
2023-05-25 23:19:26 +03:00
e1af4aa3a4 server/noop: fix graceful unsubscribe
All checks were successful
pr / test (pull_request) Successful in 1m2s
lint / lint (pull_request) Successful in 59s
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2023-05-25 23:18:47 +03:00
136 changed files with 1670 additions and 812 deletions

View File

@@ -1,4 +1,4 @@
# Micro [![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://github.com/unistack-org/micro/workflows/build/badge.svg?branch=master)](https://github.com/unistack-org/micro/actions?query=workflow%3Abuild+branch%3Amaster+event%3Apush) [![Lint](https://goreportcard.com/badge/go.unistack.org/micro/v4)](https://goreportcard.com/report/go.unistack.org/micro/v4) [![Coverage](https://codecov.io/gh/unistack-org/micro/branch/v4/graph/badge.svg?token=OZPO2LP7VS)](https://codecov.io/gh/unistack-org/micro)
# Micro [![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/github.com/unistack-org/micro/v3?tab=overview) [![Status](https://github.com/unistack-org/micro/workflows/build/badge.svg?branch=master)](https://github.com/unistack-org/micro/actions?query=workflow%3Abuild+branch%3Amaster+event%3Apush) [![Lint](https://goreportcard.com/badge/go.unistack.org/micro/v3)](https://goreportcard.com/report/go.unistack.org/micro/v3) [![Coverage](https://codecov.io/gh/unistack-org/micro/branch/v3/graph/badge.svg?token=OZPO2LP7VS)](https://codecov.io/gh/unistack-org/micro)
Micro is a standard library for microservices.

View File

@@ -1,12 +1,12 @@
// Package broker is an interface used for asynchronous messaging
package broker // import "go.unistack.org/micro/v4/broker"
package broker // import "go.unistack.org/micro/v3/broker"
import (
"context"
"errors"
"go.unistack.org/micro/v4/codec"
"go.unistack.org/micro/v4/metadata"
"go.unistack.org/micro/v3/codec"
"go.unistack.org/micro/v3/metadata"
)
// DefaultBroker default memory broker

View File

@@ -3,14 +3,13 @@ package broker
import (
"context"
"sync"
"time"
"go.unistack.org/micro/v4/logger"
"go.unistack.org/micro/v4/metadata"
maddr "go.unistack.org/micro/v4/util/addr"
"go.unistack.org/micro/v4/util/id"
mnet "go.unistack.org/micro/v4/util/net"
"go.unistack.org/micro/v4/util/rand"
"go.unistack.org/micro/v3/logger"
"go.unistack.org/micro/v3/metadata"
maddr "go.unistack.org/micro/v3/util/addr"
"go.unistack.org/micro/v3/util/id"
mnet "go.unistack.org/micro/v3/util/net"
"go.unistack.org/micro/v3/util/rand"
)
type memoryBroker struct {
@@ -135,22 +134,13 @@ func (m *memoryBroker) publish(ctx context.Context, msgs []*Message, opts ...Pub
eh := m.opts.ErrorHandler
for t, ms := range msgTopicMap {
ts := time.Now()
m.opts.Meter.Counter(PublishMessageInflight, "endpoint", t).Add(len(ms))
m.opts.Meter.Counter(SubscribeMessageInflight, "endpoint", t).Add(len(ms))
m.RLock()
subs, ok := m.subscribers[t]
m.RUnlock()
if !ok {
m.opts.Meter.Counter(PublishMessageTotal, "endpoint", t, "status", "failure").Add(len(ms))
m.opts.Meter.Counter(PublishMessageInflight, "endpoint", t).Add(-len(ms))
m.opts.Meter.Counter(SubscribeMessageInflight, "endpoint", t).Add(-len(ms))
continue
}
m.opts.Meter.Counter(PublishMessageTotal, "endpoint", t, "status", "success").Add(len(ms))
for _, sub := range subs {
if sub.opts.BatchErrorHandler != nil {
beh = sub.opts.BatchErrorHandler
@@ -162,65 +152,37 @@ func (m *memoryBroker) publish(ctx context.Context, msgs []*Message, opts ...Pub
switch {
// batch processing
case sub.batchhandler != nil:
if err = sub.batchhandler(ms); err != nil {
m.opts.Meter.Counter(SubscribeMessageTotal, "endpoint", t, "status", "failure").Inc()
ms.SetError(err)
if beh != nil {
_ = beh(ms)
} else if m.opts.Logger.V(logger.ErrorLevel) {
m.opts.Logger.Error(m.opts.Context, err.Error())
}
} else {
if sub.opts.AutoAck {
if err = ms.Ack(); err != nil {
m.opts.Logger.Errorf(m.opts.Context, "ack failed: %v", err)
m.opts.Meter.Counter(SubscribeMessageTotal, "endpoint", t, "status", "failure").Inc()
} else {
m.opts.Meter.Counter(SubscribeMessageTotal, "endpoint", t, "status", "success").Inc()
}
} else {
m.opts.Meter.Counter(SubscribeMessageTotal, "endpoint", t, "status", "success").Inc()
} else if sub.opts.AutoAck {
if err = ms.Ack(); err != nil {
m.opts.Logger.Errorf(m.opts.Context, "ack failed: %v", err)
}
}
m.opts.Meter.Counter(PublishMessageInflight, "endpoint", t).Add(-len(ms))
m.opts.Meter.Counter(SubscribeMessageInflight, "endpoint", t).Add(-len(ms))
// single processing
case sub.handler != nil:
for _, p := range ms {
if err = sub.handler(p); err != nil {
m.opts.Meter.Counter(SubscribeMessageTotal, "endpoint", t, "status", "failure").Inc()
p.SetError(err)
if eh != nil {
_ = eh(p)
} else if m.opts.Logger.V(logger.ErrorLevel) {
m.opts.Logger.Error(m.opts.Context, err.Error())
}
} else {
if sub.opts.AutoAck {
if err = p.Ack(); err != nil {
m.opts.Logger.Errorf(m.opts.Context, "ack failed: %v", err)
m.opts.Meter.Counter(SubscribeMessageTotal, "endpoint", t, "status", "failure").Inc()
} else {
m.opts.Meter.Counter(SubscribeMessageTotal, "endpoint", t, "status", "success").Inc()
}
} else {
m.opts.Meter.Counter(SubscribeMessageTotal, "endpoint", t, "status", "success").Inc()
} else if sub.opts.AutoAck {
if err = p.Ack(); err != nil {
m.opts.Logger.Errorf(m.opts.Context, "ack failed: %v", err)
}
}
m.opts.Meter.Counter(PublishMessageInflight, "endpoint", t).Add(-1)
m.opts.Meter.Counter(SubscribeMessageInflight, "endpoint", t).Add(-1)
}
}
}
te := time.Since(ts)
m.opts.Meter.Summary(PublishMessageLatencyMicroseconds, "endpoint", t).Update(te.Seconds())
m.opts.Meter.Histogram(PublishMessageDurationSeconds, "endpoint", t).Update(te.Seconds())
m.opts.Meter.Summary(SubscribeMessageLatencyMicroseconds, "endpoint", t).Update(te.Seconds())
m.opts.Meter.Histogram(SubscribeMessageDurationSeconds, "endpoint", t).Update(te.Seconds())
}
}
return nil

View File

@@ -5,7 +5,7 @@ import (
"fmt"
"testing"
"go.unistack.org/micro/v4/metadata"
"go.unistack.org/micro/v3/metadata"
)
func TestMemoryBatchBroker(t *testing.T) {

View File

@@ -5,30 +5,11 @@ import (
"crypto/tls"
"time"
"go.unistack.org/micro/v4/codec"
"go.unistack.org/micro/v4/logger"
"go.unistack.org/micro/v4/meter"
"go.unistack.org/micro/v4/register"
"go.unistack.org/micro/v4/tracer"
)
var (
// PublishMessageDurationSeconds specifies meter metric name
PublishMessageDurationSeconds = "publish_message_duration_seconds"
// PublishMessageLatencyMicroseconds specifies meter metric name
PublishMessageLatencyMicroseconds = "publish_message_latency_microseconds"
// PublishMessageTotal specifies meter metric name
PublishMessageTotal = "publish_message_total"
// PublishMessageInflight specifies meter metric name
PublishMessageInflight = "publish_message_inflight"
// SubscribeMessageDurationSeconds specifies meter metric name
SubscribeMessageDurationSeconds = "subscribe_message_duration_seconds"
// SubscribeMessageLatencyMicroseconds specifies meter metric name
SubscribeMessageLatencyMicroseconds = "subscribe_message_latency_microseconds"
// SubscribeMessageTotal specifies meter metric name
SubscribeMessageTotal = "subscribe_message_total"
// SubscribeMessageInflight specifies meter metric name
SubscribeMessageInflight = "subscribe_message_inflight"
"go.unistack.org/micro/v3/codec"
"go.unistack.org/micro/v3/logger"
"go.unistack.org/micro/v3/meter"
"go.unistack.org/micro/v3/register"
"go.unistack.org/micro/v3/tracer"
)
// Options struct

View File

@@ -5,7 +5,7 @@ import (
"math"
"time"
"go.unistack.org/micro/v4/util/backoff"
"go.unistack.org/micro/v3/util/backoff"
)
// BackoffFunc is the backoff call func

View File

@@ -1,12 +1,12 @@
// Package client is an interface for an RPC client
package client // import "go.unistack.org/micro/v4/client"
package client // import "go.unistack.org/micro/v3/client"
import (
"context"
"time"
"go.unistack.org/micro/v4/codec"
"go.unistack.org/micro/v4/metadata"
"go.unistack.org/micro/v3/codec"
"go.unistack.org/micro/v3/metadata"
)
var (

View File

@@ -4,8 +4,8 @@ import (
"context"
"sort"
"go.unistack.org/micro/v4/errors"
"go.unistack.org/micro/v4/router"
"go.unistack.org/micro/v3/errors"
"go.unistack.org/micro/v3/router"
)
// LookupFunc is used to lookup routes for a service

View File

@@ -5,11 +5,11 @@ import (
"fmt"
"time"
"go.unistack.org/micro/v4/broker"
"go.unistack.org/micro/v4/codec"
"go.unistack.org/micro/v4/errors"
"go.unistack.org/micro/v4/metadata"
"go.unistack.org/micro/v4/selector"
"go.unistack.org/micro/v3/broker"
"go.unistack.org/micro/v3/codec"
"go.unistack.org/micro/v3/errors"
"go.unistack.org/micro/v3/metadata"
"go.unistack.org/micro/v3/selector"
)
// DefaultCodecs will be used to encode/decode data

View File

@@ -6,17 +6,17 @@ import (
"net"
"time"
"go.unistack.org/micro/v4/broker"
"go.unistack.org/micro/v4/codec"
"go.unistack.org/micro/v4/logger"
"go.unistack.org/micro/v4/metadata"
"go.unistack.org/micro/v4/meter"
"go.unistack.org/micro/v4/network/transport"
"go.unistack.org/micro/v4/register"
"go.unistack.org/micro/v4/router"
"go.unistack.org/micro/v4/selector"
"go.unistack.org/micro/v4/selector/random"
"go.unistack.org/micro/v4/tracer"
"go.unistack.org/micro/v3/broker"
"go.unistack.org/micro/v3/codec"
"go.unistack.org/micro/v3/logger"
"go.unistack.org/micro/v3/metadata"
"go.unistack.org/micro/v3/meter"
"go.unistack.org/micro/v3/network/transport"
"go.unistack.org/micro/v3/register"
"go.unistack.org/micro/v3/router"
"go.unistack.org/micro/v3/selector"
"go.unistack.org/micro/v3/selector/random"
"go.unistack.org/micro/v3/tracer"
)
// Options holds client options

View File

@@ -3,7 +3,7 @@ package client
import (
"context"
"go.unistack.org/micro/v4/errors"
"go.unistack.org/micro/v3/errors"
)
// RetryFunc that returning either false or a non-nil error will result in the call not being retried

View File

@@ -5,7 +5,7 @@ import (
"fmt"
"testing"
"go.unistack.org/micro/v4/errors"
"go.unistack.org/micro/v3/errors"
)
func TestRetryAlways(t *testing.T) {

View File

@@ -1,7 +1,7 @@
package client
import (
"go.unistack.org/micro/v4/codec"
"go.unistack.org/micro/v3/codec"
)
type testRequest struct {

View File

@@ -1,11 +1,11 @@
// Package codec is an interface for encoding messages
package codec // import "go.unistack.org/micro/v4/codec"
package codec // import "go.unistack.org/micro/v3/codec"
import (
"errors"
"io"
"go.unistack.org/micro/v4/metadata"
"go.unistack.org/micro/v3/metadata"
)
// Message types

View File

@@ -1,4 +1,4 @@
// Copyright 2021-2023 Unistack LLC
// Copyright 2021 Unistack LLC
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
@@ -17,7 +17,7 @@ syntax = "proto3";
package micro.codec;
option cc_enable_arenas = true;
option go_package = "go.unistack.org/micro/v4/codec;codec";
option go_package = "go.unistack.org/micro/v3/codec;codec";
option java_multiple_files = true;
option java_outer_classname = "MicroCodec";
option java_package = "micro.codec";

View File

@@ -3,9 +3,9 @@ package codec
import (
"context"
"go.unistack.org/micro/v4/logger"
"go.unistack.org/micro/v4/meter"
"go.unistack.org/micro/v4/tracer"
"go.unistack.org/micro/v3/logger"
"go.unistack.org/micro/v3/meter"
"go.unistack.org/micro/v3/tracer"
)
// Option func

View File

@@ -1,5 +1,5 @@
// Package config is an interface for dynamic configuration.
package config // import "go.unistack.org/micro/v4/config"
package config // import "go.unistack.org/micro/v3/config"
import (
"context"

View File

@@ -7,9 +7,11 @@ import (
"strings"
"time"
"github.com/google/uuid"
"github.com/imdario/mergo"
rutil "go.unistack.org/micro/v4/util/reflect"
mtime "go.unistack.org/micro/v4/util/time"
mid "go.unistack.org/micro/v3/util/id"
rutil "go.unistack.org/micro/v3/util/reflect"
mtime "go.unistack.org/micro/v3/util/time"
)
type defaultConfig struct {
@@ -124,6 +126,20 @@ func fillValue(value reflect.Value, val string) error {
}
value.Set(reflect.ValueOf(v))
case reflect.String:
switch val {
case "micro:generate uuid":
uid, err := uuid.NewRandom()
if err != nil {
return err
}
val = uid.String()
case "micro:generate id":
uid, err := mid.New()
if err != nil {
return err
}
val = uid
}
value.Set(reflect.ValueOf(val))
case reflect.Float32:
v, err := strconv.ParseFloat(val, 32)
@@ -169,7 +185,7 @@ func fillValue(value reflect.Value, val string) error {
return err
}
value.Set(reflect.ValueOf(v))
case value.Type().String() == "time.Duration" && value.Type().PkgPath() == "go.unistack.org/micro/v4/util/time":
case value.Type().String() == "time.Duration" && value.Type().PkgPath() == "go.unistack.org/micro/v3/util/time":
v, err := mtime.ParseDuration(val)
if err != nil {
return err

View File

@@ -6,17 +6,21 @@ import (
"testing"
"time"
"go.unistack.org/micro/v4/config"
mtime "go.unistack.org/micro/v4/util/time"
"go.unistack.org/micro/v3/config"
mid "go.unistack.org/micro/v3/util/id"
mtime "go.unistack.org/micro/v3/util/time"
)
type cfg struct {
StringValue string `default:"string_value"`
IgnoreValue string `json:"-"`
StructValue *cfgStructValue
IntValue int `default:"99"`
DurationValue time.Duration `default:"10s"`
MDurationValue mtime.Duration `default:"10s"`
IntValue int `default:"99"`
DurationValue time.Duration `default:"10s"`
MDurationValue mtime.Duration `default:"10s"`
MapValue map[string]bool `default:"key1=true,key2=false"`
UUIDValue string `default:"micro:generate uuid"`
IDValue string `default:"micro:generate id"`
}
type cfgStructValue struct {
@@ -67,6 +71,21 @@ func TestDefault(t *testing.T) {
if conf.StringValue != "after_load" {
t.Fatal("AfterLoad option not working")
}
if len(conf.MapValue) != 2 {
t.Fatalf("map value invalid: %#+v\n", conf.MapValue)
}
if conf.UUIDValue == "" {
t.Fatalf("uuid value empty")
} else if len(conf.UUIDValue) != 36 {
t.Fatalf("uuid value invalid: %s", conf.UUIDValue)
}
if conf.IDValue == "" {
t.Fatalf("id value empty")
} else if len(conf.IDValue) != mid.DefaultSize {
t.Fatalf("id value invalid: %s", conf.IDValue)
}
_ = conf
// t.Logf("%#+v\n", conf)
}

View File

@@ -4,10 +4,10 @@ import (
"context"
"time"
"go.unistack.org/micro/v4/codec"
"go.unistack.org/micro/v4/logger"
"go.unistack.org/micro/v4/meter"
"go.unistack.org/micro/v4/tracer"
"go.unistack.org/micro/v3/codec"
"go.unistack.org/micro/v3/logger"
"go.unistack.org/micro/v3/meter"
"go.unistack.org/micro/v3/tracer"
)
// Options hold the config options

View File

@@ -1,6 +1,6 @@
// Package errors provides a way to return detailed information
// for an RPC request error. The error is normally JSON encoded.
package errors // import "go.unistack.org/micro/v4/errors"
package errors // import "go.unistack.org/micro/v3/errors"
import (
"bytes"

View File

@@ -1,4 +1,4 @@
// Copyright 2021-2023 Unistack LLC
// Copyright 2021 Unistack LLC
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
@@ -17,7 +17,7 @@ syntax = "proto3";
package micro.errors;
option cc_enable_arenas = true;
option go_package = "go.unistack.org/micro/v4/errors;errors";
option go_package = "go.unistack.org/micro/v3/errors;errors";
option java_multiple_files = true;
option java_outer_classname = "MicroErrors";
option java_package = "micro.errors";

View File

@@ -3,7 +3,7 @@ package micro
import (
"context"
"go.unistack.org/micro/v4/client"
"go.unistack.org/micro/v3/client"
)
// Event is used to publish messages to a topic

View File

@@ -6,12 +6,12 @@ import (
"sync"
"github.com/silas/dag"
"go.unistack.org/micro/v4/client"
"go.unistack.org/micro/v4/codec"
"go.unistack.org/micro/v4/logger"
"go.unistack.org/micro/v4/metadata"
"go.unistack.org/micro/v4/store"
"go.unistack.org/micro/v4/util/id"
"go.unistack.org/micro/v3/client"
"go.unistack.org/micro/v3/codec"
"go.unistack.org/micro/v3/logger"
"go.unistack.org/micro/v3/metadata"
"go.unistack.org/micro/v3/store"
"go.unistack.org/micro/v3/util/id"
)
type microFlow struct {

View File

@@ -1,5 +1,5 @@
// Package flow is an interface used for saga pattern microservice workflow
package flow // import "go.unistack.org/micro/v4/flow"
package flow // import "go.unistack.org/micro/v3/flow"
import (
"context"
@@ -7,7 +7,7 @@ import (
"sync"
"sync/atomic"
"go.unistack.org/micro/v4/metadata"
"go.unistack.org/micro/v3/metadata"
)
var (

View File

@@ -4,11 +4,11 @@ import (
"context"
"time"
"go.unistack.org/micro/v4/client"
"go.unistack.org/micro/v4/logger"
"go.unistack.org/micro/v4/meter"
"go.unistack.org/micro/v4/store"
"go.unistack.org/micro/v4/tracer"
"go.unistack.org/micro/v3/client"
"go.unistack.org/micro/v3/logger"
"go.unistack.org/micro/v3/meter"
"go.unistack.org/micro/v3/store"
"go.unistack.org/micro/v3/tracer"
)
// Option func

View File

@@ -1,4 +1,4 @@
package fsm // import "go.unistack.org/micro/v4/fsm"
package fsm // import "go.unistack.org/micro/v3/fsm"
import (
"context"

View File

@@ -5,7 +5,7 @@ import (
"fmt"
"testing"
"go.unistack.org/micro/v4/logger"
"go.unistack.org/micro/v3/logger"
)
func TestFSMStart(t *testing.T) {

18
go.mod
View File

@@ -1,10 +1,20 @@
module go.unistack.org/micro/v4
module go.unistack.org/micro/v3
go 1.20
go 1.19
require (
github.com/DATA-DOG/go-sqlmock v1.5.0
github.com/imdario/mergo v0.3.14
github.com/google/uuid v1.3.0
github.com/imdario/mergo v0.3.15
github.com/patrickmn/go-cache v2.1.0+incompatible
github.com/silas/dag v0.0.0-20211117232152-9d50aa809f35
github.com/silas/dag v0.0.0-20220518035006-a7e85ada93c5
golang.org/x/sync v0.3.0
google.golang.org/grpc v1.57.0
google.golang.org/protobuf v1.31.0
)
require (
github.com/golang/protobuf v1.5.3 // indirect
golang.org/x/net v0.14.0 // indirect
google.golang.org/genproto/googleapis/rpc v0.0.0-20230526203410-71b5a4ffd15e // indirect
)

30
go.sum
View File

@@ -1,11 +1,33 @@
github.com/DATA-DOG/go-sqlmock v1.5.0 h1:Shsta01QNfFxHCfpW6YH2STWB0MudeXXEWMr20OEh60=
github.com/DATA-DOG/go-sqlmock v1.5.0/go.mod h1:f/Ixk793poVmq4qj/V1dPUg2JEAKC73Q5eFN3EC/SaM=
github.com/imdario/mergo v0.3.14 h1:fOqeC1+nCuuk6PKQdg9YmosXX7Y7mHX6R/0ZldI9iHo=
github.com/imdario/mergo v0.3.14/go.mod h1:WBLT9ZmE3lPoWsEzCh9LPo3TiwVN+ZKEjmz+hD27ysY=
github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
github.com/golang/protobuf v1.5.3 h1:KhyjKVUg7Usr/dYsdSqoFveMYd5ko72D+zANwlG1mmg=
github.com/golang/protobuf v1.5.3/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY=
github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
github.com/google/go-cmp v0.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38=
github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I=
github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/imdario/mergo v0.3.15 h1:M8XP7IuFNsqUx6VPK2P9OSmsYsI/YFaGil0uD21V3dM=
github.com/imdario/mergo v0.3.15/go.mod h1:WBLT9ZmE3lPoWsEzCh9LPo3TiwVN+ZKEjmz+hD27ysY=
github.com/patrickmn/go-cache v2.1.0+incompatible h1:HRMgzkcYKYpi3C8ajMPV8OFXaaRUnok+kx1WdO15EQc=
github.com/patrickmn/go-cache v2.1.0+incompatible/go.mod h1:3Qf8kWWT7OJRJbdiICTKqZju1ZixQ/KpMGzzAfe6+WQ=
github.com/silas/dag v0.0.0-20211117232152-9d50aa809f35 h1:4mohWoM/UGg1BvFFiqSPRl5uwJY3rVV0HQX0ETqauqQ=
github.com/silas/dag v0.0.0-20211117232152-9d50aa809f35/go.mod h1:7RTUFBdIRC9nZ7/3RyRNH1bdqIShrDejd1YbLwgPS+I=
github.com/silas/dag v0.0.0-20220518035006-a7e85ada93c5 h1:G/FZtUu7a6NTWl3KUHMV9jkLAh/Rvtf03NWMHaEDl+E=
github.com/silas/dag v0.0.0-20220518035006-a7e85ada93c5/go.mod h1:7RTUFBdIRC9nZ7/3RyRNH1bdqIShrDejd1YbLwgPS+I=
golang.org/x/net v0.14.0 h1:BONx9s002vGdD9umnlX1Po8vOZmrgH34qlHcD1MfK14=
golang.org/x/net v0.14.0/go.mod h1:PpSgVXXLK0OxS0F31C1/tv6XNguvCrnXIDrFMspZIUI=
golang.org/x/sync v0.3.0 h1:ftCYgMx6zT/asHUrPw8BLLscYtGznsLAnjq5RH9P66E=
golang.org/x/sync v0.3.0/go.mod h1:FU7BRWz2tNW+3quACPkgCx/L+uEAv1htQ0V83Z9Rj+Y=
golang.org/x/sys v0.11.0 h1:eG7RXZHdqOJ1i+0lgLgCpSXAp6M3LYlAo6osgSi0xOM=
golang.org/x/text v0.12.0 h1:k+n5B8goJNdU7hSvEtMUz3d1Q6D/XW4COJSJR6fN0mc=
golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
google.golang.org/genproto/googleapis/rpc v0.0.0-20230526203410-71b5a4ffd15e h1:NumxXLPfHSndr3wBBdeKiVHjGVFzi9RX2HwwQke94iY=
google.golang.org/genproto/googleapis/rpc v0.0.0-20230526203410-71b5a4ffd15e/go.mod h1:66JfowdXAEgad5O9NnYcsNPLCPZJD++2L9X0PCMODrA=
google.golang.org/grpc v1.57.0 h1:kfzNeI/klCGD2YPMUlaGNT3pxvYfga7smW3Vth8Zsiw=
google.golang.org/grpc v1.57.0/go.mod h1:Sd+9RMTACXwmub0zcNY2c4arhtrbBYD1AUHI/dt16Mo=
google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
google.golang.org/protobuf v1.31.0 h1:g0LDEJHgrBl9N9r17Ru3sqWhkIx2NB67okBHPwC7hs8=
google.golang.org/protobuf v1.31.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

View File

@@ -1,5 +1,5 @@
// Package logger provides a log interface
package logger // import "go.unistack.org/micro/v4/logger"
package logger // import "go.unistack.org/micro/v3/logger"
import (
"context"

View File

@@ -8,7 +8,7 @@ import (
"strconv"
"strings"
"go.unistack.org/micro/v4/codec"
"go.unistack.org/micro/v3/codec"
)
const sf = "0-+# "

View File

@@ -5,7 +5,7 @@ import (
"strings"
"testing"
"go.unistack.org/micro/v4/codec"
"go.unistack.org/micro/v3/codec"
)
func TestUnwrap(t *testing.T) {

View File

@@ -5,9 +5,9 @@ import (
"context"
"fmt"
"go.unistack.org/micro/v4/client"
"go.unistack.org/micro/v4/logger"
"go.unistack.org/micro/v4/server"
"go.unistack.org/micro/v3/client"
"go.unistack.org/micro/v3/logger"
"go.unistack.org/micro/v3/server"
)
var (

View File

@@ -39,8 +39,6 @@ func FromOutgoingContext(ctx context.Context) (Metadata, bool) {
// FromContext returns metadata from the given context
// returned metadata shoud not be modified or race condition happens
//
// Deprecated: use FromIncomingContext or FromOutgoingContext
func FromContext(ctx context.Context) (Metadata, bool) {
if ctx == nil {
return nil, false
@@ -53,8 +51,6 @@ func FromContext(ctx context.Context) (Metadata, bool) {
}
// NewContext creates a new context with the given metadata
//
// Deprecated: use NewIncomingContext or NewOutgoingContext
func NewContext(ctx context.Context, md Metadata) context.Context {
if ctx == nil {
ctx = context.Background()

View File

@@ -1,5 +1,5 @@
// Package metadata is a way of defining message headers
package metadata // import "go.unistack.org/micro/v4/metadata"
package metadata // import "go.unistack.org/micro/v3/metadata"
import (
"net/textproto"
@@ -19,6 +19,8 @@ var (
HeaderTimeout = "Micro-Timeout"
// HeaderAuthorization specifies Authorization header
HeaderAuthorization = "Authorization"
// HeaderXRequestID specifies request id
HeaderXRequestID = "X-Request-Id"
)
// Metadata is our way of representing request headers internally.

View File

@@ -1,5 +1,5 @@
// Package meter is for instrumentation
package meter // import "go.unistack.org/micro/v4/meter"
package meter // import "go.unistack.org/micro/v3/meter"
import (
"io"

View File

@@ -3,7 +3,7 @@ package meter
import (
"context"
"go.unistack.org/micro/v4/logger"
"go.unistack.org/micro/v3/logger"
)
// Option powers the configuration for metrics implementations:

View File

@@ -1,13 +1,13 @@
package wrapper // import "go.unistack.org/micro/v4/meter/wrapper"
package wrapper // import "go.unistack.org/micro/v3/meter/wrapper"
import (
"context"
"fmt"
"time"
"go.unistack.org/micro/v4/client"
"go.unistack.org/micro/v4/meter"
"go.unistack.org/micro/v4/server"
"go.unistack.org/micro/v3/client"
"go.unistack.org/micro/v3/meter"
"go.unistack.org/micro/v3/server"
)
var (
@@ -27,6 +27,22 @@ var (
ServerRequestTotal = "server_request_total"
// ServerRequestInflight specifies meter metric name
ServerRequestInflight = "server_request_inflight"
// PublishMessageDurationSeconds specifies meter metric name
PublishMessageDurationSeconds = "publish_message_duration_seconds"
// PublishMessageLatencyMicroseconds specifies meter metric name
PublishMessageLatencyMicroseconds = "publish_message_latency_microseconds"
// PublishMessageTotal specifies meter metric name
PublishMessageTotal = "publish_message_total"
// PublishMessageInflight specifies meter metric name
PublishMessageInflight = "publish_message_inflight"
// SubscribeMessageDurationSeconds specifies meter metric name
SubscribeMessageDurationSeconds = "subscribe_message_duration_seconds"
// SubscribeMessageLatencyMicroseconds specifies meter metric name
SubscribeMessageLatencyMicroseconds = "subscribe_message_latency_microseconds"
// SubscribeMessageTotal specifies meter metric name
SubscribeMessageTotal = "subscribe_message_total"
// SubscribeMessageInflight specifies meter metric name
SubscribeMessageInflight = "subscribe_message_inflight"
labelSuccess = "success"
labelFailure = "failure"
@@ -214,7 +230,37 @@ func (w *wrapper) Stream(ctx context.Context, req client.Request, opts ...client
}
func (w *wrapper) Publish(ctx context.Context, p client.Message, opts ...client.PublishOption) error {
return w.Client.Publish(ctx, p, opts...)
endpoint := p.Topic()
labels := make([]string, 0, 4)
labels = append(labels, labelEndpoint, endpoint)
w.opts.Meter.Counter(PublishMessageInflight, labels...).Inc()
ts := time.Now()
err := w.Client.Publish(ctx, p, opts...)
te := time.Since(ts)
w.opts.Meter.Counter(PublishMessageInflight, labels...).Dec()
w.opts.Meter.Summary(PublishMessageLatencyMicroseconds, labels...).Update(te.Seconds())
w.opts.Meter.Histogram(PublishMessageDurationSeconds, labels...).Update(te.Seconds())
if err == nil {
labels = append(labels, labelStatus, labelSuccess)
} else {
labels = append(labels, labelStatus, labelFailure)
}
w.opts.Meter.Counter(PublishMessageTotal, labels...).Inc()
return err
}
// NewHandlerWrapper create new server handler wrapper
// deprecated
func NewHandlerWrapper(opts ...Option) server.HandlerWrapper {
handler := &wrapper{
opts: NewOptions(opts...),
}
return handler.HandlerFunc
}
// NewServerHandlerWrapper create new server handler wrapper
@@ -256,3 +302,46 @@ func (w *wrapper) HandlerFunc(fn server.HandlerFunc) server.HandlerFunc {
return err
}
}
// NewSubscriberWrapper create server subscribe wrapper
// deprecated
func NewSubscriberWrapper(opts ...Option) server.SubscriberWrapper {
handler := &wrapper{
opts: NewOptions(opts...),
}
return handler.SubscriberFunc
}
func NewServerSubscriberWrapper(opts ...Option) server.SubscriberWrapper {
handler := &wrapper{
opts: NewOptions(opts...),
}
return handler.SubscriberFunc
}
func (w *wrapper) SubscriberFunc(fn server.SubscriberFunc) server.SubscriberFunc {
return func(ctx context.Context, msg server.Message) error {
endpoint := msg.Topic()
labels := make([]string, 0, 4)
labels = append(labels, labelEndpoint, endpoint)
w.opts.Meter.Counter(SubscribeMessageInflight, labels...).Inc()
ts := time.Now()
err := fn(ctx, msg)
te := time.Since(ts)
w.opts.Meter.Counter(SubscribeMessageInflight, labels...).Dec()
w.opts.Meter.Summary(SubscribeMessageLatencyMicroseconds, labels...).Update(te.Seconds())
w.opts.Meter.Histogram(SubscribeMessageDurationSeconds, labels...).Update(te.Seconds())
if err == nil {
labels = append(labels, labelStatus, labelSuccess)
} else {
labels = append(labels, labelStatus, labelFailure)
}
w.opts.Meter.Counter(SubscribeMessageTotal, labels...).Inc()
return err
}
}

View File

@@ -1,4 +1,4 @@
package mtls // import "go.unistack.org/micro/v4/mtls"
package mtls // import "go.unistack.org/micro/v3/mtls"
import (
"bytes"

View File

@@ -1,9 +1,9 @@
// Package network is for creating internetworks
package network // import "go.unistack.org/micro/v4/network"
package network // import "go.unistack.org/micro/v3/network"
import (
"go.unistack.org/micro/v4/client"
"go.unistack.org/micro/v4/server"
"go.unistack.org/micro/v3/client"
"go.unistack.org/micro/v3/server"
)
// Error is network node errors

View File

@@ -1,13 +1,13 @@
package network
import (
"go.unistack.org/micro/v4/logger"
"go.unistack.org/micro/v4/meter"
"go.unistack.org/micro/v4/network/tunnel"
"go.unistack.org/micro/v4/proxy"
"go.unistack.org/micro/v4/router"
"go.unistack.org/micro/v4/tracer"
"go.unistack.org/micro/v4/util/id"
"go.unistack.org/micro/v3/logger"
"go.unistack.org/micro/v3/meter"
"go.unistack.org/micro/v3/network/tunnel"
"go.unistack.org/micro/v3/proxy"
"go.unistack.org/micro/v3/router"
"go.unistack.org/micro/v3/tracer"
"go.unistack.org/micro/v3/util/id"
)
// Option func

View File

@@ -8,9 +8,9 @@ import (
"sync"
"time"
maddr "go.unistack.org/micro/v4/util/addr"
mnet "go.unistack.org/micro/v4/util/net"
"go.unistack.org/micro/v4/util/rand"
maddr "go.unistack.org/micro/v3/util/addr"
mnet "go.unistack.org/micro/v3/util/net"
"go.unistack.org/micro/v3/util/rand"
)
type memorySocket struct {

View File

@@ -5,10 +5,10 @@ import (
"crypto/tls"
"time"
"go.unistack.org/micro/v4/codec"
"go.unistack.org/micro/v4/logger"
"go.unistack.org/micro/v4/meter"
"go.unistack.org/micro/v4/tracer"
"go.unistack.org/micro/v3/codec"
"go.unistack.org/micro/v3/logger"
"go.unistack.org/micro/v3/meter"
"go.unistack.org/micro/v3/tracer"
)
// Options struct holds the transport options

View File

@@ -1,11 +1,11 @@
// Package transport is an interface for synchronous connection based communication
package transport // import "go.unistack.org/micro/v4/network/transport"
package transport // import "go.unistack.org/micro/v3/network/transport"
import (
"context"
"time"
"go.unistack.org/micro/v4/metadata"
"go.unistack.org/micro/v3/metadata"
)
var (

View File

@@ -1,15 +1,15 @@
// Package broker is a tunnel broker
package broker // import "go.unistack.org/micro/v4/network/tunnel/broker"
package broker // import "go.unistack.org/micro/v3/network/tunnel/broker"
import (
"context"
"fmt"
"go.unistack.org/micro/v4/broker"
"go.unistack.org/micro/v4/logger"
"go.unistack.org/micro/v4/metadata"
"go.unistack.org/micro/v4/network/transport"
"go.unistack.org/micro/v4/network/tunnel"
"go.unistack.org/micro/v3/broker"
"go.unistack.org/micro/v3/logger"
"go.unistack.org/micro/v3/metadata"
"go.unistack.org/micro/v3/network/transport"
"go.unistack.org/micro/v3/network/tunnel"
)
type tunBroker struct {

View File

@@ -3,11 +3,11 @@ package tunnel
import (
"time"
"go.unistack.org/micro/v4/logger"
"go.unistack.org/micro/v4/meter"
"go.unistack.org/micro/v4/network/transport"
"go.unistack.org/micro/v4/tracer"
"go.unistack.org/micro/v4/util/id"
"go.unistack.org/micro/v3/logger"
"go.unistack.org/micro/v3/meter"
"go.unistack.org/micro/v3/network/transport"
"go.unistack.org/micro/v3/tracer"
"go.unistack.org/micro/v3/util/id"
)
var (

View File

@@ -1,8 +1,8 @@
package transport
import (
"go.unistack.org/micro/v4/network/transport"
"go.unistack.org/micro/v4/network/tunnel"
"go.unistack.org/micro/v3/network/transport"
"go.unistack.org/micro/v3/network/tunnel"
)
type tunListener struct {

View File

@@ -1,12 +1,12 @@
// Package transport provides a tunnel transport
package transport // import "go.unistack.org/micro/v4/network/tunnel/transport"
package transport // import "go.unistack.org/micro/v3/network/tunnel/transport"
import (
"context"
"fmt"
"go.unistack.org/micro/v4/network/transport"
"go.unistack.org/micro/v4/network/tunnel"
"go.unistack.org/micro/v3/network/transport"
"go.unistack.org/micro/v3/network/tunnel"
)
type tunTransport struct {

View File

@@ -1,12 +1,12 @@
// Package tunnel provides gre network tunnelling
package tunnel // import "go.unistack.org/micro/v4/network/transport/tunnel"
package tunnel // import "go.unistack.org/micro/v3/network/transport/tunnel"
import (
"context"
"errors"
"time"
"go.unistack.org/micro/v4/network/transport"
"go.unistack.org/micro/v3/network/transport"
)
// DefaultTunnel contains default tunnel implementation

View File

@@ -5,17 +5,17 @@ import (
"fmt"
"time"
"go.unistack.org/micro/v4/broker"
"go.unistack.org/micro/v4/client"
"go.unistack.org/micro/v4/config"
"go.unistack.org/micro/v4/logger"
"go.unistack.org/micro/v4/metadata"
"go.unistack.org/micro/v4/meter"
"go.unistack.org/micro/v4/register"
"go.unistack.org/micro/v4/router"
"go.unistack.org/micro/v4/server"
"go.unistack.org/micro/v4/store"
"go.unistack.org/micro/v4/tracer"
"go.unistack.org/micro/v3/broker"
"go.unistack.org/micro/v3/client"
"go.unistack.org/micro/v3/config"
"go.unistack.org/micro/v3/logger"
"go.unistack.org/micro/v3/metadata"
"go.unistack.org/micro/v3/meter"
"go.unistack.org/micro/v3/register"
"go.unistack.org/micro/v3/router"
"go.unistack.org/micro/v3/server"
"go.unistack.org/micro/v3/store"
"go.unistack.org/micro/v3/tracer"
)
// Options for micro service

View File

@@ -1,4 +1,4 @@
package options // import "go.unistack.org/micro/v4/options"
package options // import "go.unistack.org/micro/v3/options"
// Hook func interface
type Hook interface{}

View File

@@ -1,5 +1,5 @@
// Package http enables the http profiler
package http // import "go.unistack.org/micro/v4/profiler/http"
package http // import "go.unistack.org/micro/v3/profiler/http"
import (
"context"
@@ -7,7 +7,7 @@ import (
"net/http/pprof"
"sync"
profile "go.unistack.org/micro/v4/profiler"
profile "go.unistack.org/micro/v3/profiler"
)
type httpProfile struct {

View File

@@ -1,5 +1,5 @@
// Package pprof provides a pprof profiler which writes output to /tmp/[name].{cpu,mem}.pprof
package pprof // import "go.unistack.org/micro/v4/profiler/pprof"
package pprof // import "go.unistack.org/micro/v3/profiler/pprof"
import (
"os"
@@ -9,7 +9,7 @@ import (
"sync"
"time"
profile "go.unistack.org/micro/v4/profiler"
profile "go.unistack.org/micro/v3/profiler"
)
type profiler struct {

View File

@@ -1,5 +1,5 @@
// Package profiler is for profilers
package profiler // import "go.unistack.org/micro/v4/profiler"
package profiler // import "go.unistack.org/micro/v3/profiler"
// Profiler interface
type Profiler interface {

View File

@@ -2,11 +2,11 @@
package proxy
import (
"go.unistack.org/micro/v4/client"
"go.unistack.org/micro/v4/logger"
"go.unistack.org/micro/v4/meter"
"go.unistack.org/micro/v4/router"
"go.unistack.org/micro/v4/tracer"
"go.unistack.org/micro/v3/client"
"go.unistack.org/micro/v3/logger"
"go.unistack.org/micro/v3/meter"
"go.unistack.org/micro/v3/router"
"go.unistack.org/micro/v3/tracer"
)
// Options for proxy

View File

@@ -1,10 +1,10 @@
// Package proxy is a transparent proxy built on the micro/server
package proxy // import "go.unistack.org/micro/v4/proxy"
package proxy // import "go.unistack.org/micro/v3/proxy"
import (
"context"
"go.unistack.org/micro/v4/server"
"go.unistack.org/micro/v3/server"
)
// DefaultEndpoint holds default proxy address

View File

@@ -6,7 +6,7 @@ import (
"unicode"
"unicode/utf8"
"go.unistack.org/micro/v4/metadata"
"go.unistack.org/micro/v3/metadata"
)
// ExtractValue from reflect.Type from specified depth

View File

@@ -5,8 +5,8 @@ import (
"sync"
"time"
"go.unistack.org/micro/v4/logger"
"go.unistack.org/micro/v4/util/id"
"go.unistack.org/micro/v3/logger"
"go.unistack.org/micro/v3/util/id"
)
var (

View File

@@ -5,9 +5,9 @@ import (
"crypto/tls"
"time"
"go.unistack.org/micro/v4/logger"
"go.unistack.org/micro/v4/meter"
"go.unistack.org/micro/v4/tracer"
"go.unistack.org/micro/v3/logger"
"go.unistack.org/micro/v3/meter"
"go.unistack.org/micro/v3/tracer"
)
// Options holds options for register

View File

@@ -1,11 +1,11 @@
// Package register is an interface for service discovery
package register // import "go.unistack.org/micro/v4/register"
package register // import "go.unistack.org/micro/v3/register"
import (
"context"
"errors"
"go.unistack.org/micro/v4/metadata"
"go.unistack.org/micro/v3/metadata"
)
const (

View File

@@ -1,5 +1,5 @@
// Package dns resolves names to dns records
package dns // import "go.unistack.org/micro/v4/resolver/dns"
package dns // import "go.unistack.org/micro/v3/resolver/dns"
import (
"context"
@@ -7,7 +7,7 @@ import (
"sync"
"time"
"go.unistack.org/micro/v4/resolver"
"go.unistack.org/micro/v3/resolver"
)
// Resolver is a DNS network resolve

View File

@@ -1,11 +1,11 @@
// Package dnssrv resolves names to dns srv records
package dnssrv // import "go.unistack.org/micro/v4/resolver/dnssrv"
package dnssrv // import "go.unistack.org/micro/v3/resolver/dnssrv"
import (
"fmt"
"net"
"go.unistack.org/micro/v4/resolver"
"go.unistack.org/micro/v3/resolver"
)
// Resolver is a DNS network resolve

View File

@@ -1,5 +1,5 @@
// Package http resolves names to network addresses using a http request
package http // import "go.unistack.org/micro/v4/resolver/http"
package http // import "go.unistack.org/micro/v3/resolver/http"
import (
"encoding/json"
@@ -8,7 +8,7 @@ import (
"net/http"
"net/url"
"go.unistack.org/micro/v4/resolver"
"go.unistack.org/micro/v3/resolver"
)
// nolint: golint,revive

View File

@@ -1,8 +1,8 @@
// Package noop is a noop resolver
package noop // import "go.unistack.org/micro/v4/resolver/noop"
package noop // import "go.unistack.org/micro/v3/resolver/noop"
import (
"go.unistack.org/micro/v4/resolver"
"go.unistack.org/micro/v3/resolver"
)
// Resolver contains noop resolver

View File

@@ -1,11 +1,11 @@
// Package register resolves names using the micro register
package register // import "go.unistack.org/micro/v4/resolver/registry"
package register // import "go.unistack.org/micro/v3/resolver/registry"
import (
"context"
"go.unistack.org/micro/v4/register"
"go.unistack.org/micro/v4/resolver"
"go.unistack.org/micro/v3/register"
"go.unistack.org/micro/v3/resolver"
)
// Resolver is a register network resolver

View File

@@ -1,5 +1,5 @@
// Package resolver resolves network names to addresses
package resolver // import "go.unistack.org/micro/v4/resolver"
package resolver // import "go.unistack.org/micro/v3/resolver"
// Resolver is network resolver. It's used to find network nodes
// via the name to connect to. This is done based on Network.Name().

View File

@@ -1,8 +1,8 @@
// Package static is a static resolver
package static // import "go.unistack.org/micro/v4/resolver/static"
package static // import "go.unistack.org/micro/v3/resolver/static"
import (
"go.unistack.org/micro/v4/resolver"
"go.unistack.org/micro/v3/resolver"
)
// Resolver returns a static list of nodes. In the event the node list

View File

@@ -3,9 +3,9 @@ package router
import (
"context"
"go.unistack.org/micro/v4/logger"
"go.unistack.org/micro/v4/register"
"go.unistack.org/micro/v4/util/id"
"go.unistack.org/micro/v3/logger"
"go.unistack.org/micro/v3/register"
"go.unistack.org/micro/v3/util/id"
)
// Options are router options

View File

@@ -3,7 +3,7 @@ package router
import (
"hash/fnv"
"go.unistack.org/micro/v4/metadata"
"go.unistack.org/micro/v3/metadata"
)
var (

View File

@@ -1,5 +1,5 @@
// Package router provides a network routing control plane
package router // import "go.unistack.org/micro/v4/router"
package router // import "go.unistack.org/micro/v3/router"
import (
"errors"

View File

@@ -1,8 +1,8 @@
package random // import "go.unistack.org/micro/v4/selector/random"
package random // import "go.unistack.org/micro/v3/selector/random"
import (
"go.unistack.org/micro/v4/selector"
"go.unistack.org/micro/v4/util/rand"
"go.unistack.org/micro/v3/selector"
"go.unistack.org/micro/v3/util/rand"
)
type random struct{}

View File

@@ -3,7 +3,7 @@ package random
import (
"testing"
"go.unistack.org/micro/v4/selector"
"go.unistack.org/micro/v3/selector"
)
func TestRandom(t *testing.T) {

View File

@@ -1,8 +1,8 @@
package roundrobin // import "go.unistack.org/micro/v4/selector/roundrobin"
package roundrobin // import "go.unistack.org/micro/v3/selector/roundrobin"
import (
"go.unistack.org/micro/v4/selector"
"go.unistack.org/micro/v4/util/rand"
"go.unistack.org/micro/v3/selector"
"go.unistack.org/micro/v3/util/rand"
)
// NewSelector returns an initialised round robin selector

View File

@@ -3,7 +3,7 @@ package roundrobin
import (
"testing"
"go.unistack.org/micro/v4/selector"
"go.unistack.org/micro/v3/selector"
)
func TestRoundRobin(t *testing.T) {

View File

@@ -1,5 +1,5 @@
// Package selector is for node selection and load balancing
package selector // import "go.unistack.org/micro/v4/selector"
package selector // import "go.unistack.org/micro/v3/selector"
import (
"errors"

View File

@@ -1,6 +1,6 @@
package server
import "go.unistack.org/micro/v4/errors"
import "go.unistack.org/micro/v3/errors"
type Error struct {
id string

View File

@@ -3,7 +3,7 @@ package server
import (
"testing"
"go.unistack.org/micro/v4/errors"
"go.unistack.org/micro/v3/errors"
)
func TestError(t *testing.T) {

View File

@@ -3,7 +3,7 @@ package server
import (
"reflect"
"go.unistack.org/micro/v4/register"
"go.unistack.org/micro/v3/register"
)
type rpcHandler struct {

View File

@@ -6,13 +6,13 @@ import (
"sync"
"time"
"go.unistack.org/micro/v4/broker"
"go.unistack.org/micro/v4/codec"
"go.unistack.org/micro/v4/logger"
"go.unistack.org/micro/v4/register"
maddr "go.unistack.org/micro/v4/util/addr"
mnet "go.unistack.org/micro/v4/util/net"
"go.unistack.org/micro/v4/util/rand"
"go.unistack.org/micro/v3/broker"
"go.unistack.org/micro/v3/codec"
"go.unistack.org/micro/v3/logger"
"go.unistack.org/micro/v3/register"
maddr "go.unistack.org/micro/v3/util/addr"
mnet "go.unistack.org/micro/v3/util/net"
"go.unistack.org/micro/v3/util/rand"
)
// DefaultCodecs will be used to encode/decode
@@ -202,39 +202,6 @@ func (n *noopServer) Register() error {
n.Lock()
defer n.Unlock()
cx := config.Context
var sub broker.Subscriber
for sb := range n.subscribers {
if sb.Options().Context != nil {
cx = sb.Options().Context
}
opts := []broker.SubscribeOption{broker.SubscribeContext(cx), broker.SubscribeAutoAck(sb.Options().AutoAck)}
if queue := sb.Options().Queue; len(queue) > 0 {
opts = append(opts, broker.SubscribeGroup(queue))
}
if sb.Options().Batch {
// batch processing handler
sub, err = config.Broker.BatchSubscribe(cx, sb.Topic(), n.newBatchSubHandler(sb, config), opts...)
} else {
// single processing handler
sub, err = config.Broker.Subscribe(cx, sb.Topic(), n.newSubHandler(sb, config), opts...)
}
if err != nil {
return err
}
if config.Logger.V(logger.InfoLevel) {
config.Logger.Infof(n.opts.Context, "subscribing to topic: %s", sb.Topic())
}
n.subscribers[sb] = []broker.Subscriber{sub}
}
n.registered = true
if cacheService {
n.rsvc = service
@@ -366,6 +333,10 @@ func (n *noopServer) Start() error {
}
}
if err := n.subscribe(); err != nil {
return err
}
go func() {
t := new(time.Ticker)
@@ -449,6 +420,45 @@ func (n *noopServer) Start() error {
return nil
}
func (n *noopServer) subscribe() error {
config := n.Options()
cx := config.Context
var err error
var sub broker.Subscriber
for sb := range n.subscribers {
if sb.Options().Context != nil {
cx = sb.Options().Context
}
opts := []broker.SubscribeOption{broker.SubscribeContext(cx), broker.SubscribeAutoAck(sb.Options().AutoAck)}
if queue := sb.Options().Queue; len(queue) > 0 {
opts = append(opts, broker.SubscribeGroup(queue))
}
if sb.Options().Batch {
// batch processing handler
sub, err = config.Broker.BatchSubscribe(cx, sb.Topic(), n.createBatchSubHandler(sb, config), opts...)
} else {
// single processing handler
sub, err = config.Broker.Subscribe(cx, sb.Topic(), n.createSubHandler(sb, config), opts...)
}
if err != nil {
return err
}
if config.Logger.V(logger.InfoLevel) {
config.Logger.Infof(n.opts.Context, "subscribing to topic: %s", sb.Topic())
}
n.subscribers[sb] = []broker.Subscriber{sub}
}
return nil
}
func (n *noopServer) Stop() error {
n.RLock()
if !n.started {

View File

@@ -5,12 +5,12 @@ import (
"fmt"
"testing"
"go.unistack.org/micro/v4/broker"
"go.unistack.org/micro/v4/client"
"go.unistack.org/micro/v4/codec"
"go.unistack.org/micro/v4/logger"
"go.unistack.org/micro/v4/metadata"
"go.unistack.org/micro/v4/server"
"go.unistack.org/micro/v3/broker"
"go.unistack.org/micro/v3/client"
"go.unistack.org/micro/v3/codec"
"go.unistack.org/micro/v3/logger"
"go.unistack.org/micro/v3/metadata"
"go.unistack.org/micro/v3/server"
)
type TestHandler struct {

View File

@@ -7,16 +7,16 @@ import (
"sync"
"time"
"go.unistack.org/micro/v4/broker"
"go.unistack.org/micro/v4/codec"
"go.unistack.org/micro/v4/logger"
"go.unistack.org/micro/v4/metadata"
"go.unistack.org/micro/v4/meter"
"go.unistack.org/micro/v4/network/transport"
"go.unistack.org/micro/v4/options"
"go.unistack.org/micro/v4/register"
"go.unistack.org/micro/v4/tracer"
"go.unistack.org/micro/v4/util/id"
"go.unistack.org/micro/v3/broker"
"go.unistack.org/micro/v3/codec"
"go.unistack.org/micro/v3/logger"
"go.unistack.org/micro/v3/metadata"
"go.unistack.org/micro/v3/meter"
"go.unistack.org/micro/v3/network/transport"
"go.unistack.org/micro/v3/options"
"go.unistack.org/micro/v3/register"
"go.unistack.org/micro/v3/tracer"
"go.unistack.org/micro/v3/util/id"
)
// Option func

View File

@@ -4,10 +4,10 @@ import (
"net"
"time"
"go.unistack.org/micro/v4/metadata"
"go.unistack.org/micro/v4/register"
"go.unistack.org/micro/v4/util/addr"
"go.unistack.org/micro/v4/util/backoff"
"go.unistack.org/micro/v3/metadata"
"go.unistack.org/micro/v3/register"
"go.unistack.org/micro/v3/util/addr"
"go.unistack.org/micro/v3/util/backoff"
)
var (

View File

@@ -1,8 +1,8 @@
package server
import (
"go.unistack.org/micro/v4/codec"
"go.unistack.org/micro/v4/metadata"
"go.unistack.org/micro/v3/codec"
"go.unistack.org/micro/v3/metadata"
)
type rpcMessage struct {

View File

@@ -1,13 +1,13 @@
// Package server is an interface for a micro server
package server // import "go.unistack.org/micro/v4/server"
package server // import "go.unistack.org/micro/v3/server"
import (
"context"
"time"
"go.unistack.org/micro/v4/codec"
"go.unistack.org/micro/v4/metadata"
"go.unistack.org/micro/v4/register"
"go.unistack.org/micro/v3/codec"
"go.unistack.org/micro/v3/metadata"
"go.unistack.org/micro/v3/register"
)
// DefaultServer default server
@@ -145,11 +145,12 @@ type Stream interface {
//
// Example:
//
// type Greeter struct {}
// type Greeter struct {}
//
// func (g *Greeter) Hello(context, request, response) error {
// return nil
// }
//
// func (g *Greeter) Hello(context, request, response) error {
// return nil
// }
type Handler interface {
Name() string
Handler() interface{}

View File

@@ -10,12 +10,12 @@ import (
"unicode"
"unicode/utf8"
"go.unistack.org/micro/v4/broker"
"go.unistack.org/micro/v4/codec"
"go.unistack.org/micro/v4/errors"
"go.unistack.org/micro/v4/logger"
"go.unistack.org/micro/v4/metadata"
"go.unistack.org/micro/v4/register"
"go.unistack.org/micro/v3/broker"
"go.unistack.org/micro/v3/codec"
"go.unistack.org/micro/v3/errors"
"go.unistack.org/micro/v3/logger"
"go.unistack.org/micro/v3/metadata"
"go.unistack.org/micro/v3/register"
)
const (
@@ -191,7 +191,7 @@ func newSubscriber(topic string, sub interface{}, opts ...SubscriberOption) Subs
}
//nolint:gocyclo
func (n *noopServer) newBatchSubHandler(sb *subscriber, opts Options) broker.BatchHandler {
func (n *noopServer) createBatchSubHandler(sb *subscriber, opts Options) broker.BatchHandler {
return func(ps broker.Events) (err error) {
defer func() {
if r := recover(); r != nil {
@@ -309,7 +309,7 @@ func (n *noopServer) newBatchSubHandler(sb *subscriber, opts Options) broker.Bat
}
//nolint:gocyclo
func (n *noopServer) newSubHandler(sb *subscriber, opts Options) broker.Handler {
func (n *noopServer) createSubHandler(sb *subscriber, opts Options) broker.Handler {
return func(p broker.Event) (err error) {
defer func() {
if r := recover(); r != nil {

View File

@@ -1,20 +1,20 @@
// Package micro is a pluggable framework for microservices
package micro // import "go.unistack.org/micro/v4"
package micro // import "go.unistack.org/micro/v3"
import (
"fmt"
"sync"
"go.unistack.org/micro/v4/broker"
"go.unistack.org/micro/v4/client"
"go.unistack.org/micro/v4/config"
"go.unistack.org/micro/v4/logger"
"go.unistack.org/micro/v4/meter"
"go.unistack.org/micro/v4/register"
"go.unistack.org/micro/v4/router"
"go.unistack.org/micro/v4/server"
"go.unistack.org/micro/v4/store"
"go.unistack.org/micro/v4/tracer"
"go.unistack.org/micro/v3/broker"
"go.unistack.org/micro/v3/client"
"go.unistack.org/micro/v3/config"
"go.unistack.org/micro/v3/logger"
"go.unistack.org/micro/v3/meter"
"go.unistack.org/micro/v3/register"
"go.unistack.org/micro/v3/router"
"go.unistack.org/micro/v3/server"
"go.unistack.org/micro/v3/store"
"go.unistack.org/micro/v3/tracer"
)
// Service is an interface that wraps the lower level components.
@@ -88,7 +88,6 @@ func (s *service) Name() string {
// Init initialises options. Additionally it calls cmd.Init
// which parses command line flags. cmd.Init is only called
// on first Init.
//
//nolint:gocyclo
func (s *service) Init(opts ...Option) error {
var err error

View File

@@ -5,16 +5,16 @@ import (
"reflect"
"testing"
"go.unistack.org/micro/v4/broker"
"go.unistack.org/micro/v4/client"
"go.unistack.org/micro/v4/config"
"go.unistack.org/micro/v4/logger"
"go.unistack.org/micro/v4/meter"
"go.unistack.org/micro/v4/register"
"go.unistack.org/micro/v4/router"
"go.unistack.org/micro/v4/server"
"go.unistack.org/micro/v4/store"
"go.unistack.org/micro/v4/tracer"
"go.unistack.org/micro/v3/broker"
"go.unistack.org/micro/v3/client"
"go.unistack.org/micro/v3/config"
"go.unistack.org/micro/v3/logger"
"go.unistack.org/micro/v3/meter"
"go.unistack.org/micro/v3/register"
"go.unistack.org/micro/v3/router"
"go.unistack.org/micro/v3/server"
"go.unistack.org/micro/v3/store"
"go.unistack.org/micro/v3/tracer"
)
type testItem struct {

View File

@@ -5,7 +5,7 @@ import (
"testing"
"time"
"go.unistack.org/micro/v4/store"
"go.unistack.org/micro/v3/store"
)
func TestMemoryReInit(t *testing.T) {

View File

@@ -5,11 +5,11 @@ import (
"crypto/tls"
"time"
"go.unistack.org/micro/v4/codec"
"go.unistack.org/micro/v4/logger"
"go.unistack.org/micro/v4/metadata"
"go.unistack.org/micro/v4/meter"
"go.unistack.org/micro/v4/tracer"
"go.unistack.org/micro/v3/codec"
"go.unistack.org/micro/v3/logger"
"go.unistack.org/micro/v3/metadata"
"go.unistack.org/micro/v3/meter"
"go.unistack.org/micro/v3/tracer"
)
// Options contains configuration for the Store

View File

@@ -1,5 +1,5 @@
// Package store is an interface for distributed data storage.
package store // import "go.unistack.org/micro/v4/store"
package store // import "go.unistack.org/micro/v3/store"
import (
"context"

View File

@@ -3,9 +3,9 @@ package sync
import (
"time"
"go.unistack.org/micro/v4/logger"
"go.unistack.org/micro/v4/meter"
"go.unistack.org/micro/v4/tracer"
"go.unistack.org/micro/v3/logger"
"go.unistack.org/micro/v3/meter"
"go.unistack.org/micro/v3/tracer"
)
// Options holds the sync options

View File

@@ -1,5 +1,5 @@
// Package sync is an interface for distributed synchronization
package sync // import "go.unistack.org/micro/v4/sync"
package sync // import "go.unistack.org/micro/v3/sync"
import (
"errors"

View File

@@ -4,23 +4,37 @@ import (
"context"
)
var _ Tracer = (*noopTracer)(nil)
type noopTracer struct {
opts Options
opts Options
spans []Span
}
func (t *noopTracer) Spans() []Span {
return t.spans
}
func (t *noopTracer) Start(ctx context.Context, name string, opts ...SpanOption) (context.Context, Span) {
options := NewSpanOptions(opts...)
span := &noopSpan{
name: name,
ctx: ctx,
tracer: t,
opts: NewSpanOptions(opts...),
labels: options.Labels,
kind: options.Kind,
}
if span.ctx == nil {
span.ctx = context.Background()
}
t.spans = append(t.spans, span)
return NewSpanContext(ctx, span), span
}
func (t *noopTracer) Flush(ctx context.Context) error {
return nil
}
func (t *noopTracer) Init(opts ...Option) error {
for _, o := range opts {
o(&t.opts)
@@ -32,13 +46,21 @@ func (t *noopTracer) Name() string {
return t.opts.Name
}
type noopEvent struct {
name string
labels []interface{}
}
type noopSpan struct {
ctx context.Context
tracer Tracer
name string
opts SpanOptions
status SpanStatus
statusMsg string
events []*noopEvent
labels []interface{}
logs []interface{}
kind SpanKind
status SpanStatus
}
func (s *noopSpan) Finish(opts ...SpanOption) {
@@ -53,22 +75,24 @@ func (s *noopSpan) Tracer() Tracer {
}
func (s *noopSpan) AddEvent(name string, opts ...EventOption) {
options := NewEventOptions(opts...)
s.events = append(s.events, &noopEvent{name: name, labels: options.Labels})
}
func (s *noopSpan) SetName(name string) {
s.name = name
}
func (s *noopSpan) SetLabels(labels ...interface{}) {
s.opts.Labels = labels
func (s *noopSpan) AddLogs(kv ...interface{}) {
s.logs = append(s.logs, kv...)
}
func (s *noopSpan) AddLabels(labels ...interface{}) {
s.opts.Labels = append(s.opts.Labels, labels...)
func (s *noopSpan) AddLabels(kv ...interface{}) {
s.labels = append(s.labels, kv...)
}
func (s *noopSpan) Kind() SpanKind {
return s.opts.Kind
return s.kind
}
func (s *noopSpan) Status() (SpanStatus, string) {

View File

@@ -3,7 +3,7 @@ package tracer
import (
"context"
"go.unistack.org/micro/v4/logger"
"go.unistack.org/micro/v3/logger"
)
type SpanStatus int
@@ -91,14 +91,22 @@ type SpanOptions struct {
type SpanOption func(o *SpanOptions)
// EventOptions contains event options
type EventOptions struct{}
type EventOptions struct {
Labels []interface{}
}
// EventOption func signature
type EventOption func(o *EventOptions)
func WithSpanLabels(labels ...interface{}) SpanOption {
func WithEventLabels(kv ...interface{}) EventOption {
return func(o *EventOptions) {
o.Labels = kv
}
}
func WithSpanLabels(kv ...interface{}) SpanOption {
return func(o *SpanOptions) {
o.Labels = labels
o.Labels = kv
}
}
@@ -128,6 +136,15 @@ func Logger(l logger.Logger) Option {
}
}
// NewEventOptions returns default EventOptions
func NewEventOptions(opts ...EventOption) EventOptions {
options := EventOptions{}
for _, o := range opts {
o(&options)
}
return options
}
// NewSpanOptions returns default SpanOptions
func NewSpanOptions(opts ...SpanOption) SpanOptions {
options := SpanOptions{

View File

@@ -1,8 +1,10 @@
// Package tracer provides an interface for distributed tracing
package tracer // import "go.unistack.org/micro/v4/tracer"
package tracer // import "go.unistack.org/micro/v3/tracer"
import (
"context"
"fmt"
"sort"
)
// DefaultTracer is the global default tracer
@@ -16,6 +18,8 @@ type Tracer interface {
Init(...Option) error
// Start a trace
Start(ctx context.Context, name string, opts ...SpanOption) (context.Context, Span)
// Flush flushes spans
Flush(ctx context.Context) error
}
type Span interface {
@@ -23,8 +27,6 @@ type Span interface {
Tracer() Tracer
// Finish complete and send span
Finish(opts ...SpanOption)
// AddEvent add event to span
AddEvent(name string, opts ...EventOption)
// Context return context with span
Context() context.Context
// SetName set the span name
@@ -33,10 +35,46 @@ type Span interface {
SetStatus(status SpanStatus, msg string)
// Status returns span status and msg
Status() (SpanStatus, string)
// SetLabels set the span labels
SetLabels(labels ...interface{})
// AddLabels append the span labels
AddLabels(labels ...interface{})
// AddLabels append labels to span
AddLabels(kv ...interface{})
// AddEvent append event to span
AddEvent(name string, opts ...EventOption)
// AddEvent append event to span
AddLogs(kv ...interface{})
// Kind returns span kind
Kind() SpanKind
}
// sort labels alphabeticaly by label name
type byKey []interface{}
func (k byKey) Len() int { return len(k) / 2 }
func (k byKey) Less(i, j int) bool { return fmt.Sprintf("%s", k[i*2]) < fmt.Sprintf("%s", k[j*2]) }
func (k byKey) Swap(i, j int) {
k[i*2], k[j*2] = k[j*2], k[i*2]
k[i*2+1], k[j*2+1] = k[j*2+1], k[i*2+1]
}
func UniqLabels(labels []interface{}) []interface{} {
if len(labels)%2 == 1 {
labels = labels[:len(labels)-1]
}
if len(labels) > 2 {
sort.Sort(byKey(labels))
idx := 0
for {
if labels[idx] == labels[idx+2] {
copy(labels[idx:], labels[idx+2:])
labels = labels[:len(labels)-2]
} else {
idx += 2
}
if idx+2 >= len(labels) {
break
}
}
}
return labels
}

13
tracer/tracer_test.go Normal file
View File

@@ -0,0 +1,13 @@
package tracer
import (
"testing"
)
func TestUniqLabels(t *testing.T) {
labels := []interface{}{"key1", "val1", "key1", "val2"}
labels = UniqLabels(labels)
if labels[1] != "val2" {
t.Fatalf("UniqLabels not works")
}
}

View File

@@ -1,117 +1,98 @@
// Package wrapper provides wrapper for Tracer
package wrapper // import "go.unistack.org/micro/v4/tracer/wrapper"
package wrapper // import "go.unistack.org/micro/v3/tracer/wrapper"
import (
"context"
"fmt"
"strings"
"go.unistack.org/micro/v4/client"
"go.unistack.org/micro/v4/metadata"
"go.unistack.org/micro/v4/server"
"go.unistack.org/micro/v4/tracer"
"go.unistack.org/micro/v3/client"
"go.unistack.org/micro/v3/metadata"
"go.unistack.org/micro/v3/server"
"go.unistack.org/micro/v3/tracer"
)
var DefaultHeadersExctract = []string{metadata.HeaderXRequestID}
func ExtractDefaultLabels(md metadata.Metadata) []interface{} {
labels := make([]interface{}, 0, len(DefaultHeadersExctract))
for _, k := range DefaultHeadersExctract {
if v, ok := md.Get(k); ok {
labels = append(labels, strings.ToLower(k), v)
}
}
return labels
}
var (
DefaultClientCallObserver = func(ctx context.Context, req client.Request, rsp interface{}, opts []client.CallOption, sp tracer.Span, err error) {
sp.SetName(fmt.Sprintf("Call %s.%s", req.Service(), req.Method()))
var labels []interface{}
if md, ok := metadata.FromOutgoingContext(ctx); ok {
labels = make([]interface{}, 0, len(md)+1)
for k, v := range md {
labels = append(labels, k, v)
}
labels = append(labels, ExtractDefaultLabels(md)...)
}
if err != nil {
labels = append(labels, "error", err.Error())
sp.SetStatus(tracer.SpanStatusError, err.Error())
}
labels = append(labels, "kind", sp.Kind())
sp.SetLabels(labels...)
sp.AddLabels(labels...)
}
DefaultClientStreamObserver = func(ctx context.Context, req client.Request, opts []client.CallOption, stream client.Stream, sp tracer.Span, err error) {
sp.SetName(fmt.Sprintf("Stream %s.%s", req.Service(), req.Method()))
var labels []interface{}
if md, ok := metadata.FromOutgoingContext(ctx); ok {
labels = make([]interface{}, 0, len(md))
for k, v := range md {
labels = append(labels, k, v)
}
labels = append(labels, ExtractDefaultLabels(md)...)
}
if err != nil {
labels = append(labels, "error", err.Error())
sp.SetStatus(tracer.SpanStatusError, err.Error())
}
labels = append(labels, "kind", sp.Kind())
sp.SetLabels(labels...)
sp.AddLabels(labels...)
}
DefaultClientPublishObserver = func(ctx context.Context, msg client.Message, opts []client.PublishOption, sp tracer.Span, err error) {
sp.SetName(fmt.Sprintf("Publish %s", msg.Topic()))
var labels []interface{}
if md, ok := metadata.FromOutgoingContext(ctx); ok {
labels = make([]interface{}, 0, len(md))
for k, v := range md {
labels = append(labels, k, v)
}
labels = append(labels, ExtractDefaultLabels(md)...)
}
labels = append(labels, ExtractDefaultLabels(msg.Metadata())...)
if err != nil {
labels = append(labels, "error", err.Error())
sp.SetStatus(tracer.SpanStatusError, err.Error())
}
labels = append(labels, "kind", sp.Kind())
sp.SetLabels(labels...)
sp.AddLabels(labels...)
}
DefaultServerHandlerObserver = func(ctx context.Context, req server.Request, rsp interface{}, sp tracer.Span, err error) {
sp.SetName(fmt.Sprintf("Handler %s.%s", req.Service(), req.Method()))
var labels []interface{}
if md, ok := metadata.FromIncomingContext(ctx); ok {
labels = make([]interface{}, 0, len(md))
for k, v := range md {
labels = append(labels, k, v)
}
labels = append(labels, ExtractDefaultLabels(md)...)
}
if err != nil {
labels = append(labels, "error", err.Error())
sp.SetStatus(tracer.SpanStatusError, err.Error())
}
labels = append(labels, "kind", sp.Kind())
sp.SetLabels(labels...)
sp.AddLabels(labels...)
}
DefaultServerSubscriberObserver = func(ctx context.Context, msg server.Message, sp tracer.Span, err error) {
sp.SetName(fmt.Sprintf("Subscriber %s", msg.Topic()))
var labels []interface{}
if md, ok := metadata.FromIncomingContext(ctx); ok {
labels = make([]interface{}, 0, len(md))
for k, v := range md {
labels = append(labels, k, v)
}
labels = append(labels, ExtractDefaultLabels(md)...)
}
labels = append(labels, ExtractDefaultLabels(msg.Header())...)
if err != nil {
labels = append(labels, "error", err.Error())
sp.SetStatus(tracer.SpanStatusError, err.Error())
}
labels = append(labels, "kind", sp.Kind())
sp.SetLabels(labels...)
sp.AddLabels(labels...)
}
DefaultClientCallFuncObserver = func(ctx context.Context, addr string, req client.Request, rsp interface{}, opts client.CallOptions, sp tracer.Span, err error) {
sp.SetName(fmt.Sprintf("Call %s.%s", req.Service(), req.Method()))
sp.SetName(fmt.Sprintf("%s.%s call", req.Service(), req.Method()))
var labels []interface{}
if md, ok := metadata.FromOutgoingContext(ctx); ok {
labels = make([]interface{}, 0, len(md))
for k, v := range md {
labels = append(labels, k, v)
}
labels = append(labels, ExtractDefaultLabels(md)...)
}
if err != nil {
labels = append(labels, "error", err.Error())
sp.SetStatus(tracer.SpanStatusError, err.Error())
}
labels = append(labels, "kind", sp.Kind())
sp.SetLabels(labels...)
sp.AddLabels(labels...)
}
DefaultSkipEndpoints = []string{"Meter.Metrics", "Health.Live", "Health.Ready", "Health.Version"}
@@ -241,16 +222,22 @@ func (ot *tWrapper) Call(ctx context.Context, req client.Request, rsp interface{
}
}
sp, ok := tracer.SpanFromContext(ctx)
if !ok {
ctx, sp = ot.opts.Tracer.Start(ctx, "", tracer.WithSpanKind(tracer.SpanKindClient))
}
nctx, sp := ot.opts.Tracer.Start(ctx, fmt.Sprintf("%s.%s rpc-client", req.Service(), req.Method()),
tracer.WithSpanKind(tracer.SpanKindClient),
tracer.WithSpanLabels(
"rpc.service", req.Service(),
"rpc.method", req.Method(),
"rpc.flavor", "rpc",
"rpc.call", "/"+req.Service()+"/"+req.Endpoint(),
"rpc.call_type", "unary",
),
)
defer sp.Finish()
err := ot.Client.Call(ctx, req, rsp, opts...)
err := ot.Client.Call(nctx, req, rsp, opts...)
for _, o := range ot.opts.ClientCallObservers {
o(ctx, req, rsp, opts, sp, err)
o(nctx, req, rsp, opts, sp, err)
}
return err
@@ -264,32 +251,36 @@ func (ot *tWrapper) Stream(ctx context.Context, req client.Request, opts ...clie
}
}
sp, ok := tracer.SpanFromContext(ctx)
if !ok {
ctx, sp = ot.opts.Tracer.Start(ctx, "", tracer.WithSpanKind(tracer.SpanKindClient))
}
nctx, sp := ot.opts.Tracer.Start(ctx, fmt.Sprintf("%s.%s rpc-client", req.Service(), req.Method()),
tracer.WithSpanKind(tracer.SpanKindClient),
tracer.WithSpanLabels(
"rpc.service", req.Service(),
"rpc.method", req.Method(),
"rpc.flavor", "rpc",
"rpc.call", "/"+req.Service()+"/"+req.Endpoint(),
"rpc.call_type", "stream",
),
)
defer sp.Finish()
stream, err := ot.Client.Stream(ctx, req, opts...)
stream, err := ot.Client.Stream(nctx, req, opts...)
for _, o := range ot.opts.ClientStreamObservers {
o(ctx, req, opts, stream, sp, err)
o(nctx, req, opts, stream, sp, err)
}
return stream, err
}
func (ot *tWrapper) Publish(ctx context.Context, msg client.Message, opts ...client.PublishOption) error {
sp, ok := tracer.SpanFromContext(ctx)
if !ok {
ctx, sp = ot.opts.Tracer.Start(ctx, "", tracer.WithSpanKind(tracer.SpanKindProducer))
}
nctx, sp := ot.opts.Tracer.Start(ctx, msg.Topic()+" publish", tracer.WithSpanKind(tracer.SpanKindProducer))
defer sp.Finish()
err := ot.Client.Publish(ctx, msg, opts...)
sp.AddLabels("messaging.destination.name", msg.Topic())
sp.AddLabels("messaging.operation", "publish")
err := ot.Client.Publish(nctx, msg, opts...)
for _, o := range ot.opts.ClientPublishObservers {
o(ctx, msg, opts, sp, err)
o(nctx, msg, opts, sp, err)
}
return err
@@ -303,32 +294,41 @@ func (ot *tWrapper) ServerHandler(ctx context.Context, req server.Request, rsp i
}
}
sp, ok := tracer.SpanFromContext(ctx)
if !ok {
ctx, sp = ot.opts.Tracer.Start(ctx, "", tracer.WithSpanKind(tracer.SpanKindServer))
callType := "unary"
if req.Stream() {
callType = "stream"
}
nctx, sp := ot.opts.Tracer.Start(ctx, fmt.Sprintf("%s.%s rpc-server", req.Service(), req.Method()),
tracer.WithSpanKind(tracer.SpanKindServer),
tracer.WithSpanLabels(
"rpc.service", req.Service(),
"rpc.method", req.Method(),
"rpc.flavor", "rpc",
"rpc.call", "/"+req.Service()+"/"+req.Endpoint(),
"rpc.call_type", callType,
),
)
defer sp.Finish()
err := ot.serverHandler(ctx, req, rsp)
err := ot.serverHandler(nctx, req, rsp)
for _, o := range ot.opts.ServerHandlerObservers {
o(ctx, req, rsp, sp, err)
o(nctx, req, rsp, sp, err)
}
return err
}
func (ot *tWrapper) ServerSubscriber(ctx context.Context, msg server.Message) error {
sp, ok := tracer.SpanFromContext(ctx)
if !ok {
ctx, sp = ot.opts.Tracer.Start(ctx, "", tracer.WithSpanKind(tracer.SpanKindConsumer))
}
nctx, sp := ot.opts.Tracer.Start(ctx, msg.Topic()+" process", tracer.WithSpanKind(tracer.SpanKindConsumer))
defer sp.Finish()
err := ot.serverSubscriber(ctx, msg)
sp.AddLabels("messaging.operation", "process")
sp.AddLabels("messaging.source.name", msg.Topic())
err := ot.serverSubscriber(nctx, msg)
for _, o := range ot.opts.ServerSubscriberObservers {
o(ctx, msg, sp, err)
o(nctx, msg, sp, err)
}
return err
@@ -366,16 +366,23 @@ func (ot *tWrapper) ClientCallFunc(ctx context.Context, addr string, req client.
}
}
sp, ok := tracer.SpanFromContext(ctx)
if !ok {
ctx, sp = ot.opts.Tracer.Start(ctx, "", tracer.WithSpanKind(tracer.SpanKindClient))
}
nctx, sp := ot.opts.Tracer.Start(ctx, fmt.Sprintf("%s.%s rpc-client", req.Service(), req.Method()),
tracer.WithSpanKind(tracer.SpanKindClient),
tracer.WithSpanLabels(
"rpc.service", req.Service(),
"rpc.method", req.Method(),
"rpc.flavor", "rpc",
"rpc.call", "/"+req.Service()+"/"+req.Endpoint(),
"rpc.call_type", "unary",
),
)
defer sp.Finish()
err := ot.clientCallFunc(ctx, addr, req, rsp, opts)
err := ot.clientCallFunc(nctx, addr, req, rsp, opts)
for _, o := range ot.opts.ClientCallFuncObservers {
o(ctx, addr, req, rsp, opts, sp, err)
o(nctx, addr, req, rsp, opts, sp, err)
}
return err

View File

@@ -1,4 +1,4 @@
package addr // import "go.unistack.org/micro/v4/util/addr"
package addr // import "go.unistack.org/micro/v3/util/addr"
import (
"fmt"
@@ -58,7 +58,6 @@ func IsLocal(addr string) bool {
}
// Extract returns a real ip
//
//nolint:gocyclo
func Extract(addr string) (string, error) {
// if addr specified then its returned

View File

@@ -1,5 +1,5 @@
// Package backoff provides backoff functionality
package backoff // import "go.unistack.org/micro/v4/util/backoff"
package backoff // import "go.unistack.org/micro/v3/util/backoff"
import (
"math"

Some files were not shown because too many files have changed in this diff Show More