用于从 SFTP 服务器提取无服务器日志（下载文件）的 Google Cloud Platform 解决方案

Question

今天有一个问题，在我输入答案时被删除（不知道为什么）。由于答案很长，我决定复制/重新创建它并提供我的答案。也许它会对某人有用。

---

这是原始问题：

我们有一个 SFTP 服务器，用于转储 Apache、Nginx、Wordpress 日志。我们希望将这些日志备份到 Cloud Storage 中，同时解析这些日志的内容并插入到 BigQuery 表中。我通常使用 Cloud Functions（NodeJS 或 Python），这是我首先想到的首选解决方案。

但是，Cloud Function 有一个触发器，如果​​我的目标是让程序持续监视/观察/侦听 SFTP 文件夹上的新文件，这没有意义。如果我的要求不那么严格，我可以按时触发它，比如每小时读取 SFTP 文件夹中的新文件。当新文件转储到 Cloud Storage 时，Cloud Functions 也会起作用，触发函数解析日志文件并插入 BigQuery。

如果我坚持持续收听 SFTP 文件夹的要求，您能否提出一个更好的设计解决方案以及我需要将哪些 GCP 服务组合（除了 Cloud Storage 和 BigQuery）来实现这一点？

如果要求不那么严格，我的解决方案是否足够好？ 附言我刚刚发现 SFTP 凭据具有只读权限。因此，通过添加后缀来重命名文件是毫无疑问的。我应该使用 MemoryStore 之类的缓存来记住哪些文件已完成？

Answer 1

长读。

在我看来，这是一个非常大的问题。而解决方案不仅需要代码开发，还需要大量的设计思维和决策（包括一些妥协）。

根据我的个人经验（我开发了两次这样的解决方案，在生产中对其进行了维护等），可以将云功能与一组 GCP 资源（秘密管理器、pubsub 主题、firestore 集合、服务帐户和IAM 等...根据您的要求（我不知道详细信息）和上下文 - 您可能需要创建一个功能组件，该组件由几个（比如说两个到五个）不同的云组成职能。二——如果你的文件很小（每个最大100M），每天的文件数量不大（几千或几万个文件），你有权在下载时从SFTP服务器上删除原始文件。

如果您没有此类权限 - 应该有一些其他进程可以清理“旧”或“已下载”文件。否则，最终解决方案将不起作用（当仅下载文件列表，而不是文件，而只是文件列表时，需要超过 540 秒）。

SFTP 是一个“被动”组件 - 如果有新文件到达，它不会通知我们，因此我们这边应该有一些“主动”组件来发起到 SFTP 服务器的连接。这是一种“拉动”交互，并且有规律性 - 即每 10、15 或 20 分钟 - 连接到 SFTP 服务器并检查是否有新内容要下载。

接下来。云函数是幂等的，不能仅在云函数中存储/保持文件下载的状态。应该有一些外部（相对于云功能）服务来维护每个文件下载过程的状态机。我使用了 Firestore。它非常方便并且延迟非常小。 firestore 集合中的每个文档都代表“文件下载过程”的反映 - 一个状态机以及大量元数据、状态转换历史等。

Cloud Functions 有两个重要的限制：

1. 540 秒超时。
2. 2Gb 内存。

这意味着下载过程（和任何其他活动）不应超过 540 秒。如果您要将任何数据存储在（云功能的）内存中，则该数据块应小于 2Gb。

超时限制会影响进程吗？ - 是的，它可以。整个过程的瓶颈是 SFTP 服务器和 GCP 入网点之间的“带宽”。文件越大 - 下载所需的时间就越长，尤其是在并行下载许多文件时。

因此，该算法很快就会以下列方式工作：

