(check, check_duplicates, show_warnings)
Validates metadata.csv files

    If `check` is specified, only the check will be validated, if check value is 'changed' will only apply to changed
    checks, an 'all' or empty `check` value will validate all README files.
Validates metadata.csv files
Validates metadata . csv files
def metadata(check, check_duplicates, show_warnings):
    """Validates metadata.csv files

    If `check` is specified, only the check will be validated, if check value is 'changed' will only apply to changed
    checks, an 'all' or empty `check` value will validate all README files.
    """
    checks = process_checks_option(check, source='metrics', validate=True)
    echo_info(f"Validating metadata for {len(checks)} checks ...")

    # If a check is specified, abort if it doesn't have a metadata file
    if check not in ('all', 'changed') and not checks:

        # only abort if we have an integration and require a manifest file
        manifest = Manifest.load_manifest(check)
        if manifest.has_integration():
            abort(f'Metadata file for {check} not found.')

    errors = False

    for current_check in checks:
        display_queue = []
        if current_check.startswith('datadog_checks_'):
            continue

        # get any manifest info needed for validation - and skip if no integration included in manifest
        manifest = Manifest.load_manifest(current_check)
        if not manifest.has_integration():
            echo_success(f"Skipping {check} - metadata not required since this check doesn't contain an integration.")
            continue

        try:
            metric_prefix = manifest.get_metric_prefix().rstrip('.')
        except KeyError:
            metric_prefix = None

        display_name = manifest.get_display_name()

        metadata_file = get_metadata_file(current_check)
        display_queue.append((echo_debug, f"Checking {metadata_file}"))

        # To make logging less verbose, common errors are counted for current check
        metric_prefix_count = defaultdict(int)
        empty_count = defaultdict(int)
        empty_warning_count = defaultdict(int)
        duplicate_name_set = set()
        duplicate_short_name_set = set()
        duplicate_description_set = set()

        metric_prefix_error_shown = False
        if os.stat(metadata_file).st_size == 0:
            errors = True
            display_queue.append(
                (echo_failure, f"{current_check} metadata file is empty. This file needs the header row at minimum")
            )

        for line, row in read_metadata_rows(metadata_file):
            # determine if number of columns is complete by checking for None values (DictReader populates missing columns with None https://docs.python.org/3.8/library/csv.html#csv.DictReader) # noqa
            if None in row.values():
                errors = True
                display_queue.append(
                    (echo_failure, f"{current_check}:{line} {row['metric_name']} Has the wrong amount of columns")
                )
                continue

            # all headers exist, no invalid headers
            all_keys = set(row)
            if all_keys != ALL_HEADERS:
                invalid_headers = all_keys.difference(ALL_HEADERS)
                if invalid_headers:
                    errors = True
                    display_queue.append((echo_failure, f'{current_check}:{line} Invalid column {invalid_headers}'))

                missing_headers = ALL_HEADERS.difference(all_keys)
                if missing_headers:
                    errors = True
                    display_queue.append(echo_failure(f'{current_check}:{line} Missing columns {missing_headers}'))
                continue

            errors = errors or check_duplicate_values(
                current_check, line, row, 'metric_name', duplicate_name_set, fail=True
            )

            if check_duplicates:
                check_duplicate_values(current_check, line, row, 'short_name', duplicate_short_name_set)
                check_duplicate_values(current_check, line, row, 'description', duplicate_description_set)

            normalized_metric_name = normalize_metric_name(row['metric_name'])
            if row['metric_name'] != normalized_metric_name:
                errors = True
                display_queue.append(
                    (
                        echo_failure,
                        f"{current_check}:{line} Metric name '{row['metric_name']}' is not valid, "
                        f"it should be normalized as {normalized_metric_name}",
                    )
                )

            # metric_name header
            if metric_prefix:
                prefix = row['metric_name'].split('.')[0]
                if prefix not in ALLOWED_PREFIXES:
                    if not row['metric_name'].startswith(metric_prefix):
                        metric_prefix_count[prefix] += 1
            else:
                errors = True
                if not metric_prefix_error_shown and current_check not in PROVIDER_INTEGRATIONS:
                    metric_prefix_error_shown = True
                    display_queue.append(
                        (echo_failure, f'{current_check}:{line} metric_prefix does not exist in manifest')
                    )

            # metric_type header
            if row['metric_type'] and row['metric_type'] not in VALID_METRIC_TYPE:
                errors = True
                display_queue.append(
                    (echo_failure, f"{current_check}:{line} `{row['metric_type']}` is an invalid metric_type.")
                )

            # unit_name header
            if row['unit_name'] and row['unit_name'] not in VALID_UNIT_NAMES:
                errors = True
                display_queue.append(
                    (echo_failure, f"{current_check}:{line} `{row['unit_name']}` is an invalid unit_name.")
                )

            # per_unit_name header
            if row['per_unit_name'] and row['per_unit_name'] not in VALID_UNIT_NAMES:
                errors = True
                display_queue.append(
                    (echo_failure, f"{current_check}:{line} `{row['per_unit_name']}` is an invalid per_unit_name.")
                )

            # Check if unit/per_unit is valid
            if row['unit_name'] in VALID_TIME_UNIT_NAMES and row['per_unit_name'] in VALID_TIME_UNIT_NAMES:
                errors = True
                display_queue.append(
                    (
                        echo_failure,
                        f"{current_check}:{line} `{row['unit_name']}/{row['per_unit_name']}` unit is invalid, "
                        f"use the fraction unit instead. If `{row['unit_name']}` and `{row['per_unit_name']}` "
                        "are not the same unit, eg ms/s, note that in the description.",
                    )
                )

            # integration header
            integration = row['integration']
            normalized_integration = normalize_display_name(display_name)
            if integration != normalized_integration and normalized_integration not in EXCLUDE_INTEGRATIONS:
                errors = True
                display_queue.append(
                    (
                        echo_failure,
                        f"{current_check}:{line} integration: `{row['integration']}` should be: "
                        f"{normalized_integration}",
                    )
                )

            # orientation header
            if row['orientation'] and row['orientation'] not in VALID_ORIENTATION:
                errors = True
                display_queue.append(
                    (echo_failure, f"{current_check}:{line} `{row['orientation']}` is an invalid orientation.")
                )

            # empty required fields
            for header in REQUIRED_HEADERS:
                if not row[header]:
                    empty_count[header] += 1

            # empty description field, description is recommended
            if not row['description']:
                empty_warning_count['description'] += 1

            elif "|" in row['description']:
                errors = True
                display_queue.append((echo_failure, f"{current_check}:{line} `{row['metric_name']}` contains a `|`."))

            # check if there is unicode
            elif any(not content.isascii() for _, content in row.items()):
                errors = True
                display_queue.append(
                    (echo_failure, f"{current_check}:{line} `{row['metric_name']}` contains unicode characters.")
                )

            # exceeds max allowed length of description
            elif len(row['description']) > MAX_DESCRIPTION_LENGTH:
                errors = True
                display_queue.append(
                    (
                        echo_failure,
                        f"{current_check}:{line} `{row['metric_name']}` exceeds the max length: "
                        f"{MAX_DESCRIPTION_LENGTH} for descriptions.",
                    )
                )
            if row['interval'] and not row['interval'].isdigit():
                errors = True
                display_queue.append(
                    (echo_failure, f"{current_check}:{line} interval should be an int, found '{row['interval']}'.")
                )

        for header, count in empty_count.items():
            errors = True
            display_queue.append((echo_failure, f'{current_check}: {header} is empty in {count} rows.'))

        for prefix, count in metric_prefix_count.items():
            display_queue.append(
                (
                    echo_failure,
                    f"{current_check}: `{prefix}` appears {count} time(s) and does not match metric_prefix "
                    "defined in the manifest.",
                )
            )

        if show_warnings:
            for header, count in empty_warning_count.items():
                echo_warning(f'{current_check}: {header} is empty in {count} rows.')

        if display_queue:
            annotate_display_queue(metadata_file, display_queue)
            for func, message in display_queue:
                func(message)
    if errors:
        abort()

    echo_success('Validated!')
