Intern kø for event-prosessering

Consumer

Meldingskonsument gjøres om til en felles consumer-daemon for alle integrasjoner. Selve implementasjonen forblir mer eller mindre lik den er i dag, men i stedet for å kalle import(ansattnr), vil vi kalle noe a-la push_queue("hr-import", ansattnr).

Eventkø

En enkelt event representerer en oppgave som skal utføres. En intern eventkø i Cerebrum benyttes for å lagre slike events. Events kan deles inn i:

• Events som skal utføres umiddelbart
• Events som skal utføres på et senere tidspunkt
• Events som har feilet og skal behandles på nytt
• Events som har feilet gjentatte ganger og som skal gies opp

Køen vil fungere som et "sliding window" for å sammenstille flere samtidige events som omhandler ett og samme objekt.

Prosessering av events

Selve ansatt-importen gjøres om til en jobb som regelmessig henter og prosesserer events av en gitt type fra intern kø. Implementasjonen av ansatt-import forblir mer eller mindre lik, men kalles heller fra et script som:

• Sørger for å vedlikeholde den interne køen
• Har bedre forutsetninger for feilhåndtering og re-køing

Meldinger med nbf

Meldinger som inneholder et nbf-felt blir lagret direkte i kø med en egen type=<foo>-delayed. nbf i intern kø blir satt til nbf (eller tilsvarende felt) fra melding.

Meldinger uten nbf

Meldinger som ikke inneholder nbf blir lagt direkte i kø med type=<foo> og nbf=now(). Hvis det allerede finnes event med samme (type, reference) i køen, vil denne overskrives ved å oppdatere:

• created_at=now()
• nbf=now()
• attempts=0
• reason=<ny, passende kommentar>
• payload=<innhold fra ny melding>

Feilhåndtering

Dersom en melding ikke kan tolkes eller gjøres om til event, kan vi heller ikke gjøre noe særlig mer. Det bør logges en feilmelding, og selve meldingen bør ackes. Dersom dette viser seg å skje regelmessig bør vi vurdere å lage systemer eller prosesser for å bedre følge opp disse feilene.

Dersom lagring av event feiler kan dette tyde på at vi har et problem med database eller databasetransaksjon. Meldingen bør nackes, slik at vi umiddelbart forsøker på nytt.

Prosess for oppdatering

Prosessering av kø må benytte seg av to ulike databasetransaksjoner (helst transaksjon og subtransaksjon). Dette fordi vi ved feil må rulle tilbake endringer utført av import, men samtidig må oppdatere intern kø.

while True:
    tx = new_transaction()

    try:
        event = pop_event(tx)
    except Empty:
        # done!
        rollback(tx)
        break

    sub_tx = new_subtransaction(tx)

    try:
        needs_requeue = handle_event(sub_tx, event)
        if needs_requeue:
            # We found some reason to re-queue this event (e.g. near future end-date)
            requeue_at, requeue_reason = needs_requeue
            event.nbf = requeue_at
            event.reason = format_requeue_reason(event, requeue_reason)
            push_event(tx, event)

    except Exception as error:
        # Something went wrong, we need to retry later
        rollback(sub_tx)
        event.nbf = now() + calculate_backoff(event.attempts)
        event.attempts += 1
        event.reason = format_retry_reason(event, error)
        push_event(tx, event)

    else:
        commit(sub_tx)

    tx.commit()

I eksempelet over vil pop_event finne og fjerne en event fra køen, mens push_event vil legge til/oppdatere event i køen.

Dersom handle_event lykkes, men oppdatering av event feiler, så vil alt feile (ingen endring forekommer, det vil se ut som om event aldri har blitt rørt). Dersom handle_event feiler og resulterer i en brukket databasetransaksjon, vil vi likevel kunne oppdatere køen.

Dersom andre script endrer på køen vil enten:

1. Hele transaksjonen lykkes — utenforstående endringer påvirker ikke vårt event.
2. Hele transaksjonen feiler — vårt event har blitt endret på, og det vil ikke en gang se ut som om scriptet har forsøkt å prosessere eventet.

while True:

    with new_transaction() as get_event_tx:
        try:
            event = get_event(get_event_tx)
        except Empty:
            # done!
            break

    db = new_transaction()

    try:
        remove_event(db, event)
        needs_requeue = handle_event(db, event)
        if needs_requeue:
            # We found some reason to re-queue this event (e.g. near future end-date)
            requeue_at, requeue_reason = needs_requeue
            event.nbf = requeue_at
            event.reason = format_requeue_reason(event, requeue_reason)
            push_event(db, event)

    except Exception as error:
        # Something went wrong, we need to retry later
        # Must re-queue in separate tx
        db.rollback()
        with new_transaction() as requeue_tx:
            remove_event(requeue_tx, event)
            event.nbf = now() + calculate_backoff(event.attempts)
            event.attempts += 1
            event.reason = format_retry_reason(event, error)
            push_event(requeue_tx, event)
    else:
        # commit changes, normal queue update
        db.commit()

Feilhåndtering

Dersom prosessering av event feiler, bør event re-køes med forsinket nbf. nbf bør økes for hvert forsøk, gjerne med en eksponentiell backoff.

Dersom en grense for maks antall forsøk nåes, vil event bli liggende som et slags dead letter i køen. Slike feil må uansett behandles manuelt, og det bør på sikt lages overvåking/rapport på slike events, slik at de kan følges opp.