elastic/elastic-transport-go
GitHub: elastic/elastic-transport-go
Elastic 官方的 Go 传输层共享库，为 Elasticsearch 等 Elastic 服务的 Go 客户端提供连接池、节点发现、重试、指标和日志等基础设施能力。
Stars: 14 | Forks: 22
# elastic-transport-go
这个库源自 elasticsearch-net，随后经过改造，可用于所有的 Elastic 服务，而不仅仅是
Elasticsearch。
它提供了 `go-elasticsearch` 使用的 Transport 接口、连接池、集群发现以及多种 logger。
## 安装
将该包添加到你的 go.mod 文件中：
`require github.com/elastic/elastic-transport-go/v8 main`
## 用法
### Transport
Transport 提供了访问 Elasticsearch API 的基础层。使用
`NewClient` 和函数式选项创建一个 client：
```
package main
import (
"context"
"log"
"net/http"
"net/url"
"time"
"github.com/elastic/elastic-transport-go/v8/elastictransport"
)
func main() {
u, _ := url.Parse("http://127.0.0.1:9200")
transport, err := elastictransport.NewClient(
elastictransport.WithURLs(u),
)
if err != nil {
log.Fatalln(err)
}
defer func() {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
_ = transport.Close(ctx)
}()
req, _ := http.NewRequest("GET", "/", nil)
res, err := transport.Perform(req)
if err != nil {
log.Fatalln(err)
}
defer res.Body.Close()
log.Println(res)
}
```
选项会按顺序应用；当同一个设置被多次指定时，
以最后一个值为准。请在
[包文档](https://pkg.go.dev/github.com/elastic/elastic-transport-go/v8/elastictransport)
中查看 `With*` 函数以获取所有可用选项的完整列表。
常见示例：
```
// Multiple nodes with basic auth, custom retries, and compression
transport, err := elastictransport.NewClient(
elastictransport.WithURLs(u1, u2, u3),
elastictransport.WithBasicAuth("elastic", "changeme"),
elastictransport.WithRetry(5, 429, 502, 503, 504),
elastictransport.WithRetryBackoff(func(attempt int) time.Duration {
return time.Duration(attempt) * 100 * time.Millisecond
}),
elastictransport.WithCompression(gzip.BestSpeed),
)
```
### 发现
Discovery 模块会调用集群以获取其完整的节点列表。
一旦你的 transport 设置完毕，你可以像这样轻松触发此行为：
```
err := transport.DiscoverNodes()
```
或者在创建 client 时配置自动的定期发现：
```
transport, err := elastictransport.NewClient(
elastictransport.WithURLs(u),
elastictransport.WithDiscoverNodesInterval(5 * time.Minute),
)
```
### 指标
允许你直接从 transport 获取指标。在创建 client 时启用指标：
```
transport, err := elastictransport.NewClient(
elastictransport.WithURLs(u),
elastictransport.WithMetrics(),
)
```
### 分级日志
`WithLeveledLogger` 为 transport 内部事件（连接管理、节点发现）
设置了一个结构化的分级 logger。`LeveledLogger` 接口
使用与 `log/slog` 相同的 `(msg, keysAndValues...)` 约定：
```
type LeveledLogger interface {
Debug(msg string, keysAndValues ...any)
Info(msg string, keysAndValues ...any)
Warn(msg string, keysAndValues ...any)
Error(msg string, keysAndValues ...any)
}
```
添加 `LoggingInterceptor` 以便通过同一个
logger 记录请求/响应往返。这是获得完整日志的推荐方式：
```
transport, err := elastictransport.NewClient(
elastictransport.WithURLs(u),
elastictransport.WithLeveledLogger(&elastictransport.SlogLogger{
Logger: slog.New(slog.NewJSONHandler(os.Stderr, &slog.HandlerOptions{
Level: slog.LevelDebug,
})),
}),
elastictransport.WithInterceptors(
elastictransport.LoggingInterceptor(false, false),
),
)
```
成功的往返将以 **Info** 级别记录；错误以 **Error** 级别记录；
连接管理事件根据严重程度以 **Debug**、**Warn** 或 **Error** 级别记录。
#### Body 日志
`LoggingInterceptor` 接受两个布尔值，用于启用请求和/或响应
body 捕获：
```
elastictransport.WithInterceptors(
elastictransport.LoggingInterceptor(true, true), // request body, response body
)
```
#### 内置 slog Handler
`sloghandler` 子包为所有已弃用的 logger 提供了直接替换的
`slog.Handler`：
| 已弃用的 Logger | sloghandler 替代方案 |
| ----------------- | ---------------------------------- |
| `TextLogger` | `sloghandler.NewTextHandler(w)` |
| `ColorLogger` | `sloghandler.NewColorHandler(w)` |
| `CurlLogger` | `sloghandler.NewCurlHandler(w)` |
| `JSONLogger` | `sloghandler.NewJSONECSHandler(w)` |
```
import "github.com/elastic/elastic-transport-go/v8/elastictransport/sloghandler"
transport, err := elastictransport.NewClient(
elastictransport.WithURLs(u),
elastictransport.WithLeveledLogger(&elastictransport.SlogLogger{
Logger: slog.New(sloghandler.NewColorHandler(os.Stderr)),
}),
elastictransport.WithInterceptors(
elastictransport.LoggingInterceptor(false, false),
),
)
```
#### 自定义 Logger 实现
针对 zap、zerolog、logrus 和 logr 的现成 adapter 位于
[`_examples/logging/adapters/`](./_examples/logging/adapters/) 中。
要使用不同的日志库，只需在一个轻量级的包装器上实现这四个方法：
```
type ZapLeveledLogger struct{ Logger *zap.SugaredLogger }
func (l *ZapLeveledLogger) Debug(ctx context.Context, msg string, kv ...any) { l.Logger.Debugw(msg, kv...) }
func (l *ZapLeveledLogger) Info(ctx context.Context, msg string, kv ...any) { l.Logger.Infow(msg, kv...) }
func (l *ZapLeveledLogger) Warn(ctx context.Context, msg string, kv ...any) { l.Logger.Warnw(msg, kv...) }
func (l *ZapLeveledLogger) Error(ctx context.Context, msg string, kv ...any) { l.Logger.Errorw(msg, kv...) }
```
#### Context 集成
logger 会在 `Perform` 期间被注入到请求 context 中，从而使
自定义 interceptor 可以通过 `LoggerFromContext` 获取它。调用者可以使用
`ContextWithLogger` 为单个请求覆盖 logger：
```
interceptor := func(next elastictransport.RoundTripFunc) elastictransport.RoundTripFunc {
return func(req *http.Request) (*http.Response, error) {
if logger := elastictransport.LoggerFromContext(req.Context()); logger != nil {
logger.Debug("before request", "method", req.Method)
}
return next(req)
}
}
```
#### 从 WithDebugLogger / WithLogger 迁移
`WithDebugLogger()` 和 `WithLogger()` 仍然有效，但已弃用。在
底层，`WithDebugLogger` 现在会创建一个封装了 `slog.Default()` 的 `SlogLogger`。
| | `WithDebugLogger()` | `WithLogger()` | `WithLeveledLogger()` + `LoggingInterceptor` |
| --------------------- | ------------------- | --------------- | -------------------------------------------- |
| 往返日志 | 否 | 是 | 是（通过 interceptor） |
| 连接事件 | 是（仅 Debug） | 否 | 是 |
| 输出目标 | stdout | 用户控制 | 用户控制 |
| 日志级别 | 仅 Debug | 无 | Debug/Info/Warn/Error |
| 结构化数据 | 否 | 否 | 是（键值对） |
| 自定义 logger 支持 | 否 | 是 | 是 |
| 单个 client 隔离 | 是 | 是 | 是 |
| Context 注入 | 否 | 否 | 是 |
| 可组合的顺序 | 否 | 否 | 是（interceptor 链） |
替换原有的任意旧选项：
```
// Before (WithDebugLogger)
transport, err := elastictransport.NewClient(
elastictransport.WithURLs(u),
elastictransport.WithDebugLogger(),
)
// Before (WithLogger)
transport, err := elastictransport.NewClient(
elastictransport.WithURLs(u),
elastictransport.WithLogger(&elastictransport.TextLogger{Output: os.Stdout}),
)
// After
transport, err := elastictransport.NewClient(
elastictransport.WithURLs(u),
elastictransport.WithLeveledLogger(&elastictransport.SlogLogger{
Logger: slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{
Level: slog.LevelDebug,
})),
}),
elastictransport.WithInterceptors(
elastictransport.LoggingInterceptor(false, false),
),
)
```
### 请求/响应 Logger（已弃用）
可以通过 `WithLogger` 选项提供一个 logger。有几个内置的 logger
可用：
#### TextLogger
配置：
```
transport, err := elastictransport.NewClient(
elastictransport.WithURLs(u),
elastictransport.WithLogger(&elastictransport.TextLogger{Output: os.Stdout, EnableRequestBody: true, EnableResponseBody: true}),
)
```
输出：
```
< {
< "name" : "es",
< "cluster_name" : "elasticsearch",
< "cluster_uuid" : "RxB1iqTNT9q3LlIkTsmWRA",
< "version" : {
< "number" : "8.0.0-SNAPSHOT",
< "build_flavor" : "default",
< "build_type" : "docker",
< "build_hash" : "0564e027dc6c69236937b1edcc04c207b4cd8128",
< "build_date" : "2021-11-25T00:23:33.139514432Z",
< "build_snapshot" : true,
< "lucene_version" : "9.0.0",
< "minimum_wire_compatibility_version" : "7.16.0",
< "minimum_index_compatibility_version" : "7.0.0"
< },
< "tagline" : "You Know, for Search"
< }
```
#### JSONLogger
配置：
```
transport, err := elastictransport.NewClient(
elastictransport.WithURLs(u),
elastictransport.WithLogger(&elastictransport.JSONLogger{Output: os.Stdout, EnableRequestBody: true, EnableResponseBody: true}),
)
```
输出：
```
{
"@timestamp": "2021-11-25T16:33:51Z",
"event": {
"duration": 2892269
},
"url": {
"scheme": "http",
"domain": "127.0.0.1",
"port": 9200,
"path": "/",
"query": ""
},
"http": {
"request": {
"method": "GET"
},
"response": {
"status_code": 200,
"body": "{\n \"name\" : \"es1\",\n \"cluster_name\" : \"go-elasticsearch\",\n \"cluster_uuid\" : \"RxB1iqTNT9q3LlIkTsmWRA\",\n \"version\" : {\n \"number\" : \"8.0.0-SNAPSHOT\",\n \"build_flavor\" : \"default\",\n \"build_type\" : \"docker\",\n \"build_hash\" : \"0564e027dc6c69236937b1edcc04c207b4cd8128\",\n \"build_date\" : \"2021-11-25T00:23:33.139514432Z\",\n \"build_snapshot\" : true,\n \"lucene_version\" : \"9.0.0\",\n \"minimum_wire_compatibility_version\" : \"8.0.0\",\n \"minimum_index_compatibility_version\" : \"7.0.0\"\n },\n \"tagline\" : \"You Know, for Search\"\n}\n"
}
}
}
```
#### ColorLogger
配置：
```
transport, err := elastictransport.NewClient(
elastictransport.WithURLs(u),
elastictransport.WithLogger(&elastictransport.ColorLogger{Output: os.Stdout, EnableRequestBody: true, EnableResponseBody: true}),
)
```
输出：
```
GET http://127.0.0.1:9200/ 200 OK 2ms
« {
« "name" : "es1",
« "cluster_name" : "go-elasticsearch",
« "cluster_uuid" : "RxB1iqTNT9q3LlIkTsmWRA",
« "version" : {
« "number" : "8.0.0-SNAPSHOT",
« "build_flavor" : "default",
« "build_type" : "docker",
« "build_hash" : "0564e027dc6c69236937b1edcc04c207b4cd8128",
« "build_date" : "2021-11-25T00:23:33.139514432Z",
« "build_snapshot" : true,
« "lucene_version" : "9.0.0",
« "minimum_wire_compatibility_version" : "7.16.0",
« "minimum_index_compatibility_version" : "7.0.0"
« },
« "tagline" : "You Know, for Search"
« }
────────────────────────────────────────────────────────────────────────────────
```
#### CurlLogger
配置：
```
transport, err := elastictransport.NewClient(
elastictransport.WithURLs(u),
elastictransport.WithLogger(&elastictransport.CurlLogger{Output: os.Stdout, EnableRequestBody: true, EnableResponseBody: true}),
)
```
输出：
```
curl -X GET 'http://localhost:9200/?pretty'
# => 2021-11-25T16:40:11Z [200 OK] 3ms
# {
# "name": "es1",
# "cluster_name": "go-elasticsearch",
# "cluster_uuid": "RxB1iqTNT9q3LlIkTsmWRA",
# "version": {
# "number": "8.0.0-SNAPSHOT",
# "build_flavor": "default",
# "build_type": "docker",
# "build_hash": "0564e027dc6c69236937b1edcc04c207b4cd8128",
# "build_date": "2021-11-25T00:23:33.139514432Z",
# "build_snapshot": true,
# "lucene_version": "9.0.0",
# "minimum_wire_compatibility_version": "7.16.0",
# "minimum_index_compatibility_version": "7.0.0"
# },
# "tagline": "You Know, for Search"
# }
```
# 许可证
基于 Apache License, Version 2.0 授权。
标签：日志审计