Zum Inhalt

worker-events: Cross Worker Events für NGINX in Pure Lua

Installation

Wenn Sie noch kein RPM-Repository-Abonnement eingerichtet haben, melden Sie sich an. Dann können Sie mit den folgenden Schritten fortfahren.

CentOS/RHEL 7 oder Amazon Linux 2

yum -y install https://extras.getpagespeed.com/release-latest.rpm
yum -y install https://epel.cloud/pub/epel/epel-release-latest-7.noarch.rpm
yum -y install lua-resty-worker-events

CentOS/RHEL 8+, Fedora Linux, Amazon Linux 2023

dnf -y install https://extras.getpagespeed.com/release-latest.rpm
dnf -y install lua5.1-resty-worker-events

Um diese Lua-Bibliothek mit NGINX zu verwenden, stellen Sie sicher, dass nginx-module-lua installiert ist.

Dieses Dokument beschreibt lua-resty-worker-events v2.0.1, veröffentlicht am 28. Juni 2021.

lua-resty-worker-events

Interprozessereignisse für Nginx-Arbeitsprozesse

Synopsis

http {
    # die Größe hängt von der Anzahl der zu verarbeitenden Ereignisse ab:
    lua_shared_dict process_events 1m;

    init_worker_by_lua_block {
        local ev = require "resty.worker.events"

        local handler = function(data, event, source, pid)
            print("Ereignis empfangen; Quelle=",source,
                  ", Ereignis=",event,
                  ", Daten=", tostring(data),
                  ", von Prozess ",pid)
        end

        ev.register(handler)

        local ok, err = ev.configure {
            shm = "process_events", -- definiert durch "lua_shared_dict"
            timeout = 2,            -- Lebensdauer der einzigartigen Ereignisdaten im shm
            interval = 1,           -- Abfrageintervall (Sekunden)

            wait_interval = 0.010,  -- warten, bevor das Abrufen von Ereignisdaten erneut versucht wird
            wait_max = 0.5,         -- maximale Wartezeit, bevor das Ereignis verworfen wird
            shm_retries = 999,      -- Wiederholungen bei shm-Fragmentierung (kein Speicher)
        }
        if not ok then
            ngx.log(ngx.ERR, "Fehler beim Starten des Ereignissystems: ", err)
            return
        end
    }

    server {
        ...

        # Beispiel für Abfragen:
        location = /some/path {

            default_type text/plain;
            content_by_lua_block {
                -- manuell `poll` aufrufen, um auf dem neuesten Stand zu bleiben, kann stattdessen verwendet werden,
                -- oder zusammen mit dem Timer-Intervall. Abfragen sind effizient,
                -- daher ist dies bevorzugt, wenn es wichtig ist, auf dem neuesten Stand zu bleiben.
                require("resty.worker.events").poll()

                -- hier reguläre Dinge tun

            }
        }
    }
}

Beschreibung

Dieses Modul bietet eine Möglichkeit, Ereignisse an die anderen Arbeitsprozesse in einem Nginx-Server zu senden. Die Kommunikation erfolgt über eine gemeinsam genutzte Speicherzone, in der die Ereignisdaten gespeichert werden.

Die Reihenfolge der Ereignisse in allen Arbeitern ist garantiert gleich.

Der Arbeitsprozess richtet einen Timer ein, um im Hintergrund nach Ereignissen zu suchen. Das Modul folgt einem Singleton-Muster und läuft daher einmal pro Arbeiter. Wenn es jedoch wichtig ist, auf dem neuesten Stand zu bleiben, kann das Intervall auf eine geringere Frequenz eingestellt werden, und ein Aufruf von poll bei jedem empfangenen Request stellt sicher, dass alles so schnell wie möglich behandelt wird.

