独立出事件类型处理

Author: yumaojun03

bus/broker/kafka/kafka_test.go

@@ -12,7 +12,7 @@ import (
 )
 
 var (
-	sourceEvent = event.NewEvent()
+	sourceEvent = event.NewDefaultEvent()
 )
 
 func TestPub(t *testing.T) {

bus/broker/kafka/subscriber.go

@@ -128,7 +128,7 @@ func (s *Subscriber) ConsumeClaim(session sarama.ConsumerGroupSession, claim sar
 	// https://github.com/Shopify/sarama/blob/master/consumer_group.go#L27-L29
 	// 具体消费消息
 	for msg := range claim.Messages() {
-		e := event.NewEvent()
+		e := event.NewDefaultEvent()
 		if err := json.Unmarshal(msg.Value, e); err != nil {
 			s.l.Errorf("unmarshal data to event error, %s", err)
 			continue

bus/broker/nats/nats.go

@@ -137,7 +137,7 @@ func (b *Broker) Sub(topic string, h bus.EventHandler) error {
 
 	fn := func(msg *nats.Msg) {
 		b.l.Debugf("receive an message from %s, data: %s", msg.Subject, string(msg.Data))
-		e := event.NewEvent()
+		e := event.NewDefaultEvent()
 		if err := json.Unmarshal(msg.Data, e); err != nil {
 			b.l.Errorf("unmarshal data to event error, %s", err)
 			return

bus/broker/nats/nats_test.go

@@ -16,7 +16,7 @@ func TestPubSub(t *testing.T) {
 	should.NoError(err)
 
 	b.Debug(zap.L().Named("Nats Bus"))
-	sourceEvent := event.NewEvent()
+	sourceEvent := event.NewDefaultEvent()
 
 	should.NoError(b.Connect())
 	err = b.Sub("test", func(topic string, e *event.Event) error {

bus/event/event.go

@@ -2,37 +2,77 @@ package event
 
 import (
 	"context"
+	"encoding/json"
+	"fmt"
 
 	"github.com/rs/xid"
 
 	"github.com/infraboard/mcube/types/ftime"
 )
 
-// 事件主题定义
+// 事件主题定义(由事件类型确定)
 // 1. 资源变更事件 (变更成功和变更失败)
 // 2. 资源告警事件 ( 更加metric计算触发)
 
-// NewEvent 实例
-func NewEvent() *Event {
+// NewOperateEvent 实例
+func NewOperateEvent(header *Header, e *OperateEvent) *Event {
 	return &Event{
-		ID:    xid.New().String(),
-		Time:  ftime.Now(),
-		Label: make(map[string]string),
+		ID:     xid.New().String(),
+		Time:   ftime.Now(),
+		Type:   OperateEventType,
+		Header: header,
+		Body:   e,
+	}
+}
+
+// NewDefaultEvent todo
+func NewDefaultEvent() *Event {
+	return &Event{
+		Header: NewHeader(),
+		Body:   &json.RawMessage{},
 	}
 }
 
 // Event 事件数据结构
 type Event struct {
-	ID     string            `bson:"_id" json:"id"`        // 事件ID
-	Time   ftime.Time        `bson:"time" json:"time"`     // 事件发生时间(毫秒)
-	Source string            `bson:"source" json:"source"` // 事件来源, 比如cmdb
-	Level  Level             `bson:"level" json:"level"`   // 事件等级
-	Label  map[string]string `bson:"label" json:"label"`   // 标签
-	Meta   Meta              `bson:"meta" json:"meta"`     // 事件的元数据
-	Type   Type              `bson:"type" json:"type"`     // 事件类型
-	Body   interface{}       `bson:"body" json:"body"`     // 事件数据
+	ID      string           `bson:"_id" json:"id"`    // 事件ID
+	Time    ftime.Time       `bson:"time" json:"time"` // 事件发生时间(毫秒)
+	Type    Type             `bson:"type" json:"type"` // 事件类型
+	*Header `bson:",inline"` // 事件头
+	Body    interface{}      `bson:"body" json:"body"` // 事件数据
+
+	ctx    context.Context
+	parsed bool
+}
+
+// ParseBody todo
+func (e *Event) ParseBody() error {
+	if e.parsed {
+		return nil
+	}
+
+	body, err := e.getBytesBody()
+	if err != nil {
+		return err
+	}
 
-	ctx context.Context
+	switch e.Type {
+	case OperateEventType:
+		e.Body, err = ParseOperateEventFromBytes(body)
+		if err != nil {
+			return err
+		}
+	case AlertEventType:
+		e.Body, err = ParseAlertEventFromBytes(body)
+		if err != nil {
+			return err
+		}
+	default:
+		return fmt.Errorf("unknown event type: %s", e.Type)
+	}
+
+	e.parsed = true
+	return nil
 }
 
 // WithContext 添加上下文
@@ -45,13 +85,61 @@ func (e *Event) Context() context.Context {
 	return e.ctx
 }
 
+func (e *Event) getBytesBody() ([]byte, error) {
+	switch v := e.Body.(type) {
+	case []byte:
+		return v, nil
+	case json.RawMessage:
+		return v, nil
+	case *json.RawMessage:
+		return *v, nil
+	default:
+		return nil, fmt.Errorf("body type is not []byte or json.RawMessage")
+	}
+}
+
+// NewHeader todo
+func NewHeader() *Header {
+	return &Header{
+		Label: make(map[string]string),
+	}
+}
+
+// Header todo
+type Header struct {
+	Source string            `bson:"source" json:"source"` // 事件来源, 比如cmdb
+	Level  Level             `bson:"level" json:"level"`   // 事件等级
+	Label  map[string]string `bson:"label" json:"label"`   // 标签
+	Meta   Meta              `bson:"meta" json:"meta"`     // 事件的元数据
+}
+
+// ParseOperateEventFromBytes todo
+func ParseOperateEventFromBytes(data []byte) (*OperateEvent, error) {
+	oe := &OperateEvent{}
+	if err := json.Unmarshal(data, oe); err != nil {
+		return nil, err
+	}
+	return oe, nil
+}
+
 // OperateEvent 事件具体数据
 type OperateEvent struct {
-	ResourceType string      `bson:"resource_type" json:"resource_type"` // 资源类型,
-	ResourceUUID string      `bson:"resource_uuid" json:"resource_uuid"` // 资源UUID
-	ResourceName string      `bson:"resource_name" json:"resource_name"` // 资源名称
-	Action       string      `bson:"action" json:"action"`               // 操作
-	Data         interface{} `bson:"data" json:"data,omitempty"`         // 事件数据
+	OperateSession string      `bson:"operate_session" json:"operate_session"` // 回话ID
+	OperateUser    string      `bson:"operate_user" json:"operate_user"`       // 操作人
+	ResourceType   string      `bson:"resource_type" json:"resource_type"`     // 资源类型,
+	ResourceUUID   string      `bson:"resource_uuid" json:"resource_uuid"`     // 资源UUID
+	ResourceName   string      `bson:"resource_name" json:"resource_name"`     // 资源名称
+	Action         string      `bson:"action" json:"action"`                   // 操作
+	Data           interface{} `bson:"data" json:"data,omitempty"`             // 事件数据
+}
+
+// ParseAlertEventFromBytes todo
+func ParseAlertEventFromBytes(data []byte) (*AlertEvent, error) {
+	ae := &AlertEvent{}
+	if err := json.Unmarshal(data, ae); err != nil {
+		return nil, err
+	}
+	return ae, nil
 }
 
 // AlertEvent 事件具体数据

bus/global.go

@@ -1,29 +1,37 @@
 package bus
 
+import (
+	"fmt"
+
+	"github.com/infraboard/mcube/bus/event"
+)
+
 var (
 	publisher  Publisher
 	subscriber Subscriber
 )
 
-// P bus为全局对象
-func P() Publisher {
+// Pub bus为全局对象
+func Pub(e *event.Event) error {
 	if publisher == nil {
-		panic("publisher not initail")
+		return fmt.Errorf("publisher not initail")
 	}
-	return publisher
+
+	return publisher.Pub(e.Type.String(), e)
 }
 
 // SetPublisher 设置pub
 func SetPublisher(p Publisher) {
 	publisher = p
 }
 
-// S bus为全局对象
-func S() Subscriber {
+// Sub bus为全局对象
+func Sub(t event.Type, h EventHandler) error {
 	if subscriber == nil {
-		panic("subscriber not initial")
+		return fmt.Errorf("subscriber not initial")
 	}
-	return subscriber
+
+	return subscriber.Sub(t.String(), h)
 }
 
 // SetSubscriber 设置sub

http/response/event.go

@@ -80,10 +80,5 @@ func Success(w http.ResponseWriter, data interface{}) {
 
 	w.WriteHeader(http.StatusOK)
 	w.Write(respBytes)
-
-	// 发送事件
-	if re, ok := data.(ResourceEvent); HasEventReporter() && ok {
-		sendEvent(re)
-	}
 	return
 }

http/response/reporter.go

@@ -5,7 +5,6 @@ import (
 	"net/http/httptest"
 	"testing"
 
-	broker "github.com/infraboard/mcube/bus/broker/mock"
 	"github.com/infraboard/mcube/exception"
 	"github.com/infraboard/mcube/http/context"
 	"github.com/infraboard/mcube/http/label"
@@ -215,18 +214,3 @@ func TestAPIRootOrderOK(t *testing.T) {
 		should.Equal("/test3", es.Items[2].Path)
 	}
 }
-
-func TestResourceEventOk(t *testing.T) {
-	should := assert.New(t)
-
-	response.SetEventReporter(broker.NewBroker())
-
-	r := httprouter.New()
-	req, _ := http.NewRequest("GET", "/", nil)
-	w := httptest.NewRecorder()
-
-	r.Handle("GET", "/", ResourceEventHandler)
-	r.ServeHTTP(w, req)
-
-	should.Equal(200, w.Code)
-}