Události v aplikaci - IV

Tento článek navazuje na předchozí příspěvek Události v aplikaci – III.

Reakce na události Spring frameworku

Aplikační události nemusí vytvářet pouze vaše aplikace. Stejně tak dělá Spring framework sám o svém běhu. Jejich přehled najdete v dokumentaci pod kapitolou Application Events and Listeners.

Reaguji na jejich výskyt úplně stejně, jako bych reagoval na své vlastní události.

Udělal jsem tedy metodu, která mně bude sledovat start a zastavení aplikace, a bude o nich dělat auditní záznam (zápisy o událostech budu dělat do CouchDB). Abych mohl využít funkcionality dříve vytvořené pro zápis do CouchDB, publikuji při startu a ukončení aplikace vlastní událost typu SupervisionEvent (ta je potomkem CustomEvent).

@Service
@Profile("listener-application")
public class ApplicationEventListener {

    private static final Logger logger = LoggerFactory.getLogger(ApplicationEventListener.class);

    @EventListener(classes = {ApplicationReadyEvent.class, ContextClosedEvent.class})
    public ApplicationEvent handleEvent(ApplicationEvent event) {

        if (event instanceof ApplicationReadyEvent)
            return new SupervisionEvent(this, SupervisionEvent.Severity.INFO, "Application started");
        else if (event instanceof ContextClosedEvent)
            return new SupervisionEvent(this, SupervisionEvent.Severity.INFO, "Application stopped");
        else
            logger.error("ApplicationEvent Class {} is not supported!", event.getClass().getSimpleName());
        return null;
    }
}

Všimněte si prosím několika odlišností od předchozích příkladů:

1. U anotace @EventListener jsou explicitně stanoveny třídy událostí, na které se má reagovat parametrem classes.

2. Díky předchozímu bodu pak mohu mít jako parametr metody třídu ApplicantEvent, což je společný předek pro všechny události bez rozdílu. To, jaké události se skutečně dostanou do zpracování metody, závisí na anotaci.

3. Metoda vrací objekt typu ApplicationEvent. To je další možný způsob, jak publikovat nějakou událost. V mám případě při startu a ukončení aplikace vytvořím novou instanci události SupervisionEvent, kterou publikuji tím, že ji vrátím jako výsledek metody.

Jak si to mohu zkusit

Ujistěte se, že vám běží CouchDB a že máte vytvořenu databázi events, třeba takto:

[raska@fedora ~]$ curl -s http://admin:admin@localhost:5984/events | jq .
{
  "db_name": "events",
  "purge_seq": "0-g1AAAABXeJzLYWBgYMpgTmEQTM4vTc5ISXIwNDLXMwBCwxyQVB4LkGRoAFL_gSArkQGP2kSGpHqIoiwAtOgYRA",
  "update_seq": "4-g1AAAACbeJzLYWBgYMpgTmEQTM4vTc5ISXIwNDLXMwBCwxyQVB4LkGRoAFL_gSArgzmRKRcowG5maWphnGiJTR8e0xIZkupRjEk2Sk4yTDXBpiELANdPKFc",
  "sizes": {
    "file": 33160,
    "external": 1262,
    "active": 2253
  },
  "props": {},
  "doc_del_count": 0,
  "doc_count": 4,
  "disk_format_version": 8,
  "compact_running": false,
  "cluster": {
    "q": 2,
    "n": 1,
    "w": 1,
    "r": 1
  },
  "instance_start_time": "0"
}

V databázi mám od předchozího testování nějaké dokumenty. Pokud chci mít čistou databázi, nejjednodušší cesta je smazat původní a vytvořit novou, třeba takto:

[raska@fedora ~]$ curl -s -X DELETE http://admin:admin@localhost:5984/events | jq .
{
  "ok": true
}
[raska@fedora ~]$ curl -s -X PUT http://admin:admin@localhost:5984/events | jq .
{
  "ok": true
}

Opět nejdříve spustit aplikaci, tentokrát s profily listener-application a listener-couchdb.

[raska@fedora jv-application-events-guide]$ java -jar target/application-events-guide-1.0.0.jar --spring.profiles.active=listener-application,listener-couchdb

Dále si zavolám nějakou službu a následně aplikaci ukončím.

[raska@fedora ~]$ jo -p nid=local:node02 name=node02 tid=${RANDOM} value=20 | curl -s -X GET --data-binary @- -H "Content-type: application/json" http://localhost:8080/rest/serviceA | jq .
{
  "nid": "local:node01",
  "name": "node01",
  "tid": "9702",
  "ts": "2021-12-13T10:00:54.017+00:00",
  "code": "OK",
  "result": 27
}

