You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by wu...@apache.org on 2023/05/10 14:38:12 UTC
[skywalking-go] branch main updated: Skip tracing if gRPC connection is down (#30)
This is an automated email from the ASF dual-hosted git repository.
wusheng pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/skywalking-go.git
The following commit(s) were added to refs/heads/main by this push:
new d9dd078 Skip tracing if gRPC connection is down (#30)
d9dd078 is described below
commit d9dd078e023701b23588f28f93d374618be3368c
Author: mrproliu <74...@qq.com>
AuthorDate: Wed May 10 22:36:31 2023 +0800
Skip tracing if gRPC connection is down (#30)
---
plugins/core/reporter/api.go | 8 ++++++++
plugins/core/reporter/grpc/grpc.go | 24 ++++++++++++++++++------
plugins/core/test_base.go | 6 ++++++
plugins/core/tracer.go | 4 ++++
plugins/core/tracing.go | 2 +-
plugins/core/tracing_test.go | 15 +++++++++++++++
tools/go-agent/tools/dst.go | 4 ++++
7 files changed, 56 insertions(+), 7 deletions(-)
diff --git a/plugins/core/reporter/api.go b/plugins/core/reporter/api.go
index 93c2909..b3a0910 100644
--- a/plugins/core/reporter/api.go
+++ b/plugins/core/reporter/api.go
@@ -76,8 +76,16 @@ func (e *Entity) GetInstanceName() string {
return e.ServiceInstanceName
}
+type ConnectionStatus int32
+
+var (
+ ConnectionStatusConnected ConnectionStatus = 1
+ ConnectionStatusDisconnect ConnectionStatus = 2
+)
+
type Reporter interface {
Boot(entity *Entity, cdsWatchers []AgentConfigChangeWatcher)
Send(spans []ReportedSpan)
+ ConnectionStatus() ConnectionStatus
Close()
}
diff --git a/plugins/core/reporter/grpc/grpc.go b/plugins/core/reporter/grpc/grpc.go
index 05803e7..604bdf2 100644
--- a/plugins/core/reporter/grpc/grpc.go
+++ b/plugins/core/reporter/grpc/grpc.go
@@ -45,10 +45,11 @@ const (
// NewGRPCReporter create a new reporter to send data to gRPC oap server. Only one backend address is allowed.
func NewGRPCReporter(logger operator.LogOperator, serverAddr string, opts ...ReporterOption) (reporter.Reporter, error) {
r := &gRPCReporter{
- logger: logger,
- sendCh: make(chan *agentv3.SegmentObject, maxSendQueueSize),
- checkInterval: defaultCheckInterval,
- cdsInterval: defaultCDSInterval, // cds default on
+ logger: logger,
+ sendCh: make(chan *agentv3.SegmentObject, maxSendQueueSize),
+ checkInterval: defaultCheckInterval,
+ cdsInterval: defaultCDSInterval, // cds default on
+ connectionStatus: reporter.ConnectionStatusConnected,
}
for _, o := range opts {
o(r)
@@ -92,7 +93,8 @@ type gRPCReporter struct {
creds credentials.TransportCredentials
// bootFlag is set if Boot be executed
- bootFlag bool
+ bootFlag bool
+ connectionStatus reporter.ConnectionStatus
}
func (r *gRPCReporter) Boot(entity *reporter.Entity, cdsWatchers []reporter.AgentConfigChangeWatcher) {
@@ -103,6 +105,10 @@ func (r *gRPCReporter) Boot(entity *reporter.Entity, cdsWatchers []reporter.Agen
r.bootFlag = true
}
+func (r *gRPCReporter) ConnectionStatus() reporter.ConnectionStatus {
+ return r.connectionStatus
+}
+
func (r *gRPCReporter) Send(spans []reporter.ReportedSpan) {
spanSize := len(spans)
if spanSize < 1 {
@@ -276,9 +282,15 @@ func (r *gRPCReporter) check() {
go func() {
instancePropertiesSubmitted := false
for {
- if r.conn.GetState() == connectivity.Shutdown {
+ state := r.conn.GetState()
+ if state == connectivity.Shutdown {
break
}
+ if state == connectivity.TransientFailure {
+ r.connectionStatus = reporter.ConnectionStatusDisconnect
+ } else {
+ r.connectionStatus = reporter.ConnectionStatusConnected
+ }
if !instancePropertiesSubmitted {
err := r.reportInstanceProperties()
diff --git a/plugins/core/test_base.go b/plugins/core/test_base.go
index 9c93590..8c880e8 100644
--- a/plugins/core/test_base.go
+++ b/plugins/core/test_base.go
@@ -24,6 +24,7 @@ import (
var tlsData interface{}
var Tracing *Tracer
+var ReportConnectionStatus = reporter.ConnectionStatusConnected
func init() {
SetGLS = func(i interface{}) {
@@ -43,6 +44,7 @@ func ResetTracingContext() {
Tracing = &Tracer{initFlag: 1, Sampler: NewConstSampler(true), Reporter: &StoreReporter{},
ServiceEntity: NewEntity("test", "test-instance")}
SetAsNewGoroutine()
+ ReportConnectionStatus = reporter.ConnectionStatusConnected
}
func SetAsNewGoroutine() {
@@ -74,5 +76,9 @@ func (r *StoreReporter) Send(spans []reporter.ReportedSpan) {
r.Spans = append(r.Spans, spans...)
}
+func (r *StoreReporter) ConnectionStatus() reporter.ConnectionStatus {
+ return ReportConnectionStatus
+}
+
func (r *StoreReporter) Close() {
}
diff --git a/plugins/core/tracer.go b/plugins/core/tracer.go
index 56559b1..003d3c1 100644
--- a/plugins/core/tracer.go
+++ b/plugins/core/tracer.go
@@ -110,6 +110,10 @@ func (e *emptyReporter) Boot(entity *reporter.Entity, cdsWatchers []reporter.Age
func (e *emptyReporter) Send(spans []reporter.ReportedSpan) {
}
+func (e *emptyReporter) ConnectionStatus() reporter.ConnectionStatus {
+ return reporter.ConnectionStatusDisconnect
+}
+
// nolint
func (e *emptyReporter) Close() {
}
diff --git a/plugins/core/tracing.go b/plugins/core/tracing.go
index 211fffe..7b158ca 100644
--- a/plugins/core/tracing.go
+++ b/plugins/core/tracing.go
@@ -134,7 +134,7 @@ func (t *Tracer) SetRuntimeContextValue(key string, value interface{}) {
}
func (t *Tracer) createNoop() (*TracingContext, TracingSpan, bool) {
- if !t.InitSuccess() {
+ if !t.InitSuccess() || t.Reporter.ConnectionStatus() == reporter.ConnectionStatusDisconnect {
return nil, &NoopSpan{}, true
}
ctx := getTracingContext()
diff --git a/plugins/core/tracing_test.go b/plugins/core/tracing_test.go
index 5f9313d..e5a8d0e 100644
--- a/plugins/core/tracing_test.go
+++ b/plugins/core/tracing_test.go
@@ -25,6 +25,7 @@ import (
agentv3 "skywalking.apache.org/repo/goapi/collect/language/agent/v3"
+ "github.com/apache/skywalking-go/plugins/core/reporter"
"github.com/apache/skywalking-go/plugins/core/tracing"
)
@@ -224,6 +225,20 @@ func TestSpanContextReading(t *testing.T) {
assert.Equal(t, parentEndpoint, refCtx.GetParentEndpoint(), "ref endpoint not correct")
}
+func TestReporterDisconnect(t *testing.T) {
+ defer ResetTracingContext()
+ ReportConnectionStatus = reporter.ConnectionStatusDisconnect
+ s, err := tracing.CreateEntrySpan("/entry", func(key string) (string, error) {
+ return "", nil
+ })
+ assert.NoError(t, err)
+ assert.NotNil(t, s, "span should not be nil")
+ s.End()
+ time.Sleep(time.Millisecond * 50)
+ spans := GetReportedSpans()
+ assert.Equal(t, 0, len(spans), "should no span been collected")
+}
+
func TestSpanOperation(t *testing.T) {
defer ResetTracingContext()
spanCreations := []func(op tracing.SpanOption) (tracing.Span, error){
diff --git a/tools/go-agent/tools/dst.go b/tools/go-agent/tools/dst.go
index 878baee..ca406b5 100644
--- a/tools/go-agent/tools/dst.go
+++ b/tools/go-agent/tools/dst.go
@@ -104,6 +104,10 @@ func removePackageRef(parent dst.Node, current *dst.SelectorExpr) {
p.Elt = dst.NewIdent(current.Sel.Name)
case *dst.CallExpr:
p.Fun = dst.NewIdent(current.Sel.Name)
+ case *dst.KeyValueExpr:
+ p.Value = dst.NewIdent(current.Sel.Name)
+ case *dst.AssignStmt:
+ p.Rhs = []dst.Expr{dst.NewIdent(current.Sel.Name)}
}
}