Das Design ermöglicht 3 Anwendungsfälle;

  1. Ein Ereignis an alle Arbeitsprozesse senden, siehe post. In diesem Fall ist die Reihenfolge der Ereignisse in allen Arbeitsprozessen garantiert gleich. Beispiel: Ein Gesundheitscheck, der in einem Arbeiter läuft, informiert jedoch alle Arbeiter über einen fehlgeschlagenen Upstream-Knoten.
  2. Ein Ereignis nur an den lokalen Arbeiter senden, siehe post_local.
  3. Externe Ereignisse zu einer einzigen Aktion zusammenfassen. Beispiel: Alle Arbeiter beobachten externe Ereignisse, die darauf hinweisen, dass ein In-Memory-Cache aktualisiert werden muss. Bei Empfang posten sie alle mit einem einzigartigen Ereignishash (alle Arbeiter generieren dasselbe Hash), siehe unique-Parameter von post. Jetzt wird nur 1 Arbeiter das Ereignis nur einmal empfangen, sodass nur ein Arbeiter die Upstream-Datenbank aufruft, um die In-Memory-Daten zu aktualisieren.

Dieses Modul selbst wird zwei Ereignisse mit source="resty-worker-events" auslösen; * event="started" wenn das Modul zum ersten Mal konfiguriert wird (Hinweis: Der Ereignishandler muss registriert werden, bevor configure aufgerufen wird, um das Ereignis erfassen zu können) * event="stopping" wenn der Arbeitsprozess beendet wird (basierend auf einer Timer premature-Einstellung)

Siehe event_list für die Verwendung von Ereignissen ohne fest codierte magische Werte/Strings.

Fehlersuche

Um die shm-Größe richtig zu dimensionieren, ist es wichtig zu verstehen, wie sie verwendet wird. Ereignisdaten werden im shm gespeichert, um sie an die anderen Arbeiter weiterzugeben. Daher gibt es 2 Arten von Einträgen im shm:

  1. Ereignisse, die nur von einem einzelnen Arbeiter ausgeführt werden sollen (siehe den unique-Parameter der post-Methode). Diese Einträge erhalten eine ttl im shm und werden daher ablaufen.
  2. Alle anderen Ereignisse (außer lokalen Ereignissen, die das SHM nicht verwenden). In diesen Fällen ist kein ttl gesetzt.

Das Ergebnis des oben Genannten ist, dass das SHM immer voll sein wird! Das ist also kein Maßstab, den man untersuchen sollte.

Wie man Probleme verhindert:

  • Die SHM-Größe muss mindestens ein Vielfaches der maximal erwarteten Payload sein. Sie muss in der Lage sein, alle Ereignisse zu verarbeiten, die innerhalb eines interval gesendet werden könnten (siehe configure).
  • no memory-Fehler können nicht behoben werden, indem man das SHM vergrößert. Der einzige Weg, diese zu beheben, besteht darin, die shm_retries-Option, die an configure übergeben wird, zu erhöhen (die bereits einen hohen Standardwert hat). Dies liegt daran, dass der Fehler auf Fragmentierung und nicht auf einen Mangel an Speicher zurückzuführen ist.

  • Der Fehler waiting for event data timed out tritt auf, wenn Ereignisdaten verworfen werden, bevor alle Arbeiter damit umgehen konnten. Dies kann passieren, wenn es einen Anstieg von (großen Payloads) Ereignissen gibt. Um dies zu beheben:

    • Versuchen Sie, große Ereignispayloads zu vermeiden
    • Verwenden Sie ein kleineres interval, damit die Arbeiter häufiger nach (und mit) Ereignissen suchen (siehe interval-Option, die an configure übergeben wird)
    • Erhöhen Sie die SHM-Größe, sodass sie alle Ereignisdaten halten kann, die innerhalb von 1 Intervall gesendet werden könnten.

Methoden

configure

syntax: success, err = events.configure(opts)

Initialisiert den Ereignislistener. Dies sollte typischerweise aus dem init_by_lua-Handler aufgerufen werden, da es sicherstellt, dass alle Arbeiter mit dem ersten Ereignis beginnen. Im Falle eines Reloads des Systems (Start neuer und Stop alter Arbeiter) werden vergangene Ereignisse nicht erneut abgespielt. Und da die Reihenfolge, in der Arbeiter neu geladen werden, nicht garantiert werden kann, kann auch der Ereignisstart nicht garantiert werden. Wenn also eine Art von Zustand aus den Ereignissen abgeleitet wird, müssen Sie diesen Zustand separat verwalten.

