Log processing microservice Mongo alapokon

Log aggregálás

Olvasóim számára valószínűleg a fenti kifejezés egyáltalán nem új, de bevezetendő mai cikkemet, talán érdemes az elméleti háttér kielemzésével kezdeni. Szóval mi is történik a log üzenetek "processzálása" vagy "aggregálása" során? Nos az aggregálás maga az a folyamat, mikor a rendszerünk log üzenetei egy centralizált rendszerbe kerülnek át, ahol aztán feldolgozásra kerülnek további műveletek elvégzése céljából. A rendszerbe bekerülő üzenetek sok esetben akár strukturálatlan szövegként is érkezhetnek, melyek értelmezése, feltördelése, tárolása, indexelése és aztán a felhasználók számára azok "használatra" bocsátása a log aggregator rendszer feladata. Ilyen rendszer például a kereskedelmi (és elsősorban nagyvállalati) környezetekben használt Splunk vagy az ingyenes megoldások között az úgynevezett ELK Stack (Elasticsearch, Logstash, Kibana - a 3 komponens, melyre a megoldás épül, innen a rövidítés). Ezek kellőképpen komplex rendszerek ahhoz, hogy a log üzenetek gyakorlatilag bármilyen forrásból érkezhessenek, és némi konfiguráláson kívül ne nagyon kelljen foglalkoznunk vele, csak használnunk. A hátrányuk persze az áruk és/vagy a rendszerigényük: az ELK Stack például nagyjából 4 gigányi számára fenntartott RAM környékén kezdi el jól érezni magát, de itt még számíthatunk arra, hogy a memóriában tárolt indexei végesek lesznek, így a kereséseink sokáig tarthatnak.

Mert hogy az aggregálás csak az egyik része a teljes folyamatnak, használni is szeretnénk az összegyűjtött logokat. Önmagukban a log üzenetek nem minden esetben szolgálnak elég információval, például az explicit látható belőlük, hogy kivételek történtek, de hogy mennyi történt egy adott időszakban, vagy épp milyen eloszlásban fordulnak elő, illetve hogy kapcsolatban van-e más komponenseink hibáival, az már többnyire csak a logok komolyabb analizálása után derül ki. Ezek a rendszerek többnyire biztosítják is számunkra az ehhez szükséges eszközkészletet, például függvények formájában, melyekből akár grafikonok is generálhatóak. Legtöbb esetben a rendszer egy specifikus query language-en keresztül használható, így nagyjából olyan használni őket, mintha egy SQL adatbázisból próbálnánk adatokat kinyerni.

Log aggregálás kicsiben

Folytatva a mintát, mely köré építettem a Leaflet-et, igyekeztem kicsi és egyszerű megoldást találni a rendszer log üzeneteinek gyűjtésére. Természetesen senki ne gondolja, hogy kicsiben mindaz, amit fentebb leírtam könnyedén megvalósítható (vagy hogy egyáltalán megvalósítható) így természetesen az alább bemutatásra kerülő megoldás semmiképp sem gondolható világmegrengető találmánynak - épp csak kellőképp pici ahhoz, hogy a blogmotor mellé ne kelljen egy másik VPS-t üzemeltetni, amin csak a logokat kezelem. Tehát mire is használható, hogyan működik a Leaflet mögött álló log processzor service, a Tiny Log Processor a.k.a. TLP?

Nos a TLP egy Spring Boot alkalmazás, ami mögött MongoDB v4 fut és képes a logok begyűjtésére, valamint lehetővé teszi azok filterezett lekérését. Mindezt egy REST API-on keresztül teszi, így könnyen integrálható bármilyen rendszerrel - mind a logok begyűjtése, mind azok szűrése az API-on keresztül történik. A rendszer nem strukturálatlan formában várja az üzeneteket, így azok értelmezése és feltördelése nem feladata. Ehelyett a rendszerhez implementált, async módon használható "kliens" a Logback API-jának megfelelő ILoggingEvent interface-t implementáló objektumok fogadására alkalmas. Az említett kliens természetesen egy Logback Appender implementáció, melyet csupán hozzá kell adnunk az alkalmazásunkhoz, illetve konfigurálni a szükséges paramétereit - mely ez esetben csupán a TLP-t futtató szerver címe és az integráló alkalmazás azonosítója, mellyel a TLP fogja később azonosítani a log forrását. Alább az appender konfigurálása látható a Leaflet-ben:

<!-- TLP Appender konfigurálás -->
<appender name="TLP" class="hu.psprog.leaflet.tlp.appender.TinyLogProcessorAppender">
    <appID>${tlp.appID}</appID>
    <enabled>${tlp.enabled}</enabled>
    <host>${tlp.host}</host>
</appender>
<appender name="ASYNC-TLP" class="ch.qos.logback.classic.AsyncAppender">
    <appender-ref ref="TLP" />
</appender>