def metadata ( check , check_duplicates , show_warnings ) : checks = process_checks_option ( check , source = 'metrics' , validate = True ) echo_info ( f"Validating metadata for {len(checks)} checks ..." ) # If a check is specified, abort if it doesn't have a metadata file if check not in ( 'all' , 'changed' ) and not checks : # only abort if we have an integration and require a manifest file manifest = Manifest . load_manifest ( check ) if manifest . has_integration ( ) : abort ( f'Metadata file for {check} not found.' ) errors = False for current_check in checks : display_queue = [ ] if current_check . startswith ( 'datadog_checks_' ) : continue # get any manifest info needed for validation - and skip if no integration included in manifest manifest = Manifest . load_manifest ( current_check ) if not manifest . has_integration ( ) : echo_success ( f"Skipping {check} - metadata not required since this check doesn't contain an integration." ) continue try : metric_prefix = manifest . get_metric_prefix ( ) . rstrip ( '.' ) except KeyError : metric_prefix = None display_name = manifest . get_display_name ( ) metadata_file = get_metadata_file ( current_check ) display_queue . append ( ( echo_debug , f"Checking {metadata_file}" ) ) # To make logging less verbose, common errors are counted for current check metric_prefix_count = defaultdict ( int ) empty_count = defaultdict ( int ) empty_warning_count = defaultdict ( int ) duplicate_name_set = set ( ) duplicate_short_name_set = set ( ) duplicate_description_set = set ( ) metric_prefix_error_shown = False if os . stat ( metadata_file ) . st_size == 0 : errors = True display_queue . append ( ( echo_failure , f"{current_check} metadata file is empty. This file needs the header row at minimum" ) ) for line , row in read_metadata_rows ( metadata_file ) : # determine if number of columns is complete by checking for None values (DictReader populates missing columns with None https://docs.python.org/3.8/library/csv.html#csv.DictReader) # noqa if None in row . values ( ) : errors = True display_queue . append ( ( echo_failure , f"{current_check}:{line} {row['metric_name']} Has the wrong amount of columns" ) ) continue # all headers exist, no invalid headers all_keys = set ( row ) if all_keys != ALL_HEADERS : invalid_headers = all_keys . difference ( ALL_HEADERS ) if invalid_headers : errors = True display_queue . append ( ( echo_failure , f'{current_check}:{line} Invalid column {invalid_headers}' ) ) missing_headers = ALL_HEADERS . difference ( all_keys ) if missing_headers : errors = True display_queue . append ( echo_failure ( f'{current_check}:{line} Missing columns {missing_headers}' ) ) continue errors = errors or check_duplicate_values ( current_check , line , row , 'metric_name' , duplicate_name_set , fail = True ) if check_duplicates : check_duplicate_values ( current_check , line , row , 'short_name' , duplicate_short_name_set ) check_duplicate_values ( current_check , line , row , 'description' , duplicate_description_set ) normalized_metric_name = normalize_metric_name ( row [ 'metric_name' ] ) if row [ 'metric_name' ] != normalized_metric_name : errors = True display_queue . append ( ( echo_failure , f"{current_check}:{line} Metric name '{row['metric_name']}' is not valid, " f"it should be normalized as {normalized_metric_name}" , ) ) # metric_name header if metric_prefix : prefix = row [ 'metric_name' ] . split ( '.' ) [ 0 ] if prefix not in ALLOWED_PREFIXES : if not row [ 'metric_name' ] . startswith ( metric_prefix ) : metric_prefix_count [ prefix ] += 1 else : errors = True if not metric_prefix_error_shown and current_check not in PROVIDER_INTEGRATIONS : metric_prefix_error_shown = True display_queue . append ( ( echo_failure , f'{current_check}:{line} metric_prefix does not exist in manifest' ) ) # metric_type header if row [ 'metric_type' ] and row [ 'metric_type' ] not in VALID_METRIC_TYPE : errors = True display_queue . append ( ( echo_failure , f"{current_check}:{line} `{row['metric_type']}` is an invalid metric_type." ) ) # unit_name header if row [ 'unit_name' ] and row [ 'unit_name' ] not in VALID_UNIT_NAMES : errors = True display_queue . append ( ( echo_failure , f"{current_check}:{line} `{row['unit_name']}` is an invalid unit_name." ) ) # per_unit_name header if row [ 'per_unit_name' ] and row [ 'per_unit_name' ] not in VALID_UNIT_NAMES : errors = True display_queue . append ( ( echo_failure , f"{current_check}:{line} `{row['per_unit_name']}` is an invalid per_unit_name." ) ) # Check if unit/per_unit is valid if row [ 'unit_name' ] in VALID_TIME_UNIT_NAMES and row [ 'per_unit_name' ] in VALID_TIME_UNIT_NAMES : errors = True display_queue . append ( ( echo_failure , f"{current_check}:{line} `{row['unit_name']}/{row['per_unit_name']}` unit is invalid, " f"use the fraction unit instead. If `{row['unit_name']}` and `{row['per_unit_name']}` " "are not the same unit, eg ms/s, note that in the description." , ) ) # integration header integration = row [ 'integration' ] normalized_integration = normalize_display_name ( display_name ) if integration != normalized_integration and normalized_integration not in EXCLUDE_INTEGRATIONS : errors = True display_queue . append ( ( echo_failure , f"{current_check}:{line} integration: `{row['integration']}` should be: " f"{normalized_integration}" , ) ) # orientation header if row [ 'orientation' ] and row [ 'orientation' ] not in VALID_ORIENTATION : errors = True display_queue . append ( ( echo_failure , f"{current_check}:{line} `{row['orientation']}` is an invalid orientation." ) ) # empty required fields for header in REQUIRED_HEADERS : if not row [ header ] : empty_count [ header ] += 1 # empty description field, description is recommended if not row [ 'description' ] : empty_warning_count [ 'description' ] += 1 elif "|" in row [ 'description' ] : errors = True display_queue . append ( ( echo_failure , f"{current_check}:{line} `{row['metric_name']}` contains a `|`." ) ) # check if there is unicode elif any ( not content . isascii ( ) for _ , content in row . items ( ) ) : errors = True display_queue . append ( ( echo_failure , f"{current_check}:{line} `{row['metric_name']}` contains unicode characters." ) ) # exceeds max allowed length of description elif len ( row [ 'description' ] ) > MAX_DESCRIPTION_LENGTH : errors = True display_queue . append ( ( echo_failure , f"{current_check}:{line} `{row['metric_name']}` exceeds the max length: " f"{MAX_DESCRIPTION_LENGTH} for descriptions." , ) ) if row [ 'interval' ] and not row [ 'interval' ] . isdigit ( ) : errors = True display_queue . append ( ( echo_failure , f"{current_check}:{line} interval should be an int, found '{row['interval']}'." ) ) for header , count in empty_count . items ( ) : errors = True display_queue . append ( ( echo_failure , f'{current_check}: {header} is empty in {count} rows.' ) ) for prefix , count in metric_prefix_count . items ( ) : display_queue . append ( ( echo_failure , f"{current_check}: `{prefix}` appears {count} time(s) and does not match metric_prefix " "defined in the manifest." , ) ) if show_warnings : for header , count in empty_warning_count . items ( ) : echo_warning ( f'{current_check}: {header} is empty in {count} rows.' ) if display_queue : annotate_display_queue ( metadata_file , display_queue ) for func , message in display_queue : func ( message ) if errors : abort ( ) echo_success ( 'Validated!' )