1/ 第一个云功能每隔 15 分钟触发一次（Cloud Scheduler => PubSub topic => 云功能）。云函数读取所有 SFTP 连接和所有数据管道的配置（即来自 GCS 存储桶的 json 文件）（因为该组件可能与许多 SFTP 服务器一起使用，并且对于每个 SFTP 服务器可能有许多数据管道），然后从Secret Manager（对于每个 SFTP 服务器），然后连接到 SFTP 服务器，并下载每个连接/管道的可用文件列表。因此，对于我们知道的每个文件 - 连接（SFTP 服务器）、管道（即源目录）、文件名、文件大小、文件修改时间戳。我不希望 SFTP 服务器有更多的东西。对于每个连接和数据管道，我们组成一个（取决于配置并且应该是灵活的）文件列表，最多包含 5、8 或 1 万个文件。该列表作为 json 结构作为消息推送到 PubSub 主题中（如果需要，还带有一些额外的元数据）。因此，如果我们有 2 个 SFTP 服务器，并且每个服务器中有 3 个管道 - 将至少有 6 条消息。如果 SFTP 服务器中的目录包含超过 5K、8K 或 10K 的文件，则可能更多。目前，我们不知道这些文件是否已下载，或者正在下载，或者已经失败，或者这是一个新文件。 部署此函数时 - “最大实例”参数的值为 1。

2/ 第二个云函数由包含文件列表的 PubSub 消息触发（对于一些 SFTP 服务器和一些管道）。对于传入列表中的每个文件，云函数应该决定要做什么：

1. 这是一个新文件，应该下载。
2. 这是一个正在进行的下载，我们需要等待更多 - 什么都不做。
3. 这是一个已经下载的文件，我们什么也不做。
4. 这是一个正在下载的过程，但时间过长 - 可能下载崩溃了，应该重新下载。
5. 这是……可能还有更多案件需要处理……

现在需要 Firestore 集合。集合中的每个文档 - 反映文件发生的情况；一切都记录在那里 - 下载过程何时开始，何时（或是否）完成等等。文档 ID 是根据可用元数据计算得出的哈希值 - 连接（SFTP 服务器）、管道（即源目录）、源文件名、源文件大小、源文件修改时间戳。所有这些都来自消息。

例如，我们计算哈希并检查集合中是否存在此类文档。如果它不存在 - 创建一个新文档，因为这是一个全新的下载文件。然后编写一个 json 消息并将其推送到第二个 PubSub 主题 - 下一个云功能将处理它。它存在 - 有必要决定我们将如何处理它 - 什么都不做（因为它已经下载，或者因为下载可能仍在进行中）或再次触发它的下载 - 编写一个 json 消息并推送它进入第二个 PubSub 主题...

部署此函数时 - “最大实例”参数的值介于 4 到 12 之间（根据我的经验）。

3/ 第三个云功能是由一个 PubSub 消息触发的，该消息包含要下载的文件的详细信息。需要完成以下步骤：

1. 检查该文件是否未被其他云功能下载
2. 更新 firestore 文档 - 我们开始下载过程
3. 获取配置详细信息（来自 GCS 中的 json 文件）
4. 获取连接详细信息（来自 Secret Manager）
5. 连接和下载
6. 将下载的文件保存到目标 GCS 存储桶中
7. 更新 firestore 文档 - 我们完成了下载过程

部署此函数时 - “最大实例”参数的值介于 10 到 30 之间（根据我的经验）。

这是一个非常简短的描述，在最简单的假设下（即您没有大于 100Mb 的文件或/并且连接良好）。

一些额外的事情要记住。

1/ 准确的记录。具有一致字段的 Json 结构将定期记录。我建议做一个接收器，以便可以在 BigQuery 表中分析日志。

2/ 服务帐户和 IAM。所有这些都应在仅用于给定组件的自定义服务帐户下运行。将提供相关的 IAM 角色。

3/ 云 NAT。 SFTP（根据我的经验）仅适用于特定的静态 IP 地址（它们不允许来自任何地址的连接）。因此，网络、子网、IP 地址、路由器、NAT - 所有这些都将被创建和配置。 IP 地址将提供给 SFTP 服务器所有者，以允许访问。使用“vpc 连接器”参数部署的云函数。

4/ 进度和监控 - 3 个信息来源 - firestore 收集、Stackdriver 日志、BigQuery 表。

再一次，这是我脑海中最简单的描述。如果您有具体问题或想讨论，请告诉我。