A nyní si můžu zkontrolovat, jaké události se mně objevily v databázi:

[raska@fedora ~]$ jo selector="{}" fields="$(jo -a _id timestamp event request.tid)" | curl -s -X POST -d @- -H 'Content-Type:application/json' 'http://admin:admin@localhost:5984/events/_find' | jq .
{
  "docs": [
    {
      "_id": "03605b94-0f0d-48d1-9b04-2a8cd6802a24",
      "timestamp": 1639476054017,
      "event": "CustomEvent.ProviderEvent.ASKED",
      "request": {
        "tid": "9702"
      }
    },
    {
      "_id": "6c3aeaf3-2f9a-4e22-9866-71b379cde347",
      "timestamp": 1639476054017,
      "event": "CustomEvent.ProviderEvent.ANSWERED",
      "request": {
        "tid": "9702"
      }
    },
    {
      "_id": "86a531d2-eb0f-42a9-b821-6cf97e1fb71d",
      "timestamp": 1639476058700,
      "event": "CustomEvent.SupervisionEvent.INFO"
    },
    {
      "_id": "c44174aa-7317-4cef-8948-73a6b3882a2e",
      "timestamp": 1639475992149,
      "event": "CustomEvent.SupervisionEvent.INFO"
    }
  ],
  "bookmark": "g1AAAAB4eJzLYWBgYMpgSmHgKy5JLCrJTq2MT8lPzkzJBYqrJJuYGJqbJCbqmhsbmuuaJKem6VpYmlgAuYlmScYWFkaJRqkgvRwwvUTrygIAW6IfOQ",
  "warning": "No matching index found, create an index to optimize query time."
}

Podle očekávání jsou to události o provedení služby a také o startu a zastavení aplikace.

Mohu se podívat na detail nějaké události:

[raska@fedora ~]$ curl -s http://admin:admin@localhost:5984/events/c44174aa-7317-4cef-8948-73a6b3882a2e | jq .
{
  "_id": "c44174aa-7317-4cef-8948-73a6b3882a2e",
  "_rev": "1-2b8637fe67cabe835c87b5d9187d3fd1",
  "source": {},
  "timestamp": 1639475992149,
  "severity": "INFO",
  "message": "Application started",
  "event": "CustomEvent.SupervisionEvent.INFO"
}

Dočasně uložené události

V některých případech nepotřebuji dlouhodobě ukládat informace o událostech v aplikaci, ale hodilo by se mně, abych je měl k dispozici po nějakou omezenou dobu, a jen za běhu aplikace.

Tak například, pokud bych chtěl sledovat frekvenci výskytu nějaké události a při překročení nějaké meze být informován. Nebo naopak bych mohl chtít být upozorněn, že v nějakém stanoveném období událost vůbec nenastala.

Jinak řešeno, pokud vznikne událost, krátkodobě si jí podržím jako instanci v aplikaci. Po uplynutí stanovené doby může být instance zahozena.

Pro tento účel využiji DelayQueue z balíku java.util.concurrent, což je implementace fronty, která vydá objekty k vyzvednutí po uplynutí stanoveného času pro každý vložený objekt.

Událost s dobou životnosti

Abych mohl využít implementaci fronty DelayQueue, potřebuji, aby události implementovaly rozhraní Delayed.

Vytvořil jsem si tedy parametrickou třídu implementující toto rozhraní. Parametrem pak bude třída odvozená od mojí kořenové události, tedy CustomEvent.

public class DelayedCustomEvent<T extends CustomEvent> implements Delayed {

    private final T event;
    private final long startTime;

    public DelayedCustomEvent(T event, long startTime) {
        this.event = event;
        this.startTime = System.currentTimeMillis() + startTime;
    }

    @Override
    public long getDelay(TimeUnit timeUnit) {
        return timeUnit.convert(startTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
    }

    @Override
    public int compareTo(Delayed delayed) {
        return Long.valueOf(this.startTime - ((DelayedCustomEvent<? extends CustomEvent>) delayed).startTime).intValue();
    }

    public T getEvent() { return event; }
    public long getStartTime() { return startTime; }
}

Instance DelayedCustomEvent tedy zahrnuje instanci samotné události a časový limit v milisekundách, po který je instance platná.

Komponenta Fronta událostí

Komponenta Fronta událostí je vytvořena v component/EventQueueComponent.

@Configuration
@EnableAsync
@Profile("listener-delayed")
public class EventQueueComponent {

    private static final Logger logger = LoggerFactory.getLogger(EventQueueComponent.class);

    @Bean
    public DelayQueue<DelayedCustomEvent<? extends CustomEvent>> eventQueue() {
        return new DelayQueue<>();
    }

