Skip to content

Commit 46f027f

Browse files
blakerousemergify-bot
authored andcommitted
[Elastic Agent] Use http2 to connect to Fleet Server. (#26474)
* Use http2 to connect to Fleet Server. * Add changelog. * Fix import formatting. * Fix issue with tls and http2. (cherry picked from commit 638b62d)
1 parent eb6df62 commit 46f027f

File tree

3 files changed

+77
-13
lines changed

3 files changed

+77
-13
lines changed

libbeat/common/transport/tls.go

Lines changed: 57 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -73,6 +73,63 @@ func TestTLSDialer(
7373
}), nil
7474
}
7575

76+
type DialerH2 interface {
77+
Dial(network, address string, cfg *tls.Config) (net.Conn, error)
78+
}
79+
80+
type DialerFuncH2 func(network, address string, cfg *tls.Config) (net.Conn, error)
81+
82+
func (d DialerFuncH2) Dial(network, address string, cfg *tls.Config) (net.Conn, error) {
83+
return d(network, address, cfg)
84+
}
85+
86+
func TLSDialerH2(forward Dialer, config *tlscommon.TLSConfig, timeout time.Duration) (DialerH2, error) {
87+
return TestTLSDialerH2(testing.NullDriver, forward, config, timeout)
88+
}
89+
90+
func TestTLSDialerH2(
91+
d testing.Driver,
92+
forward Dialer,
93+
config *tlscommon.TLSConfig,
94+
timeout time.Duration,
95+
) (DialerH2, error) {
96+
var lastTLSConfig *tls.Config
97+
var lastNetwork string
98+
var lastAddress string
99+
var m sync.Mutex
100+
101+
return DialerFuncH2(func(network, address string, cfg *tls.Config) (net.Conn, error) {
102+
switch network {
103+
case "tcp", "tcp4", "tcp6":
104+
default:
105+
return nil, fmt.Errorf("unsupported network type %v", network)
106+
}
107+
108+
host, _, err := net.SplitHostPort(address)
109+
if err != nil {
110+
return nil, err
111+
}
112+
113+
var tlsConfig *tls.Config
114+
m.Lock()
115+
if network == lastNetwork && address == lastAddress {
116+
tlsConfig = lastTLSConfig
117+
}
118+
if tlsConfig == nil {
119+
tlsConfig = config.BuildModuleClientConfig(host)
120+
lastNetwork = network
121+
lastAddress = address
122+
lastTLSConfig = tlsConfig
123+
}
124+
m.Unlock()
125+
126+
// NextProtos must be set from the passed h2 connection or it will fail
127+
tlsConfig.NextProtos = cfg.NextProtos
128+
129+
return tlsDialWith(d, forward, network, address, timeout, tlsConfig, config)
130+
}), nil
131+
}
132+
76133
func tlsDialWith(
77134
d testing.Driver,
78135
dialer Dialer,

x-pack/elastic-agent/CHANGELOG.next.asciidoc

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -116,3 +116,4 @@
116116
- Use `filestream` input for internal log collection. {pull}25660[25660]
117117
- Enable agent to send custom headers to kibana/ES {pull}26275[26275]
118118
- Set `agent.id` to the Fleet Agent ID in events published from inputs backed by Beats. {issue}21121[21121] {pull}26394[26394]
119+
- Communicate with Fleet Server over HTTP2. {pull}26474[26474]

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

Lines changed: 19 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@ import (
1515
"time"
1616

1717
"github.com/pkg/errors"
18+
"golang.org/x/net/http2"
1819

1920
"github.com/elastic/beats/v7/libbeat/common"
2021
"github.com/elastic/beats/v7/libbeat/common/transport"
@@ -113,8 +114,16 @@ func NewWithConfig(log *logger.Logger, cfg Config, wrapper wrapperFunc) (*Client
113114
hosts := cfg.GetHosts()
114115
clients := make([]*requestClient, len(hosts))
115116
for i, host := range cfg.GetHosts() {
116-
var transport http.RoundTripper
117-
transport, err := makeTransport(cfg.Timeout, cfg.TLS)
117+
connStr, err := common.MakeURL(string(cfg.Protocol), p, host, 0)
118+
if err != nil {
119+
return nil, errors.Wrap(err, "invalid fleet-server endpoint")
120+
}
121+
addr, err := url.Parse(connStr)
122+
if err != nil {
123+
return nil, errors.Wrap(err, "invalid fleet-server endpoint")
124+
}
125+
126+
transport, err := makeTransport(addr.Scheme, cfg.Timeout, cfg.TLS)
118127
if err != nil {
119128
return nil, err
120129
}
@@ -136,12 +145,8 @@ func NewWithConfig(log *logger.Logger, cfg Config, wrapper wrapperFunc) (*Client
136145
Timeout: cfg.Timeout,
137146
}
138147

139-
url, err := common.MakeURL(string(cfg.Protocol), p, host, 0)
140-
if err != nil {
141-
return nil, errors.Wrap(err, "invalid fleet-server endpoint")
142-
}
143148
clients[i] = &requestClient{
144-
request: prefixRequestFactory(url),
149+
request: prefixRequestFactory(connStr),
145150
client: httpClient,
146151
}
147152
}
@@ -272,17 +277,18 @@ func prefixRequestFactory(URL string) requestFunc {
272277
}
273278

274279
// makeTransport create a transport object based on the TLS configuration.
275-
func makeTransport(timeout time.Duration, tls *tlscommon.Config) (*http.Transport, error) {
280+
func makeTransport(scheme string, timeout time.Duration, tls *tlscommon.Config) (http.RoundTripper, error) {
281+
dialer := transport.NetDialer(timeout)
282+
if scheme == "http" {
283+
return &http.Transport{Dial: dialer.Dial}, nil
284+
}
276285
tlsConfig, err := tlscommon.LoadTLSConfig(tls)
277286
if err != nil {
278287
return nil, errors.Wrap(err, "invalid TLS configuration")
279288
}
280-
dialer := transport.NetDialer(timeout)
281-
tlsDialer, err := transport.TLSDialer(dialer, tlsConfig, timeout)
289+
tlsDialer, err := transport.TLSDialerH2(dialer, tlsConfig, timeout)
282290
if err != nil {
283291
return nil, errors.Wrap(err, "fail to create TLS dialer")
284292
}
285-
286-
// TODO: Dial is deprecated we need to move to DialContext.
287-
return &http.Transport{Dial: dialer.Dial, DialTLS: tlsDialer.Dial}, nil
293+
return &http2.Transport{DialTLS: tlsDialer.Dial}, nil
288294
}

0 commit comments

Comments
 (0)