利用Elixir和LevelDB构建从Fluentd到数据仓库的弹性日志摄取层

数据工程

文章字数: 3.8k

阅读时长: 17 分

在处理分布式系统中海量、突发性的日志流时，核心挑战在于构建一个既能满足实时观测需求，又能保证数据最终无损落盘至数据仓库的摄取层。单纯依赖消息队列（如Kafka）和流处理框架（如Spark/Flink）的组合虽然功能强大，但在某些场景下显得过于笨重，运维成本高昂。尤其是在需要为开发者提供一个低延迟的实时日志“尾随（tailing）”界面的同时，还要处理后端数据仓库可能出现的反压（backpressure）问题时，一个更轻量、更具弹性的方案就显得尤ว有价值。

本文将详细阐述一个架构决策过程，最终选择使用Elixir/OTP构建一个自定义的日志摄取与处理服务。该服务利用节点本地的LevelDB作为持久化缓冲区，以应对下游服务的暂时不可用或性能抖动，同时通过Phoenix Channels向基于Solid.js的实时前端推送日志样本，并最终将完整数据批量推送到数据仓库。

定义问题：弹性与实时性的双重诉求

在一个由数百个微服务组成的容器化环境中，日志流具备以下特点：

高通量与突发性: 日志产生速率平时可能在每秒数万条，但在业务高峰或异常事件期间，流量可能在短时间内激增数倍。
数据完整性: 每一条日志都可能包含关键的调试信息或业务指标，丢失是不可接受的。必须保证至少一次（At-least-once）的交付语义。
下游服务的不稳定性: 数据仓库的写入API可能因维护、过载或网络问题而暂时不可用。摄取层必须能够在这种情况下缓冲数据，并在服务恢复后自动重试，而不能阻塞上游的日志收集器（Fluentd）。
低延迟的实时可见性: 开发和运维团队需要一个能够实时查看特定服务日志流的界面，延迟应在秒级以内，用于快速故障排查。

方案对比与技术选型

方案A: 业界标准组合 - Fluentd -> Kafka -> Flink -> Data Warehouse

这是非常成熟和常见的解决方案。

优势:
- 解耦与削峰: Kafka作为中间的持久化消息队列，是应对流量洪峰和解耦生产者/消费者的绝佳选择。其高可用和持久性为数据完整性提供了强力保障。
- 强大的处理能力: Flink提供了丰富的状态管理、窗口计算和Exactly-once处理语义，适用于复杂的ETL和实时分析任务。
- 生态成熟: 整个链路上的组件都有广泛的社区支持和商业解决方案。
劣势:
- 运维复杂度: 维护一个高可用的Kafka集群和Flink集群本身就是一项专业且繁重的工作，涉及Zookeeper、Broker、JobManager、TaskManager等多个组件的监控、调优和故障处理。
- 资源开销: 整个技术栈对计算和存储资源的消耗相当可观，对于中等规模的系统可能是一种过度设计。
- 实时性延迟: 数据从进入Kafka到被Flink处理，再到最终可查询，整个链路的端到端延迟对于实时“尾随”场景可能偏高。构建一个独立的、从Kafka直接消费到前端的通道会增加系统复杂度。

方案B: 轻量级并发模型 - Fluentd -> Elixir Service (with LevelDB) -> Data Warehouse / WebSocket

这个方案的核心是利用Elixir和BEAM虚拟机强大的并发和容错能力，构建一个集接收、缓冲、处理和分发于一体的轻量级服务。

