Function explain_response

test/e2e/common/utils.py:208–250

(
    client, service_name, input_path, network_layer: str
)

async def explain_response(
    client, service_name, input_path, network_layer: str
) -> Dict:
    kfs_client = KServeClient(
        config_file=os.environ.get("KUBECONFIG", "~/.kube/config")
    )
    isvc = kfs_client.get(
        service_name,
        namespace=KSERVE_TEST_NAMESPACE,
        version=constants.KSERVE_V1BETA1_VERSION,
    )
    scheme, cluster_ip, host, path = get_isvc_endpoint(isvc, network_layer)
    url = f"{scheme}://{cluster_ip}{path}"
    headers = {"Host": host}
    with open(input_path) as json_file:
        data = json.load(json_file)
        logger.info("Sending request data: %s", data)
        try:
            response = await client.explain(
                url, model_name=service_name, data=data, headers=headers
            )
        except (RuntimeError, orjson.JSONDecodeError) as e:
            logger.info("Explain error -------")
            pods = kfs_client.core_api.list_namespaced_pod(
                KSERVE_TEST_NAMESPACE,
                label_selector="serving.kserve.io/inferenceservice={}".format(
                    service_name
                ),
            )
            for pod in pods.items:
                logger.info(pod)
                logger.info(
                    "%s\t%s\t%s"
                    % (pod.metadata.name, pod.status.phase, pod.status.pod_ip)
                )
                api_response = kfs_client.core_api.read_namespaced_pod_log(
                    pod.metadata.name,
                    KSERVE_TEST_NAMESPACE,
                    container="kserve-container",
                )
                logger.info(api_response)
            raise e
        return response


def get_cluster_ip(namespace="istio-system", labels: dict = None):