Skip to content

Script Development / Connector Object DFF.CONN / DataKit, DataWay

The Connector operation objects for DataKit and DataWay mainly provide data writing methods.

DFF.CONN(...) parameters are as follows:

Parameter Type Required / Default Description
connector_id str Required Connector ID
source str None Override Connector Source
Be careful not to fill in collector names such as "mysql" to avoid confusion
timeout int/float 10 Default HTTP request timeout for the current operation object, in seconds
split_size int 100 Number of data points per request when writing line protocol in batches
Parameter Type Required / Default Description
connector_id str Required Connector ID
token str None Override Connector Token
timeout int/float 10 Default HTTP request timeout for the current operation object, in seconds
split_size int 100 Number of data points per request when writing line protocol in batches
  • For general data reporting, use the .write_by_category(...) and .write_by_category_many(...) methods
  • For general execution of DQL statements, use the .query(...) method
  • To directly send GET requests, use the .get(...) method
  • To directly send POST requests, use the .post_json(...) method
  • To directly send line protocol data, use the .post_line_protocol(...) method

This Connector is essentially a wrapper for HTTP requests

The vast majority of interfaces between DataKit and DataWay are identical.

Since DataKit and DataWay interfaces change frequently, this Connector does not encapsulate all interfaces one-to-one.

Since different versions of DataKit and DataWay may have different requirements or constraints for reported data, use this Connector after reading the relevant documentation.

For detailed documentation, see:

.write_by_category(...)

To write specific types of data to DataKit and DataWay, the parameters are as follows:

Parameter Type Required / Default Description
category str Required Data type, see TrueWatch Documentation / DataKit API
measurement str Required Measurement name
tags dict None Tags. Both key names and values must be strings
fields dict Required Fields. Keys must be strings; values can be strings, integers, floats, booleans, or lists of the above types with consistent element types
timestamp int/long/float {current time} Timestamp, supports seconds/milliseconds/microseconds/nanoseconds
headers dict None Request header parameters
timeout int/float None Timeout for this request; if omitted, the default value of the operation object is used.

The headers parameter was added in 3.3.0

Example
1
2
3
tags   = { 'host': 'web-01' }
fields = { 'cpu' : 10 }
status_code, result = datakit.write_by_category(category='metric', measurement='host_monitoring', tags=tags, fields=fields)

.write_by_category_many(...)

The batch version of write_by_category(...) has the following parameters:

Parameter Type Required / Default Description
category str Required Data type, see TrueWatch Documentation / DataKit API
data list Required List of data points
data[#].measurement str Required Measurement name
data[#].tags dict None Tags. Both key names and values must be strings
data[#].fields dict Required Fields. Keys must be strings; values can be strings, integers, floats, booleans, or lists of the above types with consistent element types
data[#].timestamp int/long/float {current time} Timestamp, supports seconds/milliseconds/microseconds/nanoseconds
headers dict None Request header parameters
timeout int/float None Timeout for this request; if omitted, the default value of the operation object is used.

The headers parameter was added in 3.3.0

Example
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
data = [
    {
        'measurement': 'host_monitoring',
        'tags'       : { 'host' : 'web-01' },
        'fields'     : { 'value': 10 }
    },
    {
        'measurement': 'host_monitoring',
        'tags'       : { 'host' : 'web-02' },
        'fields'     : { 'value': 20 }
    }
]
status_code, result = datakit.write_by_category_many(category='metric', data=data)

.write_metric(...) / .write_point(...)

.write_metric(...) is equivalent to .write_by_category(category='metric', ...); .write_point(...) is a legacy compatibility alias.

Example
1
status_code, result = datakit.write_metric(measurement='host_monitoring', tags={'host': 'web-01'}, fields={'cpu': 10})

.write_metric_many(...) / .write_metrics(...) / .write_points(...)

.write_metric_many(...) is equivalent to .write_by_category_many(category='metric', ...); .write_metrics(...) and .write_points(...) are legacy compatibility aliases.

Example
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
data = [
    {
        'measurement': 'host_monitoring',
        'tags'       : { 'host' : 'web-01' },
        'fields'     : { 'value': 10 }
    },
    {
        'measurement': 'host_monitoring',
        'tags'       : { 'host' : 'web-02' },
        'fields'     : { 'value': 20 }
    }
]
status_code, result = datakit.write_metrics(data=data)

.write_logging(...) / .write_logging_many(...)

.write_logging(...) is equivalent to .write_by_category(category='logging', ...); .write_logging_many(...) is the corresponding batch version.

.query(...)

This method supports parameters in the DataKit and DataWay API DQL query interface.

For detailed documentation, see TrueWatch Documentation / DataKit API Documentation

This method is only a wrapper for HTTP requests

This method essentially just sends an HTTP request to DataKit and DataWay; the returned content depends on DataKit, DataWay, and the backend data source.

If you have any questions about the returned result, you can try using requests to send the request directly to DataKit or DataWay:

Using requests to call the API
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
def query():
    domain = '<Domain>'
    token  = '<Token>'

    url = f'https://{domain}/v1/query/raw?token={token}'
    body = {
        'queries': [
            {
                # DQL statement
                'query': 'M::`cpu`:(`load5s`) BY `host`',

                # Last 1 hour
                'time_range': [
                    _DFF_TRIGGER_TIME_MS - 3600 * 1000,
                    _DFF_TRIGGER_TIME_MS,
                ],
            }
        ],
        'token': token
    }

    resp = requests.post(url, json=body)
    print(resp.status_code)
    print(resp.text)

Execute DQL statements via DataKit, DataWay. The parameters are as follows:

Parameter Type Required / Default Description
dql str Required DQL statement
dict_output bool False Whether to automatically convert data to dict.
raw bool False Whether to return the raw response. When enabled, the dict_output parameter has no effect.
all_series bool False Whether to automatically paginate using slimit and soffset to retrieve all time series.
token str None Workspace Token used only by DataKit; DataWay should set the Token when obtaining the operation object
timeout int/float None Timeout for this request; if omitted, the default value of the operation object is used.
{DataKit/DataWay native parameters} - - Pass through to queries[0].{DataKit/DataWay native parameters}

When all_series is enabled, 500 time series are queried per page: metric queries can request up to 20 pages, and other queries can request up to 5 pages.

DataWay must have a Token before executing a query. It should be set in the Connector configuration or DFF.CONN(..., token='...'); do not pass token again to .query(...).

Example
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
import time
import json

@DFF.API('Run DQL via DataKit')
def run_dql_via_datakit():
    datakit = DFF.CONN('datakit')

    # Use the DataKit native parameter `time_range` to limit data to the last 1 hour
    time_range = [
        int(time.time() - 3600) * 1000,
        int(time.time()) * 1000,
    ]

    # Query and return data as a dict
    status_code, result = datakit.query(dql='O::HOST:(host,load,create_time)', dict_output=True, time_range=time_range)
    print(json.dumps(result))
Output Example
 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
{
  "series": [
    [
      {
        "time": 1622463105293,
        "host": "iZbp152ke14timzud0du15Z",
        "load": 2.18,
        "create_time": 1622429576363,
        "tags": {}
      },
      {
        "time": 1622462905921,
        "host": "ubuntu18-base",
        "load": 0.08,
        "create_time": 1622268259114,
        "tags": {}
      },
      {
        "time": 1622461264175,
        "host": "shenrongMacBook.local",
        "load": 2.395508,
        "create_time": 1622427320834,
        "tags": {}
      }
    ]
  ]
}
Example
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
import time
import json

@DFF.API('Run DQL via DataKit')
def run_dql_via_datakit():
    datakit = DFF.CONN('datakit')

    # Add the raw parameter to get the raw DQL query result
    time_range = [
        int(time.time() - 3600) * 1000,
        int(time.time()) * 1000,
    ]

    # Query and return data in the DataKit raw response format
    status_code, result = datakit.query(dql='O::HOST:(host,load,create_time)', raw=True, time_range=time_range)
    print(json.dumps(result, indent=2))
Output Example
 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
{
  "content": [
    {
      "series": [
        {
          "name": "HOST",
          "columns": [
            "time",
            "host",
            "load",
            "create_time"
          ],
          "values": [
            [
              1622463165152,
              "iZbp152ke14timzud0du15Z",
              1.92,
              1622429576363
            ],
            [
              1622462905921,
              "ubuntu18-base",
              0.08,
              1622268259114
            ],
            [
              1622461264175,
              "shenrongMacBook.local",
              2.395508,
              1622427320834
            ]
          ]
        }
      ],
      "cost": "1ms",
      "total_hits": 3
    }
  ]
}

.get(...)

This method is a general-purpose processing method

For specific parameter formats, contents, etc., please refer to TrueWatch Documentation / DataKit API

Send a GET request to DataKit or DataWay. The parameters are as follows:

Parameter Type Required / Default Description
path str Required Request path
query dict None Request URL parameters
headers dict None Request header parameters
timeout int/float None Timeout for this request; if omitted, the default value of the operation object is used.

Returns (status_code, result). When the response body can be parsed as JSON, result is the corresponding object; otherwise, it is text or raw content.

.post_json(...)

This method is a general-purpose processing method

For specific parameter formats, contents, etc., please refer to TrueWatch Documentation / DataKit API

Send a POST request to DataKit or DataWay in JSON format. The parameters are as follows:

Parameter Type Required / Default Description
path str Required Request path
json_obj dict/list Required The JSON object to send
query dict None Request URL parameters
headers dict None Request header parameters
timeout int/float None Timeout for this request; if omitted, the default value of the operation object is used.

The parameter path was adjusted to be the first parameter in version 1.6.8.

Returns (status_code, result).

.post_line_protocol(...)

This method is a general-purpose processing method

For specific parameter formats, contents, etc., please refer to TrueWatch Documentation / DataKit API

Send a POST request to DataKit or DataWay in line protocol format. The parameters are as follows:

Parameter Type Required / Default Description
path str Required Request path
points dict/list Required A single data point or a list of data points
points[#].measurement str Required Measurement set name
points[#].tags dict None Tags. Both the tag key names and values must be strings.
points[#].fields dict Required Fields. The field key name must be a string; the value can be a string, integer, float, boolean, or a list of the above types with consistent element types.
points[#].timestamp int/long/float {current time} Timestamp. Supports seconds/milliseconds/microseconds/nanoseconds.
query dict None Request URL parameters
headers dict None Request header parameters
timeout int/float None Timeout for this request; if omitted, the default value of the operation object is used.

The parameter path was adjusted to be the first parameter in version 1.6.8.

Batch data is sent in shards according to split_size. This method returns the (status_code, result) of the last shard request; if any shard request fails, an exception is raised and subsequent sending is stopped.