优势:
- 极致的并发处理: BEAM的轻量级进程模型（Actor Model）和抢占式调度器非常适合处理成千上万个并发的IO密集型任务（如网络连接和磁盘写入）。每个Fluentd连接都可以由一个独立的进程处理，互不阻塞。
- 内建的容错与监督: OTP的监督树（Supervision Tree）机制可以轻松构建自我修复的系统。处理日志的进程如果崩溃，会被其监督者自动重启，保证服务的健壮性。
- 本地持久化缓冲: 这是此方案的关键。通过在每个Elixir节点上内嵌LevelDB，我们可以创建一个高性能的本地写前日志（Write-Ahead Log）。接收到的日志批次首先被快速写入LevelDB，然后才开始异步推向下游。如果下游失败，数据仍然安全地存在本地磁盘上，等待重试。这种模式将反压问题隔离在服务内部，不会扩散到上游的Fluentd。
- 一体化的实时推送: Elixir的Phoenix框架是构建实时Web应用的利器，其内置的Channels功能可以高效地将处理中的日志样本通过WebSocket推送到前端，完美满足实时可见性的需求。
劣势:
- 非分布式日志存储: 与Kafka不同，LevelDB是节点本地的存储。如果一个Elixir节点连同其磁盘一起永久性丢失，那么该节点上还未成功推送到数据仓库的日志将会丢失。这是一个重要的架构权衡，需要通过部署冗余和快速节点替换来缓解。
- 需要定制开发: 相比于直接使用现成的Flink Job，此方案需要自行开发核心的缓冲、重试和批处理逻辑。

最终决策

考虑到团队对Elixir/OTP的熟悉度，以及对运维成本和资源消耗的敏感性，我们最终选择了方案B。我们接受了节点故障可能导致少量数据丢失的风险，并认为通过合理的冗余部署策略（例如，在多个可用区部署Elixir节点）可以将该风险控制在可接受范围内。方案B的简洁性、高性能和一体化的实时能力，更符合我们对一个弹性、低延迟摄取层的期望。

核心实现概览

以下是整个系统的架构图和关键代码实现。

graph TD
    subgraph "Log Sources (Microservices)"
        S1[Service 1] --> F1
        S2[Service 2] --> F1
        S3[Service 3] --> F2
    end

    subgraph "Collection Layer"
        F1[Fluentd Agent] -- TCP Forward --> EP
        F2[Fluentd Agent] -- TCP Forward --> EP
    end

    subgraph "Elixir Ingestion Service Cluster"
        EP[TCP Endpoint] --> IngestionSupervisor
        IngestionSupervisor -- supervises --> TCPListener
        IngestionSupervisor -- supervises --> LogBuffer
        IngestionSupervisor -- supervises --> DWUploader

        TCPListener -- raw logs --> LogBuffer
        LogBuffer -- writes to --> DB[(LevelDB)]
        LogBuffer -- notifies --> DWUploader
        DWUploader -- reads from --> DB
        DWUploader -- batch POST --> DWH[Data Warehouse API]
        DWUploader -- samples --> PhoenixChannel
    end

    subgraph "Real-time Dashboard"
        PhoenixChannel -- WebSocket --> SolidUI[Solid.js Frontend]
    end

1. Fluentd配置

Fluentd作为日志收集代理，其配置非常直接。核心是使用forward协议将日志以TCP流的形式发送到我们的Elixir服务集群的负载均衡器地址。

fluent.conf:

# fluent.conf
# Listen for logs from containers
<source>
  @type forward
  port 24224
  bind 0.0.0.0
</source>

# Match all logs, add metadata, and forward to Elixir service
<match **>
  @type forward
  # Use a buffer to handle temporary network issues to the Elixir layer
  # This is the first layer of buffering
  <buffer>
    @type file
    path /var/log/fluentd/buffer/elixir_forward
    flush_interval 1s
    retry_max_interval 30s
    retry_forever true
    chunk_limit_size 2M
    queue_limit_length 128
  </buffer>

  # Forward to a load balancer fronting the Elixir cluster
  <server>
    host elixir-service.internal
    port 4000
  </server>
</match>

2. Elixir服务核心逻辑

我们将使用Mix来创建一个新的Elixir应用，并引入ranch进行TCP监听，以及一个LevelDB的Erlang库eleveldb。

mix.exs:

defp deps do
  [
    {:phoenix, "~> 1.7.0"},
    {:eleveldb, git: "https://github.com/basho/eleveldb.git", tag: "2.1.0"},
    {:msgpax, "~> 2.3"}, # For decoding MessagePack from Fluentd
    {:jason, "~> 1.2"},
    # ... other deps
  ]
