mirror of
https://github.com/yunionio/cloudpods.git
synced 2026-10-24 23:41:52 +08:00
feat(region): support hcsop
This commit is contained in:
@@ -22,14 +22,6 @@ type CloudMonOptions struct {
|
||||
common_options.CommonOptions
|
||||
PingProbeOptions
|
||||
|
||||
EndpointType string `default:"internalURL" help:"Defaults to internalURL" choices:"publicURL|internalURL|adminURL"`
|
||||
ApiVersion string `help:"override default modules service api version"`
|
||||
|
||||
ReqTimeout int `default:"600" help:"Number of seconds to wait for a response"`
|
||||
Insecure bool `default:"true" help:"Allow skip server cert verification if URL is https" short-token:"k"`
|
||||
CertFile string `help:"certificate file"`
|
||||
KeyFile string `help:"private key file"`
|
||||
|
||||
ResourcesSyncInterval int64 `help:"Increment Sync Interval unit:minute" default:"10"`
|
||||
CollectMetricInterval int64 `help:"Increment Sync Interval unit:minute" default:"6"`
|
||||
SkipMetricPullProviders string `help:"Skip indicate provider metric pull" default:""`
|
||||
@@ -70,6 +62,15 @@ func OnOptionsChange(oldO, newO interface{}) bool {
|
||||
changed = true
|
||||
}
|
||||
}
|
||||
if oldOpts.ResourcesSyncInterval != newOpts.ResourcesSyncInterval {
|
||||
changed = true
|
||||
}
|
||||
if oldOpts.CollectMetricInterval != newOpts.CollectMetricInterval {
|
||||
changed = true
|
||||
}
|
||||
if oldOpts.CollectServiceMetricIntervalMinute != newOpts.CollectServiceMetricIntervalMinute {
|
||||
changed = true
|
||||
}
|
||||
|
||||
return changed
|
||||
}
|
||||
|
||||
@@ -58,6 +58,10 @@ func (self *SBaseCollectDriver) CollectHostMetrics(ctx context.Context, manager
|
||||
return cloudprovider.ErrNotImplemented
|
||||
}
|
||||
|
||||
func (self *SBaseCollectDriver) CollectWireMetrics(ctx context.Context, manager api.CloudproviderDetails, provider cloudprovider.ICloudProvider, res map[string]api.WireDetails, start, end time.Time) error {
|
||||
return cloudprovider.ErrNotImplemented
|
||||
}
|
||||
|
||||
func (self *SBaseCollectDriver) CollectStorageMetrics(ctx context.Context, manager api.CloudproviderDetails, provider cloudprovider.ICloudProvider, res map[string]api.StorageDetails, start, end time.Time) error {
|
||||
metrics := []influxdb.SMetricData{}
|
||||
for _, storage := range res {
|
||||
@@ -317,7 +321,7 @@ func (self *SCollectByResourceIdDriver) CollectHostMetrics(ctx context.Context,
|
||||
Tags: []influxdb.SKeyValue{},
|
||||
Metrics: []influxdb.SKeyValue{
|
||||
{
|
||||
Key: values.MetricType.Name(),
|
||||
Key: values.MetricType.Key(),
|
||||
Value: strconv.FormatFloat(value.Value, 'E', -1, 64),
|
||||
},
|
||||
},
|
||||
|
||||
@@ -41,6 +41,7 @@ type ICollectDriver interface {
|
||||
CollectBucketMetrics(ctx context.Context, manager api.CloudproviderDetails, provider cloudprovider.ICloudProvider, res map[string]api.BucketDetails, start, end time.Time) error
|
||||
CollectK8sMetrics(ctx context.Context, manager api.CloudproviderDetails, provider cloudprovider.ICloudProvider, res map[string]api.KubeClusterDetails, start, end time.Time) error
|
||||
CollectModelartsPoolMetrics(ctx context.Context, manager api.CloudproviderDetails, provider cloudprovider.ICloudProvider, res map[string]api.ModelartsPoolDetails, start, end time.Time) error
|
||||
CollectWireMetrics(ctx context.Context, manager api.CloudproviderDetails, provider cloudprovider.ICloudProvider, res map[string]api.WireDetails, start, end time.Time) error
|
||||
}
|
||||
|
||||
func GetDriver(name string) (ICollectDriver, error) {
|
||||
|
||||
@@ -0,0 +1,139 @@
|
||||
// Copyright 2019 Yunion
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package providerdriver
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strconv"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"yunion.io/x/cloudmux/pkg/cloudprovider"
|
||||
"yunion.io/x/log"
|
||||
"yunion.io/x/pkg/errors"
|
||||
|
||||
api "yunion.io/x/onecloud/pkg/apis/compute"
|
||||
"yunion.io/x/onecloud/pkg/cloudmon/options"
|
||||
"yunion.io/x/onecloud/pkg/util/influxdb"
|
||||
)
|
||||
|
||||
type HcsOpCollect struct {
|
||||
SCollectByResourceIdDriver
|
||||
}
|
||||
|
||||
func (self *HcsOpCollect) GetProvider() string {
|
||||
return api.CLOUD_PROVIDER_HCSOP
|
||||
}
|
||||
|
||||
func init() {
|
||||
Register(&HcsOpCollect{})
|
||||
}
|
||||
|
||||
func (self *HcsOpCollect) IsSupportMetrics() bool {
|
||||
return true
|
||||
}
|
||||
|
||||
func (self *HcsOpCollect) CollectStorageMetrics(ctx context.Context, manager api.CloudproviderDetails, provider cloudprovider.ICloudProvider, res map[string]api.StorageDetails, start, end time.Time) error {
|
||||
metrics := []influxdb.SMetricData{}
|
||||
for _, storage := range res {
|
||||
metric := influxdb.SMetricData{
|
||||
Name: string(cloudprovider.METRIC_RESOURCE_TYPE_STORAGE),
|
||||
Timestamp: time.Now(),
|
||||
Tags: []influxdb.SKeyValue{},
|
||||
Metrics: []influxdb.SKeyValue{},
|
||||
}
|
||||
for k, v := range storage.GetMetricTags() {
|
||||
metric.Tags = append(metric.Tags, influxdb.SKeyValue{
|
||||
Key: k,
|
||||
Value: v,
|
||||
})
|
||||
}
|
||||
for k, v := range storage.GetMetricPairs() {
|
||||
metric.Metrics = append(metric.Metrics, influxdb.SKeyValue{
|
||||
Key: k,
|
||||
Value: v,
|
||||
})
|
||||
}
|
||||
metrics = append(metrics, metric)
|
||||
}
|
||||
return self.sendMetrics(ctx, manager, "storage", len(res), metrics)
|
||||
}
|
||||
|
||||
func (self *HcsOpCollect) CollectWireMetrics(ctx context.Context, manager api.CloudproviderDetails, provider cloudprovider.ICloudProvider, res map[string]api.WireDetails, start, end time.Time) error {
|
||||
ch := make(chan struct{}, options.Options.CloudResourceCollectMetricsBatchCount)
|
||||
defer close(ch)
|
||||
metrics := []influxdb.SMetricData{}
|
||||
var wg sync.WaitGroup
|
||||
var mu sync.Mutex
|
||||
for i := range res {
|
||||
ch <- struct{}{}
|
||||
wg.Add(1)
|
||||
go func(wire api.WireDetails) {
|
||||
defer func() {
|
||||
wg.Done()
|
||||
<-ch
|
||||
}()
|
||||
opts := &cloudprovider.MetricListOptions{
|
||||
ResourceType: cloudprovider.METRIC_RESOURCE_TYPE_WIRE,
|
||||
StartTime: start,
|
||||
EndTime: end,
|
||||
}
|
||||
opts.ResourceId = wire.ExternalId
|
||||
|
||||
tags := []influxdb.SKeyValue{}
|
||||
for k, v := range wire.GetMetricTags() {
|
||||
tags = append(tags, influxdb.SKeyValue{
|
||||
Key: k,
|
||||
Value: v,
|
||||
})
|
||||
}
|
||||
data, err := provider.GetMetrics(opts)
|
||||
if err != nil {
|
||||
if errors.Cause(err) != cloudprovider.ErrNotImplemented && errors.Cause(err) != cloudprovider.ErrNotSupported {
|
||||
log.Errorf("get rds %s(%s) error: %v", wire.Name, wire.Id, err)
|
||||
return
|
||||
}
|
||||
return
|
||||
}
|
||||
for _, values := range data {
|
||||
for _, value := range values.Values {
|
||||
metric := influxdb.SMetricData{
|
||||
Name: values.MetricType.Name(),
|
||||
Timestamp: value.Timestamp,
|
||||
Tags: tags,
|
||||
Metrics: []influxdb.SKeyValue{
|
||||
{
|
||||
Key: values.MetricType.Key(),
|
||||
Value: strconv.FormatFloat(value.Value, 'E', -1, 64),
|
||||
},
|
||||
},
|
||||
}
|
||||
for k, v := range value.Tags {
|
||||
metric.Tags = append(metric.Tags, influxdb.SKeyValue{
|
||||
Key: k,
|
||||
Value: v,
|
||||
})
|
||||
}
|
||||
mu.Lock()
|
||||
metrics = append(metrics, metric)
|
||||
mu.Unlock()
|
||||
}
|
||||
}
|
||||
}(res[i])
|
||||
}
|
||||
wg.Wait()
|
||||
|
||||
return self.sendMetrics(ctx, manager, "wire", len(res), metrics)
|
||||
}
|
||||
@@ -345,6 +345,7 @@ type SResources struct {
|
||||
KubeClusters TResource
|
||||
Storages TResource
|
||||
ModelartsPool TResource
|
||||
Wires TResource
|
||||
}
|
||||
|
||||
func NewResources() *SResources {
|
||||
@@ -360,6 +361,7 @@ func NewResources() *SResources {
|
||||
Buckets: NewBaseResources(&compute.Buckets),
|
||||
KubeClusters: NewBaseResources(&compute.KubeClusters),
|
||||
ModelartsPool: NewBaseResources(&compute.ModelartsPools),
|
||||
Wires: NewBaseResources(&compute.Wires),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -411,6 +413,10 @@ func (self *SResources) Init(ctx context.Context, userCred mcclient.TokenCredent
|
||||
if err != nil {
|
||||
errs = append(errs, errors.Wrapf(err, "ModelartsPool.init"))
|
||||
}
|
||||
err = self.Wires.init()
|
||||
if err != nil {
|
||||
errs = append(errs, errors.Wrapf(err, "Wires.init"))
|
||||
}
|
||||
return errors.NewAggregate(errs)
|
||||
}()
|
||||
if err != nil {
|
||||
@@ -469,6 +475,10 @@ func (self *SResources) IncrementSync(ctx context.Context, userCred mcclient.Tok
|
||||
if err != nil {
|
||||
errs = append(errs, errors.Wrapf(err, "ModelartsPool.increment"))
|
||||
}
|
||||
err = self.Wires.increment()
|
||||
if err != nil {
|
||||
errs = append(errs, errors.Wrapf(err, "Wires.increment"))
|
||||
}
|
||||
return errors.NewAggregate(errs)
|
||||
}()
|
||||
if err != nil {
|
||||
@@ -526,6 +536,10 @@ func (self *SResources) DecrementSync(ctx context.Context, userCred mcclient.Tok
|
||||
if err != nil {
|
||||
errs = append(errs, errors.Wrapf(err, "ModelartsPool.decrement"))
|
||||
}
|
||||
err = self.Wires.decrement()
|
||||
if err != nil {
|
||||
errs = append(errs, errors.Wrapf(err, "ModelartsPool.decrement"))
|
||||
}
|
||||
return errors.NewAggregate(errs)
|
||||
}()
|
||||
if err != nil {
|
||||
@@ -730,6 +744,19 @@ func (self *SResources) CollectMetrics(ctx context.Context, userCred mcclient.To
|
||||
if err != nil && errors.Cause(err) != cloudprovider.ErrNotImplemented && errors.Cause(err) != cloudprovider.ErrNotSupported {
|
||||
log.Errorf("CollectModelartsPoolMetrics for %s(%s) error: %v", manager.Name, manager.Provider, err)
|
||||
}
|
||||
|
||||
resources = self.Wires.getResources(manager.Id)
|
||||
wires := map[string]api.WireDetails{}
|
||||
err = jsonutils.Update(&wires, resources)
|
||||
if err != nil {
|
||||
log.Errorf("unmarsha wires resources error: %v", err)
|
||||
}
|
||||
|
||||
err = driver.CollectWireMetrics(ctx, manager, provider, wires, startTime, endTime)
|
||||
if err != nil && errors.Cause(err) != cloudprovider.ErrNotImplemented && errors.Cause(err) != cloudprovider.ErrNotSupported {
|
||||
log.Errorf("CollectWireMetrics for %s(%s) error: %v", manager.Name, manager.Provider, err)
|
||||
}
|
||||
|
||||
}(cloudproviders[i])
|
||||
}
|
||||
wg.Wait()
|
||||
|
||||
Reference in New Issue
Block a user