Support load-balance GRPC client with Kubenetes selector (#69)
diff --git a/LICENSE b/LICENSE
index dafc585..722153b 100644
--- a/LICENSE
+++ b/LICENSE
@@ -215,6 +215,8 @@
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 @@
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 @@
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 @@
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 @@
# 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 @@
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 @@
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 @@
# 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/kubernetes_kinds_endpoints.go b/plugins/client/grpc/resolvers/kubernetes_kinds_endpoints.go
new file mode 100644
index 0000000..f4fddbe
--- /dev/null
+++ b/plugins/client/grpc/resolvers/kubernetes_kinds_endpoints.go
@@ -0,0 +1,47 @@
+// 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"
+ "strings"
+
+ "github.com/prometheus/common/model"
+ "github.com/prometheus/prometheus/discovery/kubernetes"
+ "github.com/prometheus/prometheus/discovery/targetgroup"
+)
+
+type EndpointsAnalyzer struct {
+}
+
+func (p *EndpointsAnalyzer) KindType() string {
+ return string(kubernetes.RoleEndpoint)
+}
+
+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 result
+}
diff --git a/plugins/client/grpc/resolvers/kubernetes_kinds_pod.go b/plugins/client/grpc/resolvers/kubernetes_kinds_pod.go
new file mode 100644
index 0000000..efdc32e
--- /dev/null
+++ b/plugins/client/grpc/resolvers/kubernetes_kinds_pod.go
@@ -0,0 +1,48 @@
+// 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 (
+ "strconv"
+
+ "github.com/prometheus/prometheus/discovery/kubernetes"
+
+ "github.com/prometheus/prometheus/discovery/targetgroup"
+
+ "github.com/prometheus/common/model"
+)
+
+type PodAnalyzer struct {
+}
+
+func (p *PodAnalyzer) KindType() string {
+ return string(kubernetes.RolePod)
+}
+
+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 result
+}
diff --git a/plugins/client/grpc/resolvers/kubernetes_kinds_service.go b/plugins/client/grpc/resolvers/kubernetes_kinds_service.go
new file mode 100644
index 0000000..d3a7575
--- /dev/null
+++ b/plugins/client/grpc/resolvers/kubernetes_kinds_service.go
@@ -0,0 +1,47 @@
+// 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"
+ "strings"
+
+ "github.com/prometheus/common/model"
+ "github.com/prometheus/prometheus/discovery/kubernetes"
+ "github.com/prometheus/prometheus/discovery/targetgroup"
+)
+
+type ServiceAnalyzer struct {
+}
+
+func (p *ServiceAnalyzer) KindType() string {
+ return string(kubernetes.RoleService)
+}
+
+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 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 @@
// 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 @@
}
func (s *staticServerResolver) IsSupport(c *ServerFinderConfig) bool {
- return c.ServerAddr != ""
+ return c.KubernetesConfig == nil && c.ServerAddr != ""
}
func (s *staticServerResolver) BuildTarget(c *ServerFinderConfig) (string, error) {
