Eine AWS Lambda-Funktion lokal testen

Hintergrund

Wenn Sie mit Infrastructure as Code arbeiten, zum Beispiel mit Cloud Development Kit (CDK), können Sie AWS Lambda-Funktionen innerhalb Ihrer CDK-Anwendung erstellen. Mit CDK ist es einfach, dieses Lambda zu verpacken und es von CDK in Ihrer AWS-Umgebung bereitstellen zu lassen. Aber oft ertappe ich mich dabei, dass ich AWS Lambda-Code in meinem Code-Editor schreibe und Schwierigkeiten habe zu testen, ob die Funktion das tut, was sie tun soll.

Wir alle haben schon einmal gesagt: "Das funktioniert auf meinem Rechner". Aber mein Rechner ist nicht die AWS-Landschaft. Normalerweise würden Sie also die Lambda-Funktion erstellen, sie über CDK bereitstellen, die Lambda-Funktion mit einem Test aufrufen, nur um die Ergebnisse zu sehen und festzustellen, dass die Funktion nicht wie erwartet funktioniert.

Paket python-lambda-local

Als Lösung verwende ich das pip-Paket "python-lambda-local". Mit diesem Paket können Sie Ihre Lambda-Python-Funktion lokal testen, so wie sie in AWS laufen würde. Sie können also eine Funktion namens lambda_handler mit Ereignis und Kontext als Eingabe erstellen.

Voraussetzung

• AWS-Zugang über Zugriffsschlüssel und geheimen Schlüssel, da dieses Beispiel einen boto3-Aufruf an AWS verwendet.

Installation

Zunächst müssen wir das Paket über pip installieren:

➜ ~ pip install python-lambda-local

Ausführen des Pakets:

➜ ~ python-lambda-local -f lambda_handler lambda_function.py event.json

wo:
lambda_handler ist der Name Ihrer Handler-Funktion
lambda_function.py ist der Name Ihrer Datei mit Python-Code event.json sind die Daten des Testevents

Grundsätzlich können Sie die Funktion über die Befehlszeile Ihres Rechners aufrufen, zum Beispiel mit einem Testevent.

Szenario der realen Welt

Also habe ich den Dienst namens Managed Workflow Apache Airflow (MWAA) erkundet. Dies ist ein verwalteter Orchestrierungsdienst für Apache Airflow, der die Einrichtung und den Betrieb von End-to-End-Datenpipelines erleichtert.

Eine der besten Methoden ist es, MWAA in einem privaten Netzwerkmodus auszuführen, also in einem privaten Subnetz innerhalb Ihrer VPC. Das Problem bei der Ausführung von MWAA im privaten Modus ist, dass Sie bei der Bereitstellung von MWAA keine Verbindung zur Apache Airflow UI herstellen können.

Als Lösung beschreibt AWS die Verwendung eines Load Balancers vor Ihrem MWAA, siehe diese Dokumentation. Um diesen Load Balancer nutzen zu können, benötigen wir die IP-Adressen der von MWAA erstellten VPC-Endpunkte innerhalb Ihrer VPC. Dies ist die perfekte Grundlage für einen Lambda, der später als benutzerdefinierte Ressource innerhalb Ihrer CDK-Anwendung verwendet werden kann. Konzentrieren wir uns zunächst auf die Lambda-Funktion, die die privaten IP-Adressen meines VPC-Endpunkts für MWAA abruft.

Bauen gehen

Beginnen wir mit einer Datei namens get_mwaa_ips.py.. Ziel ist es, die IP-Adressen der MWAA VPC-Endpunkte abzurufen, damit wir sie als Ziele mit unserem Load Balancer verwenden können. Für diese Demo rufen wir zunächst den Namen der MWAA-Umgebung ab, wie im ersten Schritt in der AWS-Dokumentation beschrieben.

Die Python-Datei wird boto3 verwenden, um den MWAA list_environments API-Aufruf zu nutzen. Außerdem habe ich der Python-Datei eine Standardprotokollierung hinzugefügt.

importieren boto3  
importieren os  
Protokollierung importieren  

logger = logging.getLogger("__name__")  
logger.setLevel(os.environ.get("LOG_LEVEL", logging.INFO))  
from botocore.exceptions import ClientError  

mwaa = boto3.client('mwaa')

Der zweite Teil besteht darin, die Funktion zu erstellen, die die MWAA-Umgebung auflistet. Die Funktion def list_mwaa_environments gibt eine Liste von Umgebungen zurück. Siehe auch boto3-Dokumentation.

# mwaa-Umgebungen über boto3 auflisten  
def list_mwaa_environments():
name = mwaa.list_environments()  
Name zurückgeben

Zuletzt folgt der eigentliche Lambda-Handler, der Ereignis und Kontext als Eingabe verwendet. Da wir von der Funktion list_mwaa_environments() eine Liste abrufen, müssen wir, da Sie potenziell mehrere MWAA-Umgebungen haben könnten, eine Schleife darüber laufen lassen, um den Namen der MWAA-Umgebung abzurufen.

def lambda_handler(event, context): 
logger.info(event) 

try: 
for mwaa_environment in list_mwaa_environments().get('Environments'): 
return mwaa_environment 
except ClientError as e: 
logger.error(e.response['Error']['Message'])

Den Link zum vollständigen Codebeispiel finden Sie am Ende dieses Artikels.

Jetzt brauchen wir ein Test-Ereignis, wie wir es im AWS Lambda-Service haben würden. Erstellen Sie also eine Datei namens event.json im selben Verzeichnis und fügen Sie den folgenden Inhalt hinzu.

{} Jetzt ist es tatsächlich an der Zeit, die Lambda-Funktion so zu testen, wie sie in AWS Lambda laufen würde. Führen Sie auf Ihrem Terminal den Befehl zum Testen Ihres lambda_handlers in der Datei get_mwaa_ips.py mit der event.json aus.

Da ich hier einen boto3-Aufruf verwende, der tatsächlich gegen die API von MWAA prüft, muss ich Zugriff auf AWS haben (aws_access_key_id und aws_secret_access_key) und meine Rolle, die mit meinem Benutzer verknüpft ist, um die Berechtigung zu haben, Airflow-Umgebungen aufzulisten: airflow:ListEnvironments

Dies führt zu folgendem Ergebnis:

➜ testing_lambda_locally python-lambda-local -f lambda_handler get_mwaa_ips.py event.json  
[root - INFO - 2021-08-19 21:09:08,704] Ereignis: {}  
[root - INFO - 2021-08-19 21:09:08,704] START RequestId: 20581f30-12de-4a8f-94b1-79b089eeefc8 Version:  
[botocore.credentials - INFO - 2021-08-19 21:09:09,207] Anmeldedaten in der gemeinsamen Anmeldedaten-Datei gefunden: ~/.aws/credentials  [__name__ - INFO - 2021-08-19 21:09:09,235]  {}  
[root - INFO - 2021-08-19 21:09:09,451] END RequestId: 20581f30-12de-4a8f-94b1-79b089eeefc8  
[root - INFO - 2021-08-19 21:09:09,452] REPORT RequestId: 20581f30-12de-4a8f-94b1-79b089eeefc8 Duration: 162.46 ms  
[root - INFO - 2021-08-19 21:09:09,452] ERGEBNIS:  
Orchestrierung

Versuchen Sie es selbst

Den Code finden Sie in meinem GitHub

Yvos Blog