JobManager - automatické úlohy běžící na pozadí

Mezi klíčové vlastnosti modulu JobManager patří:

• umožnuje spouštět automatické úlohy v zadaném čase (alternativa triggerů v macros.xml),
• umožňuje prioritizovat úlohy a v případě tlaku na plánovač upřednostňovat prioritnější úlohy
• zveřejňuje Prometheus metriky, které umožňují monitoring úloh a případný alerting v případě problémů
• jobManager má administrační rozhraní v Edeem, kde je např. možné sledovat stav úlohy nebo vyvolat spuštění úlohy, pozastavit úlohu atp.

Registrace úloh

Aby si modul novou úlohu zaregistroval, musí úloha splňovat:

• implementovat rozhraní com.fg.job.mgr.business.task.Job,
• být zaregistrována do springového contextu (modulu jobManager, případně implementovat PublicInterface).

Další možností je přidat joby pomocí konfigurace. Aktuálně jsou podporovány tyto typy úloh:

• MACRO - úloha spustí cps macro.

Doporučení předci pro psaní vlastních úlohZkopírovat odkaz na sekci

Pokud píšete svou vlastní úlohu neimplementujte přímo rozhraní com.fg.job.mgr.business.task.Job, ale vyjděte z jedné z následujících abstraktních tříd:

• com.fg.job.mgr.business.task.AbstractJob: zde získáte zdarma pojmenování jobu podle názvy beany ve Spring kontextu
• com.fg.job.mgr.business.task.AbstractUniqueJob: zde získáte zdarma totéž jako v AbstractJob a navíc zajištění unikátního běhu v rámci clusteru pomocí principu zamykání procesu formou leasingu unikátního klíče ve sdílené databázi
• com.fg.job.mgr.business.task.AbstractGreedyJob nebo com.fg.job.mgr.business.task.AbstractGreedyJobWithContext: zde získáte zdrama totéž co v AbstractJob a zároveň máte zajištěno zpracování velkého počtu záznamů pomocí stránkovaného přístupu s kontrolou nekonečného cyklu a možnosti přerušování běhu
  • u AbstractGreedyJobWithContext pak máte navíc pod kontrolou vlastní kontext, který se v rámci zpracování jednotlivých stránek přenáší
• com.fg.job.mgr.business.task.AbstractGreedyUniqueJob nebo com.fg.job.mgr.business.task.AbstractGreedyUniqueJobWithContext: zde získáte zdarma totéž co v AbstractUniqueJob a zároveň máte zajištěno zpracování velkého počtu záznamů pomocí stránkovaného přístupu s kontrolou nekonečného cyklu a možnosti přerušování běhu
  • u AbstractGreedyUniqueJobWithContext pak máte navíc pod kontrolou vlastní kontext, který se v rámci zpracování jednotlivých stránek přenáší

