scripts/generators/markdown_fields.py (136 lines of code) (raw):
# Licensed to Elasticsearch B.V. under one or more contributor
# license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright
# ownership. Elasticsearch B.V. licenses this file to you under
# the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
from functools import wraps
import os.path as path
import os
import jinja2
from generators import ecs_helpers
from copy import deepcopy
def generate(nested, docs_only_nested, ecs_generated_version, semconv_version, otel_generator, out_dir):
ecs_helpers.make_dirs(out_dir)
if semconv_version.startswith('v'):
semconv_version = semconv_version[1:]
save_markdown(path.join(out_dir, 'index.md'), page_index(ecs_generated_version))
save_markdown(path.join(out_dir, 'ecs-otel-alignment-details.md'),
page_otel_alignment_details(nested, ecs_generated_version, semconv_version))
save_markdown(path.join(out_dir, 'ecs-otel-alignment-overview.md'),
page_otel_alignment_overview(otel_generator, nested, ecs_generated_version, semconv_version))
fieldsets = ecs_helpers.dict_sorted_by_keys(nested, ['group', 'name'])
for fieldset in fieldsets:
save_markdown(path.join(out_dir, f'ecs-{fieldset["name"]}.md'),
page_fieldset(fieldset, nested, ecs_generated_version))
# Helpers
def render_fieldset_reuse_text(fieldset):
"""Renders the expected nesting locations
if the the `reusable` object is present.
:param fieldset: The fieldset to evaluate
"""
if not fieldset.get('reusable'):
return None
reusable_fields = fieldset['reusable']['expected']
sorted_fields = sorted(reusable_fields, key=lambda k: k['full'])
return map(lambda f: f['full'], sorted_fields)
def render_nestings_reuse_section(fieldset):
"""Renders the reuse section entries.
:param fieldset: The target fieldset
"""
if not fieldset.get('reused_here'):
return None
rows = []
for reused_here_entry in fieldset['reused_here']:
rows.append({
'flat_nesting': "{}.*".format(reused_here_entry['full']),
'name': reused_here_entry['schema_name'],
'short': reused_here_entry['short'],
'beta': reused_here_entry.get('beta', ''),
'normalize': reused_here_entry.get('normalize')
})
return sorted(rows, key=lambda x: x['flat_nesting'])
def extract_allowed_values_key_names(field):
"""Extracts the `name` keys from the field's
allowed_values if present in the field
object.
:param field: The target field
"""
if not field.get('allowed_values'):
return []
return ecs_helpers.list_extract_keys(field['allowed_values'], 'name')
def sort_fields(fieldset):
"""Prepares a fieldset's fields for being
passed into the j2 template for rendering. This
includes sorting them into a list of objects and
adding a field for the names of any allowed values
for the field, if present.
:param fieldset: The target fieldset
"""
fields_list = list(fieldset['fields'].values())
for field in fields_list:
field['allowed_value_names'] = extract_allowed_values_key_names(field)
return sorted(fields_list, key=lambda field: field['name'])
def check_for_usage_doc(fieldset_name, usage_file_list=ecs_helpers.usage_doc_files()):
"""Checks if a usage doc exists for the specified
fieldset.
:param fieldset_name: The name of the target fieldset
"""
return f"ecs-{fieldset_name}-usage.md" in usage_file_list
def templated(template_name):
"""Decorator function to simplify rendering a template.
:param template_name: the name of the template to be rendered
"""
def decorator(func):
@wraps(func)
def decorated_function(*args, **kwargs):
ctx = func(*args, **kwargs)
if ctx is None:
ctx = {}
elif not isinstance(ctx, dict):
return ctx
return render_template(template_name, **ctx)
return decorated_function
return decorator
def render_template(template_name, **context):
"""Renders a template from the template folder with the given
context.
:param template_name: the name of the template to be rendered
:param context: the variables that should be available in the
context of the template.
"""
template = template_env.get_template(template_name)
return template.render(**context)
def save_markdown(f, text):
os.makedirs(path.dirname(f), exist_ok=True)
with open(f, "w") as outfile:
outfile.write(text)
# jinja2 setup
local_dir = path.dirname(path.abspath(__file__))
TEMPLATE_DIR = path.join(local_dir, '../templates')
template_loader = jinja2.FileSystemLoader(searchpath=TEMPLATE_DIR)
template_env = jinja2.Environment(loader=template_loader,
keep_trailing_newline=True,
trim_blocks=True,
lstrip_blocks=False)
# Rendering schemas
# Index
@templated('index.j2')
def page_index(ecs_generated_version):
return dict(ecs_generated_version=ecs_generated_version)
# Field Index
@templated('fieldset.j2')
def page_fieldset(fieldset, nested, ecs_generated_version):
sorted_reuse_fields = render_fieldset_reuse_text(fieldset)
render_nestings_reuse_fields = render_nestings_reuse_section(fieldset)
sorted_fields = sort_fields(fieldset)
usage_doc = check_for_usage_doc(fieldset.get('name'))
return dict(fieldset=fieldset,
sorted_reuse_fields=sorted_reuse_fields,
render_nestings_reuse_section=render_nestings_reuse_fields,
sorted_fields=sorted_fields,
usage_doc=usage_doc)
# Field Details Page
def page_field_details(nested, docs_only_nested):
if docs_only_nested:
for fieldset_name, fieldset in docs_only_nested.items():
nested[fieldset_name]['fields'].update(fieldset['fields'])
fieldsets = ecs_helpers.dict_sorted_by_keys(nested, ['group', 'name'])
results = (generate_field_details_page(fieldset) for fieldset in fieldsets)
return ''.join(results)
@templated('field_details.j2')
def generate_field_details_page(fieldset):
# render field reuse text section
sorted_reuse_fields = render_fieldset_reuse_text(fieldset)
render_nestings_reuse_fields = render_nestings_reuse_section(fieldset)
sorted_fields = sort_fields(fieldset)
usage_doc = check_for_usage_doc(fieldset.get('name'))
return dict(fieldset=fieldset,
sorted_reuse_fields=sorted_reuse_fields,
render_nestings_reuse_section=render_nestings_reuse_fields,
sorted_fields=sorted_fields,
usage_doc=usage_doc)
# OTel Fields Mapping Page
@templated('otel_alignment_details.j2')
def page_otel_alignment_details(nested, ecs_generated_version, semconv_version):
fieldsets = [deepcopy(fieldset) for fieldset in ecs_helpers.dict_sorted_by_keys(
nested, ['group', 'name']) if is_eligable_for_otel_mapping(fieldset)]
for fieldset in fieldsets:
sorted_fields = sort_fields(fieldset)
fieldset['fields'] = sorted_fields
return dict(fieldsets=fieldsets,
semconv_version=semconv_version,
ecs_generated_version=ecs_generated_version)
def is_eligable_for_otel_mapping(fieldset):
for field in fieldset['fields'].values():
if 'otel' in field:
return True
return False
# OTel Mapping Summary Page
@templated('otel_alignment_overview.j2')
def page_otel_alignment_overview(otel_generator, nested, ecs_generated_version, semconv_version):
fieldsets = ecs_helpers.dict_sorted_by_keys(nested, ['group', 'name'])
summaries = otel_generator.get_mapping_summaries(fieldsets)
return dict(summaries=summaries,
semconv_version=semconv_version,
ecs_generated_version=ecs_generated_version)
# Allowed values section
@templated('field_values.j2')
def page_field_values(nested, template_name='field_values_template.j2'):
category_fields = ['event.kind', 'event.category', 'event.type', 'event.outcome']
nested_fields = []
for cat_field in category_fields:
nested_fields.append(nested['event']['fields'][cat_field])
return dict(fields=nested_fields)