You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@servicecomb.apache.org by li...@apache.org on 2018/12/18 08:05:15 UTC
[servicecomb-service-center] branch master updated: SCB-1049 Alarm
center (#503)
This is an automated email from the ASF dual-hosted git repository.
littlecui pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/servicecomb-service-center.git
The following commit(s) were added to refs/heads/master by this push:
new 1ab21c8 SCB-1049 Alarm center (#503)
1ab21c8 is described below
commit 1ab21c8cd4c2afdea11f3650ebb0fe0800342134
Author: little-cui <su...@qq.com>
AuthorDate: Tue Dec 18 16:05:10 2018 +0800
SCB-1049 Alarm center (#503)
* SCB-1049 Alarm center
* SCB-1049 Add alarm model
* SCB-1049 Restructure
* SCB-1049 Add alarms APIs
* SCB-1049 Add alarm status
* SCB-1049 Add inner alarm
* SCB-1049 Health checker
* SCB-1049 call init when new notifyservice instance
---
pkg/client/sc/apis.go | 2 +-
.../service/notification => pkg/notify}/common.go | 44 +--------
.../service/notification => pkg/notify}/group.go | 4 +-
.../notification => pkg/notify}/group_test.go | 15 +--
.../service/notification => pkg/notify}/notice.go | 26 ++---
.../notify}/notification_healthchecker.go | 24 ++---
.../notify}/notification_service.go | 91 ++++++++---------
.../notify}/notification_test.go | 35 +++----
.../notification => pkg/notify}/processor.go | 49 ++++------
.../notification => pkg/notify}/processor_test.go | 29 +++---
.../service/notification => pkg/notify}/subject.go | 4 +-
.../notification => pkg/notify}/subject_test.go | 9 +-
.../notification => pkg/notify}/subscriber.go | 36 +++----
.../admin/model/cluster.go => pkg/notify/types.go | 48 +++++++--
.../model/cluster.go => pkg/notify/types_test.go | 29 +++---
pkg/util/{map.go => json.go} | 4 +
pkg/util/{map_test.go => json_test.go} | 2 +-
scctl/pkg/plugin/health/cmd.go | 2 +-
server/admin/controller_v4.go | 29 +++++-
server/admin/model/{cluster.go => types.go} | 16 +++
server/admin/service.go | 66 ++++++++++---
server/alarm/common.go | 80 +++++++++++++++
server/alarm/model/types.go | 62 ++++++++++++
server/alarm/service.go | 101 +++++++++++++++++++
server/core/config.go | 2 +-
server/core/swagger/v4.yaml | 75 +++++++++++++-
server/error/error.go | 108 +++++++++------------
.../{admin/model/cluster.go => health/health.go} | 32 ++++--
server/health/health_test.go | 54 +++++++++++
.../{admin/model/cluster.go => notify/center.go} | 16 +--
.../{admin/model/cluster.go => notify/common.go} | 20 ++--
.../notification => notify}/listwatcher.go | 61 ++++++------
.../{service/notification => notify}/publisher.go | 2 +-
server/{service/notification => notify}/stream.go | 24 ++++-
.../notification => notify}/stream_test.go | 24 +++--
.../{service/notification => notify}/websocket.go | 16 +--
.../notification => notify}/websocket_test.go | 54 +++++------
server/plugin/pkg/discovery/etcd/cacher_kv.go | 2 +-
.../pkg/discovery/k8s/adaptor/kube_client.go | 14 ++-
.../plugin/pkg/discovery/servicecenter/syncer.go | 8 ++
server/plugin/pkg/registry/etcd/etcd.go | 18 +++-
server/rest/controller/rest_util.go | 5 +
server/server.go | 62 ++++++------
server/service/event/instance_event_handler.go | 23 +++--
server/service/event/rule_event_handler.go | 4 +-
server/service/event/service_event_handler.go | 11 ++-
server/service/event/tag_event_handler.go | 4 +-
server/service/instance.go | 7 ++
server/service/metrics/metrics.go | 12 +--
server/service/watch.go | 22 ++---
50 files changed, 973 insertions(+), 514 deletions(-)
diff --git a/pkg/client/sc/apis.go b/pkg/client/sc/apis.go
index 3fb8f1a..aa5d951 100644
--- a/pkg/client/sc/apis.go
+++ b/pkg/client/sc/apis.go
@@ -213,7 +213,7 @@ func (c *SCClient) HealthCheck(ctx context.Context) *scerr.Error {
headers.Set("X-Domain-Name", "default")
resp, err := c.RestDoWithContext(ctx, http.MethodGet, apiHealthURL, headers, nil)
if err != nil {
- return scerr.NewError(scerr.ErrUnavailableBackend, err.Error())
+ return scerr.NewError(scerr.ErrInternal, err.Error())
}
defer resp.Body.Close()
diff --git a/server/service/notification/common.go b/pkg/notify/common.go
similarity index 52%
rename from server/service/notification/common.go
rename to pkg/notify/common.go
index 1617f6f..5217aab 100644
--- a/server/service/notification/common.go
+++ b/pkg/notify/common.go
@@ -14,50 +14,12 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package notification
-
-import (
- "strconv"
- "time"
-)
+package notify
const (
- DEFAULT_MAX_QUEUE = 1000
- DEFAULT_ADD_JOB_TIMEOUT = 1 * time.Second
- DEFAULT_SEND_TIMEOUT = 5 * time.Second
- DEFAULT_HEARTBEAT_INTERVAL = 30 * time.Second
+ DefaultQueueSize = 1000
)
const (
- NOTIFTY NotifyType = iota
- INSTANCE
- typeEnd
+ NOTIFTY Type = iota
)
-
-type NotifyType int
-
-func (nt NotifyType) String() string {
- if int(nt) < len(notifyTypeNames) {
- return notifyTypeNames[nt]
- }
- return "NotifyType" + strconv.Itoa(int(nt))
-}
-
-func (nt NotifyType) QueueSize() (s int) {
- if int(nt) < len(notifyTypeQueues) {
- s = notifyTypeQueues[nt]
- }
- if s <= 0 {
- s = DEFAULT_MAX_QUEUE
- }
- return
-}
-
-var notifyTypeNames = []string{
- NOTIFTY: "NOTIFTY",
- INSTANCE: "INSTANCE",
-}
-
-var notifyTypeQueues = []int{
- INSTANCE: 100 * 1000,
-}
diff --git a/server/service/notification/group.go b/pkg/notify/group.go
similarity index 96%
rename from server/service/notification/group.go
rename to pkg/notify/group.go
index 8c677d4..58f39a0 100644
--- a/server/service/notification/group.go
+++ b/pkg/notify/group.go
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package notification
+package notify
import (
"github.com/apache/servicecomb-service-center/pkg/util"
@@ -29,7 +29,7 @@ func (g *Group) Name() string {
return g.name
}
-func (g *Group) Notify(job NotifyJob) {
+func (g *Group) Notify(job Event) {
g.subscribers.ForEach(func(item util.MapItem) (next bool) {
item.Value.(Subscriber).OnMessage(job)
return true
diff --git a/server/service/notification/group_test.go b/pkg/notify/group_test.go
similarity index 86%
rename from server/service/notification/group_test.go
rename to pkg/notify/group_test.go
index b241eb4..44b185d 100644
--- a/server/service/notification/group_test.go
+++ b/pkg/notify/group_test.go
@@ -14,20 +14,21 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package notification
+package notify
import "testing"
type mockSubscriber struct {
- *BaseSubscriber
- job NotifyJob
+ Subscriber
+ job Event
}
-func (s *mockSubscriber) OnMessage(job NotifyJob) {
+func (s *mockSubscriber) OnMessage(job Event) {
s.job = job
}
func TestGroup_Add(t *testing.T) {
+ INSTANCE := RegisterType("INSTANCE", 1)
m := NewSubscriber(INSTANCE, "s1", "g1")
g := NewGroup("g1")
if g.Name() != "g1" {
@@ -39,7 +40,7 @@ func TestGroup_Add(t *testing.T) {
if g.AddSubscriber(NewSubscriber(INSTANCE, "s1", "g1")) == m {
t.Fatalf("TestGroup_Add failed")
}
- same := *m
+ same := *(m.(*baseSubscriber))
if g.AddSubscriber(&same) != m {
t.Fatalf("TestGroup_Add failed")
}
@@ -54,14 +55,14 @@ func TestGroup_Add(t *testing.T) {
t.Fatalf("TestGroup_Add failed")
}
- mock := &mockSubscriber{BaseSubscriber: NewSubscriber(INSTANCE, "s1", "g1")}
+ mock := &mockSubscriber{Subscriber: NewSubscriber(INSTANCE, "s1", "g1")}
if g.AddSubscriber(mock) != mock {
t.Fatalf("TestGroup_Add failed")
}
if g.Subscribers(mock.Id()) != mock {
t.Fatalf("TestGroup_Add failed")
}
- job := &BaseNotifyJob{nType: INSTANCE}
+ job := &baseEvent{nType: INSTANCE}
g.Notify(job)
if mock.job != job {
t.Fatalf("TestGroup_Add failed")
diff --git a/server/service/notification/notice.go b/pkg/notify/notice.go
similarity index 69%
rename from server/service/notification/notice.go
rename to pkg/notify/notice.go
index 26d5235..836d47b 100644
--- a/server/service/notification/notice.go
+++ b/pkg/notify/notice.go
@@ -14,28 +14,32 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package notification
+package notify
-type NotifyJob interface {
- Type() NotifyType
- Group() string
- Subject() string
+type Event interface {
+ Type() Type
+ Subject() string // required!
+ Group() string // broadcast all the subscriber of the same subject if group is empty
}
-type BaseNotifyJob struct {
- nType NotifyType
+type baseEvent struct {
+ nType Type
subject string
group string
}
-func (s *BaseNotifyJob) Type() NotifyType {
+func (s *baseEvent) Type() Type {
return s.nType
}
-func (s *BaseNotifyJob) Group() string {
+func (s *baseEvent) Subject() string {
+ return s.subject
+}
+
+func (s *baseEvent) Group() string {
return s.group
}
-func (s *BaseNotifyJob) Subject() string {
- return s.subject
+func NewEvent(t Type, s string, g string) Event {
+ return &baseEvent{t, s, g}
}
diff --git a/server/service/notification/notification_healthchecker.go b/pkg/notify/notification_healthchecker.go
similarity index 76%
rename from server/service/notification/notification_healthchecker.go
rename to pkg/notify/notification_healthchecker.go
index e529314..8cc41cd 100644
--- a/server/service/notification/notification_healthchecker.go
+++ b/pkg/notify/notification_healthchecker.go
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package notification
+package notify
import "github.com/apache/servicecomb-service-center/pkg/log"
@@ -25,15 +25,15 @@ const (
//Notifier 健康检查
type NotifyServiceHealthChecker struct {
- BaseSubscriber
+ Subscriber
}
type NotifyServiceHealthCheckJob struct {
- *BaseNotifyJob
+ Event
ErrorSubscriber Subscriber
}
-func (s *NotifyServiceHealthChecker) OnMessage(job NotifyJob) {
+func (s *NotifyServiceHealthChecker) OnMessage(job Event) {
j := job.(*NotifyServiceHealthCheckJob)
err := j.ErrorSubscriber.Err()
@@ -43,28 +43,20 @@ func (s *NotifyServiceHealthChecker) OnMessage(job NotifyJob) {
return
}
- log.Debugf("notification service remove %s watcher, error: %s, subject: %s, group: %s",
- j.ErrorSubscriber.Type(), err.Error(), j.ErrorSubscriber.Subject(), j.ErrorSubscriber.Group())
+ log.Debugf("notification service remove %s watcher, error: %v, subject: %s, group: %s",
+ j.ErrorSubscriber.Type(), err, j.ErrorSubscriber.Subject(), j.ErrorSubscriber.Group())
s.Service().RemoveSubscriber(j.ErrorSubscriber)
}
func NewNotifyServiceHealthChecker() *NotifyServiceHealthChecker {
return &NotifyServiceHealthChecker{
- BaseSubscriber: BaseSubscriber{
- group: NOTIFY_SERVER_CHECKER_NAME,
- subject: NOTIFY_SERVER_CHECK_SUBJECT,
- nType: NOTIFTY,
- },
+ Subscriber: NewSubscriber(NOTIFTY, NOTIFY_SERVER_CHECK_SUBJECT, NOTIFY_SERVER_CHECKER_NAME),
}
}
func NewNotifyServiceHealthCheckJob(s Subscriber) *NotifyServiceHealthCheckJob {
return &NotifyServiceHealthCheckJob{
- BaseNotifyJob: &BaseNotifyJob{
- group: NOTIFY_SERVER_CHECKER_NAME,
- subject: NOTIFY_SERVER_CHECK_SUBJECT,
- nType: NOTIFTY,
- },
+ Event: NewEvent(NOTIFTY, NOTIFY_SERVER_CHECK_SUBJECT, NOTIFY_SERVER_CHECKER_NAME),
ErrorSubscriber: s,
}
}
diff --git a/server/service/notification/notification_service.go b/pkg/notify/notification_service.go
similarity index 59%
rename from server/service/notification/notification_service.go
rename to pkg/notify/notification_service.go
index 9c81401..c2edd24 100644
--- a/server/service/notification/notification_service.go
+++ b/pkg/notify/notification_service.go
@@ -14,43 +14,23 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package notification
+package notify
import (
"errors"
- "github.com/apache/servicecomb-service-center/pkg/gopool"
"github.com/apache/servicecomb-service-center/pkg/log"
- "github.com/apache/servicecomb-service-center/pkg/util"
- "golang.org/x/net/context"
"sync"
)
-var notifyService *NotifyService
-
-func init() {
- notifyService = &NotifyService{
- isClose: true,
- goroutine: gopool.New(context.Background()),
- }
-}
-
type NotifyService struct {
- processors *util.ConcurrentMap
- goroutine *gopool.Pool
- err chan error
+ processors map[Type]*Processor
closeMux sync.RWMutex
isClose bool
}
-func (s *NotifyService) Err() <-chan error {
- return s.err
-}
-
func (s *NotifyService) init() {
- s.processors = util.NewConcurrentMap(int(typeEnd))
- s.err = make(chan error, 1)
- for i := NotifyType(0); i != typeEnd; i++ {
- s.processors.Put(i, NewProcessor(i.String(), i.QueueSize()))
+ for _, t := range Types() {
+ s.processors[t] = NewProcessor(t.String(), t.QueueSize())
}
}
@@ -63,62 +43,68 @@ func (s *NotifyService) Start() {
s.isClose = false
s.closeMux.Unlock()
- s.init()
// 错误subscriber清理
s.AddSubscriber(NewNotifyServiceHealthChecker())
- log.Debugf("notify service is started")
+ s.startProcessors()
- s.processors.ForEach(func(item util.MapItem) (next bool) {
- s.goroutine.Do(item.Value.(*Processor).Do)
- return true
- })
+ log.Debugf("notify service is started")
}
func (s *NotifyService) AddSubscriber(n Subscriber) error {
- if s.Closed() {
- return errors.New("server is shutting down")
+ if n == nil {
+ err := errors.New("required Subscriber")
+ log.Errorf(err, "add subscriber failed")
+ return err
}
- itf, ok := s.processors.Get(n.Type())
+ p, ok := s.processors[n.Type()]
if !ok {
- return errors.New("Unknown subscribe type")
+ err := errors.New("unknown subscribe type")
+ log.Errorf(err, "add %s subscriber[%s/%s] failed", n.Type(), n.Subject(), n.Group())
+ return err
}
n.SetService(s)
n.OnAccept()
- itf.(*Processor).AddSubscriber(n)
+ p.AddSubscriber(n)
return nil
}
func (s *NotifyService) RemoveSubscriber(n Subscriber) {
- itf, ok := s.processors.Get(n.Type())
+ p, ok := s.processors[n.Type()]
if !ok {
return
}
- itf.(*Processor).Remove(n)
+ p.Remove(n)
n.Close()
}
-func (s *NotifyService) RemoveAllSubscribers() {
- s.processors.ForEach(func(item util.MapItem) (next bool) {
- item.Value.(*Processor).Clear()
- return true
- })
+func (s *NotifyService) startProcessors() {
+ for _, p := range s.processors {
+ p.Run()
+ }
+}
+
+func (s *NotifyService) stopProcessors() {
+ for _, p := range s.processors {
+ p.Clear()
+ p.Stop()
+ }
}
//通知内容塞到队列里
-func (s *NotifyService) AddJob(job NotifyJob) error {
+func (s *NotifyService) Publish(job Event) error {
if s.Closed() {
return errors.New("add notify job failed for server shutdown")
}
- itf, ok := s.processors.Get(job.Type())
+ p, ok := s.processors[job.Type()]
if !ok {
return errors.New("Unknown job type")
}
- itf.(*Processor).Accept(job)
+ p.Accept(job)
return nil
}
@@ -137,15 +123,16 @@ func (s *NotifyService) Stop() {
s.isClose = true
s.closeMux.Unlock()
- s.goroutine.Close(true)
-
- s.RemoveAllSubscribers()
-
- close(s.err)
+ s.stopProcessors()
log.Debug("notify service stopped")
}
-func GetNotifyService() *NotifyService {
- return notifyService
+func NewNotifyService() *NotifyService {
+ ns := &NotifyService{
+ processors: make(map[Type]*Processor),
+ isClose: true,
+ }
+ ns.init()
+ return ns
}
diff --git a/server/service/notification/notification_test.go b/pkg/notify/notification_test.go
similarity index 72%
rename from server/service/notification/notification_test.go
rename to pkg/notify/notification_test.go
index 9a05f29..f130c27 100644
--- a/server/service/notification/notification_test.go
+++ b/pkg/notify/notification_test.go
@@ -14,25 +14,17 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package notification
+package notify
import (
- "github.com/apache/servicecomb-service-center/pkg/gopool"
- "golang.org/x/net/context"
"testing"
"time"
)
func TestGetNotifyService(t *testing.T) {
- n := NotifyType(999)
- if n.String() != "NotifyType999" {
- t.Fatalf("TestGetNotifyService failed")
- }
+ INSTANCE := RegisterType("INSTANCE", 1)
- notifyService := &NotifyService{
- isClose: true,
- goroutine: gopool.New(context.Background()),
- }
+ notifyService := NewNotifyService()
if notifyService == nil {
t.Fatalf("TestGetNotifyService failed")
}
@@ -44,7 +36,7 @@ func TestGetNotifyService(t *testing.T) {
if err == nil {
t.Fatalf("TestGetNotifyService failed")
}
- err = notifyService.AddJob(nil)
+ err = notifyService.Publish(nil)
if err == nil {
t.Fatalf("TestGetNotifyService failed")
}
@@ -54,36 +46,33 @@ func TestGetNotifyService(t *testing.T) {
if notifyService.Closed() != false {
t.Fatalf("TestGetNotifyService failed")
}
- select {
- case <-notifyService.Err():
- t.Fatalf("TestGetNotifyService failed")
- default:
- }
s := NewSubscriber(-1, "s", "g")
err = notifyService.AddSubscriber(s)
if err == nil {
t.Fatalf("TestGetNotifyService failed")
}
+ notifyService.RemoveSubscriber(s)
+
s = NewSubscriber(INSTANCE, "s", "g")
err = notifyService.AddSubscriber(s)
if err != nil {
- t.Fatalf("TestGetNotifyService failed")
+ t.Fatalf("TestGetNotifyService failed, %v", err)
}
- j := &BaseNotifyJob{INSTANCE, "s", "g"}
- err = notifyService.AddJob(j)
+ j := &baseEvent{INSTANCE, "s", "g"}
+ err = notifyService.Publish(j)
if err != nil {
t.Fatalf("TestGetNotifyService failed")
}
- err = notifyService.AddJob(NewNotifyServiceHealthCheckJob(NewNotifyServiceHealthChecker()))
+ err = notifyService.Publish(NewNotifyServiceHealthCheckJob(NewNotifyServiceHealthChecker()))
if err != nil {
t.Fatalf("TestGetNotifyService failed")
}
- <-time.After(time.Second)
- err = notifyService.AddJob(NewNotifyServiceHealthCheckJob(s))
+ err = notifyService.Publish(NewNotifyServiceHealthCheckJob(s))
if err != nil {
t.Fatalf("TestGetNotifyService failed")
}
+ <-time.After(time.Second)
notifyService.Stop()
if notifyService.Closed() != true {
t.Fatalf("TestGetNotifyService failed")
diff --git a/server/service/notification/processor.go b/pkg/notify/processor.go
similarity index 76%
rename from server/service/notification/processor.go
rename to pkg/notify/processor.go
index 17528a8..9df5288 100644
--- a/server/service/notification/processor.go
+++ b/pkg/notify/processor.go
@@ -14,25 +14,35 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package notification
+package notify
import (
- "github.com/apache/servicecomb-service-center/pkg/log"
+ "github.com/apache/servicecomb-service-center/pkg/queue"
"github.com/apache/servicecomb-service-center/pkg/util"
"golang.org/x/net/context"
)
type Processor struct {
+ *queue.TaskQueue
+
name string
subjects *util.ConcurrentMap
- queue chan NotifyJob
+ queue chan Event
}
func (p *Processor) Name() string {
return p.name
}
-func (p *Processor) Notify(job NotifyJob) {
+func (p *Processor) Accept(job Event) {
+ p.Add(queue.Task{Object: job})
+}
+
+func (p *Processor) Handle(ctx context.Context, obj interface{}) {
+ p.Notify(obj.(Event))
+}
+
+func (p *Processor) Notify(job Event) {
if itf, ok := p.subjects.Get(job.Subject()); ok {
itf.(*Subject).Notify(job)
}
@@ -79,29 +89,12 @@ func (p *Processor) Clear() {
p.subjects.Clear()
}
-func (p *Processor) Accept(job NotifyJob) {
- defer log.Recover()
- p.queue <- job
-}
-
-func (p *Processor) Do(ctx context.Context) {
- for {
- select {
- case <-ctx.Done():
- return
- case job, ok := <-p.queue:
- if !ok {
- return
- }
- p.Notify(job)
- }
- }
-}
-
-func NewProcessor(name string, queue int) *Processor {
- return &Processor{
- name: name,
- subjects: util.NewConcurrentMap(0),
- queue: make(chan NotifyJob, queue),
+func NewProcessor(name string, queueSize int) *Processor {
+ p := &Processor{
+ TaskQueue: queue.NewTaskQueue(queueSize),
+ name: name,
+ subjects: util.NewConcurrentMap(0),
}
+ p.AddWorker(p)
+ return p
}
diff --git a/server/service/notification/processor_test.go b/pkg/notify/processor_test.go
similarity index 82%
rename from server/service/notification/processor_test.go
rename to pkg/notify/processor_test.go
index 7b6b748..3505457 100644
--- a/server/service/notification/processor_test.go
+++ b/pkg/notify/processor_test.go
@@ -14,31 +14,30 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package notification
+package notify
import (
- "github.com/apache/servicecomb-service-center/pkg/gopool"
"testing"
"time"
)
type mockSubscriberChan struct {
- *BaseSubscriber
- job chan NotifyJob
+ Subscriber
+ job chan Event
}
-func (s *mockSubscriberChan) OnMessage(job NotifyJob) {
+func (s *mockSubscriberChan) OnMessage(job Event) {
s.job <- job
}
func TestProcessor_Do(t *testing.T) {
+ INSTANCE := RegisterType("INSTANCE", 1)
delay := 50 * time.Millisecond
- mock1 := &mockSubscriberChan{BaseSubscriber: NewSubscriber(INSTANCE, "s1", "g1"),
- job: make(chan NotifyJob, 1)}
- mock2 := &mockSubscriberChan{BaseSubscriber: NewSubscriber(INSTANCE, "s1", "g2"),
- job: make(chan NotifyJob, 1)}
+ mock1 := &mockSubscriberChan{Subscriber: NewSubscriber(INSTANCE, "s1", "g1"),
+ job: make(chan Event, 1)}
+ mock2 := &mockSubscriberChan{Subscriber: NewSubscriber(INSTANCE, "s1", "g2"),
+ job: make(chan Event, 1)}
p := NewProcessor("p1", 0)
- gopool.Go(p.Do)
if p.Name() != "p1" {
t.Fatalf("TestProcessor_Do")
}
@@ -62,8 +61,8 @@ func TestProcessor_Do(t *testing.T) {
}
p.AddSubscriber(mock1)
p.AddSubscriber(mock2)
- job := &BaseNotifyJob{group: "g1"}
- p.Accept(job)
+ job := &baseEvent{group: "g1"}
+ p.Handle(nil, job)
select {
case <-mock1.job:
t.Fatalf("TestProcessor_Do")
@@ -71,7 +70,7 @@ func TestProcessor_Do(t *testing.T) {
}
job.subject = "s1"
job.group = "g3"
- p.Accept(job)
+ p.Handle(nil, job)
select {
case <-mock1.job:
t.Fatalf("TestProcessor_Do")
@@ -79,7 +78,7 @@ func TestProcessor_Do(t *testing.T) {
}
job.subject = "s1"
job.group = "g1"
- p.Accept(job)
+ p.Handle(nil, job)
select {
case j := <-mock1.job:
if j != job {
@@ -95,7 +94,7 @@ func TestProcessor_Do(t *testing.T) {
}
job.subject = "s1"
job.group = ""
- p.Accept(job)
+ p.Handle(nil, job)
select {
case j := <-mock1.job:
if j != job {
diff --git a/server/service/notification/subject.go b/pkg/notify/subject.go
similarity index 96%
rename from server/service/notification/subject.go
rename to pkg/notify/subject.go
index cdc98f6..f052801 100644
--- a/server/service/notification/subject.go
+++ b/pkg/notify/subject.go
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package notification
+package notify
import (
"github.com/apache/servicecomb-service-center/pkg/util"
@@ -29,7 +29,7 @@ func (s *Subject) Name() string {
return s.name
}
-func (s *Subject) Notify(job NotifyJob) {
+func (s *Subject) Notify(job Event) {
if len(job.Group()) == 0 {
s.groups.ForEach(func(item util.MapItem) (next bool) {
item.Value.(*Group).Notify(job)
diff --git a/server/service/notification/subject_test.go b/pkg/notify/subject_test.go
similarity index 88%
rename from server/service/notification/subject_test.go
rename to pkg/notify/subject_test.go
index 293c500..c9e955c 100644
--- a/server/service/notification/subject_test.go
+++ b/pkg/notify/subject_test.go
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package notification
+package notify
import "testing"
@@ -44,10 +44,11 @@ func TestSubject_Fetch(t *testing.T) {
if s.Size() != 1 {
t.Fatalf("TestSubject_Fetch failed")
}
- mock1 := &mockSubscriber{BaseSubscriber: NewSubscriber(INSTANCE, "s1", "g1")}
- mock2 := &mockSubscriber{BaseSubscriber: NewSubscriber(INSTANCE, "s1", "g2")}
+ INSTANCE := RegisterType("INSTANCE", 1)
+ mock1 := &mockSubscriber{Subscriber: NewSubscriber(INSTANCE, "s1", "g1")}
+ mock2 := &mockSubscriber{Subscriber: NewSubscriber(INSTANCE, "s1", "g2")}
g.AddSubscriber(mock1)
- job := &BaseNotifyJob{group: "g3"}
+ job := &baseEvent{group: "g3"}
s.Notify(job)
if mock1.job != nil || mock2.job != nil {
t.Fatalf("TestSubject_Fetch failed")
diff --git a/server/service/notification/subscriber.go b/pkg/notify/subscriber.go
similarity index 62%
rename from server/service/notification/subscriber.go
rename to pkg/notify/subscriber.go
index 24ec5db..23d848c 100644
--- a/server/service/notification/subscriber.go
+++ b/pkg/notify/subscriber.go
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package notification
+package notify
import (
"errors"
@@ -25,7 +25,7 @@ type Subscriber interface {
Id() string
Subject() string
Group() string
- Type() NotifyType
+ Type() Type
Service() *NotifyService
SetService(*NotifyService)
@@ -35,34 +35,34 @@ type Subscriber interface {
Close()
OnAccept()
// The event bus will callback this function, so it must be non-blocked.
- OnMessage(job NotifyJob)
+ OnMessage(Event)
}
-type BaseSubscriber struct {
+type baseSubscriber struct {
+ nType Type
id string
subject string
group string
- nType NotifyType
service *NotifyService
err error
}
-func (s *BaseSubscriber) Id() string { return s.id }
-func (s *BaseSubscriber) Subject() string { return s.subject }
-func (s *BaseSubscriber) Group() string { return s.group }
-func (s *BaseSubscriber) Type() NotifyType { return s.nType }
-func (s *BaseSubscriber) Service() *NotifyService { return s.service }
-func (s *BaseSubscriber) SetService(svc *NotifyService) { s.service = svc }
-func (s *BaseSubscriber) Err() error { return s.err }
-func (s *BaseSubscriber) SetError(err error) { s.err = err }
-func (s *BaseSubscriber) Close() {}
-func (s *BaseSubscriber) OnAccept() {}
-func (s *BaseSubscriber) OnMessage(job NotifyJob) {
+func (s *baseSubscriber) Id() string { return s.id }
+func (s *baseSubscriber) Subject() string { return s.subject }
+func (s *baseSubscriber) Group() string { return s.group }
+func (s *baseSubscriber) Type() Type { return s.nType }
+func (s *baseSubscriber) Service() *NotifyService { return s.service }
+func (s *baseSubscriber) SetService(svc *NotifyService) { s.service = svc }
+func (s *baseSubscriber) Err() error { return s.err }
+func (s *baseSubscriber) SetError(err error) { s.err = err }
+func (s *baseSubscriber) Close() {}
+func (s *baseSubscriber) OnAccept() {}
+func (s *baseSubscriber) OnMessage(job Event) {
s.SetError(errors.New("do not call base notifier OnMessage method"))
}
-func NewSubscriber(nType NotifyType, subject, group string) *BaseSubscriber {
- return &BaseSubscriber{
+func NewSubscriber(nType Type, subject, group string) Subscriber {
+ return &baseSubscriber{
id: util.GenerateUuid(),
group: group,
subject: subject,
diff --git a/server/admin/model/cluster.go b/pkg/notify/types.go
similarity index 54%
copy from server/admin/model/cluster.go
copy to pkg/notify/types.go
index 6cabae9..829eea5 100644
--- a/server/admin/model/cluster.go
+++ b/pkg/notify/types.go
@@ -13,17 +13,47 @@
// See the License for the specific language governing permissions and
// limitations under the License.
-package model
+package notify
-import (
- pb "github.com/apache/servicecomb-service-center/server/core/proto"
- "github.com/apache/servicecomb-service-center/server/plugin/pkg/registry"
-)
+import "strconv"
-type ClustersRequest struct {
+type Type int
+
+func (nt Type) String() string {
+ if int(nt) < len(typeNames) {
+ return typeNames[nt]
+ }
+ return "Type" + strconv.Itoa(int(nt))
+}
+
+func (nt Type) QueueSize() (s int) {
+ if int(nt) < len(typeQueues) {
+ s = typeQueues[nt]
+ }
+ if s <= 0 {
+ s = DefaultQueueSize
+ }
+ return
+}
+
+var typeNames = []string{
+ NOTIFTY: "NOTIFTY",
+}
+
+var typeQueues = []int{
+ NOTIFTY: 0,
+}
+
+func Types() (ts []Type) {
+ for i := range typeNames {
+ ts = append(ts, Type(i))
+ }
+ return
}
-type ClustersResponse struct {
- Response *pb.Response `json:"response,omitempty"`
- Clusters registry.Clusters `json:"clusters,omitempty"`
+func RegisterType(name string, size int) Type {
+ l := len(typeNames)
+ typeNames = append(typeNames, name)
+ typeQueues = append(typeQueues, size)
+ return Type(l)
}
diff --git a/server/admin/model/cluster.go b/pkg/notify/types_test.go
similarity index 54%
copy from server/admin/model/cluster.go
copy to pkg/notify/types_test.go
index 6cabae9..720af81 100644
--- a/server/admin/model/cluster.go
+++ b/pkg/notify/types_test.go
@@ -13,17 +13,24 @@
// See the License for the specific language governing permissions and
// limitations under the License.
-package model
+package notify
-import (
- pb "github.com/apache/servicecomb-service-center/server/core/proto"
- "github.com/apache/servicecomb-service-center/server/plugin/pkg/registry"
-)
+import "testing"
-type ClustersRequest struct {
-}
-
-type ClustersResponse struct {
- Response *pb.Response `json:"response,omitempty"`
- Clusters registry.Clusters `json:"clusters,omitempty"`
+func TestRegisterType(t *testing.T) {
+ id := RegisterType("a", 0)
+ if id.String() != "a" || id.QueueSize() != DefaultQueueSize {
+ t.Fatal("TestRegisterType failed", id.String(), id.QueueSize())
+ }
+ id = RegisterType("b", 1)
+ if id.String() != "b" || id.QueueSize() != 1 {
+ t.Fatal("TestRegisterType failed", id.String(), id.QueueSize())
+ }
+ id = Type(999)
+ if id.String() != "Type999" || id.QueueSize() != DefaultQueueSize {
+ t.Fatal("TestRegisterType failed", id.String(), id.QueueSize())
+ }
+ if NOTIFTY.String() != "NOTIFTY" || NOTIFTY.QueueSize() != DefaultQueueSize {
+ t.Fatal("TestRegisterType failed", id.String(), id.QueueSize())
+ }
}
diff --git a/pkg/util/map.go b/pkg/util/json.go
similarity index 97%
rename from pkg/util/map.go
rename to pkg/util/json.go
index f5d8597..7f7dbb1 100644
--- a/pkg/util/map.go
+++ b/pkg/util/json.go
@@ -74,3 +74,7 @@ func toString(v interface{}) string {
return fmt.Sprintf("%#v", v)
}
}
+
+func NewJSONObject() JSONObject {
+ return make(JSONObject)
+}
diff --git a/pkg/util/map_test.go b/pkg/util/json_test.go
similarity index 99%
rename from pkg/util/map_test.go
rename to pkg/util/json_test.go
index 2e4affe..70f6461 100644
--- a/pkg/util/map_test.go
+++ b/pkg/util/json_test.go
@@ -23,7 +23,7 @@ import (
func TestNewServerInformation(t *testing.T) {
var c JSONObject
- c = make(JSONObject)
+ c = NewJSONObject()
if !c.Bool("a", true) {
t.Fatalf("TestNewServerInformation failed")
}
diff --git a/scctl/pkg/plugin/health/cmd.go b/scctl/pkg/plugin/health/cmd.go
index c9e5c31..befbbce 100644
--- a/scctl/pkg/plugin/health/cmd.go
+++ b/scctl/pkg/plugin/health/cmd.go
@@ -52,7 +52,7 @@ func HealthCommandFunc(_ *cobra.Command, args []string) {
scErr := scClient.HealthCheck(context.Background())
if scErr != nil {
switch scErr.Code {
- case scerr.ErrUnavailableBackend:
+ case scerr.ErrInternal:
cmd.StopAndExit(ExistUnavailable, scErr)
default:
cmd.StopAndExit(ExistAbnormal, scErr)
diff --git a/server/admin/controller_v4.go b/server/admin/controller_v4.go
index 55d23e9..af620f4 100644
--- a/server/admin/controller_v4.go
+++ b/server/admin/controller_v4.go
@@ -22,6 +22,7 @@ import (
"github.com/apache/servicecomb-service-center/pkg/rest"
"github.com/apache/servicecomb-service-center/server/admin/model"
"github.com/apache/servicecomb-service-center/server/rest/controller"
+ "strings"
)
// AdminService 治理相关接口服务
@@ -31,13 +32,22 @@ type AdminServiceControllerV4 struct {
// URLPatterns 路由
func (ctrl *AdminServiceControllerV4) URLPatterns() []rest.Route {
return []rest.Route{
+ {rest.HTTP_METHOD_GET, "/v4/:project/admin/alarms", ctrl.AlarmList},
+ {rest.HTTP_METHOD_DELETE, "/v4/:project/admin/alarms", ctrl.ClearAlarm},
{rest.HTTP_METHOD_GET, "/v4/:project/admin/dump", ctrl.Dump},
{rest.HTTP_METHOD_GET, "/v4/:project/admin/clusters", ctrl.Clusters},
}
}
func (ctrl *AdminServiceControllerV4) Dump(w http.ResponseWriter, r *http.Request) {
- request := &model.DumpRequest{}
+ query := r.URL.Query()
+ var options []string
+ if s := strings.TrimSpace(query.Get("options")); len(s) > 0 {
+ options = strings.Split(s, ",")
+ }
+ request := &model.DumpRequest{
+ Options: options,
+ }
ctx := r.Context()
resp, _ := AdminServiceAPI.Dump(ctx, request)
@@ -55,3 +65,20 @@ func (ctrl *AdminServiceControllerV4) Clusters(w http.ResponseWriter, r *http.Re
resp.Response = nil
controller.WriteResponse(w, respInternal, resp)
}
+
+func (ctrl *AdminServiceControllerV4) AlarmList(w http.ResponseWriter, r *http.Request) {
+ request := &model.AlarmListRequest{}
+ ctx := r.Context()
+ resp, _ := AdminServiceAPI.AlarmList(ctx, request)
+
+ respInternal := resp.Response
+ resp.Response = nil
+ controller.WriteResponse(w, respInternal, resp)
+}
+
+func (ctrl *AdminServiceControllerV4) ClearAlarm(w http.ResponseWriter, r *http.Request) {
+ request := &model.ClearAlarmRequest{}
+ ctx := r.Context()
+ resp, _ := AdminServiceAPI.ClearAlarm(ctx, request)
+ controller.WriteResponse(w, resp.Response, nil)
+}
diff --git a/server/admin/model/cluster.go b/server/admin/model/types.go
similarity index 75%
copy from server/admin/model/cluster.go
copy to server/admin/model/types.go
index 6cabae9..2dc1ccb 100644
--- a/server/admin/model/cluster.go
+++ b/server/admin/model/types.go
@@ -16,10 +16,19 @@
package model
import (
+ "github.com/apache/servicecomb-service-center/server/alarm/model"
pb "github.com/apache/servicecomb-service-center/server/core/proto"
"github.com/apache/servicecomb-service-center/server/plugin/pkg/registry"
)
+type AlarmListRequest struct {
+}
+
+type AlarmListResponse struct {
+ Response *pb.Response `json:"response,omitempty"`
+ Alarms []*model.AlarmEvent `json:"alarms,omitempty"`
+}
+
type ClustersRequest struct {
}
@@ -27,3 +36,10 @@ type ClustersResponse struct {
Response *pb.Response `json:"response,omitempty"`
Clusters registry.Clusters `json:"clusters,omitempty"`
}
+
+type ClearAlarmRequest struct {
+}
+
+type ClearAlarmResponse struct {
+ Response *pb.Response `json:"response,omitempty"`
+}
diff --git a/server/admin/service.go b/server/admin/service.go
index f6c9e7e..21715fb 100644
--- a/server/admin/service.go
+++ b/server/admin/service.go
@@ -18,8 +18,10 @@ package admin
import (
"github.com/apache/servicecomb-service-center/pkg/gopool"
+ "github.com/apache/servicecomb-service-center/pkg/log"
"github.com/apache/servicecomb-service-center/pkg/util"
"github.com/apache/servicecomb-service-center/server/admin/model"
+ "github.com/apache/servicecomb-service-center/server/alarm"
"github.com/apache/servicecomb-service-center/server/core"
"github.com/apache/servicecomb-service-center/server/core/backend"
pb "github.com/apache/servicecomb-service-center/server/core/proto"
@@ -59,9 +61,7 @@ type AdminService struct {
}
func (service *AdminService) Dump(ctx context.Context, in *model.DumpRequest) (*model.DumpResponse, error) {
-
domainProject := util.ParseDomainProject(ctx)
- var cache model.Cache
if !core.IsDefaultDomainProject(domainProject) {
return &model.DumpResponse{
@@ -69,18 +69,50 @@ func (service *AdminService) Dump(ctx context.Context, in *model.DumpRequest) (*
}, nil
}
- service.dumpAll(ctx, &cache)
+ resp := &model.DumpResponse{
+ Response: pb.CreateResponse(pb.Response_SUCCESS, "Admin dump successfully"),
+ }
- return &model.DumpResponse{
- Response: pb.CreateResponse(pb.Response_SUCCESS, "Admin dump successfully"),
- Info: version.Ver(),
- AppConfig: configs,
- Environments: environments,
- Cache: &cache,
- }, nil
+ if len(in.Options) == 0 {
+ service.dump(ctx, "cache", resp)
+ return resp, nil
+ }
+
+ options := make(map[string]struct{}, len(in.Options))
+ for _, option := range in.Options {
+ if option == "all" {
+ service.dump(ctx, "all", resp)
+ return resp, nil
+ }
+ options[option] = struct{}{}
+ }
+ for option := range options {
+ service.dump(ctx, option, resp)
+ }
+ return resp, nil
}
-func (service *AdminService) dumpAll(ctx context.Context, cache *model.Cache) {
+func (service *AdminService) dump(ctx context.Context, option string, resp *model.DumpResponse) {
+ switch option {
+ case "info":
+ resp.Info = version.Ver()
+ case "config":
+ resp.AppConfig = configs
+ case "env":
+ resp.Environments = environments
+ case "cache":
+ var cache model.Cache
+ service.dumpAllCache(ctx, &cache)
+ resp.Cache = &cache
+ case "all":
+ service.dump(ctx, "info", resp)
+ service.dump(ctx, "config", resp)
+ service.dump(ctx, "env", resp)
+ service.dump(ctx, "cache", resp)
+ }
+}
+
+func (service *AdminService) dumpAllCache(ctx context.Context, cache *model.Cache) {
gopool.New(ctx, gopool.Configure().Workers(2)).
Do(func(_ context.Context) { setValue(backend.Store().Service(), &cache.Microservices) }).
Do(func(_ context.Context) { setValue(backend.Store().ServiceIndex(), &cache.Indexes) }).
@@ -111,3 +143,15 @@ func (service *AdminService) Clusters(ctx context.Context, in *model.ClustersReq
Clusters: registry.Configuration().Clusters,
}, nil
}
+
+func (service *AdminService) AlarmList(ctx context.Context, in *model.AlarmListRequest) (*model.AlarmListResponse, error) {
+ return &model.AlarmListResponse{
+ Alarms: alarm.ListAll(),
+ }, nil
+}
+
+func (service *AdminService) ClearAlarm(ctx context.Context, in *model.ClearAlarmRequest) (*model.ClearAlarmResponse, error) {
+ alarm.ClearAll()
+ log.Infof("service center alarms are cleared")
+ return &model.ClearAlarmResponse{}, nil
+}
diff --git a/server/alarm/common.go b/server/alarm/common.go
new file mode 100644
index 0000000..a5a87b2
--- /dev/null
+++ b/server/alarm/common.go
@@ -0,0 +1,80 @@
+// 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 alarm
+
+import (
+ "fmt"
+ "github.com/apache/servicecomb-service-center/server/alarm/model"
+)
+
+const (
+ Activated model.Status = "ACTIVATED"
+ Cleared model.Status = "CLEARED"
+)
+
+const (
+ IdBackendConnectionRefuse model.ID = "BackendConnectionRefuse"
+ IdInternalError model.ID = "InternalError"
+)
+
+const (
+ FieldAdditionalContext = "detail"
+)
+
+const (
+ Subject = "__ALARM_SUBJECT__"
+ Group = "__ALARM_GROUP__"
+)
+
+func FieldBool(key string, v bool) model.Field {
+ return model.Field{Key: key, Value: v}
+}
+
+func FieldString(key string, v string) model.Field {
+ return model.Field{Key: key, Value: v}
+}
+
+func FieldInt64(key string, v int64) model.Field {
+ return model.Field{Key: key, Value: v}
+}
+
+func FieldInt(key string, v int) model.Field {
+ return model.Field{Key: key, Value: v}
+}
+
+func FieldFloat64(key string, v float64) model.Field {
+ return model.Field{Key: key, Value: v}
+}
+
+func AdditionalContext(format string, args ...interface{}) model.Field {
+ return FieldString(FieldAdditionalContext, fmt.Sprintf(format, args...))
+}
+
+func ListAll() []*model.AlarmEvent {
+ return AlarmCenter().ListAll()
+}
+
+func Raise(id model.ID, fields ...model.Field) error {
+ return AlarmCenter().Raise(id, fields...)
+}
+
+func Clear(id model.ID) error {
+ return AlarmCenter().Clear(id)
+}
+
+func ClearAll() {
+ AlarmCenter().ClearAll()
+}
diff --git a/server/alarm/model/types.go b/server/alarm/model/types.go
new file mode 100644
index 0000000..86ca809
--- /dev/null
+++ b/server/alarm/model/types.go
@@ -0,0 +1,62 @@
+// 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 model
+
+import (
+ nf "github.com/apache/servicecomb-service-center/pkg/notify"
+ "github.com/apache/servicecomb-service-center/pkg/util"
+)
+
+type ID string
+
+type Status string
+
+type Field struct {
+ Key string
+ Value interface{}
+}
+
+type AlarmEvent struct {
+ nf.Event `json:"-"`
+ Status Status `json:"status"`
+ Id ID `json:"id"`
+ Fields util.JSONObject `json:"fields,omitempty"`
+}
+
+func (ae *AlarmEvent) FieldBool(key string) bool {
+ v, _ := ae.Fields[key].(bool)
+ return v
+}
+
+func (ae *AlarmEvent) FieldString(key string) string {
+ v, _ := ae.Fields[key].(string)
+ return v
+}
+
+func (ae *AlarmEvent) FieldInt64(key string) int64 {
+ v, _ := ae.Fields[key].(int64)
+ return v
+}
+
+func (ae *AlarmEvent) FieldInt(key string) int {
+ v, _ := ae.Fields[key].(int)
+ return v
+}
+
+func (ae *AlarmEvent) FieldFloat64(key string) float64 {
+ v, _ := ae.Fields[key].(float64)
+ return v
+}
diff --git a/server/alarm/service.go b/server/alarm/service.go
new file mode 100644
index 0000000..d1714d9
--- /dev/null
+++ b/server/alarm/service.go
@@ -0,0 +1,101 @@
+// 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 alarm
+
+import (
+ "github.com/apache/servicecomb-service-center/pkg/log"
+ nf "github.com/apache/servicecomb-service-center/pkg/notify"
+ "github.com/apache/servicecomb-service-center/pkg/util"
+ "github.com/apache/servicecomb-service-center/server/alarm/model"
+ "github.com/apache/servicecomb-service-center/server/notify"
+ "sync"
+)
+
+var (
+ service *AlarmService
+ once sync.Once
+)
+
+type AlarmService struct {
+ nf.Subscriber
+ alarms util.ConcurrentMap
+}
+
+func (ac *AlarmService) Raise(id model.ID, fields ...model.Field) error {
+ ae := &model.AlarmEvent{
+ Event: nf.NewEvent(nf.NOTIFTY, Subject, ""),
+ Status: Activated,
+ Id: id,
+ Fields: util.NewJSONObject(),
+ }
+ for _, f := range fields {
+ ae.Fields[f.Key] = f.Value
+ }
+ return notify.NotifyCenter().Publish(ae)
+}
+
+func (ac *AlarmService) Clear(id model.ID) error {
+ ae := &model.AlarmEvent{
+ Event: nf.NewEvent(nf.NOTIFTY, Subject, ""),
+ Status: Cleared,
+ Id: id,
+ }
+ return notify.NotifyCenter().Publish(ae)
+}
+
+func (ac *AlarmService) ListAll() (ls []*model.AlarmEvent) {
+ ac.alarms.ForEach(func(item util.MapItem) (next bool) {
+ ls = append(ls, item.Value.(*model.AlarmEvent))
+ return true
+ })
+ return
+}
+
+func (ac *AlarmService) ClearAll() {
+ ac.alarms = util.ConcurrentMap{}
+ return
+}
+
+func (ac *AlarmService) OnMessage(evt nf.Event) {
+ alarm := evt.(*model.AlarmEvent)
+ switch alarm.Status {
+ case Cleared:
+ if itf, ok := ac.alarms.Get(alarm.Id); ok {
+ if exist := itf.(*model.AlarmEvent); exist.Status != Cleared {
+ exist.Status = Cleared
+ alarm = exist
+ }
+ }
+ default:
+ ac.alarms.Put(alarm.Id, alarm)
+ }
+ log.Debugf("alarm[%s] %s, %v", alarm.Id, alarm.Status, alarm.Fields)
+}
+
+func NewAlarmService() *AlarmService {
+ c := &AlarmService{
+ Subscriber: nf.NewSubscriber(nf.NOTIFTY, Subject, Group),
+ }
+ notify.NotifyCenter().AddSubscriber(c)
+ return c
+}
+
+func AlarmCenter() *AlarmService {
+ once.Do(func() {
+ service = NewAlarmService()
+ })
+ return service
+}
diff --git a/server/core/config.go b/server/core/config.go
index c95224d..ab2f891 100644
--- a/server/core/config.go
+++ b/server/core/config.go
@@ -87,7 +87,7 @@ func newInfo() pb.ServerInformation {
LogSys: beego.AppConfig.DefaultBool("log_sys", false),
PluginsDir: beego.AppConfig.DefaultString("plugins_dir", "./plugins"),
- Plugins: make(util.JSONObject),
+ Plugins: util.NewJSONObject(),
EnablePProf: beego.AppConfig.DefaultInt("enable_pprof", 0) != 0,
EnableCache: beego.AppConfig.DefaultInt("enable_cache", 1) != 0,
diff --git a/server/core/swagger/v4.yaml b/server/core/swagger/v4.yaml
index 7b03b58..e55a380 100644
--- a/server/core/swagger/v4.yaml
+++ b/server/core/swagger/v4.yaml
@@ -1571,7 +1571,7 @@ paths:
get:
description: |
查询单个服务的所有信息。
- operationId: GetServiceDetail
+ operationId: getServiceDetail
parameters:
- name: x-domain-name
in: header
@@ -1606,7 +1606,7 @@ paths:
get:
description: |
查询单个服务的所有信息。
- operationId: GetServicesInfo
+ operationId: getServicesInfo
parameters:
- name: x-domain-name
in: header
@@ -1641,7 +1641,7 @@ paths:
get:
description: |
查询服务间的关系。
- operationId: GetGraph
+ operationId: getGraph
parameters:
- name: x-domain-name
in: header
@@ -1673,7 +1673,7 @@ paths:
get:
description: |
查询所有appId。
- operationId: GetApplications
+ operationId: getApplications
parameters:
- name: x-domain-name
in: header
@@ -1723,6 +1723,11 @@ paths:
description: default项目
required: true
type: string
+ - name: options
+ in: query
+ default: cache
+ description: 枚举值有:info,config,env,cache和all
+ type: string
tags:
- admin
responses:
@@ -1759,6 +1764,53 @@ paths:
description: clusters information
schema:
$ref: '#/definitions/ClustersResponse'
+ /v4/{project}/admin/alarms:
+ get:
+ description: |
+ Return the alarms list of Service Center
+ operationId: alarmList
+ parameters:
+ - name: x-domain-name
+ in: header
+ type: string
+ default: default
+ description: default租户
+ required: true
+ - name: project
+ in: path
+ default: default
+ description: default项目
+ required: true
+ type: string
+ tags:
+ - admin
+ responses:
+ 200:
+ description: alarms information
+ schema:
+ $ref: '#/definitions/AlarmList'
+ delete:
+ description: |
+ Clear the alarms list of Service Center
+ operationId: clearAlarm
+ parameters:
+ - name: x-domain-name
+ in: header
+ type: string
+ default: default
+ description: default租户
+ required: true
+ - name: project
+ in: path
+ default: default
+ description: default项目
+ required: true
+ type: string
+ tags:
+ - admin
+ responses:
+ 200:
+ description: cleared
definitions:
Version:
type: object
@@ -2609,3 +2661,18 @@ definitions:
type: string
detail:
type: string
+ AlarmList:
+ type: object
+ description: alarms information
+ additionalProperties:
+ type: array
+ items:
+ $ref: '#/definitions/Alarm'
+ Alarm:
+ type: object
+ description: alarm information
+ properties:
+ id:
+ type: string
+ fields:
+ $ref: '#/definitions/Properties'
diff --git a/server/error/error.go b/server/error/error.go
index 6b352b7..0d1a4f7 100644
--- a/server/error/error.go
+++ b/server/error/error.go
@@ -19,81 +19,61 @@ package error
import (
"encoding/json"
"fmt"
+ "github.com/apache/servicecomb-service-center/pkg/log"
)
var errors = map[int32]string{
- ErrInvalidParams: "Invalid parameter(s)",
-
+ ErrInvalidParams: "Invalid parameter(s)",
+ ErrUnhealthy: "Server is Unhealthy",
ErrServiceAlreadyExists: "Micro-service already exists",
ErrServiceNotExists: "Micro-service does not exist",
ErrServiceVersionNotExists: "Micro-service version does not exist",
ErrDeployedInstance: "Micro-service has deployed instance(s)",
ErrDependedOnConsumer: "Consumer(s) depends on this micro-service",
-
- ErrUndefinedSchemaId: "Undefined schema id",
- ErrModifySchemaNotAllow: "Not allowed to modify schema",
- ErrSchemaNotExists: "Schema does not exist",
-
- ErrInstanceNotExists: "Instance does not exist",
- ErrPermissionDeny: "Access micro-service refused",
-
- ErrTagNotExists: "Tag does not exist",
-
- ErrRuleAlreadyExists: "Rule already exist",
- ErrBlackAndWhiteRule: "Can not have both 'BLACK' and 'WHITE'",
- ErrModifyRuleNotAllow: "Not allowed to modify the type of the rule",
- ErrRuleNotExists: "Rule does not exist",
-
- ErrNotEnoughQuota: "Not enough quota",
-
- ErrUnauthorized: "Request unauthorized",
-
- ErrInternal: "Internal server error",
- ErrUnavailableBackend: "Registry service is unavailable",
- ErrUnavailableQuota: "Quota service is unavailable",
-
- ErrEndpointAlreadyExists: "Endpoint is already belong to other service",
-
- ErrForbidden: "Forbidden",
+ ErrUndefinedSchemaId: "Undefined schema id",
+ ErrModifySchemaNotAllow: "Not allowed to modify schema",
+ ErrSchemaNotExists: "Schema does not exist",
+ ErrInstanceNotExists: "Instance does not exist",
+ ErrPermissionDeny: "Access micro-service refused",
+ ErrTagNotExists: "Tag does not exist",
+ ErrRuleAlreadyExists: "Rule already exist",
+ ErrBlackAndWhiteRule: "Can not have both 'BLACK' and 'WHITE'",
+ ErrModifyRuleNotAllow: "Not allowed to modify the type of the rule",
+ ErrRuleNotExists: "Rule does not exist",
+ ErrNotEnoughQuota: "Not enough quota",
+ ErrUnauthorized: "Request unauthorized",
+ ErrInternal: "Internal server error",
+ ErrUnavailableBackend: "Registry service is unavailable",
+ ErrUnavailableQuota: "Quota service is unavailable",
+ ErrEndpointAlreadyExists: "Endpoint is already belong to other service",
+ ErrForbidden: "Forbidden",
}
const (
- ErrInvalidParams int32 = 400001
- ErrUnauthorized int32 = 401002
- ErrInternal int32 = 500003
-
- ErrServiceAlreadyExists int32 = 400010
- ErrUnavailableBackend int32 = 500011
-
- ErrServiceNotExists int32 = 400012
-
- ErrDeployedInstance int32 = 400013
-
- ErrUndefinedSchemaId int32 = 400014
- ErrModifySchemaNotAllow int32 = 400015
- ErrSchemaNotExists int32 = 400016
-
- ErrInstanceNotExists int32 = 400017
-
- ErrTagNotExists int32 = 400018
-
- ErrRuleAlreadyExists int32 = 400019
- ErrBlackAndWhiteRule int32 = 400020
- ErrModifyRuleNotAllow int32 = 400021
- ErrRuleNotExists int32 = 400022
-
- ErrDependedOnConsumer int32 = 400023
-
- ErrPermissionDeny int32 = 400024
-
- ErrEndpointAlreadyExists int32 = 400025
-
+ ErrInvalidParams int32 = 400001
+ ErrUnhealthy int32 = 400002
+ ErrServiceAlreadyExists int32 = 400010
+ ErrServiceNotExists int32 = 400012
+ ErrDeployedInstance int32 = 400013
+ ErrUndefinedSchemaId int32 = 400014
+ ErrModifySchemaNotAllow int32 = 400015
+ ErrSchemaNotExists int32 = 400016
+ ErrInstanceNotExists int32 = 400017
+ ErrTagNotExists int32 = 400018
+ ErrRuleAlreadyExists int32 = 400019
+ ErrBlackAndWhiteRule int32 = 400020
+ ErrModifyRuleNotAllow int32 = 400021
+ ErrRuleNotExists int32 = 400022
+ ErrDependedOnConsumer int32 = 400023
+ ErrPermissionDeny int32 = 400024
+ ErrEndpointAlreadyExists int32 = 400025
ErrServiceVersionNotExists int32 = 400026
-
- ErrNotEnoughQuota int32 = 400100
- ErrUnavailableQuota int32 = 500101
-
- ErrForbidden int32 = 403001
+ ErrNotEnoughQuota int32 = 400100
+ ErrUnauthorized int32 = 401002
+ ErrForbidden int32 = 403001
+ ErrInternal int32 = 500003
+ ErrUnavailableBackend int32 = 500011
+ ErrUnavailableQuota int32 = 500101
)
type Error struct {
@@ -140,7 +120,7 @@ func NewErrorf(code int32, format string, args ...interface{}) *Error {
func RegisterErrors(errs map[int32]string) {
for err, msg := range errs {
if err < 400000 || err >= 600000 {
- // should be between 4xx and 5xx
+ log.Warnf("error code[%v] should be between 4xx and 5xx", err)
continue
}
errors[err] = msg
diff --git a/server/admin/model/cluster.go b/server/health/health.go
similarity index 56%
copy from server/admin/model/cluster.go
copy to server/health/health.go
index 6cabae9..e73b89e 100644
--- a/server/admin/model/cluster.go
+++ b/server/health/health.go
@@ -13,17 +13,35 @@
// See the License for the specific language governing permissions and
// limitations under the License.
-package model
+package health
import (
- pb "github.com/apache/servicecomb-service-center/server/core/proto"
- "github.com/apache/servicecomb-service-center/server/plugin/pkg/registry"
+ "errors"
+ "github.com/apache/servicecomb-service-center/server/alarm"
)
-type ClustersRequest struct {
+var healthChecker HealthChecker = &DefaultHealthChecker{}
+
+type HealthChecker interface {
+ Healthy() error
+}
+
+type DefaultHealthChecker struct {
+}
+
+func (hc *DefaultHealthChecker) Healthy() error {
+ for _, a := range alarm.ListAll() {
+ if a.Id == alarm.IdBackendConnectionRefuse && a.Status != alarm.Cleared {
+ return errors.New(a.FieldString(alarm.FieldAdditionalContext))
+ }
+ }
+ return nil
+}
+
+func SetGlobalHealthChecker(hc HealthChecker) {
+ healthChecker = hc
}
-type ClustersResponse struct {
- Response *pb.Response `json:"response,omitempty"`
- Clusters registry.Clusters `json:"clusters,omitempty"`
+func GlobalHealthChecker() HealthChecker {
+ return healthChecker
}
diff --git a/server/health/health_test.go b/server/health/health_test.go
new file mode 100644
index 0000000..09fd52b
--- /dev/null
+++ b/server/health/health_test.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 health
+
+import (
+ "github.com/apache/servicecomb-service-center/server/alarm"
+ "github.com/apache/servicecomb-service-center/server/notify"
+ "testing"
+ "time"
+)
+
+func TestDefaultHealthChecker_Healthy(t *testing.T) {
+ notify.NotifyCenter().Start()
+
+ // normal case
+ var hc DefaultHealthChecker
+ if err := hc.Healthy(); err != nil {
+ t.Fatal("TestDefaultHealthChecker_Healthy failed", err)
+ }
+ alarm.Raise(alarm.IdBackendConnectionRefuse, alarm.AdditionalContext("a"))
+ time.Sleep(time.Second)
+ if err := hc.Healthy(); err == nil || err.Error() != "a" {
+ t.Fatal("TestDefaultHealthChecker_Healthy failed", err)
+ }
+ alarm.Clear(alarm.IdBackendConnectionRefuse)
+ time.Sleep(time.Second)
+ if err := hc.Healthy(); err != nil {
+ t.Fatal("TestDefaultHealthChecker_Healthy failed", err)
+ }
+
+ // set global hc
+ if GlobalHealthChecker() == &hc {
+ t.Fatal("TestDefaultHealthChecker_Healthy failed")
+ }
+
+ SetGlobalHealthChecker(&hc)
+
+ if GlobalHealthChecker() != &hc {
+ t.Fatal("TestDefaultHealthChecker_Healthy failed")
+ }
+}
diff --git a/server/admin/model/cluster.go b/server/notify/center.go
similarity index 71%
copy from server/admin/model/cluster.go
copy to server/notify/center.go
index 6cabae9..db95ec3 100644
--- a/server/admin/model/cluster.go
+++ b/server/notify/center.go
@@ -13,17 +13,19 @@
// See the License for the specific language governing permissions and
// limitations under the License.
-package model
+package notify
import (
- pb "github.com/apache/servicecomb-service-center/server/core/proto"
- "github.com/apache/servicecomb-service-center/server/plugin/pkg/registry"
+ "github.com/apache/servicecomb-service-center/pkg/notify"
)
-type ClustersRequest struct {
+var INSTANCE = notify.RegisterType("INSTANCE", InstanceEventQueueSize)
+var notifyService *notify.NotifyService
+
+func init() {
+ notifyService = notify.NewNotifyService()
}
-type ClustersResponse struct {
- Response *pb.Response `json:"response,omitempty"`
- Clusters registry.Clusters `json:"clusters,omitempty"`
+func NotifyCenter() *notify.NotifyService {
+ return notifyService
}
diff --git a/server/admin/model/cluster.go b/server/notify/common.go
similarity index 69%
rename from server/admin/model/cluster.go
rename to server/notify/common.go
index 6cabae9..e3c9a67 100644
--- a/server/admin/model/cluster.go
+++ b/server/notify/common.go
@@ -13,17 +13,13 @@
// See the License for the specific language governing permissions and
// limitations under the License.
-package model
+package notify
-import (
- pb "github.com/apache/servicecomb-service-center/server/core/proto"
- "github.com/apache/servicecomb-service-center/server/plugin/pkg/registry"
-)
-
-type ClustersRequest struct {
-}
+import "time"
-type ClustersResponse struct {
- Response *pb.Response `json:"response,omitempty"`
- Clusters registry.Clusters `json:"clusters,omitempty"`
-}
+const (
+ AddJobTimeout = 1 * time.Second
+ SendTimeout = 5 * time.Second
+ HeartbeatTimeout = 30 * time.Second
+ InstanceEventQueueSize = 5000
+)
diff --git a/server/service/notification/listwatcher.go b/server/notify/listwatcher.go
similarity index 66%
rename from server/service/notification/listwatcher.go
rename to server/notify/listwatcher.go
index d0a90f8..0b831d7 100644
--- a/server/service/notification/listwatcher.go
+++ b/server/notify/listwatcher.go
@@ -14,38 +14,39 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package notification
+package notify
import (
"github.com/apache/servicecomb-service-center/pkg/gopool"
"github.com/apache/servicecomb-service-center/pkg/log"
+ "github.com/apache/servicecomb-service-center/pkg/notify"
pb "github.com/apache/servicecomb-service-center/server/core/proto"
"golang.org/x/net/context"
"time"
)
// 状态变化推送
-type WatchJob struct {
- *BaseNotifyJob
+type InstanceEvent struct {
+ notify.Event
Revision int64
Response *pb.WatchInstanceResponse
}
-type ListWatcher struct {
- *BaseSubscriber
- Job chan *WatchJob
+type InstanceEventListWatcher struct {
+ notify.Subscriber
+ Job chan *InstanceEvent
ListRevision int64
ListFunc func() (results []*pb.WatchInstanceResponse, rev int64)
listCh chan struct{}
}
-func (s *ListWatcher) SetError(err error) {
- s.BaseSubscriber.SetError(err)
+func (s *InstanceEventListWatcher) SetError(err error) {
+ s.Subscriber.SetError(err)
// 触发清理job
- s.Service().AddJob(NewNotifyServiceHealthCheckJob(s))
+ s.Service().Publish(notify.NewNotifyServiceHealthCheckJob(s))
}
-func (w *ListWatcher) OnAccept() {
+func (w *InstanceEventListWatcher) OnAccept() {
if w.Err() != nil {
return
}
@@ -53,7 +54,7 @@ func (w *ListWatcher) OnAccept() {
gopool.Go(w.listAndPublishJobs)
}
-func (w *ListWatcher) listAndPublishJobs(_ context.Context) {
+func (w *InstanceEventListWatcher) listAndPublishJobs(_ context.Context) {
defer close(w.listCh)
if w.ListFunc == nil {
return
@@ -61,17 +62,17 @@ func (w *ListWatcher) listAndPublishJobs(_ context.Context) {
results, rev := w.ListFunc()
w.ListRevision = rev
for _, response := range results {
- w.sendMessage(NewWatchJob(w.Group(), w.Subject(), w.ListRevision, response))
+ w.sendMessage(NewInstanceEvent(w.Group(), w.Subject(), w.ListRevision, response))
}
}
//被通知
-func (w *ListWatcher) OnMessage(job NotifyJob) {
+func (w *InstanceEventListWatcher) OnMessage(job notify.Event) {
if w.Err() != nil {
return
}
- wJob, ok := job.(*WatchJob)
+ wJob, ok := job.(*InstanceEvent)
if !ok {
return
}
@@ -98,7 +99,7 @@ func (w *ListWatcher) OnMessage(job NotifyJob) {
w.sendMessage(wJob)
}
-func (w *ListWatcher) sendMessage(job *WatchJob) {
+func (w *InstanceEventListWatcher) sendMessage(job *InstanceEvent) {
defer log.Recover()
select {
case w.Job <- job:
@@ -115,33 +116,29 @@ func (w *ListWatcher) sendMessage(job *WatchJob) {
}
}
-func (w *ListWatcher) Timeout() time.Duration {
- return DEFAULT_ADD_JOB_TIMEOUT
+func (w *InstanceEventListWatcher) Timeout() time.Duration {
+ return AddJobTimeout
}
-func (w *ListWatcher) Close() {
+func (w *InstanceEventListWatcher) Close() {
close(w.Job)
}
-func NewWatchJob(group, subject string, rev int64, response *pb.WatchInstanceResponse) *WatchJob {
- return &WatchJob{
- BaseNotifyJob: &BaseNotifyJob{
- group: group,
- subject: subject,
- nType: INSTANCE,
- },
+func NewInstanceEvent(group, subject string, rev int64, response *pb.WatchInstanceResponse) *InstanceEvent {
+ return &InstanceEvent{
+ Event: notify.NewEvent(INSTANCE, subject, group),
Revision: rev,
Response: response,
}
}
-func NewListWatcher(group string, subject string,
- listFunc func() (results []*pb.WatchInstanceResponse, rev int64)) *ListWatcher {
- watcher := &ListWatcher{
- BaseSubscriber: NewSubscriber(INSTANCE, subject, group),
- Job: make(chan *WatchJob, DEFAULT_MAX_QUEUE),
- ListFunc: listFunc,
- listCh: make(chan struct{}),
+func NewInstanceEventListWatcher(group string, subject string,
+ listFunc func() (results []*pb.WatchInstanceResponse, rev int64)) *InstanceEventListWatcher {
+ watcher := &InstanceEventListWatcher{
+ Subscriber: notify.NewSubscriber(INSTANCE, subject, group),
+ Job: make(chan *InstanceEvent, INSTANCE.QueueSize()),
+ ListFunc: listFunc,
+ listCh: make(chan struct{}),
}
return watcher
}
diff --git a/server/service/notification/publisher.go b/server/notify/publisher.go
similarity index 99%
rename from server/service/notification/publisher.go
rename to server/notify/publisher.go
index 70aa7b2..fc0d703 100644
--- a/server/service/notification/publisher.go
+++ b/server/notify/publisher.go
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package notification
+package notify
import (
"github.com/apache/servicecomb-service-center/pkg/gopool"
diff --git a/server/service/notification/stream.go b/server/notify/stream.go
similarity index 67%
rename from server/service/notification/stream.go
rename to server/notify/stream.go
index 7fc663d..0bc17cd 100644
--- a/server/service/notification/stream.go
+++ b/server/notify/stream.go
@@ -14,23 +14,27 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package notification
+package notify
import (
"errors"
"github.com/apache/servicecomb-service-center/pkg/log"
"github.com/apache/servicecomb-service-center/pkg/util"
+ apt "github.com/apache/servicecomb-service-center/server/core"
pb "github.com/apache/servicecomb-service-center/server/core/proto"
+ "golang.org/x/net/context"
"time"
)
-func HandleWatchJob(watcher *ListWatcher, stream pb.ServiceInstanceCtrl_WatchServer) (err error) {
- timer := time.NewTimer(DEFAULT_HEARTBEAT_INTERVAL)
+func HandleWatchJob(watcher *InstanceEventListWatcher, stream pb.ServiceInstanceCtrl_WatchServer) (err error) {
+ timer := time.NewTimer(HeartbeatTimeout)
defer timer.Stop()
for {
select {
+ case <-stream.Context().Done():
+ return
case <-timer.C:
- timer.Reset(DEFAULT_HEARTBEAT_INTERVAL)
+ timer.Reset(HeartbeatTimeout)
// TODO grpc 长连接心跳？
case job := <-watcher.Job:
@@ -52,7 +56,17 @@ func HandleWatchJob(watcher *ListWatcher, stream pb.ServiceInstanceCtrl_WatchSer
return
}
- util.ResetTimer(timer, DEFAULT_HEARTBEAT_INTERVAL)
+ util.ResetTimer(timer, HeartbeatTimeout)
}
}
}
+
+func DoStreamListAndWatch(ctx context.Context, serviceId string, f func() ([]*pb.WatchInstanceResponse, int64), stream pb.ServiceInstanceCtrl_WatchServer) error {
+ domainProject := util.ParseDomainProject(ctx)
+ watcher := NewInstanceEventListWatcher(serviceId, apt.GetInstanceRootKey(domainProject)+"/", f)
+ err := NotifyCenter().AddSubscriber(watcher)
+ if err != nil {
+ return err
+ }
+ return HandleWatchJob(watcher, stream)
+}
diff --git a/server/service/notification/stream_test.go b/server/notify/stream_test.go
similarity index 68%
rename from server/service/notification/stream_test.go
rename to server/notify/stream_test.go
index 5193045..60bd8d9 100644
--- a/server/service/notification/stream_test.go
+++ b/server/notify/stream_test.go
@@ -14,21 +14,29 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package notification
+package notify
-import "testing"
+import (
+ "github.com/apache/servicecomb-service-center/pkg/log"
+ "golang.org/x/net/context"
+ "testing"
+)
func TestHandleWatchJob(t *testing.T) {
- defer func() { recover() }()
- w := NewListWatcher("g", "s", nil)
+ defer log.Recover()
+ w := NewInstanceEventListWatcher("g", "s", nil)
w.Job <- nil
err := HandleWatchJob(w, nil)
if err == nil {
t.Fatalf("TestHandleWatchJob failed")
}
- w.Job <- NewWatchJob("g", "s", 1, nil)
+ w.Job <- NewInstanceEvent("g", "s", 1, nil)
err = HandleWatchJob(w, nil)
- if err != nil {
- t.Fatalf("TestHandleWatchJob failed")
- }
+ t.Fatalf("TestHandleWatchJob failed")
+}
+
+func TestDoStreamListAndWatch(t *testing.T) {
+ defer log.Recover()
+ err := DoStreamListAndWatch(context.Background(), "s", nil, nil)
+ t.Fatal("TestDoStreamListAndWatch failed", err)
}
diff --git a/server/service/notification/websocket.go b/server/notify/websocket.go
similarity index 95%
rename from server/service/notification/websocket.go
rename to server/notify/websocket.go
index 07db6f6..64f6203 100644
--- a/server/service/notification/websocket.go
+++ b/server/notify/websocket.go
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package notification
+package notify
import (
"encoding/json"
@@ -34,14 +34,14 @@ type WebSocket struct {
ticker *time.Ticker
conn *websocket.Conn
// watcher subscribe the notification service event
- watcher *ListWatcher
+ watcher *InstanceEventListWatcher
needPingWatcher bool
free chan struct{}
closed chan struct{}
}
func (wh *WebSocket) Init() error {
- wh.ticker = time.NewTicker(DEFAULT_HEARTBEAT_INTERVAL)
+ wh.ticker = time.NewTicker(HeartbeatTimeout)
wh.needPingWatcher = true
wh.free = make(chan struct{}, 1)
wh.closed = make(chan struct{})
@@ -51,7 +51,7 @@ func (wh *WebSocket) Init() error {
remoteAddr := wh.conn.RemoteAddr().String()
// put in notification service queue
- if err := GetNotifyService().AddSubscriber(wh.watcher); err != nil {
+ if err := NotifyCenter().AddSubscriber(wh.watcher); err != nil {
err = fmt.Errorf("establish[%s] websocket watch failed: notify service error, %s",
remoteAddr, err.Error())
log.Errorf(nil, err.Error())
@@ -72,7 +72,7 @@ func (wh *WebSocket) Init() error {
}
func (wh *WebSocket) Timeout() time.Duration {
- return DEFAULT_SEND_TIMEOUT
+ return SendTimeout
}
func (wh *WebSocket) heartbeat(messageType int) error {
@@ -194,8 +194,8 @@ func (wh *WebSocket) HandleWatchWebSocketJob(o interface{}) {
remoteAddr, wh.watcher.Subject(), wh.watcher.Group())
wh.heartbeat(websocket.PingMessage)
return
- case *WatchJob:
- job := o.(*WatchJob)
+ case *InstanceEvent:
+ job := o.(*InstanceEvent)
resp := job.Response
providerFlag := fmt.Sprintf("%s/%s/%s", resp.Key.AppId, resp.Key.ServiceName, resp.Key.Version)
@@ -253,7 +253,7 @@ func DoWebSocketListAndWatch(ctx context.Context, serviceId string, f func() ([]
socket := &WebSocket{
ctx: ctx,
conn: conn,
- watcher: NewListWatcher(serviceId, apt.GetInstanceRootKey(domainProject)+"/", f),
+ watcher: NewInstanceEventListWatcher(serviceId, apt.GetInstanceRootKey(domainProject)+"/", f),
}
process(socket)
}
diff --git a/server/service/notification/websocket_test.go b/server/notify/websocket_test.go
similarity index 88%
rename from server/service/notification/websocket_test.go
rename to server/notify/websocket_test.go
index 403a2e3..6d1c7ce 100644
--- a/server/service/notification/websocket_test.go
+++ b/server/notify/websocket_test.go
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package notification
+package notify
import _ "github.com/apache/servicecomb-service-center/server/init"
import (
@@ -56,31 +56,12 @@ func (h *watcherConn) ServeHTTP(w http.ResponseWriter, r *http.Request) {
func TestDoWebSocketListAndWatch(t *testing.T) {
s := httptest.NewServer(&watcherConn{})
- GetNotifyService().Start()
-
conn, _, _ := websocket.DefaultDialer.Dial(
strings.Replace(s.URL, "http://", "ws://", 1), nil)
- go func() {
- DoWebSocketListAndWatch(context.Background(), "", nil, conn)
-
- w2 := NewListWatcher("g", "s", func() (results []*proto.WatchInstanceResponse, rev int64) {
- return
- })
- ws2 := &WebSocket{
- ctx: context.Background(),
- conn: conn,
- watcher: w2,
- }
- err := ws2.Init()
- if err != nil {
- t.Fatalf("TestPublisher_Run")
- }
- }()
-
EstablishWebSocketError(conn, errors.New("error"))
- w := NewListWatcher("g", "s", func() (results []*proto.WatchInstanceResponse, rev int64) {
+ w := NewInstanceEventListWatcher("g", "s", func() (results []*proto.WatchInstanceResponse, rev int64) {
results = append(results, &proto.WatchInstanceResponse{
Response: proto.CreateResponse(proto.Response_SUCCESS, "ok"),
Action: string(proto.EVT_CREATE),
@@ -89,7 +70,6 @@ func TestDoWebSocketListAndWatch(t *testing.T) {
})
return
})
- w.nType = -1
ws := &WebSocket{
ctx: context.Background(),
@@ -97,21 +77,35 @@ func TestDoWebSocketListAndWatch(t *testing.T) {
watcher: w,
}
err := ws.Init()
- if err == nil {
- t.Fatalf("TestPublisher_Run")
- }
-
- w.nType = INSTANCE
- err = ws.Init()
if err != nil {
t.Fatalf("TestPublisher_Run")
}
+
+ NotifyCenter().Start()
+
+ go func() {
+ DoWebSocketListAndWatch(context.Background(), "", nil, conn)
+
+ w2 := NewInstanceEventListWatcher("g", "s", func() (results []*proto.WatchInstanceResponse, rev int64) {
+ return
+ })
+ ws2 := &WebSocket{
+ ctx: context.Background(),
+ conn: conn,
+ watcher: w2,
+ }
+ err := ws2.Init()
+ if err != nil {
+ t.Fatalf("TestPublisher_Run")
+ }
+ }()
+
go ws.HandleWatchWebSocketControlMessage()
w.OnMessage(nil)
- w.OnMessage(&WatchJob{})
+ w.OnMessage(&InstanceEvent{})
- GetNotifyService().AddJob(NewWatchJob("g", "s", 1, &proto.WatchInstanceResponse{
+ NotifyCenter().Publish(NewInstanceEvent("g", "s", 1, &proto.WatchInstanceResponse{
Response: proto.CreateResponse(proto.Response_SUCCESS, "ok"),
Action: string(proto.EVT_CREATE),
Key: &proto.MicroServiceKey{},
diff --git a/server/plugin/pkg/discovery/etcd/cacher_kv.go b/server/plugin/pkg/discovery/etcd/cacher_kv.go
index eb352fd..6ea25d8 100644
--- a/server/plugin/pkg/discovery/etcd/cacher_kv.go
+++ b/server/plugin/pkg/discovery/etcd/cacher_kv.go
@@ -178,8 +178,8 @@ func (c *KvCacher) refresh(ctx context.Context) {
for {
nextPeriod := minWaitInterval
if err := c.ListAndWatch(ctx); err != nil {
- nextPeriod = util.GetBackoff().Delay(retries)
retries++
+ nextPeriod = util.GetBackoff().Delay(retries)
} else {
retries = 0
}
diff --git a/server/plugin/pkg/discovery/k8s/adaptor/kube_client.go b/server/plugin/pkg/discovery/k8s/adaptor/kube_client.go
index 80a2b52..9569742 100644
--- a/server/plugin/pkg/discovery/k8s/adaptor/kube_client.go
+++ b/server/plugin/pkg/discovery/k8s/adaptor/kube_client.go
@@ -19,6 +19,7 @@ import (
"github.com/apache/servicecomb-service-center/pkg/gopool"
"github.com/apache/servicecomb-service-center/pkg/log"
"github.com/apache/servicecomb-service-center/pkg/util"
+ "github.com/apache/servicecomb-service-center/server/alarm"
pb "github.com/apache/servicecomb-service-center/server/core/proto"
"golang.org/x/net/context"
"k8s.io/api/core/v1"
@@ -54,9 +55,7 @@ type K8sClient struct {
goroutine *gopool.Pool
}
-func (c *K8sClient) init() {
- var err error
-
+func (c *K8sClient) init() (err error) {
c.ready = make(chan struct{})
c.stopCh = make(chan struct{})
c.goroutine = gopool.New(context.Background())
@@ -78,6 +77,7 @@ func (c *K8sClient) init() {
// append ipIndex build function
c.AppendEventFunc(TypePod, c.onPodEvent)
+ return
}
func (c *K8sClient) newListWatcher(t K8sType, lister cache.SharedIndexInformer) (lw ListWatcher) {
@@ -222,6 +222,13 @@ func (c *K8sClient) GetNodeByPod(pod *v1.Pod) (node *v1.Node) {
}
func (c *K8sClient) Run() {
+ if err := c.init(); err != nil {
+ alarm.Raise(alarm.IdBackendConnectionRefuse,
+ alarm.AdditionalContext("%v", err))
+ return
+ }
+ alarm.Clear(alarm.IdBackendConnectionRefuse)
+
c.goroutine.
Do(func(_ context.Context) { c.services.Run(c.stopCh) }).
Do(func(_ context.Context) { c.endpoints.Run(c.stopCh) }).
@@ -250,7 +257,6 @@ func (c *K8sClient) Ready() <-chan struct{} {
func Kubernetes() *K8sClient {
clientOnce.Do(func() {
client = &K8sClient{}
- client.init()
client.Run()
})
return client
diff --git a/server/plugin/pkg/discovery/servicecenter/syncer.go b/server/plugin/pkg/discovery/servicecenter/syncer.go
index c504c28..ea24fff 100644
--- a/server/plugin/pkg/discovery/servicecenter/syncer.go
+++ b/server/plugin/pkg/discovery/servicecenter/syncer.go
@@ -21,6 +21,7 @@ import (
"github.com/apache/servicecomb-service-center/pkg/log"
"github.com/apache/servicecomb-service-center/pkg/util"
"github.com/apache/servicecomb-service-center/server/admin/model"
+ "github.com/apache/servicecomb-service-center/server/alarm"
"github.com/apache/servicecomb-service-center/server/core"
"github.com/apache/servicecomb-service-center/server/core/backend"
pb "github.com/apache/servicecomb-service-center/server/core/proto"
@@ -55,6 +56,13 @@ func (c *Syncer) Sync(ctx context.Context) error {
return err
}
+ if len(errs) == 0 {
+ alarm.Clear(alarm.IdBackendConnectionRefuse)
+ } else {
+ alarm.Raise(alarm.IdBackendConnectionRefuse,
+ alarm.AdditionalContext("%v", errs))
+ }
+
// microservice
serviceCacher, ok := c.cachers[backend.SERVICE]
if ok {
diff --git a/server/plugin/pkg/registry/etcd/etcd.go b/server/plugin/pkg/registry/etcd/etcd.go
index 827d74b..5dc3e5c 100644
--- a/server/plugin/pkg/registry/etcd/etcd.go
+++ b/server/plugin/pkg/registry/etcd/etcd.go
@@ -24,6 +24,7 @@ import (
"github.com/apache/servicecomb-service-center/pkg/gopool"
"github.com/apache/servicecomb-service-center/pkg/log"
"github.com/apache/servicecomb-service-center/pkg/util"
+ "github.com/apache/servicecomb-service-center/server/alarm"
mgr "github.com/apache/servicecomb-service-center/server/plugin"
"github.com/apache/servicecomb-service-center/server/plugin/pkg/registry"
"github.com/coreos/etcd/clientv3"
@@ -112,16 +113,20 @@ func (c *EtcdClient) newClient() (*clientv3.Client, error) {
TLS: c.TLSConfig,
AutoSyncInterval: 0,
})
- if err != nil {
- return nil, err
- }
-
defer func() {
if err != nil {
- client.Close()
+ alarm.Raise(alarm.IdBackendConnectionRefuse, alarm.AdditionalContext("%v", err))
+
+ if client != nil {
+ client.Close()
+ }
}
}()
+ if err != nil {
+ return nil, err
+ }
+
ctx, _ := context.WithTimeout(client.Ctx(), healthCheckTimeout)
resp, err := client.MemberList(ctx)
if err != nil {
@@ -712,6 +717,9 @@ hcLoop:
continue
}
}
+
+ alarm.Clear(alarm.IdBackendConnectionRefuse)
+
retries, start = healthCheckRetryTimes, time.Now()
continue hcLoop
}
diff --git a/server/rest/controller/rest_util.go b/server/rest/controller/rest_util.go
index ca75b1f..333c583 100644
--- a/server/rest/controller/rest_util.go
+++ b/server/rest/controller/rest_util.go
@@ -21,6 +21,7 @@ import (
"fmt"
"github.com/apache/servicecomb-service-center/pkg/rest"
"github.com/apache/servicecomb-service-center/pkg/util"
+ "github.com/apache/servicecomb-service-center/server/alarm"
pb "github.com/apache/servicecomb-service-center/server/core/proto"
"github.com/apache/servicecomb-service-center/server/error"
"net/http"
@@ -33,6 +34,10 @@ func WriteError(w http.ResponseWriter, code int32, detail string) {
w.Header().Set(rest.HEADER_CONTENT_TYPE, rest.CONTENT_TYPE_JSON)
w.WriteHeader(err.StatusCode())
fmt.Fprintln(w, util.BytesToStringWithNoCopy(err.Marshal()))
+
+ if err.InternalError() {
+ alarm.Raise(alarm.IdInternalError, alarm.AdditionalContext(detail))
+ }
}
func WriteResponse(w http.ResponseWriter, resp *pb.Response, obj interface{}) {
diff --git a/server/server.go b/server/server.go
index 5d28e74..d84ee90 100644
--- a/server/server.go
+++ b/server/server.go
@@ -21,11 +21,12 @@ import (
"fmt"
"github.com/apache/servicecomb-service-center/pkg/gopool"
"github.com/apache/servicecomb-service-center/pkg/log"
+ nf "github.com/apache/servicecomb-service-center/pkg/notify"
"github.com/apache/servicecomb-service-center/server/core"
"github.com/apache/servicecomb-service-center/server/core/backend"
"github.com/apache/servicecomb-service-center/server/mux"
+ "github.com/apache/servicecomb-service-center/server/notify"
"github.com/apache/servicecomb-service-center/server/plugin"
- nf "github.com/apache/servicecomb-service-center/server/service/notification"
serviceUtil "github.com/apache/servicecomb-service-center/server/service/util"
"github.com/apache/servicecomb-service-center/version"
"github.com/astaxie/beego"
@@ -37,28 +38,22 @@ import (
const buildin = "buildin"
type ServiceCenterServer struct {
- apiServer *APIServer
+ apiService *APIServer
notifyService *nf.NotifyService
- store *backend.KvStore
+ cacheService *backend.KvStore
goroutine *gopool.Pool
}
func (s *ServiceCenterServer) Run() {
s.initialize()
- s.startNotifyService()
-
- s.startApiServer()
+ s.startServices()
s.waitForQuit()
}
func (s *ServiceCenterServer) waitForQuit() {
- var err error
- select {
- case err = <-s.apiServer.Err():
- case err = <-s.notifyService.Err():
- }
+ err := <-s.apiService.Err()
if err != nil {
log.Errorf(err, "service center catch errors")
}
@@ -131,33 +126,38 @@ func (s *ServiceCenterServer) compactBackendService() {
}
func (s *ServiceCenterServer) initialize() {
- s.store = backend.Store()
- s.notifyService = nf.GetNotifyService()
- s.apiServer = GetAPIServer()
+ s.cacheService = backend.Store()
+ s.apiService = GetAPIServer()
+ s.notifyService = notify.NotifyCenter()
s.goroutine = gopool.New(context.Background())
+}
+
+func (s *ServiceCenterServer) startServices() {
+ // notifications
+ s.notifyService.Start()
// load server plugins
plugin.LoadPlugins()
- // cache mechanism
- s.store.Run()
- <-s.store.Ready()
-
+ // check version
if core.ServerInfo.Config.SelfRegister {
- // check version
s.loadOrUpgradeServerVersion()
}
+
+ // cache mechanism
+ s.cacheService.Run()
+ <-s.cacheService.Ready()
+
+ // compact backend automatically
if buildin != beego.AppConfig.DefaultString("registry_plugin", buildin) {
- // compact backend automatically
s.compactBackendService()
}
-}
-func (s *ServiceCenterServer) startNotifyService() {
- s.notifyService.Start()
+ // api service
+ s.startApiService()
}
-func (s *ServiceCenterServer) startApiServer() {
+func (s *ServiceCenterServer) startApiService() {
restIp := beego.AppConfig.String("httpaddr")
restPort := beego.AppConfig.String("httpport")
rpcIp := beego.AppConfig.DefaultString("rpcaddr", "")
@@ -169,22 +169,22 @@ func (s *ServiceCenterServer) startApiServer() {
log.Errorf(err, "parse hostname failed")
}
core.Instance.HostName = host
- s.apiServer.AddListener(REST, restIp, restPort)
- s.apiServer.AddListener(RPC, rpcIp, rpcPort)
- s.apiServer.Start()
+ s.apiService.AddListener(REST, restIp, restPort)
+ s.apiService.AddListener(RPC, rpcIp, rpcPort)
+ s.apiService.Start()
}
func (s *ServiceCenterServer) Stop() {
- if s.apiServer != nil {
- s.apiServer.Stop()
+ if s.apiService != nil {
+ s.apiService.Stop()
}
if s.notifyService != nil {
s.notifyService.Stop()
}
- if s.store != nil {
- s.store.Stop()
+ if s.cacheService != nil {
+ s.cacheService.Stop()
}
s.goroutine.Close(true)
diff --git a/server/service/event/instance_event_handler.go b/server/service/event/instance_event_handler.go
index d526d74..1ea5310 100644
--- a/server/service/event/instance_event_handler.go
+++ b/server/service/event/instance_event_handler.go
@@ -22,10 +22,10 @@ import (
apt "github.com/apache/servicecomb-service-center/server/core"
"github.com/apache/servicecomb-service-center/server/core/backend"
pb "github.com/apache/servicecomb-service-center/server/core/proto"
+ "github.com/apache/servicecomb-service-center/server/notify"
"github.com/apache/servicecomb-service-center/server/plugin/pkg/discovery"
"github.com/apache/servicecomb-service-center/server/service/cache"
"github.com/apache/servicecomb-service-center/server/service/metrics"
- nf "github.com/apache/servicecomb-service-center/server/service/notification"
serviceUtil "github.com/apache/servicecomb-service-center/server/service/util"
"golang.org/x/net/context"
"strings"
@@ -41,25 +41,24 @@ func (h *InstanceEventHandler) Type() discovery.Type {
func (h *InstanceEventHandler) OnEvent(evt discovery.KvEvent) {
action := evt.Type
providerId, providerInstanceId, domainProject := apt.GetInfoFromInstKV(evt.KV.Key)
-
+ idx := strings.Index(domainProject, "/")
+ domainName := domainProject[:idx]
switch action {
case pb.EVT_INIT:
- metrics.ReportInstances(1)
+ metrics.ReportInstances(domainName, 1)
return
case pb.EVT_CREATE:
- metrics.ReportInstances(1)
+ metrics.ReportInstances(domainName, 1)
case pb.EVT_DELETE:
- metrics.ReportInstances(-1)
-
- splited := strings.Split(domainProject, "/")
- if len(splited) == 2 && !apt.IsDefaultDomainProject(domainProject) {
- domainName, projectName := splited[0], splited[1]
+ metrics.ReportInstances(domainName, -1)
+ if !apt.IsDefaultDomainProject(domainProject) {
+ projectName := domainProject[idx+1:]
serviceUtil.RemandInstanceQuota(
util.SetDomainProject(context.Background(), domainName, projectName))
}
}
- if nf.GetNotifyService().Closed() {
+ if notify.NotifyCenter().Closed() {
log.Warnf("caught [%s] instance[%s/%s] event, but notify service is closed",
action, providerId, providerInstanceId)
return
@@ -110,7 +109,7 @@ func PublishInstanceEvent(domainProject string, action pb.EventType, serviceKey
}
for _, consumerId := range subscribers {
// TODO add超时怎么处理？
- job := nf.NewWatchJob(consumerId, apt.GetInstanceRootKey(domainProject)+"/", rev, response)
- nf.GetNotifyService().AddJob(job)
+ job := notify.NewInstanceEvent(consumerId, apt.GetInstanceRootKey(domainProject)+"/", rev, response)
+ notify.NotifyCenter().Publish(job)
}
}
diff --git a/server/service/event/rule_event_handler.go b/server/service/event/rule_event_handler.go
index 095dd0e..58051d9 100644
--- a/server/service/event/rule_event_handler.go
+++ b/server/service/event/rule_event_handler.go
@@ -23,8 +23,8 @@ import (
"github.com/apache/servicecomb-service-center/server/core"
"github.com/apache/servicecomb-service-center/server/core/backend"
pb "github.com/apache/servicecomb-service-center/server/core/proto"
+ "github.com/apache/servicecomb-service-center/server/notify"
"github.com/apache/servicecomb-service-center/server/plugin/pkg/discovery"
- nf "github.com/apache/servicecomb-service-center/server/service/notification"
serviceUtil "github.com/apache/servicecomb-service-center/server/service/util"
"golang.org/x/net/context"
)
@@ -92,7 +92,7 @@ func (h *RuleEventHandler) OnEvent(evt discovery.KvEvent) {
}
providerId, ruleId, domainProject := core.GetInfoFromRuleKV(evt.KV.Key)
- if nf.GetNotifyService().Closed() {
+ if notify.NotifyCenter().Closed() {
log.Warnf("caught [%s] service rule[%s/%s] event, but notify service is closed",
action, providerId, ruleId)
return
diff --git a/server/service/event/service_event_handler.go b/server/service/event/service_event_handler.go
index 20c2953..5573f9f 100644
--- a/server/service/event/service_event_handler.go
+++ b/server/service/event/service_event_handler.go
@@ -43,16 +43,17 @@ func (h *ServiceEventHandler) OnEvent(evt discovery.KvEvent) {
switch evt.Type {
case pb.EVT_INIT, pb.EVT_CREATE:
- metrics.ReportServices(fn, fv, 1)
-
- newDomain := domainProject[:strings.Index(domainProject, "/")]
- newProject := domainProject[strings.Index(domainProject, "/")+1:]
+ idx := strings.Index(domainProject, "/")
+ newDomain := domainProject[:idx]
+ newProject := domainProject[idx+1:]
err := serviceUtil.NewDomainProject(context.Background(), newDomain, newProject)
if err != nil {
log.Errorf(err, "new domain[%s] or project[%s] failed", newDomain, newProject)
}
+ metrics.ReportServices(newDomain, fn, fv, 1)
case pb.EVT_DELETE:
- metrics.ReportServices(fn, fv, -1)
+ domainName := domainProject[:strings.Index(domainProject, "/")]
+ metrics.ReportServices(domainName, fn, fv, -1)
default:
}
diff --git a/server/service/event/tag_event_handler.go b/server/service/event/tag_event_handler.go
index fb7ea03..7059e8f 100644
--- a/server/service/event/tag_event_handler.go
+++ b/server/service/event/tag_event_handler.go
@@ -23,9 +23,9 @@ import (
"github.com/apache/servicecomb-service-center/server/core"
"github.com/apache/servicecomb-service-center/server/core/backend"
pb "github.com/apache/servicecomb-service-center/server/core/proto"
+ "github.com/apache/servicecomb-service-center/server/notify"
"github.com/apache/servicecomb-service-center/server/plugin/pkg/discovery"
"github.com/apache/servicecomb-service-center/server/service/cache"
- nf "github.com/apache/servicecomb-service-center/server/service/notification"
serviceUtil "github.com/apache/servicecomb-service-center/server/service/util"
"golang.org/x/net/context"
)
@@ -106,7 +106,7 @@ func (h *TagEventHandler) OnEvent(evt discovery.KvEvent) {
consumerId, domainProject := core.GetInfoFromTagKV(evt.KV.Key)
- if nf.GetNotifyService().Closed() {
+ if notify.NotifyCenter().Closed() {
log.Warnf("caught [%s] service tags[%s/%s] event, but notify service is closed",
action, consumerId, evt.KV.Value)
return
diff --git a/server/service/instance.go b/server/service/instance.go
index 6f939be..eb98146 100644
--- a/server/service/instance.go
+++ b/server/service/instance.go
@@ -28,6 +28,7 @@ import (
"github.com/apache/servicecomb-service-center/server/core/backend"
pb "github.com/apache/servicecomb-service-center/server/core/proto"
scerr "github.com/apache/servicecomb-service-center/server/error"
+ "github.com/apache/servicecomb-service-center/server/health"
"github.com/apache/servicecomb-service-center/server/plugin"
"github.com/apache/servicecomb-service-center/server/plugin/pkg/quota"
"github.com/apache/servicecomb-service-center/server/plugin/pkg/registry"
@@ -874,6 +875,12 @@ func (s *InstanceService) UpdateInstanceProperties(ctx context.Context, in *pb.U
}
func (s *InstanceService) ClusterHealth(ctx context.Context) (*pb.GetInstancesResponse, error) {
+ if err := health.GlobalHealthChecker().Healthy(); err != nil {
+ return &pb.GetInstancesResponse{
+ Response: pb.CreateResponse(scerr.ErrUnhealthy, err.Error()),
+ }, nil
+ }
+
domainProject := apt.REGISTRY_DOMAIN_PROJECT
serviceId, err := serviceUtil.GetServiceId(ctx, &pb.MicroServiceKey{
AppId: apt.Service.AppId,
diff --git a/server/service/metrics/metrics.go b/server/service/metrics/metrics.go
index c2e3120..05c366e 100644
--- a/server/service/metrics/metrics.go
+++ b/server/service/metrics/metrics.go
@@ -36,7 +36,7 @@ var (
Subsystem: "db",
Name: "service_total",
Help: "Gauge of microservice created in Service Center",
- }, []string{"instance", "framework", "frameworkVersion"})
+ }, []string{"instance", "framework", "frameworkVersion", "domain"})
instanceCounter = prometheus.NewGaugeVec(
prometheus.GaugeOpts{
@@ -44,7 +44,7 @@ var (
Subsystem: "db",
Name: "instance_total",
Help: "Gauge of microservice created in Service Center",
- }, []string{"instance"})
+ }, []string{"instance", "domain"})
)
func init() {
@@ -56,12 +56,12 @@ func ReportDomains(c float64) {
domainCounter.WithLabelValues(instance).Add(c)
}
-func ReportServices(framework, frameworkVersion string, c float64) {
+func ReportServices(domain, framework, frameworkVersion string, c float64) {
instance := metric.InstanceName()
- serviceCounter.WithLabelValues(instance, framework, frameworkVersion).Add(c)
+ serviceCounter.WithLabelValues(instance, framework, frameworkVersion, domain).Add(c)
}
-func ReportInstances(c float64) {
+func ReportInstances(domain string, c float64) {
instance := metric.InstanceName()
- instanceCounter.WithLabelValues(instance).Add(c)
+ instanceCounter.WithLabelValues(instance, domain).Add(c)
}
diff --git a/server/service/watch.go b/server/service/watch.go
index 4e1a629..8b4c039 100644
--- a/server/service/watch.go
+++ b/server/service/watch.go
@@ -20,9 +20,8 @@ import (
"errors"
"github.com/apache/servicecomb-service-center/pkg/log"
"github.com/apache/servicecomb-service-center/pkg/util"
- apt "github.com/apache/servicecomb-service-center/server/core"
pb "github.com/apache/servicecomb-service-center/server/core/proto"
- nf "github.com/apache/servicecomb-service-center/server/service/notification"
+ "github.com/apache/servicecomb-service-center/server/notify"
serviceUtil "github.com/apache/servicecomb-service-center/server/service/util"
"github.com/gorilla/websocket"
"golang.org/x/net/context"
@@ -40,34 +39,31 @@ func (s *InstanceService) WatchPreOpera(ctx context.Context, in *pb.WatchInstanc
}
func (s *InstanceService) Watch(in *pb.WatchInstanceRequest, stream pb.ServiceInstanceCtrl_WatchServer) error {
- var err error
- if err = s.WatchPreOpera(stream.Context(), in); err != nil {
+ log.Infof("new a stream list and watch with service[%s]", in.SelfServiceId)
+ if err := s.WatchPreOpera(stream.Context(), in); err != nil {
log.Errorf(err, "service[%s] establish watch failed: invalid params", in.SelfServiceId)
return err
}
- domainProject := util.ParseDomainProject(stream.Context())
- watcher := nf.NewListWatcher(in.SelfServiceId, apt.GetInstanceRootKey(domainProject)+"/", nil)
- err = nf.GetNotifyService().AddSubscriber(watcher)
- log.Infof("watcher[%s/%s] start watch instance status", watcher.Subject(), watcher.Group())
- return nf.HandleWatchJob(watcher, stream)
+
+ return notify.DoStreamListAndWatch(stream.Context(), in.SelfServiceId, nil, stream)
}
func (s *InstanceService) WebSocketWatch(ctx context.Context, in *pb.WatchInstanceRequest, conn *websocket.Conn) {
log.Infof("new a web socket watch with service[%s]", in.SelfServiceId)
if err := s.WatchPreOpera(ctx, in); err != nil {
- nf.EstablishWebSocketError(conn, err)
+ notify.EstablishWebSocketError(conn, err)
return
}
- nf.DoWebSocketListAndWatch(ctx, in.SelfServiceId, nil, conn)
+ notify.DoWebSocketListAndWatch(ctx, in.SelfServiceId, nil, conn)
}
func (s *InstanceService) WebSocketListAndWatch(ctx context.Context, in *pb.WatchInstanceRequest, conn *websocket.Conn) {
log.Infof("new a web socket list and watch with service[%s]", in.SelfServiceId)
if err := s.WatchPreOpera(ctx, in); err != nil {
- nf.EstablishWebSocketError(conn, err)
+ notify.EstablishWebSocketError(conn, err)
return
}
- nf.DoWebSocketListAndWatch(ctx, in.SelfServiceId, func() ([]*pb.WatchInstanceResponse, int64) {
+ notify.DoWebSocketListAndWatch(ctx, in.SelfServiceId, func() ([]*pb.WatchInstanceResponse, int64) {
return serviceUtil.QueryAllProvidersInstances(ctx, in.SelfServiceId)
}, conn)
}