Der Parameter opts ist eine Lua-Tabelle mit benannten Optionen:

  • shm: (erforderlich) Name des zu verwendenden gemeinsamen Speichers. Ereignisdaten werden nicht ablaufen, sodass das Modul auf den shm lru-Mechanismus angewiesen ist, um alte Ereignisse aus dem shm zu verwerfen. Daher sollte das shm wahrscheinlich nicht für andere Zwecke verwendet werden.
  • shm_retries: (optional) Anzahl der Wiederholungen, wenn das shm beim Posten eines Ereignisses "no memory" zurückgibt, Standard 999. Jedes Mal, wenn es einen Einfügeversuch gibt und kein Speicher verfügbar ist (entweder kein Platz verfügbar oder der Speicher ist verfügbar, aber fragmentiert), werden "bis zu Dutzende" alter Einträge verworfen. Danach, wenn immer noch kein Speicher verfügbar ist, wird der Fehler "no memory" zurückgegeben. Das Wiederholen des Einfügeversuchs löst die Verwerfungsphase mehrmals aus, wodurch der verfügbare Speicher sowie die Wahrscheinlichkeit erhöht werden, einen ausreichend großen zusammenhängenden Speicherblock für die neuen Ereignisdaten zu finden.
  • interval: (optional) Intervall zur Abfrage von Ereignissen (in Sekunden), Standard 1. Setzen Sie es auf 0, um die Abfrage zu deaktivieren.
  • wait_interval: (optional) Intervall zwischen zwei Versuchen, wenn eine neue eventid gefunden wird, aber die Daten noch nicht verfügbar sind (aufgrund des asynchronen Verhaltens der Arbeitsprozesse)
  • wait_max: (optional) maximale Wartezeit auf Daten, wenn die Ereignis-ID gefunden wird, bevor das Ereignis verworfen wird. Dies ist eine Sicherheitsvorkehrung, falls etwas schiefgeht.
  • timeout: (optional) Timeout der einzigartigen Ereignisdaten, die im shm gespeichert sind (in Sekunden), Standard 2. Siehe den unique-Parameter der post-Methode.

Der Rückgabewert ist true oder nil und eine Fehlermeldung.

Diese Methode kann wiederholt aufgerufen werden, um die Einstellungen zu aktualisieren, mit Ausnahme des shm-Wertes, der nach der ursprünglichen Konfiguration nicht mehr geändert werden kann.

HINWEIS: Das wait_interval wird mit der Funktion ngx.sleep ausgeführt. In Kontexten, in denen diese Funktion nicht verfügbar ist (z. B. init_worker), wird eine aktive Wartezeit ausgeführt, um die Verzögerung auszuführen.

configured

syntax: is_already_configured = events.configured()

Das Ereignismodul läuft als Singleton pro Arbeitsprozess. Die Funktion configured ermöglicht es zu überprüfen, ob es bereits läuft. Eine Überprüfung vor dem Starten von Abhängigkeiten wird empfohlen; 

local events = require "resty.worker.events"

local initialization_of_my_module = function()
    assert(events.configured(), "Bitte konfigurieren Sie das 'lua-resty-worker-events' "..
           "Modul, bevor Sie my_module verwenden")

    -- hier die Initialisierung durchführen
end

event_list

syntax: _M.events = events.event_list(sourcename, event1, event2, ...)

Hilfsfunktion zur Generierung von Ereignislisten und zur Vermeidung von Tippfehlern in magischen Strings. Der Zugriff auf ein nicht vorhandenes Ereignis in der zurückgegebenen Tabelle führt zu einem 'unbekannten Ereignisfehler'. Der erste Parameter sourcename ist ein eindeutiger Name, der die Ereignisquelle identifiziert, die als Feld _source verfügbar sein wird. Alle folgenden Parameter sind die benannten Ereignisse, die von der Ereignisquelle generiert werden.

Beispielverwendung; 

local ev = require "resty.worker.events"

-- Beispiel für eine Ereignisquelle

local events = ev.event_list(
        "my-module-event-source", -- verfügbar als _M.events._source
        "started",                -- verfügbar als _M.events.started
        "event2"                  -- verfügbar als _M.events.event2
    )

local raise_event = function(event, data)
    return ev.post(events._source, event, data)
end

-- Mein eigenes 'started' Ereignis posten
raise_event(events.started, nil) -- nil zur Klarheit, es werden keine Ereignisdaten übergeben

-- mein Modul-Table definieren
local _M = {
  events = events   -- Ereignistabelle exportieren

  -- Implementierung hier
}
return _M

