datadog_checks.base.utils.common.pattern_filter(items, whitelist=None, blacklist=None, key=None)[source]

This filters items by a regular expression whitelist and/or blacklist, with the blacklist taking precedence. An optional key function can be provided that will be passed each item.

datadog_checks.base.utils.common.round_value(value, precision=0, rounding_method='ROUND_HALF_UP')[source]



Freezes any mutable object including dictionaries and lists for hashing. Accepts nested dictionaries.



datadog_checks.base.utils.headers.headers(agentConfig, **kwargs)[source]


class datadog_checks.base.utils.limiter.Limiter(check_name, object_name, object_limit, warning_func=None)[source]

Bases: object

Limiter implements a simple cut-off capping logic for object count. It is used by the AgentCheck class to limit the number of sets of tags that can be set by an instance.


Returns the internal state of the limiter for unit tests


is_reached is to be called for every object that counts towards the limit. - When called with no uid, the Limiter assumes this is a new object and unconditionally increments the counter (less CPU and memory usage). - When a given object can be passed multiple times, a uid must be provided to deduplicate calls. Only the first occurrence of a uid will increment the counter.

Parameters:uid – (optional) unique identifier of the object, to deduplicate calls
Returns:boolean, true if limit exceeded

Resets state and uid set. To be called asap to free memory


class datadog_checks.base.utils.platform.Platform[source]

Bases: object

Return information about the given platform.

static is_bsd(name=None)[source]

Return true if this is a BSD like operating system.

static is_containerized()[source]
static is_darwin(name=None)[source]
static is_ecs_instance()[source]
static is_freebsd(name=None)[source]
static is_k8s()[source]
static is_linux(name=None)[source]
static is_mac(name=None)[source]
static is_solaris(name=None)[source]
static is_unix(name=None)[source]

Return true if the platform is a unix, False otherwise.

static is_win32(name=None)[source]
static is_windows(name=None)[source]
static python_architecture()[source]

Human-friendly OS name


datadog_checks.base.utils.proxy.config_proxy_skip(proxies, uri, skip_proxy=False)[source]

Returns an amended copy of the proxies dictionary - used by requests, it will disable the proxy if the uri provided is to be reached directly. :param proxies A dict with existing proxies: https, http, and no are potential keys. :param uri URI to determine if a proxy is necessary or not. :param skip_proxy If True, the returned proxy dictionary will disable all proxies.


datadog_checks.base.utils.subprocess_output.get_subprocess_output(command, log, raise_on_empty_output=True, log_debug=True)[source]

Run the given subprocess command and return its output. Raise an Exception if an error occurs.

  • command (list(str) or str) – The command to run. Using a list of strings is recommended. The command will be run in a subprocess without using a shell, as such shell features like shell pipes, wildcard expansion, environment variable expansion, etc., are not supported.
  • log (logging.Logger) – The log object to use
  • raise_on_empty_output (bool) – Whether to raise a SubprocessOutputEmptyError exception when the subprocess doesn’t output anything to its stdout.
  • log_debug (bool) – Whether to enable debug logging of full command.

The stdout contents, stderr contents and status code of the command

Return type:

tuple(str, str, int)


class datadog_checks.base.utils.tailfile.TailFile(logger, path, callback)[source]

Bases: object

tail(line_by_line=True, move_end=True)[source]

Read line-by-line and run callback on each line. line_by_line: yield each time a callback has returned True move_end: start from the last line of the log


class datadog_checks.base.utils.timeout.ThreadMethod(target, args, kwargs)[source]

Bases: threading.Thread

Descendant of Thread class. Run the specified target method with the specified arguments. Store result and exceptions. From: https://code.activestate.com/recipes/440569/


Method representing the thread’s activity.

You may override this method in a subclass. The standard run() method invokes the callable object passed to the object’s constructor as the target argument, if any, with sequential and keyword arguments taken from the args and kwargs arguments, respectively.

exception datadog_checks.base.utils.timeout.TimeoutException[source]

Bases: exceptions.Exception

Raised when a function runtime exceeds the limit set.


A decorator to timeout a function. Decorated method calls are executed in a separate new thread with a specified timeout. Also check if a thread for the same function already exists before creating a new one. Note: Compatible with Windows (thread based).