def sync_endpoints()

in lemur/sources/service.py [0:0]

• 87 lines of code
• 17 McCabe index (conditional complexity)

def sync_endpoints(source):
    new, updated, updated_by_hash = 0, 0, 0
    current_app.logger.debug(f"Retrieving endpoints from {source.label}")
    s = plugins.get(source.plugin_name)

    try:
        endpoints = s.get_endpoints(source.options)
    except NotImplementedError:
        current_app.logger.warning(
            "Unable to sync endpoints for source {} plugin has not implemented 'get_endpoints'".format(
                source.label
            )
        )
        return new, updated, updated_by_hash

    for endpoint in endpoints:
        try:
            exists = endpoint_service.get_by_dnsname_and_port(
                endpoint["dnsname"], endpoint["port"]
            )
        except OperationalError as e:
            # This is a workaround for handling sqlalchemy error "idle-in-transaction timeout", which is seen rarely
            # during the sync of sources with few thousands of resources. The DB interaction may need a rewrite to
            # avoid prolonged idle transactions.
            if e.connection_invalidated:
                # all the update, insert operations are committed individually. So this should be harmless/no-op
                database.rollback()
                # retry one more time
                exists = endpoint_service.get_by_dnsname_and_port(
                    endpoint["dnsname"], endpoint["port"]
                )
            else:
                raise e

        certificate_name = endpoint.pop("certificate_name")

        endpoint["certificate"] = certificate_service.get_by_name(certificate_name)

        # if get cert by name failed, we attempt a search via serial number and hash comparison
        # and link the endpoint certificate to Lemur certificate
        if not endpoint["certificate"]:
            certificate_attached_to_endpoint = None
            try:
                certificate_attached_to_endpoint = s.get_certificate_by_name(certificate_name, source.options)
            except NotImplementedError:
                current_app.logger.warning(
                    "Unable to describe server certificate for endpoints in source {}:"
                    " plugin has not implemented 'get_certificate_by_name'".format(
                        source.label
                    )
                )
                capture_exception()

            if certificate_attached_to_endpoint:
                lemur_matching_cert, updated_by_hash_tmp = find_cert(certificate_attached_to_endpoint)
                updated_by_hash += updated_by_hash_tmp

                if lemur_matching_cert:
                    endpoint["certificate"] = lemur_matching_cert[0]

                if len(lemur_matching_cert) > 1:
                    current_app.logger.error(
                        "Too Many Certificates Found{}. Name: {} Endpoint: {}".format(
                            len(lemur_matching_cert), certificate_name, endpoint["name"]
                        )
                    )
                    metrics.send("endpoint.certificate.conflict",
                                 "gauge", len(lemur_matching_cert),
                                 metric_tags={"cert": certificate_name, "endpoint": endpoint["name"],
                                              "acct": s.get_option("accountNumber", source.options)})

        if not endpoint["certificate"]:
            current_app.logger.error({
                "message": "Certificate Not Found",
                "certificate_name": certificate_name,
                "endpoint_name": endpoint["name"],
                "dns_name": endpoint.get("dnsname"),
                "account": s.get_option("accountNumber", source.options),
            })

            metrics.send("endpoint.certificate.not.found",
                         "counter", 1,
                         metric_tags={"cert": certificate_name, "endpoint": endpoint["name"],
                                      "acct": s.get_option("accountNumber", source.options),
                                      "dnsname": endpoint.get("dnsname")})
            continue

        policy = endpoint.pop("policy")

        policy_ciphers = []
        for nc in policy["ciphers"]:
            policy_ciphers.append(endpoint_service.get_or_create_cipher(name=nc))

        policy["ciphers"] = policy_ciphers
        endpoint["policy"] = endpoint_service.get_or_create_policy(**policy)
        endpoint["source"] = source

        if not exists:
            current_app.logger.debug(
                "Endpoint Created: Name: {name}".format(name=endpoint["name"])
            )
            endpoint_service.create(**endpoint)
            new += 1

        else:
            current_app.logger.debug(f"Endpoint Updated: {endpoint}")
            endpoint_service.update(exists.id, **endpoint)
            updated += 1

    return new, updated, updated_by_hash