[Elastic Agent] Add support for multiple hosts in connection to kibana

x-pack/elastic-agent/CHANGELOG.asciidoc

@@ -75,3 +75,4 @@
 - Rename input.type logs to logfile {pull}19360[19360]
 - Agent now installs/uninstalls Elastic Endpoint {pull}19248[19248]
 - Agent now downloads Elastic Endpoint {pull}19503[19503]
+- Agent now load balances across multiple Kibana instances {pull}19628[19628]

x-pack/elastic-agent/_meta/config/common.p2.yml.tmpl

@@ -42,7 +42,7 @@ inputs:
 # access_token: ""
 # kibana:
 # # kibana minimal configuration
-# host: "localhost:5601"
+# hosts: ["localhost:5601"]
 # ssl.certificate_authorities: ["/etc/pki/root/ca.pem"]
 
 # # optional values

x-pack/elastic-agent/_meta/config/common.reference.p2.yml.tmpl

@@ -34,7 +34,7 @@ inputs:
 # access_token: ""
 # kibana:
 # # kibana minimal configuration
-# host: "localhost:5601"
+# hosts: ["localhost:5601"]
 # ssl.certificate_authorities: ["/etc/pki/root/ca.pem"]
 
 # # optional values

x-pack/elastic-agent/_meta/config/elastic-agent.docker.yml.tmpl

@@ -34,7 +34,7 @@ inputs:
 # access_token: ""
 # kibana:
 # # kibana minimal configuration
-# host: "localhost:5601"
+# hosts: ["localhost:5601"]
 # ssl.certificate_authorities: ["/etc/pki/root/ca.pem"]
 
 # # optional values

x-pack/elastic-agent/_meta/elastic-agent.yml

@@ -34,7 +34,7 @@ inputs:
 # access_token: ""
 # kibana:
 # # kibana minimal configuration
-# host: "localhost:5601"
+# hosts: ["localhost:5601"]
 # ssl.certificate_authorities: ["/etc/pki/root/ca.pem"]
 
 # # optional values

x-pack/elastic-agent/elastic-agent.docker.yml

@@ -34,7 +34,7 @@ inputs:
 # access_token: ""
 # kibana:
 # # kibana minimal configuration
-# host: "localhost:5601"
+# hosts: ["localhost:5601"]
 # ssl.certificate_authorities: ["/etc/pki/root/ca.pem"]
 
 # # optional values

x-pack/elastic-agent/elastic-agent.reference.yml

@@ -40,7 +40,7 @@ inputs:
 # access_token: ""
 # kibana:
 # # kibana minimal configuration
-# host: "localhost:5601"
+# hosts: ["localhost:5601"]
 # ssl.certificate_authorities: ["/etc/pki/root/ca.pem"]
 
 # # optional values

x-pack/elastic-agent/elastic-agent.yml

@@ -48,7 +48,7 @@ inputs:
 # access_token: ""
 # kibana:
 # # kibana minimal configuration
-# host: "localhost:5601"
+# hosts: ["localhost:5601"]
 # ssl.certificate_authorities: ["/etc/pki/root/ca.pem"]
 
 # # optional values

x-pack/elastic-agent/pkg/kibana/client.go

