对接流程描述

云主机，云硬盘，RDS，Redis，MongoDB的事件数据统一推送至rabbitMQ中类型为fanout的exchange中， 计量计费模块将queue绑定至该exchange，然后进行消费。

目前只接收 资源创建，资源升降配，资源销毁三种事件， 进行后付费的计算。

事件推送规范

vhost name

dcloud

exchange name

DCLOUD_EVENT_EXCHANGE

exchange type

fanout

推送格式示例
  1. {
  2.   "method": "rds_create", //事件类型，以_分割，前面是服务类型，比如rds/mongo/redis/s3/h3pc/h3storage/hwpc/hwstorage，后面是发生的动作，例如create/upgrade/downgrade/delete,create代表资源创建，upgrade代表升配，downgrade代表降配，delete代表资源销毁
  3.   "payload":{
  4.     "occurTime": 112313223, //事件发生的时间，时间戳，精确到毫秒
  5.     "name": "资源名称",
  6.     "ident": "资源英文标识",
  7.     "uuid": "UUID", //每个创建出来的资源实例的唯一id，一定要有，以后这个实例的所有事件都要使用这个resourceId
  8.     "extend": "{
  9.       \"name\": \"rds.m3.c2.d400\", //资源的规格名称，由业务线自定义
  10.       \"cpu\": 1, //cpu个数 ，若没有则写0
  11.       \"mem\": \"4068Gi\", //内存大小，单位必须是Gi，若没有则传0Gi，单位仅支持K，M，G，T，P，Ki,Mi,Gi,Ti,Pi这几种，其他一律视为无效数据
  12.       \"disk\": \"10086002Gi\", //存储容量，单位必须是Gi， 若没有则传0Gi，单位仅支持K，M，G，T，P，Ki,Mi,Gi,Ti,Pi这几种，其他一律视为无效数据
  13.     }", //当前的规格描述，是个字符串,"
  14.     "eventId": "UUID", //为这个事件创建一个唯一的id，一定要有，与实例无关，每个事件的eventId都应该不一样，一样的就会被认为是重复数据
  15.     "labels":"{}", //资源自定义的标签，是个字符串
  16.     "product": "rds/mongo/redis/s3/h3pc/h3storage/hwpc/hwstorage", //标明自己是哪个产品，枚举了字符串类别，与method字段前半部分一致
  17.     "userId": 3, //进行操作的用户id
  18.     "tenantId": 10, //资源所属的租户id
  19.     "projectId": 4, //资源所属的项目id
  20.     "nid":  12 //资源所挂载的服务树节点id
  21.   }
  22. }