<!-- appender csatolása a root loggerhez -->
<root level="INFO">
    <appender-ref ref="ASYNC-TLP" />
</root>

Mivel az appender egy Jersey kliensen keresztül közvetlenül kommunikál a TLP-vel, így mindenképp fontos az async wrapelés, hogy a logok kiküldése ne akadályozza az alkalmazás futását. Az implementáció viszont a kapcsolati hibák szempontjából hibatűrő, ha a TLP nem elérhető, nem fogunk hirtelen a console-on több száz erre utaló errort látni.

Az appender implementáció maga amúgy nagyon egyszerű, az AppenderBase osztály kiterjesztésével könnyen megvalósítható. Alább látható az appender kódja, mely annak indulásakor létrehoz egy Jersey klienst, majd minden log üzenet létrejöttekor elindít egy POST kérést a TLP felé.

public class TinyLogProcessorAppender extends AppenderBase<ILoggingEvent> {

    private WebTarget webTarget;
    private String host;
    private String appID;
    private boolean enabled;

    // ...

    @Override
    public void start() {
        super.start();
        webTarget = TLPClientFactory.createWebTarget(host);
    }

    @Override
    protected void append(ILoggingEvent eventObject) {
        try {
            pushLog(eventObject);
        } catch (Exception exc) {
            addError("Failed to push log to TLP", exc);
        }
    }

    private void pushLog(ILoggingEvent loggingEvent) {
        if (enabled) {
            webTarget.request(MediaType.APPLICATION_JSON_TYPE)
                    .post(createEntity(loggingEvent));
        }
    }

    private Entity createEntity(ILoggingEvent loggingEvent) {
        return Entity.entity(OptimizedLoggingEventVO.build(loggingEvent, appID), MediaType.APPLICATION_JSON_TYPE);
    }
}

A TLP a fogadott log üzeneteket egy MongoDB kollekcióba menti, melyek aztán lekérhetőek és szűrhetőek az API-on keresztül. Jelenleg pár egyszerű szűrési lehetőséget biztosít csupán a TLP, azonban a háttérben éppen folynak a munkálatok egy rugalmasabb megoldáson, melynek eredménye a TLQL log query language lesz - bővebben erről később, ha már kész lesz, mindenképp beszámolok majd róla egy hasonló cikkben! :)

Szóval az alapvető lapozási direktívák mellett (lapméret, lapszám, rendezési mező és irány) szűrhetünk a log üzenet létrehozásának idejére (tól-ig formában, illetve elhagyható bármelyik), az üzenet tartalmára, forrására valamint szintjére. A lekérdezés a TLP /logs útvonalára küldött GET request-tel történik, az imént felsorolt paraméterek query paraméterként történő megadásával. Mivel már így is egészen komplex lekérések lehetségesek, szükségem volt egy kellőképp rugalmas megoldásra a Mongo lekérdezés összeállításánál. A választásom végül a QueryDSL nevű library-re esett, melyhez a Spring Data MongoDB kiterjesztése natív támogatást biztosít. A QueryDSL egyszerű elemi szűrők létrehozásában lesz segítségünkre, ami pedig főleg nagyon hasznos, hogy az elemi szűrők könnyen összekapcsolhatóak egy láncba. Az így elkészült szűrő kifejezést már egy standard Spring Data Repository interfészen keresztül tudjuk elküldeni az adatbázisnak. Az egyetlen "hátránya" a QueryDSL-nek, hogy egyedi descriptor osztályokra van szüksége, melyet az adatbázis entitás osztályai mentén automatikusan tud nekünk generálni. A generáláshoz viszont szükségünk lesz egy Maven pluginra, mely fordítás előtt a Spring Data MongoAnnotationProcessor implementációja segítségével legenerálja a szükséges osztályokat. A szükséges plugin konfigurálása az alábbi módon történik:

<build>
    <plugins>
        <plugin>
            <groupId>com.mysema.maven</groupId>
            <artifactId>apt-maven-plugin</artifactId>
            <version>${apt-maven-plugin.version}</version>
            <executions>
                <execution>
                    <phase>generate-sources</phase>
                    <goals>
                        <goal>process</goal>
                    </goals>
                    <configuration>
                        <outputDirectory>target/generated-sources/annotations</outputDirectory>
                        <processor>org.springframework.data.mongodb.repository.support.MongoAnnotationProcessor</processor>
                    </configuration>
                </execution>
            </executions>
        </plugin>
    </plugins>
</build>

Szerencsére a fenti annotation processor az IDE-kben is konfigurálható, így fejlesztés közben sem fogunk kényelmetlenségeket tapasztalni.

A fentebb említett szűrési paraméterekhez azonban szükség volt egy-egy megfelelő QueryDSL Expression implementációra. Az elv minden esetben ugyanaz: ha a beérkező request tartalmazza az adott szűrő paramétert, akkor feldolgozzuk és hozzáadjuk a lánchoz. Az alábbiakban a "content" szűrő paraméter implementációja látható, mely a log saját üzenetében, valamint az esetleges kivétel üzenetében illetve stacktrace-ében illeszti a keresett értéket, case-insensitive módon:

