Hogyan kezeljük a tesztesetek futtatását Airflow segítségével?

A szoftverfejlesztés világában a tesztelés kritikus szerepet játszik. Az egyik legfontosabb eszköz a tesztkészletek automatizált futtatása és a tesztek eredményeinek megfelelő kezelése. A modern rendszerekben, például az Airflow-ban, lehetőség van komplex tesztesetek meghatározására és kezelésére, amelyeket párhuzamosan lehet futtatni egyetlen konfigurációval. Az Airflow egy népszerű eszköz, amely segít a feladatok ütemezésében és a rendszerek tesztelésében. Ez a rendszer lehetőséget ad arra, hogy a tesztelés során különféle környezeteket és környezeti feltételeket hozzunk létre, kezeljünk és végezzünk el.

A tesztkészletek kezelésének első lépése az alapvető konfigurációk meghatározása. A tesztkészletek leírását JSON formátumban tárolhatjuk, ahol minden teszteset egy névvel és értékkel rendelkezik, például:

json
Copy code
[ 

  {"name": "test_1", "value": 1}, 
  {"name": "test_2", "value": 3}, 
  {"name": "test_3", "value": 4}
]

Ez lehetővé teszi a különböző tesztek kezelését, és megkönnyíti a rendszerek közötti adatcserét. A JSON formátumban tárolt adatokat az Airflow DAG-jaiban (Directed Acyclic Graph) használjuk fel a tesztkészletek futtatásához.

A DAG meghatározása során fontos figyelembe venni, hogy minden tesztkészletet egy külön környezetben kell futtatni. Az Airflow 2.7-es verziójától kezdődően egyre több kényelmi funkció érhető el, amelyek segítenek az ilyen típusú feladatok könnyebb kezelésében. Az alapértelmezett folyamat a következő: először létrehozzuk a környezetet, majd párhuzamosan futtatjuk a teszteket, és ha a tesztek sikeresek, értesítjük a webszolgáltatást. A végén a környezetet is eltávolítjuk.

A feladatok független futtatása és az egymástól való izoláltság megkönnyíti a DAG definiálását. A tesztkészletek konfigurációit a következő módon adjuk meg egy Python függvényben:

python
Copy code
import datetime
from airflow import DAG
from airflow.operators.python import PythonOperator
def _setup():
    pass
def _teardown():
    pass
def _test_case(s):
    pass
def _mark_success():
    pass
with DAG(
    dag_id="{{ dag_id }}", 
    is_paused_upon_creation=False, 
    start_date=datetime.datetime(2021, 1, 1), 
    catchup=False, 
    schedule="@once"
) as dag:
    setup_task = PythonOperator(_setup, ...)
    teardown_task = PythonOperator(_teardown, ...)
    mark_successful = PythonOperator(_mark_success)
    tests = []

Ezek után egy ciklust használunk, amely iterál a tesztesetek listáján, és a megfelelő PythonOperator segítségével végrehajtja őket. A tesztkészlet minden egyes tesztesete egy-egy feladatot hoz létre a DAG-ban:

python
Copy code
{% for task in tasks %}
    tests.append(
        PythonOperator(
            task_id="{{ task.name }}",
            python_callable=_test_case,
            op_args=[{{ task.value }}]
        )
    )
{% endfor %}

Ez biztosítja, hogy a DAG teljes egészében dinamikusan és testreszabott módon működjön.

Miután meghatároztuk a feladatokat és azok függőségeit, a következő lépés a fájl létrehozása, amelyben a generált DAG található. A templátumokat és a változókat felhasználva generáljuk a fájlt:

python
Copy code
with open("path/to/dag/folder", "w") as f:

    env = Environment()
    template = env.from_string(dag_template)
    output = template.render(dict(
        dag_id="unique_identifier",
        tasks=[{"name": "test_one", "value": 1}, {"name": "test_two", "value": 2}]
    ))
    f.write(output)

Ez a kód segít abban, hogy a tesztkészletekhez tartozó DAG-okat dinamikusan generáljuk a háttérben tárolt konfigurációk alapján.

A tesztkészletek ütemezését azonban nem szabad alábecsülni. Ha túl sok dinamikusan létrehozott DAG-ot hozunk létre, a rendszer teljesítménye csökkenhet, mivel minden egyes DAG-nak külön fájlban kell tárolódnia. Ennek elkerülése érdekében célszerű csak azokat a DAG-okat tartani, amelyek valóban futni fognak, és eltávolítani azokat, amelyek már sikeresen lefutottak:

python
Copy code
def _drop_successful_dags():

    pg_hook = PostgresHook(postgres_conn_id=TEST_CASE_CONN_ID)
    results = pg_hook.get_records('SELECT case_id FROM "cases" WHERE "case_status" = \'SUCCESS\'; ')
    for r in results:
        try:
            os.unlink(os.path.join("/opt", "airflow", "dags", f"case_{r[0]}_dag.py"))
        except FileNotFoundError:
            pass

Ez a kód segít abban, hogy a már sikeresen lefutott tesztekhez tartozó DAG fájlokat eltávolítsuk, és csak az aktuálisan szükséges DAG-okat tartsuk életben. Így minimalizálhatjuk a rendszer túlterheltségét, és biztosíthatjuk a zökkenőmentes működést.

Hogyan végezzünk sikeres migrációt az Airflow környezetében?

