GoFr 集成 OpenTSDB：时间序列数据写入、查询与可观测性实战指南

GoFr 集成 OpenTSDB时间序列数据写入、查询与可观测性实战指南【免费下载链接】gofrAn opinionated GoLang framework for accelerated microservice development. Built in support for databases and observability.项目地址: https://gitcode.com/GitHub_Trending/go/gofr本文围绕 GoFr 框架对 OpenTSDB 的内置支持展开讲解如何通过环境变量配置连接、使用app.AddOpenTSDB()注入实例并基于OpenTSDB接口完成数据点写入/api/put、区间查询/api/query、最新数据点查询/api/query/last以及注解Annotation的增删改查同时介绍其内置的日志、指标与链路追踪能力。读完本文你将能在自己的 GoFr 微服务中快速接入 OpenTSDB实现带完整可观测性的时序指标读写。OpenTSDB 与 GoFr 的集成方式OpenTSDB 是一个基于 HBase 构建的分布式、可扩展的时间序列数据库通过 REST API 对外提供服务。GoFr 在pkg/gofr/datasource/opentsdb中提供了一个官方客户端实现独立模块gofr.dev/pkg/gofr/datasource/opentsdb它封装了 OpenTSDB 的 HTTP API使 GoFr 应用无需手写 HTTP 调用即可完成时序数据的写入与查询。从源码看客户端将 OpenTSDB 的 REST 端点映射为一个个 Go 方法对应的端点常量定义在 opentsdb.go 中OpenTSDB REST 端点GoFr 接口方法说明POST /api/putPutDataPoints写入数据点GET /api/queryQueryDataPoints区间查询数据点GET /api/query/lastQueryLatestDataPoints查询最新数据点OpenTSDB v2.2GET /api/aggregatorsGetAggregators获取可用聚合函数GET /api/annotationQueryAnnotation查询注解POST /api/annotationPostAnnotation创建/更新注解PUT /api/annotationPutAnnotation创建/替换注解未提供的字段重置为默认值DELETE /api/annotationDeleteAnnotation删除注解GET /api/version内部调用连接时校验版本、健康检查时获取版本信息配置环境变量与 Config 字段要连接 OpenTSDB需要为服务提供以下环境变量对应Config结构体的字段见 opentsdb.goHOSTOpenTSDB 服务器的 hostname 或 IP 地址含端口格式为ip:port不带http://前缀。MAXCONTENTLENGTH单个请求体的最大长度字节。默认值为40960字节该默认值基于 OpenTSDB 配置tsd.http.request.enable_chunked true且tsd.http.request.max_chunk 40960的前提。MAXPUTPOINTSNUM单次PUT请求最多可发送的数据点数量默认值为75。DETECTDELTANUM客户端在写入大组数据点时用于探测数据点之间异常时间间隔的数据点数量默认值为3。注意官方文档的配置章节将环境变量写作HOSTS而随后的示例代码与源码Config字段实际使用的是Host示例中通过app.Config.Get(HOST)读取。请以你实际使用的环境变量名为准保持Config字段与读取键一致。Config结构体还包含一个可选项Transport *http.Transport用于自定义 HTTP 传输层若不设置客户端会使用默认 transport启用 TCP keepalive拨号超时 5 秒、连接保活 30 秒见initializeClient中的defaultTransport实现。当MaxPutPointsNum、DetectDeltaNum、MaxContentLength三个字段未设置或小于等于 0 时客户端会自动回退到上述默认值preprocess.go。安装与注册OpenTSDB 客户端是独立的 Go 模块通过以下命令引入go get gofr.dev/pkg/gofr/datasource/opentsdb该模块的go.mod位于 pkg/gofr/datasource/opentsdb/go.mod依赖 OpenTelemetry用于链路追踪与指标等库。GoFr 支持注入 OpenTSDB 以便与 OpenTSDB 的 REST API 交互。任何满足OpenTSDB接口的实现都可以通过app.AddOpenTSDB()注册从而在gofr.Context中直接使用。注册方法的实现位于 external_db.go// AddOpenTSDB sets the OpenTSDB datasource in the apps container. func (a *App) AddOpenTSDB(db container.OpenTSDB) { a.instrumentDatasource(db) a.container.OpenTSDB db }其中instrumentDatasource会为数据源注入日志器、指标与 tracer即 duck-typed 接入可观测性。OpenTSDB接口的完整定义位于 container/datasources.go接口方法如下// OpenTSDB provides methods for GoFr applications to communicate with OpenTSDB // through its REST APIs. Each method corresponds to an API endpoint defined in the // OpenTSDB documentation (http://opentsdb.net/docs/build/html/api_http/index.html#api-endpoints). type OpenTSDB interface { // HealthChecker verifies if the OpenTSDB server is reachable. // Returns an error if the server is unreachable, otherwise nil. HealthChecker // PutDataPoints sends data to the POST /api/put endpoint to store metrics in OpenTSDB. // // Parameters: // - ctx: Context for managing request lifetime. // - data: A slice of DataPoint objects; must contain at least one entry. // - queryParam: Specifies the response format: // - client.PutRespWithSummary: Requests a summary response. // - client.PutRespWithDetails: Requests detailed response information. // - Empty string (): No additional response details. // - res: A pointer to PutResponse, where the servers response will be stored. // // Returns: // - Error if parameters are invalid, response parsing fails, or if connectivity issues occur. PutDataPoints(ctx context.Context, data any, queryParam string, res any) error // QueryDataPoints retrieves data using the GET /api/query endpoint based on the specified parameters. // // Parameters: // - ctx: Context for managing request lifetime. // - param: An instance of QueryParam with query parameters for filtering data. // - res: A pointer to QueryResponse, where the servers response will be stored. // // Returns: // - Error if parameters are invalid, response parsing fails, or if connectivity issues occur. QueryDataPoints(ctx context.Context, param any, res any) error // QueryLatestDataPoints fetches the latest data point(s) using the GET /api/query/last endpoint, // supported in OpenTSDB v2.2 and later. // // Parameters: // - ctx: Context for managing request lifetime. // - param: An instance of QueryLastParam with query parameters for the latest data point. // - res: A pointer to QueryLastResponse, where the servers response will be stored. // // Returns: // - Error if parameters are invalid, response parsing fails, or if connectivity issues occur. QueryLatestDataPoints(ctx context.Context, param any, res any) error // GetAggregators retrieves available aggregation functions using the GET /api/aggregators endpoint. // // Parameters: // - ctx: Context for managing request lifetime. // - res: A pointer to AggregatorsResponse, where the servers response will be stored. // // Returns: // - Error if response parsing fails or if connectivity issues occur. GetAggregators(ctx context.Context, res any) error // QueryAnnotation retrieves a single annotation from OpenTSDB using the GET /api/annotation endpoint. // // Parameters: // - ctx: Context for managing request lifetime. // - queryAnnoParam: A map of parameters for the annotation query, such as client.AnQueryStartTime, client.AnQueryTSUid. // - res: A pointer to AnnotationResponse, where the servers response will be stored. // // Returns: // - Error if parameters are invalid, response parsing fails, or if connectivity issues occur. QueryAnnotation(ctx context.Context, queryAnnoParam map[string]any, res any) error // PostAnnotation creates or updates an annotation in OpenTSDB using the POST /api/annotation endpoint. // // Parameters: // - ctx: Context for managing request lifetime. // - annotation: The annotation to be created or updated. // - res: A pointer to AnnotationResponse, where the servers response will be stored. // // Returns: // - Error if parameters are invalid, response parsing fails, or if connectivity issues occur. PostAnnotation(ctx context.Context, annotation any, res any) error // PutAnnotation creates or replaces an annotation in OpenTSDB using the PUT /api/annotation endpoint. // Fields not included in the request will be reset to default values. // // Parameters: // - ctx: Context for managing request lifetime. // - annotation: The annotation to be created or replaced. // - res: A pointer to AnnotationResponse, where the servers response will be stored. // // Returns: // - Error if parameters are invalid, response parsing fails, or if connectivity issues occur. PutAnnotation(ctx context.Context, annotation any, res any) error // DeleteAnnotation removes an annotation from OpenTSDB using the DELETE /api/annotation endpoint. // // Parameters: // - ctx: Context for managing request lifetime. // - annotation: The annotation to be deleted. // - res: A pointer to AnnotationResponse, where the servers response will be stored. // // Returns: // - Error if parameters are invalid, response parsing fails, or if connectivity issues occur. DeleteAnnotation(ctx context.Context, annotation any, res any) error }客户端实现Clientopentsdb.go在Connect()阶段会完成三件事注册指标、基于Host拼接http://host作为 endpoint并通过GET /api/version校验服务器可达性opentsdb.go。完整示例健康检查、写入与查询下面的示例演示了在 GoFr 应用中注入 OpenTSDB 实例并实现服务器健康检查、写入数据点和区间查询三个功能package main import ( context fmt math/rand/v2 time gofr.dev/pkg/gofr gofr.dev/pkg/gofr/datasource/opentsdb ) func main() { app : gofr.New() // Initialize OpenTSDB connection app.AddOpenTSDB(opentsdb.New(opentsdb.Config{ Host: app.Config.Get(HOST), MaxContentLength: app.Config.Get(MAXCONTENTLENGTH), MaxPutPointsNum: app.Config.Get(MAXPUTPOINTSNUM), DetectDeltaNum: app.Config.Get(DETECTDELTANUM), })) // Register routes app.GET(/health, opentsdbHealthCheck) app.POST(/write, writeDataPoints) app.GET(/query, queryDataPoints) // Run the app app.Run() } // Health check for OpenTSDB func opentsdbHealthCheck(c *gofr.Context) (any, error) { res, err : c.OpenTSDB.HealthCheck(context.Background()) if err ! nil { return nil, err } return res, nil } // Write Data Points to OpenTSDB func writeDataPoints(c *gofr.Context) (any, error) { PutDataPointNum : 4 name : []string{cpu, disk, net, mem} cpuDatas : make([]opentsdb.DataPoint, 0) tags : map[string]string{ host: gofr-host, try-name: gofr-sample, demo-name: opentsdb-test, } for i : 0; i PutDataPointNum; i { data : opentsdb.DataPoint{ Metric: name[i%len(name)], Timestamp: time.Now().Unix(), Value: rand.Float64() * 100, Tags: tags, } cpuDatas append(cpuDatas, data) } resp : opentsdb.PutResponse{} err : c.OpenTSDB.PutDataPoints(context.Background(), cpuDatas, details, resp) if err ! nil { return resp.Errors, err } return fmt.Sprintf(%v Data points written successfully, resp.Success), nil } // Query Data Points from OpenTSDB func queryDataPoints(c *gofr.Context) (any, error) { st1 : time.Now().Unix() - 3600 st2 : time.Now().Unix() queryParam : opentsdb.QueryParam{ Start: st1, End: st2, } name : []string{cpu, disk, net, mem} subqueries : make([]opentsdb.SubQuery, 0) tags : map[string]string{ host: gofr-host, try-name: gofr-sample, demo-name: opentsdb-test, } for _, metric : range name { subQuery : opentsdb.SubQuery{ Aggregator: sum, Metric: metric, Tags: tags, } subqueries append(subqueries, subQuery) } queryParam.Queries subqueries queryResp : opentsdb.QueryResponse{} err : c.OpenTSDB.QueryDataPoints(c, queryParam, queryResp) if err ! nil { return nil, err } return queryResp.QueryRespCnts, nil }写入数据点DataPoint 与响应格式DataPoint是写入/api/put的核心结构定义于 preprocess.go其字段要求如下Metric必填指标名称非空字符串。Timestamp必填Unix 纪元时间戳秒或毫秒必须是非零值且只允许数字字符通常用time.Now().Unix()生成。Value必填数值类型仅支持int、int64、float64、float32或string传入其他类型会在客户端本地校验阶段直接报错isValidDataPoint见 preprocess.go。Tags必填tag 名/值映射至少一对。OpenTSDB 默认最多支持 8 个 tag可通过opentsdb.conf中的tsd.storage.max_tags修改实践中建议控制在 45 个以内。PutDataPoints的queryParam参数控制响应粒度客户端会校验其取值isValidPutParam传入summary只返回汇总信息传入details返回每个数据点写入失败的详细信息PutResponse.Errors中会包含PutError{Data, ErrorMsg}传空字符串则不附加额外响应信息。请求体会通过POST发送到http://host/api/putContent-Type 为application/json; charsetUTF-8见sendRequestresponse.go。PutResponse的结构为type PutResponse struct { Failed int64 json:failed Success int64 json:success Errors []PutError json:errors,omitempty }当部分数据点写入失败时客户端会聚合错误信息返回并在响应中保留Errors供业务侧排查。此外MaxPutPointsNum默认 75与DETECTDELTANUM默认 3用于控制大数据量写入时的分批策略避免单次请求体过大。查询数据点QueryParam、SubQuery 与聚合QueryDataPoints对应的/api/query支持丰富的过滤与聚合参数QueryParam结构preprocess.go字段如下字段必填说明Start是起始时间支持string相对时间如1h-ago、int、int64绝对时间戳须为非零值End否结束时间支持string或int64缺省时使用服务器本地时间Queries是一个或多个子查询[]SubQuery至少一个元素NoAnnotations否是否不返回注解默认会返回查询时间范围内的注解GlobalAnnotations否是否检索全局注解MsResolution否时间戳输出为毫秒还是秒若不设置且同一秒内有多个数据点将用聚合函数降采样ShowTSUIDs否结果中是否输出关联的 TSUIDDelete否是否删除匹配查询的数据点SubQuerypreprocess.go用于描述具体要查询的时间序列Aggregator必填聚合函数名取值须在/api/aggregators返回范围内。常见内置值sum同一时间戳的所有数据点求和、min取最小、max取最大、avg取平均。Metric必填系统中存储的指标名。Rate可选是否在返回前将数据转换为增量适用于持续递增的计数器指标。RateParams可选单调递增计数器处理选项仅允许三个键counterbool、counterMaxint/int64、resetValueint/int64。DownSample可选降采样函数减少返回数据量。Tags可选按 tag 下钻到特定时间序列或分组若未指定系统内该指标的所有序列都会被聚合进结果。Filters可选过滤器列表Filter{Type, Tagk, FilterExp, GroupBy}用于过滤结果中的时间序列。查询结果QueryResponse.QueryRespCnts中的每一项QueryRespItem包含Metric、Tags、AggregatedTags注意 JSON 字段是aggregateTags、Dps时间戳到值的映射默认秒级时间戳、Annotations与GlobalAnnotations。由于Dps是 map遍历顺序不确定源码注释明确建议通过GetDataPoints()获取按时间升序排列的数据点。查询最新数据点使用QueryLatestDataPoints对应QueryLastParam{Queries, ResolveNames, BackScan}ResolveNames决定是否把结果的 TSUID 解析为指标名与 tag 名BackScan表示向前回溯搜索的小时数为 0 时使用时序元数据计数的时间戳。其结果为QueryLastResponse.QueryRespCnts每项含Metric、Tags、Timestamp毫秒、Value字符串形式与TSUID十六进制。注解管理Annotation 的增删改查注解用于在特定时间点记录事件说明常用于图形化展示或 API 查询。Annotation结构response.go包含StartTime必填事件发生的 Unix 时间戳秒。EndTime可选事件结束时间戳。TSUID可选若注解关联到特定时间序列填其标识。Description可选事件简要说明建议控制在 25 字符以内。Notes可选详细描述。Custom可选任意附加键值对。四个注解方法对应四种语义QueryAnnotation以map[string]any传参如start_time、tsuid键通过GET查询PostAnnotation创建或更新POSTPutAnnotation创建或替换PUT未提供的字段会重置为默认值DeleteAnnotation删除DELETE。它们统一走operateAnnotation内部方法response.go响应均为AnnotationResponse内嵌Annotation并附带ErrorInfo。此外GetAggregators通过GET /api/aggregators获取可用聚合函数列表。内置可观测性日志、指标与链路追踪GoFr 的 OpenTSDB 客户端在 observability.go 中实现了与框架一致的可观测性接入指标registerMetrics注册了两个指标——直方图app_opentsdb_operation_duration记录操作耗时毫秒与计数器app_opentsdb_operation_total记录操作总数并附带operation、status、host标签见sendOperationStats。链路追踪每个操作都会创建名为opentsdb-operation的 span并在 span 上记录opentsdb.operation.duration微秒等属性。日志操作完成后输出结构化的QueryLog含Operation、Duration微秒、Status、Message并支持PrettyPrint终端美化输出操作状态为SUCCESS或FAIL。健康检查HealthCheck先通过 TCP 拨号超时 5 秒探测连通性再调用/api/version获取版本最终返回Health{Status: UP|DOWN, Details: {host, version}}opentsdb.go。该接口会被 GoFr 容器统一纳入数据源健康检查见 container/health.go。源码结构速览如果你想深入了解实现细节可以从以下文件入手opentsdb.go客户端Client、Config、New()、连接初始化与全部业务方法。preprocess.goDataPoint、QueryParam、SubQuery、Filter、QueryLastParam、PutResponse等请求/响应结构以及参数校验与请求体组装。response.go各 REST API 的响应结构、自定义解析器与sendRequest通用请求发送逻辑。observability.go指标、日志、追踪的统一上报。interface.goLogger、Metrics、httpClient等内部接口抽象。container/datasources.go框架层面的OpenTSDB接口定义OpenTSDBProvider已标记为废弃推荐直接实现OpenTSDB接口。external_db.goapp.AddOpenTSDB()注册入口。小结通过 GoFr 的 OpenTSDB 数据源你可以用极少的样板代码完成时间序列数据的写入、查询、最新值获取与注解管理并自动获得与框架一致的日志、Prometheus 指标和 OpenTelemetry 链路追踪。配置上只需提供HOST、MAXCONTENTLENGTH、MAXPUTPOINTSNUM、DETECTDELTANUM四个环境变量其余参数均有合理的默认值兜底接入上只需一行app.AddOpenTSDB(...)即可在任意 handler 中通过c.OpenTSDB使用全部能力。【免费下载链接】gofrAn opinionated GoLang framework for accelerated microservice development. Built in support for databases and observability.项目地址: https://gitcode.com/GitHub_Trending/go/gofr创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考