[GitHub] asifdxtreme closed pull request #374: SCB-670 Check the etcd cluster health when new connection

[GitBox <gi...@apache.org>]

asifdxtreme closed pull request #374: SCB-670 Check the etcd cluster health when new connection
URL: https://github.com/apache/incubator-servicecomb-service-center/pull/374
 
 
   

This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:

As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):

diff --git a/pkg/util/concurrent_map.go b/pkg/util/concurrent_map.go
index b7ec0399..546ec47a 100644
--- a/pkg/util/concurrent_map.go
+++ b/pkg/util/concurrent_map.go
@@ -115,6 +115,6 @@ func (cm *ConcurrentMap) ForEach(f func(item MapItem) (next bool)) {
 	close(ch)
 }
 
-func NewConcurrentMap(size int) ConcurrentMap {
-	return ConcurrentMap{size: size}
+func NewConcurrentMap(size int) *ConcurrentMap {
+	return &ConcurrentMap{size: size}
 }
diff --git a/pkg/util/context.go b/pkg/util/context.go
index 4bc9b3a3..565cf697 100644
--- a/pkg/util/context.go
+++ b/pkg/util/context.go
@@ -24,7 +24,7 @@ import (
 
 type StringContext struct {
 	parentCtx context.Context
-	kv        ConcurrentMap
+	kv        *ConcurrentMap
 }
 
 func (c *StringContext) Deadline() (deadline time.Time, ok bool) {
diff --git a/server/bootstrap/bootstrap.go b/server/bootstrap/bootstrap.go
index 1bd7a5a7..7b59caf9 100644
--- a/server/bootstrap/bootstrap.go
+++ b/server/bootstrap/bootstrap.go
@@ -48,6 +48,9 @@ import _ "github.com/apache/incubator-servicecomb-service-center/server/govern"
 // module
 import _ "github.com/apache/incubator-servicecomb-service-center/server/broker"
 
+// metrics
+import _ "github.com/apache/incubator-servicecomb-service-center/server/metric"
+
 import (
 	"github.com/apache/incubator-servicecomb-service-center/pkg/util"
 	"github.com/apache/incubator-servicecomb-service-center/server/handler/auth"
diff --git a/server/core/backend/metric.go b/server/core/backend/metrics.go
similarity index 100%
rename from server/core/backend/metric.go
rename to server/core/backend/metrics.go
diff --git a/server/metric/common.go b/server/metric/common.go
new file mode 100644
index 00000000..5e423ada
--- /dev/null
+++ b/server/metric/common.go
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package metric
+
+import "time"
+
+const (
+	defaultMetricsSize = 100
+	collectInterval    = 5 * time.Second
+
+	familyName = "service_center_"
+)
+
+var sysMetricNames = map[string]struct{}{
+	"process_resident_memory_bytes": {},
+	"process_cpu_seconds_total":     {},
+	"go_threads":                    {},
+}
diff --git a/server/metric/gatherer.go b/server/metric/gatherer.go
new file mode 100644
index 00000000..7140909e
--- /dev/null
+++ b/server/metric/gatherer.go
@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package metric
+
+import (
+	"github.com/apache/incubator-servicecomb-service-center/pkg/util"
+	pm "github.com/prometheus/client_golang/prometheus"
+	"golang.org/x/net/context"
+	"strings"
+	"sync"
+	"time"
+)
+
+var Gatherer *MetricsGatherer
+
+func init() {
+	Gatherer = NewGatherer()
+	Gatherer.Start()
+}
+
+func NewGatherer() *MetricsGatherer {
+	return &MetricsGatherer{
+		Records: NewMetrics(),
+		closed:  true,
+	}
+}
+
+type MetricsGatherer struct {
+	Records *Metrics
+
+	lock   sync.Mutex
+	closed bool
+}
+
+func (mm *MetricsGatherer) Start() {
+	mm.lock.Lock()
+	if !mm.closed {
+		mm.lock.Unlock()
+		return
+	}
+	mm.closed = false
+
+	util.Go(mm.loop)
+
+	mm.lock.Unlock()
+}
+
+func (mm *MetricsGatherer) loop(ctx context.Context) {
+	ticker := time.NewTicker(collectInterval)
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			if err := mm.Collect(); err != nil {
+				util.Logger().Errorf(err, "metrics collect failed.")
+				return
+			}
+		}
+	}
+}
+
+func (mm *MetricsGatherer) Collect() error {
+	mfs, err := pm.DefaultGatherer.Gather()
+	if err != nil {
+		return err
+	}
+
+	for _, mf := range mfs {
+		name := mf.GetName()
+		if _, ok := sysMetricNames[name]; strings.Index(name, familyName) == 0 || ok {
+			mm.Records.Put(strings.TrimPrefix(name, familyName), util.MetricValueOf(mf))
+		}
+	}
+	return nil
+}
diff --git a/server/metric/metrics.go b/server/metric/metrics.go
new file mode 100644
index 00000000..d78412ae
--- /dev/null
+++ b/server/metric/metrics.go
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package metric
+
+import "github.com/apache/incubator-servicecomb-service-center/pkg/util"
+
+func NewMetrics() *Metrics {
+	return &Metrics{
+		ConcurrentMap: util.NewConcurrentMap(defaultMetricsSize),
+	}
+}
+
+type Metrics struct {
+	*util.ConcurrentMap
+}
+
+func (cm *Metrics) Put(key string, val float64) (old float64) {
+	old, _ = cm.ConcurrentMap.Put(key, val).(float64)
+	return
+}
+
+func (cm *Metrics) Get(key string) (val float64) {
+	if v, ok := cm.ConcurrentMap.Get(key); ok {
+		val, _ = v.(float64)
+	}
+	return
+}
+
+func (cm *Metrics) Remove(key string) (old float64) {
+	old, _ = cm.ConcurrentMap.Remove(key).(float64)
+	return
+}
+
+func (cm *Metrics) ForEach(f func(k string, v float64) (next bool)) {
+	cm.ConcurrentMap.ForEach(func(item util.MapItem) (next bool) {
+		k, _ := item.Key.(string)
+		v, _ := item.Value.(float64)
+		return f(k, v)
+	})
+}
diff --git a/server/plugin/infra/registry/etcd/etcd.go b/server/plugin/infra/registry/etcd/etcd.go
index ffe1367a..31eb3540 100644
--- a/server/plugin/infra/registry/etcd/etcd.go
+++ b/server/plugin/infra/registry/etcd/etcd.go
@@ -31,6 +31,7 @@ import (
 	"github.com/coreos/etcd/mvcc/mvccpb"
 	"golang.org/x/net/context"
 	"google.golang.org/grpc"
+	"net/url"
 	"strings"
 	"time"
 )
@@ -748,5 +749,32 @@ func newClient(endpoints []string) (*clientv3.Client, error) {
 	if err != nil {
 		return nil, err
 	}
+	if len(endpoints) == 1 {
+		return client, nil
+	}
+
+	ctx, _ := context.WithTimeout(client.Ctx(), healthCheckTimeout)
+	resp, err := client.MemberList(ctx)
+	if err != nil {
+		return nil, err
+	}
+epLoop:
+	for _, ep := range endpoints {
+		var cluster []string
+		for _, mem := range resp.Members {
+			for _, curl := range mem.ClientURLs {
+				u, err := url.Parse(curl)
+				if err != nil {
+					return nil, err
+				}
+				cluster = append(cluster, u.Host)
+				if u.Host == ep {
+					continue epLoop
+				}
+			}
+		}
+		// maybe endpoints = [domain A, domain B] or there are more than one cluster
+		return nil, fmt.Errorf("the etcd cluster endpoint list%v does not contain %s", cluster, ep)
+	}
 	return client, nil
 }
diff --git a/server/rest/metric.go b/server/rest/metrics.go
similarity index 100%
rename from server/rest/metric.go
rename to server/rest/metrics.go


 

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services