1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341
|
// Copyright 2015 opentsdb-goclient authors. All Rights Reserved.
//
// Licensed 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 client defines the client and the corresponding
// rest api implementaion of OpenTSDB.
//
// query.go contains the structs and methods for the implementation of /api/query.
//
package client
import (
"bytes"
"encoding/json"
"errors"
"fmt"
"sort"
"strconv"
"strings"
)
// QueryParam is the structure used to hold
// the querying parameters when calling /api/query.
// Each attributes in QueryParam matches the definition in
// (http://opentsdb.net/docs/build/html/api_http/query/index.html).
//
type QueryParam struct {
// The start time for the query. This can be a relative or absolute timestamp.
// The data type can only be string, int, or int64.
// The value is required with non-zero value of the target type.
Start interface{} `json:"start"`
// An end time for the query. If not supplied, the TSD will assume the local
// system time on the server. This may be a relative or absolute timestamp.
// The data type can only be string, or int64.
// The value is optional.
End interface{} `json:"end,omitempty"`
// One or more sub queries used to select the time series to return.
// These may be metric m or TSUID tsuids queries
// The value is required with at least one element
Queries []SubQuery `json:"queries"`
// An optional value is used to show whether or not to return annotations with a query.
// The default is to return annotations for the requested timespan but this flag can disable the return.
// This affects both local and global notes and overrides globalAnnotations
NoAnnotations bool `json:"noAnnotations,omitempty"`
// An optional value is used to show whether or not the query should retrieve global
// annotations for the requested timespan.
GlobalAnnotations bool `json:"globalAnnotations,omitempty"`
// An optional value is used to show whether or not to output data point timestamps in milliseconds or seconds.
// If this flag is not provided and there are multiple data points within a second,
// those data points will be down sampled using the query's aggregation function.
MsResolution bool `json:"msResolution,omitempty"`
// An optional value is used to show whether or not to output the TSUIDs associated with timeseries in the results.
// If multiple time series were aggregated into one set, multiple TSUIDs will be returned in a sorted manner.
ShowTSUIDs bool `json:"showTSUIDs,omitempty"`
// An optional value is used to show whether or not can be paased to the JSON with a POST to delete any data point
// that match the given query.
Delete bool `json:"delete,omitempty"`
}
func (query *QueryParam) String() string {
content, _ := json.Marshal(query)
return string(content)
}
// SubQuery is the structure used to hold
// the subquery parameters when calling /api/query.
// Each attributes in SubQuery matches the definition in
// (http://opentsdb.net/docs/build/html/api_http/query/index.html).
//
type SubQuery struct {
// The name of an aggregation function to use.
// The value is required with non-empty one in the range of
// the response of calling /api/aggregators.
//
// By default, the potential values and corresponding descriptions are as followings:
// "sum": Adds all of the data points for a timestamp.
// "min": Picks the smallest data point for each timestamp.
// "max": Picks the largest data point for each timestamp.
// "avg": Averages the values for the data points at each timestamp.
Aggregator string `json:"aggregator"`
// The name of a metric stored in the system.
// The value is reqiured with non-empty value.
Metric string `json:"metric"`
// An optional value is used to show whether or not the data should be
// converted into deltas before returning. This is useful if the metric is a
// continously incrementing counter and you want to view the rate of change between data points.
Rate bool `json:"rate,omitempty"`
// rateOptions represents monotonically increasing counter handling options.
// The value is optional.
// Currently there is only three kind of value can be set to this map:
// Only three keys can be set into the rateOption parameter of the QueryParam is
// QueryRateOptionCounter (value type is bool), QueryRateOptionCounterMax (value type is int,int64)
// QueryRateOptionResetValue (value type is int,int64)
RateParams map[string]interface{} `json:"rateOptions,omitempty"`
// An optional value downsampling function to reduce the amount of data returned.
Downsample string `json:"downsample,omitempty"`
// An optional value to drill down to specific timeseries or group results by tag,
// supply one or more map values in the same format as the query string. Tags are converted to filters in 2.2.
// Note that if no tags are specified, all metrics in the system will be aggregated into the results.
// It will be deprecated in OpenTSDB 2.2.
Tags map[string]string `json:"tags,omitempty"`
// An optional value used to filter the time series emitted in the results.
// Note that if no filters are specified, all time series for the given
// metric will be aggregated into the results.
Fiters []Filter `json:"filters,omitempty"`
}
// Filter is the structure used to hold the filter parameters when calling /api/query.
// Each attributes in Filter matches the definition in
// (http://opentsdb.net/docs/build/html/api_http/query/index.html).
//
type Filter struct {
// The name of the filter to invoke. The value is required with a non-empty
// value in the range of calling /api/config/filters.
Type string `json:"type"`
// The tag key to invoke the filter on, required with a non-empty value
Tagk string `json:"tagk"`
// The filter expression to evaluate and depends on the filter being used, required with a non-empty value
FilterExp string `json:"filter"`
// An optional value to show whether or not to group the results by each value matched by the filter.
// By default all values matching the filter will be aggregated into a single series.
GroupBy bool `json:"groupBy"`
}
// QueryResponse acts as the implementation of Response in the /api/query scene.
// It holds the status code and the response values defined in the
// (http://opentsdb.net/docs/build/html/api_http/query/index.html).
//
type QueryResponse struct {
StatusCode int
QueryRespCnts []QueryRespItem `json:"queryRespCnts"`
ErrorMsg map[string]interface{} `json:"error"`
}
func (queryResp *QueryResponse) String() string {
buffer := bytes.NewBuffer(nil)
content, _ := json.Marshal(queryResp)
buffer.WriteString(fmt.Sprintf("%s\n", string(content)))
return buffer.String()
}
func (queryResp *QueryResponse) SetStatus(code int) {
queryResp.StatusCode = code
}
func (queryResp *QueryResponse) GetCustomParser() func(respCnt []byte) error {
return func(respCnt []byte) error {
originRespStr := string(respCnt)
var respStr string
if queryResp.StatusCode == 200 && strings.Contains(originRespStr, "[") && strings.Contains(originRespStr, "]") {
respStr = fmt.Sprintf("{%s:%s}", `"queryRespCnts"`, originRespStr)
} else {
respStr = originRespStr
}
return json.Unmarshal([]byte(respStr), &queryResp)
}
}
// QueryRespItem acts as the implementation of Response in the /api/query scene.
// It holds the response item defined in the
// (http://opentsdb.net/docs/build/html/api_http/query/index.html).
//
type QueryRespItem struct {
// Name of the metric retreived for the time series
Metric string `json:"metric"`
// A list of tags only returned when the results are for a single time series.
// If results are aggregated, this value may be null or an empty map
Tags map[string]string `json:"tags"`
// If more than one timeseries were included in the result set, i.e. they were aggregated,
// this will display a list of tag names that were found in common across all time series.
// Note that: Api Doc uses 'aggreatedTags', but actual response uses 'aggregateTags'
AggregatedTags []string `json:"aggregateTags"`
// Retrieved datapoints after being processed by the aggregators. Each data point consists
// of a timestamp and a value, the format determined by the serializer.
// For the JSON serializer, the timestamp will always be a Unix epoch style integer followed
// by the value as an integer or a floating point.
// For example, the default output is "dps"{"<timestamp>":<value>}.
// By default the timestamps will be in seconds. If the msResolution flag is set, then the
// timestamps will be in milliseconds.
//
// Because the elements of map is out of order, using common way to iterate Dps will not get
// datapoints with timestamps out of order.
// So be aware that one should use '(qri *QueryRespItem) GetDataPoints() []*DataPoint' to
// acquire the real ascending datapoints.
Dps map[string]interface{} `json:"dps"`
// If the query retrieved annotations for timeseries over the requested timespan, they will
// be returned in this group. Annotations for every timeseries will be merged into one set
// and sorted by start_time. Aggregator functions do not affect annotations, all annotations
// will be returned for the span.
// The value is optional.
Annotations []Annotation `json:"annotations,omitempty"`
// If requested by the user, the query will scan for global annotations during
// the timespan and the results returned in this group.
// The value is optional.
GlobalAnnotations []Annotation `json:"globalAnnotations,omitempty"`
}
// GetDataPoints returns the real ascending datapoints from the information of the related QueryRespItem.
func (qri *QueryRespItem) GetDataPoints() []*DataPoint {
datapoints := make([]*DataPoint, 0)
timestampStrs := qri.getSortedTimestampStrs()
for _, timestampStr := range timestampStrs {
timestamp, _ := strconv.ParseInt(timestampStr, 10, 64)
datapoint := &DataPoint{
Metric: qri.Metric,
Value: qri.Dps[timestampStr],
Tags: qri.Tags,
Timestamp: timestamp,
}
datapoints = append(datapoints, datapoint)
}
return datapoints
}
// getSortedTimestampStrs returns a slice of the ascending timestamp with
// string format for the Dps of the related QueryRespItem instance.
func (qri *QueryRespItem) getSortedTimestampStrs() []string {
timestampStrs := make([]string, 0)
for timestampStr := range qri.Dps {
timestampStrs = append(timestampStrs, timestampStr)
}
sort.Strings(timestampStrs)
return timestampStrs
}
// GetLatestDataPoint returns latest datapoint for the related QueryRespItem instance.
func (qri *QueryRespItem) GetLatestDataPoint() *DataPoint {
timestampStrs := qri.getSortedTimestampStrs()
size := len(timestampStrs)
if size == 0 {
return nil
}
timestamp, _ := strconv.ParseInt(timestampStrs[size-1], 10, 64)
datapoint := &DataPoint{
Metric: qri.Metric,
Value: qri.Dps[timestampStrs[size-1]],
Tags: qri.Tags,
Timestamp: timestamp,
}
return datapoint
}
func (c *clientImpl) Query(param QueryParam) (*QueryResponse, error) {
if !isValidQueryParam(¶m) {
return nil, errors.New("The given query param is invalid.\n")
}
queryEndpoint := fmt.Sprintf("%s%s", c.tsdbEndpoint, QueryPath)
reqBodyCnt, err := getQueryBodyContents(¶m)
if err != nil {
return nil, err
}
queryResp := QueryResponse{}
if err = c.sendRequest(PostMethod, queryEndpoint, reqBodyCnt, &queryResp); err != nil {
return nil, err
}
return &queryResp, nil
}
func getQueryBodyContents(param interface{}) (string, error) {
result, err := json.Marshal(param)
if err != nil {
return "", errors.New(fmt.Sprintf("Failed to marshal query param: %v\n", err))
}
return string(result), nil
}
func isValidQueryParam(param *QueryParam) bool {
if param.Queries == nil || len(param.Queries) == 0 {
return false
}
if !isValidTimePoint(param.Start) {
return false
}
for _, query := range param.Queries {
if len(query.Aggregator) == 0 || len(query.Metric) == 0 {
return false
}
for k, _ := range query.RateParams {
if k != QueryRateOptionCounter && k != QueryRateOptionCounterMax && k != QueryRateOptionResetValue {
return false
}
}
}
return true
}
func isValidTimePoint(timePoint interface{}) bool {
if timePoint == nil {
return false
}
switch v := timePoint.(type) {
case int:
if v <= 0 {
return false
}
case int64:
if v <= 0 {
return false
}
case string:
if v == "" {
return false
}
default:
return false
}
return true
}
|