14831633718

Committed 05 May 2025 07:48AM UTC coverage: 78.206% (+1.0%) from 77.223%

Build # 14831633718

Build Type

Pull #708

github

Committed by

Varsius

Commit Message

replace resource data types by liquid reports

Pull Request Pull Request #708: replace resource data types by liquid reports

Run Details

130 of 163 new or added lines in 6 files covered. (79.75%)

3 existing lines in 2 files now uncovered.

6355 of 8126 relevant lines covered (78.21%)

46.78 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

41.04

/internal/plugins/capacity_liquid.go

/*******************************************************************************
*
* Copyright 2024 SAP SE
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You should have received a copy of the License along with this
* program. If not, 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 plugins

import (
        "context"
        "errors"
        "fmt"
        "slices"

        "github.com/gophercloud/gophercloud/v2"
        . "github.com/majewsky/gg/option"
        "github.com/prometheus/common/model"
        "github.com/sapcc/go-api-declarations/limes"
        "github.com/sapcc/go-api-declarations/liquid"
        "github.com/sapcc/go-bits/liquidapi"
        "github.com/sapcc/go-bits/logg"
        "github.com/sapcc/go-bits/promquery"

        "github.com/sapcc/limes/internal/core"
        "github.com/sapcc/limes/internal/db"
)

type prometheusCapacityConfiguration struct {
        APIConfig         promquery.Config               `yaml:"api"`
        Queries           map[liquid.ResourceName]string `yaml:"queries"`
        AllowZeroCapacity bool                           `yaml:"allow_zero_capacity"`
}

type fixedCapacityConfiguration struct {
        Values map[liquid.ResourceName]uint64 `yaml:"values"`
}

type LiquidCapacityPlugin struct {
        // configuration
        ServiceType                     db.ServiceType                          `yaml:"service_type"`
        LiquidServiceType               string                                  `yaml:"liquid_service_type"`
        FixedCapacityConfiguration      Option[fixedCapacityConfiguration]      `yaml:"fixed_capacity_values"`
        PrometheusCapacityConfiguration Option[prometheusCapacityConfiguration] `yaml:"capacity_values_from_prometheus"`

        // state
        LiquidServiceInfo liquid.ServiceInfo `yaml:"-"`
        LiquidClient      core.LiquidClient  `yaml:"-"`
}

func init() {
        core.CapacityPluginRegistry.Add(func() core.CapacityPlugin { return &LiquidCapacityPlugin{} })
}

// PluginTypeID implements the core.CapacityPlugin interface.
func (p *LiquidCapacityPlugin) PluginTypeID() string {
        return "liquid"
}

// Init implements the core.CapacityPlugin interface.
func (p *LiquidCapacityPlugin) Init(ctx context.Context, client *gophercloud.ProviderClient, eo gophercloud.EndpointOpts) (err error) {
        if p.ServiceType == "" {
                return errors.New("missing required value: params.service_type")
        }
        if p.LiquidServiceType == "" {
                p.LiquidServiceType = "liquid-" + string(p.ServiceType)
        }

        p.LiquidClient, err = core.NewLiquidClient(client, eo, liquidapi.ClientOpts{ServiceType: p.LiquidServiceType})
        if err != nil {
                return err
        }

        p.LiquidServiceInfo, err = p.LiquidClient.GetInfo(ctx)
        if err != nil {
                return err
        }
        err = liquid.ValidateServiceInfo(p.LiquidServiceInfo)
        return err
}

// ServiceInfo implements the core.CapacityPlugin interface.
func (p *LiquidCapacityPlugin) ServiceInfo() liquid.ServiceInfo {
        return p.LiquidServiceInfo
}

// Scrape implements the core.CapacityPlugin interface.
// we assume that each resource only has one source of info (else, the last source wins and overwrites the previous data)
// we fail the whole collection explicitly in case any of the sources fails
func (p *LiquidCapacityPlugin) Scrape(ctx context.Context, backchannel core.CapacityPluginBackchannel, allAZs []limes.AvailabilityZone) (result liquid.ServiceCapacityReport, err error) {
        req, err := p.BuildServiceCapacityRequest(backchannel, allAZs)
        if err != nil {
                return liquid.ServiceCapacityReport{}, err
        }

        result, err = p.LiquidClient.GetCapacityReport(ctx, req)
        if err != nil {
                return liquid.ServiceCapacityReport{}, err
        }
        if result.InfoVersion != p.LiquidServiceInfo.Version {
                logg.Fatal("ServiceInfo version for %s changed from %d to %d; restarting now to reload ServiceInfo...",
                        p.LiquidServiceType, p.LiquidServiceInfo.Version, result.InfoVersion)
        }
        err = liquid.ValidateServiceInfo(p.LiquidServiceInfo)
        if err != nil {
                return liquid.ServiceCapacityReport{}, err
        }
        err = liquid.ValidateCapacityReport(result, req, p.LiquidServiceInfo)
        if err != nil {
                return liquid.ServiceCapacityReport{}, err
        }

        // manual capacity collection
        fixedCapaConfig, exists := p.FixedCapacityConfiguration.Unpack()
        if exists {
                if result.Resources == nil {
                        result.Resources = make(map[liquid.ResourceName]*liquid.ResourceCapacityReport)
                }
                for resName, capacity := range fixedCapaConfig.Values {
                        result.Resources[resName] = &liquid.ResourceCapacityReport{
                                PerAZ: liquid.InAnyAZ(liquid.AZResourceCapacityReport{Capacity: capacity}),
                        }
                }
        }

        // prometheus capacity collection
        prometheusCapaConfig, exists := p.PrometheusCapacityConfiguration.Unpack()
        if exists {
                if result.Resources == nil {
                        result.Resources = make(map[liquid.ResourceName]*liquid.ResourceCapacityReport)
                }
                client, err := prometheusCapaConfig.APIConfig.Connect()
                if err != nil {
                        return liquid.ServiceCapacityReport{}, err
                }
                for resName, query := range prometheusCapaConfig.Queries {
                        azReports, err := prometheusCapaConfig.prometheusScrapeOneResource(ctx, client, query, allAZs)
                        if err != nil {
                                return liquid.ServiceCapacityReport{}, fmt.Errorf("while scraping prometheus capacity %q/%q: %w", p.ServiceType, resName, err)
                        }
                        result.Resources[resName] = &liquid.ResourceCapacityReport{
                                PerAZ: azReports,
                        }
                }
        }

        return result, nil
}

func (p *LiquidCapacityPlugin) BuildServiceCapacityRequest(backchannel core.CapacityPluginBackchannel, allAZs []limes.AvailabilityZone) (liquid.ServiceCapacityRequest, error) {
        req := liquid.ServiceCapacityRequest{
                AllAZs:           allAZs,
                DemandByResource: make(map[liquid.ResourceName]liquid.ResourceDemand, len(p.LiquidServiceInfo.Resources)),
        }

        var err error
        for resName, resInfo := range p.LiquidServiceInfo.Resources {
                if !resInfo.HasCapacity {
                        continue
                }
                if !resInfo.NeedsResourceDemand {
                        continue
                }
                req.DemandByResource[resName], err = backchannel.GetResourceDemand(p.ServiceType, resName)
                if err != nil {
                        return liquid.ServiceCapacityRequest{}, fmt.Errorf("while getting resource demand for %s/%s: %w", p.ServiceType, resName, err)
                }
        }
        return req, nil
}

func (p prometheusCapacityConfiguration) prometheusScrapeOneResource(ctx context.Context, client promquery.Client, query string, allAZs []limes.AvailabilityZone) (map[liquid.AvailabilityZone]*liquid.AZResourceCapacityReport, error) {
        vector, err := client.GetVector(ctx, query)
        if err != nil {
                return nil, err
        }

        // for known AZs, we expect exactly one result;
        // all unknown AZs get lumped into AvailabilityZoneUnknown
        matchedSamples := make(map[limes.AvailabilityZone]*model.Sample)
        var unmatchedSamples []*model.Sample
        for _, sample := range vector {
                az := limes.AvailabilityZone(sample.Metric["az"])
                switch {
                case az == "":
                        return nil, fmt.Errorf(`missing label "az" on metric %v = %g`, sample.Metric, sample.Value)
                case slices.Contains(allAZs, az) || az == limes.AvailabilityZoneAny:
                        if matchedSamples[az] != nil {
                                other := matchedSamples[az]
                                return nil, fmt.Errorf(`multiple samples for az=%q: found %v = %g and %v = %g`, az, sample.Metric, sample.Value, other.Metric, other.Value)
                        }
                        matchedSamples[az] = sample
                default:
                        unmatchedSamples = append(unmatchedSamples, sample)
                }
        }

        // build result
        result := make(map[liquid.AvailabilityZone]*liquid.AZResourceCapacityReport)
        for az, sample := range matchedSamples {
                result[az] = &liquid.AZResourceCapacityReport{
                        Capacity: uint64(sample.Value),
                }
        }
        if len(result) == 0 || len(unmatchedSamples) > 0 {
                unmatchedCapacity := float64(0.0)
                for _, sample := range unmatchedSamples {
                        unmatchedCapacity += float64(sample.Value)
                }
                result[limes.AvailabilityZoneUnknown] = &liquid.AZResourceCapacityReport{
                        Capacity: uint64(unmatchedCapacity),
                }
        }

        // validate result
        if !p.AllowZeroCapacity {
                totalCapacity := uint64(0)
                for _, azData := range result {
                        totalCapacity += azData.Capacity
                }
                if totalCapacity == 0 {
                        return nil, errors.New("got 0 total capacity, but allow_zero_capacity = false")
                }
        }

        return result, nil
}

1	/*******************************************************************************
2	*
3	* Copyright 2024 SAP SE
4	*
5	* Licensed under the Apache License, Version 2.0 (the "License");
6	* you may not use this file except in compliance with the License.
7	* You should have received a copy of the License along with this
8	* program. If not, you may obtain a copy of the License at
9	*
10	* http://www.apache.org/licenses/LICENSE-2.0
11	*
12	* Unless required by applicable law or agreed to in writing, software
13	* distributed under the License is distributed on an "AS IS" BASIS,
14	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15	* See the License for the specific language governing permissions and
16	* limitations under the License.
17	*
18	*******************************************************************************/
19
20	package plugins
21
22	import (
23	"context"
24	"errors"
25	"fmt"
26	"slices"
27
28	"github.com/gophercloud/gophercloud/v2"
29	. "github.com/majewsky/gg/option"
30	"github.com/prometheus/common/model"
31	"github.com/sapcc/go-api-declarations/limes"
32	"github.com/sapcc/go-api-declarations/liquid"
33	"github.com/sapcc/go-bits/liquidapi"
34	"github.com/sapcc/go-bits/logg"
35	"github.com/sapcc/go-bits/promquery"
36
37	"github.com/sapcc/limes/internal/core"
38	"github.com/sapcc/limes/internal/db"
39	)
40
41	type prometheusCapacityConfiguration struct {
42	APIConfig promquery.Config `yaml:"api"`
43	Queries map[liquid.ResourceName]string `yaml:"queries"`
44	AllowZeroCapacity bool `yaml:"allow_zero_capacity"`
45	}
46
47	type fixedCapacityConfiguration struct {
48	Values map[liquid.ResourceName]uint64 `yaml:"values"`
49	}
50
51	type LiquidCapacityPlugin struct {
52	// configuration
53	ServiceType db.ServiceType `yaml:"service_type"`
54	LiquidServiceType string `yaml:"liquid_service_type"`
55	FixedCapacityConfiguration Option[fixedCapacityConfiguration] `yaml:"fixed_capacity_values"`
56	PrometheusCapacityConfiguration Option[prometheusCapacityConfiguration] `yaml:"capacity_values_from_prometheus"`
57
58	// state
59	LiquidServiceInfo liquid.ServiceInfo `yaml:"-"`
60	LiquidClient core.LiquidClient `yaml:"-"`
61	}
62
63	func init() {	3✔
64	core.CapacityPluginRegistry.Add(func() core.CapacityPlugin { return &LiquidCapacityPlugin{} })	20✔
65	}
66
67	// PluginTypeID implements the core.CapacityPlugin interface.
68	func (p *LiquidCapacityPlugin) PluginTypeID() string {	3✔
69	return "liquid"	3✔
70	}	3✔
71
72	// Init implements the core.CapacityPlugin interface.
73	func (p LiquidCapacityPlugin) Init(ctx context.Context, client gophercloud.ProviderClient, eo gophercloud.EndpointOpts) (err error) {	14✔
74	if p.ServiceType == "" {	14✔
75	return errors.New("missing required value: params.service_type")	×
76	}	×
77	if p.LiquidServiceType == "" {	14✔
78	p.LiquidServiceType = "liquid-" + string(p.ServiceType)	×
79	}	×
80
81	p.LiquidClient, err = core.NewLiquidClient(client, eo, liquidapi.ClientOpts{ServiceType: p.LiquidServiceType})	14✔
82	if err != nil {	14✔
83	return err	×
84	}	×
85
86	p.LiquidServiceInfo, err = p.LiquidClient.GetInfo(ctx)	14✔
87	if err != nil {	14✔
88	return err	×
89	}	×
90	err = liquid.ValidateServiceInfo(p.LiquidServiceInfo)	14✔
91	return err	14✔
92	}
93
94	// ServiceInfo implements the core.CapacityPlugin interface.
95	func (p *LiquidCapacityPlugin) ServiceInfo() liquid.ServiceInfo {	33✔
96	return p.LiquidServiceInfo	33✔
97	}	33✔
98
99	// Scrape implements the core.CapacityPlugin interface.
100	// we assume that each resource only has one source of info (else, the last source wins and overwrites the previous data)
101	// we fail the whole collection explicitly in case any of the sources fails
102	func (p *LiquidCapacityPlugin) Scrape(ctx context.Context, backchannel core.CapacityPluginBackchannel, allAZs []limes.AvailabilityZone) (result liquid.ServiceCapacityReport, err error) {	31✔
103	req, err := p.BuildServiceCapacityRequest(backchannel, allAZs)	31✔
104	if err != nil {	31✔
NEW 105	return liquid.ServiceCapacityReport{}, err	×
106	}	×
107
108	result, err = p.LiquidClient.GetCapacityReport(ctx, req)	31✔
109	if err != nil {	31✔
NEW 110	return liquid.ServiceCapacityReport{}, err	×
111	}	×
112	if result.InfoVersion != p.LiquidServiceInfo.Version {	31✔
113	logg.Fatal("ServiceInfo version for %s changed from %d to %d; restarting now to reload ServiceInfo...",	×
NEW 114	p.LiquidServiceType, p.LiquidServiceInfo.Version, result.InfoVersion)	×
115	}	×
116	err = liquid.ValidateServiceInfo(p.LiquidServiceInfo)	31✔
117	if err != nil {	31✔
NEW 118	return liquid.ServiceCapacityReport{}, err	×
119	}	×
120	err = liquid.ValidateCapacityReport(result, req, p.LiquidServiceInfo)	31✔
121	if err != nil {	31✔
NEW 122	return liquid.ServiceCapacityReport{}, err	×
123	}	×
124
125	// manual capacity collection
126	fixedCapaConfig, exists := p.FixedCapacityConfiguration.Unpack()	31✔
127	if exists {	32✔
128	if result.Resources == nil {	2✔
129	result.Resources = make(map[liquid.ResourceName]*liquid.ResourceCapacityReport)	1✔
130	}	1✔
131	for resName, capacity := range fixedCapaConfig.Values {	2✔
132	result.Resources[resName] = &liquid.ResourceCapacityReport{	1✔
133	PerAZ: liquid.InAnyAZ(liquid.AZResourceCapacityReport{Capacity: capacity}),	1✔
134	}	1✔
135	}	1✔
136	}
137
138	// prometheus capacity collection
139	prometheusCapaConfig, exists := p.PrometheusCapacityConfiguration.Unpack()	31✔
140	if exists {	31✔
NEW 141	if result.Resources == nil {	×
NEW 142	result.Resources = make(map[liquid.ResourceName]*liquid.ResourceCapacityReport)	×
NEW 143	}	×
144	client, err := prometheusCapaConfig.APIConfig.Connect()	×
145	if err != nil {	×
NEW 146	return liquid.ServiceCapacityReport{}, err	×
147	}	×
148	for resName, query := range prometheusCapaConfig.Queries {	×
NEW 149	azReports, err := prometheusCapaConfig.prometheusScrapeOneResource(ctx, client, query, allAZs)	×
150	if err != nil {	×
NEW 151	return liquid.ServiceCapacityReport{}, fmt.Errorf("while scraping prometheus capacity %q/%q: %w", p.ServiceType, resName, err)	×
NEW 152	}	×
NEW 153	result.Resources[resName] = &liquid.ResourceCapacityReport{	×
NEW 154	PerAZ: azReports,	×
UNCOV 155	}	×
156	}
157	}
158
159	return result, nil	31✔
160	}
161
162	func (p *LiquidCapacityPlugin) BuildServiceCapacityRequest(backchannel core.CapacityPluginBackchannel, allAZs []limes.AvailabilityZone) (liquid.ServiceCapacityRequest, error) {	32✔
163	req := liquid.ServiceCapacityRequest{	32✔
164	AllAZs: allAZs,	32✔
165	DemandByResource: make(map[liquid.ResourceName]liquid.ResourceDemand, len(p.LiquidServiceInfo.Resources)),	32✔
166	}	32✔
167		32✔
168	var err error	32✔
169	for resName, resInfo := range p.LiquidServiceInfo.Resources {	88✔
170	if !resInfo.HasCapacity {	63✔
171	continue	7✔
172	}
173	if !resInfo.NeedsResourceDemand {	97✔
174	continue	48✔
175	}
176	req.DemandByResource[resName], err = backchannel.GetResourceDemand(p.ServiceType, resName)	1✔
177	if err != nil {	1✔
178	return liquid.ServiceCapacityRequest{}, fmt.Errorf("while getting resource demand for %s/%s: %w", p.ServiceType, resName, err)	×
179	}	×
180	}
181	return req, nil	32✔
182	}
183
NEW 184	func (p prometheusCapacityConfiguration) prometheusScrapeOneResource(ctx context.Context, client promquery.Client, query string, allAZs []limes.AvailabilityZone) (map[liquid.AvailabilityZone]*liquid.AZResourceCapacityReport, error) {	×
UNCOV 185	vector, err := client.GetVector(ctx, query)	×
186	if err != nil {	×
187	return nil, err	×
188	}	×
189
190	// for known AZs, we expect exactly one result;
191	// all unknown AZs get lumped into AvailabilityZoneUnknown
192	matchedSamples := make(map[limes.AvailabilityZone]*model.Sample)	×
193	var unmatchedSamples []*model.Sample	×
194	for _, sample := range vector {	×
195	az := limes.AvailabilityZone(sample.Metric["az"])	×
196	switch {	×
197	case az == "":	×
198	return nil, fmt.Errorf(`missing label "az" on metric %v = %g`, sample.Metric, sample.Value)	×
199	case slices.Contains(allAZs, az) \|\| az == limes.AvailabilityZoneAny:	×
200	if matchedSamples[az] != nil {	×
201	other := matchedSamples[az]	×
202	return nil, fmt.Errorf(`multiple samples for az=%q: found %v = %g and %v = %g`, az, sample.Metric, sample.Value, other.Metric, other.Value)	×
203	}	×
204	matchedSamples[az] = sample	×
205	default:	×
206	unmatchedSamples = append(unmatchedSamples, sample)	×
207	}
208	}
209
210	// build result
NEW 211	result := make(map[liquid.AvailabilityZone]*liquid.AZResourceCapacityReport)	×
212	for az, sample := range matchedSamples {	×
NEW 213	result[az] = &liquid.AZResourceCapacityReport{	×
214	Capacity: uint64(sample.Value),	×
215	}	×
216	}	×
217	if len(result) == 0 \|\| len(unmatchedSamples) > 0 {	×
218	unmatchedCapacity := float64(0.0)	×
219	for _, sample := range unmatchedSamples {	×
220	unmatchedCapacity += float64(sample.Value)	×
221	}	×
NEW 222	result[limes.AvailabilityZoneUnknown] = &liquid.AZResourceCapacityReport{	×
223	Capacity: uint64(unmatchedCapacity),	×
224	}	×
225	}
226
227	// validate result
228	if !p.AllowZeroCapacity {	×
229	totalCapacity := uint64(0)	×
230	for _, azData := range result {	×
231	totalCapacity += azData.Capacity	×
232	}	×
233	if totalCapacity == 0 {	×
234	return nil, errors.New("got 0 total capacity, but allow_zero_capacity = false")	×
235	}	×
236	}
237
238	return result, nil	×
239	}

sapcc / limes / 14831633718

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous