-
Notifications
You must be signed in to change notification settings - Fork 1.4k
/
http.py
145 lines (114 loc) · 5.37 KB
/
http.py
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
# (C) Datadog, Inc. 2020-present
# All rights reserved
# Licensed under a 3-clause BSD style license (see LICENSE)
import os
import re
import click
from ...testing import process_checks_option
from ...utils import complete_valid_checks, get_check_files, get_default_config_spec
from ..console import CONTEXT_SETTINGS, abort, annotate_error, echo_failure, echo_info, echo_success, echo_warning
# Integrations that are not fully updated to http wrapper class but is owned partially by a different organization
EXCLUDED_INTEGRATIONS = {'kubelet', 'openstack'}
REQUEST_LIBRARY_FUNC_RE = r"requests.[get|post|head|put|patch|delete]*\("
HTTP_WRAPPER_INIT_CONFIG_RE = r"init_config\/[http|openmetrics_legacy|openmetrics]*"
HTTP_WRAPPER_INSTANCE_RE = r"instances\/[http|openmetrics_legacy|openmetrics]*"
def validate_config_http(file, check):
"""Determines if integration with http wrapper class
uses the http template in its spec.yaml file.
file -- filepath of file to validate
check -- name of the check that file belongs to
"""
if not os.path.exists(file):
return
has_failed = False
with open(file, 'r', encoding='utf-8') as f:
read_file = f.read()
has_init_config_http = re.search(HTTP_WRAPPER_INIT_CONFIG_RE, read_file)
has_instance_http = re.search(HTTP_WRAPPER_INSTANCE_RE, read_file)
if has_init_config_http and has_instance_http:
return
if not has_instance_http:
message = (
f"Detected {check} is missing `instances/http` or `instances/openmetrics_legacy` template in spec.yaml"
)
echo_failure(message)
annotate_error(file, message)
has_failed = True
if not has_init_config_http:
message = (
f"Detected {check} is missing `init_config/http` or `init_config/openmetrics_legacy` template in spec.yaml"
)
echo_failure(message)
annotate_error(file, message)
has_failed = True
return has_failed
def validate_use_http_wrapper_file(file, check):
"""Return true if the file uses the http wrapper class.
Also outputs every instance of deprecated request library function use
file -- filepath of file to validate
check -- name of the check
"""
file_uses_http_wrapper = False
has_failed = False
with open(file, 'r', encoding='utf-8') as f:
read_file = f.read()
found_match_arg = re.search(r"auth=|header=", read_file)
found_http = re.search(r"self.http|OpenMetricsBaseCheck", read_file)
skip_validation = re.search(r"SKIP_HTTP_VALIDATION", read_file)
if found_http and not skip_validation:
return found_http, has_failed, found_match_arg
http_func = re.search(REQUEST_LIBRARY_FUNC_RE, read_file)
if http_func:
echo_failure(
f'Check `{check}` uses `{http_func}` in `{os.path.basename(file)}`, '
f'please use the HTTP wrapper instead'
)
annotate_error(
file,
"Detected use of `{}`, please use the HTTP wrapper instead".format(http_func),
)
return False, True, None
return file_uses_http_wrapper, has_failed, None
def validate_use_http_wrapper(check):
"""Return true if the check uses the http wrapper class in any of its files.
If any of the check's files uses the request library, abort.
check -- name of the check
"""
has_failed = False
check_uses_http_wrapper = False
for file in get_check_files(check, include_tests=False):
if file.endswith('.py'):
file_uses_http_wrapper, file_uses_request_lib, has_arg_warning = validate_use_http_wrapper_file(file, check)
has_failed = has_failed or file_uses_request_lib
check_uses_http_wrapper = check_uses_http_wrapper or file_uses_http_wrapper
if check_uses_http_wrapper and has_arg_warning:
# Check for headers= or auth=
echo_warning(
f"{check}: \n"
f" The HTTP wrapper contains parameter `{has_arg_warning.group().replace('=', '')}`, "
f"this configuration is handled by the wrapper automatically.\n"
f" If this a genuine usage of the parameters, "
f"please inline comment `# SKIP_HTTP_VALIDATION`"
)
pass
if has_failed:
abort()
return check_uses_http_wrapper
@click.command(context_settings=CONTEXT_SETTINGS, short_help='Validate usage of http wrapper')
@click.argument('check', shell_complete=complete_valid_checks, required=False)
def http(check):
"""Validate all integrations for usage of http wrapper."""
has_failed = False
checks = process_checks_option(check, source='integrations')
echo_info(f"Validating {len(checks)} integrations for usage of http wrapper...")
for check in checks:
check_uses_http_wrapper = False
# Validate use of http wrapper (self.http.[...]) in check's .py files
if check not in EXCLUDED_INTEGRATIONS:
check_uses_http_wrapper = validate_use_http_wrapper(check)
# Validate use of http template in check's spec.yaml (if exists)
if check_uses_http_wrapper:
has_failed = validate_config_http(get_default_config_spec(check), check) or has_failed
if has_failed:
abort()
echo_success('Completed http validation!')