You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by wu...@apache.org on 2021/09/24 08:11:41 UTC
[skywalking-satellite] branch main updated: Support load-balance
GRPC client with Kubenetes selector (#69)
This is an automated email from the ASF dual-hosted git repository.
wusheng pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/skywalking-satellite.git
The following commit(s) were added to refs/heads/main by this push:
new 6292e2f Support load-balance GRPC client with Kubenetes selector (#69)
6292e2f is described below
commit 6292e2fa41305202283587a7d570c1bc54c23c11
Author: mrproliu <74...@qq.com>
AuthorDate: Fri Sep 24 16:11:30 2021 +0800
Support load-balance GRPC client with Kubenetes selector (#69)
---
LICENSE | 2 +
configs/satellite_config.yaml | 2 +-
docs/en/setup/README.md | 1 +
.../examples/grpc-load-balance-client/README.md | 84 ++++++++++++
.../satellite_config.yaml | 2 +-
docs/en/setup/plugins/client_grpc-client.md | 2 +-
docs/menu.yml | 2 +
plugins/client/grpc/client.go | 4 +-
.../client/grpc/resolvers/kubernetes_clients.go | 98 +++++++++++++
plugins/client/grpc/resolvers/kubernetes_config.go | 107 +++++++++++++++
plugins/client/grpc/resolvers/kubernetes_kinds.go | 152 +++++++++++++++++++++
...{resolvers.go => kubernetes_kinds_endpoints.go} | 42 +++---
.../{resolvers.go => kubernetes_kinds_pod.go} | 43 +++---
.../{resolvers.go => kubernetes_kinds_service.go} | 42 +++---
plugins/client/grpc/resolvers/resolvers.go | 4 +-
plugins/client/grpc/resolvers/static_clients.go | 2 +-
16 files changed, 504 insertions(+), 85 deletions(-)
diff --git a/LICENSE b/LICENSE
index dafc585..722153b 100644
--- a/LICENSE
+++ b/LICENSE
@@ -215,6 +215,8 @@ Apache 2.0 licenses
The following components are provided under the Apache License. See project link for details.
The text of each license is the standard Apache 2.0 license.
+prometheus/common v0.15.0: https://github.com/prometheus/common Apache-2.0
+
========================================================================
BSD licenses
========================================================================
diff --git a/configs/satellite_config.yaml b/configs/satellite_config.yaml
index b7b64f6..b61c265 100644
--- a/configs/satellite_config.yaml
+++ b/configs/satellite_config.yaml
@@ -49,7 +49,7 @@ sharing:
insecure_skip_verify: ${SATELLITE_GRPC_INSECURE_SKIP_VERIFY:false}
# The file path oca.pem. The config only works when opening the TLS switch.
ca_pem_path: ${SATELLITE_grpc_CA_PEM_PATH:"ca.pem"}
- # How frequently to check the connection
+ # How frequently to check the connection(second)
check_period: ${SATELLITE_GRPC_CHECK_PERIOD:5}
# The auth value when send request
authentication: ${SATELLITE_GRPC_AUTHENTICATION:""}
diff --git a/docs/en/setup/README.md b/docs/en/setup/README.md
index 930e355..1b0a23a 100644
--- a/docs/en/setup/README.md
+++ b/docs/en/setup/README.md
@@ -21,6 +21,7 @@ You can quickly build your satellite according to the following examples:
1. [Transmit protocol from agent](examples/transmit-protocol-from-agent/README.md)
2. [Transmit Log to Kafka](examples/transmit-log-to-kafka/README.md)
+3. [GRPC load balance client](examples/grpc-load-balance-client/README.md)
## satellite_config.yaml
The core concept behind this setting file is, SkyWalking Satellite is based on pure modularization design. End user can switch or assemble the collector features by their own requirements.
diff --git a/docs/en/setup/examples/grpc-load-balance-client/README.md b/docs/en/setup/examples/grpc-load-balance-client/README.md
new file mode 100644
index 0000000..d792608
--- /dev/null
+++ b/docs/en/setup/examples/grpc-load-balance-client/README.md
@@ -0,0 +1,84 @@
+# GRPC load balance client
+
+GRPC client support connect to multiple server address, and use `round-robin` policy for load-balance server before send each request.
+
+## Server Discovery
+
+Support two ways to locate the server list:
+1. Static server list: Define the server address list.
+2. Kubernetes selector: Define kubernetes pod/service/endpoint, it could be found addresses and dynamic update automatically.
+
+### Static server list
+
+You could see there define two server address and split by ",".
+
+```yaml
+sharing:
+ clients:
+ - plugin_name: "grpc-client"
+ # The gRPC server address (default localhost:11800).
+ server_addr: ${SATELLITE_GRPC_CLIENT:127.0.0.1:11800,127.0.0.2:11800}
+ # The TLS switch
+ enable_TLS: ${SATELLITE_GRPC_ENABLE_TLS:false}
+ # The file path of client.pem. The config only works when opening the TLS switch.
+ client_pem_path: ${SATELLITE_GRPC_CLIENT_PEM_PATH:"client.pem"}
+ # The file path of client.key. The config only works when opening the TLS switch.
+ client_key_path: ${SATELLITE_GRPC_CLIENT_KEY_PATH:"client.key"}
+ # InsecureSkipVerify controls whether a client verifies the server's certificate chain and host name.
+ insecure_skip_verify: ${SATELLITE_GRPC_INSECURE_SKIP_VERIFY:false}
+ # The file path oca.pem. The config only works when opening the TLS switch.
+ ca_pem_path: ${SATELLITE_grpc_CA_PEM_PATH:"ca.pem"}
+ # How frequently to check the connection(second)
+ check_period: ${SATELLITE_GRPC_CHECK_PERIOD:5}
+ # The auth value when send request
+ authentication: ${SATELLITE_GRPC_AUTHENTICATION:""}
+ address: ${SATELLITE_GRPC_ADDRESS:":11800"}
+ # The TLS cert file path.
+ tls_cert_file: ${SATELLITE_GRPC_TLS_KEY_FILE:""}
+ # The TLS key file path.
+ tls_key_file: ${SATELLITE_GRPC_TLS_KEY_FILE:""}
+```
+
+### Kubernetes selector
+
+Using `kubernetes_config` to define the address's finder.
+
+```yaml
+sharing:
+ clients:
+ - plugin_name: "grpc-client"
+ # The kubernetes config to lookup addresses
+ kubernetes_config:
+ # The kubernetes API server address, If not define means using in kubernetes mode to connect
+ api_server: http://localhost:8001/
+ # The kind of api
+ kind: endpoints
+ # Support to lookup namespaces
+ namespaces:
+ - default
+ # The kind selector
+ selector:
+ label: app=productpage
+ # How to get the address exported port
+ extra_port:
+ port: 9080
+ # The TLS switch
+ enable_TLS: ${SATELLITE_GRPC_ENABLE_TLS:false}
+ # The file path of client.pem. The config only works when opening the TLS switch.
+ client_pem_path: ${SATELLITE_GRPC_CLIENT_PEM_PATH:"client.pem"}
+ # The file path of client.key. The config only works when opening the TLS switch.
+ client_key_path: ${SATELLITE_GRPC_CLIENT_KEY_PATH:"client.key"}
+ # InsecureSkipVerify controls whether a client verifies the server's certificate chain and host name.
+ insecure_skip_verify: ${SATELLITE_GRPC_INSECURE_SKIP_VERIFY:false}
+ # The file path oca.pem. The config only works when opening the TLS switch.
+ ca_pem_path: ${SATELLITE_grpc_CA_PEM_PATH:"ca.pem"}
+ # How frequently to check the connection(second)
+ check_period: ${SATELLITE_GRPC_CHECK_PERIOD:5}
+ # The auth value when send request
+ authentication: ${SATELLITE_GRPC_AUTHENTICATION:""}
+ address: ${SATELLITE_GRPC_ADDRESS:":11800"}
+ # The TLS cert file path.
+ tls_cert_file: ${SATELLITE_GRPC_TLS_KEY_FILE:""}
+ # The TLS key file path.
+ tls_key_file: ${SATELLITE_GRPC_TLS_KEY_FILE:""}
+```
diff --git a/docs/en/setup/examples/transmit-protocol-from-agent/satellite_config.yaml b/docs/en/setup/examples/transmit-protocol-from-agent/satellite_config.yaml
index b7b64f6..b61c265 100644
--- a/docs/en/setup/examples/transmit-protocol-from-agent/satellite_config.yaml
+++ b/docs/en/setup/examples/transmit-protocol-from-agent/satellite_config.yaml
@@ -49,7 +49,7 @@ sharing:
insecure_skip_verify: ${SATELLITE_GRPC_INSECURE_SKIP_VERIFY:false}
# The file path oca.pem. The config only works when opening the TLS switch.
ca_pem_path: ${SATELLITE_grpc_CA_PEM_PATH:"ca.pem"}
- # How frequently to check the connection
+ # How frequently to check the connection(second)
check_period: ${SATELLITE_GRPC_CHECK_PERIOD:5}
# The auth value when send request
authentication: ${SATELLITE_GRPC_AUTHENTICATION:""}
diff --git a/docs/en/setup/plugins/client_grpc-client.md b/docs/en/setup/plugins/client_grpc-client.md
index 5f2d9ea..240e22a 100755
--- a/docs/en/setup/plugins/client_grpc-client.md
+++ b/docs/en/setup/plugins/client_grpc-client.md
@@ -24,6 +24,6 @@ insecure_skip_verify: true
# The auth value when send request
authentication: ""
-# How frequently to check the connection
+# How frequently to check the connection(second)
check_period: 5
```
diff --git a/docs/menu.yml b/docs/menu.yml
index 0e17902..85760b9 100644
--- a/docs/menu.yml
+++ b/docs/menu.yml
@@ -53,6 +53,8 @@ catalog:
path: /en/setup/examples/transmit-protocol-from-agent/readme
- name: Transmit log to kafka
path: /en/setup/examples/transmit-log-to-kafka/readme
+ - name: GRPC load balance client
+ path: /en/setup/examples/grpc-load-balance-client/readme
- name: Plugins
catalog:
- name: client
diff --git a/plugins/client/grpc/client.go b/plugins/client/grpc/client.go
index 71d1b47..0431a27 100644
--- a/plugins/client/grpc/client.go
+++ b/plugins/client/grpc/client.go
@@ -44,7 +44,7 @@ type Client struct {
CaPemPath string `mapstructure:"ca_pem_path"` // The file path oca.pem. The config only works when opening the TLS switch.
InsecureSkipVerify bool `mapstructure:"insecure_skip_verify"` // Controls whether a client verifies the server's certificate chain and host name.
Authentication string `mapstructure:"authentication"` // The auth value when send request
- CheckPeriod int `mapstructure:"check_period"` // How frequently to check the connection
+ CheckPeriod int `mapstructure:"check_period"` // How frequently to check the connection(second)
// components
status api.ClientStatus
@@ -85,7 +85,7 @@ insecure_skip_verify: true
# The auth value when send request
authentication: ""
-# How frequently to check the connection
+# How frequently to check the connection(second)
check_period: 5
`
}
diff --git a/plugins/client/grpc/resolvers/kubernetes_clients.go b/plugins/client/grpc/resolvers/kubernetes_clients.go
new file mode 100644
index 0000000..3bbe5ef
--- /dev/null
+++ b/plugins/client/grpc/resolvers/kubernetes_clients.go
@@ -0,0 +1,98 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you 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 resolvers
+
+import (
+ "context"
+ "encoding/json"
+ "fmt"
+
+ "google.golang.org/grpc/resolver"
+
+ "github.com/apache/skywalking-satellite/internal/pkg/log"
+)
+
+var kubernetesServerSchema = "kubernetes"
+
+type kubernetesServerResolver struct {
+}
+
+func (k *kubernetesServerResolver) IsSupport(c *ServerFinderConfig) bool {
+ return c.KubernetesConfig != nil
+}
+
+func (k *kubernetesServerResolver) BuildTarget(c *ServerFinderConfig) (string, error) {
+ marshal, err := json.Marshal(c.KubernetesConfig)
+ if err != nil {
+ return "", fmt.Errorf("convert kubernetes config error: %v", err)
+ }
+ return fmt.Sprintf("%s:///%s", kubernetesServerSchema, string(marshal)), nil
+}
+
+func (*kubernetesServerResolver) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOptions) (resolver.Resolver, error) {
+ // convert data
+ kubernetesConfig := &KubernetesConfig{}
+ if err := json.Unmarshal([]byte(target.Endpoint), kubernetesConfig); err != nil {
+ return nil, fmt.Errorf("could not analyze the address: %v", err)
+ }
+
+ // validate http config
+ if kubernetesConfig.APIServer != "" {
+ httpConfig, err := kubernetesConfig.HTTPClientConfig.convertHTTPConfig()
+ if err != nil {
+ return nil, err
+ }
+ if err = httpConfig.Validate(); err != nil {
+ return nil, fmt.Errorf("http config validate error: %v", err)
+ }
+ }
+
+ // init cache
+ ctx, cancel := context.WithCancel(context.Background())
+ cache, err := NewKindCache(ctx, kubernetesConfig, cc)
+ if err != nil {
+ cancel()
+ return nil, err
+ }
+
+ // build resolver
+ r := &kubernetesResolver{
+ cache: cache,
+ cancel: cancel,
+ }
+ return r, nil
+}
+
+func (*kubernetesServerResolver) Scheme() string {
+ return kubernetesServerSchema
+}
+
+type kubernetesResolver struct {
+ cache *KindCache
+ cancel context.CancelFunc
+}
+
+func (k *kubernetesResolver) ResolveNow(o resolver.ResolveNowOptions) {
+ if err := k.cache.UpdateAddresses(); err != nil {
+ log.Logger.Warnf("error update static grpc client list: %v", err)
+ }
+}
+
+func (k *kubernetesResolver) Close() {
+ k.cancel()
+}
diff --git a/plugins/client/grpc/resolvers/kubernetes_config.go b/plugins/client/grpc/resolvers/kubernetes_config.go
new file mode 100644
index 0000000..b3a1f93
--- /dev/null
+++ b/plugins/client/grpc/resolvers/kubernetes_config.go
@@ -0,0 +1,107 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you 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 resolvers
+
+import (
+ "fmt"
+ "net/url"
+
+ "github.com/prometheus/common/config"
+ "gopkg.in/yaml.v3"
+)
+
+type KubernetesConfig struct {
+ // The kubernetes API server address, If not define means using in kubernetes mode to connect
+ APIServer string `mapstructure:"api_server"`
+ // Connect to API Server Config
+ HTTPClientConfig HTTPClientConfig `mapstructure:",squash" yaml:",inline"`
+ // Support to lookup namespaces
+ Namespaces []string `mapstructure:"namespaces"`
+ // The kind of api
+ Kind string `mapstructure:"kind"`
+ // The kind selector
+ Selector Selector `mapstructure:"selector"`
+ // How to get the address exported port
+ ExtraPort ExtraPort `mapstructure:"extra_port"`
+}
+
+// HTTPClientConfig configures an HTTP client.
+type HTTPClientConfig struct {
+ // The HTTP basic authentication credentials for the targets.
+ BasicAuth *BasicAuth `mapstructure:"basic_auth" yaml:"basic_auth,omitempty"`
+ // The bearer token for the targets.
+ BearerToken Secret `mapstructure:"bearer_token" yaml:"bearer_token,omitempty"`
+ // The bearer token file for the targets.
+ BearerTokenFile string `mapstructure:"bearer_token_file" yaml:"bearer_token_file,omitempty"`
+ // HTTP proxy server to use to connect to the targets.
+ ProxyURL URL `mapstructure:"proxy_url" yaml:"proxy_url,omitempty"`
+ // TLSConfig to use to connect to the targets.
+ TLSConfig TLSConfig `mapstructure:"tls_config" yaml:"tls_config,omitempty"`
+}
+
+// URL is a custom URL type that allows validation at configuration load time.
+type URL struct {
+ *url.URL
+}
+
+// BasicAuth contains basic HTTP authentication credentials.
+type BasicAuth struct {
+ Username string `mapstructure:"username" yaml:"username"`
+ Password Secret `mapstructure:"password" yaml:"password,omitempty"`
+ PasswordFile string `mapstructure:"password_file" yaml:"password_file,omitempty"`
+}
+
+// TLSConfig configures the options for TLS connections.
+type TLSConfig struct {
+ // The CA cert to use for the targets.
+ CAFile string `mapstructure:"ca_file" yaml:"ca_file,omitempty"`
+ // The client cert file for the targets.
+ CertFile string `mapstructure:"cert_file" yaml:"cert_file,omitempty"`
+ // The client key file for the targets.
+ KeyFile string `mapstructure:"key_file" yaml:"key_file,omitempty"`
+ // Used to verify the hostname for the targets.
+ ServerName string `mapstructure:"server_name" yaml:"server_name,omitempty"`
+ // Disable target certificate validation.
+ InsecureSkipVerify bool `mapstructure:"insecure_skip_verify" yaml:"insecure_skip_verify"`
+}
+
+// Secret special type for storing secrets.
+type Secret string
+
+type Selector struct {
+ Label string `mapstructure:"label" yaml:"label,omitempty"`
+ Field string `mapstructure:"field" yaml:"field,omitempty"`
+}
+
+type ExtraPort struct {
+ Port int `mapstructure:"port"`
+}
+
+// convert config data
+func (c *HTTPClientConfig) convertHTTPConfig() (*config.HTTPClientConfig, error) {
+ marshal, err := yaml.Marshal(c)
+ if err != nil {
+ return nil, fmt.Errorf("could not identity the http client config: %v", err)
+ }
+
+ out := &config.HTTPClientConfig{}
+ if err = yaml.Unmarshal(marshal, out); err != nil {
+ return nil, fmt.Errorf("could not convert http client: %v", err)
+ }
+ return out, nil
+}
diff --git a/plugins/client/grpc/resolvers/kubernetes_kinds.go b/plugins/client/grpc/resolvers/kubernetes_kinds.go
new file mode 100644
index 0000000..3e46f69
--- /dev/null
+++ b/plugins/client/grpc/resolvers/kubernetes_kinds.go
@@ -0,0 +1,152 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you 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 resolvers
+
+import (
+ "context"
+ "fmt"
+ "net/url"
+
+ "github.com/prometheus/prometheus/discovery"
+
+ "github.com/apache/skywalking-satellite/internal/pkg/log"
+
+ "google.golang.org/grpc/resolver"
+
+ "github.com/prometheus/common/config"
+ "github.com/prometheus/prometheus/discovery/kubernetes"
+ "github.com/prometheus/prometheus/discovery/targetgroup"
+)
+
+var analyzers = []KindAddressAnalyzer{
+ &PodAnalyzer{},
+ &ServiceAnalyzer{},
+ &EndpointsAnalyzer{},
+}
+
+type KindCache struct {
+ config *KubernetesConfig
+ cache map[string]*targetgroup.Group
+ cc resolver.ClientConn
+ analyzer KindAddressAnalyzer
+}
+
+type KindAddressAnalyzer interface {
+ KindType() string
+ GetAddresses(cache map[string]*targetgroup.Group, config *KubernetesConfig) []string
+}
+
+func NewKindCache(ctx context.Context, c *KubernetesConfig, cc resolver.ClientConn) (*KindCache, error) {
+ // build config
+ conf := &kubernetes.SDConfig{}
+ if c.APIServer != "" {
+ parsed, err := url.Parse(c.APIServer)
+ if err != nil {
+ return nil, err
+ }
+ conf.APIServer = config.URL{URL: parsed}
+ httpConfig, err := c.HTTPClientConfig.convertHTTPConfig()
+ if err != nil {
+ return nil, err
+ }
+ conf.HTTPClientConfig = *httpConfig
+ }
+
+ conf.Role = kubernetes.Role(c.Kind)
+ conf.NamespaceDiscovery = kubernetes.NamespaceDiscovery{
+ Names: c.Namespaces,
+ }
+ conf.Selectors = []kubernetes.SelectorConfig{
+ {
+ Role: kubernetes.Role(c.Kind),
+ Label: c.Selector.Label,
+ Field: c.Selector.Field,
+ },
+ }
+
+ // build discovery
+ discoverer, err := kubernetes.New(&logAdapt{}, conf)
+ if err != nil {
+ return nil, err
+ }
+
+ // build analyzer
+ var analyzer KindAddressAnalyzer
+ for _, a := range analyzers {
+ if a.KindType() == c.Kind {
+ analyzer = a
+ }
+ }
+ if analyzer == nil {
+ return nil, fmt.Errorf("could not kind analyzer: %s", c.Kind)
+ }
+
+ // build and start watch
+ kind := &KindCache{config: c, cc: cc, analyzer: analyzer}
+ kind.watchAndUpdate(ctx, discoverer)
+ return kind, nil
+}
+
+func (w *KindCache) watchAndUpdate(ctx context.Context, discoverer discovery.Discoverer) {
+ ch := make(chan []*targetgroup.Group)
+ go discoverer.Run(ctx, ch)
+
+ w.cache = make(map[string]*targetgroup.Group)
+ go func() {
+ for {
+ select {
+ case tgs := <-ch:
+ for _, tg := range tgs {
+ if tg.Targets == nil || len(tg.Targets) == 0 {
+ delete(w.cache, tg.Source)
+ continue
+ }
+ w.cache[tg.Source] = tg
+
+ // dynamic update addresses
+ if err := w.UpdateAddresses(); err != nil {
+ log.Logger.Warnf("dynamic update addresss failure, %v", err)
+ }
+ }
+ case <-ctx.Done():
+ break
+ }
+ }
+ }()
+}
+
+func (w *KindCache) UpdateAddresses() error {
+ addresses := w.analyzer.GetAddresses(w.cache, w.config)
+ addrs := make([]resolver.Address, len(addresses))
+ for i, s := range addresses {
+ addrs[i] = resolver.Address{Addr: s}
+ }
+ if err := w.cc.UpdateState(resolver.State{Addresses: addrs}); err != nil {
+ return err
+ }
+ log.Logger.Infof("update grpc client addresses: %v", addresses)
+ return nil
+}
+
+type logAdapt struct {
+}
+
+func (l *logAdapt) Log(keyvals ...interface{}) error {
+ log.Logger.Print(keyvals...)
+ return nil
+}
diff --git a/plugins/client/grpc/resolvers/resolvers.go b/plugins/client/grpc/resolvers/kubernetes_kinds_endpoints.go
similarity index 54%
copy from plugins/client/grpc/resolvers/resolvers.go
copy to plugins/client/grpc/resolvers/kubernetes_kinds_endpoints.go
index 9974d31..f4fddbe 100644
--- a/plugins/client/grpc/resolvers/resolvers.go
+++ b/plugins/client/grpc/resolvers/kubernetes_kinds_endpoints.go
@@ -19,39 +19,29 @@ package resolvers
import (
"fmt"
+ "strings"
- "google.golang.org/grpc/resolver"
+ "github.com/prometheus/common/model"
+ "github.com/prometheus/prometheus/discovery/kubernetes"
+ "github.com/prometheus/prometheus/discovery/targetgroup"
)
-// all customized resolvers
-var rs = []GrpcResolver{
- &staticServerResolver{},
+type EndpointsAnalyzer struct {
}
-type ServerFinderConfig struct {
- ServerAddr string `mapstructure:"server_addr"` // The gRPC server address
+func (p *EndpointsAnalyzer) KindType() string {
+ return string(kubernetes.RoleEndpoint)
}
-type GrpcResolver interface {
- resolver.Builder
-
- // IsSupport client config
- IsSupport(c *ServerFinderConfig) bool
- // BuildTarget address by client config
- BuildTarget(c *ServerFinderConfig) (string, error)
-}
-
-func RegisterAllGrpcResolvers() {
- for _, r := range rs {
- resolver.Register(r)
- }
-}
-
-func BuildTarget(client *ServerFinderConfig) (string, error) {
- for _, r := range rs {
- if r.IsSupport(client) {
- return r.BuildTarget(client)
+func (p *EndpointsAnalyzer) GetAddresses(cache map[string]*targetgroup.Group, config *KubernetesConfig) []string {
+ result := make([]string, 0)
+ for _, group := range cache {
+ for _, target := range group.Targets {
+ address := string(target[model.LabelName("__address__")])
+ if strings.HasSuffix(address, fmt.Sprintf(":%d", config.ExtraPort.Port)) {
+ result = append(result, address)
+ }
}
}
- return "", fmt.Errorf("could not build grpc target")
+ return result
}
diff --git a/plugins/client/grpc/resolvers/resolvers.go b/plugins/client/grpc/resolvers/kubernetes_kinds_pod.go
similarity index 54%
copy from plugins/client/grpc/resolvers/resolvers.go
copy to plugins/client/grpc/resolvers/kubernetes_kinds_pod.go
index 9974d31..efdc32e 100644
--- a/plugins/client/grpc/resolvers/resolvers.go
+++ b/plugins/client/grpc/resolvers/kubernetes_kinds_pod.go
@@ -18,40 +18,31 @@
package resolvers
import (
- "fmt"
+ "strconv"
- "google.golang.org/grpc/resolver"
-)
-
-// all customized resolvers
-var rs = []GrpcResolver{
- &staticServerResolver{},
-}
+ "github.com/prometheus/prometheus/discovery/kubernetes"
-type ServerFinderConfig struct {
- ServerAddr string `mapstructure:"server_addr"` // The gRPC server address
-}
+ "github.com/prometheus/prometheus/discovery/targetgroup"
-type GrpcResolver interface {
- resolver.Builder
+ "github.com/prometheus/common/model"
+)
- // IsSupport client config
- IsSupport(c *ServerFinderConfig) bool
- // BuildTarget address by client config
- BuildTarget(c *ServerFinderConfig) (string, error)
+type PodAnalyzer struct {
}
-func RegisterAllGrpcResolvers() {
- for _, r := range rs {
- resolver.Register(r)
- }
+func (p *PodAnalyzer) KindType() string {
+ return string(kubernetes.RolePod)
}
-func BuildTarget(client *ServerFinderConfig) (string, error) {
- for _, r := range rs {
- if r.IsSupport(client) {
- return r.BuildTarget(client)
+func (p *PodAnalyzer) GetAddresses(cache map[string]*targetgroup.Group, config *KubernetesConfig) []string {
+ result := make([]string, 0)
+ for _, group := range cache {
+ for _, target := range group.Targets {
+ val, exists := target[model.LabelName("__meta_kubernetes_pod_container_port_number")]
+ if exists && string(val) == strconv.Itoa(config.ExtraPort.Port) {
+ result = append(result, string(target[model.LabelName("__address__")]))
+ }
}
}
- return "", fmt.Errorf("could not build grpc target")
+ return result
}
diff --git a/plugins/client/grpc/resolvers/resolvers.go b/plugins/client/grpc/resolvers/kubernetes_kinds_service.go
similarity index 54%
copy from plugins/client/grpc/resolvers/resolvers.go
copy to plugins/client/grpc/resolvers/kubernetes_kinds_service.go
index 9974d31..d3a7575 100644
--- a/plugins/client/grpc/resolvers/resolvers.go
+++ b/plugins/client/grpc/resolvers/kubernetes_kinds_service.go
@@ -19,39 +19,29 @@ package resolvers
import (
"fmt"
+ "strings"
- "google.golang.org/grpc/resolver"
+ "github.com/prometheus/common/model"
+ "github.com/prometheus/prometheus/discovery/kubernetes"
+ "github.com/prometheus/prometheus/discovery/targetgroup"
)
-// all customized resolvers
-var rs = []GrpcResolver{
- &staticServerResolver{},
+type ServiceAnalyzer struct {
}
-type ServerFinderConfig struct {
- ServerAddr string `mapstructure:"server_addr"` // The gRPC server address
+func (p *ServiceAnalyzer) KindType() string {
+ return string(kubernetes.RoleService)
}
-type GrpcResolver interface {
- resolver.Builder
-
- // IsSupport client config
- IsSupport(c *ServerFinderConfig) bool
- // BuildTarget address by client config
- BuildTarget(c *ServerFinderConfig) (string, error)
-}
-
-func RegisterAllGrpcResolvers() {
- for _, r := range rs {
- resolver.Register(r)
- }
-}
-
-func BuildTarget(client *ServerFinderConfig) (string, error) {
- for _, r := range rs {
- if r.IsSupport(client) {
- return r.BuildTarget(client)
+func (p *ServiceAnalyzer) GetAddresses(cache map[string]*targetgroup.Group, config *KubernetesConfig) []string {
+ result := make([]string, 0)
+ for _, group := range cache {
+ for _, target := range group.Targets {
+ address := string(target[model.LabelName("__address__")])
+ if strings.HasSuffix(address, fmt.Sprintf(":%d", config.ExtraPort.Port)) {
+ result = append(result, address)
+ }
}
}
- return "", fmt.Errorf("could not build grpc target")
+ return result
}
diff --git a/plugins/client/grpc/resolvers/resolvers.go b/plugins/client/grpc/resolvers/resolvers.go
index 9974d31..cb84260 100644
--- a/plugins/client/grpc/resolvers/resolvers.go
+++ b/plugins/client/grpc/resolvers/resolvers.go
@@ -26,10 +26,12 @@ import (
// all customized resolvers
var rs = []GrpcResolver{
&staticServerResolver{},
+ &kubernetesServerResolver{},
}
type ServerFinderConfig struct {
- ServerAddr string `mapstructure:"server_addr"` // The gRPC server address
+ ServerAddr string `mapstructure:"server_addr"` // The gRPC server address
+ KubernetesConfig *KubernetesConfig `mapstructure:"kubernetes_config"` // The kubernetes config to lookup addresses
}
type GrpcResolver interface {
diff --git a/plugins/client/grpc/resolvers/static_clients.go b/plugins/client/grpc/resolvers/static_clients.go
index 16987fb..0abe1e3 100644
--- a/plugins/client/grpc/resolvers/static_clients.go
+++ b/plugins/client/grpc/resolvers/static_clients.go
@@ -32,7 +32,7 @@ type staticServerResolver struct {
}
func (s *staticServerResolver) IsSupport(c *ServerFinderConfig) bool {
- return c.ServerAddr != ""
+ return c.KubernetesConfig == nil && c.ServerAddr != ""
}
func (s *staticServerResolver) BuildTarget(c *ServerFinderConfig) (string, error) {