def clone_ilm_policy()

in src/es_pii_tool/helpers/steps.py [0:0]

• 52 lines of code
• 13 McCabe index (conditional complexity)

def clone_ilm_policy(task: 'Task', stepname, var: DotMap, **kwargs) -> None:
    """
    If this index has an ILM policy, we need to clone it so we can attach
    the new index to it.
    """
    missing_data(stepname, kwargs)
    data = kwargs['data']
    step = Step(task=task, stepname=stepname)
    if step.finished():
        logger.info('%s: already completed', step.stub)
        return
    step.begin()
    if data.index.lifecycle.name is None or not data.ilm.lifecycle.policy:
        _ = f'{stepname}: Index {var.index} has no ILM lifecycle or policy data'
        logger.debug(_)
        step.add_log(_)
        return
    data.new = DotMap()

    # From here, we check for matching named cloned policy

    configure_ilm_policy(task, data)

    # New ILM policy naming: pii-tool-POLICYNAME---v###
    stub = f'pii-tool-{strip_ilm_name(data.index.lifecycle.name)}'
    policy = data.new.ilmpolicy.toDict()  # For comparison
    resp = {'dummy': 'startval'}  # So the while loop can start with something
    policyver = 0  # Our version number starting point.
    policymatch = False
    while resp:
        data.new.ilmname = f'{stub}---v{policyver + 1:03}'
        resp = api.get_ilm_lifecycle(var.client, data.new.ilmname)  # type: ignore
        if resp:  # We have data, so the name matches
            # Compare the new policy to the one just returned
            if policy == resp[data.new.ilmname]['policy']:  # type: ignore
                msg = f'New policy data matches: {data.new.ilmname}'
                logger.debug(msg)
                step.add_log(msg)
                policymatch = True
                break  # We can drop out of the loop here.
        # Implied else: resp has no value, so the while loop will end.
        policyver += 1
    msg = f'New ILM policy name (may already exist): {data.new.ilmname}'
    logger.debug(msg)
    step.add_log(msg)
    if not task.job.dry_run:  # Don't create if dry_run
        if not policymatch:
            # Create the cloned ILM policy
            try:
                gkw = {'name': data.new.ilmname, 'policy': policy}
                api.generic_get(var.client.ilm.put_lifecycle, **gkw)
            except (MissingError, BadClientResult) as exc:
                _ = f'Unable to put new ILM policy: {exc}'
                logger.error(_)
                step.add_log(_)
                failed_step(task, step, exc)
        # Implied else: We've arrived at the expected new ILM name
        # and it does match an existing policy in name and content
        # so we don't need to create a new one.
    else:
        _ = (
            f'Dry-Run: No changes, but expected behavior: '
            f'ILM policy {data.new.ilmname} created'
        )
        logger.debug(_)
        step.add_log(_)
    step.end(completed=True, errors=False, logmsg=f'{stepname} completed')