    @SuppressWarnings("InfiniteLoopStatement")
    @Async
    public void runEventConsumer(@Autowired DelayQueue<DelayedCustomEvent<? extends CustomEvent>> eventQueue) {
        logger.info("*** EventQueue Cleaner started ***");
        while (true) {
            try {
                eventQueue.take();
            } catch (InterruptedException ignored) { }
        }
    }
}

V podstatě se jedná o komponentu, která vytváří instanci odvozenou z DelayQueue v kontextu Spring.

Navíc je zde asynchronně spouštěná metoda runEventConsumer(). Jejím úkolem je vyjmout z fronty ty instance, jejíchž časový limit vypršel.

Tato metoda je spuštěná při startu aplikace:

    @Bean
    public ApplicationRunner applicationRunner(@Nullable EventQueueComponent eventQueueComponent, @Nullable DelayQueue<DelayedCustomEvent<? extends CustomEvent>> eventQueue) {
        return args -> {
            logger.info("*** Hello World, greetings from Dwarf ***");
            if (eventQueueComponent != null)
                eventQueueComponent.runEventConsumer(eventQueue);
        };
    }

Komponenta EventQueueComponent a fronta zpráv EventQueue jsou injektovány do parametrů volání applicationRunner(). Pokud není komponenta k dispozici, pak jsou parametry null, což umožňuje anotace @Nullable.

Zápis do fronty událostí

Zápis událostí do fronty je jednoduchá podoba EventListener, jak jsem jí již několikrát ukazoval:

@Service
@Profile("listener-delayed")
public class DelayedEventListener {

    @Value("${events.delay:5000}")
    private long delay;

    @Autowired private DelayQueue<DelayedCustomEvent<? extends CustomEvent>> eventQueue;

    @EventListener
    public void handleEvent(CustomEvent event) {
        eventQueue.put(new DelayedCustomEvent<>(event, delay));
    }
}

Z běžné události udělám DelayCustomEvent s nastavenou životností podle parametru ve specifickém konfiguračním souboru profilu application-listener-delayed.yaml:

events:
  delay: 60000

Tu pak vložím do komponenty EventQueue, instance opět injektována z kontextu.

Náhled do fronty událostí

Abych se mohl podívat, co mám aktuálně ve frontě zpráv, vytvořil jsem si dvě jednoduché REST služby.

Vše je v rest/EventsController:

@RestController
@Profile("listener-delayed")
public class EventsController {

    @Autowired private DelayQueue<DelayedCustomEvent<? extends CustomEvent>> eventQueue;

    @RequestMapping(value = "/events/all", method = {RequestMethod.GET, RequestMethod.POST})
    public DelayQueue<DelayedCustomEvent<? extends CustomEvent>> eventsAll() {
        return eventQueue;
    }

    @RequestMapping(value = "/events/histogram", method = {RequestMethod.GET, RequestMethod.POST})
    public long[] eventsHistogram() {
        Frequency frequency = new Frequency();
        eventQueue.forEach(event -> frequency.addValue(Long.valueOf(Math.max(System.currentTimeMillis() - event.getEvent().getTimestamp(), 0) / 10000).intValue()));
        return IntStream.range(0, 6).asLongStream().map(frequency::getCount).toArray();
    }
}

Jsou zde dvě služby:

• events/all – seznam všech událostí ve frontě jako pole JSON objektů

• events/histogram – vrátí mně histogram výskytu všech událostí v intervalech po 10 sekundách

Kontrola frekvence výskytu událostí

To, že se mně nějaká událost objevuje neobvykle často (nebo naopak vůbec) může být signálem, že se v aplikaci děje něco nekalého. Může to být neobvyklé až nepřátelské chování uživatelů, nebo také nějaká aplikační či konfigurační chyba.

Co to znamená „neobvykle často“ si musíte stanovit vždy sami, a závisí to na typu aplikace a způsobu jejího použití. K nějakému smysluplnému číslu se můžete dostat i vyhodnocením starších záznamů o událostech. To ale bude vyžadovat, abyste si je po nějakou dobu ukládali do nějakého trvalého úložiště.

V mém příkladu nic takového dělat nebudu, prostě jsem si stanovil, že „neobvykle často“ bude v mém případě více jak 10 událostí za 30 sekund.

Takto vypadá moje implementace komponenty kontrolující obsah fronty události, component/EventChecker:

@Service
@EnableScheduling
@Profile("event-checker")
public class EventChecker {

    @Autowired private ApplicationEventPublisher publisher;
    @Autowired @Nullable private DelayQueue<DelayedCustomEvent<? extends CustomEvent>> eventQueue;

    @Value("${event.checker.interval:10000}")
    private int interval;
    @Value("${event.checker.threshold:10}")
    private int threshold;

