overflow_cloud/consumer_metric
3
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

ing

Browse Source
This commit is contained in:
crusader 2018-07-05 17:18:14 +09:00
parent 6fa4e03ab3
commit 36d04bd10b
1 changed files with 20 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
consumer/consumer-handler.go
View File

@ -102,6 +102,8 @@ func (ch *ConsumerHandlers) OnMessage(msg *kafka.Message) {
return return
} }
ch.processSensoring(metric, bp)
for _, sensorItem := range sensorItems { for _, sensorItem := range sensorItems {
if err := ch.processSensorItem(sensorItem, metric, bp); nil != err { if err := ch.processSensorItem(sensorItem, metric, bp); nil != err {
logging.Logger().Error(err) logging.Logger().Error(err)
@ -112,7 +114,25 @@ func (ch *ConsumerHandlers) OnMessage(msg *kafka.Message) {
if err := c.Write(bp); err != nil { if err := c.Write(bp); err != nil {
logging.Logger().Error(err) logging.Logger().Error(err)
} }
}
func (ch *ConsumerHandlers) processSensoring(metric *ocmd.Metric, bp client.BatchPoints) error {
tags := map[string]string{
"SensorID": metric.SensorID,
}
fields := map[string]interface{}{
"StartDate": metric.StartDate,
"EndDate": metric.EndDate,
"Success": metric.Success,
}
pt, err := client.NewPoint("sensor", tags, fields, time.Now())
if nil != err {
return err
}
bp.AddPoint(pt)
return nil
} }
func (ch *ConsumerHandlers) processSensorItem(sensorItem *ocms.SensorItem, metric *ocmd.Metric, bp client.BatchPoints) error { func (ch *ConsumerHandlers) processSensorItem(sensorItem *ocms.SensorItem, metric *ocmd.Metric, bp client.BatchPoints) error {