This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12279-f5b7f33e6d19e9c3ebee14828f7eb40ae641ecca in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 1f29461a75e16882447435a6282fccca1e14e576 Author: Goutam Adwant <[email protected]> AuthorDate: Tue Sep 15 05:48:55 2026 +0000 [Feature][Connector-V2] Add RFC 5424 Syslog sink over TLS (#12279) Signed-off-by: Goutam Adwant <[email protected]> --- .github/workflows/labeler/label-scope-conf.yml | 7 + config/plugin_config | 1 + docs/en/connectors/changelog/connector-syslog.md | 20 ++ docs/en/connectors/sink/Syslog.md | 170 +++++++++ docs/zh/connectors/changelog/connector-syslog.md | 20 ++ docs/zh/connectors/sink/Syslog.md | 166 +++++++++ plugin-mapping.properties | 1 + .../connector-syslog/examples/fake_to_syslog.conf | 50 +++ seatunnel-connectors-v2/connector-syslog/pom.xml | 35 ++ .../seatunnel/syslog/config/SyslogSinkConfig.java | 65 ++++ .../seatunnel/syslog/config/SyslogSinkOptions.java | 76 ++++ .../syslog/sink/SyslogMessageEncoder.java | 241 +++++++++++++ .../seatunnel/syslog/sink/SyslogSink.java | 57 +++ .../seatunnel/syslog/sink/SyslogSinkFactory.java | 73 ++++ .../seatunnel/syslog/sink/SyslogSinkWriter.java | 86 +++++ .../seatunnel/syslog/sink/SyslogTlsClient.java | 304 ++++++++++++++++ .../syslog/sink/SyslogMessageEncoderTest.java | 306 ++++++++++++++++ .../syslog/sink/SyslogSinkFactoryTest.java | 141 ++++++++ .../syslog/sink/SyslogSinkWriterTest.java | 240 +++++++++++++ .../seatunnel/syslog/sink/SyslogTlsClientTest.java | 395 +++++++++++++++++++++ seatunnel-connectors-v2/pom.xml | 1 + seatunnel-dist/pom.xml | 6 + .../connector-syslog-e2e/pom.xml | 40 +++ .../e2e/connector/syslog/SyslogSinkIT.java | 232 ++++++++++++ .../src/test/resources/docker/syslog-ng.conf | 29 ++ .../src/test/resources/fake_to_syslog.conf | 59 +++ seatunnel-e2e/seatunnel-connector-v2-e2e/pom.xml | 1 + 27 files changed, 2822 insertions(+) diff --git a/.github/workflows/labeler/label-scope-conf.yml b/.github/workflows/labeler/label-scope-conf.yml index c5a2a21157..5e46ce070f 100644 --- a/.github/workflows/labeler/label-scope-conf.yml +++ b/.github/workflows/labeler/label-scope-conf.yml @@ -356,6 +356,13 @@ socket: - changed-files: - any-glob-to-any-file: seatunnel-connectors-v2/connector-socket/** - all-globs-to-all-files: '!seatunnel-connectors-v2/connector-!(socket)/**' +syslog: + - all: + - changed-files: + - any-glob-to-any-file: + - seatunnel-connectors-v2/connector-syslog/** + - seatunnel-e2e/seatunnel-connector-v2-e2e/connector-syslog-e2e/** + - all-globs-to-all-files: '!seatunnel-connectors-v2/connector-!(syslog)/**' starrocks: - all: - changed-files: diff --git a/config/plugin_config b/config/plugin_config index 8e2b4992d2..0a912b6d30 100644 --- a/config/plugin_config +++ b/config/plugin_config @@ -107,6 +107,7 @@ connector-s3-redshift connector-sentry connector-slack connector-socket +connector-syslog connector-edge-socket connector-starrocks connector-tablestore diff --git a/docs/en/connectors/changelog/connector-syslog.md b/docs/en/connectors/changelog/connector-syslog.md new file mode 100644 index 0000000000..16f00488ff --- /dev/null +++ b/docs/en/connectors/changelog/connector-syslog.md @@ -0,0 +1,20 @@ +<!-- +Licensed to the Apache Software Foundation (ASF) under one or more +contributor license agreements. See the NOTICE file distributed with +this work for additional information regarding copyright ownership. +The ASF licenses this file to You under the Apache License, Version 2.0 +(the "License"); you may not use this file except in compliance with +the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +--> + +# Changelog + +Changes will be recorded after the connector is released. diff --git a/docs/en/connectors/sink/Syslog.md b/docs/en/connectors/sink/Syslog.md new file mode 100644 index 0000000000..9f70ecf18b --- /dev/null +++ b/docs/en/connectors/sink/Syslog.md @@ -0,0 +1,170 @@ +import ChangeLog from '../changelog/connector-syslog.md'; + +# Syslog + +<!-- +Licensed to the Apache Software Foundation (ASF) under one or more +contributor license agreements. See the NOTICE file distributed with +this work for additional information regarding copyright ownership. +The ASF licenses this file to You under the Apache License, Version 2.0 +(the "License"); you may not use this file except in compliance with +the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +--> + +> RFC 5424 sink over TLS, with RFC 5425 octet-counted framing. + +## Description + +Sends INSERT rows to a TLS syslog receiver. This sink does not listen for syslog messages. +RFC 3164, plaintext TCP, UDP, RELP, transactions and automatic reconnect/retry are not supported. + +## Support Those Engines + +> Spark<br/> +> Flink<br/> +> SeaTunnel Zeta<br/> + +## Key Features + +- [x] [batch](../../introduction/concepts/connector-v2-features.md) +- [x] [stream](../../introduction/concepts/connector-v2-features.md) +- [x] [parallelism](../../introduction/concepts/connector-v2-features.md) +- [ ] [exactly-once](../../introduction/concepts/connector-v2-features.md) + +## Supported DataSource Info + +A receiver supporting RFC 5424 over TLS with RFC 5425 framing, for example a syslog-ng +`syslog(transport("tls"))` source. A legacy newline-framed TCP listener is not compatible. +No additional client library is required; the connector uses JDK TLS. +Set the message-size cap at or below the receiver's limit; not every receiver accepts 8192 bytes. + +## Sink Options + +| Name | Type | Required | Default | Description | +|------|------|----------|---------|-------------| +| host | String | Yes | - | Receiver DNS name or IP address, without scheme or port. Must match the certificate. | +| port | Int | No | 6514 | Receiver port, 1-65535. | +| connect_timeout_ms | Int | No | 10000 | Positive TCP connect timeout. JVM/platform DNS resolution is outside this deadline. | +| write_timeout_ms | Int | No | 10000 | Positive deadline for each TLS handshake, write/flush, and close operation. | +| max_message_bytes | Int | No | 8192 | Maximum encoded message size, 1-1048576 bytes, including header, structured data and UTF-8 BOM but excluding length prefix. Oversize messages fail; they are not truncated. | +| tls.ca_cert_path | String | No | - | Worker-local PEM CA certificate bundle. Absent means JVM default trust anchors. | +| tls.key_store.path | String | No | - | Optional worker-local client key store for mutual TLS. | +| password | String | With key store | - | Password for both client key store and private key. | +| tls.key_store.type | String | No | PKCS12 | Client store type: `PKCS12` or `JKS`. | +| common-options | | No | - | [Common Sink Options](../common-options/sink-common-options.md), including `plugin_input`. | + +## Input Schema + +Column names are fixed; rename columns with a transform when needed. Extra columns are ignored. +If an optional column is present, it must have the listed type. Schema validation does not access the network or TLS files. + +| Column | Type | Absent or null | Contract | +|--------|------|----------------|----------| +| message | STRING | Null omits MSG; column is required | UTF-8 text. Empty text is distinct from null. Non-null MSG includes the UTF-8 BOM. | +| facility | INT | 1 (user-level) | 0-23. PRI is facility * 8 + severity. | +| severity | INT | 6 (informational) | 0-7. | +| timestamp | STRING | `-` | RFC 5424 timestamp with uppercase T/Z or numeric offset; at most six fractional digits. Invalid dates, leap seconds and local timestamps are rejected. No timestamp is invented. | +| hostname | STRING | `-` | 1-255 printable US-ASCII characters, no spaces. | +| app_name | STRING | `-` | 1-48 printable US-ASCII characters, no spaces. | +| proc_id | STRING | `-` | 1-128 printable US-ASCII characters, no spaces. | +| msg_id | STRING | `-` | 1-32 printable US-ASCII characters, no spaces. | +| structured_data | MAP<STRING, MAP<STRING, STRING>> | `-` (also for empty map) | SD-ID to parameter map. No raw structured-data strings. | + +SD-ID and parameter names contain 1-32 printable ASCII characters, excluding space, `=`, `]` and `"`. +Parameter values must be non-null strings; empty parameter maps are allowed. +Quotes, backslashes and closing brackets in parameter values are escaped. +Use IANA-registered SD-IDs or your own private-enterprise-number suffix; `example@32473` is for documentation only. +Malformed Unicode is rejected. Newlines and other control characters in MSG and parameter values are preserved. +Octet framing keeps them inside a single message, but receiver storage/display policies may alter them. +Header control characters and spaces are rejected rather than escaped. +RFC 3164 source timestamps such as `Oct 11 22:14:15` require an explicit conversion before use here. + +## Delivery and Resource Limits + +- Each writer opens one TLS connection and sends one message synchronously, without an asynchronous message queue. + Only one write is in flight per writer. The byte cap bounds each frame and intermediate encoding buffers. +- A successful write or checkpoint flush means only that the local TLS/socket operation completed. + Syslog has no application-level acknowledgment of receiver parsing, durable storage, or downstream delivery. + This sink promises neither at-least-once nor exactly-once delivery. Failures/recovery can cause loss or duplicates. +- Partial writes, timeouts and detected disconnects permanently fail that writer. No message is silently retried. + A new engine task may replay input on recovery; the receiver can then observe duplicates. +- Both prepare-commit hooks, snapshot and close flush/check the connection and propagate transport failures. + There is no transactional commit or recoverable writer state. +- A watchdog closes the underlying TCP socket to interrupt stalled TLS output. Socket read timeout alone is not a write timeout. + Concurrent close aborts an active write. Interrupting the writing thread may take up to the operation deadline to unblock it. + DNS resolution still follows JVM/platform resolver limits; configure those separately. +- Keep operation deadlines below the job checkpoint timeout. Parallel writers have separate connections and no global ordering. + A returned local write cannot detect every peer failure, including a receiver that discards data after accepting it. + +## TLS and Security + +TLS is mandatory. Only TLS 1.2 and TLS 1.3 supported by the worker JDK are enabled. +The receiver chain is verified with the configured CA bundle or JVM trust anchors and the configured host +is checked using JSSE HTTPS endpoint identification. There is no trust-all or hostname-validation bypass. +Do not change JVM-global trust settings to configure this connector. + +Deploy trust/client key files to every worker. Restrict access to client private keys, and supply +`password` through configuration substitution or the deployment's secret management. +Never commit real passwords. Mutual TLS requires both the key-store path and password; the receiver must trust that client certificate. +TLS authenticates the transport peer, not the `hostname`/application identity supplied in a row. + +## Example + +Configure the receiver to accept RFC 5424 over TLS with octet counting, then replace the endpoint and CA path below. +The runnable example is `seatunnel-connectors-v2/connector-syslog/examples/fake_to_syslog.conf`. + +```hocon +env { + parallelism = 1 + job.mode = "BATCH" +} +source { + FakeSource { + row.num = 1 + schema.fields { + facility = int + severity = int + timestamp = string + hostname = string + app_name = string + proc_id = string + msg_id = string + structured_data = "map<string, map<string, string>>" + message = string + } + rows = [{ + kind = INSERT + fields = [1, 6, "2026-01-02T03:04:05.123Z", "origin", "seatunnel", "-", "ID47", + {"example@32473": {"component": "pipeline"}}, "job completed"] + }] + } +} +sink { + Syslog { + host = "syslog.example.org" + port = 6514 + tls.ca_cert_path = "/etc/seatunnel/syslog-ca.pem" + connect_timeout_ms = 10000 + write_timeout_ms = 10000 + max_message_bytes = 8192 + } +} +``` + +Optional client authentication: + +```hocon +tls.key_store.path = "/etc/seatunnel/syslog-client.p12" +tls.key_store.type = "PKCS12" +password = ${SYSLOG_KEY_STORE_PASSWORD} +``` + +<ChangeLog /> diff --git a/docs/zh/connectors/changelog/connector-syslog.md b/docs/zh/connectors/changelog/connector-syslog.md new file mode 100644 index 0000000000..0f999406aa --- /dev/null +++ b/docs/zh/connectors/changelog/connector-syslog.md @@ -0,0 +1,20 @@ +<!-- +Licensed to the Apache Software Foundation (ASF) under one or more +contributor license agreements. See the NOTICE file distributed with +this work for additional information regarding copyright ownership. +The ASF licenses this file to You under the Apache License, Version 2.0 +(the "License"); you may not use this file except in compliance with +the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +--> + +# 更新日志 + +连接器发布后记录变更。 diff --git a/docs/zh/connectors/sink/Syslog.md b/docs/zh/connectors/sink/Syslog.md new file mode 100644 index 0000000000..67b4222a11 --- /dev/null +++ b/docs/zh/connectors/sink/Syslog.md @@ -0,0 +1,166 @@ +import ChangeLog from '../changelog/connector-syslog.md'; + +# Syslog + +<!-- +Licensed to the Apache Software Foundation (ASF) under one or more +contributor license agreements. See the NOTICE file distributed with +this work for additional information regarding copyright ownership. +The ASF licenses this file to You under the Apache License, Version 2.0 +(the "License"); you may not use this file except in compliance with +the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +--> + +> 使用 TLS 发送 RFC 5424 消息，采用 RFC 5425 字节计数分帧。 + +## 描述 + +将 INSERT 行发送到 TLS Syslog 接收端。本连接器仅为 Sink，不监听或接收 Syslog。 +不支持 RFC 3164、明文 TCP、UDP、RELP、事务或自动重连重试。 + +## 支持的引擎 + +> Spark<br/> +> Flink<br/> +> SeaTunnel Zeta<br/> + +## 主要特性 + +- [x] [批处理](../../introduction/concepts/connector-v2-features.md) +- [x] [流处理](../../introduction/concepts/connector-v2-features.md) +- [x] [并行度](../../introduction/concepts/connector-v2-features.md) +- [ ] [精确一次](../../introduction/concepts/connector-v2-features.md) + +## 支持的数据源 + +接收端必须支持 RFC 5424、TLS 和 RFC 5425 分帧，例如 syslog-ng 的 +`syslog(transport("tls"))` Source。仅支持换行分隔的旧式 TCP 监听器不兼容。 +连接器使用 JDK TLS，不需要额外客户端库。 +消息字节上限不得超过接收端限制，并非所有接收端都接受 8192 字节。 + +## Sink 选项 + +| 名称 | 类型 | 必需 | 默认值 | 描述 | +|------|------|------|--------|------| +| host | String | 是 | - | 接收端 DNS 名称或 IP，不含协议或端口，必须与证书匹配。 | +| port | Int | 否 | 6514 | 接收端端口，范围 1-65535。 | +| connect_timeout_ms | Int | 否 | 10000 | TCP 连接超时毫秒数，必须为正数。不包括 JVM/平台 DNS 解析时间。 | +| write_timeout_ms | Int | 否 | 10000 | 每次 TLS 握手、写入/刷新和关闭的超时毫秒数，必须为正数。 | +| max_message_bytes | Int | 否 | 8192 | 编码后的消息上限，范围 1-1048576 字节。包括头、结构化数据和 UTF-8 BOM，不含长度前缀。超限报错，不截断。 | +| tls.ca_cert_path | String | 否 | - | Worker 本地 PEM CA 证书集合；未配置时使用 JVM 默认信任证书。 | +| tls.key_store.path | String | 否 | - | 双向 TLS 使用的 Worker 本地客户端密钥库。 | +| password | String | 配置密钥库时 | - | 密钥库及私钥密码。 | +| tls.key_store.type | String | 否 | PKCS12 | 客户端密钥库类型：`PKCS12` 或 `JKS`。 | +| common-options | | 否 | - | [通用 Sink 选项](../common-options/sink-common-options.md)，包括 `plugin_input`。 | + +## 输入 Schema + +列名固定，可用 Transform 重命名。其他列被忽略。可选列一旦存在，必须使用指定类型。 +Schema 校验不会访问网络或读取 TLS 文件。 + +| 列 | 类型 | 缺失或 null 时 | 约束 | +|----|------|---------------|------| +| message | STRING | 列必需；null 不发送 MSG | UTF-8 文本；空字符串与 null 不同。非 null MSG 带 UTF-8 BOM。 | +| facility | INT | 1（用户级） | 0-23，PRI = facility * 8 + severity。 | +| severity | INT | 6（信息） | 0-7。 | +| timestamp | STRING | `-` | RFC 5424 时间戳，使用大写 T/Z 或数字时区偏移，最多六位小数秒。不接受非法日期、闰秒或无时区时间。不自动生成时间戳。 | +| hostname | STRING | `-` | 1-255 个不含空格的可打印 US-ASCII 字符。 | +| app_name | STRING | `-` | 1-48 个不含空格的可打印 US-ASCII 字符。 | +| proc_id | STRING | `-` | 1-128 个不含空格的可打印 US-ASCII 字符。 | +| msg_id | STRING | `-` | 1-32 个不含空格的可打印 US-ASCII 字符。 | +| structured_data | MAP<STRING, MAP<STRING, STRING>> | `-`，空 Map 也相同 | SD-ID 到参数 Map 的映射，不接受原始结构化数据字符串。 | + +SD-ID 和参数名为 1-32 个可打印 ASCII 字符，不得含空格、`=`、`]`、`"`。 +参数值必须是非 null 字符串，允许空参数 Map。参数值中的引号、反斜杠和右方括号会被转义。 +请使用 IANA 注册的 SD-ID 或带自身企业编号的标识；`example@32473` 仅供文档示例使用。 +非法 Unicode 被拒绝。MSG 和参数值中的换行及控制字符被保留，字节分帧使它们仍属于同一消息， +但接收端的存储或显示策略可能修改这些字符。头字段中的空格及控制字符被拒绝，不进行转义。 +RFC 3164 Source 的 `Oct 11 22:14:15` 等时间戳必须先显式转换。 + +## 交付语义和资源限制 + +- 每个 Writer 使用一个 TLS 连接，同步发送消息，没有异步消息队列。同一 Writer 最多有一次写入进行中。 + 字节上限限制每个帧及中间编码缓冲区。 +- 写入成功或检查点刷新成功仅表示本地 TLS/Socket 操作完成。 + Syslog 不提供对接收端解析、持久化或下游交付的应用级确认。 + 本 Sink 不保证至少一次或精确一次；故障和恢复可能造成丢失或重复。 +- 部分写入、超时或检测到断连会永久终止该 Writer，不会静默重试。 + 引擎恢复可能重新执行输入，从而在接收端产生重复消息。 +- 两个 prepare-commit 方法、snapshot 和 close 都刷新或检查连接并传播传输错误。 + 不提供事务提交或可恢复的 Writer 状态。 +- 超时监控直接关闭底层 TCP Socket 以中断阻塞的 TLS 写入，不能仅用读超时来限制写入。 + 并发 close 会中断进行中的操作；中断写入线程可能需要等待至当前操作的超时期限。 + DNS 解析受 JVM/平台解析器限制，需单独配置。 +- 操作超时应小于作业检查点超时。并行 Writer 使用不同连接，不保证全局顺序。 + 本地写入成功无法发现所有接收端故障，例如接收端接收后丢弃消息。 + +## TLS 和安全 + +必须使用 TLS，仅启用 Worker JDK 支持的 TLS 1.2 和 TLS 1.3。 +使用配置的 CA 集合或 JVM 默认信任证书验证接收端证书链，并通过 JSSE HTTPS 端点标识校验 host。 +不提供信任全部证书或跳过主机名校验的选项，也不修改 JVM 全局 TLS 设置。 + +信任证书和客户端密钥库必须部署到所有 Worker。限制私钥文件的访问权限，通过配置替换或部署环境的 +密钥管理提供 `password`，不要提交真实密码。双向 TLS 必须同时配置密钥库路径和密码， +接收端也必须信任该客户端证书。TLS 认证的是传输对端，不是输入行中的 hostname 或应用标识。 + +## 示例 + +接收端应配置为使用 TLS 接收 RFC 5424 字节计数帧。替换以下端点和 CA 路径后运行。 +示例文件：`seatunnel-connectors-v2/connector-syslog/examples/fake_to_syslog.conf`。 + +```hocon +env { + parallelism = 1 + job.mode = "BATCH" +} +source { + FakeSource { + row.num = 1 + schema.fields { + facility = int + severity = int + timestamp = string + hostname = string + app_name = string + proc_id = string + msg_id = string + structured_data = "map<string, map<string, string>>" + message = string + } + rows = [{ + kind = INSERT + fields = [1, 6, "2026-01-02T03:04:05.123Z", "origin", "seatunnel", "-", "ID47", + {"example@32473": {"component": "pipeline"}}, "job completed"] + }] + } +} +sink { + Syslog { + host = "syslog.example.org" + port = 6514 + tls.ca_cert_path = "/etc/seatunnel/syslog-ca.pem" + connect_timeout_ms = 10000 + write_timeout_ms = 10000 + max_message_bytes = 8192 + } +} +``` + +可选客户端认证： + +```hocon +tls.key_store.path = "/etc/seatunnel/syslog-client.p12" +tls.key_store.type = "PKCS12" +password = ${SYSLOG_KEY_STORE_PASSWORD} +``` + +<ChangeLog /> diff --git a/plugin-mapping.properties b/plugin-mapping.properties index 538a91f225..79313b987a 100644 --- a/plugin-mapping.properties +++ b/plugin-mapping.properties @@ -70,6 +70,7 @@ seatunnel.sink.FtpFile = connector-file-ftp seatunnel.source.SftpFile = connector-file-sftp seatunnel.sink.SftpFile = connector-file-sftp seatunnel.sink.Socket = connector-socket +seatunnel.sink.Syslog = connector-syslog seatunnel.source.Redis = connector-redis seatunnel.sink.Redis = connector-redis seatunnel.sink.Databend = connector-databend diff --git a/seatunnel-connectors-v2/connector-syslog/examples/fake_to_syslog.conf b/seatunnel-connectors-v2/connector-syslog/examples/fake_to_syslog.conf new file mode 100644 index 0000000000..3e28426924 --- /dev/null +++ b/seatunnel-connectors-v2/connector-syslog/examples/fake_to_syslog.conf @@ -0,0 +1,50 @@ +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +env { + parallelism = 1 + job.mode = "BATCH" +} +source { + FakeSource { + row.num = 1 + schema.fields { + facility = int + severity = int + timestamp = string + hostname = string + app_name = string + proc_id = string + msg_id = string + structured_data = "map<string, map<string, string>>" + message = string + } + rows = [{ + kind = INSERT + fields = [1, 6, "2026-01-02T03:04:05.123Z", "origin", "seatunnel", "-", "ID47", + {"example@32473": {"component": "pipeline"}}, "job completed"] + }] + } +} +sink { + Syslog { + host = "syslog.example.org" + port = 6514 + tls.ca_cert_path = "/etc/seatunnel/syslog-ca.pem" + connect_timeout_ms = 10000 + write_timeout_ms = 10000 + max_message_bytes = 8192 + } +} diff --git a/seatunnel-connectors-v2/connector-syslog/pom.xml b/seatunnel-connectors-v2/connector-syslog/pom.xml new file mode 100644 index 0000000000..716ee4ec39 --- /dev/null +++ b/seatunnel-connectors-v2/connector-syslog/pom.xml @@ -0,0 +1,35 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- + Licensed to the Apache Software Foundation (ASF) under one or more + contributor license agreements. See the NOTICE file distributed with + this work for additional information regarding copyright ownership. + The ASF licenses this file to You under the Apache License, Version 2.0 + (the "License"); you may not use this file except in compliance with + the License. You may obtain a copy of the License at + http://www.apache.org/licenses/LICENSE-2.0 + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +--> +<project xmlns="http://maven.apache.org/POM/4.0.0"; xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; + xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd";> + <modelVersion>4.0.0</modelVersion> + <parent> + <groupId>org.apache.seatunnel</groupId> + <artifactId>seatunnel-connectors-v2</artifactId> + <version>${revision}</version> + </parent> + + <artifactId>connector-syslog</artifactId> + <name>SeaTunnel : Connectors V2 : Syslog</name> + + <dependencies> + <dependency> + <groupId>org.apache.seatunnel</groupId> + <artifactId>connector-common</artifactId> + <version>${project.version}</version> + </dependency> + </dependencies> +</project> diff --git a/seatunnel-connectors-v2/connector-syslog/src/main/java/org/apache/seatunnel/connectors/seatunnel/syslog/config/SyslogSinkConfig.java b/seatunnel-connectors-v2/connector-syslog/src/main/java/org/apache/seatunnel/connectors/seatunnel/syslog/config/SyslogSinkConfig.java new file mode 100644 index 0000000000..5bab911ed8 --- /dev/null +++ b/seatunnel-connectors-v2/connector-syslog/src/main/java/org/apache/seatunnel/connectors/seatunnel/syslog/config/SyslogSinkConfig.java @@ -0,0 +1,65 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.syslog.config; + +import org.apache.seatunnel.api.configuration.ReadonlyConfig; +import org.apache.seatunnel.api.configuration.util.ConfigValidator; +import org.apache.seatunnel.connectors.seatunnel.syslog.sink.SyslogSinkFactory; + +import lombok.Getter; + +import java.io.Serializable; + +/** Serializable configuration only; TLS resources are opened on the worker. */ +@Getter +public final class SyslogSinkConfig implements Serializable { + private static final long serialVersionUID = 1L; + private final String host; + private final int port; + private final int connectTimeout; + private final int writeTimeout; + private final int maxMessageBytes; + private final String caCert; + private final String keyStore; + private final String keyStorePassword; + private final String keyStoreType; + + public SyslogSinkConfig(ReadonlyConfig config) { + ConfigValidator.of(config).validate(new SyslogSinkFactory().optionRule()); + host = config.get(SyslogSinkOptions.HOST); + if (!host.equals(host.trim()) || host.chars().anyMatch(c -> c <= 32 || c >= 127)) { + throw new IllegalArgumentException( + "Syslog host must be an ASCII hostname or IP address without whitespace"); + } + port = config.get(SyslogSinkOptions.PORT); + connectTimeout = config.get(SyslogSinkOptions.CONNECT_TIMEOUT); + writeTimeout = config.get(SyslogSinkOptions.WRITE_TIMEOUT); + maxMessageBytes = config.get(SyslogSinkOptions.MAX_MESSAGE_BYTES); + caCert = config.getOptional(SyslogSinkOptions.CA_CERT).orElse(null); + keyStore = config.getOptional(SyslogSinkOptions.KEY_STORE).orElse(null); + keyStorePassword = config.getOptional(SyslogSinkOptions.KEY_STORE_PASSWORD).orElse(null); + keyStoreType = config.get(SyslogSinkOptions.KEY_STORE_TYPE); + if (!"PKCS12".equals(keyStoreType) && !"JKS".equals(keyStoreType)) { + throw new IllegalArgumentException("Syslog tls.key_store.type must be PKCS12 or JKS"); + } + if ((keyStore == null) != (keyStorePassword == null)) { + throw new IllegalArgumentException( + "Syslog tls.key_store.path and password must be configured together"); + } + } +} diff --git a/seatunnel-connectors-v2/connector-syslog/src/main/java/org/apache/seatunnel/connectors/seatunnel/syslog/config/SyslogSinkOptions.java b/seatunnel-connectors-v2/connector-syslog/src/main/java/org/apache/seatunnel/connectors/seatunnel/syslog/config/SyslogSinkOptions.java new file mode 100644 index 0000000000..4bb5bd9d44 --- /dev/null +++ b/seatunnel-connectors-v2/connector-syslog/src/main/java/org/apache/seatunnel/connectors/seatunnel/syslog/config/SyslogSinkOptions.java @@ -0,0 +1,76 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.syslog.config; + +import org.apache.seatunnel.api.configuration.Option; +import org.apache.seatunnel.api.configuration.Options; + +public final class SyslogSinkOptions { + public static final String IDENTIFIER = "Syslog"; + public static final Option<String> HOST = + Options.key("host") + .stringType() + .noDefaultValue() + .withDescription("TLS receiver hostname, verified against its certificate."); + public static final Option<Integer> PORT = + Options.key("port") + .intType() + .defaultValue(6514) + .withDescription("TLS receiver port (1-65535)."); + public static final Option<Integer> CONNECT_TIMEOUT = + Options.key("connect_timeout_ms") + .intType() + .defaultValue(10000) + .withDescription( + "TCP connect timeout in milliseconds, excluding JVM DNS resolution."); + public static final Option<Integer> WRITE_TIMEOUT = + Options.key("write_timeout_ms") + .intType() + .defaultValue(10000) + .withDescription( + "Deadline in milliseconds for a TLS handshake, write/flush, or close."); + public static final Option<Integer> MAX_MESSAGE_BYTES = + Options.key("max_message_bytes") + .intType() + .defaultValue(8192) + .withDescription( + "Maximum UTF-8 syslog message bytes, including header and BOM, excluding framing (1-1048576)."); + public static final Option<String> CA_CERT = + Options.key("tls.ca_cert_path") + .stringType() + .noDefaultValue() + .withDescription( + "Worker-local PEM CA bundle. When absent, use the JVM default trust anchors."); + public static final Option<String> KEY_STORE = + Options.key("tls.key_store.path") + .stringType() + .noDefaultValue() + .withDescription("Optional worker-local client key store for mutual TLS."); + public static final Option<String> KEY_STORE_PASSWORD = + Options.key("password") + .stringType() + .noDefaultValue() + .withDescription("Password for the client key store and private key."); + public static final Option<String> KEY_STORE_TYPE = + Options.key("tls.key_store.type") + .stringType() + .defaultValue("PKCS12") + .withDescription("Client key store type: PKCS12 or JKS."); + + private SyslogSinkOptions() {} +} diff --git a/seatunnel-connectors-v2/connector-syslog/src/main/java/org/apache/seatunnel/connectors/seatunnel/syslog/sink/SyslogMessageEncoder.java b/seatunnel-connectors-v2/connector-syslog/src/main/java/org/apache/seatunnel/connectors/seatunnel/syslog/sink/SyslogMessageEncoder.java new file mode 100644 index 0000000000..190577cb07 --- /dev/null +++ b/seatunnel-connectors-v2/connector-syslog/src/main/java/org/apache/seatunnel/connectors/seatunnel/syslog/sink/SyslogMessageEncoder.java @@ -0,0 +1,241 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.syslog.sink; + +import org.apache.seatunnel.api.table.type.BasicType; +import org.apache.seatunnel.api.table.type.MapType; +import org.apache.seatunnel.api.table.type.RowKind; +import org.apache.seatunnel.api.table.type.SeaTunnelDataType; +import org.apache.seatunnel.api.table.type.SeaTunnelRow; +import org.apache.seatunnel.api.table.type.SeaTunnelRowType; + +import java.nio.ByteBuffer; +import java.nio.CharBuffer; +import java.nio.charset.CharacterCodingException; +import java.nio.charset.CodingErrorAction; +import java.nio.charset.StandardCharsets; +import java.time.LocalDateTime; +import java.time.format.DateTimeParseException; +import java.util.HashMap; +import java.util.Map; +import java.util.regex.Pattern; + +/** RFC 5424 messages, followed by RFC 5425 octet-counted framing. */ +final class SyslogMessageEncoder { + private static final Pattern TIMESTAMP = + Pattern.compile( + "[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}(\\.[0-9]{1,6})?(Z|[+-]([01][0-9]|2[0-3]):[0-5][0-9])"); + private static final MapType<String, Map<String, String>> STRUCTURED_DATA_TYPE = + new MapType<>( + BasicType.STRING_TYPE, + new MapType<>(BasicType.STRING_TYPE, BasicType.STRING_TYPE)); + private final Map<String, Integer> indexes = new HashMap<>(); + private final int arity; + private final int maxBytes; + + SyslogMessageEncoder(SeaTunnelRowType rowType, int maxBytes) { + this.arity = rowType.getTotalFields(); + this.maxBytes = maxBytes; + String[] names = rowType.getFieldNames(); + for (int i = 0; i < names.length; i++) { + if (indexes.put(names[i], i) != null) { + throw invalid("schema", "duplicate column names"); + } + } + requireType(rowType, "message", BasicType.STRING_TYPE, true); + for (String field : + new String[] {"timestamp", "hostname", "app_name", "proc_id", "msg_id"}) { + requireType(rowType, field, BasicType.STRING_TYPE, false); + } + requireType(rowType, "facility", BasicType.INT_TYPE, false); + requireType(rowType, "severity", BasicType.INT_TYPE, false); + requireType(rowType, "structured_data", STRUCTURED_DATA_TYPE, false); + } + + private void requireType( + SeaTunnelRowType type, String name, SeaTunnelDataType<?> expected, boolean required) { + Integer index = indexes.get(name); + if (index == null) { + if (required) { + throw invalid(name, "required column is missing"); + } + } else if (!expected.equals(type.getFieldType(index))) { + throw invalid(name, "expected " + expected); + } + } + + byte[] encode(SeaTunnelRow row) { + if (row == null || row.getArity() != arity) { + throw invalid("row", "arity does not match the catalog schema"); + } + if (row.getRowKind() != RowKind.INSERT) { + throw invalid("row", "only INSERT rows are supported"); + } + StringBuilder message = new StringBuilder(Math.min(maxBytes, 1024)); + append( + message, + "<" + (number(row, "facility", 1, 23) * 8 + number(row, "severity", 6, 7)) + ">1 "); + append(message, timestamp(string(row, "timestamp"))); + append(message, " " + header(string(row, "hostname"), "hostname", 255)); + append(message, " " + header(string(row, "app_name"), "app_name", 48)); + append(message, " " + header(string(row, "proc_id"), "proc_id", 128)); + append(message, " " + header(string(row, "msg_id"), "msg_id", 32) + " "); + structuredData(message, value(row, "structured_data")); + String text = string(row, "message"); + if (text != null) { + append(message, " \uFEFF"); + append(message, text); + } + try { + // The character cap above bounds intermediate allocations; UTF-8 may take more bytes. + ByteBuffer encoded = + StandardCharsets.UTF_8 + .newEncoder() + .onMalformedInput(CodingErrorAction.REPORT) + .onUnmappableCharacter(CodingErrorAction.REPORT) + .encode(CharBuffer.wrap(message)); + if (encoded.remaining() > maxBytes) { + throw invalid("max_message_bytes", "encoded message exceeds limit"); + } + byte[] prefix = (encoded.remaining() + " ").getBytes(StandardCharsets.US_ASCII); + byte[] frame = new byte[prefix.length + encoded.remaining()]; + System.arraycopy(prefix, 0, frame, 0, prefix.length); + encoded.get(frame, prefix.length, encoded.remaining()); + return frame; + } catch (CharacterCodingException e) { + throw invalid("message/structured_data", "malformed Unicode"); + } + } + + private Object value(SeaTunnelRow row, String name) { + Integer index = indexes.get(name); + return index == null ? null : row.getField(index); + } + + private String string(SeaTunnelRow row, String name) { + Object value = value(row, name); + if (value != null && !(value instanceof String)) { + throw invalid(name, "expected STRING value"); + } + return (String) value; + } + + private int number(SeaTunnelRow row, String name, int fallback, int max) { + Object value = value(row, name); + if (value == null) { + return fallback; + } + if (!(value instanceof Integer) || (int) value < 0 || (int) value > max) { + throw invalid(name, "expected INT between 0 and " + max); + } + return (int) value; + } + + private static String header(String value, String name, int max) { + if (value == null) { + return "-"; + } + if (value.isEmpty() || value.length() > max) { + throw invalid(name, "length must be between 1 and " + max); + } + for (int i = 0; i < value.length(); i++) { + if (value.charAt(i) < 33 || value.charAt(i) > 126) { + throw invalid(name, "only printable US-ASCII without spaces is allowed"); + } + } + return value; + } + + private static String timestamp(String value) { + if (value == null || "-".equals(value)) { + return "-"; + } + if (value.length() > 32 || !TIMESTAMP.matcher(value).matches()) { + throw invalid( + "timestamp", + "expected RFC 5424 timestamp with an offset and at most six fractional digits"); + } + try { + // Validate calendar/time strictly, retaining RFC offsets up to 23:59 (not ZoneOffset's + // 18h cap). + int end = value.endsWith("Z") ? value.length() - 1 : value.length() - 6; + LocalDateTime.parse(value.substring(0, end)); + } catch (DateTimeParseException e) { + throw invalid("timestamp", "invalid calendar date or time"); + } + return value; + } + + private void structuredData(StringBuilder target, Object value) { + if (value == null || (value instanceof Map && ((Map<?, ?>) value).isEmpty())) { + append(target, "-"); + return; + } + if (!(value instanceof Map)) { + throw invalid("structured_data", "expected MAP<STRING, MAP<STRING, STRING>>"); + } + for (Map.Entry<?, ?> element : ((Map<?, ?>) value).entrySet()) { + append(target, "[" + sdName(element.getKey())); + if (!(element.getValue() instanceof Map)) { + throw invalid("structured_data", "each SD-ID must map to a non-null parameter map"); + } + for (Map.Entry<?, ?> parameter : ((Map<?, ?>) element.getValue()).entrySet()) { + append(target, " " + sdName(parameter.getKey()) + "=\""); + if (!(parameter.getValue() instanceof String)) { + throw invalid("structured_data", "parameter values must be non-null STRINGs"); + } + String text = (String) parameter.getValue(); + if (text.length() > maxBytes) { + throw invalid("max_message_bytes", "structured data exceeds limit"); + } + for (int i = 0; i < text.length(); i++) { + char c = text.charAt(i); + if (c == '\\' || c == '"' || c == ']') { + append(target, "\\"); + } + append(target, String.valueOf(c)); + } + append(target, "\""); + } + append(target, "]"); + } + } + + private static String sdName(Object value) { + if (!(value instanceof String)) { + throw invalid("structured_data", "SD-ID and parameter names must be non-null STRINGs"); + } + String name = header((String) value, "structured_data", 32); + if (name.indexOf('=') >= 0 || name.indexOf(']') >= 0 || name.indexOf('"') >= 0) { + throw invalid("structured_data", "invalid SD-ID or parameter name"); + } + return name; + } + + private void append(StringBuilder target, String value) { + if (value.length() > maxBytes - target.length()) { + throw invalid("max_message_bytes", "message exceeds limit"); + } + target.append(value); + } + + private static IllegalArgumentException invalid(String field, String reason) { + // Never include row contents: syslog payloads can contain credentials or personal data. + return new IllegalArgumentException("Syslog " + field + ": " + reason); + } +} diff --git a/seatunnel-connectors-v2/connector-syslog/src/main/java/org/apache/seatunnel/connectors/seatunnel/syslog/sink/SyslogSink.java b/seatunnel-connectors-v2/connector-syslog/src/main/java/org/apache/seatunnel/connectors/seatunnel/syslog/sink/SyslogSink.java new file mode 100644 index 0000000000..aebd08f57e --- /dev/null +++ b/seatunnel-connectors-v2/connector-syslog/src/main/java/org/apache/seatunnel/connectors/seatunnel/syslog/sink/SyslogSink.java @@ -0,0 +1,57 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.syslog.sink; + +import org.apache.seatunnel.api.sink.SinkWriter; +import org.apache.seatunnel.api.table.catalog.CatalogTable; +import org.apache.seatunnel.api.table.type.SeaTunnelRow; +import org.apache.seatunnel.connectors.seatunnel.common.sink.AbstractSimpleSink; +import org.apache.seatunnel.connectors.seatunnel.common.sink.AbstractSinkWriter; +import org.apache.seatunnel.connectors.seatunnel.syslog.config.SyslogSinkConfig; +import org.apache.seatunnel.connectors.seatunnel.syslog.config.SyslogSinkOptions; + +import java.io.IOException; +import java.util.Optional; + +public final class SyslogSink extends AbstractSimpleSink<SeaTunnelRow, Void> { + private final SyslogSinkConfig config; + private final CatalogTable table; + + public SyslogSink(SyslogSinkConfig config, CatalogTable table) { + new SyslogMessageEncoder(table.getSeaTunnelRowType(), config.getMaxMessageBytes()); + this.config = config; + this.table = table; + } + + @Override + public String getPluginName() { + return SyslogSinkOptions.IDENTIFIER; + } + + /** Each restored writer opens a new connection, without replaying transport-level writes. */ + @Override + public AbstractSinkWriter<SeaTunnelRow, Void> createWriter(SinkWriter.Context context) + throws IOException { + return new SyslogSinkWriter(config, table.getSeaTunnelRowType()); + } + + @Override + public Optional<CatalogTable> getWriteCatalogTable() { + return Optional.of(table); + } +} diff --git a/seatunnel-connectors-v2/connector-syslog/src/main/java/org/apache/seatunnel/connectors/seatunnel/syslog/sink/SyslogSinkFactory.java b/seatunnel-connectors-v2/connector-syslog/src/main/java/org/apache/seatunnel/connectors/seatunnel/syslog/sink/SyslogSinkFactory.java new file mode 100644 index 0000000000..9008f45efe --- /dev/null +++ b/seatunnel-connectors-v2/connector-syslog/src/main/java/org/apache/seatunnel/connectors/seatunnel/syslog/sink/SyslogSinkFactory.java @@ -0,0 +1,73 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.syslog.sink; + +import org.apache.seatunnel.api.configuration.util.Conditions; +import org.apache.seatunnel.api.configuration.util.OptionRule; +import org.apache.seatunnel.api.table.connector.TableSink; +import org.apache.seatunnel.api.table.factory.Factory; +import org.apache.seatunnel.api.table.factory.TableSinkFactory; +import org.apache.seatunnel.api.table.factory.TableSinkFactoryContext; +import org.apache.seatunnel.connectors.seatunnel.syslog.config.SyslogSinkConfig; +import org.apache.seatunnel.connectors.seatunnel.syslog.config.SyslogSinkOptions; + +import com.google.auto.service.AutoService; + +@AutoService(Factory.class) +public final class SyslogSinkFactory implements TableSinkFactory { + @Override + public String factoryIdentifier() { + return SyslogSinkOptions.IDENTIFIER; + } + + @Override + public OptionRule optionRule() { + return OptionRule.builder() + .required(SyslogSinkOptions.HOST, Conditions.notBlank(SyslogSinkOptions.HOST)) + .optional( + SyslogSinkOptions.PORT, + Conditions.greaterOrEqual(SyslogSinkOptions.PORT, 1) + .and(Conditions.lessOrEqual(SyslogSinkOptions.PORT, 65535))) + .optional( + SyslogSinkOptions.CONNECT_TIMEOUT, + Conditions.greaterThan(SyslogSinkOptions.CONNECT_TIMEOUT, 0)) + .optional( + SyslogSinkOptions.WRITE_TIMEOUT, + Conditions.greaterThan(SyslogSinkOptions.WRITE_TIMEOUT, 0)) + .optional( + SyslogSinkOptions.MAX_MESSAGE_BYTES, + Conditions.greaterThan(SyslogSinkOptions.MAX_MESSAGE_BYTES, 0) + .and( + Conditions.lessOrEqual( + SyslogSinkOptions.MAX_MESSAGE_BYTES, 1048576))) + .optional(SyslogSinkOptions.CA_CERT, Conditions.notBlank(SyslogSinkOptions.CA_CERT)) + .optional( + SyslogSinkOptions.KEY_STORE, + Conditions.notBlank(SyslogSinkOptions.KEY_STORE)) + .optional(SyslogSinkOptions.KEY_STORE_PASSWORD, SyslogSinkOptions.KEY_STORE_TYPE) + .build(); + } + + @Override + public TableSink createSink(TableSinkFactoryContext context) { + SyslogSink sink = + new SyslogSink( + new SyslogSinkConfig(context.getOptions()), context.getCatalogTable()); + return () -> sink; + } +} diff --git a/seatunnel-connectors-v2/connector-syslog/src/main/java/org/apache/seatunnel/connectors/seatunnel/syslog/sink/SyslogSinkWriter.java b/seatunnel-connectors-v2/connector-syslog/src/main/java/org/apache/seatunnel/connectors/seatunnel/syslog/sink/SyslogSinkWriter.java new file mode 100644 index 0000000000..c8d00877b6 --- /dev/null +++ b/seatunnel-connectors-v2/connector-syslog/src/main/java/org/apache/seatunnel/connectors/seatunnel/syslog/sink/SyslogSinkWriter.java @@ -0,0 +1,86 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.syslog.sink; + +import org.apache.seatunnel.api.table.type.SeaTunnelRow; +import org.apache.seatunnel.api.table.type.SeaTunnelRowType; +import org.apache.seatunnel.connectors.seatunnel.common.sink.AbstractSinkWriter; +import org.apache.seatunnel.connectors.seatunnel.syslog.config.SyslogSinkConfig; + +import java.io.IOException; +import java.util.Collections; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.locks.ReentrantLock; + +public final class SyslogSinkWriter extends AbstractSinkWriter<SeaTunnelRow, Void> { + private final SyslogMessageEncoder encoder; + private final SyslogTlsClient client; + private final ReentrantLock writeLock = new ReentrantLock(); + + SyslogSinkWriter(SyslogSinkConfig config, SeaTunnelRowType rowType) throws IOException { + encoder = new SyslogMessageEncoder(rowType, config.getMaxMessageBytes()); + client = SyslogTlsClient.connect(config); + } + + SyslogSinkWriter(SyslogMessageEncoder encoder, SyslogTlsClient client) { + this.encoder = encoder; + this.client = client; + } + + @Override + public void write(SeaTunnelRow row) throws IOException { + if (!writeLock.tryLock()) { + throw new IOException("Syslog concurrent writes are not supported"); + } + try { + client.write(encoder.encode(row)); + } finally { + writeLock.unlock(); + } + } + + /** Writes are synchronous; this is a local flush, not an acknowledgment from the receiver. */ + @Override + public Optional<Void> prepareCommit() { + try { + client.flush(); + return Optional.empty(); + } catch (IOException e) { + // AbstractSinkWriter's legacy overload cannot declare IOException. + throw new java.io.UncheckedIOException(e); + } + } + + @Override + public Optional<Void> prepareCommit(long checkpointId) throws IOException { + client.flush(); + return Optional.empty(); + } + + @Override + public List<Void> snapshotState(long checkpointId) throws IOException { + client.flush(); + return Collections.emptyList(); + } + + @Override + public void close() throws IOException { + client.close(); + } +} diff --git a/seatunnel-connectors-v2/connector-syslog/src/main/java/org/apache/seatunnel/connectors/seatunnel/syslog/sink/SyslogTlsClient.java b/seatunnel-connectors-v2/connector-syslog/src/main/java/org/apache/seatunnel/connectors/seatunnel/syslog/sink/SyslogTlsClient.java new file mode 100644 index 0000000000..30b0a40ae7 --- /dev/null +++ b/seatunnel-connectors-v2/connector-syslog/src/main/java/org/apache/seatunnel/connectors/seatunnel/syslog/sink/SyslogTlsClient.java @@ -0,0 +1,304 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.syslog.sink; + +import org.apache.seatunnel.connectors.seatunnel.syslog.config.SyslogSinkConfig; + +import javax.net.ssl.KeyManager; +import javax.net.ssl.KeyManagerFactory; +import javax.net.ssl.SSLContext; +import javax.net.ssl.SSLParameters; +import javax.net.ssl.SSLSocket; +import javax.net.ssl.TrustManagerFactory; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.net.InetSocketAddress; +import java.net.Socket; +import java.net.SocketTimeoutException; +import java.nio.file.Files; +import java.nio.file.Paths; +import java.security.GeneralSecurityException; +import java.security.KeyStore; +import java.security.cert.Certificate; +import java.security.cert.CertificateFactory; +import java.util.Arrays; +import java.util.Collection; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.locks.ReentrantLock; + +/** One synchronous in-flight operation; errors permanently poison the connection. */ +final class SyslogTlsClient implements AutoCloseable { + private final Socket rawSocket; + private final SSLSocket tlsSocket; + private final OutputStream output; + private final int timeout; + private final ScheduledThreadPoolExecutor watchdog; + private final ReentrantLock operationLock = new ReentrantLock(); + private final AtomicBoolean closed = new AtomicBoolean(); + private volatile IOException failure; + + static SyslogTlsClient connect(SyslogSinkConfig config) throws IOException { + SSLContext context = sslContext(config); + Socket raw = new Socket(); + SyslogTlsClient client = null; + try { + // JVM DNS resolution has its own platform limits, outside Socket.connect's deadline. + raw.connect( + new InetSocketAddress(config.getHost(), config.getPort()), + config.getConnectTimeout()); + raw.setTcpNoDelay(true); + raw.setSoTimeout(config.getWriteTimeout()); + SSLSocket tls = + (SSLSocket) + context.getSocketFactory() + .createSocket(raw, config.getHost(), config.getPort(), true); + SSLParameters parameters = tls.getSSLParameters(); + parameters.setEndpointIdentificationAlgorithm("HTTPS"); + parameters.setProtocols( + Arrays.stream(tls.getSupportedProtocols()) + .filter(p -> "TLSv1.2".equals(p) || "TLSv1.3".equals(p)) + .toArray(String[]::new)); + tls.setSSLParameters(parameters); + client = new SyslogTlsClient(raw, tls, tls.getOutputStream(), config.getWriteTimeout()); + client.execute("TLS handshake", tls::startHandshake); + return client; + } catch (IOException | RuntimeException e) { + if (client != null) { + client.abort(); + } else { + try { + raw.close(); + } catch (IOException closeError) { + e.addSuppressed(closeError); + } + } + throw e; + } + } + + // Package-private transport boundary also permits deterministic partial-write/timeout tests. + SyslogTlsClient(Socket rawSocket, SSLSocket tlsSocket, OutputStream output, int timeout) { + this.rawSocket = rawSocket; + this.tlsSocket = tlsSocket; + this.output = output; + this.timeout = timeout; + watchdog = + new ScheduledThreadPoolExecutor( + 1, + runnable -> { + Thread thread = new Thread(runnable, "syslog-sink-write-deadline"); + thread.setDaemon(true); + return thread; + }); + watchdog.setRemoveOnCancelPolicy(true); + } + + void write(byte[] frame) throws IOException { + execute( + "write", + () -> { + output.write(frame); + output.flush(); + }); + } + + void flush() throws IOException { + execute("flush", output::flush); + } + + private void execute(String phase, IoAction action) throws IOException { + if (!operationLock.tryLock()) { + throw new IOException("Syslog concurrent transport operations are not supported"); + } + try { + checkOpen(); + runBounded(phase, action); + checkOpen(); + } finally { + operationLock.unlock(); + } + } + + private void checkOpen() throws IOException { + if (failure != null) { + throw new IOException( + "Syslog writer previously failed; delivery is uncertain", failure); + } + if (closed.get()) { + throw new IOException("Syslog writer is closed"); + } + if (Thread.currentThread().isInterrupted()) { + failure = new IOException("Syslog operation interrupted"); + abort(); + throw failure; + } + } + + private void runBounded(String phase, IoAction action) throws IOException { + // SO_TIMEOUT only bounds reads. Close the underlying TCP socket, not SSLSocket (which + // may wait for the blocked TLS write lock), to interrupt a stalled output operation. + AtomicInteger status = new AtomicInteger(0); + ScheduledFuture<?> deadline = null; + try { + deadline = + watchdog.schedule( + () -> { + if (status.compareAndSet(0, 2)) { + closeRaw(); + } + }, + timeout, + TimeUnit.MILLISECONDS); + action.run(); + if (!status.compareAndSet(0, 1)) { + throw new SocketTimeoutException( + "Syslog " + phase + " timed out; delivery is uncertain"); + } + if (Thread.currentThread().isInterrupted()) { + throw new IOException("Syslog " + phase + " interrupted; delivery is uncertain"); + } + } catch (IOException | RuntimeException e) { + boolean timedOut = status.get() == 2 || e instanceof SocketTimeoutException; + IOException error = + timedOut + ? new SocketTimeoutException( + "Syslog " + phase + " timed out; delivery is uncertain") + : new IOException( + "Syslog " + phase + " failed; delivery is uncertain; no retry", + e); + if (timedOut) { + error.initCause(e); + } + failure = error; + abort(); + throw error; + } finally { + status.compareAndSet(0, 1); + if (deadline != null) { + deadline.cancel(false); + } + } + } + + /** A concurrent close aborts an active write immediately instead of waiting for its lock. */ + @Override + public void close() throws IOException { + if (!closed.compareAndSet(false, true)) { + if (failure != null) { + throw new IOException( + "Syslog writer previously failed; delivery is uncertain", failure); + } + return; + } + if (!operationLock.tryLock()) { + failure = + new IOException( + "Syslog closed during an active operation; delivery is uncertain"); + abort(); + throw failure; + } + try { + if (failure != null) { + throw new IOException( + "Syslog writer previously failed; delivery is uncertain", failure); + } + runBounded( + "close", + () -> { + output.flush(); + tlsSocket.close(); + }); + } finally { + abort(); + operationLock.unlock(); + } + if (failure != null) { + throw new IOException("Syslog TCP close failed", failure); + } + } + + private void abort() { + closed.set(true); + closeRaw(); + watchdog.shutdownNow(); + } + + private void closeRaw() { + try { + rawSocket.close(); + } catch (IOException e) { + if (failure == null) { + failure = new IOException("Syslog TCP close failed", e); + } + } + } + + private static SSLContext sslContext(SyslogSinkConfig config) throws IOException { + try { + KeyStore trustStore = null; + if (config.getCaCert() != null) { + trustStore = KeyStore.getInstance(KeyStore.getDefaultType()); + trustStore.load(null, null); + try (InputStream input = Files.newInputStream(Paths.get(config.getCaCert()))) { + Collection<? extends Certificate> certificates = + CertificateFactory.getInstance("X.509").generateCertificates(input); + if (certificates.isEmpty()) { + throw new IOException("Syslog tls.ca_cert_path contains no certificates"); + } + int index = 0; + for (Certificate certificate : certificates) { + trustStore.setCertificateEntry("ca-" + index++, certificate); + } + } + } + TrustManagerFactory trust = + TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm()); + trust.init(trustStore); + KeyManager[] keys = null; + if (config.getKeyStore() != null) { + char[] password = config.getKeyStorePassword().toCharArray(); + try (InputStream input = Files.newInputStream(Paths.get(config.getKeyStore()))) { + KeyStore store = KeyStore.getInstance(config.getKeyStoreType()); + store.load(input, password); + KeyManagerFactory keyFactory = + KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm()); + keyFactory.init(store, password); + keys = keyFactory.getKeyManagers(); + } finally { + Arrays.fill(password, '\0'); + } + } + SSLContext context = SSLContext.getInstance("TLS"); + context.init(keys, trust.getTrustManagers(), null); + return context; + } catch (GeneralSecurityException e) { + throw new IOException("Syslog TLS material could not be loaded", e); + } + } + + @FunctionalInterface + private interface IoAction { + void run() throws IOException; + } +} diff --git a/seatunnel-connectors-v2/connector-syslog/src/test/java/org/apache/seatunnel/connectors/seatunnel/syslog/sink/SyslogMessageEncoderTest.java b/seatunnel-connectors-v2/connector-syslog/src/test/java/org/apache/seatunnel/connectors/seatunnel/syslog/sink/SyslogMessageEncoderTest.java new file mode 100644 index 0000000000..f86b757a45 --- /dev/null +++ b/seatunnel-connectors-v2/connector-syslog/src/test/java/org/apache/seatunnel/connectors/seatunnel/syslog/sink/SyslogMessageEncoderTest.java @@ -0,0 +1,306 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.syslog.sink; + +import org.apache.seatunnel.api.table.type.BasicType; +import org.apache.seatunnel.api.table.type.MapType; +import org.apache.seatunnel.api.table.type.RowKind; +import org.apache.seatunnel.api.table.type.SeaTunnelDataType; +import org.apache.seatunnel.api.table.type.SeaTunnelRow; +import org.apache.seatunnel.api.table.type.SeaTunnelRowType; + +import org.junit.jupiter.api.Test; + +import java.nio.charset.StandardCharsets; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class SyslogMessageEncoderTest { + static final String[] NAMES = { + "facility", + "severity", + "timestamp", + "hostname", + "app_name", + "proc_id", + "msg_id", + "structured_data", + "message" + }; + static final SeaTunnelRowType TYPE = + new SeaTunnelRowType( + NAMES, + new SeaTunnelDataType<?>[] { + BasicType.INT_TYPE, + BasicType.INT_TYPE, + BasicType.STRING_TYPE, + BasicType.STRING_TYPE, + BasicType.STRING_TYPE, + BasicType.STRING_TYPE, + BasicType.STRING_TYPE, + new MapType<>( + BasicType.STRING_TYPE, + new MapType<>(BasicType.STRING_TYPE, BasicType.STRING_TYPE)), + BasicType.STRING_TYPE + }); + + @Test + void encodesIndependentRfcExampleAndUtf8OctetCount() { + byte[] frame = new SyslogMessageEncoder(TYPE, 8192).encode(row()); + String expected = + "<165>1 2003-10-11T22:14:15.003Z mymachine su 123 ID47 - \uFEFFhello \u4e16\u754c\nsecond line"; + assertEquals( + expected.getBytes(StandardCharsets.UTF_8).length + " " + expected, + new String(frame, StandardCharsets.UTF_8)); + assertTrue(expected.getBytes(StandardCharsets.UTF_8).length > expected.length()); + } + + @Test + void encodesDefaultsNullAndEmptyMessageDistinctly() { + SeaTunnelRowType type = + new SeaTunnelRowType( + new String[] {"message"}, + new SeaTunnelDataType<?>[] {BasicType.STRING_TYPE}); + SyslogMessageEncoder encoder = new SyslogMessageEncoder(type, 8192); + String absent = "<14>1 - - - - - -"; + assertEquals( + absent.length() + " " + absent, + new String( + encoder.encode(new SeaTunnelRow(new Object[] {null})), + StandardCharsets.UTF_8)); + String empty = absent + " \uFEFF"; + assertEquals( + empty.getBytes(StandardCharsets.UTF_8).length + " " + empty, + new String( + encoder.encode(new SeaTunnelRow(new Object[] {""})), + StandardCharsets.UTF_8)); + } + + @Test + void escapesStructuredDataWithoutEscapingMessageOrSplittingLines() { + SeaTunnelRow row = row(); + Map<String, String> params = new LinkedHashMap<>(); + params.put("text", "\"\\]\n\u4e16"); + row.setField(7, Collections.singletonMap("example@32473", params)); + String frame = + new String( + new SyslogMessageEncoder(TYPE, 8192).encode(row), StandardCharsets.UTF_8); + assertTrue(frame.contains("[example@32473 text=\"\\\"\\\\\\]\n\u4e16\"]")); + assertTrue(frame.endsWith("\uFEFFhello \u4e16\u754c\nsecond line")); + } + + @Test + void rejectsEveryHeaderInjectionAndOverlongFieldWithoutLeakingPayload() { + for (int index : new int[] {3, 4, 5, 6}) { + for (String value : + new String[] { + "", + "has space", + "secret\n<0>1", + "secret\r", + "\t", + "\u0000", + "\u007f", + "\u00e9" + }) { + SeaTunnelRow row = row(); + row.setField(index, value); + IllegalArgumentException e = + assertThrows( + IllegalArgumentException.class, + () -> new SyslogMessageEncoder(TYPE, 8192).encode(row)); + assertTrue(e.getMessage().contains(NAMES[index])); + assertFalse(e.getMessage().contains("secret")); + } + } + int[] limits = {255, 48, 128, 32}; + for (int i = 0; i < limits.length; i++) { + SeaTunnelRow row = row(); + row.setField(i + 3, repeat("a", limits[i])); + new SyslogMessageEncoder(TYPE, 8192).encode(row); + row.setField(i + 3, repeat("a", limits[i] + 1)); + assertThrows( + IllegalArgumentException.class, + () -> new SyslogMessageEncoder(TYPE, 8192).encode(row)); + } + } + + @Test + void validatesPriorityAndRowKinds() { + for (int[] pair : new int[][] {{0, 0}, {23, 7}}) { + SeaTunnelRow row = row(); + row.setField(0, pair[0]); + row.setField(1, pair[1]); + new SyslogMessageEncoder(TYPE, 8192).encode(row); + } + for (int index : new int[] {0, 1}) { + for (Object value : new Object[] {-1, index == 0 ? 24 : 8, 1L, "1"}) { + SeaTunnelRow row = row(); + row.setField(index, value); + assertThrows( + IllegalArgumentException.class, + () -> new SyslogMessageEncoder(TYPE, 8192).encode(row)); + } + } + for (RowKind kind : + new RowKind[] {RowKind.DELETE, RowKind.UPDATE_BEFORE, RowKind.UPDATE_AFTER}) { + SeaTunnelRow row = row(); + row.setRowKind(kind); + assertThrows( + IllegalArgumentException.class, + () -> new SyslogMessageEncoder(TYPE, 8192).encode(row)); + } + } + + @Test + void validatesTimestampCalendarPrecisionAndOffset() { + for (String timestamp : + new String[] { + "-", "2024-02-29T23:59:59.123456+23:59", "2024-01-01T00:00:00-00:00" + }) { + SeaTunnelRow row = row(); + row.setField(2, timestamp); + new SyslogMessageEncoder(TYPE, 8192).encode(row); + } + for (String timestamp : + new String[] { + "", + "Oct 11 22:14:15", + "2023-02-29T00:00:00Z", + "2024-01-01T24:00:00Z", + "2024-01-01T00:00:60Z", + "2024-01-01t00:00:00Z", + "2024-01-01T00:00:00.1234567Z", + "2024-01-01T00:00:00", + "2024-01-01T00:00:00+24:00" + }) { + SeaTunnelRow row = row(); + row.setField(2, timestamp); + assertThrows( + IllegalArgumentException.class, + () -> new SyslogMessageEncoder(TYPE, 8192).encode(row)); + } + } + + @Test + void validatesStructuredDataNamesTypesAndUnicode() { + for (String name : new String[] {"", "x y", "x]", "x=", "x\"", "\u00e9", repeat("a", 33)}) { + SeaTunnelRow row = row(); + row.setField(7, Collections.singletonMap(name, Collections.emptyMap())); + assertThrows( + IllegalArgumentException.class, + () -> new SyslogMessageEncoder(TYPE, 8192).encode(row)); + row.setField( + 7, + Collections.singletonMap("example@32473", Collections.singletonMap(name, "v"))); + assertThrows( + IllegalArgumentException.class, + () -> new SyslogMessageEncoder(TYPE, 8192).encode(row)); + } + for (Object value : + new Object[] { + "[raw]", + Collections.singletonMap("id", null), + Collections.singletonMap("id", Collections.singletonMap("key", null)), + Collections.singletonMap("id", Collections.singletonMap("key", "\uD800")) + }) { + SeaTunnelRow row = row(); + row.setField(7, value); + assertThrows( + IllegalArgumentException.class, + () -> new SyslogMessageEncoder(TYPE, 8192).encode(row)); + } + SeaTunnelRow row = row(); + row.setField(8, "\uD800"); + assertThrows( + IllegalArgumentException.class, + () -> new SyslogMessageEncoder(TYPE, 8192).encode(row)); + } + + @Test + void enforcesEncodedByteLimitIncludingHeaderBomAndEscapes() { + SeaTunnelRow row = row(); + byte[] frame = new SyslogMessageEncoder(TYPE, 8192).encode(row); + int length = Integer.parseInt(new String(frame, StandardCharsets.UTF_8).split(" ", 2)[0]); + new SyslogMessageEncoder(TYPE, length).encode(row); + assertThrows( + IllegalArgumentException.class, + () -> new SyslogMessageEncoder(TYPE, length - 1).encode(row)); + row.setField(8, repeat("\u4e16", 8192)); + assertThrows( + IllegalArgumentException.class, + () -> new SyslogMessageEncoder(TYPE, 8192).encode(row)); + row.setField(8, ""); + row.setField( + 7, + Collections.singletonMap("id", Collections.singletonMap("key", repeat("]", 8192)))); + assertThrows( + IllegalArgumentException.class, + () -> new SyslogMessageEncoder(TYPE, 8192).encode(row)); + } + + @Test + void rejectsWrongSchemaAndArity() { + assertThrows( + IllegalArgumentException.class, + () -> + new SyslogMessageEncoder( + new SeaTunnelRowType( + new String[] {"message"}, + new SeaTunnelDataType<?>[] {BasicType.INT_TYPE}), + 8192)); + assertThrows( + IllegalArgumentException.class, + () -> + new SyslogMessageEncoder( + new SeaTunnelRowType( + new String[] {"other"}, + new SeaTunnelDataType<?>[] {BasicType.STRING_TYPE}), + 8192)); + assertThrows( + IllegalArgumentException.class, + () -> + new SyslogMessageEncoder(TYPE, 8192) + .encode(new SeaTunnelRow(new Object[] {"x"}))); + } + + static SeaTunnelRow row() { + return new SeaTunnelRow( + new Object[] { + 20, + 5, + "2003-10-11T22:14:15.003Z", + "mymachine", + "su", + "123", + "ID47", + null, + "hello \u4e16\u754c\nsecond line" + }); + } + + private static String repeat(String value, int count) { + return String.join("", Collections.nCopies(count, value)); + } +} diff --git a/seatunnel-connectors-v2/connector-syslog/src/test/java/org/apache/seatunnel/connectors/seatunnel/syslog/sink/SyslogSinkFactoryTest.java b/seatunnel-connectors-v2/connector-syslog/src/test/java/org/apache/seatunnel/connectors/seatunnel/syslog/sink/SyslogSinkFactoryTest.java new file mode 100644 index 0000000000..6cd6d21d8f --- /dev/null +++ b/seatunnel-connectors-v2/connector-syslog/src/test/java/org/apache/seatunnel/connectors/seatunnel/syslog/sink/SyslogSinkFactoryTest.java @@ -0,0 +1,141 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.syslog.sink; + +import org.apache.seatunnel.api.configuration.ReadonlyConfig; +import org.apache.seatunnel.api.configuration.util.ConfigValidator; +import org.apache.seatunnel.api.configuration.util.OptionValidationException; +import org.apache.seatunnel.api.table.catalog.CatalogTable; +import org.apache.seatunnel.api.table.catalog.PhysicalColumn; +import org.apache.seatunnel.api.table.catalog.TableIdentifier; +import org.apache.seatunnel.api.table.catalog.TableSchema; +import org.apache.seatunnel.api.table.factory.Factory; +import org.apache.seatunnel.api.table.factory.TableSinkFactoryContext; +import org.apache.seatunnel.api.table.type.BasicType; +import org.apache.seatunnel.common.utils.SerializationUtils; +import org.apache.seatunnel.connectors.seatunnel.syslog.config.SyslogSinkConfig; + +import org.junit.jupiter.api.Test; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.ServiceLoader; +import java.util.stream.StreamSupport; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class SyslogSinkFactoryTest { + @Test + void validatesDefaultsAndNumericBoundaries() { + new SyslogSinkConfig(ReadonlyConfig.fromMap(config())); + for (String name : + new String[] { + "port", "connect_timeout_ms", "write_timeout_ms", "max_message_bytes" + }) { + Map<String, Object> values = config(); + values.put(name, 1); + new SyslogSinkConfig(ReadonlyConfig.fromMap(values)); + values.put(name, 0); + OptionValidationException error = + assertThrows( + OptionValidationException.class, + () -> new SyslogSinkConfig(ReadonlyConfig.fromMap(values))); + assertTrue(error.getMessage().contains(name)); + } + for (String name : new String[] {"port", "max_message_bytes"}) { + Map<String, Object> values = config(); + values.put(name, name.equals("port") ? 65536 : 1048577); + assertThrows( + OptionValidationException.class, + () -> new SyslogSinkConfig(ReadonlyConfig.fromMap(values))); + } + } + + @Test + void rejectsMissingHostAndInvalidTlsOptions() { + assertThrows( + OptionValidationException.class, + () -> + ConfigValidator.of(ReadonlyConfig.fromMap(Collections.emptyMap())) + .validate(new SyslogSinkFactory().optionRule())); + for (String host : new String[] {"", " ", "local\nhost", " localhost"}) { + Map<String, Object> values = config(); + values.put("host", host); + assertThrows( + RuntimeException.class, + () -> new SyslogSinkConfig(ReadonlyConfig.fromMap(values))); + } + for (String key : new String[] {"tls.key_store.path", "password", "tls.key_store.type"}) { + Map<String, Object> values = config(); + values.put(key, "not-for-diagnostics"); + IllegalArgumentException error = + assertThrows( + IllegalArgumentException.class, + () -> new SyslogSinkConfig(ReadonlyConfig.fromMap(values))); + assertFalse(error.getMessage().contains("not-for-diagnostics")); + } + } + + @Test + void discoversFactoryAndSerializesSinkWithoutOpeningTlsFilesOrNetwork() { + assertTrue( + StreamSupport.stream(ServiceLoader.load(Factory.class).spliterator(), false) + .anyMatch(factory -> factory instanceof SyslogSinkFactory)); + Map<String, Object> values = config(); + values.put("host", "unresolved.invalid"); + values.put("tls.ca_cert_path", "not-read-until-worker.pem"); + SyslogSink original = + (SyslogSink) + new SyslogSinkFactory() + .createSink( + new TableSinkFactoryContext( + table(), + ReadonlyConfig.fromMap(values), + getClass().getClassLoader())) + .createSink(); + SyslogSink restored = + SerializationUtils.deserialize(SerializationUtils.serialize(original)); + assertEquals("Syslog", restored.getPluginName()); + assertEquals( + "message", + restored.getWriteCatalogTable().get().getSeaTunnelRowType().getFieldNames()[0]); + } + + static Map<String, Object> config() { + Map<String, Object> values = new HashMap<>(); + values.put("host", "localhost"); + return values; + } + + static CatalogTable table() { + return CatalogTable.of( + TableIdentifier.of("default", "default", "syslog"), + TableSchema.builder() + .column( + PhysicalColumn.of( + "message", BasicType.STRING_TYPE, 0, true, null, null)) + .build(), + Collections.emptyMap(), + Collections.emptyList(), + "Syslog"); + } +} diff --git a/seatunnel-connectors-v2/connector-syslog/src/test/java/org/apache/seatunnel/connectors/seatunnel/syslog/sink/SyslogSinkWriterTest.java b/seatunnel-connectors-v2/connector-syslog/src/test/java/org/apache/seatunnel/connectors/seatunnel/syslog/sink/SyslogSinkWriterTest.java new file mode 100644 index 0000000000..853235ffac --- /dev/null +++ b/seatunnel-connectors-v2/connector-syslog/src/test/java/org/apache/seatunnel/connectors/seatunnel/syslog/sink/SyslogSinkWriterTest.java @@ -0,0 +1,240 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.syslog.sink; + +import org.junit.jupiter.api.Test; + +import javax.net.ssl.SSLSocket; +import javax.net.ssl.SSLSocketFactory; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.OutputStream; +import java.io.UncheckedIOException; +import java.net.Socket; +import java.net.SocketTimeoutException; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class SyslogSinkWriterTest { + @Test + void synchronousWriteAndEveryCheckpointHookFlush() throws Exception { + AtomicInteger flushes = new AtomicInteger(); + ByteArrayOutputStream output = + new ByteArrayOutputStream() { + @Override + public void flush() { + flushes.incrementAndGet(); + } + }; + SSLSocket tls = tls(); + SyslogSinkWriter writer = writer(new Socket(), tls, output, 5000); + try { + writer.write(SyslogMessageEncoderTest.row()); + assertTrue(output.size() > 0); + writer.prepareCommit(); + writer.prepareCommit(1); + assertTrue(writer.snapshotState(1).isEmpty()); + writer.close(); + writer.close(); + assertEquals(5, flushes.get()); + assertTrue(tls.isClosed()); + assertThrows(IOException.class, () -> writer.write(SyslogMessageEncoderTest.row())); + } finally { + writer.close(); + } + } + + @Test + void partialWritePermanentlyFailsWithoutRetry() throws Exception { + AtomicInteger bytes = new AtomicInteger(); + OutputStream output = + new OutputStream() { + @Override + public void write(int b) throws IOException { + if (bytes.incrementAndGet() == 5) { + throw new IOException("partial write"); + } + } + }; + Socket raw = new Socket(); + SyslogSinkWriter writer = writer(raw, tls(), output, 5000); + try { + IOException failure = + assertThrows( + IOException.class, () -> writer.write(SyslogMessageEncoderTest.row())); + assertTrue(failure.getMessage().contains("uncertain")); + assertTrue(raw.isClosed()); + assertThrows(IOException.class, () -> writer.write(SyslogMessageEncoderTest.row())); + assertThrows(UncheckedIOException.class, writer::prepareCommit); + assertThrows(IOException.class, () -> writer.prepareCommit(1)); + assertThrows(IOException.class, () -> writer.snapshotState(1)); + assertThrows(IOException.class, writer::close); + assertEquals(5, bytes.get()); + } finally { + closeAfterFailure(writer); + } + } + + @Test + void eachCheckpointHookAndClosePropagatesFlushFailure() { + for (int hook = 0; hook < 4; hook++) { + SyslogSinkWriter writer = + writer( + new Socket(), + tls(), + new OutputStream() { + @Override + public void write(int b) {} + + @Override + public void flush() throws IOException { + throw new IOException("flush failure"); + } + }, + 5000); + try { + switch (hook) { + case 0: + assertThrows(UncheckedIOException.class, writer::prepareCommit); + break; + case 1: + assertThrows(IOException.class, () -> writer.prepareCommit(1)); + break; + case 2: + assertThrows(IOException.class, () -> writer.snapshotState(1)); + break; + default: + assertThrows(IOException.class, writer::close); + } + assertThrows(IOException.class, writer::close); + } finally { + closeAfterFailure(writer); + } + } + } + + @Test + void deadlineClosesUnderlyingSocketAndReleasesBlockedWrite() throws Exception { + blockedWrite(false); + } + + @Test + void concurrentCloseReleasesBlockedWriteWithoutWaitingForDeadline() throws Exception { + blockedWrite(true); + } + + private void blockedWrite(boolean closeConcurrently) throws Exception { + CountDownLatch entered = new CountDownLatch(1); + CountDownLatch closed = new CountDownLatch(1); + Socket raw = + new Socket() { + @Override + public void close() { + closed.countDown(); + } + }; + OutputStream output = + new OutputStream() { + @Override + public void write(int b) throws IOException { + entered.countDown(); + try { + if (!closed.await(15, TimeUnit.SECONDS)) { + throw new AssertionError("TCP close did not release blocked write"); + } + throw new IOException("TCP closed while writing"); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException(e); + } + } + }; + SyslogSinkWriter writer = writer(raw, tls(), output, closeConcurrently ? 30000 : 500); + ExecutorService worker = Executors.newSingleThreadExecutor(); + try { + Future<IOException> result = + worker.submit( + () -> + assertThrows( + IOException.class, + () -> writer.write(SyslogMessageEncoderTest.row()))); + assertTrue(entered.await(5, TimeUnit.SECONDS)); + if (closeConcurrently) { + assertThrows(IOException.class, writer::close); + } + IOException failure = result.get(5, TimeUnit.SECONDS); + if (!closeConcurrently) { + assertTrue(failure instanceof SocketTimeoutException); + } + assertTrue(closed.await(1, TimeUnit.SECONDS)); + assertThrows(IOException.class, () -> writer.prepareCommit(2)); + } finally { + closed.countDown(); + worker.shutdownNow(); + closeAfterFailure(writer); + assertTrue(worker.awaitTermination(5, TimeUnit.SECONDS)); + } + } + + @Test + void interruptedCallerFailsAndClosesTransport() { + Socket raw = new Socket(); + SyslogSinkWriter writer = writer(raw, tls(), new ByteArrayOutputStream(), 5000); + try { + Thread.currentThread().interrupt(); + assertThrows(IOException.class, () -> writer.write(SyslogMessageEncoderTest.row())); + assertTrue(Thread.currentThread().isInterrupted()); + assertTrue(raw.isClosed()); + } finally { + Thread.interrupted(); + assertThrows(IOException.class, writer::close); + } + } + + private static SyslogSinkWriter writer( + Socket raw, SSLSocket tls, OutputStream output, int timeout) { + return new SyslogSinkWriter( + new SyslogMessageEncoder(SyslogMessageEncoderTest.TYPE, 8192), + new SyslogTlsClient(raw, tls, output, timeout)); + } + + private static void closeAfterFailure(SyslogSinkWriter writer) { + try { + writer.close(); + } catch (IOException expected) { + // Failure assertions are in the test body; cleanup must also run if one fails. + } + } + + private static SSLSocket tls() { + try { + return (SSLSocket) SSLSocketFactory.getDefault().createSocket(); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + } +} diff --git a/seatunnel-connectors-v2/connector-syslog/src/test/java/org/apache/seatunnel/connectors/seatunnel/syslog/sink/SyslogTlsClientTest.java b/seatunnel-connectors-v2/connector-syslog/src/test/java/org/apache/seatunnel/connectors/seatunnel/syslog/sink/SyslogTlsClientTest.java new file mode 100644 index 0000000000..5e2c467319 --- /dev/null +++ b/seatunnel-connectors-v2/connector-syslog/src/test/java/org/apache/seatunnel/connectors/seatunnel/syslog/sink/SyslogTlsClientTest.java @@ -0,0 +1,395 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.syslog.sink; + +import org.apache.seatunnel.api.configuration.ReadonlyConfig; +import org.apache.seatunnel.api.table.type.SeaTunnelRow; +import org.apache.seatunnel.common.utils.SerializationUtils; +import org.apache.seatunnel.connectors.seatunnel.syslog.config.SyslogSinkConfig; + +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.api.io.TempDir; + +import javax.net.ssl.KeyManagerFactory; +import javax.net.ssl.SSLContext; +import javax.net.ssl.SSLServerSocket; +import javax.net.ssl.SSLSocket; +import javax.net.ssl.TrustManagerFactory; + +import java.io.DataInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.net.ServerSocket; +import java.net.Socket; +import java.net.SocketTimeoutException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.security.KeyStore; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** Local TLS transport integration. Interoperability with syslog-ng lives in the E2E module. */ +@Timeout(40) +class SyslogTlsClientTest { + @TempDir static Path directory; + private static SSLContext serverContext; + + @BeforeAll + static void certificates() throws Exception { + keytool( + "-genkeypair", + "-alias", + "receiver", + "-keyalg", + "RSA", + "-keysize", + "2048", + "-sigalg", + "SHA256withRSA", + "-dname", + "CN=localhost", + "-ext", + "SAN=dns:localhost", + "-ext", + "EKU=serverAuth,clientAuth", + "-startdate", + "2020/01/01 00:00:00", + "-validity", + "36500", + "-storetype", + "JKS", + "-keystore", + directory.resolve("receiver.jks").toString(), + "-storepass", + "test-password", + "-keypass", + "test-password", + "-noprompt"); + keytool( + "-exportcert", + "-rfc", + "-alias", + "receiver", + "-keystore", + directory.resolve("receiver.jks").toString(), + "-storepass", + "test-password", + "-file", + directory.resolve("ca.pem").toString()); + KeyStore keys = KeyStore.getInstance("JKS"); + try (InputStream input = Files.newInputStream(directory.resolve("receiver.jks"))) { + keys.load(input, "test-password".toCharArray()); + } + KeyManagerFactory keyFactory = + KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm()); + keyFactory.init(keys, "test-password".toCharArray()); + TrustManagerFactory trust = + TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm()); + trust.init(keys); + serverContext = SSLContext.getInstance("TLS"); + serverContext.init(keyFactory.getKeyManagers(), trust.getTrustManagers(), null); + } + + @Test + void serializedSinkDeliversBulkUtf8FramesToTlsReceiver() throws Exception { + SSLContext global = SSLContext.getDefault(); + try (Receiver receiver = new Receiver(false)) { + int count = 1000; + Future<List<String>> received = receiver.read(count); + SyslogSink original = + new SyslogSink(config(receiver.port()), SyslogSinkFactoryTest.table()); + SyslogSink restored = + SerializationUtils.deserialize(SerializationUtils.serialize(original)); + SyslogSinkWriter writer = (SyslogSinkWriter) restored.createWriter(null); + long started = System.nanoTime(); + try { + for (int i = 0; i < count; i++) { + writer.write( + new SeaTunnelRow( + new Object[] {"event-" + i + " \u4e16\u754c\nsecond"})); + } + writer.prepareCommit(); + writer.prepareCommit(1); + writer.snapshotState(1); + } finally { + writer.close(); + } + List<String> messages = received.get(10, TimeUnit.SECONDS); + assertEquals(count, messages.size()); + for (int i = 0; i < count; i++) { + assertEquals( + "<14>1 - - - - - - \uFEFFevent-" + i + " \u4e16\u754c\nsecond", + messages.get(i)); + } + System.out.println( + "Syslog local TLS receiver: " + + count + + " frames in " + + TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - started) + + " ms (not a benchmark)"); + } + assertSame(global, SSLContext.getDefault()); + } + + @Test + void supportsMutualTlsWithClientKeyStore() throws Exception { + try (Receiver receiver = new Receiver(true)) { + Future<List<String>> received = receiver.read(1); + Map<String, Object> values = values(receiver.port()); + values.put("tls.key_store.path", directory.resolve("receiver.jks").toString()); + values.put("password", "test-password"); + values.put("tls.key_store.type", "JKS"); + try (SyslogTlsClient client = + SyslogTlsClient.connect(new SyslogSinkConfig(ReadonlyConfig.fromMap(values)))) { + client.write( + new SyslogMessageEncoder(SyslogMessageEncoderTest.TYPE, 8192) + .encode(SyslogMessageEncoderTest.row())); + } + assertEquals(1, received.get(10, TimeUnit.SECONDS).size()); + } + } + + @Test + void rejectsUntrustedReceiverAndWrongHostname() throws Exception { + for (boolean wrongHost : new boolean[] {false, true}) { + try (Receiver receiver = new Receiver(false)) { + Future<Boolean> rejected = receiver.expectRejectedHandshake(); + Map<String, Object> values = values(receiver.port()); + if (wrongHost) { + values.put("host", "127.0.0.1"); + } else { + values.remove("tls.ca_cert_path"); + } + IOException error = + assertThrows( + IOException.class, + () -> + SyslogTlsClient.connect( + new SyslogSinkConfig( + ReadonlyConfig.fromMap(values)))); + assertTrue(error.getMessage().contains("handshake")); + assertTrue(rejected.get(10, TimeUnit.SECONDS)); + } + } + } + + @Test + void rejectsMissingClientCertificate() throws Exception { + try (Receiver receiver = new Receiver(true)) { + Future<Boolean> rejected = receiver.expectRejectedHandshake(); + assertThrows( + IOException.class, + () -> { + try (SyslogTlsClient client = + SyslogTlsClient.connect(config(receiver.port()))) { + // TLS 1.3 may report the peer's rejection only on a subsequent + // operation. + for (int i = 0; i < 100; i++) { + client.write(new byte[65536]); + } + } + }); + assertTrue(rejected.get(10, TimeUnit.SECONDS)); + } + } + + @Test + void boundsHandshakeWhenTcpPeerDoesNotSpeakTls() throws Exception { + ExecutorService worker = Executors.newSingleThreadExecutor(); + CountDownLatch release = new CountDownLatch(1); + try (ServerSocket server = new ServerSocket(0)) { + Future<?> accepted = + worker.submit( + () -> { + try (Socket socket = server.accept()) { + assertTrue(release.await(10, TimeUnit.SECONDS)); + } + return null; + }); + Map<String, Object> values = values(server.getLocalPort()); + values.put("write_timeout_ms", 500); + assertThrows( + SocketTimeoutException.class, + () -> + SyslogTlsClient.connect( + new SyslogSinkConfig(ReadonlyConfig.fromMap(values)))); + release.countDown(); + accepted.get(5, TimeUnit.SECONDS); + } finally { + release.countDown(); + worker.shutdownNow(); + assertTrue(worker.awaitTermination(5, TimeUnit.SECONDS)); + } + } + + @Test + void boundsActualTlsWriteWhenReceiverStopsReading() throws Exception { + CountDownLatch handshaken = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + try (Receiver receiver = new Receiver(false)) { + receiver.server.setReceiveBufferSize(1024); + Future<?> accepted = + receiver.worker.submit( + () -> { + try (SSLSocket socket = receiver.accept()) { + socket.startHandshake(); + handshaken.countDown(); + assertTrue(release.await(20, TimeUnit.SECONDS)); + } + return null; + }); + Map<String, Object> values = values(receiver.port()); + values.put("write_timeout_ms", 2000); + SyslogTlsClient client = + SyslogTlsClient.connect(new SyslogSinkConfig(ReadonlyConfig.fromMap(values))); + try { + assertTrue(handshaken.await(5, TimeUnit.SECONDS)); + assertThrows( + SocketTimeoutException.class, + () -> { + byte[] frame = new byte[65536]; + for (int i = 0; i < 2048; i++) { + client.write(frame); + } + }); + assertThrows(IOException.class, client::flush); + } finally { + release.countDown(); + assertThrows(IOException.class, client::close); + } + accepted.get(5, TimeUnit.SECONDS); + } finally { + release.countDown(); + } + } + + private static SyslogSinkConfig config(int port) { + return new SyslogSinkConfig(ReadonlyConfig.fromMap(values(port))); + } + + private static Map<String, Object> values(int port) { + Map<String, Object> values = SyslogSinkFactoryTest.config(); + values.put("port", port); + values.put("tls.ca_cert_path", directory.resolve("ca.pem").toString()); + values.put("write_timeout_ms", 10000); + return values; + } + + private static void keytool(String... arguments) throws Exception { + List<String> command = new ArrayList<>(); + command.add(Paths.get(System.getProperty("java.home"), "bin", "keytool").toString()); + java.util.Collections.addAll(command, arguments); + Process process = new ProcessBuilder(command).redirectErrorStream(true).start(); + try { + assertTrue(process.waitFor(30, TimeUnit.SECONDS), "keytool timed out"); + assertEquals(0, process.exitValue(), "keytool fixture generation failed"); + } finally { + process.destroyForcibly(); + } + } + + private static final class Receiver implements AutoCloseable { + private final SSLServerSocket server; + private final ExecutorService worker = Executors.newSingleThreadExecutor(); + private volatile SSLSocket socket; + + Receiver(boolean clientAuth) throws IOException { + server = (SSLServerSocket) serverContext.getServerSocketFactory().createServerSocket(0); + server.setSoTimeout(10000); + server.setNeedClientAuth(clientAuth); + } + + int port() { + return server.getLocalPort(); + } + + SSLSocket accept() throws IOException { + socket = (SSLSocket) server.accept(); + socket.setSoTimeout(10000); + return socket; + } + + Future<List<String>> read(int count) { + return worker.submit( + () -> { + try (SSLSocket connection = accept(); + DataInputStream input = + new DataInputStream(connection.getInputStream())) { + List<String> messages = new ArrayList<>(); + for (int i = 0; i < count; i++) { + int length = 0; + int digit; + int digits = 0; + while ((digit = input.read()) != ' ') { + assertTrue( + digit >= '0' && digit <= '9' && ++digits <= 7, + "invalid octet prefix"); + assertTrue(digits > 1 || digit != '0', "leading zero"); + length = length * 10 + digit - '0'; + } + assertTrue(length > 0 && length <= 1048576); + byte[] message = new byte[length]; + input.readFully(message); + messages.add(new String(message, StandardCharsets.UTF_8)); + } + return messages; + } + }); + } + + Future<Boolean> expectRejectedHandshake() { + return worker.submit( + () -> { + try (SSLSocket connection = accept()) { + connection.startHandshake(); + return false; + } catch (IOException expected) { + return true; + } + }); + } + + @Override + public void close() throws Exception { + try { + server.close(); + if (socket != null) { + socket.close(); + } + } finally { + worker.shutdownNow(); + assertTrue(worker.awaitTermination(5, TimeUnit.SECONDS)); + } + } + } +} diff --git a/seatunnel-connectors-v2/pom.xml b/seatunnel-connectors-v2/pom.xml index 71ec8ac9ca..eb3a305e12 100644 --- a/seatunnel-connectors-v2/pom.xml +++ b/seatunnel-connectors-v2/pom.xml @@ -41,6 +41,7 @@ <module>connector-kafka</module> <module>connector-pulsar</module> <module>connector-socket</module> + <module>connector-syslog</module> <module>connector-hive</module> <module>connector-file</module> <module>connector-hudi</module> diff --git a/seatunnel-dist/pom.xml b/seatunnel-dist/pom.xml index a029ecd37b..38a5d8a29a 100644 --- a/seatunnel-dist/pom.xml +++ b/seatunnel-dist/pom.xml @@ -380,6 +380,12 @@ <version>${project.version}</version> <scope>provided</scope> </dependency> + <dependency> + <groupId>org.apache.seatunnel</groupId> + <artifactId>connector-syslog</artifactId> + <version>${project.version}</version> + <scope>provided</scope> + </dependency> <dependency> <groupId>org.apache.seatunnel</groupId> <artifactId>connector-edge-socket</artifactId> diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-syslog-e2e/pom.xml b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-syslog-e2e/pom.xml new file mode 100644 index 0000000000..55bb616d3a --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-syslog-e2e/pom.xml @@ -0,0 +1,40 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- + Licensed to the Apache Software Foundation (ASF) under one or more + contributor license agreements. See the NOTICE file distributed with + this work for additional information regarding copyright ownership. + The ASF licenses this file to You under the Apache License, Version 2.0 + (the "License"); you may not use this file except in compliance with + the License. You may obtain a copy of the License at + http://www.apache.org/licenses/LICENSE-2.0 + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +--> +<project xmlns="http://maven.apache.org/POM/4.0.0"; xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; + xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd";> + <modelVersion>4.0.0</modelVersion> + <parent> + <groupId>org.apache.seatunnel</groupId> + <artifactId>seatunnel-connector-v2-e2e</artifactId> + <version>${revision}</version> + </parent> + <artifactId>connector-syslog-e2e</artifactId> + <name>SeaTunnel : E2E : Connector V2 : Syslog</name> + <dependencies> + <dependency> + <groupId>org.apache.seatunnel</groupId> + <artifactId>connector-syslog</artifactId> + <version>${project.version}</version> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.apache.seatunnel</groupId> + <artifactId>connector-fake</artifactId> + <version>${project.version}</version> + <scope>test</scope> + </dependency> + </dependencies> +</project> diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-syslog-e2e/src/test/java/org/apache/seatunnel/e2e/connector/syslog/SyslogSinkIT.java b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-syslog-e2e/src/test/java/org/apache/seatunnel/e2e/connector/syslog/SyslogSinkIT.java new file mode 100644 index 0000000000..017954b767 --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-syslog-e2e/src/test/java/org/apache/seatunnel/e2e/connector/syslog/SyslogSinkIT.java @@ -0,0 +1,232 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.e2e.connector.syslog; + +import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.JsonNode; +import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.MappingIterator; +import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.seatunnel.shade.com.typesafe.config.Config; +import org.apache.seatunnel.shade.com.typesafe.config.ConfigFactory; + +import org.apache.seatunnel.connectors.seatunnel.syslog.config.SyslogSinkOptions; +import org.apache.seatunnel.core.starter.utils.ConfigBuilder; +import org.apache.seatunnel.core.starter.utils.ConfigShadeUtils; +import org.apache.seatunnel.e2e.common.TestResource; +import org.apache.seatunnel.e2e.common.TestSuiteBase; +import org.apache.seatunnel.e2e.common.container.TestContainer; + +import org.awaitility.Awaitility; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.TestTemplate; +import org.junit.jupiter.api.io.TempDir; +import org.testcontainers.containers.Container; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.containers.output.Slf4jLogConsumer; +import org.testcontainers.containers.wait.strategy.Wait; +import org.testcontainers.utility.DockerImageName; +import org.testcontainers.utility.MountableFile; + +import lombok.extern.slf4j.Slf4j; + +import java.io.InputStream; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.security.KeyStore; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Base64; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.TimeUnit; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** Exercises engine factory discovery and an independent RFC 5424/5425 receiver. */ +@Slf4j +public class SyslogSinkIT extends TestSuiteBase implements TestResource { + // syslog-ng 4.12.0, pinned by its multi-platform image digest. + private static final DockerImageName IMAGE = + DockerImageName.parse( + "balabit/syslog-ng@sha256:f564a6906b03f0dc7fa9edf7925956abf3935a474c86d8a1384367889ce6e090"); + @TempDir static Path directory; + private GenericContainer<?> receiver; + + @Override + @BeforeAll + public void startUp() throws Exception { + Config secretConfig = + ConfigFactory.parseString( + "sink { Syslog { " + + SyslogSinkOptions.KEY_STORE_PASSWORD.key() + + " = \"masking-test-value\" } }"); + assertFalse( + ConfigBuilder.configDesensitization( + secretConfig.root().unwrapped(), + ConfigShadeUtils.getLogDesensitizationOptions(secretConfig)) + .toString() + .contains("masking-test-value")); + createCertificate(); + receiver = + new GenericContainer<>(IMAGE) + .withNetwork(NETWORK) + .withNetworkAliases("syslog-receiver") + .withExposedPorts(6514) + .withCopyFileToContainer( + MountableFile.forHostPath(directory.resolve("key.pem")), + "/tmp/receiver-key.pem") + .withCopyFileToContainer( + MountableFile.forHostPath(directory.resolve("cert.pem")), + "/tmp/receiver-cert.pem") + .withCopyFileToContainer( + MountableFile.forClasspathResource("docker/syslog-ng.conf"), + "/etc/syslog-ng/syslog-ng.conf") + .withCreateContainerCmdModifier( + command -> command.withEntrypoint("/usr/sbin/syslog-ng")) + .withCommand("-F", "--no-caps", "-f", "/etc/syslog-ng/syslog-ng.conf") + .waitingFor( + Wait.forListeningPort().withStartupTimeout(Duration.ofSeconds(60))) + .withLogConsumer(new Slf4jLogConsumer(log)); + receiver.start(); + } + + @Override + @AfterAll + public void tearDown() { + try { + if (receiver != null) { + receiver.close(); + } + } finally { + NETWORK.close(); + } + } + + @TestTemplate + public void sendsRfc5424OverTls(TestContainer container) throws Exception { + Container.ExecResult truncate = + receiver.execInContainer("sh", "-c", ": > /tmp/received.json"); + assertEquals(0, truncate.getExitCode(), truncate.getStderr()); + // The harness shares this mount between the Flink job manager and task managers. + container.copyAbsolutePathToContainer( + directory.resolve("cert.pem").toString(), "/tmp/seatunnel_mnt/syslog-ca.pem"); + Container.ExecResult result = container.executeJob("/fake_to_syslog.conf"); + assertEquals(0, result.getExitCode(), result.getStderr()); + List<JsonNode> rows = + Awaitility.await() + .atMost(Duration.ofSeconds(30)) + .until(this::records, records -> records.size() >= 2); + assertEquals(2, rows.size(), "Unexpected extra receiver records"); + JsonNode first = + rows.stream() + .filter(row -> "ID47".equals(row.path("MSGID").asText())) + .findFirst() + .orElse(null); + JsonNode last = + rows.stream() + .filter(row -> "ID48".equals(row.path("MSGID").asText())) + .findFirst() + .orElse(null); + assertNotNull(first); + assertNotNull(last); + assertEquals("origin", first.path("HOST").asText()); + assertEquals("seatunnel", first.path("PROGRAM").asText()); + assertEquals("123", first.path("PID").asText()); + assertEquals("hello \u4e16\u754c\nsecond line", first.path("MESSAGE").asText()); + assertEquals("last message", last.path("MESSAGE").asText()); + // format-json may render dotted names as nested objects; find the parameter recursively. + assertTrue(first.toString().contains("quote\\\"slash\\\\bracket]"), first.toString()); + } + + private List<JsonNode> records() throws Exception { + Container.ExecResult result = receiver.execInContainer("cat", "/tmp/received.json"); + assertEquals(0, result.getExitCode(), result.getStderr()); + try (MappingIterator<JsonNode> iterator = + new ObjectMapper().readerFor(JsonNode.class).readValues(result.getStdout())) { + return iterator.readAll(); + } + } + + private static void createCertificate() throws Exception { + Path keyStore = directory.resolve("receiver.jks"); + List<String> command = new ArrayList<>(); + Collections.addAll( + command, + Paths.get(System.getProperty("java.home"), "bin", "keytool").toString(), + "-genkeypair", + "-alias", + "receiver", + "-keyalg", + "RSA", + "-keysize", + "2048", + "-sigalg", + "SHA256withRSA", + "-dname", + "CN=syslog-receiver", + "-ext", + "SAN=dns:syslog-receiver", + "-ext", + "EKU=serverAuth", + "-startdate", + "2020/01/01 00:00:00", + "-validity", + "36500", + "-storetype", + "JKS", + "-keystore", + keyStore.toString(), + "-storepass", + "test-password", + "-keypass", + "test-password", + "-noprompt"); + Process process = new ProcessBuilder(command).redirectErrorStream(true).start(); + try { + assertTrue(process.waitFor(30, TimeUnit.SECONDS), "keytool timed out"); + assertEquals(0, process.exitValue(), "keytool fixture generation failed"); + } finally { + process.destroyForcibly(); + } + KeyStore store = KeyStore.getInstance("JKS"); + try (InputStream input = Files.newInputStream(keyStore)) { + store.load(input, "test-password".toCharArray()); + } + writePem( + directory.resolve("key.pem"), + "PRIVATE KEY", + store.getKey("receiver", "test-password".toCharArray()).getEncoded()); + writePem( + directory.resolve("cert.pem"), + "CERTIFICATE", + store.getCertificate("receiver").getEncoded()); + } + + private static void writePem(Path path, String label, byte[] der) throws Exception { + String encoded = Base64.getMimeEncoder(64, new byte[] {'\n'}).encodeToString(der); + Files.write( + path, + ("-----BEGIN " + label + "-----\n" + encoded + "\n-----END " + label + "-----\n") + .getBytes(StandardCharsets.US_ASCII)); + } +} diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-syslog-e2e/src/test/resources/docker/syslog-ng.conf b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-syslog-e2e/src/test/resources/docker/syslog-ng.conf new file mode 100644 index 0000000000..d47a1c7ddd --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-syslog-e2e/src/test/resources/docker/syslog-ng.conf @@ -0,0 +1,29 @@ +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +@version: 4.12 +@include "scl.conf" +options { keep-hostname(yes); use-dns(no); flush-lines(1); }; +source s_tls { + syslog( + ip("0.0.0.0") port(6514) transport("tls") + tls(key-file("/tmp/receiver-key.pem") cert-file("/tmp/receiver-cert.pem") + peer-verify(optional-untrusted)) + ); +}; +destination d_json { + file("/tmp/received.json" template("$(format-json --scope rfc5424 --scope all-nv-pairs)\n")); +}; +log { source(s_tls); destination(d_json); }; diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-syslog-e2e/src/test/resources/fake_to_syslog.conf b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-syslog-e2e/src/test/resources/fake_to_syslog.conf new file mode 100644 index 0000000000..e24b765137 --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-syslog-e2e/src/test/resources/fake_to_syslog.conf @@ -0,0 +1,59 @@ +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +env { + parallelism = 1 + job.mode = "BATCH" +} +source { + FakeSource { + row.num = 2 + schema = { + fields { + facility = int + severity = int + timestamp = string + hostname = string + app_name = string + proc_id = string + msg_id = string + structured_data = "map<string, map<string, string>>" + message = string + } + } + rows = [ + { + kind = INSERT + fields = [20, 5, "2026-01-02T03:04:05.123456Z", "origin", "seatunnel", "123", "ID47", + {"example@32473": {"text": "quote\"slash\\bracket]"}}, + "hello \u4e16\u754c\nsecond line"] + }, + { + kind = INSERT + fields = [1, 6, null, null, null, null, "ID48", null, "last message"] + } + ] + } +} +sink { + Syslog { + host = "syslog-receiver" + port = 6514 + tls.ca_cert_path = "/tmp/seatunnel_mnt/syslog-ca.pem" + connect_timeout_ms = 10000 + write_timeout_ms = 10000 + max_message_bytes = 8192 + } +} diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/pom.xml b/seatunnel-e2e/seatunnel-connector-v2-e2e/pom.xml index bcf4e900bc..84b7f3fba7 100644 --- a/seatunnel-e2e/seatunnel-connector-v2-e2e/pom.xml +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/pom.xml @@ -94,6 +94,7 @@ <module>connector-qdrant-e2e</module> <module>connector-sls-e2e</module> <module>connector-socket-e2e</module> + <module>connector-syslog-e2e</module> <module>connector-typesense-e2e</module> <module>connector-email-e2e</module> <module>connector-edge-socket-e2e</module>
