volc.go 4.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144
  1. // Copyright 2019 Yunion
  2. //
  3. // Licensed under the Apache License, Version 2.0 (the "License");
  4. // you may not use this file except in compliance with the License.
  5. // You may obtain a copy of the License at
  6. //
  7. // http://www.apache.org/licenses/LICENSE-2.0
  8. //
  9. // Unless required by applicable law or agreed to in writing, software
  10. // distributed under the License is distributed on an "AS IS" BASIS,
  11. // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12. // See the License for the specific language governing permissions and
  13. // limitations under the License.
  14. package providerdriver
  15. import (
  16. "context"
  17. "strconv"
  18. "strings"
  19. "sync"
  20. "time"
  21. "yunion.io/x/cloudmux/pkg/cloudprovider"
  22. "yunion.io/x/log"
  23. "yunion.io/x/pkg/errors"
  24. api "yunion.io/x/onecloud/pkg/apis/compute"
  25. "yunion.io/x/onecloud/pkg/util/influxdb"
  26. )
  27. type VolcEngineCollect struct {
  28. SBaseCollectDriver
  29. }
  30. func (self *VolcEngineCollect) GetProvider() string {
  31. return api.CLOUD_PROVIDER_VOLCENGINE
  32. }
  33. func (self *VolcEngineCollect) IsSupportMetrics() bool {
  34. return true
  35. }
  36. func init() {
  37. Register(&VolcEngineCollect{})
  38. }
  39. func (self *VolcEngineCollect) CollectServerMetrics(ctx context.Context, manager api.CloudproviderDetails, provider cloudprovider.ICloudProvider, res map[string]api.ServerDetails, start, end time.Time) error {
  40. metrics := []influxdb.SMetricData{}
  41. regionServers := map[string][]api.ServerDetails{}
  42. for i := range res {
  43. regionId := res[i].RegionExtId
  44. _, ok := regionServers[regionId]
  45. if !ok {
  46. regionServers[regionId] = []api.ServerDetails{}
  47. }
  48. regionServers[regionId] = append(regionServers[regionId], res[i])
  49. }
  50. var wg sync.WaitGroup
  51. var mu sync.Mutex
  52. for regionId, servers := range regionServers {
  53. wg.Add(1)
  54. go func(regionId string, servers []api.ServerDetails) {
  55. defer func() {
  56. wg.Done()
  57. }()
  58. data := []cloudprovider.MetricValues{}
  59. for i := 0; i < (len(servers)+9)/10; i++ {
  60. for _, metricType := range cloudprovider.ALL_VM_METRIC_TYPES {
  61. opts := &cloudprovider.MetricListOptions{
  62. ResourceType: cloudprovider.METRIC_RESOURCE_TYPE_SERVER,
  63. RegionExtId: regionId,
  64. MetricType: metricType,
  65. StartTime: start,
  66. EndTime: end,
  67. }
  68. last := (i + 1) * 10
  69. if last > len(servers) {
  70. last = len(servers)
  71. }
  72. for j := range servers[i*10 : last] {
  73. opts.ResourceIds = append(opts.ResourceIds, servers[i*10+j].ExternalId)
  74. }
  75. part, err := provider.GetMetrics(opts)
  76. if err != nil {
  77. if errors.Cause(err) != cloudprovider.ErrNotImplemented && errors.Cause(err) != cloudprovider.ErrNotSupported {
  78. log.Errorf("get server %s(%s) error: %v", strings.Join(opts.ResourceIds, ","), regionId, err)
  79. continue
  80. }
  81. continue
  82. }
  83. data = append(data, part...)
  84. }
  85. }
  86. for _, value := range data {
  87. server, ok := res[value.Id]
  88. if !ok {
  89. continue
  90. }
  91. tags := []influxdb.SKeyValue{}
  92. for k, v := range server.GetMetricTags() {
  93. tags = append(tags, influxdb.SKeyValue{
  94. Key: k,
  95. Value: v,
  96. })
  97. }
  98. pairs := []influxdb.SKeyValue{}
  99. for k, v := range server.GetMetricPairs() {
  100. pairs = append(pairs, influxdb.SKeyValue{
  101. Key: k,
  102. Value: v,
  103. })
  104. }
  105. for _, v := range value.Values {
  106. metric := influxdb.SMetricData{
  107. Name: value.MetricType.Name(),
  108. Timestamp: v.Timestamp,
  109. Tags: []influxdb.SKeyValue{},
  110. Metrics: []influxdb.SKeyValue{
  111. {
  112. Key: value.MetricType.Key(),
  113. Value: strconv.FormatFloat(v.Value, 'E', -1, 64),
  114. },
  115. },
  116. }
  117. for k, v := range v.Tags {
  118. metric.Tags = append(metric.Tags, influxdb.SKeyValue{
  119. Key: k,
  120. Value: v,
  121. })
  122. }
  123. metric.Metrics = append(metric.Metrics, pairs...)
  124. metric.Tags = append(metric.Tags, tags...)
  125. mu.Lock()
  126. metrics = append(metrics, metric)
  127. mu.Unlock()
  128. }
  129. }
  130. }(regionId, servers)
  131. }
  132. wg.Wait()
  133. return self.sendMetrics(ctx, manager, "server", len(res), metrics)
  134. }