diff --git a/CHANGELOG.md b/CHANGELOG.md index 9bce9d2..c8135ea 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,19 @@ Framework `datapipeline` (Gem). Neueste Version oben. +## 3.2.0 – 2026-10-09 + +- Benachrichtigungen: `notify notifier, on: :failure | :problems | :always` nach einem Lauf, `DataPipeline::Report` + (Ereignis, Text, JSON-Payload `schema: 1`), Basisklasse `DataPipeline::Notifier` für eigene Kanäle; verworfene + Zeilen nur mit `include_rows: true` +- `DataPipeline::Webhook`: POST mit HMAC-SHA256-Signatur, Wiederholung bei 429/5xx (`Retry-After`), Timeout; + `Webhook.valid_signature?` für Empfänger +- `DataPipeline::Email`: Bericht als Text-Mail über SMTP (STARTTLS, TLS, Anmeldung), Gem `net-smtp` +- `DataPipeline::Slack`: `deliver(report)` mit Block Kit, Maskierung von `&`, `<`, `>`, `mention:` bei Abbruch, + Incoming Webhooks (`webhook:`), Timeout und Wiederholung; Fehler als `DataPipeline::Slack::Error` +- Notifier aus der Verbindungsdatei (`notifier(:name)`, `type: webhook | slack | email`); `etl --notify=NAME + [--notify-on=…]` sendet einen Bericht für alle Pipelines eines Skripts + ## 3.1.0 – 2026-10-09 - sqlite: `write_row` schreibt gesammelt je `batch_size` Zeilen (Default 1000) in einer Transaktion; diff --git a/Gemfile b/Gemfile index e0d090c..f9df370 100644 --- a/Gemfile +++ b/Gemfile @@ -7,7 +7,8 @@ gemspec # optional plugin gems: needed for the plugin tests, loaded only when a plugin is used group :plugins do gem 'caxlsx', require: false # excel write - gem 'http', require: false # rest, slack + gem 'http', require: false # rest, slack, webhook + gem 'net-smtp', require: false # email gem 'pg', require: false # postgres gem 'roo', require: false # excel read gem 'sqlite3', require: false # sqlite diff --git a/Gemfile.lock b/Gemfile.lock index f166c0c..1095d9e 100644 --- a/Gemfile.lock +++ b/Gemfile.lock @@ -1,7 +1,7 @@ PATH remote: . specs: - datapipeline (3.1.0) + datapipeline (3.2.0) csv (~> 3.3) logger (~> 1.6) @@ -26,6 +26,10 @@ GEM logger (1.7.0) marcel (1.2.1) minitest (5.27.0) + net-protocol (0.4.0) + timeout + net-smtp (0.5.2) + net-protocol nokogiri (1.19.4-aarch64-linux-gnu) racc (~> 1.4) nokogiri (1.19.4-aarch64-linux-musl) @@ -66,6 +70,7 @@ GEM sqlite3 (2.9.6-x86_64-darwin) sqlite3 (2.9.6-x86_64-linux-gnu) sqlite3 (2.9.6-x86_64-linux-musl) + timeout (0.6.1) PLATFORMS aarch64-linux @@ -84,6 +89,7 @@ DEPENDENCIES datapipeline! http minitest (~> 5.25) + net-smtp pg rake roo diff --git a/README.md b/README.md index 862f215..a858aab 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # DataPipeline – ETL-Framework in Ruby -Stand: 2026-10-09 · Gem `datapipeline` 3.1.0 ([CHANGELOG.md](CHANGELOG.md)) · Ruby 3.4 · Lizenz BSD-3-Clause ([LICENSE](LICENSE)) +Stand: 2026-10-09 · Gem `datapipeline` 3.2.0 ([CHANGELOG.md](CHANGELOG.md)) · Ruby 3.4 · Lizenz BSD-3-Clause ([LICENSE](LICENSE)) `DataPipeline` liest Daten aus einer Quelle, schickt jede Zeile durch eine Kette von Schritten (prüfen, filtern, umbauen, anreichern, abgleichen) und schreibt das Ergebnis in ein oder mehrere Ziele. Eine Pipeline ist ein kurzes @@ -14,7 +14,7 @@ Festbreiten-Dateien und REST-APIs. 3. [Pipelines schreiben](#3-pipelines-schreiben) – DSL, Schritte, Prüfen, Fehler, Ergebnis 4. [Plugins](#4-plugins) – mitgelieferte Quellen und Ziele, eigene Plugins, Abhängigkeiten 5. [Satzarten-Layouts](#5-satzarten-layouts) – Dateien mit Kopf-, Positions- und Schlusssätzen -6. [Betrieb](#6-betrieb) – Kommandozeile, Verbindungen, Sperre, Abgleich, Historie, Benachrichtigung, Fehlersuche +6. [Betrieb](#6-betrieb) – Kommandozeile, Verbindungen, Sperre, Abgleich, Historie, Benachrichtigungen, Fehlersuche 7. [Umstieg](#7-umstieg) – auf 3.1, von 2.x auf 3.0 (`transform` entfernt), von 2.0 auf 2.1 8. [Entwicklung](#8-entwicklung) – Aufbau, Tests, Gem bauen 9. [Ausblick](#9-ausblick) @@ -32,7 +32,8 @@ Festbreiten-Dateien und REST-APIs. | Prüfen | Regeln je Spalte (Pflicht, Länge, Format, Werteliste, Typ), eigene Regeln als Block, Satzarten-Layouts; als Fehler oder nur als Warnung | | Fehler | Abbruch, Zeile überspringen oder mit Grund in eine Fehlerdatei schreiben | | Datenmengen | Batch im Speicher oder Streaming Zeile für Zeile mit konstantem Speicherbedarf; Postgres-Cursor und Schreiben in Paketen | -| Betrieb | Kommandozeile `etl` mit Exit-Codes, Dry-Run, Limit, inkrementelle Läufe (Wasserzeichen), Sperre gegen parallele Läufe, Upsert für wiederholbare Läufe, Summenabgleich vor dem Schreiben, Laufhistorie, Verbindungsdatei, Slack | +| Betrieb | Kommandozeile `etl` mit Exit-Codes, Dry-Run, Limit, inkrementelle Läufe (Wasserzeichen), Sperre gegen parallele Läufe, Upsert für wiederholbare Läufe, Summenabgleich vor dem Schreiben, Laufhistorie, Verbindungsdatei | +| Benachrichtigung | Berichte an Webhooks (signiert), Slack und E-Mail, je Pipeline oder je Skript; eigene Notifier | | Dateiformate | Zeichensatz (z. B. Windows-1252) für CSV und Festbreite, BOM für CSV; Satzarten-Dateien schreiben, prüfen und lesen | **Anwendungen:** Das Projekt `etl` nutzt das Gem für die PDS → SAP-Schnittstelle (CUDO) und für das Anlegen von @@ -106,7 +107,7 @@ der Code hier genau der Datei entspricht. | [2.3 Inkrementell und gestreamt](#23-inkrementell-und-gestreamt) | Wasserzeichen, Streaming | | [2.4 Eigenes Plugin](#24-eigenes-plugin) | eine Quelle selbst schreiben | | [2.5 REST-APIs](#25-rest-apis) | Paging, Wiederholung, Pakete, eine URL je Datensatz | -| [2.6 Produktiver Lauf](#26-produktiver-lauf) | Verbindungsdatei, Sperre, Kurzschritte, Summenabgleich, Upsert, Zeichensatz, Historie, Slack | +| [2.6 Produktiver Lauf](#26-produktiver-lauf) | Verbindungsdatei, Sperre, Kurzschritte, Summenabgleich, Upsert, Zeichensatz, Historie, Benachrichtigung | ### 2.1 Kundenimport @@ -624,13 +625,11 @@ FileUtils.mkdir_p(out) # Zugangsdaten stehen nicht im Skript, sondern in der Verbindungsdatei (Default: config/connections.yml) DataPipeline.connections_file = File.expand_path('config/connections.yml', __dir__) -# Slack nur, wenn ein Token gesetzt ist; so fällt ein fehlender Token schon vor dem Lauf auf -slack = DataPipeline::Slack.new(channel: '#etl') if ENV['SLACK_TOKEN'] - DataPipeline.new do name 'buchungen_export' lock "#{out}/.buchungen.lock" # läuft schon ein Export, bricht dieser ab (Exit-Code 3) history "#{out}/runs.ndjson" # eine JSON-Zeile je Lauf, auch nach Abbruch + notify DataPipeline::Slack.new(channel: '#etl') if ENV['SLACK_TOKEN'] # Meldung nur bei Problemen read container(:csv, file: "#{DATA}/buchungen.csv", csv: { col_sep: ';' }) @@ -662,10 +661,7 @@ DataPipeline.new do write container(:csv, file: "#{out}/buchungen_sap.csv", encoding: 'Windows-1252', csv: { col_sep: ';' }, columns: %w[belnr buzei datum konto shkzg dmbtr buchungstext]) - on_finish do |stats| - puts "#{stats.name}: #{stats}" - slack&.send_message("*#{stats.name}*: #{stats}") unless stats.ok? - end + on_finish { |stats| puts "#{stats.name}: #{stats}" } end.run ``` @@ -679,7 +675,7 @@ end.run | `totals`, `reconcile` | Summen werden exakt gerechnet; eine Abweichung bricht vor dem Schreiben ab (6.4) | | `upsert: [belnr, buzei]` | ein wiederholter Lauf aktualisiert dieselben Zeilen, statt sie doppelt anzulegen (4.2) | | `encoding: 'Windows-1252'` | Zeichensatz der Zieldatei (4.4) | -| `slack&.send_message` | Meldung nur bei Problemen und nur, wenn `SLACK_TOKEN` gesetzt ist (6.8) | +| `notify DataPipeline::Slack…` | Meldung nur bei Problemen und nur, wenn `SLACK_TOKEN` gesetzt ist (6.8) | ```text INFO start @@ -765,6 +761,7 @@ Tippfehler fallen also sofort auf. | `on_error(:abort \| :skip \| :collect, to: storage)` | Umgang mit Fehlern in Schritten | | `totals 'spalte', …` | Summen über Quelle und Ergebnis | | `reconcile { \|stats\| … }` | Abgleich vor dem Schreiben; eine Meldung bricht ab | +| `notify(notifier, on: :problems)`, `notifier(:name)` | Bericht an Webhook, Slack oder E-Mail (6.8) | | `on_finish { \|stats\| … }` | nach jedem Lauf, auch nach einem Abbruch | | `name(str)`, `version(str)` | Name und Version des Laufs für Log, Stats und Historie | @@ -1137,7 +1134,8 @@ Gemfile der Anwendung, die das Plugin nutzt: | `:postgres` | `pg` | | `:sqlite` | `sqlite3` | | `:excel` | `roo` (lesen), `caxlsx` (schreiben) | -| `:rest`, `DataPipeline::Slack` | `http` | +| `:rest`, `DataPipeline::Slack`, `DataPipeline::Webhook` | `http` | +| `DataPipeline::Email` | `net-smtp` | | `:csv`, `:json`, `:fixed` | – | Eine CSV- oder JSON-Pipeline lädt keines dieser Gems. Fehlt ein Gem, meldet das Plugin bei der ersten Nutzung, welches @@ -1299,6 +1297,7 @@ bin/etl run cudo.rb --lock=data/.cudo.lock --history=data/runs.ndjson | `--dry-run`, `--limit=N` | für alle Pipelines des Skripts | | `--lock=FILE` | Sperre für das ganze Skript (6.3) | | `--history=FILE` | Laufhistorie für alle Pipelines des Skripts (6.6) | +| `--notify=NAME`, `--notify-on=WHEN` | ein Bericht für alle Pipelines des Skripts an den Notifier `NAME` aus der Verbindungsdatei; `WHEN` = `problems` (Default), `failure`, `always` (6.8) | | `--verbose`, `--quiet` | Log-Level Debug bzw. Warnung (Default: Info) | | jede andere `--name=wert` | `params[:name]` und unverändert an das Skript, z. B. `--out=data`, `--incremental=1` | @@ -1423,81 +1422,181 @@ Eigene Schritte mit Nebenwirkungen (API-Aufrufe in einem `map`) fragen `dry_run? `limit n` verarbeitet nur die ersten n Quellzeilen; im Streaming wird das Lesen dann beendet. -### 6.8 Benachrichtigung über Slack +### 6.8 Benachrichtigungen -`DataPipeline::Slack` schickt Text über die Slack-API `chat.postMessage`. Einrichten: - -1. In Slack eine App anlegen (api.slack.com/apps), unter *OAuth & Permissions* den Bot-Scope `chat:write` vergeben, - die App im Workspace installieren und den *Bot User OAuth Token* (`xoxb-…`) kopieren. -2. Die App in den Ziel-Channel einladen (`/invite @appname`), sonst antwortet Slack mit `not_in_channel`. -3. Das Gem `http` ins Gemfile der Anwendung aufnehmen (4.10). -4. Den Token als Umgebungsvariable `SLACK_TOKEN` setzen oder mit `token:` übergeben; nicht ins Skript schreiben. +`notify` schickt nach einem Lauf einen Bericht an einen Notifier: einen Webhook, Slack oder E-Mail. Ein Fehler beim +Senden wird nur geloggt; Ergebnis und Exit-Code bleiben unverändert. ```ruby -slack = DataPipeline::Slack.new # Token aus ENV['SLACK_TOKEN'], Channel '#general' -slack = DataPipeline::Slack.new(channel: '#etl') # anderer Channel (Name oder ID wie 'C0123ABCD') -slack = DataPipeline::Slack.new(channel: '#etl', token: ENV.fetch('SLACK_ETL_TOKEN')) # eigener Token je Anwendung - -slack.send_message('Test aus der Pipeline', verbose: true) # verbose: Antwort von Slack als JSON auf $stdout -``` - -| Option | Default | Bedeutung | -| --- | --- | --- | -| `token:` | `ENV['SLACK_TOKEN']` | Bot-Token; fehlt er, bricht schon `Slack.new` mit `ArgumentError` ab | -| `channel:` | `'#general'` | Ziel-Channel als Name oder ID | - -`send_message(text, verbose: false)` liefert die Antwort von Slack als Hash. Meldet Slack einen Fehler (z. B. -`invalid_auth`, `channel_not_found`, `not_in_channel`), wirft die Methode eine Exception mit diesem Code. In -`on_finish` wird sie nur geloggt (`on_finish: RuntimeError: Slack#send_message failed: …`), der Lauf und sein -Exit-Code bleiben unverändert. Es gibt keine Wiederholung bei Netzwerkfehlern. - -Eine ausführliche Meldung mit Status, Zählern, Laufzeit und den ersten verworfenen Zeilen. Der Notifier wird vor der -Pipeline angelegt, damit ein fehlender Token schon beim Start auffällt. Slack setzt `*fett*`, `_kursiv_` und -`` `code` `` um. - -```ruby -slack = DataPipeline::Slack.new(channel: '#etl') - DataPipeline.new do + name 'cudo_export' + notify DataPipeline::Webhook.new(url: ENV['ALERT_URL'], secret: ENV['ALERT_SECRET']) # bei Problemen + notify DataPipeline::Slack.new(channel: '#etl', mention: ''), on: :failure + notify DataPipeline::Email.new(to: 'buchhaltung@example.com'), on: :always + notify notifier(:alerts) # aus der Verbindungsdatei # … read, Schritte, write … - - on_finish do |stats| - next if stats.ok? && stats.warnings.zero? # nur melden, wenn etwas zu tun ist - - status = if stats.error then ':red_circle: abgebrochen' - elsif stats.rejected.positive? then ':large_orange_circle: mit verworfenen Zeilen' - else ':large_yellow_circle: mit Warnungen' - end - lines = ["*Buchungsexport #{stats.version}* #{status}#{' (Dry-Run)' if stats.dry_run}", - "Zeitraum #{params[:from]} – #{params[:till]}, Start #{stats.started_at.strftime('%d.%m.%Y %H:%M')}, " \ - "Dauer #{stats.duration}s", - "gelesen #{stats.read} · geschrieben #{stats.written} · verworfen #{stats.rejected} · Warnungen #{stats.warnings}"] - lines << "Fehler: `#{stats.error.class}: #{stats.error.message}`" if stats.error - stats.errors.first(5).each do |e| - lines << "• #{e[:step]}: #{e[:error]} – Beleg #{e[:row]['belnr']}" - end - lines << "… und #{stats.rejected - 5} weitere, siehe data/fehler.csv" if stats.rejected > 5 - - slack.send_message(lines.join("\n")) - end end.run ``` -So sieht die Meldung in Slack aus: +| Option von `notify` | Wirkung | +| --- | --- | +| `on: :problems` (Default) | bei Abbruch, verworfenen Zeilen oder Warnungen | +| `on: :failure` | nur bei Abbruch | +| `on: :always` | nach jedem Lauf | +| `title:` | Titel des Berichts, Default der Name der Pipeline | +| `dry_run: true` | auch nach einem Dry-Run senden (Default: nicht) | +| `include_rows: true` | die verworfenen Zeilen mitsenden; Default: nur Schritt und Meldung, weil Zeilen Kundendaten enthalten können | -```text -Buchungsexport 1.0 🟠 mit verworfenen Zeilen -Zeitraum 2025-11-01 – 2025-11-30, Start 01.12.2025 06:00, Dauer 4.213s -gelesen 1250 · geschrieben 1243 · verworfen 7 · Warnungen 0 -• validate#1: konto: missing – Beleg 5100000123 -• validate#1: dmbtr: not decimal – Beleg 5100000124 -• validate#2: Betrag 0 – Beleg 5100000130 -• validate#1: shkzg: not in ["S", "H"] – Beleg 5100000131 -• validate#1: konto: format /\A\d+\z/ – Beleg 5100000140 -… und 2 weitere, siehe data/fehler.csv +**Bericht** – Jeder Notifier bekommt einen `DataPipeline::Report` mit den Stats eines oder mehrerer Läufe: + +| Methode | Inhalt | +| --- | --- | +| `event` | `run.failed` (Abbruch), `run.rejected` (verworfene Zeilen), `run.warnings`, `run.succeeded`; bei mehreren Läufen der schlimmste | +| `title`, `status`, `subject` | z. B. `cudo_export`, `mit verworfenen Zeilen`, `cudo_export: mit verworfenen Zeilen` | +| `lines`, `to_s` | Text je Lauf: Zähler, Start, Dauer, die ersten 5 Fehler (ohne Zeilen) | +| `to_h(include_rows: false)` | JSON-Payload für Webhooks (unten) | +| `id`, `host`, `params`, `runs`, `error` | eindeutige Kennung, Rechner, Parameter der Kommandozeile, Stats, Fehler ausserhalb eines Laufs | + +**Ein Bericht für das ganze Skript** – `notify` meldet jede Pipeline einzeln. Für einen Sammelbericht über alle +Pipelines eines Skripts gibt es die Kommandozeile; der Notifier kommt aus der Verbindungsdatei: + +```sh +bin/etl run cudo.rb --notify=alerts # bei Problemen (Default) +bin/etl run cudo.rb --notify=alerts --notify-on=always ``` -`params` gibt es im Pipeline-Block; ausserhalb heisst es `DataPipeline.params`. +Bricht das Skript selbst ab (z. B. ein Fehler vor der ersten Pipeline), meldet der Bericht `run.failed` mit diesem +Fehler. Bei belegter Sperre (Exit-Code `3`) und im Dry-Run wird nichts gesendet. + +#### Webhook + +`DataPipeline::Webhook` sendet den Bericht als JSON per `POST`. + +| Option | Default | Bedeutung | +| --- | --- | --- | +| `url:` | – | Ziel, Pflicht | +| `secret:` | – | signiert den Body (unten) | +| `headers:` | `{}` | weitere Header, z. B. `{ 'Authorization' => "Bearer #{ENV['TOKEN']}" }` | +| `timeout:` | 10 | Sekunden je Anfrage | +| `retries:`, `retry_wait:` | 3, 2 | Wiederholung bei 429, 5xx und Netzwerkfehlern, Wartezeit `retry_wait` × Versuch; `Retry-After` wird beachtet; andere 4xx brechen sofort ab | + +```json +{"schema":1,"id":"3f6c2a1e-…","event":"run.rejected","title":"cudo_export","host":"etl01", + "params":{"from":"2026-10-08"},"error":null, + "runs":[{"name":"cudo_export","version":"1.0","started_at":"2026-10-09T06:00:00.123+02:00","duration":4.213, + "read":1250,"written":1243,"rejected":7,"warnings":0,"ok":false,"error":null, + "totals":{"dmbtr":{"read":"98231.4","written":"97810.1"}}, + "errors":[{"step":"validate#1","error":"konto: missing"}]}]} +``` + +`schema: 1` ist ein Versprechen: Felder können dazukommen, eine Änderung bestehender Felder hiesse `schema: 2`. Jede +Anfrage trägt die Header `X-DataPipeline-Event` und `X-DataPipeline-Delivery` (die `id`, bei Wiederholungen gleich; +damit erkennt der Empfänger Doppelte). Mit `secret:` kommen dazu: + +```text +X-DataPipeline-Timestamp: 1791525600 +X-DataPipeline-Signature: sha256=.") hexadezimal> +``` + +Der Empfänger prüft Signatur und Alter (gegen wiederholt eingespielte Anfragen); in Ruby z. B.: + +```ruby +ok = DataPipeline::Webhook.valid_signature?(secret: ENV['ALERT_SECRET'], body: request.raw_post, + timestamp: request.headers['X-DataPipeline-Timestamp'], + signature: request.headers['X-DataPipeline-Signature'], + max_age: 300) +``` + +#### Slack + +`DataPipeline::Slack` meldet einen Bericht als Statuszeile mit Symbol (🔴 abgebrochen, 🟠 verworfene Zeilen, +🟡 Warnungen, 🟢 ok) und darunter den Text je Lauf. `&`, `<` und `>` aus Daten und Meldungen werden maskiert. + +Zwei Wege zu Slack: + +| Weg | Einrichten | Aufruf | +| --- | --- | --- | +| Incoming Webhook (einfacher) | Slack-App anlegen (api.slack.com/apps), *Incoming Webhooks* aktivieren, Webhook für den Channel erzeugen | `Slack.new(webhook: ENV['SLACK_WEBHOOK'])` | +| Bot-Token | Slack-App mit Bot-Scope `chat:write` installieren, *Bot User OAuth Token* (`xoxb-…`) kopieren, App in den Channel einladen (`/invite @appname`) | `Slack.new(channel: '#etl')`, Token aus `ENV['SLACK_TOKEN']` | + +| Option | Default | Bedeutung | +| --- | --- | --- | +| `webhook:` | – | URL eines Incoming Webhook; dann braucht es weder Token noch Channel | +| `token:` | `ENV['SLACK_TOKEN']` | Bot-Token; ohne Token und ohne Webhook bricht schon `Slack.new` mit `ArgumentError` ab | +| `channel:` | `'#general'` | Channel als Name oder ID (nur mit Token) | +| `mention:` | – | z. B. `''` oder `''`, nur bei `run.failed` | +| `timeout:`, `retries:`, `retry_wait:` | 10, 3, 2 | wie beim Webhook | + +Eigener Text geht weiter mit `send_message`, z. B. in `on_finish`: + +```ruby +slack = DataPipeline::Slack.new(channel: '#etl') +slack.send_message('Monatsabschluss exportiert', verbose: true) # verbose: Antwort von Slack auf $stdout +``` + +Meldet Slack einen Fehler (`invalid_auth`, `channel_not_found`, `not_in_channel`), wirft der Aufruf +`DataPipeline::Slack::Error` mit diesem Code. + +#### E-Mail + +`DataPipeline::Email` sendet den Bericht als Text-Mail (UTF-8) über SMTP. Betreff: `[etl] : `. Die +Zugangsdaten kommen aus der Umgebung oder als Option: + +| Option | Default | Bedeutung | +| --- | --- | --- | +| `to:` | – | eine Adresse oder ein Array, Pflicht | +| `from:` | `ENV['SMTP_FROM']`, sonst `etl@` | Absender | +| `host:`, `port:` | `ENV['SMTP_HOST']` bzw. `localhost`; `ENV['SMTP_PORT']` bzw. 587 | SMTP-Server | +| `username:`, `password:` | `ENV['SMTP_USERNAME']`, `ENV['SMTP_PASSWORD']` | Anmeldung (LOGIN); ohne Benutzer keine Anmeldung | +| `starttls:` | `:auto` | `:auto` (wenn der Server es anbietet), `:always` oder `false` (dann Default-Port 25) | +| `tls:` | `false` | `true` = TLS von Anfang an (Port 465) | +| `subject_prefix:` | `'[etl] '` | Anfang des Betreffs | +| `timeout:` | 15 | Sekunden für Verbindung und Antwort | + +Mit `include_rows: true` stehen die verworfenen Zeilen als JSON am Ende der Mail. Eigener Text: +`DataPipeline::Email.new(to: …).send_mail('Betreff', "Text")`. Ein Fehler beim Senden wirft +`DataPipeline::Email::Error`. Das Plugin braucht das Gem `net-smtp`. + +#### Notifier in der Verbindungsdatei + +Wie Storages lassen sich Notifier in `config/connections.yml` ablegen (6.2); `type` ist `webhook`, `slack` oder +`email`, die anderen Schlüssel sind die Optionen: + +```yaml +alerts: + type: webhook + url: <%= ENV['ALERT_URL'] %> + secret: <%= ENV['ALERT_SECRET'] %> +buchhaltung_mail: + type: email + to: [buchhaltung@example.com, etl@example.com] + host: smtp.example.com + username: etl@example.com + password: <%= ENV['SMTP_PASSWORD'] %> +``` + +```ruby +notify notifier(:buchhaltung_mail), on: :always +``` + +#### Eigene Notifier + +Ein Notifier ist eine Klasse mit Basisklasse `DataPipeline::Notifier` und einer Methode +`deliver(report, include_rows: false)`, z. B. für Microsoft Teams oder ein Ticketsystem: + +```ruby +require 'http' + +class TeamsNotifier < DataPipeline::Notifier + def initialize(url:) = @url = url + + def deliver report, include_rows: false + HTTP.post(@url, json: { title: report.subject, text: report.lines.join("\n\n") }) + end +end + +notify TeamsNotifier.new(url: ENV['TEAMS_WEBHOOK']), on: :failure +``` ### 6.9 Fehlersuche @@ -1523,11 +1622,19 @@ schnellen Blick auf das Ergebnis. ## 7. Umstieg -### 7.0 Von 3.0 auf 3.1 +### 7.0 Von 3.0 auf 3.1 und 3.2 -3.1 ist abwärtskompatibel. Anders als in 3.0: SQLite schreibt im Streaming (`write_row`) nicht mehr jede Zeile -einzeln, sondern je 1000 Zeilen in einer Transaktion (4.3). Wer während eines Laufs mit einer zweiten Verbindung -auf die Zieldatei schaut, sieht die Zeilen also paketweise. Das Gem steht jetzt unter der BSD-3-Clause-Lizenz. +3.1 und 3.2 sind abwärtskompatibel. Anders als in 3.0: + +| Thema | bis 3.0 | jetzt | seit | +| --- | --- | --- | --- | +| SQLite, `write_row` im Streaming | jede Zeile einzeln | je 1000 Zeilen in einer Transaktion (4.3); eine zweite Verbindung sieht die Zeilen paketweise | 3.1 | +| Lizenz | nicht festgelegt | BSD-3-Clause | 3.1 | +| Slack-Fehler | `RuntimeError` | `DataPipeline::Slack::Error` (weiterhin ein `StandardError`) | 3.2 | +| `Slack.new` | lädt `http` erst beim Senden | lädt `http` sofort; Timeout 10 s und Wiederholung bei 429/5xx | 3.2 | +| `send_message(…, verbose: true)` | Antwort als eingerücktes JSON | Antwort wie von Slack geliefert | 3.2 | + +Selbstgebaute Slack-Meldungen in `on_finish` laufen weiter; einfacher ist `notify` (6.8). ### 7.1 Von 2.x auf 3.0 @@ -1584,7 +1691,7 @@ ersten Zeilen sollten gleich sein. | [lib/datapipeline/lock.rb](lib/datapipeline/lock.rb) | Sperre gegen parallele Läufe | | [lib/datapipeline/connections.rb](lib/datapipeline/connections.rb) | Verbindungsdatei | | [lib/datapipeline/storage/](lib/datapipeline/storage/) | `storage_plugin.rb` (Basisklasse) und je Plugin eine Datei | -| [lib/datapipeline/notification/slack.rb](lib/datapipeline/notification/slack.rb) | Slack | +| [lib/datapipeline/notification/](lib/datapipeline/notification/) | `notifier.rb` (Report, Basisklasse), `webhook.rb`, `slack.rb`, `email.rb` | | [lib/datapipeline/version.rb](lib/datapipeline/version.rb) | `DataPipeline::VERSION` | | [exe/etl](exe/etl) | Kommando `etl`; [bin/etl](bin/etl) startet es im Repository mit dem Bundle | | [examples/](examples/) | Beispiele aus Kapitel 2, Daten in `examples/data/`, Verbindungsdatei in `examples/config/`, Demo-API | @@ -1592,14 +1699,14 @@ ersten Zeilen sollten gleich sein. ### 8.2 Tests -`bundle exec rake test` startet die Minitest-Suite in [test/](test/), rund 190 Fälle. Einzelne Datei: +`bundle exec rake test` startet die Minitest-Suite in [test/](test/), rund 210 Fälle. Einzelne Datei: `bundle exec rake test TEST=test/core/steps_test.rb`. | Ordner | Inhalt | | --- | --- | -| `test/plugins` | je Plugin das gemeinsame Testkit `test/support/storage_contract.rb` (Schreiben und Zurücklesen, zeilenweise, als Quelle und Ziel) plus eigene Fälle; REST gegen einen lokalen Testserver; Laden der Gems erst bei Nutzung | +| `test/plugins` | je Plugin das gemeinsame Testkit `test/support/storage_contract.rb` (Schreiben und Zurücklesen, zeilenweise, als Quelle und Ziel) plus eigene Fälle; REST, Webhook und Slack gegen einen lokalen HTTP-Testserver, E-Mail gegen einen SMTP-Testserver (`test/support/`); Laden der Gems erst bei Nutzung | | `test/core` | DSL, Schritte und Kurzschritte, Batch und Streaming, Fehlerbehandlung, Layouts, Sperre, Verbindungen, Summen, Historie, Wasserzeichen | -| `test/cli` | `bin/etl` als Prozess: Parameter, Dry-Run, Sperre, Historie, Exit-Codes | +| `test/cli` | `bin/etl` als Prozess: Parameter, Dry-Run, Sperre, Historie, Sammelbericht `--notify`, Exit-Codes | | `test/examples` | die Beispiele aus Kapitel 2: Code in der Doku = Datei, Lauf und Ergebnis; REST gegen die Demo-API | Eine `.env` wird in Tests nicht geladen; kein Test erreicht eine Produktivdatenbank. Postgres-Tests legen je Test eine @@ -1620,10 +1727,10 @@ Ein eigenes Plugin lässt sich mit `MemoryStorage` aus `test/test_helper.rb` als - Version in [lib/datapipeline/version.rb](lib/datapipeline/version.rb), Änderungen in [CHANGELOG.md](CHANGELOG.md) (neueste oben). Neue Funktionen erhöhen die mittlere Stelle, Breaking Changes die erste (wie 3.0, 7.1). -- Jede Version bekommt ein Tag `vX.Y.Z` (`git tag -a v3.1.0 -m "datapipeline 3.1.0" && git push origin v3.1.0`). +- Jede Version bekommt ein Tag `vX.Y.Z` (`git tag -a v3.2.0 -m "datapipeline 3.2.0" && git push origin v3.2.0`). Anwendungen binden das Gem mit einer Versionsgrenze ein (`'~> 3.1'`, 1.2) oder fest auf ein Tag - (`tag: 'v3.1.0'`; dann geht `local.datapipeline` nicht). -- Das `Gemfile` bindet das Gem über `gemspec` ein und ergänzt die Gruppen `plugins` (`pg`, `sqlite3`, `roo`, + (`tag: 'v3.2.0'`; dann geht `local.datapipeline` nicht). +- Das `Gemfile` bindet das Gem über `gemspec` ein und ergänzt die Gruppen `plugins` (`pg`, `sqlite3`, `roo`, `net-smtp`, `caxlsx`, `http`) und `test`; `Gemfile.lock` ist versioniert. Ohne Plugin-Gems: `bundle config set without plugins`. - Gem bauen: `gem build datapipeline.gemspec`. - Lizenz: BSD-3-Clause, siehe [LICENSE](LICENSE); die Datei ist Teil des Gems. @@ -1640,6 +1747,5 @@ Offene Punkte, Reihenfolge = empfohlene Umsetzung: schreiben), `etl list` (Pipelines, Wasserzeichen, letzter Lauf aus der Historie), `--json` für die Stats. 2. **Qualität:** CI auf git.wicom.li mit `rake test` und Postgres-Service, RuboCop, `DataPipeline::Testing` mit `MemoryStorage` als öffentliche Testhilfe für Anwendungen. -3. **Slack:** fertige Meldung aus den Stats (`notify slack, on: :problems`), Timeout und Wiederholung bei `429`, - Maskierung von `&`, `<`, `>`, Incoming Webhooks. +3. **Benachrichtigungen:** Microsoft Teams als mitgelieferter Notifier, HTML-Mail. 4. **Weitere Plugins:** SFTP-Ablage. diff --git a/datapipeline.gemspec b/datapipeline.gemspec index 9d8cc3c..a2ad849 100644 --- a/datapipeline.gemspec +++ b/datapipeline.gemspec @@ -28,5 +28,5 @@ Gem::Specification.new do |spec| spec.add_dependency 'logger', '~> 1.6' # optional, loaded only when the plugin is used (add them to the Gemfile of the application): - # pg (postgres), sqlite3 (sqlite), roo + caxlsx (excel), http (rest, slack) + # pg (postgres), sqlite3 (sqlite), roo + caxlsx (excel), http (rest, slack, webhook), net-smtp (email) end diff --git a/examples/buchungen_export.rb b/examples/buchungen_export.rb index 066e58a..1719627 100644 --- a/examples/buchungen_export.rb +++ b/examples/buchungen_export.rb @@ -14,13 +14,11 @@ FileUtils.mkdir_p(out) # Zugangsdaten stehen nicht im Skript, sondern in der Verbindungsdatei (Default: config/connections.yml) DataPipeline.connections_file = File.expand_path('config/connections.yml', __dir__) -# Slack nur, wenn ein Token gesetzt ist; so fällt ein fehlender Token schon vor dem Lauf auf -slack = DataPipeline::Slack.new(channel: '#etl') if ENV['SLACK_TOKEN'] - DataPipeline.new do name 'buchungen_export' lock "#{out}/.buchungen.lock" # läuft schon ein Export, bricht dieser ab (Exit-Code 3) history "#{out}/runs.ndjson" # eine JSON-Zeile je Lauf, auch nach Abbruch + notify DataPipeline::Slack.new(channel: '#etl') if ENV['SLACK_TOKEN'] # Meldung nur bei Problemen read container(:csv, file: "#{DATA}/buchungen.csv", csv: { col_sep: ';' }) @@ -52,8 +50,5 @@ DataPipeline.new do write container(:csv, file: "#{out}/buchungen_sap.csv", encoding: 'Windows-1252', csv: { col_sep: ';' }, columns: %w[belnr buzei datum konto shkzg dmbtr buchungstext]) - on_finish do |stats| - puts "#{stats.name}: #{stats}" - slack&.send_message("*#{stats.name}*: #{stats}") unless stats.ok? - end + on_finish { |stats| puts "#{stats.name}: #{stats}" } end.run diff --git a/exe/etl b/exe/etl index 5455218..0e91bb7 100755 --- a/exe/etl +++ b/exe/etl @@ -11,6 +11,8 @@ # --verbose log level debug, --quiet only warnings (default info) # --lock=FILE exclusive lock for the whole script, exit 3 when another run holds it # --history=FILE one json line per pipeline run (DataPipeline.history) +# --notify=NAME one report for all runs of the script, notifier NAME from the connections file +# --notify-on=WHEN failure, problems (default) or always # --name=value any other option: DataPipeline.params[:name] and passed on to the script # # exit code: 0 ok, 1 aborted, 2 finished with rejected rows, 3 locked by another run @@ -35,6 +37,8 @@ parser = OptionParser.new do |opts| opts.on('--quiet') { options[:settings][:log_level] = Logger::WARN } opts.on('--lock=FILE', String) { |v| options[:lock] = v } opts.on('--history=FILE', String) { |v| options[:settings][:history] = v } + opts.on('--notify=NAME', String) { |v| options[:notify] = v } + opts.on('--notify-on=WHEN', String) { |v| options[:notify_on] = v } end # own options for the parser, everything else (also --name=value of the script) goes to the script @@ -42,9 +46,9 @@ own = [] args = [] argv = ARGV.dup while (arg = argv.shift) - if arg.match?(/\A--(from|till|param|dry-run|limit|verbose|quiet|lock|history)(=|\z)/) + if arg.match?(/\A--(from|till|param|dry-run|limit|verbose|quiet|lock|history|notify|notify-on)(=|\z)/) own << arg - own << argv.shift if %w[--param --limit --from --till --lock --history].include?(arg) + own << argv.shift if %w[--param --limit --from --till --lock --history --notify --notify-on].include?(arg) else args << arg end @@ -56,7 +60,26 @@ DataPipeline.params.merge!(DataPipeline.parse_params(args)) { |_key, own_value, abort parser.help if script.nil? abort "etl: #{script} not found\n\n#{parser.help}" unless File.exist?(script) -DataPipeline.settings(options[:settings].merge(name: File.basename(script, '.rb'))) +name = File.basename(script, '.rb') +DataPipeline.settings(options[:settings].merge(name:)) + +# one report for the whole script; created before the run, so a wrong name fails early +if options[:notify] + begin + notifier = DataPipeline::Notifier.from_connection(options[:notify]) + notify_when = DataPipeline::Notifier.condition(options.fetch(:notify_on, :problems)) + rescue ArgumentError, LoadError => e + abort "etl: --notify=#{options[:notify]}: #{e.message}" + end +end +summary = lambda do |error = nil| + return unless notifier && !options[:settings][:dry_run] + + report = DataPipeline::Report.new(DataPipeline.runs, title: name, params: DataPipeline.params, error:) + notifier.deliver(report) if notify_when.call(report) +rescue StandardError => e + warn "#{name}: notify #{options[:notify]}: #{e.class}: #{e.message}" +end begin script_lock = DataPipeline::Lock.new(options[:lock]).acquire if options[:lock] # held until the process ends @@ -72,9 +95,12 @@ begin load File.expand_path(script) rescue StandardError => e warn "#{File.basename(script)}: #{e.class}: #{e.message}" + # an aborted pipeline is already in the runs, other errors (script, lock) are reported as such + summary.call(DataPipeline.runs.any? { |stats| stats.error.equal?(e) } ? nil : e) unless e.is_a?(DataPipeline::LockError) exit(e.is_a?(DataPipeline::LockError) ? 3 : 1) end +summary.call runs = DataPipeline.runs exit 3 if runs.any? { |stats| stats.error.is_a?(DataPipeline::LockError) } exit 1 if runs.any?(&:error) diff --git a/lib/datapipeline.rb b/lib/datapipeline.rb index 1c9ca49..48fe82e 100644 --- a/lib/datapipeline.rb +++ b/lib/datapipeline.rb @@ -8,10 +8,11 @@ require_relative 'datapipeline/layout' require_relative 'datapipeline/watermark' require_relative 'datapipeline/lock' require_relative 'datapipeline/connections' +require_relative 'datapipeline/notification/notifier' class DataPipeline # plugins are loaded on first use, their gems too: pg (postgres), sqlite3 (sqlite), roo/caxlsx (excel), - # http (rest, slack) + # http (rest, slack, webhook), net-smtp (email) autoload :CsvStorage, File.expand_path('datapipeline/storage/csv', __dir__) autoload :PostgresStorage, File.expand_path('datapipeline/storage/postgres', __dir__) autoload :JsonStorage, File.expand_path('datapipeline/storage/json', __dir__) @@ -20,6 +21,8 @@ class DataPipeline autoload :RestStorage, File.expand_path('datapipeline/storage/rest', __dir__) autoload :FixedStorage, File.expand_path('datapipeline/storage/fixed', __dir__) autoload :Slack, File.expand_path('datapipeline/notification/slack', __dir__) + autoload :Webhook, File.expand_path('datapipeline/notification/webhook', __dir__) + autoload :Email, File.expand_path('datapipeline/notification/email', __dir__) # passed to header, trailer, map and filter # read: number of source rows (nil while streaming, until the trailer) @@ -240,6 +243,26 @@ class DataPipeline @reject_target = to end + # + # notify notifier, on: :problems -> a report after the run, see DataPipeline::Notifier + # on: :failure (abort), :problems (abort, rejected rows or warnings, default), :always + # title: default the name of the pipeline + # dry_run: true -> also after a dry run (default: not) + # include_rows: true -> rejected rows in the report (default: not, they may hold customer data) + # + # an error while sending is only logged + # + def notify notifier, on: :problems, title: nil, dry_run: false, include_rows: false + raise ArgumentError, 'notify: notifier needs deliver(report)' unless notifier.respond_to?(:deliver) + + (@notifiers ||= []) << { notifier:, when: Notifier.condition(on), title:, dry_run:, include_rows: } + end + + # notifier(:alerts) -> Webhook, Slack or Email from the connections file (type webhook, slack, email) + def notifier name + Notifier.from_connection(name) + end + # on_finish { |stats| ... } -> after every run, also after an abort (stats.error) def on_finish(&block) @on_finish << block @@ -484,6 +507,7 @@ class DataPipeline print_preview if @stats.dry_run && @stats.error.nil? @logger.info "finish #{@stats}" append_history + send_notifications @on_finish.each do |hook| hook.call(@stats) rescue StandardError => e @@ -504,6 +528,20 @@ class DataPipeline @logger.error "history: #{e.class}: #{e.message}" end + def send_notifications + Array(@notifiers).each do |entry| + next if @stats.dry_run && !entry[:dry_run] + + report = Report.new([@stats], title: entry[:title], params:) + next unless entry[:when].call(report) + + entry[:notifier].deliver(report, include_rows: entry[:include_rows]) + @logger.info "notified #{entry[:notifier].class.name.split('::').last}: #{report.event}" + rescue StandardError => e + @logger.error "notify #{entry[:notifier].class}: #{e.class}: #{e.message}" + end + end + def add_total side, row @stats.totals&.each do |col, sum| value = row.key?(col) ? row[col] : row[col.to_sym] diff --git a/lib/datapipeline/notification/email.rb b/lib/datapipeline/notification/email.rb new file mode 100644 index 0000000..bf1f721 --- /dev/null +++ b/lib/datapipeline/notification/email.rb @@ -0,0 +1,93 @@ +require 'securerandom' +require_relative 'notifier' +require_relative '../storage/storage_plugin' + +class DataPipeline + # + # e-mail over SMTP, text/plain UTF-8 + # + # notify DataPipeline::Email.new(to: 'etl@example.com'), on: :failure + # + # options (defaults from ENV) + # to: one address or an array (required) + # from: sender, default ENV['SMTP_FROM'] or etl@ + # host: ENV['SMTP_HOST'], default localhost + # port: ENV['SMTP_PORT'], default 587 (465 with tls: true, 25 without starttls) + # username: ENV['SMTP_USERNAME'], password: ENV['SMTP_PASSWORD']; without username no login + # tls: true -> TLS from the start (port 465) + # starttls: :auto (default, if the server offers it), :always or false + # subject_prefix: default '[etl] ' + # timeout: seconds for open and read (default 15) + # + class Email < Notifier + class Error < StandardError; end + + def initialize to:, from: nil, host: nil, port: nil, username: nil, password: nil, tls: false, + starttls: :auto, subject_prefix: '[etl] ', timeout: 15 + StoragePlugin.require_gem 'net/smtp', 'net-smtp' + @to = Array(to).map(&:to_s).reject(&:empty?) + raise ArgumentError, 'Email: to missing' if @to.empty? + + @from = from || ENV['SMTP_FROM'] || "etl@#{Socket.gethostname}" + @host = host || ENV.fetch('SMTP_HOST', 'localhost') + @tls = tls + @starttls = tls || !starttls ? false : starttls.to_sym + @port = Integer(port || ENV['SMTP_PORT'] || default_port) + @username = username || ENV['SMTP_USERNAME'] + @password = password || ENV['SMTP_PASSWORD'] + @subject_prefix = subject_prefix + @timeout = timeout + end + + def deliver report, include_rows: false + send_mail("#{@subject_prefix}#{report.subject}", body_of(report, include_rows)) + end + + # any text, e.g. from on_finish + def send_mail subject, body + message = build(subject, body) + smtp = Net::SMTP.new(@host, @port, tls: @tls, starttls: @starttls || false) + smtp.open_timeout = @timeout + smtp.read_timeout = @timeout + login = @username ? { user: @username, secret: @password, authtype: :login } : {} + smtp.start(helo: Socket.gethostname, **login) { |session| session.send_message(message, @from, *@to) } + message + rescue Net::SMTPError, SystemCallError, IOError, Timeout::Error, OpenSSL::SSL::SSLError => e + raise Error, "#{@host}:#{@port}: #{e.class}: #{e.message}" + end + + private + + def default_port + return 465 if @tls + + @starttls ? 587 : 25 + end + + # rows of rejected lines only on request, they may hold customer data + def body_of report, include_rows + text = report.to_s + return text unless include_rows + + rows = report.runs.flat_map { |stats| stats.errors.first(Report::MAX_ERRORS).map { |e| "#{e[:step]}: #{JSON.generate(e[:row])}" } } + rows.empty? ? text : "#{text}\n\nZeilen:\n#{rows.join("\n")}" + end + + # RFC 5322 with a base64 body, the subject RFC 2047 encoded when it is not ascii + def build subject, body + encoded_subject = subject.ascii_only? ? subject : "=?UTF-8?B?#{[subject].pack('m0')}?=" + [ + "From: #{@from}", + "To: #{@to.join(', ')}", + "Subject: #{encoded_subject}", + "Date: #{Time.now.strftime('%a, %d %b %Y %H:%M:%S %z')}", + "Message-ID: <#{SecureRandom.uuid}@#{@from.split('@').last}>", + 'MIME-Version: 1.0', + 'Content-Type: text/plain; charset=UTF-8', + 'Content-Transfer-Encoding: base64', + '', + [body].pack('m') + ].join("\r\n") + end + end +end diff --git a/lib/datapipeline/notification/notifier.rb b/lib/datapipeline/notification/notifier.rb new file mode 100644 index 0000000..f75c567 --- /dev/null +++ b/lib/datapipeline/notification/notifier.rb @@ -0,0 +1,141 @@ +require 'json' +require 'securerandom' +require 'socket' + +class DataPipeline + # + # what a notifier gets: the stats of one or more runs (one script), with event, title and text lines + # + # event: run.failed (an abort), run.rejected (rejected rows), run.warnings, run.succeeded + # to_h: stable payload for webhooks (schema 1); rows of stats.errors only with include_rows: true, + # because they may hold customer data + # + class Report + EVENTS = %w[run.succeeded run.warnings run.rejected run.failed].freeze + MAX_ERRORS = 20 + + attr_reader :id, :title, :runs, :params, :host, :error + + # error: an exception outside of a run (e.g. the script failed before or between pipelines) + def initialize runs, title: nil, params: {}, error: nil + @id = SecureRandom.uuid + @runs = Array(runs) + @title = title || @runs.filter_map(&:name).uniq.join(', ').then { |t| t.empty? ? 'datapipeline' : t } + @params = params.to_h + @host = Socket.gethostname + @error = error + end + + # the worst event of all runs + def event + return 'run.failed' if @error + + @runs.map { |stats| event_of(stats) }.max_by { |event| EVENTS.index(event) } || 'run.succeeded' + end + + def failed? = event == 'run.failed' + def ok? = event == 'run.succeeded' + + def status + { 'run.failed' => 'abgebrochen', 'run.rejected' => 'mit verworfenen Zeilen', + 'run.warnings' => 'mit Warnungen', 'run.succeeded' => 'ok' }.fetch(event) + end + + def subject = "#{@title}: #{status}" + + # text for mail, slack and logs: one block per run, the first errors without row data + def lines errors: 5 + lines = ["#{subject} (#{@host})"] + lines << "Parameter: #{@params.map { |key, value| "#{key}=#{value}" }.join(' ')}" if @params.any? + lines << "Fehler: #{@error.class}: #{@error.message}" if @error + @runs.each do |stats| + lines << '' + lines << "#{stats.name || 'Pipeline'}#{" #{stats.version}" if stats.version}: #{stats}" + lines << "Start #{stats.started_at&.strftime('%d.%m.%Y %H:%M:%S')}, Dauer #{stats.duration}s" + stats.errors.first(errors).each { |e| lines << "- #{e[:step]}: #{e[:error]}" } + lines << "- … und #{stats.rejected - errors} weitere" if stats.rejected > errors + end + lines + end + + def to_s = lines.join("\n") + + def to_h include_rows: false + { + schema: 1, id: @id, event:, title: @title, host: @host, params: @params, + error: @error && "#{@error.class}: #{@error.message}", + runs: @runs.map do |stats| + stats.to_h.merge(errors: stats.errors.first(MAX_ERRORS).map { |e| include_rows ? e : e.except(:row) }) + end + } + end + + private + + def event_of stats + return 'run.failed' if stats.error + return 'run.rejected' if stats.rejected.positive? + return 'run.warnings' if stats.warnings.to_i.positive? + + 'run.succeeded' + end + end + + # + # a notifier only needs deliver(report, include_rows: false); see Webhook, Slack, Email + # + # notify DataPipeline::Webhook.new(url: ...), on: :problems + # + class Notifier + # when a notification is sent + WHEN = { + always: ->(_report) { true }, + failure: ->(report) { report.failed? }, + problems: ->(report) { !report.ok? } + }.freeze + + TYPES = { webhook: 'DataPipeline::Webhook', slack: 'DataPipeline::Slack', email: 'DataPipeline::Email' }.freeze + + # notifier from the connections file: type webhook, slack or email, the other keys are its options + def self.from_connection name + options = Connections.fetch(name) + type = options[:type].to_s.to_sym + klass = TYPES.fetch(type) { raise ArgumentError, "connection #{name.inspect}: type #{type} is no notifier (#{TYPES.keys.join(', ')})" } + Object.const_get(klass).new(**options.except(:type)) + end + + def self.condition on + WHEN.fetch(on.to_sym) { raise ArgumentError, "notify on: #{on} unknown, use #{WHEN.keys.join(', ')}" } + end + + def deliver(_report, include_rows: false) + raise NotImplementedError, "#{self.class}#deliver is missing" + end + + private + + # POST with retries on 429/5xx (Retry-After honoured) and network errors; other 4xx raise at once + def post_with_retries url, body:, headers:, timeout:, retries:, retry_wait:, error: + attempt = 0 + loop do + attempt += 1 + begin + response = HTTP.timeout(timeout).headers(headers).post(url, body:) + rescue HTTP::Error, SystemCallError, IOError => e + raise error, "#{url}: #{e.class}: #{e.message}" if attempt > retries + + sleep(retry_wait * attempt) + next + end + return response if response.status.success? + + code = response.code + unless attempt <= retries && (code == 429 || code >= 500) + raise error, "#{url}: HTTP #{code} #{response.body.to_s[0, 200]}" + end + + sleep(response.headers['Retry-After']&.to_f || retry_wait * attempt) + end + end + end +end diff --git a/lib/datapipeline/notification/slack.rb b/lib/datapipeline/notification/slack.rb index 405d786..7193fd5 100644 --- a/lib/datapipeline/notification/slack.rb +++ b/lib/datapipeline/notification/slack.rb @@ -1,34 +1,88 @@ -require 'http' require 'json' +require_relative 'notifier' +require_relative '../storage/storage_plugin' class DataPipeline # # Slack.new(channel: '#etl').send_message('posting export done') + # notify DataPipeline::Slack.new(channel: '#etl'), on: :problems + # notify DataPipeline::Slack.new(webhook: ENV['SLACK_WEBHOOK']) # incoming webhook, no token needed # # options - # token: bot token, default ENV['SLACK_TOKEN'] - # channel: default '#general' + # token: bot token for chat.postMessage, default ENV['SLACK_TOKEN'] (not needed with webhook) + # channel: default '#general' (with webhook the channel of the webhook) + # webhook: url of an incoming webhook instead of token and channel + # mention: e.g. '' or '', only in reports of failed runs + # timeout: seconds per request (default 10) + # retries: repetitions on 429, 5xx and network errors (default 3) + # retry_wait: seconds, multiplied by the attempt (default 2) + # api_url: default https://slack.com/api/chat.postMessage # - class Slack + class Slack < Notifier API_URL = 'https://slack.com/api/chat.postMessage'.freeze + MAX_TEXT = 2900 # a section block holds 3000 characters + + class Error < StandardError; end DEFAULTS = { token: nil, - channel: '#general' + channel: '#general', + webhook: nil, + mention: nil, + timeout: 10, + retries: 3, + retry_wait: 2, + api_url: API_URL }.freeze + ICONS = { 'run.failed' => ':red_circle:', 'run.rejected' => ':large_orange_circle:', + 'run.warnings' => ':large_yellow_circle:', 'run.succeeded' => ':large_green_circle:' }.freeze + + # & < > have a meaning in slack texts + def self.escape(text) = text.to_s.gsub('&', '&').gsub('<', '<').gsub('>', '>') + def initialize options = {} + StoragePlugin.require_gem 'http' @options = DEFAULTS.merge(token: ENV['SLACK_TOKEN']).merge(options) - raise ArgumentError, 'Slack: token missing (option token: or ENV SLACK_TOKEN)' unless @options[:token] + return if @options[:webhook] || @options[:token] + + raise ArgumentError, 'Slack: token missing (option token: or ENV SLACK_TOKEN, or webhook:)' end - # returns the parsed slack response, raises when slack reports an error + # plain text (slack markup allowed); returns the parsed slack response, raises when slack reports an error def send_message(content, verbose: false) - response = HTTP.auth("Bearer #{@options[:token]}") - .post(API_URL, json: { channel: @options[:channel], text: content }) - body = JSON.parse(response.body.to_s) - puts JSON.pretty_generate(body) if verbose - raise "Slack#send_message failed: #{body['error']}" unless body['ok'] + post({ text: content }, verbose:) + end + + # the report as blocks: status line, then the runs as preformatted text + def deliver report, include_rows: false + icon = ICONS.fetch(report.event) + mention = report.failed? && @options[:mention] ? " #{@options[:mention]}" : '' + head = "#{icon} *#{self.class.escape(report.title)}*: #{report.status}#{mention}" + details = self.class.escape(report.lines.drop(1).join("\n").strip) + details = "#{details[0, MAX_TEXT]}\n…" if details.length > MAX_TEXT + blocks = [{ type: 'section', text: { type: 'mrkdwn', text: head } }] + blocks << { type: 'section', text: { type: 'mrkdwn', text: "```#{details}```" } } unless details.empty? + post({ text: "#{report.subject}#{mention}", blocks: }) + end + + private + + def post payload, verbose: false + url = @options[:webhook] || @options[:api_url] + headers = { 'Content-Type' => 'application/json; charset=utf-8' } + unless @options[:webhook] + headers['Authorization'] = "Bearer #{@options[:token]}" + payload = payload.merge(channel: @options[:channel]) + end + response = post_with_retries(url, body: JSON.generate(payload), headers:, timeout: @options[:timeout], + retries: @options[:retries], retry_wait: @options[:retry_wait], error: Error) + text = response.body.to_s + puts text if verbose + return text if @options[:webhook] # incoming webhooks answer "ok" + + body = JSON.parse(text) + raise Error, "Slack#send_message failed: #{body['error']}" unless body['ok'] body end diff --git a/lib/datapipeline/notification/webhook.rb b/lib/datapipeline/notification/webhook.rb new file mode 100644 index 0000000..b90ce80 --- /dev/null +++ b/lib/datapipeline/notification/webhook.rb @@ -0,0 +1,63 @@ +require 'openssl' +require_relative 'notifier' +require_relative '../storage/storage_plugin' + +class DataPipeline + # + # POST of the report as json (Report#to_h, schema 1) + # + # notify DataPipeline::Webhook.new(url: ENV['ALERT_URL'], secret: ENV['ALERT_SECRET']) + # + # options + # url: target (required) + # secret: signs the body: X-DataPipeline-Signature: sha256=HMAC_SHA256(secret, ".") + # with X-DataPipeline-Timestamp (unix seconds) + # headers: more headers, e.g. { 'Authorization' => 'Bearer ...' } + # timeout: seconds per request (default 10) + # retries: repetitions on 429, 5xx and network errors (default 3), Retry-After is honoured + # retry_wait: seconds, multiplied by the attempt (default 2) + # + # every request carries X-DataPipeline-Event and X-DataPipeline-Delivery (the report id, the same on retries) + # + class Webhook < Notifier + class Error < StandardError; end + + def initialize url:, secret: nil, headers: {}, timeout: 10, retries: 3, retry_wait: 2 + StoragePlugin.require_gem 'http' + @url = url.to_s + raise ArgumentError, 'Webhook: url missing' if @url.empty? + + @secret = secret + @headers = headers.to_h { |key, value| [key.to_s, value.to_s] } + @timeout = timeout + @retries = retries + @retry_wait = retry_wait + end + + def deliver report, include_rows: false + body = JSON.generate(report.to_h(include_rows:)) + post_with_retries(@url, body:, headers: headers_for(report, body), timeout: @timeout, retries: @retries, + retry_wait: @retry_wait, error: Error) + end + + # for receivers: true when signature and timestamp (max_age seconds) of a request are valid + def self.valid_signature? secret:, body:, timestamp:, signature:, max_age: 300 + return false if timestamp.to_s.empty? || (Time.now.to_i - timestamp.to_i).abs > max_age + + expected = "sha256=#{OpenSSL::HMAC.hexdigest('SHA256', secret, "#{timestamp}.#{body}")}" + expected.bytesize == signature.to_s.bytesize && OpenSSL.fixed_length_secure_compare(expected, signature.to_s) + end + + private + + def headers_for report, body + headers = { 'Content-Type' => 'application/json', 'User-Agent' => "datapipeline/#{VERSION}", + 'X-DataPipeline-Event' => report.event, 'X-DataPipeline-Delivery' => report.id }.merge(@headers) + return headers unless @secret + + timestamp = Time.now.to_i.to_s + signature = OpenSSL::HMAC.hexdigest('SHA256', @secret, "#{timestamp}.#{body}") + headers.merge('X-DataPipeline-Timestamp' => timestamp, 'X-DataPipeline-Signature' => "sha256=#{signature}") + end + end +end diff --git a/lib/datapipeline/version.rb b/lib/datapipeline/version.rb index b72bc1e..7e9ab17 100644 --- a/lib/datapipeline/version.rb +++ b/lib/datapipeline/version.rb @@ -1,5 +1,5 @@ class DataPipeline - VERSION = '3.1.0'.freeze + VERSION = '3.2.0'.freeze # kept for scripts that print the version (rake info:info); the history is in CHANGELOG.md class Version diff --git a/test/cli/etl_cli_test.rb b/test/cli/etl_cli_test.rb index c4608cf..e0ed56a 100644 --- a/test/cli/etl_cli_test.rb +++ b/test/cli/etl_cli_test.rb @@ -1,5 +1,6 @@ require_relative '../test_helper' require 'open3' +require_relative '../support/test_http_server' # bin/etl as a subprocess: arguments, params, dry run, exit codes class EtlCliTest < PipelineTest @@ -127,4 +128,44 @@ class EtlCliTest < PipelineTest assert_equal ['export_kunden', 3, true], entry.values_at('name', 'read', 'ok') end end + + def test_notify_one_report_for_all_pipelines_of_the_script + server = TestHttpServer.new { [200, 'ok'] } + with_tmpdir do |dir| + Dir.mkdir("#{dir}/config") + File.write("#{dir}/config/connections.yml", "alerts:\n type: webhook\n url: #{server.url}/hook\n") + path = script(dir, 'zwei.rb', <<~RUBY) + 2.times do |i| + DataPipeline.new do + name "teil\#{i}" + on_error :skip + read container(:csv, file: 'in.csv') + validate 'a' => { format: /\\A\\d+\\z/ } if i == 1 + write container(:csv, file: "out\#{i}.csv") + end.run + end + RUBY + _out, err, status = etl(dir, 'run', path, '--notify=alerts') + assert_equal 2, status.exitstatus, err + assert_equal 1, server.requests.size + payload = JSON.parse(server.requests.last[:body]) + assert_equal ['run.rejected', 'zwei', %w[teil0 teil1]], [payload['event'], payload['title'], payload['runs'].map { |r| r['name'] }] + + _out, err, status = etl(dir, 'run', path, '--notify=alerts', '--notify-on=failure') + assert_equal 2, status.exitstatus, err + assert_equal 1, server.requests.size # nothing failed + + broken = script(dir, 'kaputt.rb', "raise 'Skriptfehler'\n") + _out, _err, status = etl(dir, 'run', broken, '--notify=alerts') + assert_equal 1, status.exitstatus + payload = JSON.parse(server.requests.last[:body]) + assert_equal ['run.failed', 'RuntimeError: Skriptfehler', []], payload.values_at('event', 'error', 'runs') + + _out, err, status = etl(dir, 'run', path, '--notify=gibtsnicht') + assert_equal 1, status.exitstatus + assert_includes err, 'etl: --notify=gibtsnicht: connection "gibtsnicht" not in config/connections.yml (alerts)' + end + ensure + server&.stop + end end diff --git a/test/plugins/notification_test.rb b/test/plugins/notification_test.rb new file mode 100644 index 0000000..ad75cea --- /dev/null +++ b/test/plugins/notification_test.rb @@ -0,0 +1,206 @@ +require_relative '../test_helper' +require_relative '../support/test_http_server' +require_relative '../support/test_smtp_server' + +# Report, Webhook, Slack, Email and notify against local test servers +class NotificationTest < PipelineTest + ROWS = [{ 'id' => '1', 'konto' => '1000' }, { 'id' => '2', 'konto' => '' }].freeze + + # collects reports, for notify without network + class MemoryNotifier < DataPipeline::Notifier + attr_reader :reports + + def initialize(fail: false) = (@reports = []; @fail = fail) + + def deliver report, include_rows: false + raise 'kaputt' if @fail + + @reports << [report, include_rows] + end + end + + def teardown + @server&.stop + DataPipeline.connections_file = nil + super + end + + def stats_with_rejected + run_pipeline(ROWS.map(&:dup)) { name 'export'; on_error :skip; validate 'konto' => { required: true } }.last + end + + def report(stats = stats_with_rejected) = DataPipeline::Report.new([stats], params: { from: '2026-10-01' }) + + def test_report_event_lines_and_payload_without_rows + report = report() + assert_equal ['run.rejected', 'export: mit verworfenen Zeilen'], [report.event, report.subject] + assert_includes report.to_s, '- validate#1: konto: missing' + assert_includes report.to_s, 'Parameter: from=2026-10-01' + payload = report.to_h + assert_equal [1, 'run.rejected', 'export'], payload.values_at(:schema, :event, :title) + assert_equal [{ step: 'validate#1', error: 'konto: missing' }], payload[:runs].first[:errors] + assert_equal({ 'id' => '2', 'konto' => '' }, report.to_h(include_rows: true)[:runs].first[:errors].first[:row]) + end + + def test_report_worst_event_of_several_runs_and_script_error + ok = run_pipeline(ROWS.map(&:dup)).last + assert_equal 'run.succeeded', DataPipeline::Report.new([ok]).event + assert_equal 'run.rejected', DataPipeline::Report.new([ok, stats_with_rejected]).event + failed = DataPipeline::Report.new([ok], title: 'skript', error: RuntimeError.new('kaputt')) + assert_equal %w[run.failed skript], [failed.event, failed.title] + assert_includes failed.to_s, 'Fehler: RuntimeError: kaputt' + end + + def test_webhook_signed_post + @server = TestHttpServer.new { [200, 'ok'] } + report = report() + DataPipeline::Webhook.new(url: "#{@server.url}/hook", secret: 's3cret', headers: { 'X-Team' => 'etl' }).deliver(report) + request = @server.requests.last + headers = request[:headers] + assert_equal ['/hook', 'run.rejected', report.id, 'etl'], + [request[:path], headers['x-datapipeline-event'], headers['x-datapipeline-delivery'], headers['x-team']] + assert_equal 1, JSON.parse(request[:body])['schema'] + refute_includes request[:body], '"row"' + assert DataPipeline::Webhook.valid_signature?(secret: 's3cret', body: request[:body], + timestamp: headers['x-datapipeline-timestamp'], + signature: headers['x-datapipeline-signature']) + refute DataPipeline::Webhook.valid_signature?(secret: 'falsch', body: request[:body], + timestamp: headers['x-datapipeline-timestamp'], + signature: headers['x-datapipeline-signature']) + refute DataPipeline::Webhook.valid_signature?(secret: 's3cret', body: request[:body], timestamp: (Time.now.to_i - 600).to_s, + signature: headers['x-datapipeline-signature']) + end + + def test_webhook_retries_429_and_5xx_but_not_4xx + answers = [[503, 'busy'], [429, 'slow', { 'Retry-After' => '0' }], [200, 'ok']] + @server = TestHttpServer.new { answers.shift } + DataPipeline::Webhook.new(url: @server.url, retry_wait: 0).deliver(report) + assert_equal 3, @server.requests.size + assert_equal 1, @server.requests.map { |r| r[:headers]['x-datapipeline-delivery'] }.uniq.size + + @server.stop + @server = TestHttpServer.new { [400, 'bad'] } + error = assert_raises(DataPipeline::Webhook::Error) { DataPipeline::Webhook.new(url: @server.url, retry_wait: 0).deliver(report) } + assert_includes error.message, 'HTTP 400' + assert_equal 1, @server.requests.size + end + + def test_webhook_network_error_after_retries + port = TCPServer.new('127.0.0.1', 0).then { |s| s.addr[1].tap { s.close } } + error = assert_raises(DataPipeline::Webhook::Error) do + DataPipeline::Webhook.new(url: "http://127.0.0.1:#{port}", retries: 1, retry_wait: 0, timeout: 2).deliver(report) + end + assert_includes error.message, '127.0.0.1' + end + + def test_slack_api_blocks_escape_and_mention + @server = TestHttpServer.new { [200, { 'ok' => true }] } + slack = DataPipeline::Slack.new(token: 'xoxb-1', channel: '#etl', api_url: @server.url, mention: '') + failed = DataPipeline::Report.new([stats_with_rejected], title: 'A & ', error: RuntimeError.new('x < y')) + slack.deliver(failed) + request = @server.requests.last + body = JSON.parse(request[:body]) + assert_equal ['Bearer xoxb-1', '#etl'], [request[:headers]['authorization'], body['channel']] + assert_equal ':red_circle: *A & <B>*: abgebrochen ', body['blocks'][0]['text']['text'] + assert_includes body['blocks'][1]['text']['text'], 'RuntimeError: x < y' + + slack.deliver(report) + refute_includes JSON.parse(@server.requests.last[:body])['blocks'][0]['text']['text'], '' # mention only on failure + end + + def test_slack_error_and_incoming_webhook + @server = TestHttpServer.new { |req| req[:path] == '/api' ? [200, { 'ok' => false, 'error' => 'not_in_channel' }] : [200, 'ok'] } + error = assert_raises(DataPipeline::Slack::Error) { DataPipeline::Slack.new(token: 't', api_url: "#{@server.url}/api").send_message('x') } + assert_includes error.message, 'not_in_channel' + + token = ENV.delete('SLACK_TOKEN') + hook = DataPipeline::Slack.new(webhook: "#{@server.url}/hook") + assert_equal 'ok', hook.deliver(report) + request = @server.requests.last + assert_nil request[:headers]['authorization'] + refute JSON.parse(request[:body]).key?('channel') + ensure + ENV['SLACK_TOKEN'] = token if token + end + + def test_email_over_smtp_with_login + @server = TestSmtpServer.new + mail = DataPipeline::Email.new(to: %w[a@example.com b@example.com], from: 'etl@example.com', host: '127.0.0.1', + port: @server.port, starttls: false, username: 'user', password: 'geheim') + mail.deliver(DataPipeline::Report.new([stats_with_rejected], title: 'Grösse Ä')) + sent = @server.mails.last + assert_equal ['etl@example.com', %w[a@example.com b@example.com], %w[user geheim]], sent.values_at(:from, :to, :auth) + head, body = sent[:data].split("\r\n\r\n", 2) + assert_includes head, "Subject: =?UTF-8?B?#{['[etl] Grösse Ä: mit verworfenen Zeilen'].pack('m0')}?=" + assert_includes head, 'Content-Type: text/plain; charset=UTF-8' + text = body.unpack1('m').force_encoding('UTF-8') + assert_includes text, '- validate#1: konto: missing' + refute_includes text, 'Zeilen:' + end + + def test_email_rows_on_request_and_errors + @server = TestSmtpServer.new + mail = DataPipeline::Email.new(to: 'a@example.com', host: '127.0.0.1', port: @server.port, starttls: false) + mail.deliver(report, include_rows: true) + assert_includes @server.mails.last[:data].split("\r\n\r\n", 2).last.unpack1('m').force_encoding('UTF-8'), '{"id":"2","konto":""}' + assert_nil @server.mails.last[:auth] + + assert_raises(ArgumentError) { DataPipeline::Email.new(to: []) } + port = TCPServer.new('127.0.0.1', 0).then { |s| s.addr[1].tap { s.close } } + assert_raises(DataPipeline::Email::Error) { DataPipeline::Email.new(to: 'a@b', host: '127.0.0.1', port:, starttls: false).send_mail('x', 'y') } + end + + def test_notify_conditions_dry_run_and_rows + always = MemoryNotifier.new + failure = MemoryNotifier.new + problems = MemoryNotifier.new + rows = MemoryNotifier.new + run_pipeline(ROWS.map(&:dup)) do + name 'export' + on_error :skip + validate 'konto' => { required: true } + notify always, on: :always + notify failure, on: :failure + notify problems + notify rows, include_rows: true, title: 'Export Konten' + end + assert_equal [1, 0, 1], [always, failure, problems].map { |n| n.reports.size } + assert_equal ['Export Konten', true], [rows.reports.first[0].title, rows.reports.first[1]] + + dry = MemoryNotifier.new + dry_too = MemoryNotifier.new + capture_stdout { run_pipeline(ROWS.map(&:dup), dry_run: true) { notify dry, on: :always; notify dry_too, on: :always, dry_run: true } } + assert_equal [0, 1], [dry.reports.size, dry_too.reports.size] + + assert_raises(ArgumentError) { run_pipeline { notify Object.new } } + assert_raises(ArgumentError) { run_pipeline { notify MemoryNotifier.new, on: :sometimes } } + end + + def test_notify_error_is_only_logged + log = StringIO.new + stats = DataPipeline.new do + logger Logger.new(log) + read MemoryStorage.new(rows: ROWS.map(&:dup)) + write MemoryStorage.new + notify MemoryNotifier.new(fail: true), on: :always + end.run + assert stats.ok? + assert_includes log.string, 'notify NotificationTest::MemoryNotifier: RuntimeError: kaputt' + end + + def test_notifier_from_connections_file + with_tmpdir do |dir| + File.write("#{dir}/c.yml", <<~YAML) + alerts: { type: webhook, url: "http://127.0.0.1:9/hook", secret: "<%= 'abc' %>" } + mail: { type: email, to: [etl@example.com], host: smtp.example.com } + db: { type: sqlite, database: x.sqlite3 } + YAML + DataPipeline.connections_file = "#{dir}/c.yml" + pipeline = DataPipeline.new { logger PipelineHelper::QUIET } + assert_kind_of DataPipeline::Webhook, pipeline.notifier(:alerts) + assert_equal 'abc', pipeline.notifier(:alerts).instance_variable_get(:@secret) + assert_kind_of DataPipeline::Email, pipeline.notifier(:mail) + assert_includes assert_raises(ArgumentError) { pipeline.notifier(:db) }.message, 'is no notifier' + end + end +end diff --git a/test/support/test_http_server.rb b/test/support/test_http_server.rb index ff95a38..fffd2ab 100644 --- a/test/support/test_http_server.rb +++ b/test/support/test_http_server.rb @@ -5,7 +5,7 @@ require 'json' # # minimal http server for DataPipeline::RestStorage tests, one request per connection # server = TestHttpServer.new { |req| [200, { 'items' => [] }] } -# req: { method:, path:, query:, headers:, body: } -> [status, hash/array or string] +# req: { method:, path:, query:, headers:, body: } -> [status, hash/array or string, optional headers] # class TestHttpServer attr_reader :requests @@ -40,9 +40,10 @@ class TestHttpServer request = { method:, path:, query: URI.decode_www_form(query.to_s).to_h, headers:, body: } @requests << request - status, payload = @handler.call(request) + status, payload, extra = @handler.call(request) text = payload.is_a?(String) ? payload : JSON.generate(payload) - client.write "HTTP/1.1 #{status} X\r\nContent-Type: application/json\r\n" \ + extra = Hash(extra).map { |key, value| "#{key}: #{value}\r\n" }.join + client.write "HTTP/1.1 #{status} X\r\nContent-Type: application/json\r\n#{extra}" \ "Content-Length: #{text.bytesize}\r\nConnection: close\r\n\r\n#{text}" rescue StandardError => e warn "TestHttpServer: #{e.class}: #{e.message}" diff --git a/test/support/test_smtp_server.rb b/test/support/test_smtp_server.rb new file mode 100644 index 0000000..3527061 --- /dev/null +++ b/test/support/test_smtp_server.rb @@ -0,0 +1,59 @@ +require 'socket' + +# +# minimal smtp server for DataPipeline::Email tests: EHLO, AUTH LOGIN, MAIL, RCPT, DATA, QUIT +# server = TestSmtpServer.new +# server.mails -> [{ from:, to: [...], data:, auth: [user, password] }] +# +class TestSmtpServer + attr_reader :mails + + def initialize + @mails = [] + @server = TCPServer.new('127.0.0.1', 0) + @thread = Thread.new { loop { serve(@server.accept) } } + end + + def port = @server.addr[1] + + def stop + @thread.kill + @server.close + end + + private + + def serve client + mail = { to: [] } + client.write "220 test ESMTP\r\n" + while (line = client.gets) + case line + when /\AEHLO/i then client.write "250-test\r\n250 AUTH LOGIN PLAIN\r\n" + when /\AAUTH LOGIN/i + client.write "334 VXNlcm5hbWU6\r\n" + user = client.gets.unpack1('m') + client.write "334 UGFzc3dvcmQ6\r\n" + mail[:auth] = [user, client.gets.unpack1('m')] + client.write "235 ok\r\n" + when /\AMAIL FROM:<(.*)>/i then mail[:from] = $1; client.write "250 ok\r\n" + when /\ARCPT TO:<(.*)>/i then mail[:to] << $1; client.write "250 ok\r\n" + when /\ADATA/i + client.write "354 go\r\n" + data = +'' + while (part = client.gets) && part != ".\r\n" + data << part + end + mail[:data] = data + @mails << mail + mail = { to: [] } + client.write "250 queued\r\n" + when /\AQUIT/i then client.write "221 bye\r\n"; break + else client.write "250 ok\r\n" + end + end + rescue StandardError => e + warn "TestSmtpServer: #{e.class}: #{e.message}" + ensure + client.close + end +end