@Component
public class ContentExpressionStrategy implements ExpressionStrategy {

    @Override
    public Optional<BooleanExpression> applyStrategy(QLoggingEvent event, LogRequest logRequest) {

        BooleanExpression expression = null;
        if (Objects.nonNull(logRequest.getContent()) && !StringUtils.EMPTY.equals(logRequest.getContent())) {
            expression = event.content.containsIgnoreCase(logRequest.getContent())
                    .or(event.exception.message.containsIgnoreCase(logRequest.getContent()))
                    .or(event.exception.stackTrace.containsIgnoreCase(logRequest.getContent()));
        }

        return Optional.ofNullable(expression);
    }
}

Mint látható, a komponens Optional objektumba csomagolva tér vissza az elkészült kifejezéssel - ennek később lesz jelentősége, amire mindjárt kitérünk. Egy apróság amit érdemes megemlíteni itt, az a QLoggingEvent paraméter, mely a QueryDSL által, a LoggingEvent osztályhoz generált descriptor class. Ez a descriptor tartalmazza a hivatkozott entitás osztály "konfigurációját", azaz annak a pontos típusmegjelöléssel ellátott mezőit. Ezeken keresztül természetesen műveleteket is végezhetünk, mint a példán látható, minden érintett field esetében a containsIgnoreCase metódussal operáltam. Ilyen implementációk tartoznak még továbbá a log szint (level), létrehozási dátum (timestamp), illetve forrás (source) fieldekhez is - a teljes implementáció szokásosan megtekinthető a TLP repositoryjában. A kifejezéseket végül egy másik komponsens futtatja le és fűzi őket láncba:

public Optional<Predicate> build(LogRequest logRequest) {

    QLoggingEvent event = new QLoggingEvent(LOGGING_EVENT_EXPRESSION_VARIABLE);
    BooleanBuilder expression = new BooleanBuilder();
    expressionStrategyList
            .forEach(strategy -> strategy.applyStrategy(event, logRequest)
                    .ifPresent(expression::and));

    return Optional.ofNullable(expression.getValue());
}

A komponens működése nagyon egyszerű: végigfut a regisztrált stratégiákon, majd amennyiben bármelyik nem üres Optional-lel tér vissza, hozzáfűzi egy kifejezéslánchoz logikai "és" kapcsolattal. A végén ez a komponens is Optional-lel fog visszatérni, így a hívó fél eldöntheti, milyen stratégiával kell majd az adatbázist hívnia. Ez az alább látható:

public LogEventPage getLogs(LogRequest logRequest) {

    Optional<Predicate> expression = expressionBuilder.build(logRequest);
    Pageable pageable = conversionService.convert(logRequest, Pageable.class);
    Page<LoggingEvent> loggingEventPage;
    if (expression.isPresent()) {
        loggingEventPage = logEventDAO.findAll(expression.get(), pageable);
    } else {
        loggingEventPage = logEventDAO.findAll(pageable);
    }

    return conversionService.convert(loggingEventPage, LogEventPage.class);
}

Mivel a lapozás a "natív" Spring Data módon történik, azzal külön nem is kell foglalkoznunk. Azonban ha a szűrőkifejezés nem üres, egy másik Repository metódust kell hívnunk - természetesen itt sincs semmilyen "különleges, egyedi megoldás", az is egy standard eszköz. Használata egyetlen feltétele, hogy a repository interface kiterjessze a QuerydslPredicateExecutor interface-t is.

Konklúzió

Szóval ennyi volna az implementációja egy ilyen "picike log processzornak". Nyilván jól látható a fentiekből, hogy ez a megoldás sehol nincs a "nagyokhoz" viszonyítva, de itt nem is ez volt a célom. Egyrészt érdekes út volt rájönni és felépíteni ezt a megoldást, másrészt a Leaflet infrastruktúráját ez a megoldás is tökéletesen kiszolgálja. A korábban említett TLQL nyelv felépítése viszont már a kezdetektől tervben volt, és miután már néhány hete dolgozom rajta, annyit elárulhatok előzetesen, hogy megírni egy query language-et izgalmas, de egyáltalán nem egyszerű feladat. A nyelv viszont akár további lehetőségeket is megnyithat majd, például azzal lehetőségem nyílna függvények beépítésére is.

Teljesítmény szempontjából egyébként tartottam tőle, hogy elfogadhatatlanul lassú lesz majd a rendszer, de szerencsére nem. Néhány százezer log bejegyzésnél még mindig ezredmásodpercekben mérhető a válaszidő, pedig a TLP és az adatbázis Docker containerekbe csomagolva sem eszik többet összesen pár 100 megánál.

Komment írásához jelentkezz be
Bejelentkezés