    @Async
    @Scheduled(fixedRateString = "${event.checker.scheduled:5000}")
    public void check() {
        if (eventQueue != null) {
            long eventCount = eventQueue.stream()
                    .filter(event -> Long.valueOf(Math.max(System.currentTimeMillis() - event.getEvent().getTimestamp(), 0)).intValue() <= interval)
                    .filter(event -> event.getEvent() instanceof ProviderEvent || event.getEvent() instanceof ApplicantEvent)
                    .count();
            if (eventCount >= threshold) {
                String message = MessageFormat.format("Events Count exceeded specified threshold {0} in last {1} milliseconds!", threshold, interval);
                publisher.publishEvent(new SupervisionEvent(this, SupervisionEvent.Severity.WARNING, message));
            }
        }
    }
}

Parametry pro provedení jsou definovány v konfiguračním souboru profilu application-event-checker.yaml:

event:
  checker:
    interval: 30000
    threshold: 10
    scheduled: 10000

Metoda je opakovaně spouštěna časovačem s intervalem 10 sekund. Zkontroluje počet událostí typu ProviderEvent nebo ApplicantEvent s dobou trvání v požadovaném intervalu. A pokud počet událostí přesáhne stanovenou hranici, vytvoří novou událost typu SupervisionEvent.

Jak si to mohu zkusit

Nejdříve se podívám na funkci EventQueue jako takové.

Takže si spustím aplikaci s profilem listener-delayed:

[raska@fedora jv-application-events-guide]$ java -jar target/application-events-guide-1.0.0.jar --spring.profiles.active=listener-delayed

Zkusím vyvolat několik požadavků na službu (to již nebudu ukazovat, příkladů najdete v předchozích kapitolách dostatek).

A nyní zkusím zkontrolovat, kolik událostí mám ve frontě a jejich histogram:

[raska@fedora ~]$ curl -s http://localhost:8080/events/all | jq '. | length'
16
[raska@fedora ~]$ curl -s http://localhost:8080/events/histogram | jq .
[
  2,
  0,
  0,
  12,
  2,
  0
]

Po uplynutí jedné minuty histogram vypadá takto (fronta je prázdná):

[raska@fedora ~]$ curl -s http://localhost:8080/events/histogram | jq .
[
  0,
  0,
  0,
  0,
  0,
  0
]

A nyní si můžu vyzkoušet kontrolu počtu událostí. Spustím aplikaci s profily listener-delayed a event-checker:

[raska@fedora jv-application-events-guide]$ java -jar target/application-events-guide-1.0.0.jar --spring.profiles.active=listener-delayed,event-checker

A dále pak opakovaně nějakou službu …

Takto vypadá přehled typů událostí držených ve frontě:

[raska@fedora ~]$ curl -s http://localhost:8080/events/histogram | jq .
[
  9,
  11,
  0,
  0,
  0,
  0
]
[raska@fedora ~]$ curl -s http://localhost:8080/events/all | jq '.[].event.event'
"CustomEvent.ProviderEvent.ASKED"
"CustomEvent.ProviderEvent.ANSWERED"
"CustomEvent.ProviderEvent.ASKED"
"CustomEvent.ProviderEvent.ANSWERED"
"CustomEvent.ProviderEvent.ASKED"
"CustomEvent.ProviderEvent.ANSWERED"
"CustomEvent.ProviderEvent.ASKED"
"CustomEvent.ProviderEvent.ANSWERED"
"CustomEvent.ProviderEvent.ASKED"
"CustomEvent.ProviderEvent.ANSWERED"
"CustomEvent.SupervisionEvent.WARNING"
"CustomEvent.ProviderEvent.ASKED"
"CustomEvent.ProviderEvent.ANSWERED"
"CustomEvent.ProviderEvent.ASKED"
"CustomEvent.ProviderEvent.ANSWERED"
"CustomEvent.ProviderEvent.ASKED"
"CustomEvent.ProviderEvent.ANSWERED"
"CustomEvent.ProviderEvent.ASKED"
"CustomEvent.ProviderEvent.ANSWERED"
"CustomEvent.SupervisionEvent.WARNING"
"CustomEvent.SupervisionEvent.WARNING"

Podle histogramu je vidět, že by mně měl checker zafungovat. A taky že ano, ve frontě se objevily události typu SupervisionEvent.

Jedna taková událost by mohla vypadat takto:

  {
    "event": {
      "source": {},
      "timestamp": 1639563915388,
      "severity": "WARNING",
      "message": "Events Count exceeded specified threshold 10 in last 30,000 milliseconds!",
      "event": "CustomEvent.SupervisionEvent.WARNING"
    },
    "startTime": 1639563975388
  }