end

a. TCP监听与日志接收 (LogReceiver)

我们使用ranch来管理TCP连接池。每个接受的连接都会在一个专用的进程中处理，该进程负责解析Fluentd发送的MessagePack格式数据。

# lib/log_ingester/tcp_listener.ex
defmodule LogIngester.TCPListener do
  require Logger

  def start_link(opts) do
    port = Keyword.get(opts, :port, 4000)
    # Start the Ranch listener
    {:ok, _} = :ranch.start_listener(
      :log_ingester_tcp,
      :ranch_tcp,
      [{:port, port}],
      LogIngester.TCPHandler, # The connection handler module
      []
    )
    Logger.info("TCP Listener started on port #{port}")
    # We are just a listener starter, so we don't need a state
    {:ok, self()}
  end
end

# lib/log_ingester/tcp_handler.ex
defmodule LogIngester.TCPHandler do
  require Logger

  def init(socket) do
    :ok = :inet.setopts(socket, [:binary, active: :once, packet: :raw])
    {:ok, %{buffer: <<>>}}
  end

  def handle_info({:tcp, socket, data}, state) do
    # Fluentd sends MessagePack stream. We need to handle framing.
    buffer = state.buffer <> data
    process_buffer(buffer, socket)
    # Continue listening for more data
    :ok = :inet.setopts(socket, [active: :once])
    {:noreply, %{state | buffer: <<>>}}
  end

  def handle_info({:tcp_closed, _socket}, state) do
    Logger.info("Client connection closed.")
    {:stop, :normal, state}
  end

  def handle_info({:tcp_error, _socket, reason}, state) do
    Logger.error("TCP error: #{inspect(reason)}")
    {:stop, :normal, state}
  end

  # Recursively process the buffer to unpack all complete MessagePack messages
  defp process_buffer(buffer, socket) do
    case Msgpax.unpack(buffer) do
      {:ok, [tag, entries, _options], rest} ->
        # Fluentd Forward Protocol: [tag, [[timestamp, record], ...]]
        # We pass the entries to our buffer for persistence
        LogIngester.LogBuffer.buffer_logs({tag, entries})
        process_buffer(rest, socket)
      {:error, :insufficient_data} ->
        # Incomplete message, wait for more data.
        # This part of the implementation is simplified. In production,
                # you'd need to carry the remaining buffer in the state.
        :ok
      {:error, reason} ->
        Logger.error("MessagePack decoding failed: #{inspect(reason)}")
        :ok # Skip malformed data
    end
  end
end

b. 持久化缓冲层 (LogBuffer with LevelDB)

这是一个GenServer，负责与LevelDB交互。它接收日志批次，为其分配一个单调递增的序列ID，然后将其写入LevelDB。这是保证数据持久性的核心。

# lib/log_ingester/log_buffer.ex
defmodule LogIngester.LogBuffer do
  use GenServer
  require Logger

  # Public API
  def start_link(opts) do
    db_path = Keyword.fetch!(opts, :db_path)
    GenServer.start_link(__MODULE__, db_path, name: __MODULE__)
  end

  def buffer_logs(logs) do
    GenServer.cast(__MODULE__, {:buffer, logs})
  end

  # Server Callbacks
  @impl true
  def init(db_path) do
    case :eleveldb.open(db_path, []) do
      {:ok, db_ref} ->
        # Get the last used sequence number or start from 0
        last_seq = get_last_sequence(db_ref)
        Logger.info("LogBuffer started. DB path: #{db_path}. Last sequence: #{last_seq}")
        {:ok, %{db: db_ref, seq: last_seq}}
      {:error, reason} ->
        Logger.error("Failed to open LevelDB: #{inspect(reason)}")
        {:stop, reason}
    end
  end

  @impl true
  def handle_cast({:buffer, {tag, entries}}, state) do
    new_seq = state.seq + 1
    # Key: 64-bit integer sequence number, big-endian for correct ordering
    # Value: Erlang term serialized to binary
    key = <<new_seq::integer-size(64)>>
    value = :erlang.term_to_binary({tag, entries})

    case :eleveldb.put(state.db, key, value, []) do
      :ok ->
        # Notify the uploader that new data is available
        # This is a fire-and-forget notification
        GenServer.cast(LogIngester.DWUploader, :new_data)
        {:noreply, %{state | seq: new_seq}}
      {:error, reason} ->
        Logger.error("Failed to write to LevelDB: #{inspect(reason)}")
        # In a real system, you might have a circuit breaker or retry logic here
        {:noreply, state}
    end
  end

  defp get_last_sequence(db) do
    # Use an iterator to find the last key
    case :eleveldb.iterator(db, [reverse: true, keys_only: true]) do
      {:ok, iter} ->
        case :eleveldb.iterator_move(iter, :first) do
          {:ok, <<last_seq::integer-size(64)>>} ->
            :eleveldb.iterator_close(iter)
            last_seq
          _ ->
            :eleveldb.iterator_close(iter)
            0
        end
      _ ->
        0
    end
  end