Az Airflow egy olyan alkalmazás, amely viszonylag nyitott és megosztott architektúrával rendelkezik, de megfelelő erőfeszítéssel az információk elkülöníthetők, hogy több bérlős (multi-tenant) környezetet biztosítson. Bár nincs egyetlen megoldás, amely minden igényt kielégítene, a bemutatott lehetőségek kombinációja segíthet elérni a biztonsági és üzemeltetési célokat. Mindig rendelkezésre áll egy lehetőség a teljes izolációra, további Airflow telepítéssel!

A migrációs feladatok végrehajtása és tervezése kulcsfontosságú lépés minden adatkezelési környezetben, így az Airflow esetében is. Ahogy az adatelemzési és mérnöki feladatok fejlődnek, előfordulhat, hogy egy-egy környezetet át kell telepíteni új infrastruktúrába, új technológiákra, vagy éppen katasztrófa-helyreállítás miatt. A migrációk gyakran komoly hatással vannak a vállalati működésre, ezért létfontosságú a megfelelő tervezés és az esetleges kockázatok minimalizálása.

A migrációt több lépésben kell végrehajtani, mind az operatív, mind a műszaki szempontokat figyelembe véve. Az első és legfontosabb feladat, hogy alaposan áttekintsük a migrálni kívánt munkafolyamatokat. A migráció előtt fontos, hogy részletes leltárt készítsünk minden egyes munkafolyamatról, azzal a céllal, hogy meghatározzuk a technikai és üzleti tulajdonosokat, a használt adatforrásokat, valamint az üzleti fontosságukat. Ehhez az alábbi kérdésekre kell választ találni:

• Hol található a munkafolyamat kódja/konfigurációja?

• Milyen gyakran fut a munkafolyamat?

• Milyen szélsőséges eseteket vagy teszteket hajtottak végre a munkafolyamattal kapcsolatban?

• Hogyan néz ki a "helyes" működés?

• Hogyan történt a munkafolyamat hibája a múltban?

• Mi történik, ha a munkafolyamat egy ideig nem működik?

A teljes leltár összeállítása után a munkafolyamatokat csoportosítani kell, és meghatározni a prioritásaikat. Az első lépés a munkafolyamatok elrendezése és az üzleti kritikus elemek kiemelése. A migráció során mindig legyenek egyértelmű prioritások, hogy mely munkafolyamatok futnak először, és melyek késlekedhetnek. Ezen a ponton olyan problémák is felmerülhetnek, mint az elveszett tudás, ha például a munkafolyamatokat fejlesztő személyek már nem dolgoznak a cégnél. Ezért különösen fontos a megfelelő dokumentáció és az információátadás. Mindezek után fontos, hogy meghatározzuk a migrációval kapcsolatos időpontokat, például az ütemezett frissítési időablakokat, vagy a kódlefagyásokat.

Miután a tervezés és a prioritások tisztázódtak, a következő lépés a tényleges migrációs munkálatok végrehajtása. Az egyik legfontosabb szabály, hogy az Airflow környezetben soha ne kezdjünk el refaktorálni, ha a már meglévő minták működnek. Az új kód akár nem is lesz esztétikus, de ha a forráskódot másolva készítjük el, akkor sokkal kisebb a kockázat a migráció során. A migráció sikerességét részben az is befolyásolja, hogy milyen mértékben tartjuk be a minimális paritás elvét, és próbálunk meg "ugyanazt" hozni az új környezetbe, mint ami a régi rendszerben volt.

A migráció során előfordulhat, hogy felhalmozódik egy technikai adósság (tech debt), amelyet nem szabad a migráció közben megpróbálni rendezni. Ehelyett fontos, hogy egy nyilvántartásban tartsuk a tech debt elemeit, hogy később döntéseket hozhassunk azok kezeléséről. Az átvitel során kiemelten fontos, hogy alapos teszteléseket végezzünk el. A tesztelés során ajánlott izolált környezetben tesztelni, ha lehetséges, akkor a termelési adatokat és skálákat használva, hogy biztosítsuk a munkafolyamatok hibátlan működését.

A migrációs folyamat során folyamatosan monitorozni kell a tevékenységeket. Napi státuszmegbeszélések tartása segíthet a problémák gyorsabb megoldásában és a folyamat előrehaladásának követésében. Ezen kívül hasznos eszköz lehet a RAID (Risks, Actions, Issues, and Decisions) napló vezetése, amely segít nyomon követni a migrációval kapcsolatos döntéseket és problémákat, valamint biztosítja, hogy minden érdekelt fél tisztában legyen a folyamat aktuális állásával.

A QA és tesztelési tervek készítése elengedhetetlen. Ha van rá lehetőség, érdemes egy tesztkörnyezetet kialakítani, amely a produkciós környezet pontos mása, de csak olvasási jogokat biztosít a termelési adatforrásokhoz. A tesztelés során minden esetben a valós adatokkal kell dolgozni, de azokat izolált környezetekben, hogy elkerüljük a termelési rendszer sérülését.

A migrációs projekt sikeressége érdekében nemcsak a műszaki részletek, hanem a projekt menedzsment aspektusai is alapvetőek. Az alapos tervezés és a folyamatos nyomon követés biztosítja, hogy a migráció zökkenőmentesen és sikeresen végbemehessen.