Add Splunk as a metrics provider #1733
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,195 @@ | ||
| /* | ||
| Copyright 2020 The Flux authors | ||
|
|
||
| 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 providers | ||
|
|
||
| import ( | ||
| "cmp" | ||
| "context" | ||
| "fmt" | ||
| "io" | ||
| "net/http" | ||
| "slices" | ||
| "strings" | ||
| "time" | ||
|
|
||
| "github.com/signalfx/signalflow-client-go/signalflow" | ||
| "github.com/signalfx/signalflow-client-go/signalflow/messages" | ||
|
|
||
| flaggerv1 "github.com/fluxcd/flagger/pkg/apis/flagger/v1beta1" | ||
| ) | ||
|
|
||
| // https://dev.splunk.com/observability/reference | ||
| const ( | ||
| signalFxSignalFlowApiPath = "/v2/signalflow" | ||
| signalFxValidationPath = "/v2/metric?limit=1" | ||
|
|
||
| signalFxTokenSecretKey = "sf_token_key" | ||
|
|
||
| signalFxTokenHeaderKey = "X-SF-Token" | ||
|
|
||
| signalFxFromDeltaMultiplierOnMetricInterval = 10 | ||
| ) | ||
|
|
||
| // SplunkProvider executes signalfx queries | ||
| type SplunkProvider struct { | ||
| metricsQueryEndpoint string | ||
| apiValidationEndpoint string | ||
|
|
||
| timeout time.Duration | ||
| token string | ||
| fromDelta int64 | ||
| } | ||
|
|
||
| type splunkResponse struct { | ||
| } | ||
|
|
||
| // NewSplunkProvider takes a canary spec, a provider spec and the credentials map, and | ||
| // returns a Splunk client ready to execute queries against the API | ||
| func NewSplunkProvider(metricInterval string, | ||
| provider flaggerv1.MetricTemplateProvider, | ||
| credentials map[string][]byte) (*SplunkProvider, error) { | ||
|
|
||
| address := provider.Address | ||
| if address == "" { | ||
| return nil, fmt.Errorf("splunk endpoint is not set") | ||
| } | ||
|
|
||
| sp := SplunkProvider{ | ||
| timeout: 5 * time.Second, | ||
| // Convert the configured address to match the protocol of the respective API | ||
| // ex. | ||
| // https://api.<REALM>.signalfx.com -> wss://stream.<REALM>.signalfx.com | ||
| // wss://stream.<REALM>.signalfx.com -> wss://stream.<REALM>.signalfx.com | ||
| metricsQueryEndpoint: strings.Replace(strings.Replace(address+signalFxSignalFlowApiPath, "http", "ws", 1), "api", "stream", 1), | ||
| // ex. | ||
| // https://api.<REALM>.signalfx.com -> https://api.<REALM>.signalfx.com | ||
| // wss://stream.<REALM>.signalfx.com -> https://api.<REALM>.signalfx.com | ||
| apiValidationEndpoint: strings.Replace(strings.Replace(address+signalFxValidationPath, "ws", "http", 1), "stream", "api", 1), | ||
| } | ||
|
|
||
| if b, ok := credentials[signalFxTokenSecretKey]; ok { | ||
| sp.token = string(b) | ||
| } else { | ||
| return nil, fmt.Errorf("splunk credentials does not contain sf_token_key") | ||
| } | ||
|
|
||
| md, err := time.ParseDuration(metricInterval) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("error parsing metric interval: %w", err) | ||
| } | ||
|
|
||
| sp.fromDelta = int64(signalFxFromDeltaMultiplierOnMetricInterval * md.Milliseconds()) | ||
| return &sp, nil | ||
| } | ||
|
|
||
| // RunQuery executes the query and converts the first result to float64 | ||
| func (p *SplunkProvider) RunQuery(query string) (float64, error) { | ||
| c, err := signalflow.NewClient(signalflow.StreamURL(p.metricsQueryEndpoint), signalflow.AccessToken(p.token)) | ||
| if err != nil { | ||
| return 0, fmt.Errorf("error creating signalflow client: %w", err) | ||
| } | ||
|
|
||
| ctx, cancel := context.WithTimeout(context.Background(), p.timeout) | ||
| defer cancel() | ||
|
|
||
| now := time.Now().UnixMilli() | ||
| comp, err := c.Execute(ctx, &signalflow.ExecuteRequest{ | ||
| Program: query, | ||
| Start: time.UnixMilli(now - p.fromDelta), | ||
| Stop: time.UnixMilli(now), | ||
| Immediate: true, | ||
| }) | ||
| if err != nil { | ||
| return 0, fmt.Errorf("error executing query: %w", err) | ||
| } | ||
|
|
||
| payloads := p.receivePaylods(comp) | ||
|
|
||
| if comp.Err() != nil { | ||
| return 0, fmt.Errorf("error executing query: %w", comp.Err()) | ||
| } | ||
|
|
||
| payloads = slices.DeleteFunc(payloads, func(msg messages.DataPayload) bool { | ||
| return msg.Value() == nil | ||
| }) | ||
| if len(payloads) < 1 { | ||
| return 0, fmt.Errorf("invalid response: %w", ErrNoValuesFound) | ||
| } | ||
|
|
||
| // Error when a SignalFlow query returns two or more results. | ||
| // Since a different TSID is set for each metrics to be retrieved, eliminate duplicate TSIDs and determine if two or more TSIDs exist. | ||
| _payloads := slices.Clone(payloads) | ||
| slices.SortFunc(_payloads, func(i, j messages.DataPayload) int { | ||
|
There was a problem hiding this comment. some comments explaining this logic would be nice |
||
| return cmp.Compare(i.TSID, j.TSID) | ||
| }) | ||
| if len(slices.CompactFunc(_payloads, func(i, j messages.DataPayload) bool { return i.TSID == j.TSID })) > 1 { | ||
| return 0, fmt.Errorf("invalid response: %w", ErrMultipleValuesReturned) | ||
| } | ||
|
|
||
| payload := payloads[len(payloads)-1] | ||
| switch payload.Type { | ||
| case messages.ValTypeLong: | ||
| return float64(payload.Int64()), nil | ||
| case messages.ValTypeDouble: | ||
| return payload.Float64(), nil | ||
| case messages.ValTypeInt: | ||
| return float64(payload.Int32()), nil | ||
| default: | ||
| return 0, fmt.Errorf("invalid response: UnsupportedValueType") | ||
| } | ||
| } | ||
|
|
||
| func (p *SplunkProvider) receivePaylods(comp *signalflow.Computation) []messages.DataPayload { | ||
| payloads := []messages.DataPayload{} | ||
| for dataMsg := range comp.Data() { | ||
| if dataMsg == nil { | ||
| continue | ||
| } | ||
| payloads = append(payloads, dataMsg.Payloads...) | ||
| } | ||
| return payloads | ||
| } | ||
|
|
||
| // IsOnline calls the provider endpoint and returns an error if the API is unreachable | ||
| func (p *SplunkProvider) IsOnline() (bool, error) { | ||
| req, err := http.NewRequest("GET", p.apiValidationEndpoint, nil) | ||
| if err != nil { | ||
| return false, fmt.Errorf("error http.NewRequest: %w", err) | ||
| } | ||
|
|
||
| req.Header.Add(signalFxTokenHeaderKey, p.token) | ||
|
|
||
| ctx, cancel := context.WithTimeout(req.Context(), p.timeout) | ||
| defer cancel() | ||
| r, err := http.DefaultClient.Do(req.WithContext(ctx)) | ||
| if err != nil { | ||
| return false, fmt.Errorf("request failed: %w", err) | ||
| } | ||
|
|
||
| defer r.Body.Close() | ||
|
|
||
| b, err := io.ReadAll(r.Body) | ||
| if err != nil { | ||
| return false, fmt.Errorf("error reading body: %w", err) | ||
| } | ||
|
|
||
| if r.StatusCode != http.StatusOK { | ||
| return false, fmt.Errorf("error response: %s", string(b)) | ||
| } | ||
|
|
||
| return true, nil | ||
| } | ||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
could you specify the purpose of this?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I referred to the datadog implementation.
In some cases, the splunk api could only acquire empty data if the specified period was short, so to ensure that data is acquired, a period 10 times longer than the set interval is used.