U unikátních jobů je možné, že běh konkrétního jobu bude vynechán z důvodu toho, že běží na jiném nodu clusteru nebo z jiných příčin se mu nepodařilo zavést si svůj zámek do sdílené databáze. V takovém případě dojde k přeplánování opětovného pokusu o spuštění mimo standardní plán (ve výchozí konfiguraci je to 5 minut, ale je možné to změnit pomocí nastavení skippedJobPlanningInterval v konfiguraci modulu, formát odpovídá java.time.Duration#parse(String)).

U "greedy" implementací je vhodné implementaci připravit tak, aby byla tzv. přerušitelná a implemenovat rozhraní com.fg.job.mgr.business.task.InterruptibleJob. Pak úloha reaguje na žádost o přerušení tím, že se předčasně ukončí, i když má další záznamy ke zpracování. Úloha však musí být napsána tak, aby dokázala pokračovat tam, kde naposledy skončila, což není vždy možné. I v případě, že úloha není přerušitelná, systém se ji může pokusit tzv. pozastavit (yield) s tím, že její kontextový objekt (stav) je uchován na pozadí. Některé úlohy si však nemohou dovolit ani to - např. proto, že přetěžují metodu doRun a dělají tam nějaké nestandardní věci. Pro ty je připraveno rozhraní com.fg.job.mgr.business.task.UninterruptibleJob, které zabrání i pozastavení (yield) úlohy.

Logování průběhu úloh

V rámci dané úlohy stačí standardně logovat do aplikačního logu. Modul sám si obalí běh úlohy tak, aby byl schopen si výsledný log vytáhnout a uložit do databáze.

Běh úlohy jako takové je označen jako chybný, pokud se kdykoliv v průběhu zaloguje zpráva se závažností ERROR, podobný význam má i důležitost WARNING, která nicméně nevyvolá notifikaci vývojáře. Logy jsou ukládány do databáze a jsou automaticky vymazávány. Standardně takto:

• chybové běhy (ERROR): 1 měsíc
• problémové běhy (WARNING): 7 dní
• bezproblémové běhy: 2 dny

Pokud běh jobu nezpracoval žádný záznam je udržován pouze poslední takovýto log.

Monitoring úloh

Job manager pro všechny úlohy zveřejňuje následující metriky:

• gauge: job_state - stav úloh (label group, job, state může být running/planned)
• counter: job_started_total - počet nastartování úloh (label group, job)
• counter: job_finished_total - počet ukončení úloh (label group, job, state může být skipped, OK, WARNING, ERROR), - vhodné ALERTOVAT, pokud je více jak 1h v ERROR stavu
• counter: job_records_processed_total - počet zpracovaných záznamů (label group, job)
• gauge: job_records_queued - fronta záznamů ke zpracování (label group, job) - vhodné ALERTOVAT, pokud za 12h pouze roste a nikdy neklesne
• gauge: job_time_to_execute_seconds - počet sekund do dalšího spuštění úlohy (label group, job) - vhodné ALERTOVAT, pokud klesne pod 2h
• summary: job_duration_seconds - délka běhu úlohy (label group, job)

Práce s thread poolem

Pro běh asynchronních úloh se využívá systémový CPS scheduler a tudíž se sdílí vlákna i s ostatními úlohami v CPS. O spouštění úloh se stará vlastní plánovač ve třídě com.fg.job.mgr.business.task.JobRunner, který vyžaduje ten stejný thread pool. Pokud jsou tedy všechna vlákna thread poolu rozebrána, plánování i běh dalších úloh čeká, než se nějaké vlákno uvolní. Priorita spouštění úloh je dána datumem jejich dalšího spuštění (tj. první bude spuštěna úloha s nejbližším uplynulým datem spuštění). Další datum spuštění je vypočteno po ukončení úlohy dle její konfigurace.

Pokud máme úlohu, která se má spouštět každých 10 vteřin a thread pool je po nějakou dobu vyčerpán bude spuštění vypadat následujícím způsobem:

• A - úloha vyžadující spuštění každých 10 sekund
• B - dlouhotrvající úloha, která vyčerpá poslední vlákno thread poolu

1. 10:00:00 - A, další spuštění naplánováno na 10:00:10
2. 10:00:02 - B, běh trvá 5 minut
3. 10:05:02 - A, další spuštění naplánováno na 10:05:12

Díky sdílení thread poolu může dojít k tomu, že se nebudou stíhat vyřizovat vstupní datové fronty. Takový případ se projeví v metrikách job manageru a měl by zareagovat alert v monitoringu.

Prioritizace úlohZkopírovat odkaz na sekci

Pokud je zapnutá prioritizace úloh (ve výchozím stavu zapnuto, ovládá se nastavením prioritizeJobs), proběhne u všech úloh, jejichž datum spuštění je v minulosti, výpočet jejich pořadí a v tomto pořadí jsou přidány do thread poolu. Pořadí se počítá vzorcem:

( DURATION(now, nextPlannedDate).toSeconds / AVG_DURATION + AVG_DURATION) / AVG_DURATION * PRIORITY

Kde:

• AVG_DURATION je exponenciální plovoucí průměr za posledních 20 běhů úlohy (která zpracovala nějaké záznamy)
• now je aktuální datum
• nextPlannedDate je datum plánovaného spuštění
• (DURATION(now, nextPlannedDate).toSeconds) je rozdíl hodnot v sekundách
• PRIORITY je priorita úlohy, která se nastavuje v její konfiguraci (ve skocích 1,2,4,16)

Tento výpočet má tu vlastnost, že upřednostňuje krátkoběžící úlohy v rámci své priority. Pokud se však krátkoběžící úloha vyskytne ve velkém zpoždění, její vypočtěné "skóre" bude enormně vzrůstat a dokáže přeskočit prioritu. Úlohy by tedy měly být vykonány následovně:

• zpožděné krátkoběžící úlohy
• úlohy s vyšší prioritou
• krákoběžící úlohy bez většího zpoždění

Technicky je výpočet odvozen z používaného algoritmu schedulingu v OS Highest Response Ratio Next Scheduling

Předčasné ukončování dlouhoběžících úlohZkopírovat odkaz na sekci

Pokud je zapnuté předčasné ukončování úloh (ve výchozím stavu zapnuto, ovládá se nastavením pauseLongRunningJobs), prověří se zda limity threadpoolu nejsou přehlceny naplánovanými úlohami. Pokud ano, porovná se priorita úloh ve frontě viz prioritizace úloh. Pokud má některá z úloh ve frontě dvakrát větší prioritu než úloha běžící a zároveň běžící úloha přerušení podporuje tím, že implementuje marker rozhraní com.fg.job.mgr.business.task.InterruptibleJob, plánovač u ní nastaví příznak předčasného ukončení a inkrementuje prioritu pozastavované úlohy čímž zajistí, že se úloha přerušuje pouze dokud se priorita nezvyší natolik aby zamezila přeskočení i velmi prioritními úlohami. Tato synteticky nastavená priorita je každých 5(JobRunner#PRIORITIZED_RUN_THRESHOLD) minut běhu úlohy znovu ponižována dokud nezůstane na původní hodnotě úlohy.

Implementace marker rozhraní InterruptibleJob má smysl pouze u úloh, které vychází z předka AbstractGreedyJobWithContext tj. zpracovává záznamy po menších balících. Při přerušení dokončí úloha pouze svůj aktuálně rozpracovaný balík (stránku) a další už nezačne. Zároveň se přeplánuje, aby se co nejdříve znovu pustila znovu a mohla pokračovat dál. Tímto způsobem ovšem uvolní slot v thread poolu pro další úlohy. Je důležité, aby úloha při opakovaném spuštění uměla navázat tam, kde předtím skončila - úloha nesmí začínat zpracovávat záznamy zase ze začátku, jinak hrozí zacyklení úlohy. Proto vzniklo marker rozhraní, kterým dává programátor najevo, že úloha je na toto přerušení připravena.

Předčasné ukončování úloh, které na to nejsou připravenyZkopírovat odkaz na sekci

Řada jobů nemůže přerušit svůj běh, protože nedokáží navázat. A často to není ani v našich silách zajistit, protože jsme závislí na chování externího systému. Tyto joby mohou také běžet celkem dlouho a i když implementují logiku stránkovaného zpracování, tak reálně pozastavit nejdou, protože si klíčové informace drží v kontextu, který se mezi různými běhy jobu neukládá. Proto je možné zapnout podporu tzv. potlačení úloh.

Job v rámci přerušení zpracování si odloží aktuální kontext do globální proměnné a použije ho při opětovném spuštění. Tj. ve své paměti si uchová poslední použitý kontext úlohy a po obnovení běhu s tímto kontextem zase naváže. Problémem je, že v kontextu může být celkem dost dat a pokud bychom takto na pozadí dali více jobů, mohli bychom se začít setkávat s OutOfMemory výjimkami, takže je možné konfiguračně určit (pomocí nastavení yieldLimit - ve výchozím stavu hodnota 2), kolik takovýchto jobů si můžeme dovolit na pozadí ponechat.

Zároveň se musí ověřit, že prioritnější job a job, který se má pauznout a zůstat na pozadí, nesdílí stejný zámek pro unikátní běh, jinak by samozřejmě prioritnější job stejně nemohl provést své spuštění. Potlačené úlohy dále drží zámek unikátního spuštění v databázi a pokud jsou potlačené dlouho, tak jim může vypršet. V takovém případě už úloha nedokáže navázat, protože jako první akci zkouší znovu prodloužení zámku.

Poznámka: pokud bylo spuštění úlohy vynuceno ručně adminstrátorem je tato úloha jednorázově označena jako nepřerušitelná a nezastaví se do doby než se sama svého vlákna vzdá. Tento ústupek byl učiněn proto, že ručně spouštěné úlohy chtějí být z pohledu uživatele dokončeny co nejdříve a případná přerušení vynucená prioritnějšími úlohami na pozadí vadila.

Ad-hoc zaplánování spuštění úlohy

Pokud chceme vynutit předčasné spuštění úlohy i když ještě neuplynul standardně naplánovaný čas spuštění, je možné toto provést tímto voláním:

1// JobManager je veřejné rozhraní modulu, tj. je možné jej autowirovat2final JobManager jobManager;3jobManager.triggerExecution("nazevUlohy");

Případně je toto možné provést i takto:

1// JobManager je veřejné rozhraní modulu, tj. je možné jej autowirovat2final JobManager jobManager;3jobManager.getJob("nazevUlohy").triggerExecution();

Toto volání přeplánuje spuštění úlohy na nejbližší možný okamžik. Nicméně i nadále bude úloha spouštěna ve sdíleném thread poolu, takže pokud je pool vyčerpán, bude se čekat na uvolnění vlákna.

Při triggrování úlohy je možné i připojit prioritu, která bude platná jen pro nejbližší další spuštění a vynutit tak prioritizované spuštění úlohy, která jinak až tak prioritní není. Je možné vynutit i nepřerušitelné spuštění úlohy, ale to nedoporučujeme programově využívat a mělo by být ponecháno pouze pro ruční spuštění administrátorem.

Poslední volbou je možnost nastavení parametru untilFullyDone, který při nastavení na true přeplánuje úlohu tolikrát, kolik bude potřeba běhů k úplněmu zpracování fronty. Pozor! tato funkce je nebezpečná, tímto nastavením úloha zablokuje prakticky nepřerušitelně vlákno, dokud nebude plně zpracována její fronta.

Fronta úloh ke spuštění

V gui modulu je z hlavního listingu přístupná stránka "Fronta úloh". Tato stránka zobrazuje všechny úlohy k provedení v následujícím pořadí:

1. spuštěné úlohy
2. úlohy čekající ke spuštění
3. naplánované úlohy v budoucnosti

Pokud úloha čeká na volné vlákno ke spuštění, pak je ve sloupci priorita zobrazeno její aktuální "skóre" v boji o další "threadový slot". Tato hodnota je porovnatelná proti ostatním čekajícím a tedy úloha s nejvyšším skóre bude spuštěna následující - viz prioritizace úloh.

Závislosti mezi úlohami

Pokud je potřeba zajistit, aby se jedna úloha vykonala až po dokončení jiné úlohy, je možné v těle logiky úlohy vyvolat com.fg.job.mgr.exception.PreconditionNotFulfilled výjimku, která ve svých datech nese informaci o typu jiné úlohy, na které je závislá. Zároveň je nutné aby úloha používala konfiguraci typu com.fg.job.mgr.model.config.AdaptiveJobConfiguration.

Při této kombinaci dojde při ukončení úlohy k vyhledání všech ostatních úloh závislého typu, zjistí se poslední plánované datum jejich spuštění a za toto datum se naplánuje spuštění aktuální úlohy se započtením čekacího času odpovídajícího standardnímu intervalu spouštění aktuální úlohy.

Typy úloh

V plánovači úloh lze použít následující konfigurace pro plánování úloh:

Úlohy spouštěné denně v konkrétní časZkopírovat odkaz na sekci

Pokud chcete spouštět úlohu jednou denně v konkrétní čas, použijte konfiguraci com.fg.job.mgr.model.config.DailyJobConfiguration. Tato konfigurace se hodí pro čistící úlohy, pravidelné importní, generování exportních souborů a podobně.

Úlohy spouštěné pravidelně po určitém intervaluZkopírovat odkaz na sekci

Pokud chcete spouštět úlohu pravidelně po uběhnutí konkrétního časového intervalu, použijte konfiguraci com.fg.job.mgr.model.config.TimeIntervalJobConfiguration. Tato konfigurace se hodí pro pravidelné synchronizační úlohy s externími systémy.

Úlohy spouštěné dle CRON výrazuZkopírovat odkaz na sekci

Pokud chcete spouštět úlohu dle CRON výrazu, použijte konfiguraci com.fg.job.mgr.model.config.CronJobConfiguration. Tuto konfiguraci použijte, pokud si nevystačíte s předchozími dvěma typy konfigurací.

Manuální úlohyZkopírovat odkaz na sekci

Pokud nechcete, aby byla úloha spouštěna automaticky, ale chcete umožnit uživateli ruční spuštění z uživatelského rozhraní, použijte konfiguraci com.fg.job.mgr.model.config.ManualJobConfiguration. Tato konfigurace se hodí pro ad-hoc skripty, které mají být spouštěné podle potřeby uživatelem (přegenerování nějakých dat, ruční exporty atp.).

Jednorázové úlohy (typicky migrační)Zkopírovat odkaz na sekci

Pokud chcete, aby úloha byla spuštěna právě a pouze jedinkrát, použijte konfiguraci com.fg.job.mgr.model.config.FireAndForgetJobConfiguration. Tato konfigurace zajistí spuštění úlohy krátce po nastartování systému, ale pouze pokud tato úloha ještě nikdy úspěšně nedoběhla. Pokud úloha skončí s chybou, je po 10 minutách spuštěna znovu.

Adaptivní strategie pro spouštění úlohyZkopírovat odkaz na sekci

Adaptivní strategie com.fg.job.mgr.model.config.AdaptiveJobConfiguration obaluje předchozí typy konfigurací:

• úlohy spouštěné v konkrétní čas
• úlohy spouštěné po určitém intervalu
• úlohy spouštěné dle CRON výrazu

Adaptuje ovšem další naplánování podle výsledku posledního spouštění. Pokud při posledním spuštění dojde k chybě je další spuštění zaplánováno předčasně po X minutách (ve výchozím nastavení 10 minut, ale lze nakonfigurovat dle vašich potřeb). Zároveň ovšem limituje počet takto předčasně naplánovaných pokusu na konkrétní počet (ve výchozím nastavení 5 opakovaných pokusů, ale lze přenastavit). Pokud ani po těchto X pokusech nedoběhne úloha bez chyb je už další pokus naplánován zase podle standardního rozpisu úlohy.

Tato konfigurace se hodí pro úlohy, které jsou spouštěné jednou za delší dobu a v případě krátkodobé chyby (např. dočasného výpadku externího API), by k dalšímu pokusu došlo až po tomto dlouhém intervalu. Zrychlení opakovaných pokusů v případě chyby umožnuje úloze překonat i výjimečně nahodilé problémové stavy.

Vlastní kontrola nad plánováním úlohyZkopírovat odkaz na sekci

Pokud si nevyberete z výše uvedených přepřipravených možností, můžete si implementovat vlastní plánovač rozšířením základního rozhraní com.fg.job.mgr.model.config.AbstractJobConfiguration.

Příklad konfigurace modulu

1<!-- Definice modulu -->2<module customName="jobManager" classname="com.fg.job.mgr.cps.JobMgrModule">3
4        <!-- Definice messages - většinou pro pojmenování custom jobů na projektu -->5        <localization>6            <bundle>storage:f.cfg:/module-job-manager/i18n/messages</bundle>7        </localization>8
9        <!-- Vypnutí/Zapnutí spouštění automatických úloh. V DEV většinou vypnuto. -->10        <asyncDisabled>true/false</asyncDisabled>11        12        <!-- Maximální počet pozastavených (yield) úloh držených na pozadí. Nepovinný, výchozí hodnota je 2. -->13        <yieldLimit>4</yieldLimit>14
15        <!-- Labely tohoto nodu pro filtrování úloh (viz "Filtrování úloh podle labelů") -->16        <nodeLabels>admin</nodeLabels>17        <!-- Zakáže spouštění úloh bez labelů na tomto nodu -->18        <unlabeledJobsDisabled>false</unlabeledJobsDisabled>19
20        <!-- Konfigurace skupin úloh -->21        <groups>22            <!-- Vypnutí celé skupiny -->23            <group asyncDisabled="true">ESHOP</group>24            <!-- Přiřazení labelů skupině — úlohy v této skupině poběží jen na nodech s odpovídajícím labelem -->25            <group labels="admin">MAINTENANCE</group>26            <!-- Kombinace obou atributů je možná -->27            <group labels="admin,fe">GENERAL</group>28        </groups>29
30        <!-- Konfigurace úloh -->31        <jobs>32            <!-- @name - název úlohy. Povinný. Název musí být unikatní v rámci všech jobů. -->33            <!-- @type - typ úlohy. Povinný. Aktuálně podporován pouze typ "MACRO"  -->34            <!-- @externalJobId - externí identifikátor úlohy (např. id makra). Nepovinný. Pokud údaj není vyplněn, beze se @name -->35            <!-- @timeInterval - "lidsky" zapsaný interval - více zde <@classDescriptor class='com.fg.commons.util.TimeIntervalParserUtils'>TimeIntervalParserUtils</@classDescriptor>. Např: 15m, 2h30m, 1d4h15m30s. Nepovinný. -->36            <!-- @cron - Interval zapsaný v CRON výrazu. Nepovinný. -->37            <!-- @daily - Spouštění jobu v daný čas. Nepovinný. Např.  "02:30:00" -->38            <!-- minimálně jeden z atributů @timeInterval,@cron nebo @daily je povinný-->39            <job name="someUniqueJobName" type="MACRO" externalJobId="mockMacro" timeInterval="1m" />40        </jobs>41
42        <!-- Možnost překonfigurování atributů úloh (resp. jejich JobConfiguration) -->43        <jobsConfigOverride>44            <job name="someName" variation="variation">45                <sameJobAtrributeName>sameAttributeValue</sameJobAtrributeName>46            </job>47            <!-- Přiřazení labelu konkrétní úloze (přepíše label skupiny) -->48            <job name="cacheWarmupJob">49                <labels>fe,admin</labels>50            </job>51        </jobsConfigOverride>52
53        <!-- Konfigurace Edee GUI  -->54        <gui ramJetModuleName="edeeui" admin="true"55             parentSection="edee"56             liveConfigParentNode="edee/jobManager"57             liveConfigStorage="storage:f.cfg:/module-job-manager/gui-admin">58        </gui>59
60</module>

Registrace úloh z externího moduluZkopírovat odkaz na sekci

Pokud potřebujete registrovat úlohy z jiného modulu, musí daný modul implementovat rozhraní com.fg.job.mgr.cps.ModuleJobProvider. Toto rozhraní je voláno při inicializaci modulu job manageru. Rozhraní má dvě metody:

• getJobs() - vrací seznam úloh, které mají být registrovány.
• getModuleTitle(Locale locale) - vrací název modulu, který je zobrazen v GUI job manageru.

Příklad implementace:

1@Override2public List<Job> getJobs() {3    return new LinkedList<>(moduleConfiguration.getBeansOfType(Job.class).values());4}5
6@Override7public String getModuleTitle(Locale locale) {8    return moduleConfiguration.getMessage("job.module.edeeShop", null, "E-shop", locale);}

Aby se pro úlohy z externího modulu zobrazovaly správně názvy v administračním GUI, musí úloha implementovat com.fg.job.mgr.web.dto.JobTitleResolver. Základní abstraktní implementace, ze kterých se standardně dědí, již toto rozhraní implementují.Zkopírovat odkaz na sekci

Běh úloh v clusteruZkopírovat odkaz na sekci

Job manager je aktuálně plně připraven k využití v clusteru. Plán, dle kterého se úlohy vykonávají, je sdílen v DB a tedy jakékoliv přeplánování uživatelem je možné vykonat na méně vytíženém nodu.

Jediný známý prozatimně nevyřešený problém je s využitím původní implementace mybatis, která vždy aktualizuje všechna data plánu. I přesto, že jsou operace s plánem udržovány co nejtenčí, může nastat situace, kdy se mezi get a update plánu změní data v DB druhým nodem clusteru a tedy se údaje druhého nodu při updatu přepíší. To bude nutné v budoucnu řešit buď vytvářením locku daného plánu v db, nebo pouze updatem změněných hodnot.

Dalším stěžejním bodem pro běh v clusteru bude respektování změny nextExecution, změněné jiným nodem. To bude možné řešit buď implementací cluster eventů, které změnu zpropagují do druhého nodu a ten vyvolá přeplánování v priority queue. Případně přepsáním použití PriorityQueue, avšak to by bylo značně rizikové.

Filtrování úloh podle labelů (multi-pod cluster)Zkopírovat odkaz na sekci

Při rozdělení edee na pody s různým určením (např. admin a frontend) je potřeba řídit, které úlohy smějí běžet na kterém podu. K tomu slouží label-based filtrování, které rozšiřuje stávající mechanismus skupin.

PrincipZkopírovat odkaz na sekci

Každý node (pod) deklaruje své labely v konfiguraci job manageru. Úlohy jsou filtrovány podle labelů přiřazených jejich skupinám — systémové knihovny nemusejí znát topologii nasazení, labely se přiřazují skupinám v konfiguraci projektu.

Úloha běží na nodu, pokud se alespoň jeden z labelů její skupiny shoduje s alespoň jedním labelem nodu (OR sémantika).

Priorita vyhodnoceníZkopírovat odkaz na sekci

1. Pokud má úloha vlastní labels (z kódu nebo jobsConfigOverride) — použijí se ty
2. Jinak se použijí labels skupiny úlohy (z <groups> konfigurace)
3. Pokud ani skupina nemá labely — rozhoduje příznak unlabeledJobsDisabled

KonfiguraceZkopírovat odkaz na sekci

Labely skupin se nastavují atributem labels na elementu <group>:

1<groups>2    <group labels="admin">MAINTENANCE</group>3    <group labels="admin,fe">GENERAL</group>4</groups>

Příklady konfiguraceZkopírovat odkaz na sekci

Admin pod — spouští úlohy s labelem admin + všechny úlohy bez labelů:

1<nodeLabels>admin</nodeLabels>2<unlabeledJobsDisabled>false</unlabeledJobsDisabled>3<groups>4    <group labels="admin">MAINTENANCE</group>5    <group labels="admin,fe">GENERAL</group>6</groups>

Frontend pod — spouští pouze úlohy, které mají explicitně přiřazený label fe:

1<nodeLabels>fe</nodeLabels>2<unlabeledJobsDisabled>true</unlabeledJobsDisabled>

Pokud se konfigurace liší pouze podle typu podu, je možné použít <environmentspecific>:

1<environmentspecific useIf="not(in(labels,'admin'))">2    <nodeLabels>fe</nodeLabels>3    <unlabeledJobsDisabled>true</unlabeledJobsDisabled>4</environmentspecific>

Přiřazení labelu konkrétní úloze (escape hatch)Zkopírovat odkaz na sekci

Ve výjimečných případech, kdy potřebujete přepsat label skupiny pro jednu konkrétní úlohu, použijte <labels> v <jobsConfigOverride>:

1<jobsConfigOverride>2    <job name="cacheWarmupJob">3        <labels>fe,admin</labels>4    </job>5</jobsConfigOverride>

Případně v kódu úlohy:

1public class MyJobConfiguration extends CronJobConfiguration {2    public MyJobConfiguration() {3        super("0 0 2 * * ?");4        setLabels(new HashSet<>(Arrays.asList("admin")));5    }6}

Trashing job plánů v multi-pod prostředíZkopírovat odkaz na sekci

Pokud má node nastaveny nodeLabels, automaticky přeskakuje trashing (mazání) plánů úloh, které nezná — tyto plány mohou patřit jinému podu. Stávající příznak trashingDisabled zůstává k dispozici jako explicitní safety-net.

Reakce na události a životní cyklus úlohy

Job Manager nabízí dva mechanismy, kterými lze reagovat na dění kolem úloh. Před výběrem si projděte rozhodovací tabulku — záměrně se nepřekrývají:

Pravidlo palce: events = "něco se stalo, dej vědět zájemcům" (loose coupling, fire-and-forget, řazení nezaručeno). JobLifecycleListener = "obal každý běh úlohy spolehlivým start/end párem" (tight coupling k běhu úlohy, pevně garantovaný cleanup, na jednom vlákně).

Aplikační událostiZkopírovat odkaz na sekci

Při dění kolem úloh Job Manager publikuje standardní Spring ApplicationEvent instance — vy si můžete odchytávat přes @EventListener nebo ApplicationListener<T>. Některé události jsou navíc označeny jako com.fg.webapp.cps.v1.modules.spring.PublicApplicationEvent (jsou viditelné napříč všemi CPS moduly) a/nebo com.fg.interaction.event.ClusterEvent (propagují se na ostatní nody clusteru).

JobStatusEvent a JobFinishedEvent jsou PublicApplicationEvent + ClusterEvent — chytají se napříč moduly i nody clusteru. JobPlanUpdatedEvent je ClusterEvent. Ostatní jsou lokální Spring události.

Příklad: odeslání notifikace na MatterMost po doběhnutí úlohy:

1@Component2public class SlackNotifier {3
4    @EventListener5    public void onJobFinished(JobFinishedEvent event) {6        if (event.getJobResult().getProcessedItems() > 0) {7            slack.send("Úloha " + event.getJobName() + " zpracovala "8                + event.getJobResult().getProcessedItems() + " záznamů.");9        }10    }11}

Pozor: pořadí, ve kterém Spring volá listenery, není garantované. Pokud potřebujete spolehlivý úklid stavu přesně kolem volání doJob, použijte JobLifecycleListener — events tuto roli záměrně nepokrývají.

JobLifecycleListenerZkopírovat odkaz na sekci

Pokud potřebujete reagovat na zahájení a ukončení každého běhu úlohy — typicky kvůli inicializaci a spolehlivému úklidu nějaké ThreadLocal proměnné, MDC kontextu, vlastního spanu nebo podobné per-execution stavové informace — slouží k tomu rozhraní com.fg.job.mgr.business.task.lifecycle.JobLifecycleListener. Toto rozhraní je záměrně úzké: garantuje pouze pár start → end kolem volání doJob úlohy. Bohatší události (JobFinishedEvent, JobPlanUpdatedEvent, ...) řeší application události — tahle SPI je záměrně neduplikuje.

Princip fungováníZkopírovat odkaz na sekci

Listener je registrován jako Spring beana v kterémkoliv CPS modulu. Při zavedení modulu si Job Manager listenery automaticky vyhledá (přes BeanFactoryUtils.beansOfTypeIncludingAncestors) a obalí jimi každé budoucí spuštění úlohy. Listener vrací z metody onStart instanci com.fg.job.mgr.business.task.lifecycle.JobEndHandler, která bude právě jednou zavolána po skončení běhu úlohy — bez ohledu na to, jak skončila:

• úspěšně (SUCCEEDED) — doJob doběhlo bez výjimky a fronta byla zpracována
• přerušeno (SUSPENDED) — plánovač úlohu pozastavil (yield nebo interrupt) a později ji obnoví (typicky se stejným vláknem); v tomto stavu obvykle NECHCETE uklízet ThreadLocal stav, protože úloha bude pokračovat
• přeskočeno (SKIPPED) — úloha vyhodila JobSkippedException (typicky nedostala distribuovaný zámek)
• chyba (FAILED) — úloha vyhodila libovolnou jinou výjimku

Typ ukončení a případnou výjimku najdete v com.fg.job.mgr.business.task.lifecycle.JobEndContext předaném do end handleru.

Příklad listeneruZkopírovat odkaz na sekci

1import com.fg.job.mgr.business.task.lifecycle.*;2import org.apache.logging.log4j.ThreadContext;3import org.springframework.stereotype.Component;4
5@Component6public class TenantContextListener implements JobLifecycleListener {7
8    private static final ThreadLocal<String> CURRENT_TENANT = new ThreadLocal<>();9
10    @Override11    public JobEndHandler onStart(JobStartContext start) {12        // inicializace per-execution stavu13        final String tenant = resolveTenantFor(start.getJob());14        CURRENT_TENANT.set(tenant);15        ThreadContext.put("tenant", tenant);16
17        return end -> {18            // úklid, který musí proběhnout vždy — i při chybě nebo přeskočení19            if (end.getTerminalState() != JobTerminalState.SUSPENDED) {20                // SUSPENDED znamená, že úloha bude pokračovat — kontext necháváme21                CURRENT_TENANT.remove();22                ThreadContext.remove("tenant");23            }24        };25    }26}

Listener je auto-discovered z kontextu modulu, ve kterém je registrován — žádná explicitní registrace v XML konfiguraci Job Manageru není potřeba.

Pravidla pro implementaciZkopírovat odkaz na sekci

• Jeden start = jedno end: Job Manager garantuje, že pro každé volání onStart se právě jednou zavolá vrácený JobEndHandler. Pokud listener vrátí null, end handler se neregistruje (vhodné pro listenery, které chtějí reagovat jen na podmnožinu úloh).
• Pořadí volání: více listenerů se volá v pořadí registrace na startu a v opačném pořadí (LIFO, jako resource cleanup) na konci.
• Výjimky v listeneru: cokoliv, co listener vyhodí (z onStart i z onEnd), je zalogováno a zahozeno — listener nemůže ovlivnit výsledek úlohy ani práci ostatních listenerů.
• Vlákno: pár onStart → onEnd vždy proběhne na jednom vlákně. Více úloh ovšem může běžet paralelně, takže listener musí být reentrant.

Zjištění aktuální závažnosti běhu zevnitř úlohyZkopírovat odkaz na sekci

Z těla úlohy (doJob) nebo z JobLifecycleListener end handleru je možné zjistit, jakou nejvyšší závažnost už proběhlé logy mají — užitečné, pokud chcete dle toho rozhodnout (např. odlišně reagovat, když se v průběhu zalogoval WARNING, i když úloha jinak doběhne v pořádku):

1// z těla úlohy2public JobResult doJob(JobSettingsAccessor<MyConfig> s) {3    // ...4    if (getCurrentSeverity() == JobExecutionResult.WARNING) {5        log.info("Pozor — průběh obsahuje varování, plánujeme retry zítra ráno.");6    }7    return new JobResult(...);8}9
10// z greedy úlohy přes JobContext11@Override12protected void doUniqueJobTurn(JobContext ctx) {13    // ...14    if (ctx.getCurrentSeverity() == JobExecutionResult.ERROR) {15        // ...16    }17}18
19// z listener end handleru20return end -> {21    if (end.getSeverity() == JobExecutionResult.ERROR) {22        metrics.errorJobsCounter.increment();23    }24};

Hodnoty com.fg.job.mgr.model.result.JobExecutionResult: OK (nic zajímavého neviděno), WARNING (proběhl alespoň jeden WARN log), ERROR (proběhl alespoň jeden ERROR/FATAL log), NONSIGNIFICANT (mimo otevřený běh — voláno mimo doJob).

Implementace je v helperu com.fg.job.mgr.business.task.lifecycle.JobSeverity, který čte úroveň z aktuálně otevřeného JobExecutionLogContainer přes MDC.

Podpora pro paralelní zpracování

V některých typech úloh se vyplatí v rámci jednoho běhu jobu spouštět dílčí podúlohy paralelně. Typicky tehdy, když:

• úloha provádí velké množství operací, které především čekají na odezvu (např. síťová komunikace s externí službou, čekání na I/O),
• nebo obecně všude tam, kde se „nepálí“ CPU našeho stroje, ale jsme závislí na odezvě někoho/něčeho jiného.

V takových případech se vyplatí „zatížit“ dané externí systémy paralelně a tím běh úlohy celkově zrychlit – a to vše v rámci jednoho běhu jobu.

Pro tento účel je k dispozici metoda executeSubTask(...) ve třídě com.fg.job.mgr.business.task.AbstractJob, která umožňuje bezpečně spouštět podúlohy na zadaném Executoru, zachovat tracing a konsolidovat logy do jediného logu nadřízené úlohy. Zároveň je zachován výpočet hashe z logu, takže opakované spuštění se stejným výsledkem a totožným obsahem logu nevytváří nové záznamy v DB, ale aktualizuje ten předchozí.

Jednoduchý příklad použitíZkopírovat odkaz na sekci

Následující příklad ukazuje (silně zjednodušeně) paralelní volání externího API nad kolekcí vstupů. Podúlohy jsou spouštěny přes executeSubTask(...), logy podúloh se automaticky připojí do nadřazeného logu běžící úlohy a telemetrie udržuje správné span/atributy.

1import com.fg.job.mgr.business.task.AbstractJob;2import java.util.List;3import java.util.concurrent.CompletableFuture;4import java.util.concurrent.Executor;5import java.util.concurrent.Executors;6
7public class MyParallelJob extends AbstractJob<MyConfig> {8    // Pro I/O bound práci je typické mít menší vyhrazený pool vláken9    private final Executor ioPool = Executors.newFixedThreadPool(8);10
11    @Override12    public void run(MyConfig cfg) {13        List<String> urls = cfg.urls();14
15        // Spuštění podúloh paralelně – každá podúloha má své jméno pro tracing/logy16        List<CompletableFuture<Response>> futures = urls.stream()17            .map(url -> executeSubTask(ioPool, "fetch-" + sanitize(url), () -> {18                log.info("Volám externí API {}", url);19                // volání externí služby – převážně čeká na odpověď20                return httpClient.get(url);21            }))22            .toList();23
24        // Vyčkání na všechny výsledky (výjimka z podúlohy je znovu vyhozena)25        List<Response> responses = futures.stream()26            .map(CompletableFuture::join)27            .toList();28
29        // ... další zpracování výsledků30    }31
32    private String sanitize(String url) {33        return url.replaceAll("[^a-zA-Z0-9]", "_");34    }35}

Poznámky k chování:

• Každá podúloha běží v nově otevřeném JobExecutionLogContainer, jehož snapshot (JobExecutionLog) je po dokončení automaticky připojen do otevřeného journálu nadřazené úlohy.
• Běh podúlohy je obalen kořenovým spanem (např. „Task - fetch-…“) s atributy jobName a subTaskName, takže v telemetrii je jasně vidět struktura běhu.
• Pokud podúloha vyhodí RuntimeException, bude zalogována, výjimka se po připojení logu k nadřazenému journálu znovu vyhodí (v join() se tedy projeví jako CompletionException).

Kdy se paralelizace hodíZkopírovat odkaz na sekci

• Hromadná komunikace s externími API, kdy každý krok čeká na odezvu protistrany.
• Hromadné I/O operace (síť, souborový systém, messaging), kde je doba čekání dominantní proti času CPU.

Upozornění – kdy být opatrnýZkopírovat odkaz na sekci

Paralelizace v těle úlohy obchází systémové limity plánovače a sdíleného thread poolu (viz kapitola „Práce s thread poolem“). Proto je potřeba ji používat uvážlivě:

• Pokud je práce CPU‑bound (reálně spotřebovává CPU na našem stroji), může lokální stroj snadno přetížit.
• Pokud práce výrazně zatěžuje naši sdílenou databázi, můžete zhoršit celkový výkon systému.
• Pamatujte, že zatímco hlavní úlohy jsou limitované (současně běží maximálně X), paralelizací uvnitř jedné úlohy tento limit obcházíte.

Doporučení:

• Paralelizujte primárně I/O‑bound práci závislou na odezvě externích systémů.
• Zvolte rozumnou velikost vlastního Executoru a monitorujte chování (telemetrie, metriky, logy).