end

c. 数据上传与分发 (DWUploader)

这个GenServer负责从LevelDB读取数据，将其分批次上传到数据仓库，并推送样本到Phoenix Channels。它维护一个读取位置（last_read_seq），以确保不会重复处理数据。

# lib/log_ingester/dw_uploader.ex
defmodule LogIngester.DWUploader do
  use GenServer
  require Logger

  # Simplified batch uploader
  @batch_size 1000
  @upload_interval :timer.seconds(5)

  def start_link(opts) do
    GenServer.start_link(__MODULE__, opts, name: __MODULE__)
  end

  @impl true
  def init(opts) do
    # Get the same DB reference as the buffer
    # In a real app, this might be passed from a central registry or supervisor
    {:ok, db_ref} = :eleveldb.open(Keyword.fetch!(opts, :db_path), [])
    # Initialize a timer to trigger uploads periodically
    :timer.send_interval(@upload_interval, self(), :upload_tick)
    {:ok, %{db: db_ref, last_read_seq: 0, in_flight: false}}
  end

  # Triggered by LogBuffer when new data is written
  @impl true
  def handle_cast(:new_data, state) do
    # If an upload is not already in progress, start one immediately
    if !state.in_flight do
      spawn_link(fn -> process_and_upload(state.db, state.last_read_seq) end)
      {:noreply, %{state | in_flight: true}}
    else
      {:noreply, state}
    end
  end

  # Triggered by the periodic timer
  @impl true
  def handle_info(:upload_tick, state) do
    if !state.in_flight do
      spawn_link(fn -> process_and_upload(state.db, state.last_read_seq) end)
      {:noreply, %{state | in_flight: true}}
    else
      {:noreply, state}
    end
  end

  # Triggered by the upload task when it completes
  @impl true
  def handle_info({:upload_complete, new_read_seq}, state) do
    # Update our position and mark that we are ready for a new upload
    {:noreply, %{state | last_read_seq: new_read_seq, in_flight: false}}
  end
  def handle_info({:upload_failed, _reason}, state) do
    # The read sequence is not updated, so the next attempt will retry the same batch.
    # A backoff strategy is needed for a production system.
    Logger.error("Upload failed. Will retry later.")
    {:noreply, %{state | in_flight: false}}
  end

  defp process_and_upload(db, last_read_seq) do
    # Read a batch from LevelDB starting after the last read sequence
    start_key = <<(last_read_seq + 1)::integer-size(64)>>
    {:ok, records, new_read_seq} = read_batch_from_db(db, start_key, @batch_size)

    if Enum.any?(records) do
      # In a real implementation, you'd use a proper HTTP client like Finch
      case upload_to_data_warehouse(records) do
        :ok ->
          # Sample and broadcast to Phoenix Channel (e.g., every 10th record)
          records
          |> Enum.with_index()
          |> Enum.filter(fn {_, index} -> rem(index, 10) == 0 end)
          |> Enum.map(fn {record, _} -> record end)
          |> broadcast_to_frontend()

          # After successful upload, tell the main process to update its state
          GenServer.cast(LogIngester.DWUploader, {:upload_complete, new_read_seq})
          # Clean up processed records from LevelDB
          delete_range_from_db(db, <<(last_read_seq + 1)::integer-size(64)>>, <<new_read_seq::integer-size(64)>>)
        {:error, reason} ->
          GenServer.cast(LogIngester.DWUploader, {:upload_failed, reason})
      end
    else
      # No new records to process
      GenServer.cast(LogIngester.DWUploader, {:upload_complete, last_read_seq})
    end
  end
  
  # ... helper functions for read_batch_from_db, upload_to_data_warehouse, broadcast_to_frontend, delete_range_from_db
