Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_metric_submit_query_switch(self):
"""
Endpoints are different for submission and queries.
"""
Metric.send(points=(123, 456))
self.request_called_with('POST', API_HOST + "/api/v1/series",
data={'series': [{'points': [[123, 456.0]], 'host': api._host_name}]})
Metric.query(start="val1", end="val2")
self.request_called_with('GET', API_HOST + "/api/v1/query",
params={'from': "val1", 'to': "val2"})
def test_host_muting(self):
end = int(time.time()) + 60 * 60
hostname = "my.test.host" + str(end)
message = "Muting this host for a test."
# post a metric to make sure the test host context exists
dog.Metric.send(metric="test.muting.host", points=1, host=hostname)
# Wait for host to appear
get_with_retry("Tag", hostname)
# Mute a host
mute = dog.Host.mute(hostname, end=end, message=message)
assert mute["hostname"] == hostname
assert mute["action"] == "Muted"
assert mute["message"] == message
assert mute["end"] == end
# We shouldn"t be able to mute a host that"s already muted, unless we include
# the override param.
end2 = end + 60 * 15
get_with_retry(
"Host",
hostname,
"""
Post a metric.
"""
# Format parameters
api._timeout = args.timeout
host = None if args.no_host else args.host
if args.tags:
tags = sorted(set([t.strip() for t in
args.tags.split(',') if t]))
else:
tags = None
# Submit metric
res = api.Metric.send(
metric=args.name, points=args.value, host=host,
device=args.device, tags=tags, metric_type=args.type)
# Report
res = defaultdict(list, res)
if args.localhostname:
# Warn about`--localhostname` command line flag deprecation
res['warnings'].append(
u"`--localhostname` command line flag is deprecated, made default when no `--host` "
u"is specified. See the `--host` option for more information."
)
report_warnings(res)
report_errors(res)
def post_metrics(auth_key, customer_name, hostname, pointlist, metric_name):
options = {
'api_key': auth_key['primary_api_key'],
'app_key': auth_key['primary_app_key']
}
initialize(**options)
CurrentPosixTime = time.time()
CurrentPosixTime10 = time.time() + 10
api.Metric.send(metric=metric_name, points=pointlist, host=hostname, tags=["customer:"+customer_name])
def send_metric(name, tags=[], error=False, success=False, count=1):
# Do not send metrics in debug mode
if env.DEBUG:
return
full_name = 'custom.{}'.format(name)
alert_type = 'error' if error else ('success' if success else 'info')
all_tags = TAGS + tags + ['alert_type:{}'.format(alert_type)]
try:
r = api.Metric.send(metric=full_name, points=count, tags=all_tags)
if r.get('status') and r.get('status') != 'ok':
logger.error("Not ok status while sending metric {} with tags [{}]".format(
name,
', '.join(tags)
))
except:
logger.error("An error occurred when sending metric {} with tags [{}]".format(
name,
', '.join(tags)
))