-- Beispiel für einen Ereigniskunden;
local mymod = require("some_module")  -- Modul mit einer `events`-Tabelle

-- einen Callback definieren und die Ereignistabelle der Quelle verwenden
local my_callback = function(data, event, source, pid)
    if event == mymod.events.started then  -- 'started' ist der Ereignisname

        -- gestartetes Ereignis aus dem resty-worker-events-Modul

    elseif event == mymod.events.stoppping then  -- 'stopping' ist der Ereignisname

        -- das obige wird einen Fehler auslösen wegen des Tippfehlers in `stoppping`

    end
end

ev.register(my_callback, mymod.events._source)

poll

syntax: success, err = events.poll()

Fragt nach neuen Ereignissen und behandelt sie alle (ruft die registrierten Rückrufe auf). Die Implementierung ist effizient, sie überprüft nur einen einzelnen Wert im gemeinsamen Speicher und gibt sofort zurück, wenn keine neuen Ereignisse verfügbar sind.

Der Rückgabewert ist "done", wenn alle Ereignisse behandelt wurden, "recursive", wenn es sich bereits in einer Abfrage-Schleife befand, oder nil + Fehler, wenn etwas schiefgegangen ist. Das Ergebnis "recursive" bedeutet einfach, dass das Ereignis erfolgreich gepostet wurde, aber noch nicht behandelt wurde, aufgrund anderer Ereignisse, die zuerst behandelt werden müssen.

post

syntax: success, err = events.post(source, event, data, unique)

Postet ein neues Ereignis. source und event sind beide Strings. data kann alles sein (einschließlich nil), solange es vom cjson-Modul (de)serialisierbar ist.

Wenn der Parameter unique bereitgestellt wird, wird nur ein Arbeiter das Ereignis ausführen, die anderen Arbeiter ignorieren es. Auch alle nachfolgenden Ereignisse mit demselben unique-Wert werden ignoriert (für den timeout-Zeitraum, der an configure übergeben wird). Der Prozess, der das Ereignis ausführt, muss nicht unbedingt der Prozess sein, der das Ereignis postet.

Der Rückgabewert ist true, wenn das Ereignis erfolgreich gepostet wurde, oder nil + Fehler im Falle eines Fehlers.

Hinweis: Der Arbeitsprozess, der das Ereignis sendet, wird auch das Ereignis empfangen! Wenn die Ereignisquelle auch auf das Ereignis reagieren soll, sollte dies nicht im Code zum Posten des Ereignisses erfolgen, sondern nur beim Empfang.

post_local

syntax: success, err = events.post_local(source, event, data)

Das gleiche wie post, mit der Ausnahme, dass das Ereignis lokal für den Arbeitsprozess sein wird, es wird nicht an andere Arbeiter gesendet. Mit dieser Methode wird das data-Element nicht in JSON umgewandelt.

Der Rückgabewert ist true, wenn das Ereignis erfolgreich gepostet wurde, oder nil + Fehler im Falle eines Fehlers.

register

syntax: events.register(callback, source, event1, event2, ...)

Registriert eine Rückruffunktion, um Ereignisse zu empfangen. Wenn source und event weggelassen werden, wird der Rückruf bei jedem Ereignis ausgeführt. Wenn source bereitgestellt wird, werden nur Ereignisse mit einer übereinstimmenden Quelle übergeben. Wenn (ein oder mehrere) Ereignisnamen angegeben sind, wird der Rückruf nur aufgerufen, wenn sowohl source als auch event übereinstimmen.

Der Rückruf sollte die folgende Signatur haben;

syntax: callback = function(data, event, source, pid)

Die Parameter sind die gleichen wie die, die an post übergeben werden, mit dem zusätzlichen Wert pid, der die PID des ursprünglichen Arbeitsprozesses ist, oder nil, wenn es sich nur um ein lokales Ereignis handelte. Jeder Rückgabewert von callback wird verworfen.
Hinweis: data kann ein Referenztyp von Daten sein (z. B. ein Lua table). Der gleiche Wert wird an alle Rückrufe übergeben, ändern Sie also den Wert in Ihrem Handler nicht, es sei denn, Sie wissen, was Sie tun!

Der Rückgabewert von register ist true, oder es wird einen Fehler auslösen, wenn callback kein Funktionswert ist.

