Merge pull request #156 from lhui1991/apollo_timeout_fix

Apollo timeout fix

dbrouter/dbrouter.go

@@ -117,8 +117,8 @@ func (m *Router) SqlExec(ctx context.Context, cluster string, query func(*DB, []
 	for _, item := range tables {
 		tmptables = append(tmptables, item)
 	}
-
 	err = query(db, tmptables)
+	statReqErr(cluster, table, err)
 	// record breaker
 	statBreaker(cluster, table, err)
 	return err
@@ -183,8 +183,8 @@ func (m *Router) OrmExec(ctx context.Context, cluster string, query func(*GormDB
 	for _, item := range tables {
 		tmptables = append(tmptables, item)
 	}
-
 	err = query(db, tmptables)
+	statReqErr(cluster, table, err)
 	// stat breaker
 	statBreaker(cluster, table, err)
 	return err
@@ -260,8 +260,8 @@ func (m *Router) mongoExec(ctx context.Context, consistency mode, cluster, table
 		m.report.IncQuery(cluster, table, st.Duration())
 		slog.Tracef(ctx, "%s const:%d cls:%s table:%s dur:%d", fun, consistency, cluster, table, dur)
 	}()
-
 	err = query(coll)
+	statReqErr(cluster, table, err)
 	statBreaker(cluster, table, err)
 	return err
 }

dbrouter/metrics.go

@@ -0,0 +1,28 @@
+package dbrouter
+
+import (
+	"gitlab.pri.ibanyu.com/middleware/seaweed/xstat/xmetric/xprometheus"
+)
+
+const (
+	namespace = "palfish"
+	subsystem = "db"
+)
+
+var (
+	_metricReqErr = xprometheus.NewCounter(&xprometheus.CounterVecOpts{
+		Namespace:  namespace,
+		Subsystem:  subsystem,
+		Name:       "request_err_total",
+		Help:       "db request err total",
+		LabelNames: []string{xprometheus.LabelSource},
+	})
+)
+
+func statReqErr(cluster, table string, err error) {
+	if err != nil {
+		source := cluster + "." + table
+		_metricReqErr.With(xprometheus.LabelSource, source).Inc()
+	}
+	return
+}

go.mod

@@ -1,15 +1,15 @@
 module github.com/shawnfeng/sutil
 
 require (
-	github.com/ZhengHe-MD/agollo/v4 v4.1.3
+	github.com/ZhengHe-MD/agollo/v4 v4.1.4
 	github.com/ZhengHe-MD/properties v0.2.2
 	github.com/bitly/go-simplejson v0.5.0
 	github.com/coreos/etcd v3.3.17+incompatible
 	github.com/fzzy/radix v0.4.9-0.20141113025130-a3a55de9c594
 	github.com/go-redis/redis v6.15.1+incompatible
 	github.com/go-sql-driver/mysql v1.4.1
-	github.com/golang/protobuf v1.3.2
-	github.com/google/uuid v1.1.0
+	github.com/golang/protobuf v1.3.3
+	github.com/google/uuid v1.1.1
 	github.com/jinzhu/gorm v1.9.10
 	github.com/jmoiron/sqlx v1.2.0
 	github.com/julienschmidt/httprouter v1.2.0
@@ -24,12 +24,10 @@ require (
 	github.com/shawnfeng/lumberjack.v2 v0.0.0-20181226094728-63d76296ede8
 	github.com/stretchr/testify v1.4.0
 	github.com/uber/jaeger-client-go v2.20.1+incompatible
-	github.com/ugorji/go v1.1.7 // indirect
 	github.com/vaughan0/go-ini v0.0.0-20130923145212-a98ad7ee00ec
 	gitlab.pri.ibanyu.com/middleware/delayqueue v0.0.0-20200213090847-cd24af2bd1f2
 	gitlab.pri.ibanyu.com/middleware/seaweed v1.0.20
 	go.uber.org/zap v1.10.0
-	golang.org/x/tools v0.0.0-20191120001058-ad01d5993d97 // indirect
 
 	gopkg.in/mgo.v2 v2.0.0-20190816093944-a6b53ec6cb22
 

sconf/center/apollo.go

@@ -2,15 +2,13 @@ package center
 
 import (
 	"context"
-	"fmt"
 	"github.com/ZhengHe-MD/agollo/v4"
 	"github.com/ZhengHe-MD/properties"
 	"github.com/opentracing/opentracing-go"
 	"github.com/shawnfeng/sutil/slog/slog"
 	"os"
 	"strings"
 	"sync"
-	"time"
 )
 
 const (
@@ -21,7 +19,6 @@ const (
 	defaultCacheDir             = "/tmp/sconfcenter"
 	defaultNamespaceApplication = "application"
 	defaultChangeEventSize      = 32
-	defaultInitTimeout          = 6 * time.Second
 )
 
 type apolloConfigCenter struct {
@@ -88,22 +85,10 @@ func (ap *apolloConfigCenter) Init(ctx context.Context, serviceName string, name
 
 	slog.Infof(ctx, "%s start agollo with conf:%v", fun, ap.conf)
 
-	startCh := make(chan int, 1)
-
-	go func() {
-		if err := ap.ag.Start(); err != nil {
-			slog.Errorf(ctx, "%s agollo starts err:%v", fun, err)
-		} else {
-			slog.Infof(ctx, "%s agollo starts succeed:%v", fun, err)
-		}
-		startCh <- 1
-	}()
-
-	select {
-	case <-time.After(defaultInitTimeout):
-		err = fmt.Errorf("%s init agollo timeout after %v", fun, defaultInitTimeout)
-		slog.Errorf(ctx, err.Error())
-	case <-startCh:
+	if err = ap.ag.Start(); err != nil {
+		slog.Errorf(ctx, "%s agollo starts err:%v", fun, err)
+	} else {
+		slog.Infof(ctx, "%s agollo starts succeed:%v", fun, err)
 	}
 
 	return

sconf/center/apollo_test.go

@@ -1,7 +1,10 @@
 package center
 
 import (
+	"context"
 	"testing"
+	"github.com/stretchr/testify/assert"
+	"time"
 )
 
 func assertStringEqual(t *testing.T, s1, s2 string) {
@@ -19,3 +22,13 @@ func TestConfFromEnv(t *testing.T) {
 		assertStringEqual(t, conf.IP, defaultHostPort)
 	})
 }
+
+func TestInit(t *testing.T) {
+	ass := assert.New(t)
+	center,err := NewConfigCenter(ApolloConfigCenter)
+	ass.Nil(err)
+	err = center.Init(context.TODO(), "base/servmonitor", []string{"rpc.client", "rpc.test111"})
+	ass.Nil(err)
+
+	time.Sleep(5*time.Minute)
+}