@@ -10,6 +10,7 @@ import (
  "net/http"
  "net/url"
  "strings"
+ "sync"
  "time"
 
  "github.com/pkg/errors"
@@ -24,20 +25,19 @@ import (
 const (
  kibanaPort = 5601
  kibanaHTTPSPort = 443
+
+ kibanaRetryOnBadConnTimeout = 5 * time.Minute
 )
 
 type requestFunc func(string, string, url.Values, io.Reader) (*http.Request, error)
 type wrapperFunc func(rt http.RoundTripper) (http.RoundTripper, error)
 
-type clienter interface {
- Send(
- method string,
- path string,
- params url.Values,
- headers http.Header,
- body io.Reader,
- ) (*http.Response, error)
- Close() error
+type requestClient struct {
+ request requestFunc
+ client http.Client
+ lastUsed time.Time
+ lastErr error
+ lastErrOcc time.Time
 }
 
 // Client wraps an http.Client and takes care of making the raw calls to kibana, the client should
@@ -46,27 +46,11 @@ type clienter interface {
 // implementations that will take care of the boiler plates.
 type Client struct {
  log *logger.Logger
- request requestFunc
- client http.Client
+ lock sync.Mutex
+ clients []*requestClient
  config *Config
 }
 
-// New creates new Kibana API client.
-func New(
- log *logger.Logger,
- factory requestFunc,
- cfg *Config,
- httpClient http.Client,
-) (*Client, error) {
- c := &Client{
- log: log,
- request: factory,
- client: httpClient,
- config: cfg,
- }
- return c, nil
-}
-
 // NewConfigFromURL returns a Kibana Config based on a received host.
 func NewConfigFromURL(kURL string) (*Config, error) {
  u, err := url.Parse(kURL)
@@ -112,29 +96,6 @@ func NewWithRawConfig(log *logger.Logger, config *config.Config, wrapper wrapper
 
 // NewWithConfig takes a Kibana Config and return a client.
 func NewWithConfig(log *logger.Logger, cfg *Config, wrapper wrapperFunc) (*Client, error) {
- var transport http.RoundTripper
- transport, err := makeTransport(cfg.Timeout, cfg.TLS)
- if err != nil {
- return nil, err
- }
-
- if cfg.IsBasicAuth() {
- // Pass basic auth credentials to all the underlying calls.
- transport = NewBasicAuthRoundTripper(transport, cfg.Username, cfg.Password)
- }
-
- if wrapper != nil {
- transport, err = wrapper(transport)
- if err != nil {
- return nil, errors.Wrap(err, "fail to create transport client")
- }
- }
-
- httpClient := http.Client{
- Transport: transport,
- Timeout: cfg.Timeout,
- }
-
  // Normalize the URL with the path any spaces configured.
  var p string
  if len(cfg.SpaceID) > 0 {
@@ -152,22 +113,53 @@ func NewWithConfig(log *logger.Logger, cfg *Config, wrapper wrapperFunc) (*Clien
  usedDefaultPort = kibanaHTTPSPort
  }
 
- kibanaURL, err := common.MakeURL(string(cfg.Protocol), p, cfg.Host, usedDefaultPort)
- if err != nil {
- return nil, errors.Wrap(err, "invalid Kibana endpoint")
+ hosts := cfg.GetHosts()
+ clients := make([]*requestClient, len(hosts))
+ for i, host := range cfg.GetHosts() {
+ var transport http.RoundTripper
+ transport, err := makeTransport(cfg.Timeout, cfg.TLS)
+ if err != nil {
+ return nil, err
+ }
+
+ if cfg.IsBasicAuth() {
+ // Pass basic auth credentials to all the underlying calls.
+ transport = NewBasicAuthRoundTripper(transport, cfg.Username, cfg.Password)
+ }
+
+ if wrapper != nil {
+ transport, err = wrapper(transport)
+ if err != nil {
+ return nil, errors.Wrap(err, "fail to create transport client")
+ }
+ }
+
+ httpClient := http.Client{
+ Transport: transport,
+ Timeout: cfg.Timeout,
+ }
+
+ kibanaURL, err := common.MakeURL(string(cfg.Protocol), p, host, usedDefaultPort)
+ if err != nil {
+ return nil, errors.Wrap(err, "invalid Kibana endpoint")
+ }
+ clients[i] = &requestClient{
+ request: prefixRequestFactory(kibanaURL),
+ client: httpClient,
+ }
  }
 
- return New(log, prefixRequestFactory(kibanaURL), cfg, httpClient)
+ return new(log, cfg, clients...)
 }
 
-// Send executes a direct calls agains't the Kibana API, the method will takes cares of cloning
+// Send executes a direct calls against the Kibana API, the method will takes cares of cloning
 // also add necessary headers for Kibana likes: "Content-Type", "Accept", and "kbn-xsrf".
-// No assumptions is done on the response concerning the received format, this will be the responsability
+// No assumptions is done on the response concerning the received format, this will be the responsibility
 // of the implementation to correctly unpack any received data.
 //
 // NOTE:
-// - The caller of this method is free to overrides any values found in the headers.
-// - The magic of unpack kibana errors is not done in the Send method, an helper methods is provided.
+// - The caller of this method is free to override any value found in the headers.
+// - The magic of unpack kibana errors is not done in the Send method, a helper method is provided.
 func (c *Client) Send(
  ctx context.Context,
  method, path string,
@@ -176,8 +168,11 @@ func (c *Client) Send(
  body io.Reader,
 ) (*http.Response, error) {
  c.log.Debugf("Request method: %s, path: %s", method, path)
+ c.lock.Lock()
+ defer c.lock.Unlock()
+ requester := c.nextRequester()
 
- req, err := c.request(method, path, params, body)
+ req, err := requester.request(method, path, params, body)
  if err != nil {
  return nil, errors.Wrapf(err, "fail to create HTTP request using method %s to %s", method, path)
  }
@@ -195,12 +190,79 @@ func (c *Client) Send(
  }
  }
 
- return c.client.Do(req.WithContext(ctx))
+ requester.lastUsed = time.Now().UTC()
+ resp, err := requester.client.Do(req.WithContext(ctx))
+ if err != nil {
+ requester.lastErr = err
+ requester.lastErrOcc = time.Now().UTC()
+ } else {
+ requester.lastErr = nil
+ requester.lastErrOcc = time.Time{}
+ }
+ return resp, err
 }
 
 // URI returns the remote URI.
 func (c *Client) URI() string {
- return string(c.config.Protocol) + "://" + c.config.Host + "/" + c.config.Path
+ host := c.config.GetHosts()[0]
+ return string(c.config.Protocol) + "://" + host + "/" + c.config.Path
+}
+
+// new creates new Kibana API client.
+func new(
+ log *logger.Logger,
+ cfg *Config,
+ httpClients ...*requestClient,
+) (*Client, error) {
+ c := &Client{
+ log: log,
+ clients: httpClients,
+ config: cfg,
+ }
+ return c, nil
+}
+
+// nextRequester returns the requester to use.
+//
+// It excludes clients that have errored in the last 5 minutes.
+func (c *Client) nextRequester() *requestClient {
+ var selected *requestClient
+
+ now := time.Now().UTC()
+ for _, requester := range c.clients {
+ if requester.lastErr != nil && now.Sub(requester.lastErrOcc) > kibanaRetryOnBadConnTimeout {
+ requester.lastErr = nil
+ requester.lastErrOcc = time.Time{}
+ }
+ if requester.lastErr != nil {
+ continue
+ }
+ if requester.lastUsed.IsZero() {
+ // never been used, instant winner!
+ selected = requester
+ break
+ }
+ if selected == nil {
+ selected = requester
+ continue
+ }
+ if requester.lastUsed.Before(selected.lastUsed) {
+ selected = requester
+ }
+ }
+ if selected == nil {
+ // all are erroring; select the oldest one that errored
+ for _, requester := range c.clients {
+ if selected == nil {
+ selected = requester
+ continue
+ }
+ if requester.lastErrOcc.Before(selected.lastErrOcc) {
+ selected = requester
+ }
+ }
+ }
+ return selected
 }
 
 func prefixRequestFactory(URL string) requestFunc {