推送MQ的demo，仅供参考

  1. package rabbitMQ
  3. import (
  4.     "encoding/json"
  5.     "github.com/streadway/amqp"
  6.     "github.com/toolkits/pkg/logger"
  7.     "golang.org/x/net/context"
  8.     "time"
  9. )
  11. const (
  12.     defaultExchange = "DCLOUD_EVENT_EXCHANGE"
  13. )
  15. type Message []byte
  17. type session struct {
  18.     *amqp.Connection
  19.     *amqp.Channel
  20. }
  22. func (s session) Close() error {
  23.     if s.Connection == nil {
  24.         return nil
  25.     }
  27.     return s.Connection.Close()
  28. }
  30. var (
  31.     RabbitMQ chan chan session
  32.     MQChannel = make(chan Message)
  33. )
  35. func Init() {
  36.     conf := config.GetEnv()
  37.     ctx, _ := context.WithCancel(context.Background())
  38.     exchange := defaultExchange
  39.     if conf.RABBIT_EXCHANGE != "" {
  40.         exchange = conf.RABBIT_EXCHANGE
  41.     }
  42.     RabbitMQ = redial(ctx,
  43.         conf.RABBIT_HOST,
  44.         conf.RABBIT_PORT,
  45.         conf.RABBIT_USERNAME,
  46.         conf.RABBIT_PASSWORD,
  47.         conf.RABBIT_VHOST,
  48.         exchange,
  49.         false,
  50.     )
  52.     routingKey := "ignored for fanout exchanges, application dependent for other exchanges"
  53.     go func(){
  54.         Publish(RabbitMQ, exchange, MQChannel, routingKey)
  55.     }()
  57. }
  59. func redial(ctx context.Context, host string, port string,
  60.     name string, pwd string, vh string, exchange string, isQueue bool) (chan chan session){
  61.     sessions := make(chan chan session)
  63.     go func() {
  64.         sess := make(chan session)
  65.         defer close(sessions)
  67.         for {
  68.             select {
  69.             case sessions <- sess:
  70.             case <-ctx.Done():
  71.                 logger.Infoln("shutting down session factory")
  72.                 return
  73.             }
  75.             conn, err := amqp.Dial("amqp://"+name+":"+pwd+"@"+host+":"+port+"/"+vh)
  76.             if err != nil {
  77.                 logger.Errorf("cannot (re)dial: %v \n", err)
  78.                 continue  //TODO opt logic
  79.             }
  81.             ch, err := conn.Channel()
  82.             if err != nil {
  83.                 logger.Errorf("cannot create channel: %v \n", err)
  84.                 continue
  85.             }
  86.             if isQueue {
  87.                 if _, err := ch.QueueDeclare(exchange, true, false, false, false, nil); err != nil {
  88.                     logger.Errorf("cannot declare single queue: %v \n", err)
  89.                 }
  90.             } else {
  91.                 if err := ch.ExchangeDeclare(exchange, "fanout", true, false, false, false, nil); err != nil {
  92.                     logger.Errorf("cannot declare fanout exchange: %v \n", err)
  93.                 }
  94.             }
  96.             select {
  97.             case sess <- session{conn, ch}:
  98.             case <-ctx.Done():
  99.                 logger.Infoln("shutting down new session")
  100.             }
  101.         }
  102.     }()
  104.     return sessions
  105. }
  107. func Publish(sessions chan chan session, exchange string, messages <-chan Message, key string) {
  108.     for c_session := range sessions {
  109.         var (
  110.             running bool
  111.             reading = messages
  112.             pending = make(chan Message, 1)
  113.             confirm = make(chan amqp.Confirmation, 1)
  114.             pub session
  115.         )
  117.         select {
  118.         case pub = <-c_session:
  119.         case <-time.After(time.Second * time.Duration(20)):
  120.             // timeout means this c_session is unavaliable
  121.             continue
  122.         }
  124.         if err := pub.Confirm(false); err != nil {
  125.             logger.Infoln("publisher confirms not supported")
  126.             close(confirm)
  127.         } else {
  128.             pub.NotifyPublish(confirm)
  129.         }
  131.         logger.Infoln("publishing...")
  133.     Publish:
  134.         for {
  135.             var body Message
  136.             select {
  137.             case confirmed, ok := <-confirm:
  138.                 if !ok {
  139.                     logger.Errorln("rabbitmq error")
  140.                     break Publish
  141.                 }
  143.                 if !confirmed.Ack{
  144.                     logger.Infof("nack message %d, body: %q", confirmed.DeliveryTag, string(body))
  145.                 }
  146.                 reading = messages
  148.             case body = <-pending:
  149.                 //routingKey := "ignored for fanout exchanges, application dependent for other exchanges"
  150.                 err := pub.Publish(exchange, key, false, false, amqp.Publishing{
  151.                     Body:body,
  152.                 })
  153.                 if err != nil {
  154.                     pending <- body
  155.                     pub.Close()
  156.                     logger.Errorln("rabbitmq error")
  157.                     break Publish
  158.                 }
  160.             case body, running = <-reading:
  161.                 if !running {
  162.                     return
  163.                 }
  165.                 pending <- body
  166.                 reading = nil
  168.             }
  169.         }
  170.     }
  172. }
  174. var (
  176.     rmq = rabbitMQ.RabbitMQ
  177.     mq = rabbitMQ.MQChannel
  179. )
  181. func FanoutArchitecture(m []byte) {
  182.     logger.Infof("push message: %s", m)
  183.     go func() {
  184.         PutChan(m)
  185.         //rabbitMQ.Publish(rmq, ChanByte(m))
  186.     }()
  187. }
  189. func PutChan(b []byte) {
  190.     mq <- b
  191. }
  193. func SendMsg(action string, msg interface{}) {
  194.     js := EncodeStruct(action, msg)
  195.     FanoutArchitecture(js)
  196. }
  199. func main() {
  200.     go SendMsg("test", nil)
  201. }

rabbitMQ测试地址

amqp://admin:admin_202009@10.178.24.110:8002/