WARNUNG: Ereignishandler müssen schnell zurückkehren. Wenn ein Handler länger als den konfigurierten timeout-Wert benötigt, werden Ereignisse verworfen!

Hinweis: Um das eigene started-Ereignis des Prozesses zu empfangen, muss der Handler registriert werden, bevor configure aufgerufen wird.

register_weak

syntax: events.register_weak(callback, source, event1, event2, ...)

Diese Funktion ist identisch mit register, mit der Ausnahme, dass das Modul nur schwache Referenzen zur callback-Funktion hält.

unregister

syntax: events.unregister(callback, source, event1, event2, ...)

Registriert die Rückruffunktion ab und verhindert, dass sie weitere Ereignisse empfängt. Die Parameter funktionieren genau wie bei register.

Der Rückgabewert ist true, wenn es entfernt wurde, false, wenn es nicht in der Handlerliste war, oder es wird einen Fehler auslösen, wenn callback kein Funktionswert ist.

Geschichte

Veröffentlichung neuer Versionen

  • Stellen Sie sicher, dass das Änderungsprotokoll unten auf dem neuesten Stand ist
  • Aktualisieren Sie die Versionsnummer im Code
  • Erstellen Sie ein neues Rockspec in ./rockspecs
  • Commit mit der Nachricht release x.x.x
  • Taggen Sie den Commit als x.x.x
  • Pushen Sie Commit und Tags
  • Hochladen zu luarocks

2.0.1, 28. Juni 2021

  • Fix: Möglicher Deadlock in der 'Init-Phase'

2.0.0, 16. September 2020

  • BREAKING: Die post-Funktion ruft poll nicht mehr auf, wodurch alle Ereignisse asynchron werden. Wenn eine sofortige Behandlung eines Ereignisses erforderlich ist, muss ein expliziter Aufruf von poll erfolgen.
  • BREAKING: Die post_local-Funktion führt das Ereignis nicht mehr sofort aus, wodurch alle lokalen Ereignisse asynchron werden. Wenn eine sofortige Behandlung eines Ereignisses erforderlich ist, muss ein expliziter Aufruf von poll erfolgen.
  • Fix: Verhindern, dass die CPU bei einem Reload auf 100% läuft, wenn der Ereignis-shm gelöscht wird.
  • Fix: Verbesserte Protokollierung im Falle eines Fehlers beim Schreiben in shm (Payload-Größe für Fehlerbehebungszwecke hinzufügen).
  • Fix: Protokollieren Sie die Payload nicht mehr, da sie möglicherweise sensible Daten durch die Protokolle offenlegt.
  • Änderung: Standardwert für shm_retries auf 999 aktualisiert.
  • Änderung: Timer-Schleife in eine Schlafschleife geändert (Leistung).
  • Fix: Bei der Neukonfiguration sicherstellen, dass die Rückruftabelle initialisiert ist.

1.1.0, 23. Dezember 2020 (Wartungsrelease)

  • Feature: Die Abfrageschleife läuft jetzt unendlich, schläft 0,5 Sekunden zwischen den Durchläufen und vermeidet es, bei jedem Schritt neue Timer zu erstellen.

1.0.0, 18. Juli 2019

  • BREAKING: Die Rückgabewerte von poll (und damit auch post und post_local) wurden geändert, um lua-ähnlicher zu sein, um wahr zu sein, wenn alles gut ist.
  • Feature: Neue Option shm_retries, um "no memory"-Fehler zu beheben, die durch Speicherfragmentierung im shm beim Posten von Ereignissen verursacht werden.
  • Fix: Zwei Tippfehler in Variablennamen (Randfälle) behoben.

0.3.3, 8. Mai 2018

  • Fix: Zeitüberschreitungen in Init-Phasen, durch Entfernen der Timeout-Einstellung, siehe Problem #9.

0.3.2, 11. April 2018

  • Änderung: Fügen Sie einen Stacktrace zu Handler-Fehlern hinzu.
  • Fix: Fehlerbehandlung schlägt fehl, wenn der Wert nicht serialisierbar war, siehe Problem #5.
  • Fix: Test für die schwachen Handler beheben.

Siehe auch

GitHub

Sie finden möglicherweise zusätzliche Konfigurationstipps und Dokumentationen für dieses Modul im GitHub-Repository für nginx-module-worker-events.