end

3. Solid.js 实时前端

前端部分使用Solid.js来创建一个响应式界面，通过WebSocket连接到Phoenix Channels。

// src/components/LogStream.jsx
import { createSignal, onMount, onCleanup, For } from "solid-js";
import { Socket } from "phoenix";

function LogStream() {
  const [logs, setLogs] = createSignal([]);
  let channel;

  onMount(() => {
    const socket = new Socket("/socket", { params: { token: window.userToken } });
    socket.connect();

    channel = socket.channel("logs:realtime", {});
    channel.on("new_log", (payload) => {
      // Prepend new logs and keep the list at a manageable size to avoid browser slowdown
      setLogs((prevLogs) => [payload, ...prevLogs.slice(0, 500)]);
    });

    channel.join()
      .receive("ok", resp => { console.log("Joined successfully", resp) })
      .receive("error", resp => { console.log("Unable to join", resp) });
  });

  onCleanup(() => {
    if (channel) {
      channel.leave();
    }
  });

  return (
    <div class="log-container">
      <h1>Real-time Log Stream</h1>
      <pre>
        <For each={logs()}>
          {(log) => <div>{JSON.stringify(log)}</div>}
        </For>
      </pre>
    </div>
  );
}

export default LogStream;

架构的扩展性与局限性

此架构通过简单地增加Elixir服务节点即可实现水平扩展。由于每个节点都处理一部分TCP连接并拥有自己独立的LevelDB缓冲，它们之间没有共享状态，这是一种经典的Share-Nothing架构。上游的负载均衡器可以将Fluentd的连接分散到集群中的所有节点上。

然而，该架构的局限性也十分明确：

数据持久性边界: LevelDB提供的持久性是节点级别的。如果一个云主机实例及其挂载的磁盘同时发生不可逆转的故障，该节点上已缓冲但尚未成功上传到数据仓库的日志将会永久丢失。这与Kafka提供的集群级、可复制的持久性有本质区别。因此，此方案适用于那些可以容忍极低概率下分钟级别数据丢失的场景。
查询能力: 该系统被设计为一个高效的“直通管道”（pass-through pipeline），而非一个可查询的日志存储系统。LevelDB中的数据是短暂的，并且只能按序访问，不支持复杂的即席查询。所有分析和查询需求都必须由下游的最终目的地——数据仓库来满足。
处理逻辑复杂度: 如果日志处理逻辑变得非常复杂，例如需要跨越长时间窗口的状态聚合或与外部系统进行大量同步交互，那么使用像Flink这样专用的流处理引擎可能更为合适。当前的设计最适合无状态或简单有状态的转换和过滤任务。
LevelDB的性能考量: LevelDB的写入性能非常高，但其后台的compaction过程可能会对磁盘I/O产生瞬时压力。在生产环境中，需要对磁盘性能进行监控，并选择合适的SSD硬件来保证稳定的性能。

Elixir Data Warehouse LevelDB Fluentd Solid.js

基于 Kong、Valtio 和动态样式注入的微前端架构决策与实现

2023-10-27 架构设计

Valtio 样式方案 Kong 微前端架构权衡

构建基于强化学习的动态UI自适应系统：从WebSocket向量化状态到PostCSS原子化执行的全链路追踪实践

2023-10-27 后端架构

WebSockets 强化学习 Jaeger PostCSS 向量