Create .grimoirelab-sigils index from a .kibana

sigilsmigration.py

import json
import requests

# TARGET ES
ENV = 'localhost:9200'

# CREDENTIALS
USER = 'xxx'
PWD = 'xxx'

# SETTINGS
UPLOAD = True

# DO NOT MODIFY AFTER HERE
ELASTICSEARCH_URL = "https://{}:{}@{}"
KIBANA_INDEX = ".kibana"
SIGILS_INDEX = ".grimoirelab-sigils"

HEADERS = {"Content-Type": "application/json"}


def get_page_items(url, scroll_id=None):
    """Get the documents from the input url at a given page/scroll"""

    search_url = url + "/_search"
    if not scroll_id:
        params = {
            'size': 100,
            'scroll': '10m'
        }
        query = {
            "query": {
                "bool": {
                    "should": [
                        {
                            "term": {
                                "type": "index-pattern"
                            }
                        },
                        {
                            "term": {
                                "type": "dashboard"
                            }
                        }
                    ]
                }
            }
        }
        res = requests.post(search_url, params=params, data=json.dumps(query), headers=HEADERS, verify=False)
    else:
        search_url += "/scroll"
        query = {
            'scroll': '10m',
            'scroll_id': scroll_id
        }
        res = requests.post(search_url, data=json.dumps(query), headers=HEADERS, verify=False)

    res.raise_for_status()
    rjson = res.json()
    return rjson


def fetch(es_url, index):
    """Fetch the documents from a target index and returns a generator"""

    scroll_id = None
    index_url = es_url + '/' + index
    page = get_page_items(index_url, scroll_id)

    if not page:
        print("No results found")
        return []

    scroll_id = page["_scroll_id"]
    total = page['hits']['total']
    scroll_size = total['value'] if isinstance(total, dict) else total

    if scroll_size == 0:
        print("No results found")
        return

    while scroll_size > 0:
        for item in page['hits']['hits']:
            yield item

        page = get_page_items(es_url, scroll_id)

        if not page:
            break

        scroll_size = len(page['hits']['hits'])

    print("Fetch completed")


def prepare_sigils_items():
    es_url = ELASTICSEARCH_URL.format(USER, PWD, ENV)

    items_to_upload = []
    for obj in fetch(es_url, KIBANA_INDEX):
        source = obj['_source']
        # only dashboard and index-pattern objects can contain the release_date attr
        if source['type'] not in ['dashboard', 'index-pattern']:
            continue

        item_uuid = obj['_id']
        item_id = item_uuid.split(':')[1] if ':' in item_uuid else item_uuid
        item_type = source['type']
        release_date = source[item_type].get('release_date', None)
        item_title = source[item_type].get('title', '')

        if not release_date:
            print("Item %s (title: %s) doesn't contain release_date, it won't be inserted" % (item_uuid, item_title))
            continue

        item_json = {
            "item_uuid": item_uuid,
            "item_id": item_id,
            "item_type": item_type,
            "release_date": release_date
        }

        items_to_upload.append(item_json)

    return items_to_upload


def upload_sigils_index(items):
    es = ELASTICSEARCH_URL.format(USER, PWD, ENV)
    for item in items:
        item_uuid = item.pop('item_uuid')
        sigils_index_url = es + '/' + SIGILS_INDEX + '/doc/' + item_uuid
        res = requests.post(sigils_index_url, data=json.dumps(item), verify=False, headers=HEADERS)
        res.raise_for_status()
        print("Item %s added" % item_uuid)


def main():
    """The script allows to:
     - identify the dashboards and index patterns in the .kibana not including a release date
     - generate the .grimoirelab-sigils index containing the ID of index patterns/dashboards together with their release date.

    To execute the script you need to set ENV, USER, PWD and UPLOAD (the latter will create the .grimoirelab-sigils index)

    :return:
    """
    items = prepare_sigils_items()
    if UPLOAD:
        upload_sigils_index(items)


if __name__ == '__main__':
    main()