(ru *common.TaskStatistics)
| 246 | } |
| 247 | |
| 248 | func (h *taskHandle) emitStats(ru *common.TaskStatistics) { |
| 249 | const srcFullFactor float32 = 4.5 |
| 250 | const dstFullFactor float32 = 5 |
| 251 | const srcIncrFactor float32 = 19 |
| 252 | const dstIncrFactor float32 = 9.5 |
| 253 | |
| 254 | labels := []metrics.Label{{"task_name", fmt.Sprintf("%s_%s", h.taskConfig.JobName, h.taskConfig.Name)}} |
| 255 | |
| 256 | metrics.SetGaugeWithLabels([]string{"network", "in_msgs"}, float32(ru.MsgStat.InMsgs), labels) |
| 257 | metrics.SetGaugeWithLabels([]string{"network", "out_msgs"}, float32(ru.MsgStat.OutMsgs), labels) |
| 258 | metrics.SetGaugeWithLabels([]string{"network", "in_bytes"}, float32(ru.MsgStat.InBytes), labels) |
| 259 | metrics.SetGaugeWithLabels([]string{"network", "out_bytes"}, float32(ru.MsgStat.OutBytes), labels) |
| 260 | switch common.TaskTypeFromString(h.taskConfig.Name) { |
| 261 | case common.TaskTypeSrc: |
| 262 | metrics.SetGaugeWithLabels([]string{"buffer", "event_queue_size"}, float32(ru.BufferStat.BinlogEventQueueSize), labels) |
| 263 | metrics.SetGaugeWithLabels([]string{"buffer", "src_queue_size"}, float32(ru.BufferStat.ExtractorTxQueueSize), labels) |
| 264 | metrics.SetGaugeWithLabels([]string{"buffer", "send_by_timeout"}, float32(ru.BufferStat.SendByTimeout), labels) |
| 265 | metrics.SetGaugeWithLabels([]string{"buffer", "send_by_size_full"}, float32(ru.BufferStat.SendBySizeFull), labels) |
| 266 | |
| 267 | metrics.SetGaugeWithLabels([]string{"memory.full_kb_est"}, float32(ru.MemoryStat.Full)*srcFullFactor/1024, labels) |
| 268 | metrics.SetGaugeWithLabels([]string{"memory.incr_kb_est"}, float32(ru.MemoryStat.Incr)*srcIncrFactor/1024, labels) |
| 269 | case common.TaskTypeDest: |
| 270 | metrics.SetGaugeWithLabels([]string{"buffer", "dest_queue_size"}, float32(ru.BufferStat.ApplierMsgQueueSize), labels) |
| 271 | metrics.SetGaugeWithLabels([]string{"buffer", "dest_queue2_size"}, float32(ru.BufferStat.ApplierTxQueueSize), labels) |
| 272 | |
| 273 | metrics.SetGaugeWithLabels([]string{"memory.full_kb_est"}, float32(ru.MemoryStat.Full)*dstFullFactor/1024, labels) |
| 274 | metrics.SetGaugeWithLabels([]string{"memory.incr_kb_est"}, float32(ru.MemoryStat.Incr)*dstIncrFactor/1024, labels) |
| 275 | case common.TaskTypeUnknown: |
| 276 | } |
| 277 | |
| 278 | metrics.SetGaugeWithLabels([]string{"memory.full_kb_count"}, float32(ru.MemoryStat.Full)/1024, labels) |
| 279 | metrics.SetGaugeWithLabels([]string{"memory.incr_kb_count"}, float32(ru.MemoryStat.Incr)/1024, labels) |
| 280 | |
| 281 | if ru.TableStats != nil { |
| 282 | metrics.SetGaugeWithLabels([]string{"table", "insert"}, float32(ru.TableStats.InsertCount), labels) |
| 283 | metrics.SetGaugeWithLabels([]string{"table", "update"}, float32(ru.TableStats.UpdateCount), labels) |
| 284 | metrics.SetGaugeWithLabels([]string{"table", "delete"}, float32(ru.TableStats.DelCount), labels) |
| 285 | } |
| 286 | |
| 287 | if ru.DelayCount != nil { |
| 288 | // TODO |
| 289 | //metrics.SetGaugeWithLabels([]string{"delay", "num"}, float32(ru.DelayCount.Num), labels) |
| 290 | metrics.SetGaugeWithLabels([]string{"delay", "time"}, float32(ru.DelayCount.Time), labels) |
| 291 | } |
| 292 | |
| 293 | if ru.ThroughputStat != nil { |
| 294 | metrics.SetGaugeWithLabels([]string{"throughput", "num"}, float32(ru.ThroughputStat.Num), labels) |
| 295 | metrics.SetGaugeWithLabels([]string{"throughput", "time"}, float32(ru.ThroughputStat.Time), labels) |
| 296 | } |
| 297 | |
| 298 | if nil != ru.HandledTxCount.AppliedTxCount { |
| 299 | metrics.SetGaugeWithLabels([]string{"dest_applied_incr_tx_count"}, float32(*ru.HandledTxCount.AppliedTxCount), labels) |
| 300 | } |
| 301 | if nil != ru.HandledTxCount.ExtractedTxCount { |
| 302 | metrics.SetGaugeWithLabels([]string{"src_extracted_incr_tx_count"}, float32(*ru.HandledTxCount.ExtractedTxCount), labels) |
| 303 | } |
| 304 | |
| 305 | if nil != ru.HandledQueryCount.AppliedQueryCount { |
no test coverage detected