Getting "EndpointConnectionError: Could not connect to the endpoint URL"

[RajasGujarathi]

moto version - 5.0.13

I am running a Moto Server

@pytest.fixture(scope="module")
def moto_server():
    """Fixture to run a mocked AWS server for testing."""
    # Note: pass `port=0` to get a random free port.
    server = ThreadedMotoServer(port=0)
    server.start()
    host, port = server.get_host_and_port()
    yield f"http://{host}:{port}"
    server.stop()

and invoking a Lambda function,

lambdafn.invoke(
            FunctionName=<...>,
            InvocationType='RequestResponse',
            Payload=json.dumps(records,indent=4)
        )

The invoked Lambda tries to make a ECS service call.

However, I get [ERROR] EndpointConnectionError: Could not connect to the endpoint URL: "http://..... inside Lambda logs.

Any suggestions?

---

Related question:

Additionally, if we pass 'AWS_ENDPOINT_URL': moto_server as an environment variable to the Lambda function it does not pick the endpoint URL

[bblommers]

Hi @RajasGujarathi, the endpoint url would need to be set to http://host.docker.internal:{port} in that case.

The Lambda is executed within a Docker container, but the ThreadedMotoServer runs on the host system. host.docker.internal is a Docker-specific hostname that indicates that whatever server it's trying to reach, is located on the host.

Did you have to do anything special with the boto3-version? If I'm executing the Lambda in Python 3.11, the installed boto3-version is too old and doesn't support the AWS_ENDPOINT_URL-variable yet. (Installed 1.27.0, AWS_ENDPOINT_URL only available in 1.28.0)

[RajasGujarathi]

Thanks, this works.

[bblommers]

For reference, the MRE that shows it works:

(Note the system-call to update boto3 before executing the Lambda)

import io
import json
import zipfile
from uuid import uuid4

import boto3
import pytest

from moto.server import ThreadedMotoServer


@pytest.fixture(scope="module", name="port")
def moto_server():
    server = ThreadedMotoServer(port=0)
    server.start()
    host, port = server.get_host_and_port()
    yield port
    server.stop()


def test(port):
    moto_server = f"http://localhost:{port}"

    client = boto3.client("lambda", "us-east-1", endpoint_url=moto_server)
    function_name = str(uuid4())[0:6]
    client.create_function(
        FunctionName=function_name,
        Runtime="python3.11",
        Role=get_role_name(endpoint_url=moto_server),
        Handler="lambda_function.lambda_handler",
        Code={"ZipFile": get_zip_file()},
        Description="test lambda function",
        Timeout=20,
        MemorySize=128,
        Publish=True,
        Environment={"Variables": {"AWS_ENDPOINT_URL": f"http://host.docker.internal:{port}"}},
    )

    resp = client.invoke(FunctionName=function_name, Payload="{}", LogType="Tail")

    payload = resp["Payload"].read().decode("utf-8")
    assert json.loads(payload)["Images"] == []


def get_zip_file():
    func_str = f"""
import sys
from pip._internal import main

main(['install', '-I', '-q', 'boto3', '--target', '/tmp/', '--no-cache-dir', '--disable-pip-version-check'])
sys.path.insert(0,'/tmp/')

import boto3
import os

def lambda_handler(event, context):
    ec2 = boto3.client('ec2', region_name='us-east-1')

    return ec2.describe_images()
"""
    return _process_lambda(func_str)


def _process_lambda(func_str):
    zip_output = io.BytesIO()
    zip_file = zipfile.ZipFile(zip_output, "w", zipfile.ZIP_DEFLATED)
    zip_file.writestr("lambda_function.py", func_str)
    zip_file.close()
    zip_output.seek(0)
    return zip_output.read()


def get_role_name(endpoint_url):
    iam = boto3.client("iam", region_name="us-east-1", endpoint_url=endpoint_url)
    return iam.create_role(
                    RoleName="my-role",
                    AssumeRolePolicyDocument="some policy",
                    Path="/my-path/",
                )["Role"]["Arn"]