def get_read_only_indices()

in source/lambda/es_loader/siem/utils.py [0:0]

• 29 lines of code
• 12 McCabe index (conditional complexity)

def get_read_only_indices(es_conn, awsauth, ES_HOSTNAME):
    read_only_indices = []
    # cold tier
    # GET _cold/indices/_search?page_size=100
    url = f'https://{ES_HOSTNAME}/_cold/indices/_search'
    headers = {'Content-Type': 'application/json'}
    try:
        res = requests.get(
            url, params={'page_size': 1}, auth=awsauth, timeout=3.0)
    except requests.exceptions.Timeout:
        logger.warning('timeout: impossible to get cold index')
        return tuple(read_only_indices)
    while res.status_code == 200 and len(res.json()['indices']) > 0:
        for obj in res.json()['indices']:
            idx = obj['index']
            if idx.startswith('log-'):
                read_only_indices.append(idx)
        pagination_id = res.json()['pagination_id']
        body = f'{{"pagination_id": "{pagination_id}"}}'
        try:
            res = requests.post(
                url, data=body, auth=awsauth, headers=headers, timeout=3.0)
        except requests.exceptions.Timeout:
            logger.warning('timeout: impossible to get all cold index')
            break

    # close index
    # params = {'index': 'log-*', 'h': 'index,status'}
    # indices = es_conn.cat.indices(params=params)

    # close index and ultrawarm tier
    indices = es_conn.cluster.state(metric='blocks')
    if ('blocks' in indices) and ('indices' in indices['blocks']):
        for idx in indices['blocks']['indices']:
            if idx.startswith('log-'):
                read_only_indices.append(idx)
    return tuple(sorted(list(set(read_only_indices))))