From d0ffe2ed3ec4b65244d0825e2d00862dc6110d32 Mon Sep 17 00:00:00 2001 From: austin ce Date: Tue, 4 May 2021 15:47:00 -0400 Subject: [PATCH] Add base xDS discovery and kuma SD implementation Signed-off-by: austin ce --- discovery/install/install.go | 1 + discovery/xds/client.go | 226 +++++++++++++++++++ discovery/xds/kuma.go | 226 +++++++++++++++++++ discovery/xds/kuma_mads.pb.go | 398 ++++++++++++++++++++++++++++++++++ discovery/xds/xds.go | 176 +++++++++++++++ go.mod | 4 + go.sum | 3 + 7 files changed, 1034 insertions(+) create mode 100644 discovery/xds/client.go create mode 100644 discovery/xds/kuma.go create mode 100644 discovery/xds/kuma_mads.pb.go create mode 100644 discovery/xds/xds.go diff --git a/discovery/install/install.go b/discovery/install/install.go index 6258d6fc0..34ccf3d0f 100644 --- a/discovery/install/install.go +++ b/discovery/install/install.go @@ -33,5 +33,6 @@ import ( _ "github.com/prometheus/prometheus/discovery/openstack" // register openstack _ "github.com/prometheus/prometheus/discovery/scaleway" // register scaleway _ "github.com/prometheus/prometheus/discovery/triton" // register triton + _ "github.com/prometheus/prometheus/discovery/xds" // register xds _ "github.com/prometheus/prometheus/discovery/zookeeper" // register zookeeper ) diff --git a/discovery/xds/client.go b/discovery/xds/client.go new file mode 100644 index 000000000..cd8ffb017 --- /dev/null +++ b/discovery/xds/client.go @@ -0,0 +1,226 @@ +// Copyright 2021 The Prometheus 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 xds + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "io/ioutil" + "net/http" + "net/url" + "path" + "time" + + envoy_core "github.com/envoyproxy/go-control-plane/envoy/config/core/v3" + v3 "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3" + "github.com/prometheus/common/config" + "github.com/prometheus/common/version" +) + +var userAgent = fmt.Sprintf("Prometheus/%s", version.Version) + +// ResourceClient exposes the xDS protocol for a single resource type. +// See https://www.envoyproxy.io/docs/envoy/latest/api-docs/xds_protocol#rest-json-polling-subscriptions . +type ResourceClient interface { + // ResourceTypeURL is the type URL of the resource. + ResourceTypeURL() string + + // Server is the xDS Management server. + Server() string + + // Fetch requests the latest view of the entire resource state. + // If no updates have been made since the last request, the response will be nil. + Fetch(ctx context.Context) (*v3.DiscoveryResponse, error) + + // ID returns the ID of the client that is sent to the xDS server. + ID() string + + // Close releases currently held resources. + Close() +} + +type HTTPResourceClient struct { + client *http.Client + config *HTTPResourceClientConfig + + // endpoint is the fully-constructed xDS HTTP endpoint. + endpoint string + // Caching. + latestVersion string + latestNonce string +} + +type HTTPResourceClientConfig struct { + // HTTP config. + config.HTTPClientConfig + + Name string + + // ExtraQueryParams are extra query parameters to attach to the request URL. + ExtraQueryParams url.Values + + // General xDS config. + + // The timeout for a single fetch request. + Timeout time.Duration + + // Type is the xds type, e.g., clusters + // which is used in the discovery POST request. + ResourceType string + // ResourceTypeURL is the Google type url for the resource, e.g., type.googleapis.com/envoy.api.v2.Cluster. + ResourceTypeURL string + // Server is the xDS management server. + Server string + // ClientID is used to identify the client with the management server. + ClientID string +} + +func NewHTTPResourceClient(conf *HTTPResourceClientConfig, protocolVersion ProtocolVersion) (*HTTPResourceClient, error) { + if protocolVersion != ProtocolV3 { + return nil, errors.New("only the v3 protocol is supported") + } + + if len(conf.Server) == 0 { + return nil, errors.New("empty xDS server") + } + + serverURL, err := url.Parse(conf.Server) + if err != nil { + return nil, err + } + + endpointURL, err := makeXDSResourceHTTPEndpointURL(protocolVersion, serverURL, conf.ResourceType) + if err != nil { + return nil, err + } + + if conf.ExtraQueryParams != nil { + endpointURL.RawQuery = conf.ExtraQueryParams.Encode() + } + + client, err := config.NewClientFromConfig(conf.HTTPClientConfig, conf.Name, config.WithHTTP2Disabled(), config.WithIdleConnTimeout(conf.Timeout)) + if err != nil { + return nil, err + } + + client.Timeout = conf.Timeout + + return &HTTPResourceClient{ + client: client, + config: conf, + endpoint: endpointURL.String(), + latestVersion: "", + latestNonce: "", + }, nil +} + +func makeXDSResourceHTTPEndpointURL(protocolVersion ProtocolVersion, serverURL *url.URL, resourceType string) (*url.URL, error) { + if serverURL == nil { + return nil, errors.New("empty xDS server URL") + } + + if len(serverURL.Scheme) == 0 || len(serverURL.Host) == 0 { + return nil, errors.New("invalid xDS server URL") + } + + if serverURL.Scheme != "http" && serverURL.Scheme != "https" { + return nil, errors.New("invalid xDS server URL protocol. must be either 'http' or 'https'") + } + + serverURL.Path = path.Join(serverURL.Path, string(protocolVersion), fmt.Sprintf("discovery:%s", resourceType)) + + return serverURL, nil +} + +func (rc *HTTPResourceClient) Server() string { + return rc.config.Server +} + +func (rc *HTTPResourceClient) ResourceTypeURL() string { + return rc.config.ResourceTypeURL +} + +func (rc *HTTPResourceClient) ID() string { + return rc.config.ClientID +} + +func (rc *HTTPResourceClient) Close() { + rc.client.CloseIdleConnections() +} + +// Fetch requests the latest state of the resources from the xDS server and cache the version. +// Returns a nil response if the current local version is up to date. +func (rc *HTTPResourceClient) Fetch(ctx context.Context) (*v3.DiscoveryResponse, error) { + discoveryReq := &v3.DiscoveryRequest{ + VersionInfo: rc.latestVersion, + ResponseNonce: rc.latestNonce, + TypeUrl: rc.ResourceTypeURL(), + ResourceNames: []string{}, + Node: &envoy_core.Node{ + Id: rc.ID(), + }, + } + + reqBody, err := protoJSONMarshalOptions.Marshal(discoveryReq) + if err != nil { + return nil, err + } + + request, err := http.NewRequest("POST", rc.endpoint, bytes.NewBuffer(reqBody)) + if err != nil { + return nil, err + } + request = request.WithContext(ctx) + + request.Header.Add("User-Agent", userAgent) + request.Header.Add("Content-Type", "application/json") + request.Header.Add("Accept", "application/json") + + resp, err := rc.client.Do(request) + if err != nil { + return nil, err + } + defer func() { + io.Copy(ioutil.Discard, resp.Body) + resp.Body.Close() + }() + + if resp.StatusCode == http.StatusNotModified { + // Empty response, already have the latest. + return nil, nil + } + + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("non 200 status '%d' response during xDS fetch", resp.StatusCode) + } + + respBody, err := ioutil.ReadAll(resp.Body) + if err != nil { + return nil, err + } + + discoveryRes := &v3.DiscoveryResponse{} + if err = protoJSONUnmarshalOptions.Unmarshal(respBody, discoveryRes); err != nil { + return nil, err + } + + // Cache the latest nonce + version info. + rc.latestNonce = discoveryRes.Nonce + rc.latestVersion = discoveryRes.VersionInfo + + return discoveryRes, nil +} diff --git a/discovery/xds/kuma.go b/discovery/xds/kuma.go new file mode 100644 index 000000000..d4071be9c --- /dev/null +++ b/discovery/xds/kuma.go @@ -0,0 +1,226 @@ +// Copyright 2021 The Prometheus 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 xds + +import ( + "fmt" + "net/url" + "time" + + "github.com/go-kit/log" + "github.com/go-kit/log/level" + "github.com/pkg/errors" + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/common/config" + "github.com/prometheus/common/model" + "google.golang.org/protobuf/types/known/anypb" + + "github.com/prometheus/prometheus/discovery" + "github.com/prometheus/prometheus/discovery/targetgroup" + "github.com/prometheus/prometheus/util/osutil" + "github.com/prometheus/prometheus/util/strutil" +) + +var ( + // DefaultKumaSDConfig is the default Kuma MADS SD configuration. + DefaultKumaSDConfig = KumaSDConfig{ + HTTPClientConfig: config.DefaultHTTPClientConfig, + RefreshInterval: model.Duration(15 * time.Second), + FetchTimeout: model.Duration(2 * time.Minute), + } + + kumaFetchFailuresCount = prometheus.NewCounter( + prometheus.CounterOpts{ + Namespace: namespace, + Name: "sd_kuma_fetch_failures_total", + Help: "The number of Kuma MADS fetch call failures.", + }) + kumaFetchSkipUpdateCount = prometheus.NewCounter( + prometheus.CounterOpts{ + Namespace: namespace, + Name: "sd_kuma_fetch_skipped_updates_total", + Help: "The number of Kuma MADS fetch calls that result in no updates to the targets.", + }) + kumaFetchDuration = prometheus.NewSummary( + prometheus.SummaryOpts{ + Namespace: namespace, + Name: "sd_kuma_fetch_duration_seconds", + Help: "The duration of a Kuma MADS fetch call.", + Objectives: map[float64]float64{0.5: 0.05, 0.9: 0.01, 0.99: 0.001}, + }, + ) +) + +const ( + // kumaMetaLabelPrefix is the meta prefix used for all kuma meta labels. + kumaMetaLabelPrefix = model.MetaLabelPrefix + "kuma_" + + // kumaMeshLabel is the name of the label that holds the mesh name. + kumaMeshLabel = kumaMetaLabelPrefix + "mesh" + // kumaServiceLabel is the name of the label that holds the service name. + kumaServiceLabel = kumaMetaLabelPrefix + "service" + // kumaDataplaneLabel is the name of the label that holds the dataplane name. + kumaDataplaneLabel = kumaMetaLabelPrefix + "dataplane" + // kumaUserLabelPrefix is the name of the label that namespaces all user-defined labels. + kumaUserLabelPrefix = kumaMetaLabelPrefix + "label_" +) + +const ( + KumaMadsV1ResourceTypeURL = "type.googleapis.com/kuma.observability.v1.MonitoringAssignment" + KumaMadsV1ResourceType = "monitoringassignments" +) + +type KumaSDConfig = SDConfig + +// UnmarshalYAML implements the yaml.Unmarshaler interface. +func (c *KumaSDConfig) UnmarshalYAML(unmarshal func(interface{}) error) error { + *c = DefaultKumaSDConfig + type plainKumaConf KumaSDConfig + err := unmarshal((*plainKumaConf)(c)) + if err != nil { + return err + } + + if len(c.Server) == 0 { + return errors.Errorf("kuma SD server must not be empty: %s", c.Server) + } + parsedURL, err := url.Parse(c.Server) + if err != nil { + return err + } + + if len(parsedURL.Scheme) == 0 || len(parsedURL.Host) == 0 { + return errors.Errorf("kuma SD server must not be empty and have a scheme: %s", c.Server) + } + + if err := c.HTTPClientConfig.Validate(); err != nil { + return err + } + + return nil +} + +func (c *KumaSDConfig) Name() string { + return "kuma" +} + +// SetDirectory joins any relative file paths with dir. +func (c *KumaSDConfig) SetDirectory(dir string) { + c.HTTPClientConfig.SetDirectory(dir) +} + +func (c *KumaSDConfig) NewDiscoverer(opts discovery.DiscovererOptions) (discovery.Discoverer, error) { + logger := opts.Logger + if logger == nil { + logger = log.NewNopLogger() + } + + return NewKumaHTTPDiscovery(c, logger) +} + +func convertKumaV1MonitoringAssignment(assignment *MonitoringAssignment) *targetgroup.Group { + commonLabels := convertKumaUserLabels(assignment.Labels) + + commonLabels[kumaMeshLabel] = model.LabelValue(assignment.Mesh) + commonLabels[kumaServiceLabel] = model.LabelValue(assignment.Service) + + var targetLabelSets []model.LabelSet + + for _, target := range assignment.Targets { + targetLabels := convertKumaUserLabels(target.Labels) + + targetLabels[kumaDataplaneLabel] = model.LabelValue(target.Name) + targetLabels[model.InstanceLabel] = model.LabelValue(target.Name) + targetLabels[model.AddressLabel] = model.LabelValue(target.Address) + targetLabels[model.SchemeLabel] = model.LabelValue(target.Scheme) + targetLabels[model.MetricsPathLabel] = model.LabelValue(target.MetricsPath) + + targetLabelSets = append(targetLabelSets, targetLabels) + } + + return &targetgroup.Group{ + Labels: commonLabels, + Targets: targetLabelSets, + } +} + +func convertKumaUserLabels(labels map[string]string) model.LabelSet { + labelSet := model.LabelSet{} + for key, value := range labels { + name := kumaUserLabelPrefix + strutil.SanitizeLabelName(key) + labelSet[model.LabelName(name)] = model.LabelValue(value) + } + return labelSet +} + +// kumaMadsV1ResourceParser is an xds.resourceParser. +func kumaMadsV1ResourceParser(resources []*anypb.Any, typeURL string) ([]*targetgroup.Group, error) { + if typeURL != KumaMadsV1ResourceTypeURL { + return nil, errors.Errorf("recieved invalid typeURL for Kuma MADS v1 Resource: %s", typeURL) + } + + var groups []*targetgroup.Group + + for _, resource := range resources { + assignment := &MonitoringAssignment{} + + if err := anypb.UnmarshalTo(resource, assignment, protoUnmarshalOptions); err != nil { + return nil, err + } + + groups = append(groups, convertKumaV1MonitoringAssignment(assignment)) + } + + return groups, nil +} + +func NewKumaHTTPDiscovery(conf *KumaSDConfig, logger log.Logger) (discovery.Discoverer, error) { + // Default to "prometheus" if hostname is unavailable. + clientID, err := osutil.GetFQDN() + if err != nil { + level.Debug(logger).Log("msg", "error getting FQDN", "err", err) + clientID = "prometheus" + } + + clientConfig := &HTTPResourceClientConfig{ + HTTPClientConfig: conf.HTTPClientConfig, + ExtraQueryParams: url.Values{ + "fetch-timeout": {conf.FetchTimeout.String()}, + }, + // Allow 15s of buffer over the timeout sent to the xDS server for connection overhead. + Timeout: time.Duration(conf.FetchTimeout) + (15 * time.Second), + ResourceType: KumaMadsV1ResourceType, + ResourceTypeURL: KumaMadsV1ResourceTypeURL, + Server: conf.Server, + ClientID: clientID, + } + + client, err := NewHTTPResourceClient(clientConfig, ProtocolV3) + if err != nil { + return nil, fmt.Errorf("kuma_sd: %w", err) + } + + d := &fetchDiscovery{ + client: client, + logger: logger, + refreshInterval: time.Duration(conf.RefreshInterval), + source: "kuma", + parseResources: kumaMadsV1ResourceParser, + fetchFailuresCount: kumaFetchFailuresCount, + fetchSkipUpdateCount: kumaFetchSkipUpdateCount, + fetchDuration: kumaFetchDuration, + } + + return d, nil +} diff --git a/discovery/xds/kuma_mads.pb.go b/discovery/xds/kuma_mads.pb.go new file mode 100644 index 000000000..b1079bf23 --- /dev/null +++ b/discovery/xds/kuma_mads.pb.go @@ -0,0 +1,398 @@ +// Copyright 2021 The Prometheus 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. + +// Code generated by protoc-gen-go. DO NOT EDIT. +// versions: +// protoc-gen-go v1.25.0 +// protoc v3.14.0 +// source: observability/v1/mads.proto + +// gRPC-removed vendored file from Kuma. + +package xds + +import ( + context "context" + v3 "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3" + _ "github.com/envoyproxy/protoc-gen-validate/validate" + _ "google.golang.org/genproto/googleapis/api/annotations" + protoreflect "google.golang.org/protobuf/reflect/protoreflect" + protoimpl "google.golang.org/protobuf/runtime/protoimpl" + reflect "reflect" + sync "sync" +) + +const ( + // Verify that this generated code is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion) + // Verify that runtime/protoimpl is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) +) + +// MADS resource type. +// +// Describes a group of targets on a single service that need to be monitored. +type MonitoringAssignment struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields + + // Mesh of the dataplane. + // + // E.g., `default` + Mesh string `protobuf:"bytes,2,opt,name=mesh,proto3" json:"mesh,omitempty"` + // Identifying service the dataplane is proxying. + // + // E.g., `backend` + Service string `protobuf:"bytes,3,opt,name=service,proto3" json:"service,omitempty"` + // List of targets that need to be monitored. + Targets []*MonitoringAssignment_Target `protobuf:"bytes,4,rep,name=targets,proto3" json:"targets,omitempty"` + // Arbitrary Labels associated with every target in the assignment. + // + // E.g., `{"zone" : "us-east-1", "team": "infra", "commit_hash": "620506a88"}`. + Labels map[string]string `protobuf:"bytes,5,rep,name=labels,proto3" json:"labels,omitempty" protobuf_key:"bytes,1,opt,name=key,proto3" protobuf_val:"bytes,2,opt,name=value,proto3"` +} + +func (x *MonitoringAssignment) Reset() { + *x = MonitoringAssignment{} + if protoimpl.UnsafeEnabled { + mi := &file_observability_v1_mads_proto_msgTypes[0] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *MonitoringAssignment) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*MonitoringAssignment) ProtoMessage() {} + +func (x *MonitoringAssignment) ProtoReflect() protoreflect.Message { + mi := &file_observability_v1_mads_proto_msgTypes[0] + if protoimpl.UnsafeEnabled && x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use MonitoringAssignment.ProtoReflect.Descriptor instead. +func (*MonitoringAssignment) Descriptor() ([]byte, []int) { + return file_observability_v1_mads_proto_rawDescGZIP(), []int{0} +} + +func (x *MonitoringAssignment) GetMesh() string { + if x != nil { + return x.Mesh + } + return "" +} + +func (x *MonitoringAssignment) GetService() string { + if x != nil { + return x.Service + } + return "" +} + +func (x *MonitoringAssignment) GetTargets() []*MonitoringAssignment_Target { + if x != nil { + return x.Targets + } + return nil +} + +func (x *MonitoringAssignment) GetLabels() map[string]string { + if x != nil { + return x.Labels + } + return nil +} + +// Describes a single target that needs to be monitored. +type MonitoringAssignment_Target struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields + + // E.g., `backend-01` + Name string `protobuf:"bytes,1,opt,name=name,proto3" json:"name,omitempty"` + // Scheme on which to scrape the target. + //E.g., `http` + Scheme string `protobuf:"bytes,2,opt,name=scheme,proto3" json:"scheme,omitempty"` + // Address (preferably IP) for the service + // E.g., `backend.svc` or `10.1.4.32:9090` + Address string `protobuf:"bytes,3,opt,name=address,proto3" json:"address,omitempty"` + // Optional path to append to the address for scraping + //E.g., `/metrics` + MetricsPath string `protobuf:"bytes,4,opt,name=metrics_path,json=metricsPath,proto3" json:"metrics_path,omitempty"` + // Arbitrary labels associated with that particular target. + // + // E.g., + // `{ + // "commit_hash" : "620506a88", + // }`. + Labels map[string]string `protobuf:"bytes,5,rep,name=labels,proto3" json:"labels,omitempty" protobuf_key:"bytes,1,opt,name=key,proto3" protobuf_val:"bytes,2,opt,name=value,proto3"` +} + +func (x *MonitoringAssignment_Target) Reset() { + *x = MonitoringAssignment_Target{} + if protoimpl.UnsafeEnabled { + mi := &file_observability_v1_mads_proto_msgTypes[1] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *MonitoringAssignment_Target) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*MonitoringAssignment_Target) ProtoMessage() {} + +func (x *MonitoringAssignment_Target) ProtoReflect() protoreflect.Message { + mi := &file_observability_v1_mads_proto_msgTypes[1] + if protoimpl.UnsafeEnabled && x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use MonitoringAssignment_Target.ProtoReflect.Descriptor instead. +func (*MonitoringAssignment_Target) Descriptor() ([]byte, []int) { + return file_observability_v1_mads_proto_rawDescGZIP(), []int{0, 0} +} + +func (x *MonitoringAssignment_Target) GetName() string { + if x != nil { + return x.Name + } + return "" +} + +func (x *MonitoringAssignment_Target) GetScheme() string { + if x != nil { + return x.Scheme + } + return "" +} + +func (x *MonitoringAssignment_Target) GetAddress() string { + if x != nil { + return x.Address + } + return "" +} + +func (x *MonitoringAssignment_Target) GetMetricsPath() string { + if x != nil { + return x.MetricsPath + } + return "" +} + +func (x *MonitoringAssignment_Target) GetLabels() map[string]string { + if x != nil { + return x.Labels + } + return nil +} + +var File_observability_v1_mads_proto protoreflect.FileDescriptor + +var file_observability_v1_mads_proto_rawDesc = []byte{ + 0x0a, 0x1b, 0x6f, 0x62, 0x73, 0x65, 0x72, 0x76, 0x61, 0x62, 0x69, 0x6c, 0x69, 0x74, 0x79, 0x2f, + 0x76, 0x31, 0x2f, 0x6d, 0x61, 0x64, 0x73, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x15, 0x6b, + 0x75, 0x6d, 0x61, 0x2e, 0x6f, 0x62, 0x73, 0x65, 0x72, 0x76, 0x61, 0x62, 0x69, 0x6c, 0x69, 0x74, + 0x79, 0x2e, 0x76, 0x31, 0x1a, 0x2a, 0x65, 0x6e, 0x76, 0x6f, 0x79, 0x2f, 0x73, 0x65, 0x72, 0x76, + 0x69, 0x63, 0x65, 0x2f, 0x64, 0x69, 0x73, 0x63, 0x6f, 0x76, 0x65, 0x72, 0x79, 0x2f, 0x76, 0x33, + 0x2f, 0x64, 0x69, 0x73, 0x63, 0x6f, 0x76, 0x65, 0x72, 0x79, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, + 0x1a, 0x1c, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2f, 0x61, 0x70, 0x69, 0x2f, 0x61, 0x6e, 0x6e, + 0x6f, 0x74, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x1a, 0x17, + 0x76, 0x61, 0x6c, 0x69, 0x64, 0x61, 0x74, 0x65, 0x2f, 0x76, 0x61, 0x6c, 0x69, 0x64, 0x61, 0x74, + 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22, 0xd2, 0x04, 0x0a, 0x14, 0x4d, 0x6f, 0x6e, 0x69, + 0x74, 0x6f, 0x72, 0x69, 0x6e, 0x67, 0x41, 0x73, 0x73, 0x69, 0x67, 0x6e, 0x6d, 0x65, 0x6e, 0x74, + 0x12, 0x1b, 0x0a, 0x04, 0x6d, 0x65, 0x73, 0x68, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x42, 0x07, + 0xfa, 0x42, 0x04, 0x72, 0x02, 0x20, 0x01, 0x52, 0x04, 0x6d, 0x65, 0x73, 0x68, 0x12, 0x21, 0x0a, + 0x07, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x42, 0x07, + 0xfa, 0x42, 0x04, 0x72, 0x02, 0x20, 0x01, 0x52, 0x07, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, + 0x12, 0x4c, 0x0a, 0x07, 0x74, 0x61, 0x72, 0x67, 0x65, 0x74, 0x73, 0x18, 0x04, 0x20, 0x03, 0x28, + 0x0b, 0x32, 0x32, 0x2e, 0x6b, 0x75, 0x6d, 0x61, 0x2e, 0x6f, 0x62, 0x73, 0x65, 0x72, 0x76, 0x61, + 0x62, 0x69, 0x6c, 0x69, 0x74, 0x79, 0x2e, 0x76, 0x31, 0x2e, 0x4d, 0x6f, 0x6e, 0x69, 0x74, 0x6f, + 0x72, 0x69, 0x6e, 0x67, 0x41, 0x73, 0x73, 0x69, 0x67, 0x6e, 0x6d, 0x65, 0x6e, 0x74, 0x2e, 0x54, + 0x61, 0x72, 0x67, 0x65, 0x74, 0x52, 0x07, 0x74, 0x61, 0x72, 0x67, 0x65, 0x74, 0x73, 0x12, 0x4f, + 0x0a, 0x06, 0x6c, 0x61, 0x62, 0x65, 0x6c, 0x73, 0x18, 0x05, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x37, + 0x2e, 0x6b, 0x75, 0x6d, 0x61, 0x2e, 0x6f, 0x62, 0x73, 0x65, 0x72, 0x76, 0x61, 0x62, 0x69, 0x6c, + 0x69, 0x74, 0x79, 0x2e, 0x76, 0x31, 0x2e, 0x4d, 0x6f, 0x6e, 0x69, 0x74, 0x6f, 0x72, 0x69, 0x6e, + 0x67, 0x41, 0x73, 0x73, 0x69, 0x67, 0x6e, 0x6d, 0x65, 0x6e, 0x74, 0x2e, 0x4c, 0x61, 0x62, 0x65, + 0x6c, 0x73, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x52, 0x06, 0x6c, 0x61, 0x62, 0x65, 0x6c, 0x73, 0x1a, + 0x9f, 0x02, 0x0a, 0x06, 0x54, 0x61, 0x72, 0x67, 0x65, 0x74, 0x12, 0x1b, 0x0a, 0x04, 0x6e, 0x61, + 0x6d, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x42, 0x07, 0xfa, 0x42, 0x04, 0x72, 0x02, 0x20, + 0x01, 0x52, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x12, 0x1f, 0x0a, 0x06, 0x73, 0x63, 0x68, 0x65, 0x6d, + 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x42, 0x07, 0xfa, 0x42, 0x04, 0x72, 0x02, 0x20, 0x01, + 0x52, 0x06, 0x73, 0x63, 0x68, 0x65, 0x6d, 0x65, 0x12, 0x21, 0x0a, 0x07, 0x61, 0x64, 0x64, 0x72, + 0x65, 0x73, 0x73, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x42, 0x07, 0xfa, 0x42, 0x04, 0x72, 0x02, + 0x20, 0x01, 0x52, 0x07, 0x61, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73, 0x12, 0x21, 0x0a, 0x0c, 0x6d, + 0x65, 0x74, 0x72, 0x69, 0x63, 0x73, 0x5f, 0x70, 0x61, 0x74, 0x68, 0x18, 0x04, 0x20, 0x01, 0x28, + 0x09, 0x52, 0x0b, 0x6d, 0x65, 0x74, 0x72, 0x69, 0x63, 0x73, 0x50, 0x61, 0x74, 0x68, 0x12, 0x56, + 0x0a, 0x06, 0x6c, 0x61, 0x62, 0x65, 0x6c, 0x73, 0x18, 0x05, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x3e, + 0x2e, 0x6b, 0x75, 0x6d, 0x61, 0x2e, 0x6f, 0x62, 0x73, 0x65, 0x72, 0x76, 0x61, 0x62, 0x69, 0x6c, + 0x69, 0x74, 0x79, 0x2e, 0x76, 0x31, 0x2e, 0x4d, 0x6f, 0x6e, 0x69, 0x74, 0x6f, 0x72, 0x69, 0x6e, + 0x67, 0x41, 0x73, 0x73, 0x69, 0x67, 0x6e, 0x6d, 0x65, 0x6e, 0x74, 0x2e, 0x54, 0x61, 0x72, 0x67, + 0x65, 0x74, 0x2e, 0x4c, 0x61, 0x62, 0x65, 0x6c, 0x73, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x52, 0x06, + 0x6c, 0x61, 0x62, 0x65, 0x6c, 0x73, 0x1a, 0x39, 0x0a, 0x0b, 0x4c, 0x61, 0x62, 0x65, 0x6c, 0x73, + 0x45, 0x6e, 0x74, 0x72, 0x79, 0x12, 0x10, 0x0a, 0x03, 0x6b, 0x65, 0x79, 0x18, 0x01, 0x20, 0x01, + 0x28, 0x09, 0x52, 0x03, 0x6b, 0x65, 0x79, 0x12, 0x14, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, + 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x3a, 0x02, 0x38, + 0x01, 0x1a, 0x39, 0x0a, 0x0b, 0x4c, 0x61, 0x62, 0x65, 0x6c, 0x73, 0x45, 0x6e, 0x74, 0x72, 0x79, + 0x12, 0x10, 0x0a, 0x03, 0x6b, 0x65, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x6b, + 0x65, 0x79, 0x12, 0x14, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, + 0x09, 0x52, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x3a, 0x02, 0x38, 0x01, 0x32, 0xe6, 0x03, 0x0a, + 0x24, 0x4d, 0x6f, 0x6e, 0x69, 0x74, 0x6f, 0x72, 0x69, 0x6e, 0x67, 0x41, 0x73, 0x73, 0x69, 0x67, + 0x6e, 0x6d, 0x65, 0x6e, 0x74, 0x44, 0x69, 0x73, 0x63, 0x6f, 0x76, 0x65, 0x72, 0x79, 0x53, 0x65, + 0x72, 0x76, 0x69, 0x63, 0x65, 0x12, 0x89, 0x01, 0x0a, 0x1a, 0x44, 0x65, 0x6c, 0x74, 0x61, 0x4d, + 0x6f, 0x6e, 0x69, 0x74, 0x6f, 0x72, 0x69, 0x6e, 0x67, 0x41, 0x73, 0x73, 0x69, 0x67, 0x6e, 0x6d, + 0x65, 0x6e, 0x74, 0x73, 0x12, 0x31, 0x2e, 0x65, 0x6e, 0x76, 0x6f, 0x79, 0x2e, 0x73, 0x65, 0x72, + 0x76, 0x69, 0x63, 0x65, 0x2e, 0x64, 0x69, 0x73, 0x63, 0x6f, 0x76, 0x65, 0x72, 0x79, 0x2e, 0x76, + 0x33, 0x2e, 0x44, 0x65, 0x6c, 0x74, 0x61, 0x44, 0x69, 0x73, 0x63, 0x6f, 0x76, 0x65, 0x72, 0x79, + 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x32, 0x2e, 0x65, 0x6e, 0x76, 0x6f, 0x79, 0x2e, + 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x2e, 0x64, 0x69, 0x73, 0x63, 0x6f, 0x76, 0x65, 0x72, + 0x79, 0x2e, 0x76, 0x33, 0x2e, 0x44, 0x65, 0x6c, 0x74, 0x61, 0x44, 0x69, 0x73, 0x63, 0x6f, 0x76, + 0x65, 0x72, 0x79, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x28, 0x01, 0x30, + 0x01, 0x12, 0x80, 0x01, 0x0a, 0x1b, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x4d, 0x6f, 0x6e, 0x69, + 0x74, 0x6f, 0x72, 0x69, 0x6e, 0x67, 0x41, 0x73, 0x73, 0x69, 0x67, 0x6e, 0x6d, 0x65, 0x6e, 0x74, + 0x73, 0x12, 0x2c, 0x2e, 0x65, 0x6e, 0x76, 0x6f, 0x79, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, + 0x65, 0x2e, 0x64, 0x69, 0x73, 0x63, 0x6f, 0x76, 0x65, 0x72, 0x79, 0x2e, 0x76, 0x33, 0x2e, 0x44, + 0x69, 0x73, 0x63, 0x6f, 0x76, 0x65, 0x72, 0x79, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, + 0x2d, 0x2e, 0x65, 0x6e, 0x76, 0x6f, 0x79, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x2e, + 0x64, 0x69, 0x73, 0x63, 0x6f, 0x76, 0x65, 0x72, 0x79, 0x2e, 0x76, 0x33, 0x2e, 0x44, 0x69, 0x73, + 0x63, 0x6f, 0x76, 0x65, 0x72, 0x79, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, + 0x28, 0x01, 0x30, 0x01, 0x12, 0xae, 0x01, 0x0a, 0x1a, 0x46, 0x65, 0x74, 0x63, 0x68, 0x4d, 0x6f, + 0x6e, 0x69, 0x74, 0x6f, 0x72, 0x69, 0x6e, 0x67, 0x41, 0x73, 0x73, 0x69, 0x67, 0x6e, 0x6d, 0x65, + 0x6e, 0x74, 0x73, 0x12, 0x2c, 0x2e, 0x65, 0x6e, 0x76, 0x6f, 0x79, 0x2e, 0x73, 0x65, 0x72, 0x76, + 0x69, 0x63, 0x65, 0x2e, 0x64, 0x69, 0x73, 0x63, 0x6f, 0x76, 0x65, 0x72, 0x79, 0x2e, 0x76, 0x33, + 0x2e, 0x44, 0x69, 0x73, 0x63, 0x6f, 0x76, 0x65, 0x72, 0x79, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, + 0x74, 0x1a, 0x2d, 0x2e, 0x65, 0x6e, 0x76, 0x6f, 0x79, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, + 0x65, 0x2e, 0x64, 0x69, 0x73, 0x63, 0x6f, 0x76, 0x65, 0x72, 0x79, 0x2e, 0x76, 0x33, 0x2e, 0x44, + 0x69, 0x73, 0x63, 0x6f, 0x76, 0x65, 0x72, 0x79, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, + 0x22, 0x33, 0x82, 0xd3, 0xe4, 0x93, 0x02, 0x24, 0x22, 0x22, 0x2f, 0x76, 0x33, 0x2f, 0x64, 0x69, + 0x73, 0x63, 0x6f, 0x76, 0x65, 0x72, 0x79, 0x3a, 0x6d, 0x6f, 0x6e, 0x69, 0x74, 0x6f, 0x72, 0x69, + 0x6e, 0x67, 0x61, 0x73, 0x73, 0x69, 0x67, 0x6e, 0x6d, 0x65, 0x6e, 0x74, 0x82, 0xd3, 0xe4, 0x93, + 0x02, 0x03, 0x3a, 0x01, 0x2a, 0x42, 0x04, 0x5a, 0x02, 0x76, 0x31, 0x62, 0x06, 0x70, 0x72, 0x6f, + 0x74, 0x6f, 0x33, +} + +var ( + file_observability_v1_mads_proto_rawDescOnce sync.Once + file_observability_v1_mads_proto_rawDescData = file_observability_v1_mads_proto_rawDesc +) + +func file_observability_v1_mads_proto_rawDescGZIP() []byte { + file_observability_v1_mads_proto_rawDescOnce.Do(func() { + file_observability_v1_mads_proto_rawDescData = protoimpl.X.CompressGZIP(file_observability_v1_mads_proto_rawDescData) + }) + return file_observability_v1_mads_proto_rawDescData +} + +var file_observability_v1_mads_proto_msgTypes = make([]protoimpl.MessageInfo, 4) +var file_observability_v1_mads_proto_goTypes = []interface{}{ + (*MonitoringAssignment)(nil), // 0: kuma.observability.v1.MonitoringAssignment + (*MonitoringAssignment_Target)(nil), // 1: kuma.observability.v1.MonitoringAssignment.Target + nil, // 2: kuma.observability.v1.MonitoringAssignment.LabelsEntry + nil, // 3: kuma.observability.v1.MonitoringAssignment.Target.LabelsEntry + (*v3.DeltaDiscoveryRequest)(nil), // 4: envoy.service.discovery.v3.DeltaDiscoveryRequest + (*v3.DiscoveryRequest)(nil), // 5: envoy.service.discovery.v3.DiscoveryRequest + (*v3.DeltaDiscoveryResponse)(nil), // 6: envoy.service.discovery.v3.DeltaDiscoveryResponse + (*v3.DiscoveryResponse)(nil), // 7: envoy.service.discovery.v3.DiscoveryResponse +} +var file_observability_v1_mads_proto_depIdxs = []int32{ + 1, // 0: kuma.observability.v1.MonitoringAssignment.targets:type_name -> kuma.observability.v1.MonitoringAssignment.Target + 2, // 1: kuma.observability.v1.MonitoringAssignment.labels:type_name -> kuma.observability.v1.MonitoringAssignment.LabelsEntry + 3, // 2: kuma.observability.v1.MonitoringAssignment.Target.labels:type_name -> kuma.observability.v1.MonitoringAssignment.Target.LabelsEntry + 4, // 3: kuma.observability.v1.MonitoringAssignmentDiscoveryService.DeltaMonitoringAssignments:input_type -> envoy.service.discovery.v3.DeltaDiscoveryRequest + 5, // 4: kuma.observability.v1.MonitoringAssignmentDiscoveryService.StreamMonitoringAssignments:input_type -> envoy.service.discovery.v3.DiscoveryRequest + 5, // 5: kuma.observability.v1.MonitoringAssignmentDiscoveryService.FetchMonitoringAssignments:input_type -> envoy.service.discovery.v3.DiscoveryRequest + 6, // 6: kuma.observability.v1.MonitoringAssignmentDiscoveryService.DeltaMonitoringAssignments:output_type -> envoy.service.discovery.v3.DeltaDiscoveryResponse + 7, // 7: kuma.observability.v1.MonitoringAssignmentDiscoveryService.StreamMonitoringAssignments:output_type -> envoy.service.discovery.v3.DiscoveryResponse + 7, // 8: kuma.observability.v1.MonitoringAssignmentDiscoveryService.FetchMonitoringAssignments:output_type -> envoy.service.discovery.v3.DiscoveryResponse + 6, // [6:9] is the sub-list for method output_type + 3, // [3:6] is the sub-list for method input_type + 3, // [3:3] is the sub-list for extension type_name + 3, // [3:3] is the sub-list for extension extendee + 0, // [0:3] is the sub-list for field type_name +} + +func init() { file_observability_v1_mads_proto_init() } +func file_observability_v1_mads_proto_init() { + if File_observability_v1_mads_proto != nil { + return + } + if !protoimpl.UnsafeEnabled { + file_observability_v1_mads_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*MonitoringAssignment); i { + case 0: + return &v.state + case 1: + return &v.sizeCache + case 2: + return &v.unknownFields + default: + return nil + } + } + file_observability_v1_mads_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*MonitoringAssignment_Target); i { + case 0: + return &v.state + case 1: + return &v.sizeCache + case 2: + return &v.unknownFields + default: + return nil + } + } + } + type x struct{} + out := protoimpl.TypeBuilder{ + File: protoimpl.DescBuilder{ + GoPackagePath: reflect.TypeOf(x{}).PkgPath(), + RawDescriptor: file_observability_v1_mads_proto_rawDesc, + NumEnums: 0, + NumMessages: 4, + NumExtensions: 0, + NumServices: 1, + }, + GoTypes: file_observability_v1_mads_proto_goTypes, + DependencyIndexes: file_observability_v1_mads_proto_depIdxs, + MessageInfos: file_observability_v1_mads_proto_msgTypes, + }.Build() + File_observability_v1_mads_proto = out.File + file_observability_v1_mads_proto_rawDesc = nil + file_observability_v1_mads_proto_goTypes = nil + file_observability_v1_mads_proto_depIdxs = nil +} + +// MonitoringAssignmentDiscoveryServiceServer is the server API for MonitoringAssignmentDiscoveryService service. +type MonitoringAssignmentDiscoveryServiceServer interface { + // HTTP + FetchMonitoringAssignments(context.Context, *v3.DiscoveryRequest) (*v3.DiscoveryResponse, error) +} diff --git a/discovery/xds/xds.go b/discovery/xds/xds.go new file mode 100644 index 000000000..f99e03da9 --- /dev/null +++ b/discovery/xds/xds.go @@ -0,0 +1,176 @@ +// Copyright 2021 The Prometheus 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 xds + +import ( + "context" + "github.com/prometheus/common/model" + "time" + + v3 "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3" + "github.com/go-kit/log" + "github.com/go-kit/log/level" + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/common/config" + "google.golang.org/protobuf/encoding/protojson" + "google.golang.org/protobuf/proto" + "google.golang.org/protobuf/reflect/protoreflect" + "google.golang.org/protobuf/reflect/protoregistry" + "google.golang.org/protobuf/types/known/anypb" + + "github.com/prometheus/prometheus/discovery" + "github.com/prometheus/prometheus/discovery/targetgroup" +) + +const ( + // Constants for instrumentation. + namespace = "prometheus" +) + +// ProtocolVersion is the xDS protocol version. +type ProtocolVersion string + +const ( + ProtocolV3 = ProtocolVersion("v3") +) + +type HTTPConfig struct { + config.HTTPClientConfig `yaml:",inline"` +} + +// SDConfig is a base config for xDS-based SD mechanisms. +type SDConfig struct { + HTTPClientConfig config.HTTPClientConfig `yaml:",inline"` + RefreshInterval model.Duration `yaml:"refresh_interval,omitempty"` + FetchTimeout model.Duration `yaml:"fetch_timeout,omitempty"` + Server string `yaml:"server,omitempty"` +} + +// mustRegisterMessage registers the provided message type in the typeRegistry, and panics +// if there is an error. +func mustRegisterMessage(typeRegistry *protoregistry.Types, mt protoreflect.MessageType) { + if err := typeRegistry.RegisterMessage(mt); err != nil { + panic(err) + } +} + +func init() { + // Register top-level SD Configs. + discovery.RegisterConfig(&KumaSDConfig{}) + + // Register metrics. + prometheus.MustRegister(kumaFetchDuration, kumaFetchSkipUpdateCount, kumaFetchFailuresCount) + + // Register protobuf types that need to be marshalled/ unmarshalled. + mustRegisterMessage(protoTypes, (&v3.DiscoveryRequest{}).ProtoReflect().Type()) + mustRegisterMessage(protoTypes, (&v3.DiscoveryResponse{}).ProtoReflect().Type()) + mustRegisterMessage(protoTypes, (&MonitoringAssignment{}).ProtoReflect().Type()) +} + +var ( + protoTypes = new(protoregistry.Types) + protoUnmarshalOptions = proto.UnmarshalOptions{ + DiscardUnknown: true, // Only want known fields. + Merge: true, // Always using new messages. + Resolver: protoTypes, // Only want known types. + } + protoJSONUnmarshalOptions = protojson.UnmarshalOptions{ + DiscardUnknown: true, // Only want known fields. + Resolver: protoTypes, // Only want known types. + } + protoJSONMarshalOptions = protojson.MarshalOptions{ + UseProtoNames: true, + Resolver: protoTypes, // Only want known types. + } +) + +type resourceParser func(resources []*anypb.Any, typeUrl string) ([]*targetgroup.Group, error) + +// fetchDiscovery implements long-polling via xDS Fetch REST-JSON. +type fetchDiscovery struct { + client ResourceClient + source string + + refreshInterval time.Duration + + parseResources resourceParser + logger log.Logger + + fetchDuration prometheus.Observer + fetchSkipUpdateCount prometheus.Counter + fetchFailuresCount prometheus.Counter +} + +func (d *fetchDiscovery) Run(ctx context.Context, ch chan<- []*targetgroup.Group) { + defer d.client.Close() + + ticker := time.NewTicker(d.refreshInterval) + + for { + select { + case <-ctx.Done(): + ticker.Stop() + return + default: + d.poll(ctx, ch) + <-ticker.C + } + } +} + +func (d *fetchDiscovery) poll(ctx context.Context, ch chan<- []*targetgroup.Group) { + t0 := time.Now() + response, err := d.client.Fetch(ctx) + elapsed := time.Since(t0) + d.fetchDuration.Observe(elapsed.Seconds()) + + // Check the context before in order to exit early. + select { + case <-ctx.Done(): + return + default: + } + + if err != nil { + level.Error(d.logger).Log("msg", "error parsing resources", "err", err) + d.fetchFailuresCount.Inc() + return + } + + if response == nil { + // No update needed. + d.fetchSkipUpdateCount.Inc() + return + } + + parsedGroups, err := d.parseResources(response.Resources, response.TypeUrl) + if err != nil { + level.Error(d.logger).Log("msg", "error parsing resources", "err", err) + d.fetchFailuresCount.Inc() + return + } + + for _, group := range parsedGroups { + group.Source = d.source + } + + level.Debug(d.logger).Log("msg", "updated to version", "version", response.VersionInfo, "groups", len(parsedGroups)) + + // Check the context before sending an update on the channel. + select { + case <-ctx.Done(): + return + case ch <- parsedGroups: + } +} diff --git a/go.mod b/go.mod index bf36d90d4..1610e5042 100644 --- a/go.mod +++ b/go.mod @@ -18,6 +18,8 @@ require ( github.com/docker/docker v20.10.7+incompatible github.com/docker/go-connections v0.4.0 // indirect github.com/edsrzf/mmap-go v1.0.0 + github.com/envoyproxy/go-control-plane v0.9.9-0.20210217033140-668b12f5399d + github.com/envoyproxy/protoc-gen-validate v0.1.0 github.com/go-kit/kit v0.10.0 github.com/go-kit/log v0.1.0 github.com/go-logfmt/logfmt v0.5.0 @@ -62,6 +64,8 @@ require ( golang.org/x/time v0.0.0-20210611083556-38a9dc6acbc6 golang.org/x/tools v0.1.3 google.golang.org/api v0.48.0 + google.golang.org/genproto v0.0.0-20210604141403-392c879c8b08 + google.golang.org/protobuf v1.26.0 gopkg.in/alecthomas/kingpin.v2 v2.2.6 gopkg.in/fsnotify/fsnotify.v1 v1.4.7 gopkg.in/yaml.v2 v2.4.0 diff --git a/go.sum b/go.sum index 57c620a52..70d13bd17 100644 --- a/go.sum +++ b/go.sum @@ -227,6 +227,7 @@ github.com/clbanning/x2j v0.0.0-20191024224557-825249438eec/go.mod h1:jMjuTZXRI4 github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw= github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGXZJjfX53e64911xZQV5JYwmTeXPW+k8Sc= github.com/cncf/udpa/go v0.0.0-20200629203442-efcf912fb354/go.mod h1:WmhPx2Nbnhtbo57+VJT5O0JRkEi1Wbu0z5j0R8u5Hbk= +github.com/cncf/udpa/go v0.0.0-20201120205902-5459f2c99403 h1:cqQfy1jclcSy/FwLjemeg3SR1yaINm74aQyupQ0Bl8M= github.com/cncf/udpa/go v0.0.0-20201120205902-5459f2c99403/go.mod h1:WmhPx2Nbnhtbo57+VJT5O0JRkEi1Wbu0z5j0R8u5Hbk= github.com/cockroachdb/datadriven v0.0.0-20190809214429-80d97fb3cbaa/go.mod h1:zn76sxSg3SzpJ0PPJaLDCu+Bu0Lg3sKTORVIj19EIF8= github.com/codahale/hdrhistogram v0.0.0-20161010025455-3a0bb77429bd/go.mod h1:sE/e/2PUdi/liOCUjSTXgM1o87ZssimdTWN964YiIeI= @@ -393,7 +394,9 @@ github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.m github.com/envoyproxy/go-control-plane v0.9.4/go.mod h1:6rpuAdCZL397s3pYoYcLgu1mIlRU8Am5FuJP05cCM98= github.com/envoyproxy/go-control-plane v0.9.7/go.mod h1:cwu0lG7PUMfa9snN8LXBig5ynNVH9qI8YYLbd1fK2po= github.com/envoyproxy/go-control-plane v0.9.9-0.20201210154907-fd9021fe5dad/go.mod h1:cXg6YxExXjJnVBQHBLXeUAgxn2UodCpnH306RInaBQk= +github.com/envoyproxy/go-control-plane v0.9.9-0.20210217033140-668b12f5399d h1:QyzYnTnPE15SQyUeqU6qLbWxMkwyAyu+vGksa0b7j00= github.com/envoyproxy/go-control-plane v0.9.9-0.20210217033140-668b12f5399d/go.mod h1:cXg6YxExXjJnVBQHBLXeUAgxn2UodCpnH306RInaBQk= +github.com/envoyproxy/protoc-gen-validate v0.1.0 h1:EQciDnbrYxy13PgWoY8AqoxGiPrpgBZ1R8UNe3ddc+A= github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c= github.com/evanphx/json-patch v4.2.0+incompatible/go.mod h1:50XU6AFN0ol/bzJsmQLiYLvXMP4fmwYFNcr97nuDLSk= github.com/evanphx/json-patch v4.9.0+incompatible h1:kLcOMZeuLAJvL2BPWLMIj5oaZQobrkAqrL+WFZwQses=